From 8a754e0858d922e955e71b253c139e071ecec432 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 28 Apr 2024 18:04:21 +0200 Subject: Adding upstream version 2.14.3. Signed-off-by: Daniel Baumann --- .../module_utils/facts/network/generic_bsd.py | 321 +++++++++++++++++++++ 1 file changed, 321 insertions(+) create mode 100644 lib/ansible/module_utils/facts/network/generic_bsd.py (limited to 'lib/ansible/module_utils/facts/network/generic_bsd.py') diff --git a/lib/ansible/module_utils/facts/network/generic_bsd.py b/lib/ansible/module_utils/facts/network/generic_bsd.py new file mode 100644 index 0000000..8d640f2 --- /dev/null +++ b/lib/ansible/module_utils/facts/network/generic_bsd.py @@ -0,0 +1,321 @@ +# 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 . + +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_ 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 default + # route -n get -inet6 default + # to find out the default outgoing interface, address, and gateway + + command = dict(v4=[route_path, '-n', 'get', 'default'], + v6=[route_path, '-n', 'get', '-inet6', 'default']) + + 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.strip().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] + # help pick the right interface address on OpenBSD + if words[0] == 'if address': + interface[v]['address'] = words[1] + # help pick the right interface address on NetBSD + if words[0] == 'local addr': + interface[v]['address'] = 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 macOS 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 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: + # Don't just assume columns, use "netmask" as the index for the prior column + try: + netmask_idx = words.index('netmask') + 1 + except ValueError: + netmask_idx = 3 + + # deal with hex netmask + if re.match('([0-9a-f]){8}$', words[netmask_idx]): + netmask = '0x' + words[netmask_idx] + else: + netmask = words[netmask_idx] + + if netmask.startswith('0x'): + address['netmask'] = socket.inet_ntoa(struct.pack('!L', int(netmask, base=16))) + else: + # otherwise assume this is a dotted quad + address['netmask'] = netmask + # 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 + try: + broadcast_idx = words.index('broadcast') + 1 + except ValueError: + address['broadcast'] = socket.inet_ntoa(struct.pack('!L', address_bin | (~netmask_bin & 0xffffffff))) + else: + address['broadcast'] = words[broadcast_idx] + + # 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] + + ipinfo = [] + if 'address' in defaults: + ipinfo = [x for x in ifinfo[ip_type] if x['address'] == defaults['address']] + + if len(ipinfo) == 0: + ipinfo = ifinfo[ip_type] + + if len(ipinfo) > 0: + for item in ipinfo[0]: + defaults[item] = ipinfo[0][item] -- cgit v1.2.3