From 923594700ec4ca9fdca383ce21b82819e21875dd Mon Sep 17 00:00:00 2001 From: PengpengSun Date: Thu, 19 Dec 2024 09:02:27 +0000 Subject: [PATCH 1/2] feat(vmware): Convert imc network configuration to Networking config Version 2hhhhh --- .../sources/helpers/vmware/imc/config_nic.py | 310 +++++++------ .../helpers/vmware/imc/guestcust_util.py | 28 +- .../sources/vmware/test_vmware_config_file.py | 421 ++++++++---------- 3 files changed, 356 insertions(+), 403 deletions(-) diff --git a/cloudinit/sources/helpers/vmware/imc/config_nic.py b/cloudinit/sources/helpers/vmware/imc/config_nic.py index b07214a228b..c99ea6027a0 100644 --- a/cloudinit/sources/helpers/vmware/imc/config_nic.py +++ b/cloudinit/sources/helpers/vmware/imc/config_nic.py @@ -1,38 +1,31 @@ # Copyright (C) 2015 Canonical Ltd. -# Copyright (C) 2016 VMware INC. +# Copyright (C) 2006-2024 Broadcom. All Rights Reserved. +# Broadcom Confidential. The term "Broadcom" refers to Broadcom Inc. +# and/or its subsidiaries. # # Author: Sankar Tanguturi +# Pengpeng Sun # # This file is part of cloud-init. See LICENSE file for license information. +import ipaddress import logging import os import re from cloudinit import net, subp, util -from cloudinit.net.network_state import ipv4_mask_to_net_prefix +from cloudinit.net.network_state import ( + ipv4_mask_to_net_prefix, + ipv6_mask_to_net_prefix, +) logger = logging.getLogger(__name__) -def gen_subnet(ip, netmask): - """ - Return the subnet for a given ip address and a netmask - @return (str): the subnet - @param ip: ip address - @param netmask: netmask - """ - ip_array = ip.split(".") - mask_array = netmask.split(".") - result = [] - for index in list(range(4)): - result.append(int(ip_array[index]) & int(mask_array[index])) - - return ".".join([str(x) for x in result]) - - class NicConfigurator: - def __init__(self, nics, use_system_devices=True): + def __init__( + self, nics, name_servers, dns_suffixes, use_system_devices=True + ): """ Initialize the Nic Configurator @param nics (list) an array of nics to configure @@ -41,9 +34,9 @@ def __init__(self, nics, use_system_devices=True): the specified nics. """ self.nics = nics + self.name_servers = name_servers + self.dns_suffixes = dns_suffixes self.mac2Name = {} - self.ipv4PrimaryGateway = None - self.ipv6PrimaryGateway = None if use_system_devices: self.find_devices() @@ -87,10 +80,10 @@ def find_devices(self): name = section.split(":", 1)[0] self.mac2Name[mac] = name - def gen_one_nic(self, nic): + def gen_one_nic_v2(self, nic): """ - Return the config list needed to configure a nic - @return (list): the subnets and routes list to configure the nic + Return the config dict needed to configure a nic + @return (dict): the config dict to configure the nic @param nic (NicBase): the nic to configure """ mac = nic.mac.lower() @@ -98,137 +91,166 @@ def gen_one_nic(self, nic): if not name: raise ValueError("No known device has MACADDR: %s" % nic.mac) - nics_cfg_list = [] - - cfg = {"type": "physical", "name": name, "mac_address": mac} - - subnet_list = [] - route_list = [] - - # Customize IPv4 - (subnets, routes) = self.gen_ipv4(name, nic) - subnet_list.extend(subnets) - route_list.extend(routes) - - # Customize IPv6 - (subnets, routes) = self.gen_ipv6(name, nic) - subnet_list.extend(subnets) - route_list.extend(routes) - - cfg.update({"subnets": subnet_list}) - - nics_cfg_list.append(cfg) - if route_list: - nics_cfg_list.extend(route_list) - - return nics_cfg_list - - def gen_ipv4(self, name, nic): - """ - Return the set of subnets and routes needed to configure the - IPv4 settings of a nic - @return (set): the set of subnet and routes to configure the gateways - @param name (str): subnet and route list for the nic - @param nic (NicBase): the nic to configure - """ - - subnet = {} - route_list = [] - + nic_config_dict = {} + # match + match = self.gen_match(mac) + if match: + nic_config_dict.update(match) + # set-name + set_name = self.gen_set_name(name) + if set_name: + nic_config_dict.update(set_name) + # wakeonlan + wakeonlan = self.gen_wakeonlan(nic) + if wakeonlan: + nic_config_dict.update(wakeonlan) + # dhcp4 + dhcp4 = self.gen_dhcp4(nic) + if dhcp4: + nic_config_dict.update(dhcp4) + # dhcp6 + dhcp6 = self.gen_dhcp6(nic) + if dhcp6: + nic_config_dict.update(dhcp6) + # addresses + addresses = self.gen_addresses(nic) + if addresses: + nic_config_dict.update(addresses) + # routes + routes = self.gen_routes(nic) + if routes: + nic_config_dict.update(routes) + # nameservers + nameservers = self.gen_nameservers() + if nameservers: + nic_config_dict.update(nameservers) + + return {name: nic_config_dict} + + def gen_match(self, mac): + return {"match": {"macaddress": mac}} + + def gen_set_name(self, name): + return {"set-name": name} + + def gen_wakeonlan(self, nic): if nic.onboot: - subnet.update({"control": "auto"}) + return {"wakeonlan": True} + else: + return {"wakeonlan": False} + def gen_dhcp4(self, nic): + dhcp4 = {} bootproto = nic.bootProto.lower() if nic.ipv4_mode.lower() == "disabled": bootproto = "manual" - if bootproto != "static": - subnet.update({"type": "dhcp"}) - return ([subnet], route_list) + dhcp4.update({"dhcp4": True}) + # dhcp4-overrides + if self.name_servers or self.dns_suffixes: + dhcp4.update({"dhcp4-overrides": {"use-dns": False}}) else: - subnet.update({"type": "static"}) - - # Static Ipv4 - addrs = nic.staticIpv4 - if not addrs: - return ([subnet], route_list) - - v4 = addrs[0] - if v4.ip: - subnet.update({"address": v4.ip}) - if v4.netmask: - subnet.update({"netmask": v4.netmask}) - - # Add the primary gateway - if nic.primary and v4.gateways: - self.ipv4PrimaryGateway = v4.gateways[0] - subnet.update({"gateway": self.ipv4PrimaryGateway}) - return ([subnet], route_list) - - # Add routes if there is no primary nic - if not self._primaryNic and v4.gateways: - subnet.update( - {"routes": self.gen_ipv4_route(nic, v4.gateways, v4.netmask)} - ) - - return ([subnet], route_list) - - def gen_ipv4_route(self, nic, gateways, netmask): - """ - Return the routes list needed to configure additional Ipv4 route - @return (list): the route list to configure the gateways - @param nic (NicBase): the nic to configure - @param gateways (str list): the list of gateways - """ - route_list = [] - - cidr = ipv4_mask_to_net_prefix(netmask) - - for gateway in gateways: - destination = "%s/%d" % (gen_subnet(gateway, netmask), cidr) - route_list.append( - { - "destination": destination, - "type": "route", - "gateway": gateway, - "metric": 10000, - } - ) - - return route_list + dhcp4.update({"dhcp4": False}) + if dhcp4: + return dhcp4 + else: + return None - def gen_ipv6(self, name, nic): - """ - Return the set of subnets and routes needed to configure the - gateways for a nic - @return (set): the set of subnets and routes to configure the gateways - @param name (str): name of the nic - @param nic (NicBase): the nic to configure - """ + def gen_dhcp6(self, nic): + if nic.staticIpv6: + return {"dhcp6": False} + else: + # TODO: nic shall explicitly tell it's DHCP6 + # TODO: set dhcp6-overrides + return None - if not nic.staticIpv6: - return ([], []) + def gen_addresses(self, nic): + address_list = [] + v4_cidr = 32 + v6_cidr = 128 - subnet_list = [] + # Static Ipv4 + v4_addrs = nic.staticIpv4 + if v4_addrs: + v4 = v4_addrs[0] + if v4.netmask: + v4_cidr = ipv4_mask_to_net_prefix(v4.netmask) + if v4.ip: + address_list.append(f"{v4.ip}/{v4_cidr}") # Static Ipv6 - addrs = nic.staticIpv6 - - for addr in addrs: - subnet = { - "type": "static6", - "address": addr.ip, - "netmask": addr.netmask, - } - subnet_list.append(subnet) - - # TODO: Add the primary gateway + v6_addrs = nic.staticIpv6 + if v6_addrs: + for v6 in v6_addrs: + v6_cidr = ipv6_mask_to_net_prefix(v6.netmask) + address_list.append(f"{v6.ip}/{v6_cidr}") + + if address_list: + return {"addresses": address_list} + else: + return None + def gen_routes(self, nic): route_list = [] - # TODO: Add routes if there is no primary nic - # if not self._primaryNic: - # route_list.extend(self._genIpv6Route(name, nic, addrs)) + v4_cidr = 32 + v6_cidr = 128 + + # Ipv4 routes + v4_addrs = nic.staticIpv4 + if v4_addrs: + v4 = v4_addrs[0] + # Add the ipv4 default route + if nic.primary and v4.gateways: + route_list.append({"to": "default", "via": v4.gateways[0]}) + # Add ipv4 static routes if there is no primary nic + if not self._primaryNic and v4.gateways: + if v4.netmask: + v4_cidr = ipv4_mask_to_net_prefix(v4.netmask) + for gateway in v4.gateways: + v4_subnet = ipaddress.IPv4Network( + f"{gateway}/{v4_cidr}", strict=False + ) + route_list.append({"to": f"{v4_subnet}", "via": gateway}) + # Ipv6 routes + v6_addrs = nic.staticIpv6 + if v6_addrs: + for v6 in v6_addrs: + v6_cidr = ipv6_mask_to_net_prefix(v6.netmask) + # Add the ipv6 default route + if nic.primary and v6.gateway: + route_list.append({"to": "default", "via": v6.gateway}) + # Add ipv6 static routes if there is no primary nic + if not self._primaryNic and v6.gateway: + v6_subnet = ipaddress.IPv6Network( + f"{v6.gateway}/{v6_cidr}", strict=False + ) + route_list.append( + {"to": f"{v6_subnet}", "via": v6.gateway} + ) - return (subnet_list, route_list) + if route_list: + return {"routes": route_list} + else: + return None + + def gen_nameservers(self): + nameservers_dict = {} + search_list = [] + addresses_list = [] + if self.dns_suffixes: + for dns_suffix in self.dns_suffixes: + search_list.append(dns_suffix) + if self.name_servers: + for name_server in self.name_servers: + addresses_list.append(name_server) + if search_list: + nameservers_dict.update({"search": search_list}) + if addresses_list: + nameservers_dict.update({"addresses": addresses_list}) + + if nameservers_dict: + return {"nameservers": nameservers_dict} + else: + return None def generate(self, configure=False, osfamily=None): """Return the config elements that are needed to configure the nics""" @@ -236,12 +258,12 @@ def generate(self, configure=False, osfamily=None): logger.info("Configuring the interfaces file") self.configure(osfamily) - nics_cfg_list = [] + ethernets_dict = {} for nic in self.nics: - nics_cfg_list.extend(self.gen_one_nic(nic)) + ethernets_dict.update(self.gen_one_nic_v2(nic)) - return nics_cfg_list + return ethernets_dict def clear_dhcp(self): logger.info("Clearing DHCP leases") diff --git a/cloudinit/sources/helpers/vmware/imc/guestcust_util.py b/cloudinit/sources/helpers/vmware/imc/guestcust_util.py index 1be10c9708b..7456db84986 100644 --- a/cloudinit/sources/helpers/vmware/imc/guestcust_util.py +++ b/cloudinit/sources/helpers/vmware/imc/guestcust_util.py @@ -1,8 +1,10 @@ # Copyright (C) 2016 Canonical Ltd. -# Copyright (C) 2016-2023 VMware Inc. +# Copyright (C) 2006-2024 Broadcom. All Rights Reserved. +# Broadcom Confidential. The term "Broadcom" refers to Broadcom Inc. +# and/or its subsidiaries. # # Author: Sankar Tanguturi -# Pengpeng Sun +# Pengpeng Sun # # This file is part of cloud-init. See LICENSE file for license information. @@ -325,23 +327,19 @@ def get_non_network_data_from_vmware_cust_cfg(cust_cfg): def get_network_data_from_vmware_cust_cfg( cust_cfg, use_system_devices=True, configure=False, osfamily=None ): - nicConfigurator = NicConfigurator(cust_cfg.nics, use_system_devices) - nics_cfg_list = nicConfigurator.generate(configure, osfamily) - - return get_v1_network_config( - nics_cfg_list, cust_cfg.name_servers, cust_cfg.dns_suffixes + nicConfigurator = NicConfigurator( + cust_cfg.nics, + cust_cfg.name_servers, + cust_cfg.dns_suffixes, + use_system_devices, ) + ethernets_dict = nicConfigurator.generate(configure, osfamily) + return gen_v2_network_config(ethernets_dict) -def get_v1_network_config(nics_cfg_list=None, nameservers=None, search=None): - config_list = nics_cfg_list - - if nameservers or search: - config_list.append( - {"type": "nameserver", "address": nameservers, "search": search} - ) - return {"version": 1, "config": config_list} +def gen_v2_network_config(ethernets_dict): + return {"version": 2, "ethernets": ethernets_dict} def connect_nics(cust_cfg_dir): diff --git a/tests/unittests/sources/vmware/test_vmware_config_file.py b/tests/unittests/sources/vmware/test_vmware_config_file.py index fd4bb481e46..dab61da6703 100644 --- a/tests/unittests/sources/vmware/test_vmware_config_file.py +++ b/tests/unittests/sources/vmware/test_vmware_config_file.py @@ -1,8 +1,10 @@ # Copyright (C) 2015 Canonical Ltd. -# Copyright (C) 2016-2022 VMware INC. +# Copyright (C) 2006-2024 Broadcom. All Rights Reserved. +# Broadcom Confidential. The term "Broadcom" refers to Broadcom Inc. +# and/or its subsidiaries. # # Author: Sankar Tanguturi -# Pengpeng Sun +# Pengpeng Sun # # This file is part of cloud-init. See LICENSE file for license information. @@ -17,10 +19,7 @@ from cloudinit.sources.helpers.vmware.imc.config_file import ( ConfigFile as WrappedConfigFile, ) -from cloudinit.sources.helpers.vmware.imc.config_nic import ( - NicConfigurator, - gen_subnet, -) +from cloudinit.sources.helpers.vmware.imc.config_nic import NicConfigurator from cloudinit.sources.helpers.vmware.imc.guestcust_util import ( get_network_data_from_vmware_cust_cfg, get_non_network_data_from_vmware_cust_cfg, @@ -164,36 +163,21 @@ def test_get_config_nameservers(self): network_config = get_network_data_from_vmware_cust_cfg(config, False) - self.assertEqual(1, network_config.get("version")) - - config_types = network_config.get("config") - name_servers = None - dns_suffixes = None + self.assertEqual(2, network_config.get("version")) - for type in config_types: - if type.get("type") == "nameserver": - name_servers = type.get("address") - dns_suffixes = type.get("search") - break + ethernets = network_config.get("ethernets") - self.assertEqual(["10.20.145.1", "10.20.145.2"], name_servers, "dns") - self.assertEqual( - ["eng.vmware.com", "proxy.vmware.com"], dns_suffixes, "suffixes" - ) - - def test_gen_subnet(self): - """Tests if gen_subnet properly calculates network subnet from - IPv4 address and netmask""" - ip_subnet_list = [ - ["10.20.87.253", "255.255.252.0", "10.20.84.0"], - ["10.20.92.105", "255.255.252.0", "10.20.92.0"], - ["192.168.0.10", "255.255.0.0", "192.168.0.0"], - ] - for entry in ip_subnet_list: + for _, config in ethernets.items(): + self.assertTrue(isinstance(config, dict)) + name_servers = config.get("nameservers").get("addresses") + dns_suffixes = config.get("nameservers").get("search") + self.assertEqual( + ["10.20.145.1", "10.20.145.2"], name_servers, "dns" + ) self.assertEqual( - entry[2], - gen_subnet(entry[0], entry[1]), - "Subnet for a specified ip and netmask", + ["eng.vmware.com", "proxy.vmware.com"], + dns_suffixes, + "suffixes", ) def test_get_config_dns_suffixes(self): @@ -206,162 +190,141 @@ def test_get_config_dns_suffixes(self): network_config = get_network_data_from_vmware_cust_cfg(config, False) - self.assertEqual(1, network_config.get("version")) + self.assertEqual(2, network_config.get("version")) - config_types = network_config.get("config") - name_servers = None - dns_suffixes = None + ethernets = network_config.get("ethernets") - for type in config_types: - if type.get("type") == "nameserver": - name_servers = type.get("address") - dns_suffixes = type.get("search") - break - - self.assertEqual([], name_servers, "dns") - self.assertEqual(["eng.vmware.com"], dns_suffixes, "suffixes") + for _, config in ethernets.items(): + self.assertTrue(isinstance(config, dict)) + name_servers = config.get("nameservers").get("addresses") + dns_suffixes = config.get("nameservers").get("search") + self.assertEqual(None, name_servers, "dns") + self.assertEqual(["eng.vmware.com"], dns_suffixes, "suffixes") def test_get_nics_list_dhcp(self): - """Tests if NicConfigurator properly calculates network subnets + """Tests if NicConfigurator properly calculates ethernets for a configuration with a list of DHCP NICs""" cf = ConfigFile("tests/data/vmware/cust-dhcp-2nic.cfg") config = Config(cf) - nicConfigurator = NicConfigurator(config.nics, False) - nics_cfg_list = nicConfigurator.generate() - - self.assertEqual(2, len(nics_cfg_list), "number of config elements") - - nic1 = {"name": "NIC1"} - nic2 = {"name": "NIC2"} - for cfg in nics_cfg_list: - if cfg.get("name") == nic1.get("name"): - nic1.update(cfg) - elif cfg.get("name") == nic2.get("name"): - nic2.update(cfg) - - self.assertEqual("physical", nic1.get("type"), "type of NIC1") - self.assertEqual("NIC1", nic1.get("name"), "name of NIC1") - self.assertEqual( - "00:50:56:a6:8c:08", nic1.get("mac_address"), "mac address of NIC1" - ) - subnets = nic1.get("subnets") - self.assertEqual(1, len(subnets), "number of subnets for NIC1") - subnet = subnets[0] - self.assertEqual("dhcp", subnet.get("type"), "DHCP type for NIC1") - self.assertEqual("auto", subnet.get("control"), "NIC1 Control type") - - self.assertEqual("physical", nic2.get("type"), "type of NIC2") - self.assertEqual("NIC2", nic2.get("name"), "name of NIC2") - self.assertEqual( - "00:50:56:a6:5a:de", nic2.get("mac_address"), "mac address of NIC2" + nicConfigurator = NicConfigurator( + config.nics, config.name_servers, config.dns_suffixes, False ) - subnets = nic2.get("subnets") - self.assertEqual(1, len(subnets), "number of subnets for NIC2") - subnet = subnets[0] - self.assertEqual("dhcp", subnet.get("type"), "DHCP type for NIC2") - self.assertEqual("auto", subnet.get("control"), "NIC2 Control type") + ethernets_dict = nicConfigurator.generate() + + self.assertTrue(isinstance(ethernets_dict, dict)) + self.assertEqual(2, len(ethernets_dict), "number of ethernets") + + for name, config in ethernets_dict.items(): + if name == "NIC1": + self.assertEqual( + "00:50:56:a6:8c:08", + config.get("match").get("macaddress"), + "mac address of NIC1", + ) + self.assertEqual( + True, config.get("wakeonlan"), "wakeonlan of NIC1" + ) + self.assertEqual( + True, config.get("dhcp4"), "DHCPv4 enablement of NIC1" + ) + self.assertEqual( + False, + config.get("dhcp4-overrides").get("use-dns"), + "use-dns enablement for dhcp4-overrides of NIC1", + ) + if name == "NIC2": + self.assertEqual( + "00:50:56:a6:5a:de", + config.get("match").get("macaddress"), + "mac address of NIC2", + ) + self.assertEqual( + True, config.get("wakeonlan"), "wakeonlan of NIC2" + ) + self.assertEqual( + True, config.get("dhcp4"), "DHCPv4 enablement of NIC2" + ) + self.assertEqual( + False, + config.get("dhcp4-overrides").get("use-dns"), + "use-dns enablement for dhcp4-overrides of NIC2", + ) def test_get_nics_list_static(self): - """Tests if NicConfigurator properly calculates network subnets + """Tests if NicConfigurator properly calculates ethernets for a configuration with 2 static NICs""" cf = ConfigFile("tests/data/vmware/cust-static-2nic.cfg") config = Config(cf) - nicConfigurator = NicConfigurator(config.nics, False) - nics_cfg_list = nicConfigurator.generate() - - self.assertEqual(2, len(nics_cfg_list), "number of elements") - - nic1 = {"name": "NIC1"} - nic2 = {"name": "NIC2"} - route_list = [] - for cfg in nics_cfg_list: - cfg_type = cfg.get("type") - if cfg_type == "physical": - if cfg.get("name") == nic1.get("name"): - nic1.update(cfg) - elif cfg.get("name") == nic2.get("name"): - nic2.update(cfg) - - self.assertEqual("physical", nic1.get("type"), "type of NIC1") - self.assertEqual("NIC1", nic1.get("name"), "name of NIC1") - self.assertEqual( - "00:50:56:a6:8c:08", nic1.get("mac_address"), "mac address of NIC1" - ) - - subnets = nic1.get("subnets") - self.assertEqual(2, len(subnets), "Number of subnets") - - static_subnet = [] - static6_subnet = [] - - for subnet in subnets: - subnet_type = subnet.get("type") - if subnet_type == "static": - static_subnet.append(subnet) - elif subnet_type == "static6": - static6_subnet.append(subnet) - else: - self.assertEqual(True, False, "Unknown type") - if "route" in subnet: - for route in subnet.get("routes"): - route_list.append(route) - - self.assertEqual(1, len(static_subnet), "Number of static subnet") - self.assertEqual(1, len(static6_subnet), "Number of static6 subnet") - - subnet = static_subnet[0] - self.assertEqual( - "10.20.87.154", - subnet.get("address"), - "IPv4 address of static subnet", - ) - self.assertEqual( - "255.255.252.0", subnet.get("netmask"), "NetMask of static subnet" - ) - self.assertEqual( - "auto", subnet.get("control"), "control for static subnet" - ) - - subnet = static6_subnet[0] - self.assertEqual( - "fc00:10:20:87::154", - subnet.get("address"), - "IPv6 address of static subnet", - ) - self.assertEqual( - "64", subnet.get("netmask"), "NetMask of static6 subnet" - ) - - route_set = set(["10.20.87.253", "10.20.87.105", "192.168.0.10"]) - for route in route_list: - self.assertEqual(10000, route.get("metric"), "metric of route") - gateway = route.get("gateway") - if gateway in route_set: - route_set.discard(gateway) - else: - self.assertEqual(True, False, "invalid gateway %s" % (gateway)) - - self.assertEqual("physical", nic2.get("type"), "type of NIC2") - self.assertEqual("NIC2", nic2.get("name"), "name of NIC2") - self.assertEqual( - "00:50:56:a6:ef:7d", nic2.get("mac_address"), "mac address of NIC2" - ) - - subnets = nic2.get("subnets") - self.assertEqual(1, len(subnets), "Number of subnets for NIC2") - - subnet = subnets[0] - self.assertEqual("static", subnet.get("type"), "Subnet type") - self.assertEqual( - "192.168.6.102", subnet.get("address"), "Subnet address" - ) - self.assertEqual( - "255.255.0.0", subnet.get("netmask"), "Subnet netmask" + nicConfigurator = NicConfigurator( + config.nics, config.name_servers, config.dns_suffixes, False ) + ethernets_dict = nicConfigurator.generate() + + self.assertTrue(isinstance(ethernets_dict, dict)) + self.assertEqual(2, len(ethernets_dict), "number of ethernets") + + for name, config in ethernets_dict.items(): + print(config) + if name == "NIC1": + self.assertEqual( + "00:50:56:a6:8c:08", + config.get("match").get("macaddress"), + "mac address of NIC1", + ) + self.assertEqual( + True, config.get("wakeonlan"), "wakeonlan of NIC1" + ) + self.assertEqual( + False, config.get("dhcp4"), "DHCPv4 enablement of NIC1" + ) + self.assertEqual( + False, config.get("dhcp6"), "DHCPv6 enablement of NIC1" + ) + self.assertEqual( + ["10.20.87.154/22", "fc00:10:20:87::154/64"], + config.get("addresses"), + "IP addresses of NIC1", + ) + self.assertEqual( + [ + {"to": "10.20.84.0/22", "via": "10.20.87.253"}, + {"to": "10.20.84.0/22", "via": "10.20.87.105"}, + { + "to": "fc00:10:20:87::/64", + "via": "fc00:10:20:87::253", + }, + ], + config.get("routes"), + "routes of NIC1", + ) + if name == "NIC2": + self.assertEqual( + "00:50:56:a6:ef:7d", + config.get("match").get("macaddress"), + "mac address of NIC2", + ) + self.assertEqual( + True, config.get("wakeonlan"), "wakeonlan of NIC2" + ) + self.assertEqual( + False, config.get("dhcp4"), "DHCPv4 enablement of NIC2" + ) + self.assertEqual( + ["192.168.6.102/16"], + config.get("addresses"), + "IP addresses of NIC2", + ) + self.assertEqual( + [ + {"to": "192.168.0.0/16", "via": "192.168.0.10"}, + ], + config.get("routes"), + "routes of NIC2", + ) def test_custom_script(self): cf = ConfigFile("tests/data/vmware/cust-dhcp-2nic.cfg") @@ -408,7 +371,12 @@ def _get_NicConfigurator(self, text): fp.write(text) fp.close() cfg = Config(ConfigFile(fp.name)) - return NicConfigurator(cfg.nics, use_system_devices=False) + return NicConfigurator( + cfg.nics, + cfg.name_servers, + cfg.dns_suffixes, + use_system_devices=False, + ) finally: if fp: os.unlink(fp.name) @@ -437,21 +405,15 @@ def test_non_primary_nic_without_gateway(self): ) nc = self._get_NicConfigurator(config) self.assertEqual( - [ - { - "type": "physical", - "name": "NIC1", - "mac_address": "00:50:56:a6:8c:08", - "subnets": [ - { - "control": "auto", - "type": "static", - "address": "10.20.87.154", - "netmask": "255.255.252.0", - } - ], + { + "NIC1": { + "match": {"macaddress": "00:50:56:a6:8c:08"}, + "wakeonlan": True, + "dhcp4": False, + "addresses": ["10.20.87.154/22"], + "set-name": "NIC1", } - ], + }, nc.generate(), ) @@ -480,29 +442,16 @@ def test_non_primary_nic_with_gateway(self): ) nc = self._get_NicConfigurator(config) self.assertEqual( - [ - { - "type": "physical", - "name": "NIC1", - "mac_address": "00:50:56:a6:8c:08", - "subnets": [ - { - "control": "auto", - "type": "static", - "address": "10.20.87.154", - "netmask": "255.255.252.0", - "routes": [ - { - "type": "route", - "destination": "10.20.84.0/22", - "gateway": "10.20.87.253", - "metric": 10000, - } - ], - } - ], + { + "NIC1": { + "match": {"macaddress": "00:50:56:a6:8c:08"}, + "wakeonlan": True, + "dhcp4": False, + "addresses": ["10.20.87.154/22"], + "routes": [{"to": "10.20.84.0/22", "via": "10.20.87.253"}], + "set-name": "NIC1", } - ], + }, nc.generate(), ) @@ -540,29 +489,19 @@ def test_cust_non_primary_nic_with_gateway_(self): ) nc = self._get_NicConfigurator(config) self.assertEqual( - [ - { - "type": "physical", - "name": "NIC1", - "mac_address": "00:50:56:ac:d1:8a", - "subnets": [ - { - "control": "auto", - "type": "static", - "address": "100.115.223.75", - "netmask": "255.255.255.0", - "routes": [ - { - "type": "route", - "destination": "100.115.223.0/24", - "gateway": "100.115.223.254", - "metric": 10000, - } - ], - } + { + "NIC1": { + "match": {"macaddress": "00:50:56:ac:d1:8a"}, + "wakeonlan": True, + "dhcp4": False, + "addresses": ["100.115.223.75/24"], + "routes": [ + {"to": "100.115.223.0/24", "via": "100.115.223.254"} ], + "set-name": "NIC1", + "nameservers": {"addresses": ["8.8.8.8"]}, } - ], + }, nc.generate(), ) @@ -592,22 +531,16 @@ def test_a_primary_nic_with_gateway(self): ) nc = self._get_NicConfigurator(config) self.assertEqual( - [ - { - "type": "physical", - "name": "NIC1", - "mac_address": "00:50:56:a6:8c:08", - "subnets": [ - { - "control": "auto", - "type": "static", - "address": "10.20.87.154", - "netmask": "255.255.252.0", - "gateway": "10.20.87.253", - } - ], + { + "NIC1": { + "match": {"macaddress": "00:50:56:a6:8c:08"}, + "wakeonlan": True, + "dhcp4": False, + "addresses": ["10.20.87.154/22"], + "routes": [{"to": "default", "via": "10.20.87.253"}], + "set-name": "NIC1", } - ], + }, nc.generate(), ) From 52d7507114e9afa7ef47ed571ebcec17013e5fe1 Mon Sep 17 00:00:00 2001 From: PengpengSun Date: Thu, 6 Feb 2025 07:59:26 +0000 Subject: [PATCH 2/2] Address comments from TheRealFalcon and sshedi --- .../sources/helpers/vmware/imc/config_nic.py | 77 ++++++------------- .../sources/vmware/test_vmware_config_file.py | 35 +++++++-- 2 files changed, 55 insertions(+), 57 deletions(-) diff --git a/cloudinit/sources/helpers/vmware/imc/config_nic.py b/cloudinit/sources/helpers/vmware/imc/config_nic.py index c99ea6027a0..78526fdf7c7 100644 --- a/cloudinit/sources/helpers/vmware/imc/config_nic.py +++ b/cloudinit/sources/helpers/vmware/imc/config_nic.py @@ -92,38 +92,19 @@ def gen_one_nic_v2(self, nic): raise ValueError("No known device has MACADDR: %s" % nic.mac) nic_config_dict = {} - # match - match = self.gen_match(mac) - if match: - nic_config_dict.update(match) - # set-name - set_name = self.gen_set_name(name) - if set_name: - nic_config_dict.update(set_name) - # wakeonlan - wakeonlan = self.gen_wakeonlan(nic) - if wakeonlan: - nic_config_dict.update(wakeonlan) - # dhcp4 - dhcp4 = self.gen_dhcp4(nic) - if dhcp4: - nic_config_dict.update(dhcp4) - # dhcp6 - dhcp6 = self.gen_dhcp6(nic) - if dhcp6: - nic_config_dict.update(dhcp6) - # addresses - addresses = self.gen_addresses(nic) - if addresses: - nic_config_dict.update(addresses) - # routes - routes = self.gen_routes(nic) - if routes: - nic_config_dict.update(routes) - # nameservers - nameservers = self.gen_nameservers() - if nameservers: - nic_config_dict.update(nameservers) + generators = [ + self.gen_match(mac), + self.gen_set_name(name), + self.gen_wakeonlan(nic), + self.gen_dhcp4(nic), + self.gen_dhcp6(nic), + self.gen_addresses(nic), + self.gen_routes(nic), + self.gen_nameservers(), + ] + for value in generators: + if value: + nic_config_dict.update(value) return {name: nic_config_dict} @@ -134,10 +115,7 @@ def gen_set_name(self, name): return {"set-name": name} def gen_wakeonlan(self, nic): - if nic.onboot: - return {"wakeonlan": True} - else: - return {"wakeonlan": False} + return {"wakeonlan": nic.onboot} def gen_dhcp4(self, nic): dhcp4 = {} @@ -151,23 +129,19 @@ def gen_dhcp4(self, nic): dhcp4.update({"dhcp4-overrides": {"use-dns": False}}) else: dhcp4.update({"dhcp4": False}) - if dhcp4: - return dhcp4 - else: - return None + return dhcp4 def gen_dhcp6(self, nic): + dhcp6 = {} if nic.staticIpv6: - return {"dhcp6": False} - else: - # TODO: nic shall explicitly tell it's DHCP6 - # TODO: set dhcp6-overrides - return None + dhcp6.update({"dhcp6": False}) + # TODO: nic shall explicitly tell it's DHCP6 + # TODO: set dhcp6-overrides + return dhcp6 def gen_addresses(self, nic): address_list = [] v4_cidr = 32 - v6_cidr = 128 # Static Ipv4 v4_addrs = nic.staticIpv4 @@ -187,12 +161,11 @@ def gen_addresses(self, nic): if address_list: return {"addresses": address_list} else: - return None + return {} def gen_routes(self, nic): route_list = [] v4_cidr = 32 - v6_cidr = 128 # Ipv4 routes v4_addrs = nic.staticIpv4 @@ -200,7 +173,7 @@ def gen_routes(self, nic): v4 = v4_addrs[0] # Add the ipv4 default route if nic.primary and v4.gateways: - route_list.append({"to": "default", "via": v4.gateways[0]}) + route_list.append({"to": "0.0.0.0/0", "via": v4.gateways[0]}) # Add ipv4 static routes if there is no primary nic if not self._primaryNic and v4.gateways: if v4.netmask: @@ -217,7 +190,7 @@ def gen_routes(self, nic): v6_cidr = ipv6_mask_to_net_prefix(v6.netmask) # Add the ipv6 default route if nic.primary and v6.gateway: - route_list.append({"to": "default", "via": v6.gateway}) + route_list.append({"to": "::/0", "via": v6.gateway}) # Add ipv6 static routes if there is no primary nic if not self._primaryNic and v6.gateway: v6_subnet = ipaddress.IPv6Network( @@ -230,7 +203,7 @@ def gen_routes(self, nic): if route_list: return {"routes": route_list} else: - return None + return {} def gen_nameservers(self): nameservers_dict = {} @@ -250,7 +223,7 @@ def gen_nameservers(self): if nameservers_dict: return {"nameservers": nameservers_dict} else: - return None + return {} def generate(self, configure=False, osfamily=None): """Return the config elements that are needed to configure the nics""" diff --git a/tests/unittests/sources/vmware/test_vmware_config_file.py b/tests/unittests/sources/vmware/test_vmware_config_file.py index dab61da6703..837efa218d3 100644 --- a/tests/unittests/sources/vmware/test_vmware_config_file.py +++ b/tests/unittests/sources/vmware/test_vmware_config_file.py @@ -381,8 +381,8 @@ def _get_NicConfigurator(self, text): if fp: os.unlink(fp.name) - def test_non_primary_nic_without_gateway(self): - """A non primary nic set is not required to have a gateway.""" + def test_static_nic_without_ipv4_netmask(self): + """netmask is optional for static ipv4 configuration.""" config = textwrap.dedent( """\ [NETWORK] @@ -400,7 +400,6 @@ def test_non_primary_nic_without_gateway(self): IPv4_MODE = BACKWARDS_COMPATIBLE BOOTPROTO = static IPADDR = 10.20.87.154 - NETMASK = 255.255.252.0 """ ) nc = self._get_NicConfigurator(config) @@ -410,13 +409,39 @@ def test_non_primary_nic_without_gateway(self): "match": {"macaddress": "00:50:56:a6:8c:08"}, "wakeonlan": True, "dhcp4": False, - "addresses": ["10.20.87.154/22"], + "addresses": ["10.20.87.154/32"], "set-name": "NIC1", } }, nc.generate(), ) + def test_static_nic_without_ipv6_netmask(self): + """netmask is mandatory for static ipv6 configuration.""" + config = textwrap.dedent( + """\ + [NETWORK] + NETWORKING = yes + BOOTPROTO = dhcp + HOSTNAME = myhost1 + DOMAINNAME = eng.vmware.com + + [NIC-CONFIG] + NICS = NIC1 + + [NIC1] + MACADDR = 00:50:56:a6:8c:08 + ONBOOT = yes + IPv4_MODE = BACKWARDS_COMPATIBLE + BOOTPROTO = static + IPADDR = 10.20.87.154 + IPv6ADDR|1 = fc00:10:20:87::154 + """ + ) + nc = self._get_NicConfigurator(config) + with self.assertRaises(ValueError): + nc.generate() + def test_non_primary_nic_with_gateway(self): """A non primary nic set can have a gateway.""" config = textwrap.dedent( @@ -537,7 +562,7 @@ def test_a_primary_nic_with_gateway(self): "wakeonlan": True, "dhcp4": False, "addresses": ["10.20.87.154/22"], - "routes": [{"to": "default", "via": "10.20.87.253"}], + "routes": [{"to": "0.0.0.0/0", "via": "10.20.87.253"}], "set-name": "NIC1", } },