Add docker-ip-addr-manager initial app

Apps/docker-ip-addr-manager/backend/app/ip_commands.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import subprocess
+from typing import Callable
+
+
+class CommandError(RuntimeError):
+    pass
+
+
+Runner = Callable[[list[str]], subprocess.CompletedProcess[str]]
+
+
+def _default_runner(args: list[str]) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(args, capture_output=True, text=True, check=False)
+
+
+class IpAddressManager:
+    def __init__(self, runner: Runner | None = None):
+        self._runner = runner or _default_runner
+
+    def is_present(self, ip: str, cidr: int, device: str) -> bool:
+        target = f"{ip}/{cidr}"
+        result = self._runner(["ip", "-4", "-o", "addr", "show", "dev", device])
+        if result.returncode != 0:
+            raise CommandError(f"Failed to query addresses on {device}: {result.stderr.strip()}")
+
+        for line in result.stdout.splitlines():
+            tokens = line.split()
+            if target in tokens:
+                return True
+        return False
+
+    def ensure_present(self, ip: str, cidr: int, device: str) -> None:
+        if self.is_present(ip, cidr, device):
+            return
+
+        result = self._runner(["ip", "addr", "add", f"{ip}/{cidr}", "dev", device])
+        if result.returncode != 0 and "File exists" not in result.stderr:
+            raise CommandError(f"Failed to add address {ip}/{cidr} on {device}: {result.stderr.strip()}")
+
+    def ensure_absent(self, ip: str, cidr: int, device: str) -> None:
+        if not self.is_present(ip, cidr, device):
+            return
+
+        result = self._runner(["ip", "addr", "del", f"{ip}/{cidr}", "dev", device])
+        if result.returncode != 0 and "Cannot assign requested address" not in result.stderr:
+            raise CommandError(f"Failed to remove address {ip}/{cidr} on {device}: {result.stderr.strip()}")