Python netifaces 模块，AF_LINK 实例源码

def get_host_devices(iface_prefix=''):

    rst = {}

    for ifacename in netifaces.interfaces():

        if not ifacename.startswith(iface_prefix):
            continue

        addrs = netifaces.ifaddresses(ifacename)

        if netifaces.AF_INET in addrs and netifaces.AF_LINK in addrs:

            ips = [addr['addr'] for addr in addrs[netifaces.AF_INET]]

            for ip in ips:
                if is_ip4_loopback(ip):
                    break
            else:
                rst[ifacename] = {'INET': addrs[netifaces.AF_INET],
                                  'LINK': addrs[netifaces.AF_LINK]}

    return rst

def get_if_raw_hwaddr(iff):
    err = create_string_buffer(PCAP_ERRBUF_SIZE)
    devs = POINTER(pcap_if_t)()
    ret = b"\0\0\0\0\0\0"
    if pcap_findalldevs(byref(devs), err) < 0:
      return ret
    try:
      p = devs
      while p:
        if p.contents.name.endswith(iff.encode('ascii')):
          a = p.contents.addresses
          while a:
            if hasattr(socket, 'AF_LINK') and a.contents.addr.contents.sa_family == socket.AF_LINK:
              ap = a.contents.addr
              val = cast(ap, POINTER(sockaddr_dl))
              ret = bytes(val.contents.sdl_data[ val.contents.sdl_nlen : val.contents.sdl_nlen + val.contents.sdl_alen ])
            a = a.contents.next
          break
        p = p.contents.next
      return ret
    finally:
      pcap_freealldevs(devs)

def get_if_raw_hwaddr(iff):
    err = create_string_buffer(PCAP_ERRBUF_SIZE)
    devs = POINTER(pcap_if_t)()
    ret = b"\0\0\0\0\0\0"
    if pcap_findalldevs(byref(devs), err) < 0:
      return ret
    try:
      p = devs
      while p:
        if p.contents.name.endswith(iff.encode('ascii')):
          a = p.contents.addresses
          while a:
            if hasattr(socket, 'AF_LINK') and a.contents.addr.contents.sa_family == socket.AF_LINK:
              ap = a.contents.addr
              val = cast(ap, POINTER(sockaddr_dl))
              ret = bytes(val.contents.sdl_data[ val.contents.sdl_nlen : val.contents.sdl_nlen + val.contents.sdl_alen ])
            a = a.contents.next
          break
        p = p.contents.next
      return ret
    finally:
      pcap_freealldevs(devs)

def do_inet(self, _AF):
        """Usage : inet [option], see 'man' for options"""

        if _AF == 'AF_INET':
            print("\033[1;96m{} : \033[1;m".format(__global_iface__))
            print(netifaces.ifaddresses(__global_iface__)[netifaces.AF_INET])

        elif _AF == 'AF_LINK':
            print("\033[1;96m{} : \033[1;m".format(__global_iface__))
            print(netifaces.ifaddresses(__global_iface__)[netifaces.AF_LINK])

        elif _AF:
            syntax_err('inet')

        else:
            print("\033[1;96m{} : \033[1;m".format(__global_iface__))
            print(netifaces.ifaddresses(__global_iface__))

def get_interfaces():
    """
    Return dict with available network interfaces and their addresses.
    """

    interfaces = {}

    for interface in nif.interfaces():
        new_interface = {}
        new_interface['name'] = interface

        addrs = nif.ifaddresses(interface)

        try:
            new_interface['ip'] = addrs[nif.AF_INET][0]['addr']
        except KeyError:
            new_interface['ip'] = 'None'

        new_interface['hwaddr'] = addrs[nif.AF_LINK][0]['addr']

        interfaces[interface] = new_interface

    return interfaces

