Coverage for lib/ansible/module_utils/facts/network/generic_bsd.py : 9%

# This file is part of Ansible 

# 

# Ansible is free software: you can redistribute it and/or modify 

# it under the terms of the GNU General Public License as published by 

# the Free Software Foundation, either version 3 of the License, or 

# (at your option) any later version. 

# 

# Ansible is distributed in the hope that it will be useful, 

# but WITHOUT ANY WARRANTY; without even the implied warranty of 

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

# GNU General Public License for more details. 

# 

# You should have received a copy of the GNU General Public License 

# along with Ansible. If not, see <http://www.gnu.org/licenses/>. 

from __future__ import (absolute_import, division, print_function) 

__metaclass__ = type 

import re 

import socket 

import struct 

from ansible.module_utils.facts.network.base import Network 

class GenericBsdIfconfigNetwork(Network): 

""" 

This is a generic BSD subclass of Network using the ifconfig command. 

It defines 

- interfaces (a list of interface names) 

- interface_<name> dictionary of ipv4, ipv6, and mac address information. 

- all_ipv4_addresses and all_ipv6_addresses: lists of all configured addresses. 

""" 

platform = 'Generic_BSD_Ifconfig' 

def populate(self, collected_facts=None): 

network_facts = {} 

ifconfig_path = self.module.get_bin_path('ifconfig') 

if ifconfig_path is None: 

return network_facts 

route_path = self.module.get_bin_path('route') 

if route_path is None: 

return network_facts 

default_ipv4, default_ipv6 = self.get_default_interfaces(route_path) 

interfaces, ips = self.get_interfaces_info(ifconfig_path) 

interfaces = self.detect_type_media(interfaces) 

self.merge_default_interface(default_ipv4, interfaces, 'ipv4') 

self.merge_default_interface(default_ipv6, interfaces, 'ipv6') 

network_facts['interfaces'] = sorted(list(interfaces.keys())) 

for iface in interfaces: 

network_facts[iface] = interfaces[iface] 

network_facts['default_ipv4'] = default_ipv4 

network_facts['default_ipv6'] = default_ipv6 

network_facts['all_ipv4_addresses'] = ips['all_ipv4_addresses'] 

network_facts['all_ipv6_addresses'] = ips['all_ipv6_addresses'] 

return network_facts 

def detect_type_media(self, interfaces): 

for iface in interfaces: 

if 'media' in interfaces[iface]: 

if 'ether' in interfaces[iface]['media'].lower(): 

interfaces[iface]['type'] = 'ether' 

return interfaces 

def get_default_interfaces(self, route_path): 

# Use the commands: 

# route -n get 8.8.8.8 -> Google public DNS 

# route -n get -inet6 2404:6800:400a:800::1012 -> ipv6.google.com 

# to find out the default outgoing interface, address, and gateway 

command = dict(v4=[route_path, '-n', 'get', '8.8.8.8'], 

v6=[route_path, '-n', 'get', '-inet6', '2404:6800:400a:800::1012']) 

interface = dict(v4={}, v6={}) 

for v in 'v4', 'v6': 

if v == 'v6' and not socket.has_ipv6: 

continue 

rc, out, err = self.module.run_command(command[v]) 

if not out: 

# v6 routing may result in 

# RTNETLINK answers: Invalid argument 

continue 

for line in out.splitlines(): 

words = line.split() 

# Collect output from route command 

if len(words) > 1: 

if words[0] == 'interface:': 

interface[v]['interface'] = words[1] 

if words[0] == 'gateway:': 

interface[v]['gateway'] = words[1] 

return interface['v4'], interface['v6'] 

def get_interfaces_info(self, ifconfig_path, ifconfig_options='-a'): 

interfaces = {} 

current_if = {} 

ips = dict( 

all_ipv4_addresses=[], 

all_ipv6_addresses=[], 

) 

# FreeBSD, DragonflyBSD, NetBSD, OpenBSD and OS X all implicitly add '-a' 

# when running the command 'ifconfig'. 

# Solaris must explicitly run the command 'ifconfig -a'. 

rc, out, err = self.module.run_command([ifconfig_path, ifconfig_options]) 

for line in out.splitlines(): 

if line: 

words = line.split() 

if words[0] == 'pass': 

continue 

elif re.match(r'^\S', line) and len(words) > 3: 

current_if = self.parse_interface_line(words) 

interfaces[current_if['device']] = current_if 

elif words[0].startswith('options='): 

self.parse_options_line(words, current_if, ips) 

elif words[0] == 'nd6': 

self.parse_nd6_line(words, current_if, ips) 

elif words[0] == 'ether': 

self.parse_ether_line(words, current_if, ips) 

elif words[0] == 'media:': 

self.parse_media_line(words, current_if, ips) 

elif words[0] == 'status:': 

self.parse_status_line(words, current_if, ips) 

elif words[0] == 'lladdr': 

self.parse_lladdr_line(words, current_if, ips) 

elif words[0] == 'inet': 

self.parse_inet_line(words, current_if, ips) 

elif words[0] == 'inet6': 

self.parse_inet6_line(words, current_if, ips) 

elif words[0] == 'tunnel': 

self.parse_tunnel_line(words, current_if, ips) 

else: 

self.parse_unknown_line(words, current_if, ips) 

return interfaces, ips 

def parse_interface_line(self, words): 

device = words[0][0:-1] 

current_if = {'device': device, 'ipv4': [], 'ipv6': [], 'type': 'unknown'} 

