infinite leases

core/libs/commonwealth/commonwealth/utils/DHCPServerManager.py

@@ -10,16 +10,19 @@
 
 
 # pylint: disable=too-many-arguments
+# pylint: disable=too-many-instance-attributes
 class Dnsmasq:
     def __init__(
         self,
         interface: str,
         ipv4_gateway: IPv4Address,
         subnet_mask: Optional[IPv4Address] = None,
         ipv4_lease_range: Optional[tuple[IPv4Address, IPv4Address]] = None,
-        lease_time: str = "24h",
+        lease_time: str = "infinite",
+        leases_file_path: Optional[str] = None,
     ) -> None:
         self._subprocess: Optional[Any] = None
+        self._leases_file_path = leases_file_path
 
         if interface not in psutil.net_if_stats():
             raise ValueError(f"Interface '{interface}' not found. Available interfaces are {psutil.net_if_stats()}.")
@@ -76,8 +79,8 @@ def command_list(self) -> List[Union[str, pathlib.Path]]:
         """List of arguments to be used in the command line call.
         Refer to https://thekelleys.org.uk/dnsmasq/docs/dnsmasq-man.html for details about each argument."""
 
-        return [
-            self.binary(),
+        options: List[str | pathlib.Path] = [
+            self.binary().as_posix(),
             "--no-daemon",
             f"--interface={self._interface}",
             f"--dhcp-range={self._ipv4_lease_range[0]},{self._ipv4_lease_range[1]},{self._subnet_mask},{self._lease_time}",  # fmt: skip
@@ -93,6 +96,12 @@ def command_list(self) -> List[Union[str, pathlib.Path]]:
             "--port=0",
             "--user=root",
         ]
+        if self._leases_file_path:
+            if pathlib.Path(self._leases_file_path).parent.exists():
+                options.append(f"--dhcp-leasefile={self._leases_file_path}")
+            logger.error(f"Parent folder does not exist: {pathlib.Path(self._leases_file_path).parent}")
+            logger.error("Ignoring option")
+        return options
 
     def start(self) -> None:
         try:

core/services/cable_guy/api/manager.py

@@ -2,6 +2,7 @@
 import subprocess
 import time
 from enum import Enum
+from pathlib import Path
 from socket import AddressFamily
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -91,6 +92,7 @@ def __init__(self, default_configs: List[NetworkInterface]) -> None:
                 self.set_configuration(NetworkInterface(**item))
             except Exception as error:
                 logger.error(f"Failed loading saved configuration. {error}")
+        self.dhcp_leases_path = self.settings.settings_path / "leases.txt"
 
     def save(self) -> None:
         """Save actual configuration"""
@@ -131,7 +133,7 @@ def set_configuration(self, interface: NetworkInterface) -> None:
             if address.mode == AddressMode.Unmanaged:
                 self.add_static_ip(interface.name, address.ip)
             elif address.mode == AddressMode.Server:
-                self.add_dhcp_server_to_interface(interface.name, address.ip)
+                self.add_dhcp_server_to_interface(interface.name, address.ip, leases_file=self.dhcp_leases_path)
 
     def _get_wifi_interfaces(self) -> List[str]:
         """Get wifi interface list
@@ -656,14 +658,16 @@ def remove_dhcp_server_from_interface(self, interface_name: str) -> DHCPServerMa
         except Exception as error:
             raise RuntimeError("Cannot remove DHCP server from interface.") from error
 
-    def add_dhcp_server_to_interface(self, interface_name: str, ipv4_gateway: str) -> None:
+    def add_dhcp_server_to_interface(
+        self, interface_name: str, ipv4_gateway: str, leases_file: Optional[Path] = None
+    ) -> None:
         if self._is_dhcp_server_running_on_interface(interface_name):
             self.remove_dhcp_server_from_interface(interface_name)
         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)
         logger.info(f"Adding DHCP server with gateway '{ipv4_gateway}' to interface '{interface_name}'.")
-        self._dhcp_servers.append(DHCPServerManager(interface_name, ipv4_gateway))
+        self._dhcp_servers.append(DHCPServerManager(interface_name, ipv4_gateway, leases_file_path=leases_file))
 
     def stop(self) -> None:
         """Perform steps necessary to properly stop the manager."""