Adds Netpreserver module

This commit adds NetPreserver module inside OSMorphing in order to
preserve static IPs on Linux.

coriolis/osmorphing/base.py

@@ -12,6 +12,7 @@ from six import with_metaclass
 import yaml
 
 from coriolis import exception
+from coriolis.osmorphing.netpreserver import factory
 from coriolis import utils
 
 GRUB2_SERIAL = "serial --word=8 --stop=1 --speed=%d --parity=%s --unit=0"
@@ -572,3 +573,41 @@ class BaseLinuxOSMorphingTools(BaseOSMorphingTools):
         self._set_grub2_cmdline(config_obj, options)
         self._apply_grub2_config(
             config_obj, execute_update_grub)
+
+    def _add_net_udev_rules(self, net_ifaces_info):
+        udev_file = "etc/udev/rules.d/70-persistent-net.rules"
+        if not self._test_path(udev_file):
+            if net_ifaces_info:
+                content = utils.get_udev_net_rules(net_ifaces_info)
+                self._write_file_sudo(udev_file, content)
+
+    def _setup_network_preservation(self, nics_info) -> None:
+        net_ifaces_info = dict()
+
+        net_preserver_class = factory.get_net_preserver(self)
+        LOG.info("Using network preserver class: %s", net_preserver_class)
+        netpreserver = net_preserver_class(self)
+        netpreserver.parse_network()
+        LOG.info("Parsed network configuration: %s",
+                 netpreserver.interface_info)
+
+        for nic in nics_info:
+            mac_address = nic.get('mac_address')
+            ip_addresses = nic.get('ip_addresses')
+            for iface, info in netpreserver.interface_info.items():
+                if mac_address:
+                    if info["mac_address"] == mac_address:
+                        net_ifaces_info[iface] = mac_address
+                    elif ip_addresses:
+                        for ip in ip_addresses:
+                            if ip in info["ip_addresses"] and mac_address:
+                                net_ifaces_info[iface] = mac_address
+                else:
+                    LOG.warning(
+                        "Could not find MAC address or IP addresses for "
+                        "interface '%s' in network configuration %s",
+                        iface, nic)
+
+        self._add_net_udev_rules(net_ifaces_info.items())
+
+        return

coriolis/osmorphing/debian.py

@@ -165,8 +165,8 @@ class BaseDebianMorphingTools(base.BaseLinuxOSMorphingTools):
 
     def set_net_config(self, nics_info, dhcp):
         if not dhcp:
-            if self._test_path(self.netplan_base):
-                self._preserve_static_netplan_configuration(nics_info)
+            LOG.info("Setting static IP configuration")
+            self._setup_network_preservation(nics_info)
             return
 
         self.disable_predictable_nic_names()

coriolis/osmorphing/netpreserver/base.py

@@ -0,0 +1,15 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+
+class BaseNetPreserver(object):
+
+    def __init__(self, osmorphing_tool):
+        self.osmorphing_tool = osmorphing_tool
+        self.interface_info = dict()
+
+    def check_net_preserver(self):
+        pass
+
+    def parse_network(self, nics_info):
+        pass

coriolis/osmorphing/netpreserver/factory.py

@@ -0,0 +1,20 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis.osmorphing.netpreserver import ifcfg
+from coriolis.osmorphing.netpreserver import interfaces
+from coriolis.osmorphing.netpreserver import netplan
+from coriolis.osmorphing.netpreserver import nmconnection
+
+NET_PRESERVERS = [netplan.NetplanNetPreserver,
+                  nmconnection.NmconnectionNetPreserver,
+                  ifcfg.IfcfgNetPreserver,
+                  interfaces.InterfacesNetPreserver]
+
+
+def get_net_preserver(osmorphing_tool):
+    for net_preserver_class in NET_PRESERVERS:
+        net_preserver = net_preserver_class(osmorphing_tool)
+        if net_preserver.check_net_preserver():
+            return net_preserver_class
+    return None

coriolis/osmorphing/netpreserver/ifcfg.py

