diff --git a/daemon/handlers/network.py b/daemon/handlers/network.py index 2f08a31..aff6110 100644 --- a/daemon/handlers/network.py +++ b/daemon/handlers/network.py @@ -103,13 +103,14 @@ def get_interfaces(_request: Any, _body: Any) -> dict[str, Any]: runtime: dict[str, Any] = {} with contextlib.suppress(Exception): - raw = run(["networkctl", "status", "--all"], sudo=True) + raw = run(["networkctl", "status", "--json=short", "--all"], sudo=True) runtime = parse_networkctl_status(raw) merged: dict[str, Any] = {} - for name, config_entry in ifaces_cfg.items(): + all_names = set(ifaces_cfg.keys()) | set(runtime.keys()) - {"lo"} + for name in sorted(all_names): merged[name] = { - "config": config_entry, + "config": ifaces_cfg.get(name, {}), "runtime": runtime.get(name, {}), } @@ -131,7 +132,7 @@ def get_interface(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]: runtime: dict[str, Any] = {} with contextlib.suppress(Exception): - raw = run(["networkctl", "status", "--all"], sudo=True) + raw = run(["networkctl", "status", "--json=short", "--all"], sudo=True) runtime = parse_networkctl_status(raw) return { @@ -160,7 +161,7 @@ def save_interface(_request: Any, body: dict[str, Any] | None) -> dict[str, Any] ) with contextlib.suppress(Exception): - raw = run(["networkctl", "status", "--all"], sudo=True) + raw = run(["networkctl", "status", "--json=short", "--all"], sudo=True) runtime = parse_networkctl_status(raw) if name not in runtime: logger.warning( diff --git a/lib/network.py b/lib/network.py index 317add8..c0ff4b8 100644 --- a/lib/network.py +++ b/lib/network.py @@ -1,10 +1,12 @@ """Networkd/IP configuration module. Reads/writes config/network/config.json, renders .network INI files, -and parses networkctl status output for runtime state. +and parses networkctl JSON output for runtime state. """ +import contextlib import ipaddress +import json import logging from pathlib import Path from typing import Any @@ -427,70 +429,83 @@ def render_network_file(iface_name: str, cfg_entry: dict[str, Any]) -> str: return "\n".join(lines) +def _bytes_to_ip(addr_bytes: list[int], family: int) -> str: + """Convert networkctl JSON address byte array to string.""" + if family == 2: + return str(ipaddress.ip_address(bytes(addr_bytes))) + return str(ipaddress.IPv6Address(bytes(addr_bytes))) + + def parse_networkctl_status(output: str) -> dict[str, Any]: - """Parse ``networkctl status --all`` output into runtime state dict. + """Parse ``networkctl status --json=short --all`` JSON output into runtime state dict. Args: - output: Raw command output from networkctl status. + output: JSON command output from networkctl status. Returns: Dict mapping interface names to their runtime state including addresses, gateway, DNS, and link state. """ + try: + data = json.loads(output) + except (json.JSONDecodeError, TypeError): + return {} + result: dict[str, Any] = {} - current_iface: dict[str, Any] | None = None - def _is_iface_header(line: str) -> bool: - """Check if a line looks like an interface header (digits:name ...).""" - colon_idx = line.find(":") - if colon_idx < 0: - return False - header = line[:colon_idx].strip() - return bool(header) and header[-1].isdigit() - - for raw_line in output.splitlines(): - stripped = raw_line.strip() - if not stripped: + for iface in data.get("Interfaces", []): + name = iface.get("Name") + if not name: continue - # Interface header: "1: eth0" or similar - if _is_iface_header(raw_line): - parts = raw_line.split(":", 1)[1].strip().split() - if parts: - iface_name = parts[0] - current_iface = { - "addresses": [], - "gateway": None, - "dns": [], - "mac": None, - "state": "unknown", - "link": parts[1] if len(parts) > 1 else "unknown", - } - result[iface_name] = current_iface + # Addresses + addresses = [] + for a in iface.get("Addresses", []): + try: + ip = _bytes_to_ip(a["Address"], a["Family"]) + addresses.append(f"{ip}/{a['PrefixLength']}") + except (KeyError, ValueError, TypeError): continue - if current_iface is None: - continue + # Gateway — find default route (Destination 0.0.0.0/0) + gateway = None + for route in iface.get("Routes", []): + if route.get("Family") != 2: + continue + dest = route.get("Destination", []) + prefix = route.get("DestinationPrefixLength", 32) + if len(dest) == 4 and all(d == 0 for d in dest) and prefix == 0: + gw_bytes = route.get("Gateway") + if gw_bytes: + with contextlib.suppress(ValueError, TypeError): + gateway = _bytes_to_ip(gw_bytes, 2) + break - if stripped.startswith("State:"): - current_iface["state"] = stripped.split(":", 1)[1].strip() - elif stripped.startswith("Gateway:"): - gw = stripped.split(":", 1)[1].strip() - if gw and gw.lower() not in ("n/a", ""): - current_iface["gateway"] = gw - elif stripped.startswith("DNS:"): - dns_str = stripped.split(":", 1)[1].strip() - if dns_str and dns_str.lower() != "n/a": - current_iface["dns"] = [d.strip() for d in dns_str.split() if d.strip()] - elif stripped.startswith("Hardware Address:"): - current_iface["mac"] = stripped.split(":", 2)[2].strip() - elif stripped.startswith("Addresses:"): - addr_str = stripped.split(":", 1)[1].strip() - if addr_str and addr_str.lower() != "n/a": - for tok in addr_str.split(): - addr = tok.rstrip(",") - if "/" in addr: - current_iface["addresses"].append(addr) + # DNS + dns = [] + for d in iface.get("DNS", []): + try: + dns.append(_bytes_to_ip(d["Address"], d["Family"])) + except (KeyError, ValueError, TypeError): + continue + + # MAC + mac = None + hw = iface.get("HardwareAddress") + if hw: + mac = ":".join(f"{b:02x}" for b in hw) + + # State + state = iface.get("OperationalState") or "unknown" + + result[name] = { + "addresses": addresses, + "gateway": gateway, + "dns": dns, + "mac": mac, + "state": state, + "link": iface.get("Type", "unknown"), + } return result diff --git a/lib/state.py b/lib/state.py index 23c9388..64297c6 100644 --- a/lib/state.py +++ b/lib/state.py @@ -807,7 +807,7 @@ def _collect_networkd() -> dict[str, Any]: result: dict[str, dict[str, Any]] = {} try: - raw = run(["networkctl", "status", "--all"], sudo=True) + raw = run(["networkctl", "status", "--json=short", "--all"], sudo=True) result = parse_networkctl_status(raw) if not result: return {"interfaces": {}, "timestamp": _now_iso()} diff --git a/tests/test_network.py b/tests/test_network.py index 5379a8e..baf7692 100644 --- a/tests/test_network.py +++ b/tests/test_network.py @@ -1,5 +1,7 @@ """Tests for lib.network module — networkd config, rendering, and parsing.""" +import json + import pytest from lib import network as _net @@ -282,12 +284,36 @@ class TestParseNetworkctlStatus: assert _net.parse_networkctl_status("") == {} def test_single_interface(self): - output = ( - "1: eth0 ethernet 192.168.1.0/24 routable\n" - " State: routable\n" - " Addresses: 192.168.1.1/24,\n" - " Gateway: 192.168.1.254\n" - " DNS: 8.8.8.8 8.8.4.4\n" + output = json.dumps( + { + "Interfaces": [ + { + "Name": "eth0", + "Type": "ether", + "AdministrativeState": "configured", + "OperationalState": "routable", + "Addresses": [ + { + "Family": 2, + "Address": [192, 168, 1, 1], + "PrefixLength": 24, + } + ], + "DNS": [ + {"Family": 2, "Address": [8, 8, 8, 8]}, + {"Family": 2, "Address": [8, 8, 4, 4]}, + ], + "Routes": [ + { + "Family": 2, + "Destination": [0, 0, 0, 0], + "DestinationPrefixLength": 0, + "Gateway": [192, 168, 1, 254], + } + ], + } + ] + } ) result = _net.parse_networkctl_status(output) assert "eth0" in result @@ -298,24 +324,68 @@ class TestParseNetworkctlStatus: assert "8.8.4.4" in iface["dns"] def test_unmanaged(self): - output = "2: lo loopback 127.0.0.1/8 unmanaged\n" + output = json.dumps( + { + "Interfaces": [ + { + "Name": "lo", + "Type": "loopback", + "AdministrativeState": "unmanaged", + "OperationalState": "carrier", + "Addresses": [], + } + ] + } + ) result = _net.parse_networkctl_status(output) assert "lo" in result def test_no_addresses(self): - output = "1: eth0 ethernet (none) degraded\n State: degraded\n" + output = json.dumps( + { + "Interfaces": [ + { + "Name": "eth0", + "Type": "ether", + "AdministrativeState": "degraded", + "OperationalState": "degraded", + "Addresses": [], + } + ] + } + ) result = _net.parse_networkctl_status(output) assert "eth0" in result assert result["eth0"]["addresses"] == [] def test_multiple_interfaces(self): - output = ( - "1: eth0 ethernet 192.168.1.0/24 routable\n" - " State: routable\n" - " Addresses: 192.168.1.1/24,\n" - "2: eth1 ethernet 10.0.0.0/24 routable\n" - " State: routable\n" - " Addresses: 10.0.0.1/24,\n" + output = json.dumps( + { + "Interfaces": [ + { + "Name": "eth0", + "Type": "ether", + "AdministrativeState": "configured", + "OperationalState": "routable", + "Addresses": [ + { + "Family": 2, + "Address": [192, 168, 1, 1], + "PrefixLength": 24, + } + ], + }, + { + "Name": "eth1", + "Type": "ether", + "AdministrativeState": "configured", + "OperationalState": "routable", + "Addresses": [ + {"Family": 2, "Address": [10, 0, 0, 1], "PrefixLength": 24} + ], + }, + ] + } ) result = _net.parse_networkctl_status(output) assert "eth0" in result diff --git a/tests/test_network_integration.py b/tests/test_network_integration.py index ce1ee99..78ec124 100644 --- a/tests/test_network_integration.py +++ b/tests/test_network_integration.py @@ -3,6 +3,7 @@ Tests TF-8 (DNS upstream sync), TF-9 (DHCP range inference), TF-10 (zone inference). """ +import json from pathlib import Path from unittest.mock import patch @@ -290,12 +291,34 @@ class TestStateParserDedup: import lib.state as _state with patch("lib.state.run") as mock_run: - mock_run.return_value = ( - "1: eth0 ethernet 10.0.0.0/24 routable\n" - " State: routable\n" - " Addresses: 10.0.0.1/24,\n" - " Gateway: 10.0.0.254\n" - " DNS: 8.8.8.8\n" + mock_run.return_value = json.dumps( + { + "Interfaces": [ + { + "Name": "eth0", + "Type": "ether", + "OperationalState": "routable", + "Addresses": [ + { + "Family": 2, + "Address": [10, 0, 0, 1], + "PrefixLength": 24, + } + ], + "DNS": [ + {"Family": 2, "Address": [8, 8, 8, 8]}, + ], + "Routes": [ + { + "Family": 2, + "Destination": [0, 0, 0, 0], + "DestinationPrefixLength": 0, + "Gateway": [10, 0, 0, 254], + } + ], + } + ] + } ) result = _state._collect_networkd()