current_if['flags'] = self.get_options(words[1]) 

if 'LOOPBACK' in current_if['flags']: 

current_if['type'] = 'loopback' 

current_if['macaddress'] = 'unknown' # will be overwritten later 

if len(words) >= 5: # Newer FreeBSD versions 

current_if['metric'] = words[3] 

current_if['mtu'] = words[5] 

else: 

current_if['mtu'] = words[3] 

return current_if 

def parse_options_line(self, words, current_if, ips): 

# Mac has options like this... 

current_if['options'] = self.get_options(words[0]) 

def parse_nd6_line(self, words, current_if, ips): 

# FreeBSD has options like this... 

current_if['options'] = self.get_options(words[1]) 

def parse_ether_line(self, words, current_if, ips): 

current_if['macaddress'] = words[1] 

current_if['type'] = 'ether' 

def parse_media_line(self, words, current_if, ips): 

# not sure if this is useful - we also drop information 

current_if['media'] = words[1] 

if len(words) > 2: 

current_if['media_select'] = words[2] 

if len(words) > 3: 

current_if['media_type'] = words[3][1:] 

if len(words) > 4: 

current_if['media_options'] = self.get_options(words[4]) 

def parse_status_line(self, words, current_if, ips): 

current_if['status'] = words[1] 

def parse_lladdr_line(self, words, current_if, ips): 

current_if['lladdr'] = words[1] 

def parse_inet_line(self, words, current_if, ips): 

# netbsd show aliases like this 

# lo0: flags=8049<UP,LOOPBACK,RUNNING,MULTICAST> mtu 33184 

# inet 127.0.0.1 netmask 0xff000000 

# inet alias 127.1.1.1 netmask 0xff000000 

if words[1] == 'alias': 

del words[1] 

address = {'address': words[1]} 

# cidr style ip address (eg, 127.0.0.1/24) in inet line 

# used in netbsd ifconfig -e output after 7.1 

if '/' in address['address']: 

ip_address, cidr_mask = address['address'].split('/') 

address['address'] = ip_address 

netmask_length = int(cidr_mask) 

netmask_bin = (1 << 32) - (1 << 32 >> int(netmask_length)) 

address['netmask'] = socket.inet_ntoa(struct.pack('!L', netmask_bin)) 

if len(words) > 5: 

address['broadcast'] = words[3] 

else: 

# deal with hex netmask 

if re.match('([0-9a-f]){8}', words[3]) and len(words[3]) == 8: 

words[3] = '0x' + words[3] 

if words[3].startswith('0x'): 

address['netmask'] = socket.inet_ntoa(struct.pack('!L', int(words[3], base=16))) 

else: 

# otherwise assume this is a dotted quad 

address['netmask'] = words[3] 

# calculate the network 

address_bin = struct.unpack('!L', socket.inet_aton(address['address']))[0] 

netmask_bin = struct.unpack('!L', socket.inet_aton(address['netmask']))[0] 

address['network'] = socket.inet_ntoa(struct.pack('!L', address_bin & netmask_bin)) 

if 'broadcast' not in address: 

# broadcast may be given or we need to calculate 

if len(words) > 5: 

address['broadcast'] = words[5] 

else: 

address['broadcast'] = socket.inet_ntoa(struct.pack('!L', address_bin | (~netmask_bin & 0xffffffff))) 

# add to our list of addresses 

if not words[1].startswith('127.'): 

ips['all_ipv4_addresses'].append(address['address']) 

current_if['ipv4'].append(address) 

def parse_inet6_line(self, words, current_if, ips): 

address = {'address': words[1]} 

# using cidr style addresses, ala NetBSD ifconfig post 7.1 

if '/' in address['address']: 

ip_address, cidr_mask = address['address'].split('/') 

address['address'] = ip_address 

address['prefix'] = cidr_mask 

if len(words) > 5: 

address['scope'] = words[5] 

else: 

if (len(words) >= 4) and (words[2] == 'prefixlen'): 

address['prefix'] = words[3] 

if (len(words) >= 6) and (words[4] == 'scopeid'): 

address['scope'] = words[5] 

localhost6 = ['::1', '::1/128', 'fe80::1%lo0'] 

if address['address'] not in localhost6: 

ips['all_ipv6_addresses'].append(address['address']) 

current_if['ipv6'].append(address) 

def parse_tunnel_line(self, words, current_if, ips): 

current_if['type'] = 'tunnel' 

def parse_unknown_line(self, words, current_if, ips): 

# we are going to ignore unknown lines here - this may be 

# a bad idea - but you can override it in your subclass 

pass 

# TODO: these are module scope static function candidates 

# (most of the class is really...) 

def get_options(self, option_string): 

start = option_string.find('<') + 1 

end = option_string.rfind('>') 

if (start > 0) and (end > 0) and (end > start + 1): 

option_csv = option_string[start:end] 

return option_csv.split(',') 

else: 

return [] 

def merge_default_interface(self, defaults, interfaces, ip_type): 

if 'interface' not in defaults: 

return 

if not defaults['interface'] in interfaces: 

return 

ifinfo = interfaces[defaults['interface']] 

# copy all the interface values across except addresses 

for item in ifinfo: 

if item != 'ipv4' and item != 'ipv6': 

defaults[item] = ifinfo[item] 

if len(ifinfo[ip_type]) > 0: 

for item in ifinfo[ip_type][0]: 

defaults[item] = ifinfo[ip_type][0][item]