Joachim Friberg
49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
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()}")
|