@@ -0,0 +1,57 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import os
+import re
+
+from oslo_log import log as logging
+
+from coriolis.osmorphing.netpreserver import base
+
+LOG = logging.getLogger(__name__)
+
+
+class IfcfgNetPreserver(base.BaseNetPreserver):
+
+    def __init__(self, osmorphing_tool):
+        super(IfcfgNetPreserver, self).__init__(osmorphing_tool)
+        self.network_scripts_path = "etc/sysconfig/network-scripts"
+
+    def check_net_preserver(self):
+        if self.osmorphing_tool._test_path(self.network_scripts_path):
+            ifcfg_files = self._get_net_config_files(self.network_scripts_path)
+            if ifcfg_files:
+                ifcfgs_ethernet = self._get_ifcfgs_by_type(
+                    "Ethernet", self.network_scripts_path)
+                if ifcfgs_ethernet:
+                    return True
+        return False
+
+    def parse_network(self):
+        ifcfgs_ethernet = self._get_ifcfgs_by_type(
+            "Ethernet", self.network_scripts_path)
+        if ifcfgs_ethernet:
+            for ifcfg_file, ifcfg in ifcfgs_ethernet:
+                name = ifcfg.get("DEVICE")
+                if not name:
+                    # Get the name from the config file
+                    name = re.match("^.*/ifcfg-(.*)", ifcfg_file).groups()[0]
+                mac_address = ifcfg.get("HWADDR")
+                ip_address = ifcfg.get("IPADDR")
+                self.interface_info[name] = {
+                    "mac_address": mac_address,
+                    "ip_addresses": [ip_address] if ip_address else []
+                }
+
+    def _get_net_config_files(self, network_scripts_path):
+        dir_content = self.osmorphing_tool._list_dir(network_scripts_path)
+        return [os.path.join(network_scripts_path, f) for f in
+                dir_content if re.match("^ifcfg-(.*)", f)]
+
+    def _get_ifcfgs_by_type(self, ifcfg_type, network_scripts_path):
+        ifcfgs = []
+        for ifcfg_file in self._get_net_config_files(network_scripts_path):
+            ifcfg = self.osmorphing_tool._read_config_file(ifcfg_file)
+            if ifcfg.get("TYPE") == ifcfg_type:
+                ifcfgs.append((ifcfg_file, ifcfg))
+        return ifcfgs

coriolis/osmorphing/netpreserver/interfaces.py

@@ -0,0 +1,74 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import os
+
+from oslo_log import log as logging
+
+from coriolis.osmorphing.netpreserver import base
+
+LOG = logging.getLogger(__name__)
+
+
+class InterfacesNetPreserver(base.BaseNetPreserver):
+
+    def __init__(self, osmorphing_tool):
+            super(InterfacesNetPreserver, self).__init__(osmorphing_tool)
+            self.ifaces_file = "etc/network/interfaces"
+
+    def check_net_preserver(self):
+        if self.osmorphing_tool._test_path(self.ifaces_file):
+            if self.osmorphing_tool._list_dir(
+                    os.path.dirname(self.ifaces_file)):
+                return True
+        return False
+
+    def parse_network(self):
+        paths = [self.ifaces_file]
+
+        def _parse_iface_file(interface_file):
+            nonlocal paths
+            curr_ip_address = None
+            interfaces_contents = self.osmorphing_tool._read_file_sudo(
+                interface_file).decode()
+            LOG.debug(
+                "Fetched %s contents: %s", interface_file, interfaces_contents)
+            for line in interfaces_contents.splitlines():
+                if line.strip().startswith("iface"):
+                    words = line.split()
+                    if len(words) > 1:
+                        curr_iface = words[1]
+                        self.interface_info[curr_iface] = {
+                            "mac_address": "",
+                            "ip_addresses": []
+                        }
+                elif line.strip().startswith("hwaddress ether"):
+                    words = line.split()
+                    if len(words) > 2:
+                        if not curr_iface:
+                            LOG.warn("Found MAC address %s does not belong to "
+                                     "any interface stanza. Skipping.",
+                                     words[2])
+                            continue
+
+                        self.interface_info[curr_iface][
+                            "mac_address"] = words[2]
+                elif line.strip().startswith("address"):
+                    words = line.split()
+                    if len(words) > 1:
+                        curr_ip_address = words[1].split('/')[0]
+                        if curr_iface:
+                            self.interface_info[curr_iface][
+                                "ip_addresses"].append(curr_ip_address)
+
+                elif line.strip().startswith("source"):
+                    words = line.split()
+                    if len(words) > 1:
+                        source_path = words[1]
+                        if self.osmorphing_tool._test_path(source_path):
+                            paths += self.osmorphing_tool._exec_cmd_chroot(
+                                'ls -1 %s' % source_path).splitlines()
+
+        while paths:
+            _parse_iface_file(paths[0])
+            paths.pop(0)

coriolis/osmorphing/netpreserver/netplan.py

