Skip to content

Commit

Permalink
refactor: remove dependency on netifaces
Browse files Browse the repository at this point in the history
netifaces is no longer being maintained and is only used by the VMWare data
source.  As such this commit replaces the calls to netifaces with cloudinit's
native netinfo.
  • Loading branch information
catmsred committed Nov 30, 2023
1 parent f5b0bad commit 5b3565e
Show file tree
Hide file tree
Showing 2 changed files with 69 additions and 83 deletions.
150 changes: 68 additions & 82 deletions cloudinit/sources/DataSourceVMware.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,9 +72,7 @@
import socket
import time

import netifaces

from cloudinit import atomic_helper, dmi, log, net, sources, util
from cloudinit import atomic_helper, dmi, log, net, netinfo, sources, util
from cloudinit.sources.helpers.vmware.imc import guestcust_util
from cloudinit.subp import ProcessExecutionError, subp, which

Expand Down Expand Up @@ -776,91 +774,52 @@ def get_default_ip_addrs():
addresses associated with the device used by the default route for a given
address.
"""
# TODO(promote and use netifaces in cloudinit.net* modules)
gateways = netifaces.gateways()
if "default" not in gateways:
return None, None

default_gw = gateways["default"]
if (
netifaces.AF_INET not in default_gw
and netifaces.AF_INET6 not in default_gw
):
return None, None

ipv4 = None
ipv6 = None
routes = netinfo.route_info()
for route in routes:
if "G" in route["flags"]:
if ipv4 and ipv4 != route["gateway"]:
LOG.debug(
"multiple ipv4 gateways: %s, %s", ipv4, route["gateway"]
)
ipv4 = route["gateway"]
elif "UG" in route["flags"]:
if ipv6 and ipv6 != route["gateway"]:
LOG.debug(
"multiple ipv6 gateways: %s, %s", ipv6, route["gateway"]
)
ipv6 = route["gateway"]

gw4 = default_gw.get(netifaces.AF_INET)
if gw4:
_, dev4 = gw4
addr4_fams = netifaces.ifaddresses(dev4)
if addr4_fams:
af_inet4 = addr4_fams.get(netifaces.AF_INET)
if af_inet4:
if len(af_inet4) > 1:
LOG.debug(
"device %s has more than one ipv4 address: %s",
dev4,
af_inet4,
)
elif "addr" in af_inet4[0]:
ipv4 = af_inet4[0]["addr"]

# Try to get the default IPv6 address by first seeing if there is a default
# IPv6 route.
gw6 = default_gw.get(netifaces.AF_INET6)
if gw6:
_, dev6 = gw6
addr6_fams = netifaces.ifaddresses(dev6)
if addr6_fams:
af_inet6 = addr6_fams.get(netifaces.AF_INET6)
if af_inet6:
if len(af_inet6) > 1:
LOG.debug(
"device %s has more than one ipv6 address: %s",
dev6,
af_inet6,
)
elif "addr" in af_inet6[0]:
ipv6 = af_inet6[0]["addr"]
# If there is an IPv4 gateway and an IPv6 gateway, return immediately to
# avoid extra work
if ipv4 and ipv6:
return ipv4, ipv4

netdev = netinfo.netdev_info()
# If there is a default IPv4 address but not IPv6, then see if there is a
# single IPv6 address associated with the same device associated with the
# default IPv4 address.
if ipv4 and not ipv6:
af_inet6 = addr4_fams.get(netifaces.AF_INET6)
if af_inet6:
if len(af_inet6) > 1:
LOG.debug(
"device %s has more than one ipv6 address: %s",
dev4,
af_inet6,
)
elif "addr" in af_inet6[0]:
ipv6 = af_inet6[0]["addr"]
if ipv4 is not None and ipv6 is None:
for dev_name in netdev.keys():
for addr in netdev[dev_name]["ipv4"]:
if addr["ip"] == ipv4 and len(netdev[dev_name]["ipv6"]) == 1:
ipv6 = netdev[dev_name]["ipv6"][0]["ip"]
break

# If there is a default IPv6 address but not IPv4, then see if there is a
# single IPv4 address associated with the same device associated with the
# default IPv6 address.
if not ipv4 and ipv6:
af_inet4 = addr6_fams.get(netifaces.AF_INET)
if af_inet4:
if len(af_inet4) > 1:
LOG.debug(
"device %s has more than one ipv4 address: %s",
dev6,
af_inet4,
)
elif "addr" in af_inet4[0]:
ipv4 = af_inet4[0]["addr"]
if ipv4 is None and ipv6 is not None:
for dev_name in netdev:
for addr in netdev[dev_name]["ipv6"]:
if addr["ip"] == ipv6 and len(netdev[dev_name]["ipv4"]) == 1:
ipv4 = netdev[dev_name]["ipv6"][0]["ip"]
break

return ipv4, ipv6


# patched socket.getfqdn() - see https://bugs.python.org/issue5004


def getfqdn(name=""):
"""Get fully qualified domain name from name.
An empty argument is interpreted as meaning the local host.
Expand Down Expand Up @@ -895,6 +854,33 @@ def is_valid_ip_addr(val):
)


def convert_to_netifaces_format(addr):
"""
Takes a cloudinit.netinfo formatted address and converts to netifaces
format, since this module was originally written with netifaces as the
network introspection module.
netifaces format:
{
"broadcast": "10.15.255.255",
"netmask": "255.240.0.0",
"addr": "10.0.1.4"
}
cloudinit.netinfo format:
{
"ip": "10.0.1.4",
"mask": "255.240.0.0",
"bcast": "10.15.255.255",
"scope": "global",
}
"""
return {
"broadcast": addr["bcast"],
"netmask": addr["mask"],
"addr": addr["ip"],
}


def get_host_info():
"""
Returns host information such as the host name and network interfaces.
Expand Down Expand Up @@ -925,16 +911,16 @@ def get_host_info():
by_ipv4 = host_info["network"]["interfaces"]["by-ipv4"]
by_ipv6 = host_info["network"]["interfaces"]["by-ipv6"]

ifaces = netifaces.interfaces()
ifaces = netinfo.netdev_info()
for dev_name in ifaces:
addr_fams = netifaces.ifaddresses(dev_name)
af_link = addr_fams.get(netifaces.AF_LINK)
af_inet4 = addr_fams.get(netifaces.AF_INET)
af_inet6 = addr_fams.get(netifaces.AF_INET6)

mac = None
if af_link and "addr" in af_link[0]:
mac = af_link[0]["addr"]
af_inet4 = []
af_inet6 = []
for addr in ifaces[dev_name]["ipv4"]:
af_inet4.append(convert_to_netifaces_format(addr))
for addr in ifaces[dev_name]["ipv6"]:
af_inet6.append(convert_to_netifaces_format(addr))

mac = ifaces[dev_name].get("hwaddr")

# Do not bother recording localhost
if mac == "00:00:00:00:00:00":
Expand Down
2 changes: 1 addition & 1 deletion tests/unittests/sources/test_vmware.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ def common_patches():
is_FreeBSD=mock.Mock(return_value=False),
),
mock.patch(
"cloudinit.sources.DataSourceVMware.netifaces.interfaces",
"cloudinit.netinfo.netdev_info",
return_value=[],
),
mock.patch(
Expand Down

0 comments on commit 5b3565e

Please sign in to comment.