core: services: cable_guy: api: manager: Remove subnet from deleted IP

core/services/cable_guy/api/manager.py

@@ -19,6 +19,7 @@
 from networksetup import AbstractNetworkHandler, NetworkHandlerDetector
 from pyroute2 import IW, NDB, IPRoute
 from pyroute2.netlink.exceptions import NetlinkError
+from pyroute2.netlink.rtnl import rtprotos
 from pyroute2.netlink.rtnl.ifaddrmsg import ifaddrmsg
 from typedefs import (
     AddressMode,
@@ -380,8 +381,38 @@ def remove_ip(self, interface_name: str, ip_address: str) -> None:
                 address for address in saved_interface.addresses if address.mode != AddressMode.Client
             ]
 
+        self._remove_orphaned_subnet_route(interface_name, ip_address, saved_interface)
+
         self._update_interface_settings(interface_name, saved_interface)
 
+    def _remove_orphaned_subnet_route(
+        self, interface_name: str, ip_address: str, saved_interface: NetworkInterface
+    ) -> None:
+        # A subnet route previously adopted will outlives its IP and can hijack traffic
+        if not self.weak_is_ip_address(ip_address) or ip_address == "0.0.0.0":
+            return
+
+        # /24 mirrors the prefix used when adding/removing static IPs
+        subnet = ip_network(f"{ip_address}/24", strict=False)
+        try:
+            remaining = self.get_interface_by_name(interface_name).addresses
+        except Exception as error:
+            logger.error(f"Could not check remaining addresses on {interface_name}: {error}")
+            return
+        if any(self.weak_is_ip_address(address.ip) and IPv4Address(address.ip) in subnet for address in remaining):
+            return
+
+        try:
+            self.ipr.route("del", dst=str(subnet), oif=self._get_interface_index(interface_name))
+            logger.info(f"Removed orphaned route {subnet} from interface {interface_name}.")
+        except NetlinkError as error:
+            # the kernel already removes its own connected routes on address deletion, so a missing route is fine
+            if error.code not in (errno.ESRCH, errno.ENOENT):
+                logger.error(f"Failed to remove orphaned route {subnet} on {interface_name}: {error}")
+                return
+
+        saved_interface.routes = [route for route in saved_interface.routes if route.destination != str(subnet)]
+
     def get_interface_by_name(self, name: str, include_dhcp_markers: bool = False) -> NetworkInterface:
         """Get interface by name.
 
@@ -689,6 +720,11 @@ def get_routes(self, interface_name: str, ignore_unmanaged: bool = True) -> Set[
 
         routes: Set[Route] = set()
         for raw_route in raw_routes:
+            # Kernel-maintained connected routes are created/removed automatically alongside their
+            # interface addresses. Adopting them would turn them into persistent routes
+            # that outlive their IP, so we ignore them entirely.
+            if raw_route["proto"] == rtprotos["RTPROT_KERNEL"]:
+                continue
             try:
                 route = self._parse_route(raw_route)