@@ -0,0 +1,73 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import yaml
+
+from oslo_log import log as logging
+
+from coriolis.osmorphing.netpreserver import base
+from coriolis import utils
+
+LOG = logging.getLogger(__name__)
+
+
+class NetplanNetPreserver(base.BaseNetPreserver):
+
+    def __init__(self, osmorphing_tool):
+        super(NetplanNetPreserver, self).__init__(osmorphing_tool)
+        self.netplan_base = "etc/netplan"
+        self.netplan_files = []
+
+    def check_net_preserver(self):
+        if self.osmorphing_tool._test_path(self.netplan_base):
+            return any(f.endswith((".yaml", ".yml"))
+                       for f in self.osmorphing_tool._list_dir(
+                           self.netplan_base))
+        return False
+
+    def parse_network(self):
+        self.netplan_files = [f for f in self.osmorphing_tool._list_dir(
+                              self.netplan_base)
+                              if f.endswith(".yaml") or f.endswith(".yml")]
+        for cfg in self.netplan_files:
+            cfg_path = "%s/%s" % (self.netplan_base, cfg)
+            try:
+                contents = yaml.safe_load(self.osmorphing_tool._read_file_sudo(
+                                          cfg_path).decode())
+                LOG.info("Parsing netplan configuration '%s'", cfg_path)
+                ifaces = contents.get('network', {}).get('ethernets', {})
+                for iface, net_cfg in ifaces.items():
+                    self.interface_info[iface] = {"mac_address": "",
+                                                  "ip_addresses": []}
+                    mac_address = net_cfg.get('match', {}).get('macaddress')
+                    if mac_address:
+                        self.interface_info[iface]["mac_address"] = mac_address
+                    else:
+                        LOG.warn(
+                            "Could not find MAC address or IP addresses for "
+                            "interface '%s' in netplan configuration '%s'",
+                            iface, cfg_path)
+
+                    for ip in net_cfg.get('addresses', []):
+                        if isinstance(ip, dict):
+                            ip_keys = list(ip.keys())
+                            if not ip_keys:
+                                LOG.warning(
+                                    "Found empty IP address object entry. "
+                                    "Skipping")
+                                continue
+                            ip_obj_addrs = [
+                                addr.split('/')[0] for addr in ip_keys]
+                            self.interface_info[iface]["ip_addresses"].extend(
+                                ip_obj_addrs)
+                        else:
+                            self.interface_info[iface]["ip_addresses"].append(
+                                ip.split('/')[0])
+                if not self.interface_info[iface]["ip_addresses"]:
+                    LOG.warning(
+                        "No IP addresses found for interface '%s' in netplan"
+                        "config '%s'", iface, cfg_path)
+            except yaml.YAMLError:
+                LOG.warn(
+                    "Could not parse netplan configuration '%s'. Invalid YAML "
+                    "file: %s", cfg_path, utils.get_exception_details())

coriolis/osmorphing/netpreserver/nmconnection.py

@@ -0,0 +1,71 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import os
+import re
+
+from oslo_log import log as logging
+
+from coriolis.osmorphing.netpreserver import base
+
+LOG = logging.getLogger(__name__)
+
+
+class NmconnectionNetPreserver(base.BaseNetPreserver):
+
+    def __init__(self, osmorphing_tool):
+        super(NmconnectionNetPreserver, self).__init__(osmorphing_tool)
+        self.nmconnection_file = "etc/NetworkManager/system-connections"
+
+    def check_net_preserver(self):
+        if self.osmorphing_tool._test_path(self.nmconnection_file):
+            nmconnection_files = self._get_nmconnection_files(
+                self.nmconnection_file)
+            if nmconnection_files:
+                nmconnection_ethernet = self._get_keyfiles_by_type(
+                    "ethernet", self.nmconnection_file)
+                if nmconnection_ethernet:
+                    return True
+        return False
+
+    def parse_network(self):
+        nmconnection_ethernet = self._get_keyfiles_by_type(
+            "ethernet", self.nmconnection_file)
+        if nmconnection_ethernet:
+            for nmconn_file, nmconn in nmconnection_ethernet:
+                name = nmconn.get("connection", {}).get("id")
+                if not name:
+                    name = re.match(r"^.*/(.*)\.nmconnection$",
+                                    nmconn_file).groups()[0]
+                mac_address = nmconn.get("mac-address")
+                self.interface_info[name] = {
+                    "mac_address": mac_address,
+                    "ip_addresses": []
+                }
+                for key, value in nmconn.items():
+                    if key.lower().startswith("address") and value:
+                        for addr in value.split(','):
+                            addr = addr.strip()
+                            ip = addr.strip().split('/')[0]
+                            self.interface_info[name][
+                                "ip_addresses"].append(ip)
+                if not mac_address and not self.interface_info[
+                    name]["ip_addresses"]:
+                    LOG.warning(
+                        "Could not find MAC address or IP addresses for "
+                        "interface '%s' in nmconnection configuration "
+                        "'%s'", name, nmconn_file)
+
+    def _get_nmconnection_files(self, network_scripts_path):
+        dir_content = self.osmorphing_tool._list_dir(network_scripts_path)
+        return [os.path.join(network_scripts_path, f)
+                for f in dir_content if re.match(
+                    r"^(.*\.nmconnection)$", f)]
+
+    def _get_keyfiles_by_type(self, nmconnection_type, network_scripts_path):
+        keyfiles = []
+        for file in self._get_nmconnection_files(network_scripts_path):
+            keyfile = self.osmorphing_tool._read_config_file(file)
+            if keyfile.get("type") == nmconnection_type:
+                keyfiles.append((file, keyfile))
+        return keyfiles

