ACIL FM

Nama	Ukuran	Aksi
__pycache__	-	Open
aix.py	5.91 MB	View DL Edit
base.py	2.41 MB	View DL Edit
darwin.py	1.96 MB	View DL Edit
dragonfly.py	1.17 MB	View DL Edit
fc_wwn.py	5.04 MB	View DL Edit
freebsd.py	1.16 MB	View DL Edit
generic_bsd.py	12.35 MB	View DL Edit
hpux.py	2.95 MB	View DL Edit
hurd.py	3.05 MB	View DL Edit
iscsi.py	4.7 MB	View DL Edit
linux.py	16.69 MB	View DL Edit
netbsd.py	1.76 MB	View DL Edit
nvme.py	2.02 MB	View DL Edit
openbsd.py	1.56 MB	View DL Edit
sunos.py	4.69 MB	View DL Edit
__init__.py	0 B	View DL Edit

Edit file: /usr/lib/python3.9/site-packages/ansible/module_utils/facts/network/generic_bsd.py

# 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 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<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:
            # 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]

Batal