def get_if_raw_hwaddr(iff):
    err = create_string_buffer(PCAP_ERRBUF_SIZE)
    devs = POINTER(pcap_if_t)()
    ret = b"\0\0\0\0\0\0"
    if pcap_findalldevs(byref(devs), err) < 0:
      return ret
    try:
      p = devs
      while p:
        if p.contents.name.endswith(iff.encode('ascii')):
          a = p.contents.addresses
          while a:
            if hasattr(socket, 'AF_LINK') and a.contents.addr.contents.sa_family == socket.AF_LINK:
              ap = a.contents.addr
              val = cast(ap, POINTER(sockaddr_dl))
              ret = bytes(val.contents.sdl_data[ val.contents.sdl_nlen : val.contents.sdl_nlen + val.contents.sdl_alen ])
            a = a.contents.next
          break
        p = p.contents.next
      return ret
    finally:
      pcap_freealldevs(devs)

def upload(directory='.', oauth='~/oauth', remove=False,
           uploader_id=netifaces.ifaddresses('eth0')[netifaces.AF_LINK][0]['addr'].upper()):
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)
    logger.info("Init Daemon - Press Ctrl+C to quit")

    api = Musicmanager()
    event_handler = MusicToUpload()
    event_handler.api = api
    event_handler.path = directory
    event_handler.willDelete = remove
    event_handler.logger = logger
    if not api.login(oauth, uploader_id):
        print("Error with oauth credentials")
        sys.exit(1)
    if remove:
        files = [file for file in glob.glob(directory + '/**/*', recursive=True)]
        for file_path in files:
            if os.path.isfile(file_path):
                logger.info("Uploading : " + file_path)
                uploaded, matched, not_uploaded = api.upload(file_path, True)
                if uploaded or matched:
                    os.remove(file_path)
    observer = Observer()
    observer.schedule(event_handler, directory, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--directory", '-d', default='.', help="Music Folder to upload from (default: .)")
    parser.add_argument("--oauth", '-a', default='~/oauth', help="Path to oauth file (default: ~/oauth)")
    parser.add_argument("-r", "--remove", action='store_true', help="Remove files if present (default: False)")
    parser.add_argument("--uploader_id", '-u',
                        default=netifaces.ifaddresses('eth0')[netifaces.AF_LINK][0]['addr'].upper(),
                        help="Uploader identification (should be an uppercase MAC address) (default: <current eth0 MAC address>)")
    args = parser.parse_args()
    upload(args.directory, args.oauth, args.remove, args.uploader_id)

def mac_address():
    """
    Return the mac address of computer.
    """
    interface = netifaces.ifaddresses('en0')
    info = interface[netifaces.AF_LINK]
    if info:
        return interface[netifaces.AF_LINK][0]["addr"]

def mac_address():
    """
    Return the mac address of computer.
    """
    interface = netifaces.ifaddresses('en0')
    info = interface[netifaces.AF_LINK]
    if info:
        return interface[netifaces.AF_LINK][0]["addr"]

def get_if_raw_hwaddr(iff):
      if iff == scapy.arch.LOOPBACK_NAME:
          return (772, '\x00'*6)
      try:
          s = netifaces.ifaddresses(iff)[netifaces.AF_LINK][0]['addr']
          return struct.pack('BBBBBB', *[ int(i, 16) for i in s.split(':') ])
      except:
          raise Scapy_Exception("Error in attempting to get hw address for interface [%s]" % iff)
      return l

def mac(self, interface_name):
        """
            interface: str , Interface Name
            return: List
                    [{'addr': '00:12:34:56:78:9a'}]

        """
        interface_check = self.interfaces

        if not interface_name in interface_check:
            raise WrongInterfaceName("Wrong Interface Name %s" % interface_name)

        addrs = netifaces.ifaddresses(interface_name)

        return addrs[netifaces.AF_LINK]

def get_addresses(interface):
    addrs = netifaces.ifaddresses(interface)
    link_addr = addrs[netifaces.AF_LINK]
    iface_addrs = addrs[netifaces.AF_INET]
    iface_dict = iface_addrs[0]
    link_dict = link_addr[0]
    hwaddr = link_dict.get('addr')
    iface_addr = iface_dict.get('addr')
    iface_broadcast = iface_dict.get('broadcast')
    iface_netmask = iface_dict.get('netmask')
    return hwaddr, iface_addr, iface_broadcast, iface_netmask

def get_addresses(interface):
    addrs = netifaces.ifaddresses(interface)
    link_addr = addrs[netifaces.AF_LINK]
    iface_addrs = addrs[netifaces.AF_INET]
    iface_dict = iface_addrs[0]
    link_dict = link_addr[0]
    hwaddr = link_dict.get('addr')
    iface_addr = iface_dict.get('addr')
    iface_broadcast = iface_dict.get('broadcast')
    iface_netmask = iface_dict.get('netmask')
    return hwaddr, iface_addr, iface_broadcast, iface_netmask

def get_addresses(interface):
    addrs = netifaces.ifaddresses(interface)
    link_addr = addrs[netifaces.AF_LINK]
    iface_addrs = addrs[netifaces.AF_INET]
    iface_dict = iface_addrs[0]
    link_dict = link_addr[0]
    hwaddr = link_dict.get('addr')
    iface_addr = iface_dict.get('addr')
    iface_broadcast = iface_dict.get('broadcast')
    iface_netmask = iface_dict.get('netmask')
    return hwaddr, iface_addr, iface_broadcast, iface_netmask

def get_if_raw_hwaddr(iff):
      if iff == scapy.arch.LOOPBACK_NAME:
          return (772, '\x00'*6)
      try:
          s = netifaces.ifaddresses(iff)[netifaces.AF_LINK][0]['addr']
          return struct.pack('BBBBBB', *[ int(i, 16) for i in s.split(':') ])
      except:
          raise Scapy_Exception("Error in attempting to get hw address for interface [%s]" % iff)
      return l

def get_device_mac(self):
        try:
            return ifaddresses(self.interface)[AF_LINK][0]['addr']
        except KeyError:
            return "00:00:00:00:00:00"

def get_netiface_mac_address(interface_name):
        try:
            mac_address = str(ifaddresses(interface_name)[AF_LINK][0]['addr'])
        except:
            mac_address = None
        return mac_address

def get_netiface_ipv6_link_address(interface_name):
        try:
            mac_address = str(ifaddresses(interface_name)[AF_LINK][0]['addr'])
            parts = mac_address.split(":")
            parts.insert(3, "ff")
            parts.insert(4, "fe")
            parts[0] = "%x" % (int(parts[0], 16) ^ 2)
            ipv6Parts = []
            for i in range(0, len(parts), 2):
                ipv6Parts.append("".join(parts[i:i + 2]))
            ipv6_address = "fe80::%s" % (":".join(ipv6Parts))
        except:
            ipv6_address = None
        return ipv6_address

def get_local_mac(self, iface):
        return ni.ifaddresses(iface)[ni.AF_LINK][0]['addr']

def get_urmac():
    AF_LINK = netifaces.ifaddresses(__global_iface__)[netifaces.AF_LINK]
    urmac = str(AF_LINK).split("'addr': '")[1].split("'}")[0]
    return urmac

def get_mac_by_name(name: str) -> str:
        """
        Get MAC address of interface specified by name of the interface.
        :type name: str
        :param name: name of the network interface

        :rtype: str
        :return: string MAC address
        """
        ifa = netifaces.ifaddresses(name)
        mac = ifa[netifaces.AF_LINK][0]['addr']
        return mac

def getMac():
    """Get MAC address."""
    addrs = netifaces.ifaddresses('eno1')
    maca = addrs[netifaces.AF_LINK][0]['addr']
    return maca

def get_mac(cls, interface):
        addresses = netifaces.ifaddresses(interface)
        if netifaces.AF_LINK in addresses:
            return addresses[netifaces.AF_LINK][0]["addr"]
        return "0"

def determine_src_mac(iface):
    if iface in netifaces.interfaces():
        return netifaces.ifaddresses(iface)[netifaces.AF_LINK][0]['addr']
    return None

def match_device(self):
        interfaces = netifaces.interfaces()
        for device in interfaces:
            addresses = netifaces.ifaddresses(device)
            link_addresses = addresses[netifaces.AF_LINK]
            for address in link_addresses:
                device_hwaddr = address['addr']
                if self.mac.lower() == device_hwaddr.lower():
                    self.device_name=device
                    return device
    return -1

def get_local_port_mac(self, bridge_name):
        mac_address = netifaces.ifaddresses(
            bridge_name)[netifaces.AF_LINK][0].get('addr')
        return mac_address

def getMacAddress(iface):
  if(iface in netifaces.interfaces()):
    return netifaces.ifaddresses(iface)[netifaces.AF_LINK][0]['addr']
  else:
    return '0'

def get_if_raw_hwaddr(iff):
      if iff == scapy.arch.LOOPBACK_NAME:
          return (772, '\x00'*6)
      try:
          s = netifaces.ifaddresses(iff)[netifaces.AF_LINK][0]['addr']
          return struct.pack('BBBBBB', *[ int(i, 16) for i in s.split(':') ])
      except:
          raise Scapy_Exception("Error in attempting to get hw address for interface [%s]" % iff)
      return l

def __init__(self):
        self.__records = []
        for name in netifaces.interfaces():
            if not name.startswith("lo"):
                iface = netifaces.ifaddresses(name)
                if netifaces.AF_LINK in iface:
                    self.__records.append(
                        "MAC_{}={}".format(str(name).upper(), iface[netifaces.AF_LINK][0]['addr']))
                if netifaces.AF_INET in iface:
                    self.__records.append("IP_{}={}".format(str(name).upper(), iface[netifaces.AF_INET][0]['addr']))

        with open("/home/tipi/tipi/version.txt", 'r') as fh_in:
            for line in fh_in.readlines():
                parts = line.split('=')
                self.__records.append("{}={}".format( 
                        str(parts[0]).strip().upper(),
                        str(parts[1]).strip()))

        with open("/home/tipi/tipi.uuid", 'r') as fh_in:
            self.__records.append("UUID={}".format(fh_in.readline()))

        # This needs to work even if there is no network.. thus a catch all.
        try:
            upgradeCheck = check_output(["/home/tipi/tipi/setup/upgrade.sh"])
            latest = upgradeCheck.split('\n')[1]
            if latest.startswith("Latest Version: "):
                self.__records.append("LATEST={}".format(latest.split(':')[1].strip()))
        except Exception as e:
            logger.warn("failed to fetch latest version info")

def test_ensure_bridge_brings_up_interface(self):
        # We have to bypass the CONF.fake_network check so that netifaces
        # is actually called.
        self.flags(fake_network=False)
        fake_mac = 'aa:bb:cc:00:11:22'
        fake_ifaces = {
            netifaces.AF_LINK: [{'addr': fake_mac}]
        }
        calls = {
            'device_exists': [mock.call('bridge')],
            '_execute': [
                mock.call('brctl', 'addif', 'bridge', 'eth0',
                          run_as_root=True, check_exit_code=False),
                mock.call('ip', 'link', 'set', 'bridge', 'address', fake_mac,
                          run_as_root=True),
                mock.call('ip', 'link', 'set', 'eth0', 'up',
                          run_as_root=True, check_exit_code=False),
                mock.call('ip', 'route', 'show', 'dev', 'eth0'),
                mock.call('ip', 'addr', 'show', 'dev', 'eth0', 'scope',
                          'global'),
                ]
            }
        with test.nested(
            mock.patch.object(linux_net, 'device_exists', return_value=True),
            mock.patch.object(linux_net, '_execute', return_value=('', '')),
            mock.patch.object(netifaces, 'ifaddresses')
        ) as (device_exists, _execute, ifaddresses):
            ifaddresses.return_value = fake_ifaces
            driver = linux_net.LinuxBridgeInterfaceDriver()
            driver.ensure_bridge('bridge', 'eth0')
            device_exists.assert_has_calls(calls['device_exists'])
            _execute.assert_has_calls(calls['_execute'])
            ifaddresses.assert_called_once_with('eth0')

def get_host_ip4(iface_prefix=None, exclude_prefix=None):

    if iface_prefix is None:
        iface_prefix = ['']

    if type(iface_prefix) in types.StringTypes:
        iface_prefix = [iface_prefix]

    if exclude_prefix is not None:
        if type(exclude_prefix) in types.StringTypes:
            exclude_prefix = [exclude_prefix]

    ips = []

    for ifacename in netifaces.interfaces():

        matched = False

        for t in iface_prefix:
            if ifacename.startswith(t):
                matched = True
                break

        if exclude_prefix is not None:
            for ex in exclude_prefix:
                if ifacename.startswith(ex):
                    matched = False
                    break

        if not matched:
            continue

        addrs = netifaces.ifaddresses(ifacename)

        if netifaces.AF_INET in addrs and netifaces.AF_LINK in addrs:

            for addr in addrs[netifaces.AF_INET]:

                ip = addr['addr']

                if not is_ip4_loopback(ip):
                    ips.append(ip)

    return ips

def __init__(self, interface, network, default, elements, loggers, tunnels):
        """Function initialized the dipatcher
        Args:
            interface : name of the network interface to listen
            network : networkx graph representation of the network
            default : default template
            elements : elements of the network
            loggers : instances of the logger modules
            tunnels : tunnel configuration
        """
        self.interface = interface
        self.mac = netifaces.ifaddresses(self.interface)[netifaces.AF_LINK][0]['addr']
        self.network = network
        try:
            post('http://localhost:8080/network', json=dumps(json_graph.node_link_data(self.network)))
        except:
            logger.exception('Exception: Cannot connect to local server.')
        self.default = default
        self.devices, self.routes, self.externals = elements
        self.hpfeeds, self.dblogger = loggers
        self.tunnels = tunnels
        self.packet_queue = dict()
        self.entry_points = list()
        self.unreach_list = list()
        self.pcapy_object = pcapy.open_live(self.interface, 65535, 1, 10)
        self.decoder = ImpactDecoder.EthDecoder()
        self.ip_decoder = ImpactDecoder.IPDecoder()
        self.ip_icmp_decoder = ImpactDecoder.IPDecoderForICMP()
        self.mac_set = set([self.mac])
        for d in self.devices:
            if len(d.mac):
                self.mac_set.add(d.mac)
        for r in self.routes:
            if r.entry:
                self.entry_points.append(r)
            self.unreach_list.extend(r.unreach_list)
        logger.info('Started dispatcher listening on interface %s', self.interface)
        while True:
            try:
                (hdr, pkt) = self.pcapy_object.next()
                self.callback(hdr, pkt)
            except KeyboardInterrupt:
                return

def getNetworkInfo(self):
        """Get network information as a dict

        Returned dict example::

            {
                "eth0": {
                    "ip": {
                        "address": "192.168.0.25",
                    },
                    "ipv6": [{
                        "address": "2001:db8:3c4d::1a2f:1a2b",
                    }],
                    "mac": "aa:bb:cc:dd:ee:ff",
                },
                "wlan0": {
                    "ipv6": [{
                        "address": "2001:db8:3c4d::1a2f:1a2b",
                    }],
                    "mac": "aa:bb:cc:dd:ee:ff"
                }
            }
        """
        network_info = {}
        try:
            for interface in netifaces.interfaces():
                addresses = netifaces.ifaddresses(interface)
                interface_info = {}
                try:
                    addr = addresses[netifaces.AF_INET][0]['addr']
                    interface_info['ip'] = {}
                    interface_info['ip']['address'] = addr
                except:
                    pass
                try:
                    interface_info['ipv6'] = [{'address': addr['addr'].split('%')[0]} for addr in addresses[netifaces.AF_INET6]]
                except:
                    pass
                try:
                    interface_info['mac'] = addresses[netifaces.AF_LINK][0]['addr']
                except:
                    pass
                if interface_info:
                    network_info[interface] = interface_info
        except:
            exception('Error getting network info')
        info = {}
        info['list'] = network_info
        return info