coriolis/osmorphing/redhat.py

@@ -81,35 +81,6 @@ class BaseRedHatMorphingTools(base.BaseLinuxOSMorphingTools):
             "could not determine grub location."
             " boot partition not mounted?")
 
-    def _get_net_ifaces_info(self, ifcfgs_ethernet, mac_addresses):
-        net_ifaces_info = []
-
-        for ifcfg_file, ifcfg in ifcfgs_ethernet:
-            mac_address = ifcfg.get("HWADDR")
-            if not mac_address:
-                if len(ifcfgs_ethernet) == 1 and len(mac_addresses) == 1:
-                    mac_address = mac_addresses[0]
-                    LOG.info("HWADDR not defined in: %s, using migration "
-                             "configuration mac_address: %s",
-                             ifcfg_file, mac_address)
-            if not mac_address:
-                self._event_manager.warn(
-                    "HWADDR not defined, skipping: %s" % ifcfg_file)
-                continue
-            name = ifcfg.get("NAME")
-            if not name:
-                # Get the name from the config file
-                name = re.match("^.*/ifcfg-(.*)", ifcfg_file).groups()[0]
-            net_ifaces_info.append((name, mac_address))
-        return net_ifaces_info
-
-    def _add_net_udev_rules(self, net_ifaces_info):
-        udev_file = "etc/udev/rules.d/70-persistent-net.rules"
-        if not self._test_path(udev_file):
-            if net_ifaces_info:
-                content = utils.get_udev_net_rules(net_ifaces_info)
-                self._write_file_sudo(udev_file, content)
-
     def _has_systemd(self):
         try:
             self._exec_cmd_chroot("rpm -q systemd")
@@ -141,14 +112,6 @@ class BaseRedHatMorphingTools(base.BaseLinuxOSMorphingTools):
             del network_cfg["GATEWAY"]
             self._write_config_file(network_cfg_file, network_cfg)
 
-    def _get_ifcfgs_by_type(self, ifcfg_type):
-        ifcfgs = []
-        for ifcfg_file in self._get_net_config_files():
-            ifcfg = self._read_config_file(ifcfg_file)
-            if ifcfg.get("TYPE") == ifcfg_type:
-                ifcfgs.append((ifcfg_file, ifcfg))
-        return ifcfgs
-
     def _write_nic_configs(self, nics_info):
         for idx, _ in enumerate(nics_info):
             dev_name = "eth%d" % idx
@@ -203,11 +166,8 @@ class BaseRedHatMorphingTools(base.BaseLinuxOSMorphingTools):
             self._write_nic_configs(nics_info)
             return
 
-        ifcfgs_ethernet = self._get_ifcfgs_by_type("Ethernet")
-        mac_addresses = [ni.get("mac_address") for ni in nics_info]
-        net_ifaces_info = self._get_net_ifaces_info(ifcfgs_ethernet,
-                                                    mac_addresses)
-        self._add_net_udev_rules(net_ifaces_info)
+        LOG.info("Setting static IP configuration")
+        self._setup_network_preservation(nics_info)
 
     def get_installed_packages(self):
         cmd = 'rpm -qa --qf "%{NAME}\\n"'
@@ -329,8 +289,3 @@ class BaseRedHatMorphingTools(base.BaseLinuxOSMorphingTools):
     def _get_config_file_content(self, config):
         return "%s\n" % "\n".join(
             ['%s="%s"' % (k, v) for k, v in config.items()])
-
-    def _get_net_config_files(self):
-        dir_content = self._list_dir(self._NETWORK_SCRIPTS_PATH)
-        return [os.path.join(self._NETWORK_SCRIPTS_PATH, f) for f in
-                dir_content if re.match("^ifcfg-(.*)", f)]