feat(vmware): Convert imc network config to Networking config Version 2

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 <stanguturi@vmware.com>
+#         Pengpeng Sun <pengpeng.sun@broadcom.com>
 #
 # 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,161 +80,190 @@ 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()
         name = self.mac2Name.get(mac)
         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"""
         if configure:
             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")