diff options
author | James Falcon <james.falcon@canonical.com> | 2021-12-15 20:16:38 -0600 |
---|---|---|
committer | GitHub <noreply@github.com> | 2021-12-15 19:16:38 -0700 |
commit | bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf (patch) | |
tree | 1fbb3269fc87e39832e3286ef42eefd2b23fcd44 /cloudinit/net/network_state.py | |
parent | 2bcf4fa972fde686c2e3141c58e640640b44dd00 (diff) | |
download | vyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.tar.gz vyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.zip |
Adopt Black and isort (SC-700) (#1157)
Applied Black and isort, fixed any linting issues, updated tox.ini
and CI.
Diffstat (limited to 'cloudinit/net/network_state.py')
-rw-r--r-- | cloudinit/net/network_state.py | 734 |
1 files changed, 403 insertions, 331 deletions
diff --git a/cloudinit/net/network_state.py b/cloudinit/net/network_state.py index 4862bf91..d7c9144f 100644 --- a/cloudinit/net/network_state.py +++ b/cloudinit/net/network_state.py @@ -10,52 +10,70 @@ import logging import socket import struct -from cloudinit import safeyaml -from cloudinit import util +from cloudinit import safeyaml, util LOG = logging.getLogger(__name__) NETWORK_STATE_VERSION = 1 -IPV6_DYNAMIC_TYPES = ['dhcp6', - 'ipv6_slaac', - 'ipv6_dhcpv6-stateless', - 'ipv6_dhcpv6-stateful'] +IPV6_DYNAMIC_TYPES = [ + "dhcp6", + "ipv6_slaac", + "ipv6_dhcpv6-stateless", + "ipv6_dhcpv6-stateful", +] NETWORK_STATE_REQUIRED_KEYS = { - 1: ['version', 'config', 'network_state'], + 1: ["version", "config", "network_state"], } NETWORK_V2_KEY_FILTER = [ - 'addresses', 'dhcp4', 'dhcp4-overrides', 'dhcp6', 'dhcp6-overrides', - 'gateway4', 'gateway6', 'interfaces', 'match', 'mtu', 'nameservers', - 'renderer', 'set-name', 'wakeonlan', 'accept-ra' + "addresses", + "dhcp4", + "dhcp4-overrides", + "dhcp6", + "dhcp6-overrides", + "gateway4", + "gateway6", + "interfaces", + "match", + "mtu", + "nameservers", + "renderer", + "set-name", + "wakeonlan", + "accept-ra", ] NET_CONFIG_TO_V2 = { - 'bond': {'bond-ad-select': 'ad-select', - 'bond-arp-interval': 'arp-interval', - 'bond-arp-ip-target': 'arp-ip-target', - 'bond-arp-validate': 'arp-validate', - 'bond-downdelay': 'down-delay', - 'bond-fail-over-mac': 'fail-over-mac-policy', - 'bond-lacp-rate': 'lacp-rate', - 'bond-miimon': 'mii-monitor-interval', - 'bond-min-links': 'min-links', - 'bond-mode': 'mode', - 'bond-num-grat-arp': 'gratuitious-arp', - 'bond-primary': 'primary', - 'bond-primary-reselect': 'primary-reselect-policy', - 'bond-updelay': 'up-delay', - 'bond-xmit-hash-policy': 'transmit-hash-policy'}, - 'bridge': {'bridge_ageing': 'ageing-time', - 'bridge_bridgeprio': 'priority', - 'bridge_fd': 'forward-delay', - 'bridge_gcint': None, - 'bridge_hello': 'hello-time', - 'bridge_maxage': 'max-age', - 'bridge_maxwait': None, - 'bridge_pathcost': 'path-cost', - 'bridge_portprio': 'port-priority', - 'bridge_stp': 'stp', - 'bridge_waitport': None}} + "bond": { + "bond-ad-select": "ad-select", + "bond-arp-interval": "arp-interval", + "bond-arp-ip-target": "arp-ip-target", + "bond-arp-validate": "arp-validate", + "bond-downdelay": "down-delay", + "bond-fail-over-mac": "fail-over-mac-policy", + "bond-lacp-rate": "lacp-rate", + "bond-miimon": "mii-monitor-interval", + "bond-min-links": "min-links", + "bond-mode": "mode", + "bond-num-grat-arp": "gratuitious-arp", + "bond-primary": "primary", + "bond-primary-reselect": "primary-reselect-policy", + "bond-updelay": "up-delay", + "bond-xmit-hash-policy": "transmit-hash-policy", + }, + "bridge": { + "bridge_ageing": "ageing-time", + "bridge_bridgeprio": "priority", + "bridge_fd": "forward-delay", + "bridge_gcint": None, + "bridge_hello": "hello-time", + "bridge_maxage": "max-age", + "bridge_maxwait": None, + "bridge_pathcost": "path-cost", + "bridge_portprio": "port-priority", + "bridge_stp": "stp", + "bridge_waitport": None, + }, +} def from_state_file(state_file): @@ -77,17 +95,16 @@ class InvalidCommand(Exception): def ensure_command_keys(required_keys): - def wrapper(func): - @functools.wraps(func) def decorator(self, command, *args, **kwargs): if required_keys: missing_keys = diff_keys(required_keys, command) if missing_keys: - raise InvalidCommand("Command missing %s of required" - " keys %s" % (missing_keys, - required_keys)) + raise InvalidCommand( + "Command missing %s of required keys %s" + % (missing_keys, required_keys) + ) return func(self, command, *args, **kwargs) return decorator @@ -102,29 +119,28 @@ class CommandHandlerMeta(type): 'handle_' and on finding those will populate a class attribute mapping so that those methods can be quickly located and called. """ + def __new__(cls, name, parents, dct): command_handlers = {} for attr_name, attr in dct.items(): - if callable(attr) and attr_name.startswith('handle_'): - handles_what = attr_name[len('handle_'):] + if callable(attr) and attr_name.startswith("handle_"): + handles_what = attr_name[len("handle_") :] if handles_what: command_handlers[handles_what] = attr - dct['command_handlers'] = command_handlers - return super(CommandHandlerMeta, cls).__new__(cls, name, - parents, dct) + dct["command_handlers"] = command_handlers + return super(CommandHandlerMeta, cls).__new__(cls, name, parents, dct) class NetworkState(object): - def __init__(self, network_state, version=NETWORK_STATE_VERSION): self._network_state = copy.deepcopy(network_state) self._version = version - self.use_ipv6 = network_state.get('use_ipv6', False) + self.use_ipv6 = network_state.get("use_ipv6", False) self._has_default_route = None @property def config(self): - return self._network_state['config'] + return self._network_state["config"] @property def version(self): @@ -133,14 +149,14 @@ class NetworkState(object): @property def dns_nameservers(self): try: - return self._network_state['dns']['nameservers'] + return self._network_state["dns"]["nameservers"] except KeyError: return [] @property def dns_searchdomains(self): try: - return self._network_state['dns']['search'] + return self._network_state["dns"]["search"] except KeyError: return [] @@ -151,7 +167,7 @@ class NetworkState(object): return self._has_default_route def iter_interfaces(self, filter_func=None): - ifaces = self._network_state.get('interfaces', {}) + ifaces = self._network_state.get("interfaces", {}) for iface in ifaces.values(): if filter_func is None: yield iface @@ -160,7 +176,7 @@ class NetworkState(object): yield iface def iter_routes(self, filter_func=None): - for route in self._network_state.get('routes', []): + for route in self._network_state.get("routes", []): if filter_func is not None: if filter_func(route): yield route @@ -172,38 +188,37 @@ class NetworkState(object): if self._is_default_route(route): return True for iface in self.iter_interfaces(): - for subnet in iface.get('subnets', []): - for route in subnet.get('routes', []): + for subnet in iface.get("subnets", []): + for route in subnet.get("routes", []): if self._is_default_route(route): return True return False def _is_default_route(self, route): - default_nets = ('::', '0.0.0.0') + default_nets = ("::", "0.0.0.0") return ( - route.get('prefix') == 0 - and route.get('network') in default_nets + route.get("prefix") == 0 and route.get("network") in default_nets ) class NetworkStateInterpreter(metaclass=CommandHandlerMeta): initial_network_state = { - 'interfaces': {}, - 'routes': [], - 'dns': { - 'nameservers': [], - 'search': [], + "interfaces": {}, + "routes": [], + "dns": { + "nameservers": [], + "search": [], }, - 'use_ipv6': False, - 'config': None, + "use_ipv6": False, + "config": None, } def __init__(self, version=NETWORK_STATE_VERSION, config=None): self._version = version self._config = config self._network_state = copy.deepcopy(self.initial_network_state) - self._network_state['config'] = config + self._network_state["config"] = config self._parsed = False self._interface_dns_map = {} @@ -213,41 +228,41 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): @property def use_ipv6(self): - return self._network_state.get('use_ipv6') + return self._network_state.get("use_ipv6") @use_ipv6.setter def use_ipv6(self, val): - self._network_state.update({'use_ipv6': val}) + self._network_state.update({"use_ipv6": val}) def dump(self): state = { - 'version': self._version, - 'config': self._config, - 'network_state': self._network_state, + "version": self._version, + "config": self._config, + "network_state": self._network_state, } return safeyaml.dumps(state) def load(self, state): - if 'version' not in state: - LOG.error('Invalid state, missing version field') - raise ValueError('Invalid state, missing version field') + if "version" not in state: + LOG.error("Invalid state, missing version field") + raise ValueError("Invalid state, missing version field") - required_keys = NETWORK_STATE_REQUIRED_KEYS[state['version']] + required_keys = NETWORK_STATE_REQUIRED_KEYS[state["version"]] missing_keys = diff_keys(required_keys, state) if missing_keys: - msg = 'Invalid state, missing keys: %s' % (missing_keys) + msg = "Invalid state, missing keys: %s" % (missing_keys) LOG.error(msg) raise ValueError(msg) # v1 - direct attr mapping, except version - for key in [k for k in required_keys if k not in ['version']]: + for key in [k for k in required_keys if k not in ["version"]]: setattr(self, key, state[key]) def dump_network_state(self): return safeyaml.dumps(self._network_state) def as_dict(self): - return {'version': self._version, 'config': self._config} + return {"version": self._version, "config": self._config} def get_network_state(self): ns = self.network_state @@ -263,7 +278,7 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): def parse_config_v1(self, skip_broken=True): for command in self._config: - command_type = command['type'] + command_type = command["type"] try: handler = self.command_handlers[command_type] except KeyError as e: @@ -276,28 +291,29 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): if not skip_broken: raise else: - LOG.warning("Skipping invalid command: %s", command, - exc_info=True) + LOG.warning( + "Skipping invalid command: %s", command, exc_info=True + ) LOG.debug(self.dump_network_state()) for interface, dns in self._interface_dns_map.items(): iface = None try: - iface = self._network_state['interfaces'][interface] + iface = self._network_state["interfaces"][interface] except KeyError as e: raise ValueError( - 'Nameserver specified for interface {0}, ' - 'but interface {0} does not exist!'.format(interface) + "Nameserver specified for interface {0}, " + "but interface {0} does not exist!".format(interface) ) from e if iface: nameservers, search = dns - iface['dns'] = { - 'addresses': nameservers, - 'search': search, + iface["dns"] = { + "addresses": nameservers, + "search": search, } def parse_config_v2(self, skip_broken=True): for command_type, command in self._config.items(): - if command_type in ['version', 'renderer']: + if command_type in ["version", "renderer"]: continue try: handler = self.command_handlers[command_type] @@ -312,17 +328,18 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): if not skip_broken: raise else: - LOG.warning("Skipping invalid command: %s", command, - exc_info=True) + LOG.warning( + "Skipping invalid command: %s", command, exc_info=True + ) LOG.debug(self.dump_network_state()) - @ensure_command_keys(['name']) + @ensure_command_keys(["name"]) def handle_loopback(self, command): return self.handle_physical(command) - @ensure_command_keys(['name']) + @ensure_command_keys(["name"]) def handle_physical(self, command): - ''' + """ command = { 'type': 'physical', 'mac_address': 'c0:d6:9f:2c:e8:80', @@ -332,119 +349,122 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): ], 'accept-ra': 'true' } - ''' + """ - interfaces = self._network_state.get('interfaces', {}) - iface = interfaces.get(command['name'], {}) - for param, val in command.get('params', {}).items(): + interfaces = self._network_state.get("interfaces", {}) + iface = interfaces.get(command["name"], {}) + for param, val in command.get("params", {}).items(): iface.update({param: val}) # convert subnet ipv6 netmask to cidr as needed - subnets = _normalize_subnets(command.get('subnets')) + subnets = _normalize_subnets(command.get("subnets")) # automatically set 'use_ipv6' if any addresses are ipv6 if not self.use_ipv6: for subnet in subnets: - if (subnet.get('type').endswith('6') or - is_ipv6_addr(subnet.get('address'))): + if subnet.get("type").endswith("6") or is_ipv6_addr( + subnet.get("address") + ): self.use_ipv6 = True break - accept_ra = command.get('accept-ra', None) + accept_ra = command.get("accept-ra", None) if accept_ra is not None: accept_ra = util.is_true(accept_ra) - wakeonlan = command.get('wakeonlan', None) + wakeonlan = command.get("wakeonlan", None) if wakeonlan is not None: wakeonlan = util.is_true(wakeonlan) - iface.update({ - 'name': command.get('name'), - 'type': command.get('type'), - 'mac_address': command.get('mac_address'), - 'inet': 'inet', - 'mode': 'manual', - 'mtu': command.get('mtu'), - 'address': None, - 'gateway': None, - 'subnets': subnets, - 'accept-ra': accept_ra, - 'wakeonlan': wakeonlan, - }) - self._network_state['interfaces'].update({command.get('name'): iface}) + iface.update( + { + "name": command.get("name"), + "type": command.get("type"), + "mac_address": command.get("mac_address"), + "inet": "inet", + "mode": "manual", + "mtu": command.get("mtu"), + "address": None, + "gateway": None, + "subnets": subnets, + "accept-ra": accept_ra, + "wakeonlan": wakeonlan, + } + ) + self._network_state["interfaces"].update({command.get("name"): iface}) self.dump_network_state() - @ensure_command_keys(['name', 'vlan_id', 'vlan_link']) + @ensure_command_keys(["name", "vlan_id", "vlan_link"]) def handle_vlan(self, command): - ''' - auto eth0.222 - iface eth0.222 inet static - address 10.10.10.1 - netmask 255.255.255.0 - hwaddress ether BC:76:4E:06:96:B3 - vlan-raw-device eth0 - ''' - interfaces = self._network_state.get('interfaces', {}) + """ + auto eth0.222 + iface eth0.222 inet static + address 10.10.10.1 + netmask 255.255.255.0 + hwaddress ether BC:76:4E:06:96:B3 + vlan-raw-device eth0 + """ + interfaces = self._network_state.get("interfaces", {}) self.handle_physical(command) - iface = interfaces.get(command.get('name'), {}) - iface['vlan-raw-device'] = command.get('vlan_link') - iface['vlan_id'] = command.get('vlan_id') - interfaces.update({iface['name']: iface}) + iface = interfaces.get(command.get("name"), {}) + iface["vlan-raw-device"] = command.get("vlan_link") + iface["vlan_id"] = command.get("vlan_id") + interfaces.update({iface["name"]: iface}) - @ensure_command_keys(['name', 'bond_interfaces', 'params']) + @ensure_command_keys(["name", "bond_interfaces", "params"]) def handle_bond(self, command): - ''' - #/etc/network/interfaces - auto eth0 - iface eth0 inet manual - bond-master bond0 - bond-mode 802.3ad - - auto eth1 - iface eth1 inet manual - bond-master bond0 - bond-mode 802.3ad - - auto bond0 - iface bond0 inet static - address 192.168.0.10 - gateway 192.168.0.1 - netmask 255.255.255.0 - bond-slaves none - bond-mode 802.3ad - bond-miimon 100 - bond-downdelay 200 - bond-updelay 200 - bond-lacp-rate 4 - ''' + """ + #/etc/network/interfaces + auto eth0 + iface eth0 inet manual + bond-master bond0 + bond-mode 802.3ad + + auto eth1 + iface eth1 inet manual + bond-master bond0 + bond-mode 802.3ad + + auto bond0 + iface bond0 inet static + address 192.168.0.10 + gateway 192.168.0.1 + netmask 255.255.255.0 + bond-slaves none + bond-mode 802.3ad + bond-miimon 100 + bond-downdelay 200 + bond-updelay 200 + bond-lacp-rate 4 + """ self.handle_physical(command) - interfaces = self._network_state.get('interfaces') - iface = interfaces.get(command.get('name'), {}) - for param, val in command.get('params').items(): + interfaces = self._network_state.get("interfaces") + iface = interfaces.get(command.get("name"), {}) + for param, val in command.get("params").items(): iface.update({param: val}) - iface.update({'bond-slaves': 'none'}) - self._network_state['interfaces'].update({iface['name']: iface}) + iface.update({"bond-slaves": "none"}) + self._network_state["interfaces"].update({iface["name"]: iface}) # handle bond slaves - for ifname in command.get('bond_interfaces'): + for ifname in command.get("bond_interfaces"): if ifname not in interfaces: cmd = { - 'name': ifname, - 'type': 'bond', + "name": ifname, + "type": "bond", } # inject placeholder self.handle_physical(cmd) - interfaces = self._network_state.get('interfaces', {}) + interfaces = self._network_state.get("interfaces", {}) bond_if = interfaces.get(ifname) - bond_if['bond-master'] = command.get('name') + bond_if["bond-master"] = command.get("name") # copy in bond config into slave - for param, val in command.get('params').items(): + for param, val in command.get("params").items(): bond_if.update({param: val}) - self._network_state['interfaces'].update({ifname: bond_if}) + self._network_state["interfaces"].update({ifname: bond_if}) - @ensure_command_keys(['name', 'bridge_interfaces']) + @ensure_command_keys(["name", "bridge_interfaces"]) def handle_bridge(self, command): - ''' + """ auto br0 iface br0 inet static address 10.10.10.1 @@ -469,89 +489,91 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): "bridge_stp", "bridge_waitport", ] - ''' + """ # find one of the bridge port ifaces to get mac_addr # handle bridge_slaves - interfaces = self._network_state.get('interfaces', {}) - for ifname in command.get('bridge_interfaces'): + interfaces = self._network_state.get("interfaces", {}) + for ifname in command.get("bridge_interfaces"): if ifname in interfaces: continue cmd = { - 'name': ifname, + "name": ifname, } # inject placeholder self.handle_physical(cmd) - interfaces = self._network_state.get('interfaces', {}) + interfaces = self._network_state.get("interfaces", {}) self.handle_physical(command) - iface = interfaces.get(command.get('name'), {}) - iface['bridge_ports'] = command['bridge_interfaces'] - for param, val in command.get('params', {}).items(): + iface = interfaces.get(command.get("name"), {}) + iface["bridge_ports"] = command["bridge_interfaces"] + for param, val in command.get("params", {}).items(): iface.update({param: val}) # convert value to boolean - bridge_stp = iface.get('bridge_stp') + bridge_stp = iface.get("bridge_stp") if bridge_stp is not None and type(bridge_stp) != bool: - if bridge_stp in ['on', '1', 1]: + if bridge_stp in ["on", "1", 1]: bridge_stp = True - elif bridge_stp in ['off', '0', 0]: + elif bridge_stp in ["off", "0", 0]: bridge_stp = False else: raise ValueError( - 'Cannot convert bridge_stp value ({stp}) to' - ' boolean'.format(stp=bridge_stp)) - iface.update({'bridge_stp': bridge_stp}) + "Cannot convert bridge_stp value ({stp}) to" + " boolean".format(stp=bridge_stp) + ) + iface.update({"bridge_stp": bridge_stp}) - interfaces.update({iface['name']: iface}) + interfaces.update({iface["name"]: iface}) - @ensure_command_keys(['name']) + @ensure_command_keys(["name"]) def handle_infiniband(self, command): self.handle_physical(command) def _parse_dns(self, command): nameservers = [] search = [] - if 'address' in command: - addrs = command['address'] + if "address" in command: + addrs = command["address"] if not type(addrs) == list: addrs = [addrs] for addr in addrs: nameservers.append(addr) - if 'search' in command: - paths = command['search'] + if "search" in command: + paths = command["search"] if not isinstance(paths, list): paths = [paths] for path in paths: search.append(path) return nameservers, search - @ensure_command_keys(['address']) + @ensure_command_keys(["address"]) def handle_nameserver(self, command): - dns = self._network_state.get('dns') + dns = self._network_state.get("dns") nameservers, search = self._parse_dns(command) - if 'interface' in command: - self._interface_dns_map[command['interface']] = ( - nameservers, search + if "interface" in command: + self._interface_dns_map[command["interface"]] = ( + nameservers, + search, ) else: - dns['nameservers'].extend(nameservers) - dns['search'].extend(search) + dns["nameservers"].extend(nameservers) + dns["search"].extend(search) - @ensure_command_keys(['address']) + @ensure_command_keys(["address"]) def _handle_individual_nameserver(self, command, iface): - _iface = self._network_state.get('interfaces') + _iface = self._network_state.get("interfaces") nameservers, search = self._parse_dns(command) - _iface[iface]['dns'] = {'nameservers': nameservers, 'search': search} + _iface[iface]["dns"] = {"nameservers": nameservers, "search": search} - @ensure_command_keys(['destination']) + @ensure_command_keys(["destination"]) def handle_route(self, command): - self._network_state['routes'].append(_normalize_route(command)) + self._network_state["routes"].append(_normalize_route(command)) # V2 handlers def handle_bonds(self, command): - ''' + """ v2_command = { bond0: { 'interfaces': ['interface0', 'interface1'], @@ -578,12 +600,12 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): } } - ''' - self._handle_bond_bridge(command, cmd_type='bond') + """ + self._handle_bond_bridge(command, cmd_type="bond") def handle_bridges(self, command): - ''' + """ v2_command = { br0: { 'interfaces': ['interface0', 'interface1'], @@ -604,11 +626,11 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): } } - ''' - self._handle_bond_bridge(command, cmd_type='bridge') + """ + self._handle_bond_bridge(command, cmd_type="bridge") def handle_ethernets(self, command): - ''' + """ ethernets: eno1: match: @@ -644,34 +666,38 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): {'type': 'dhcp4'} ] } - ''' + """ for eth, cfg in command.items(): phy_cmd = { - 'type': 'physical', - 'name': cfg.get('set-name', eth), + "type": "physical", + "name": cfg.get("set-name", eth), } - match = cfg.get('match', {}) - mac_address = match.get('macaddress', None) + match = cfg.get("match", {}) + mac_address = match.get("macaddress", None) if not mac_address: - LOG.debug('NetworkState Version2: missing "macaddress" info ' - 'in config entry: %s: %s', eth, str(cfg)) - phy_cmd['mac_address'] = mac_address - driver = match.get('driver', None) + LOG.debug( + 'NetworkState Version2: missing "macaddress" info ' + "in config entry: %s: %s", + eth, + str(cfg), + ) + phy_cmd["mac_address"] = mac_address + driver = match.get("driver", None) if driver: - phy_cmd['params'] = {'driver': driver} - for key in ['mtu', 'match', 'wakeonlan', 'accept-ra']: + phy_cmd["params"] = {"driver": driver} + for key in ["mtu", "match", "wakeonlan", "accept-ra"]: if key in cfg: phy_cmd[key] = cfg[key] subnets = self._v2_to_v1_ipcfg(cfg) if len(subnets) > 0: - phy_cmd.update({'subnets': subnets}) + phy_cmd.update({"subnets": subnets}) - LOG.debug('v2(ethernets) -> v1(physical):\n%s', phy_cmd) + LOG.debug("v2(ethernets) -> v1(physical):\n%s", phy_cmd) self.handle_physical(phy_cmd) def handle_vlans(self, command): - ''' + """ v2_vlans = { 'eth0.123': { 'id': 123, @@ -687,41 +713,43 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): 'vlan_id': 123, 'subnets': [{'type': 'dhcp4'}], } - ''' + """ for vlan, cfg in command.items(): vlan_cmd = { - 'type': 'vlan', - 'name': vlan, - 'vlan_id': cfg.get('id'), - 'vlan_link': cfg.get('link'), + "type": "vlan", + "name": vlan, + "vlan_id": cfg.get("id"), + "vlan_link": cfg.get("link"), } - if 'mtu' in cfg: - vlan_cmd['mtu'] = cfg['mtu'] + if "mtu" in cfg: + vlan_cmd["mtu"] = cfg["mtu"] subnets = self._v2_to_v1_ipcfg(cfg) if len(subnets) > 0: - vlan_cmd.update({'subnets': subnets}) - LOG.debug('v2(vlans) -> v1(vlan):\n%s', vlan_cmd) + vlan_cmd.update({"subnets": subnets}) + LOG.debug("v2(vlans) -> v1(vlan):\n%s", vlan_cmd) self.handle_vlan(vlan_cmd) def handle_wifis(self, command): - LOG.warning('Wifi configuration is only available to distros with' - ' netplan rendering support.') + LOG.warning( + "Wifi configuration is only available to distros with" + " netplan rendering support." + ) def _v2_common(self, cfg): - LOG.debug('v2_common: handling config:\n%s', cfg) + LOG.debug("v2_common: handling config:\n%s", cfg) for iface, dev_cfg in cfg.items(): - if 'set-name' in dev_cfg: - set_name_iface = dev_cfg.get('set-name') + if "set-name" in dev_cfg: + set_name_iface = dev_cfg.get("set-name") if set_name_iface: iface = set_name_iface - if 'nameservers' in dev_cfg: - search = dev_cfg.get('nameservers').get('search', []) - dns = dev_cfg.get('nameservers').get('addresses', []) - name_cmd = {'type': 'nameserver'} + if "nameservers" in dev_cfg: + search = dev_cfg.get("nameservers").get("search", []) + dns = dev_cfg.get("nameservers").get("addresses", []) + name_cmd = {"type": "nameserver"} if len(search) > 0: - name_cmd.update({'search': search}) + name_cmd.update({"search": search}) if len(dns) > 0: - name_cmd.update({'address': dns}) + name_cmd.update({"address": dns}) self.handle_nameserver(name_cmd) self._handle_individual_nameserver(name_cmd, iface) @@ -729,98 +757,110 @@ class NetworkStateInterpreter(metaclass=CommandHandlerMeta): """Common handler for bond and bridge types""" # inverse mapping for v2 keynames to v1 keynames - v2key_to_v1 = dict((v, k) for k, v in - NET_CONFIG_TO_V2.get(cmd_type).items()) + v2key_to_v1 = dict( + (v, k) for k, v in NET_CONFIG_TO_V2.get(cmd_type).items() + ) for item_name, item_cfg in command.items(): - item_params = dict((key, value) for (key, value) in - item_cfg.items() if key not in - NETWORK_V2_KEY_FILTER) + item_params = dict( + (key, value) + for (key, value) in item_cfg.items() + if key not in NETWORK_V2_KEY_FILTER + ) # we accept the fixed spelling, but write the old for compatibility # Xenial does not have an updated netplan which supports the # correct spelling. LP: #1756701 - params = item_params.get('parameters', {}) - grat_value = params.pop('gratuitous-arp', None) + params = item_params.get("parameters", {}) + grat_value = params.pop("gratuitous-arp", None) if grat_value: - params['gratuitious-arp'] = grat_value + params["gratuitious-arp"] = grat_value v1_cmd = { - 'type': cmd_type, - 'name': item_name, - cmd_type + '_interfaces': item_cfg.get('interfaces'), - 'params': dict((v2key_to_v1[k], v) for k, v in params.items()) + "type": cmd_type, + "name": item_name, + cmd_type + "_interfaces": item_cfg.get("interfaces"), + "params": dict((v2key_to_v1[k], v) for k, v in params.items()), } - if 'mtu' in item_cfg: - v1_cmd['mtu'] = item_cfg['mtu'] + if "mtu" in item_cfg: + v1_cmd["mtu"] = item_cfg["mtu"] subnets = self._v2_to_v1_ipcfg(item_cfg) if len(subnets) > 0: - v1_cmd.update({'subnets': subnets}) + v1_cmd.update({"subnets": subnets}) - LOG.debug('v2(%s) -> v1(%s):\n%s', cmd_type, cmd_type, v1_cmd) + LOG.debug("v2(%s) -> v1(%s):\n%s", cmd_type, cmd_type, v1_cmd) if cmd_type == "bridge": self.handle_bridge(v1_cmd) elif cmd_type == "bond": self.handle_bond(v1_cmd) else: - raise ValueError('Unknown command type: {cmd_type}'.format( - cmd_type=cmd_type)) + raise ValueError( + "Unknown command type: {cmd_type}".format( + cmd_type=cmd_type + ) + ) def _v2_to_v1_ipcfg(self, cfg): """Common ipconfig extraction from v2 to v1 subnets array.""" def _add_dhcp_overrides(overrides, subnet): - if 'route-metric' in overrides: - subnet['metric'] = overrides['route-metric'] + if "route-metric" in overrides: + subnet["metric"] = overrides["route-metric"] subnets = [] - if cfg.get('dhcp4'): - subnet = {'type': 'dhcp4'} - _add_dhcp_overrides(cfg.get('dhcp4-overrides', {}), subnet) + if cfg.get("dhcp4"): + subnet = {"type": "dhcp4"} + _add_dhcp_overrides(cfg.get("dhcp4-overrides", {}), subnet) subnets.append(subnet) - if cfg.get('dhcp6'): - subnet = {'type': 'dhcp6'} + if cfg.get("dhcp6"): + subnet = {"type": "dhcp6"} self.use_ipv6 = True - _add_dhcp_overrides(cfg.get('dhcp6-overrides', {}), subnet) + _add_dhcp_overrides(cfg.get("dhcp6-overrides", {}), subnet) subnets.append(subnet) gateway4 = None gateway6 = None nameservers = {} - for address in cfg.get('addresses', []): + for address in cfg.get("addresses", []): subnet = { - 'type': 'static', - 'address': address, + "type": "static", + "address": address, } if ":" in address: - if 'gateway6' in cfg and gateway6 is None: - gateway6 = cfg.get('gateway6') - subnet.update({'gateway': gateway6}) + if "gateway6" in cfg and gateway6 is None: + gateway6 = cfg.get("gateway6") + subnet.update({"gateway": gateway6}) else: - if 'gateway4' in cfg and gateway4 is None: - gateway4 = cfg.get('gateway4') - subnet.update({'gateway': gateway4}) + if "gateway4" in cfg and gateway4 is None: + gateway4 = cfg.get("gateway4") + subnet.update({"gateway": gateway4}) - if 'nameservers' in cfg and not nameservers: - addresses = cfg.get('nameservers').get('addresses') + if "nameservers" in cfg and not nameservers: + addresses = cfg.get("nameservers").get("addresses") if addresses: - nameservers['dns_nameservers'] = addresses - search = cfg.get('nameservers').get('search') + nameservers["dns_nameservers"] = addresses + search = cfg.get("nameservers").get("search") if search: - nameservers['dns_search'] = search + nameservers["dns_search"] = search subnet.update(nameservers) subnets.append(subnet) routes = [] - for route in cfg.get('routes', []): - routes.append(_normalize_route( - {'destination': route.get('to'), 'gateway': route.get('via')})) + for route in cfg.get("routes", []): + routes.append( + _normalize_route( + { + "destination": route.get("to"), + "gateway": route.get("via"), + } + ) + ) # v2 routes are bound to the interface, in v1 we add them under # the first subnet since there isn't an equivalent interface level. if len(subnets) and len(routes): - subnets[0]['routes'] = routes + subnets[0]["routes"] = routes return subnets @@ -830,18 +870,25 @@ def _normalize_subnet(subnet): subnet = copy.deepcopy(subnet) normal_subnet = dict((k, v) for k, v in subnet.items() if v) - if subnet.get('type') in ('static', 'static6'): + if subnet.get("type") in ("static", "static6"): normal_subnet.update( - _normalize_net_keys(normal_subnet, address_keys=( - 'address', 'ip_address',))) - normal_subnet['routes'] = [_normalize_route(r) - for r in subnet.get('routes', [])] + _normalize_net_keys( + normal_subnet, + address_keys=( + "address", + "ip_address", + ), + ) + ) + normal_subnet["routes"] = [ + _normalize_route(r) for r in subnet.get("routes", []) + ] def listify(snet, name): if name in snet and not isinstance(snet[name], list): snet[name] = snet[name].split() - for k in ('dns_search', 'dns_nameservers'): + for k in ("dns_search", "dns_nameservers"): listify(normal_subnet, k) return normal_subnet @@ -865,15 +912,16 @@ def _normalize_net_keys(network, address_keys=()): addr_key = key break if not addr_key: - message = ( - 'No config network address keys [%s] found in %s' % - (','.join(address_keys), network)) + message = "No config network address keys [%s] found in %s" % ( + ",".join(address_keys), + network, + ) LOG.error(message) raise ValueError(message) addr = net.get(addr_key) ipv6 = is_ipv6_addr(addr) - netmask = net.get('netmask') + netmask = net.get("netmask") if "/" in addr: addr_part, _, maybe_prefix = addr.partition("/") net[addr_key] = addr_part @@ -884,23 +932,26 @@ def _normalize_net_keys(network, address_keys=()): prefix = mask_to_net_prefix(maybe_prefix) elif netmask: prefix = mask_to_net_prefix(netmask) - elif 'prefix' in net: - prefix = int(net['prefix']) + elif "prefix" in net: + prefix = int(net["prefix"]) else: prefix = 64 if ipv6 else 24 - if 'prefix' in net and str(net['prefix']) != str(prefix): - LOG.warning("Overwriting existing 'prefix' with '%s' in " - "network info: %s", prefix, net) - net['prefix'] = prefix + if "prefix" in net and str(net["prefix"]) != str(prefix): + LOG.warning( + "Overwriting existing 'prefix' with '%s' in network info: %s", + prefix, + net, + ) + net["prefix"] = prefix if ipv6: # TODO: we could/maybe should add this back with the very uncommon # 'netmask' for ipv6. We need a 'net_prefix_to_ipv6_mask' for that. - if 'netmask' in net: - del net['netmask'] + if "netmask" in net: + del net["netmask"] else: - net['netmask'] = net_prefix_to_ipv4_mask(net['prefix']) + net["netmask"] = net_prefix_to_ipv4_mask(net["prefix"]) return net @@ -913,25 +964,28 @@ def _normalize_route(route): 'prefix': the network prefix for address as an integer. 'metric': integer metric (only if present in input). 'netmask': netmask (string) equivalent to prefix iff network is ipv4. - """ + """ # Prune None-value keys. Specifically allow 0 (a valid metric). - normal_route = dict((k, v) for k, v in route.items() - if v not in ("", None)) - if 'destination' in normal_route: - normal_route['network'] = normal_route['destination'] - del normal_route['destination'] + normal_route = dict( + (k, v) for k, v in route.items() if v not in ("", None) + ) + if "destination" in normal_route: + normal_route["network"] = normal_route["destination"] + del normal_route["destination"] normal_route.update( _normalize_net_keys( - normal_route, address_keys=('network', 'destination'))) + normal_route, address_keys=("network", "destination") + ) + ) - metric = normal_route.get('metric') + metric = normal_route.get("metric") if metric: try: - normal_route['metric'] = int(metric) + normal_route["metric"] = int(metric) except ValueError as e: raise TypeError( - 'Route config metric {} is not an integer'.format(metric) + "Route config metric {} is not an integer".format(metric) ) from e return normal_route @@ -952,10 +1006,10 @@ def subnet_is_ipv6(subnet): """Common helper for checking network_state subnets for ipv6.""" # 'static6', 'dhcp6', 'ipv6_dhcpv6-stateful', 'ipv6_dhcpv6-stateless' or # 'ipv6_slaac' - if subnet['type'].endswith('6') or subnet['type'] in IPV6_DYNAMIC_TYPES: + if subnet["type"].endswith("6") or subnet["type"] in IPV6_DYNAMIC_TYPES: # This is a request either static6 type or DHCPv6. return True - elif subnet['type'] == 'static' and is_ipv6_addr(subnet.get('address')): + elif subnet["type"] == "static" and is_ipv6_addr(subnet.get("address")): return True return False @@ -967,7 +1021,8 @@ def net_prefix_to_ipv4_mask(prefix): 24 -> "255.255.255.0" Also supports input as a string.""" mask = socket.inet_ntoa( - struct.pack(">I", (0xffffffff << (32 - int(prefix)) & 0xffffffff))) + struct.pack(">I", (0xFFFFFFFF << (32 - int(prefix)) & 0xFFFFFFFF)) + ) return mask @@ -990,14 +1045,14 @@ def ipv4_mask_to_net_prefix(mask): else: raise TypeError("mask '%s' is not a string or int") - if '.' not in mask: + if "." not in mask: raise ValueError("netmask '%s' does not contain a '.'" % mask) toks = mask.split(".") if len(toks) != 4: raise ValueError("netmask '%s' had only %d parts" % (mask, len(toks))) - return sum([bin(int(x)).count('1') for x in toks]) + return sum([bin(int(x)).count("1") for x in toks]) def ipv6_mask_to_net_prefix(mask): @@ -1017,14 +1072,30 @@ def ipv6_mask_to_net_prefix(mask): else: raise TypeError("mask '%s' is not a string or int") - if ':' not in mask: + if ":" not in mask: raise ValueError("mask '%s' does not have a ':'") - bitCount = [0, 0x8000, 0xc000, 0xe000, 0xf000, 0xf800, 0xfc00, 0xfe00, - 0xff00, 0xff80, 0xffc0, 0xffe0, 0xfff0, 0xfff8, 0xfffc, - 0xfffe, 0xffff] + bitCount = [ + 0, + 0x8000, + 0xC000, + 0xE000, + 0xF000, + 0xF800, + 0xFC00, + 0xFE00, + 0xFF00, + 0xFF80, + 0xFFC0, + 0xFFE0, + 0xFFF0, + 0xFFF8, + 0xFFFC, + 0xFFFE, + 0xFFFF, + ] prefix = 0 - for word in mask.split(':'): + for word in mask.split(":"): if not word or int(word, 16) == 0: break prefix += bitCount.index(int(word, 16)) @@ -1052,11 +1123,12 @@ def mask_and_ipv4_to_bcast_addr(mask, ip): """Calculate the broadcast address from the subnet mask and ip addr. Supports ipv4 only.""" - ip_bin = int(''.join([bin(int(x) + 256)[3:] for x in ip.split('.')]), 2) + ip_bin = int("".join([bin(int(x) + 256)[3:] for x in ip.split(".")]), 2) mask_dec = ipv4_mask_to_net_prefix(mask) - bcast_bin = ip_bin | (2**(32 - mask_dec) - 1) - bcast_str = '.'.join([str(bcast_bin >> (i << 3) & 0xFF) - for i in range(4)[::-1]]) + bcast_bin = ip_bin | (2 ** (32 - mask_dec) - 1) + bcast_str = ".".join( + [str(bcast_bin >> (i << 3) & 0xFF) for i in range(4)[::-1]] + ) return bcast_str @@ -1066,8 +1138,8 @@ def parse_net_config_data(net_config, skip_broken=True) -> NetworkState: :param net_config: curtin network config dict """ state = None - version = net_config.get('version') - config = net_config.get('config') + version = net_config.get("version") + config = net_config.get("config") if version == 2: # v2 does not have explicit 'config' key so we # pass the whole net-config as-is |