cable-guy: testing dynamic ip

cable-guy: troubleshooting dynamic ip

good

goow?

core/services/cable_guy/api/manager.py

@@ -100,11 +100,6 @@ def set_configuration(self, interface: NetworkInterface) -> None:
         self.flush_interface(interface.name)
         self.remove_dhcp_server_from_interface(interface.name)
 
-        # Even if it happened to receive more than one dynamic IP, only one trigger is necessary
-        if any(address.mode == AddressMode.Client for address in interface.addresses):
-            logger.info(f"Triggering dynamic IP acquisition for interface '{interface.name}'.")
-            self.trigger_dynamic_ip_acquisition(interface.name)
-
         logger.info(f"Configuring addresses for interface '{interface.name}': {interface.addresses}.")
         for address in interface.addresses:
             if address.mode == AddressMode.Unmanaged:
@@ -113,6 +108,10 @@ def set_configuration(self, interface: NetworkInterface) -> None:
             elif address.mode == AddressMode.Server:
                 logger.info(f"Adding DHCP server with gateway '{address.ip}' to interface '{interface.name}'.")
                 self.add_dhcp_server_to_interface(interface.name, address.ip)
+        # Even if it happened to receive more than one dynamic IP, only one trigger is necessary
+        if any(address.mode == AddressMode.Client for address in interface.addresses):
+            logger.info(f"Triggering dynamic IP acquisition for interface '{interface.name}'.")
+            self.trigger_dynamic_ip_acquisition(interface.name)
 
     def _get_wifi_interfaces(self) -> List[str]:
         """Get wifi interface list
@@ -246,10 +245,16 @@ def trigger_dynamic_ip_acquisition(self, interface_name: str) -> None:
         Args:
             interface_name (str): Interface name
         """
-        logger.info(f"Restarting interface {interface_name} to trigger dynamic IP acquisition.")
-        self.enable_interface(interface_name, False)
-        time.sleep(1)
-        self.enable_interface(interface_name, True)
+        try:
+            self.network_handler.trigger_dynamic_ip_acquisition(interface_name)
+            return
+        except NotImplementedError as error:
+            logger.info(f"Handler does not support triggering dynamic IP acquisition. {error}")
+            logger.info(f"Restarting interface {interface_name} to trigger dynamic IP acquisition.")
+            self.enable_interface(interface_name, enable=False)
+            self.enable_interface(interface_name, enable=True)
+        except Exception as error:
+            logger.error(f"Failed to trigger dynamic IP acquisition for interface {interface_name}. {error}")
 
     def add_static_ip(self, interface_name: str, ip: str) -> None:
         """Set ip address for a specific interface
@@ -513,6 +518,7 @@ def add_dhcp_server_to_interface(self, interface_name: str, ipv4_gateway: str) -
         if self._is_ip_on_interface(interface_name, ipv4_gateway):
             self.remove_ip(interface_name, ipv4_gateway)
         self.add_static_ip(interface_name, ipv4_gateway)
+        time.sleep(1)
         logger.info(f"Adding DHCP server with gateway '{ipv4_gateway}' to interface '{interface_name}'.")
         self._dhcp_servers.append(DHCPServerManager(interface_name, ipv4_gateway))
 

core/services/cable_guy/networksetup.py

@@ -1,9 +1,10 @@
 import os
-import time
+import re
 from typing import List
 
 import sdbus
 from loguru import logger
+from pyroute2 import IPRoute
 from sdbus_block.networkmanager import (
     NetworkConnectionSettings,
     NetworkDeviceGeneric,
@@ -41,8 +42,15 @@ def add_static_ip(self, interface_name: str, ip: str) -> None:
     def remove_static_ip(self, interface_name: str, ip: str) -> None:
         pass
 
+    def trigger_dynamic_ip_acquisition(self, interface_name: str) -> None:
+        raise NotImplementedError("This Handler does not support setting interface priority")
+
 
 class NetworkManagerHandler(AbstractNetworkHandler):
+    def __init__(self) -> None:
+        super().__init__()
+        self.ipr = IPRoute()
+
     def detect(self) -> bool:
         try:
             all_devices = {path: NetworkDeviceGeneric(path) for path in network_manager.devices}
@@ -69,23 +77,29 @@ def remove_static_ip(self, interface_name: str, ip: str) -> None:
     def add_static_ip(self, interface_name: str, ip: str) -> None:
         networkmanager_settings = NetworkManagerSettings()
 
-        time.sleep(1)
         for connection_path in networkmanager_settings.connections:
             try:
                 settings = NetworkConnectionSettings(connection_path)
                 data = settings.get_profile()
                 if data.connection.interface_name != interface_name:
                     continue
-                if any(ip in addressData.address for addressData in data.ipv4.address_data):
-                    logger.info(f"IP {ip} already exists for {interface_name}")
-                    continue
+
+                for address_data in data.ipv4.address_data:
+                    if ip == address_data.address:
+                        logger.info(f"IP {ip} already exists for {interface_name}")
+                        return
+
                 new_ip = AddressData(address=ip, prefix=24)
+                data.ipv4.address_data.append(new_ip)
+                settings.update_profile(data)
+
                 properties = settings.get_settings()
-                properties["ipv4"]["method"] = ("s", "shared")
+                # Keep method as manual to allow both DHCP client and server
+                properties["ipv4"]["method"] = ("s", "manual")
+                # Preserve DHCP client if it was enabled
+                properties["ipv4"]["may-fail"] = ("b", True)
                 settings.update(properties)
                 settings.save()
-                data.ipv4.address_data.append(new_ip)
-                settings.update_profile(data)
                 network_manager.activate_connection(connection_path)
                 return
             except Exception as e:
@@ -99,9 +113,22 @@ def enable_dhcp_client(self, interface_name: str) -> None:
                 properties = settings.get_settings()
                 if properties["connection"]["interface-name"][1] != interface_name:
                     continue
+
+                # Get current profile to preserve address data
+                data = settings.get_profile()
+                existing_ips = data.ipv4.address_data
+
+                # Update to shared mode while preserving existing IPs
                 properties["ipv4"]["method"] = ("s", "shared")
                 properties["ipv4"]["may-fail"] = ("b", True)
                 settings.update(properties)
+
+                # Restore existing IPs
+                if existing_ips:
+                    data = settings.get_profile()
+                    data.ipv4.address_data = existing_ips
+                    settings.update_profile(data)
+
                 settings.save()
                 network_manager.activate_connection(connection_path)
             except Exception as e:
@@ -135,6 +162,45 @@ def set_interfaces_priority(self, interfaces: List[NetworkInterfaceMetricApi]) -
                 settings.update(properties)
                 network_manager.activate_connection(connection_path)
 
+    def _get_dhcp_address_using_dhclient(self, interface_name: str) -> str | None:
+        """Run dhclient to get a new IP address and return it.
+
+        Args:
+            interface_name: Name of the interface to get IP for
+
+        Returns:
+            The IP address acquired from DHCP, or None if failed
+        """
+        try:
+            # Just run dhclient without releasing existing IPs
+            dhclient_output = os.popen(f"dhclient -v {interface_name} 2>&1").read()
+
+            bound_ip_match = re.search(r"bound to ([0-9.]+)", dhclient_output)
+            if bound_ip_match:
+                return bound_ip_match.group(1)
+
+            logger.error(f"Could not find bound IP in dhclient output: {dhclient_output}")
+            return None
+
+        except Exception as e:
+            logger.error(f"Failed to run dhclient: {e}")
+            return None
+
+    def trigger_dynamic_ip_acquisition(self, interface_name: str) -> None:
+        """Get a new IP from DHCP using dhclient.
+        The IP will be managed by dhclient and not added to NetworkManager's configuration.
+
+        Args:
+            interface_name: Name of the interface to get IP for
+        """
+        # Get new IP using dhclient
+        new_ip = self._get_dhcp_address_using_dhclient(interface_name)
+        if not new_ip:
+            logger.error(f"Failed to get DHCP-acquired IP for {interface_name}")
+            return
+
+        logger.info(f"Got new IP {new_ip} from DHCP for {interface_name}")
+
 
 class DHCPCD(AbstractNetworkHandler):
     dhcpcd_conf_path = "/etc/dhcpcd.conf"
@@ -245,6 +311,9 @@ def _remove_dhcpcd_configuration(self) -> None:
         with open("/etc/dhcpcd.conf", "w", encoding="utf-8") as f:
             f.writelines(lines)
 
+    def trigger_dynamic_ip_acquisition(self, interface_name: str) -> None:
+        raise NotImplementedError("This Handler does not support triggering dynamic IP acquisition")
+
     def set_interfaces_priority(self, interfaces: List[NetworkInterfaceMetricApi]) -> None:
         """Sets network interface priority..