"""Host network settings (spec §15): agent ops, gating, web API. Two layers: * the agent protocol itself — validation, whitelist refusals, auto-snapshot + restore, audit trail, config loading (FakeBackend; the Linux backend shares the same dispatch in `handle`) * the web API — the FR-15.2.1/2 gates (404 when off), readonly 403, mutation mapping through an InProcessClient-backed HostNetService (the socket path is exercised only on real Linux hosts). """ from __future__ import annotations import pytest from wago_sim.hostnet import (FakeBackend, HostNetAgent, HostNetError, HostNetRefused, HostNetService, HostNetUnavailable, InProcessClient, hostnet_enabled, load_agent_config, parse_iface, parse_ipv4, parse_prefix) from wago_sim.webui.app import build_webui @pytest.fixture def agent(): """Agent with two whitelisted interfaces on the fake backend.""" return HostNetAgent( backend=FakeBackend({"eth0": ["192.168.1.10/24"], "eth1": ["10.0.0.5/8"], "lo": ["127.0.0.1/8"]}), whitelist=["eth0", "eth1"], ) @pytest.fixture def service(agent): return HostNetService(InProcessClient(agent)) @pytest.fixture async def hostnet_client(service): """httpx client against a web UI with the hostnet service mounted.""" from wago_sim.app import build_app from wago_sim.modules import DI_8, DO_8 sim = build_app(modules=[DI_8, DO_8]) webui = build_webui(sim, user="admin", password="simu", hostnet_service=service) import httpx transport = httpx.ASGITransport(app=webui) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: r = await c.post("/login", json={"user": "admin", "password": "simu"}) assert r.status_code == 200 yield c @pytest.fixture async def plain_client(): """httpx client against a web UI withOUT the hostnet service.""" from wago_sim.app import build_app from wago_sim.modules import DI_8, DO_8 sim = build_app(modules=[DI_8, DO_8]) webui = build_webui(sim, user="admin", password="simu") import httpx transport = httpx.ASGITransport(app=webui) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: r = await c.post("/login", json={"user": "admin", "password": "simu"}) assert r.status_code == 200 yield c # --------------------------------------------------------------------------- # Validation helpers (FR-15.2.4 — used by both the API and the agent) # --------------------------------------------------------------------------- class TestValidation: def test_ipv4_normalized(self): assert parse_ipv4(" 192.168.1.33 ") == "192.168.1.33" @pytest.mark.parametrize("bad", ["banana", "1.2.3", "1.2.3.999", "", None, 33, "192.168.1.1/24"]) def test_ipv4_rejected(self, bad): with pytest.raises(HostNetError): parse_ipv4(bad) def test_prefix_bounds(self): assert parse_prefix(24) == 24 assert parse_prefix("28") == 28 for bad in (0, 33, "x", None, -1): with pytest.raises(HostNetError): parse_prefix(bad) def test_iface_name_shape(self): assert parse_iface("eth0") == "eth0" assert parse_iface("enp3s0") == "enp3s0" assert parse_iface("br-123_4") == "br-123_4" for bad in ("", None, "eth 0", "eth0; rm -rf /", "x" * 16, "/proc/x"): with pytest.raises(HostNetError): parse_iface(bad) # --------------------------------------------------------------------------- # Agent protocol (FR-15.2.4 / 15.2.5) # --------------------------------------------------------------------------- class TestAgentOps: def test_health_reports_whitelist(self, agent): r = agent.handle({"op": "health"}) assert r["ok"] is True assert r["whitelist"] == ["eth0", "eth1"] assert r["protocol"] == 1 def test_list_is_live_state(self, agent): r = agent.handle({"op": "list"}) names = {i["name"] for i in r["interfaces"]} assert names == {"eth0", "eth1", "lo"} eth0 = next(i for i in r["interfaces"] if i["name"] == "eth0") assert eth0["addresses"] == [{"address": "192.168.1.10", "prefix": 24}] def test_add_address(self, agent): r = agent.handle({"op": "add_address", "iface": "eth0", "address": "192.168.1.33", "prefix": 24}) assert r["ok"] is True eth0 = next(i for i in agent.handle({"op": "list"})["interfaces"] if i["name"] == "eth0") assert {"address": "192.168.1.33", "prefix": 24} in eth0["addresses"] def test_add_duplicate_rejected(self, agent): agent.handle({"op": "add_address", "iface": "eth0", "address": "192.168.1.33", "prefix": 24}) r = agent.handle({"op": "add_address", "iface": "eth0", "address": "192.168.1.33", "prefix": 24}) assert r["ok"] is False assert "already exists" in r["error"] def test_remove_address(self, agent): agent.handle({"op": "add_address", "iface": "eth0", "address": "192.168.1.33", "prefix": 24}) r = agent.handle({"op": "remove_address", "iface": "eth0", "address": "192.168.1.33"}) assert r["ok"] is True eth0 = next(i for i in agent.handle({"op": "list"})["interfaces"] if i["name"] == "eth0") assert all(a["address"] != "192.168.1.33" for a in eth0["addresses"]) def test_remove_missing_rejected(self, agent): r = agent.handle({"op": "remove_address", "iface": "eth0", "address": "9.9.9.9"}) assert r["ok"] is False assert "not assigned" in r["error"] def test_change_address(self, agent): r = agent.handle({"op": "change_address", "iface": "eth0", "old": "192.168.1.10", "new": "192.168.1.20", "prefix": 24}) assert r["ok"] is True eth0 = next(i for i in agent.handle({"op": "list"})["interfaces"] if i["name"] == "eth0") assert {"address": "192.168.1.20", "prefix": 24} in eth0["addresses"] assert all(a["address"] != "192.168.1.10" for a in eth0["addresses"]) def test_whitelist_refusal(self, agent): # lo exists but is NOT whitelisted -> refused + audited r = agent.handle({"op": "add_address", "iface": "lo", "address": "127.0.0.2", "prefix": 8}) assert r["ok"] is False assert r.get("refused") is True assert "whitelist" in r["error"] def test_unknown_interface_refused(self, agent): r = agent.handle({"op": "snapshot", "iface": "wlan9"}) assert r["ok"] is False assert r.get("refused") is True def test_unknown_op(self, agent): r = agent.handle({"op": "reboot_host"}) assert r["ok"] is False assert "unknown operation" in r["error"] def test_invalid_input_rejected(self, agent): r = agent.handle({"op": "add_address", "iface": "eth0", "address": "not-an-ip", "prefix": 24}) assert r["ok"] is False assert "invalid IPv4" in r["error"] class TestSnapshotRestore: """FR-15.2.5: snapshot before every mutation; restore walks back.""" def _addresses(self, agent, iface): return [a["address"] for i in agent.handle({"op": "list"})["interfaces"] if i["name"] == iface for a in i["addresses"]] def test_auto_snapshot_on_add(self, agent): before = self._addresses(agent, "eth0") agent.handle({"op": "add_address", "iface": "eth0", "address": "192.168.1.33", "prefix": 24}) # restore walks back to the pre-add state r = agent.handle({"op": "restore", "iface": "eth0"}) assert r["ok"] is True assert self._addresses(agent, "eth0") == before def test_restore_after_remove(self, agent): agent.handle({"op": "remove_address", "iface": "eth0", "address": "192.168.1.10"}) assert "192.168.1.10" not in self._addresses(agent, "eth0") agent.handle({"op": "restore", "iface": "eth0"}) assert "192.168.1.10" in self._addresses(agent, "eth0") def test_restore_without_snapshot_rejected(self, agent): r = agent.handle({"op": "restore", "iface": "eth1"}) # eth1 had no mutation yet -> nothing to restore assert r["ok"] is False assert "no snapshot" in r["error"] def test_explicit_snapshot_op(self, agent): r = agent.handle({"op": "snapshot", "iface": "eth0"}) assert r["ok"] is True assert [a[0] for a in r["addresses"]] == ["192.168.1.10"] class TestAudit: def test_mutations_and_refusals_audited(self, agent): agent.handle({"op": "add_address", "iface": "eth0", "address": "192.168.1.33", "prefix": 24, "operator": "admin"}) agent.handle({"op": "add_address", "iface": "lo", "address": "127.0.0.2", "prefix": 8}) lines = agent.handle({"op": "audit", "limit": 50})["lines"] joined = "\n".join(lines) assert "add eth0 192.168.1.33/24 by admin" in joined assert "REFUSED" in joined assert "whitelist" in joined def test_audit_bounded(self, agent): for i in range(10): agent.handle({"op": "add_address", "iface": "eth0", "address": f"10.60.{i}.1", "prefix": 24}) lines = agent.handle({"op": "audit", "limit": 5})["lines"] assert len(lines) == 5 def test_operator_tagged(self, agent): agent.handle({"op": "add_address", "iface": "eth0", "address": "10.9.9.1", "prefix": 24, "operator": "commissioner"}) joined = "\n".join(agent.handle({"op": "audit"})["lines"]) assert "by commissioner" in joined # --------------------------------------------------------------------------- # Gating (FR-15.2.1 / 15.2.2) # --------------------------------------------------------------------------- class TestGating: def test_off_without_flag(self, tmp_path): assert hostnet_enabled({"WAGO_SIM_HOSTNET": ""}, []) is False assert hostnet_enabled({}, []) is False def test_flag_alone_is_not_enough(self, tmp_path): # flag set, docker markers absent, no env marker -> off assert hostnet_enabled( {"WAGO_SIM_HOSTNET": "1", "WAGO_SIM_IN_DOCKER": "0"}, []) is False def test_flag_and_docker_env(self): assert hostnet_enabled( {"WAGO_SIM_HOSTNET": "1", "WAGO_SIM_IN_DOCKER": "1"}, []) is True def test_flag_and_marker_file(self, tmp_path): marker = tmp_path / ".dockerenv" marker.write_text("") assert hostnet_enabled({"WAGO_SIM_HOSTNET": "true"}, [marker]) is True @pytest.mark.parametrize("value", ["1", "true", "yes", "on", "YES", " On "]) def test_flag_truthy_forms(self, value): assert hostnet_enabled( {"WAGO_SIM_HOSTNET": value, "WAGO_SIM_IN_DOCKER": "1"}, []) is True def test_flag_falsy_form(self): assert hostnet_enabled( {"WAGO_SIM_HOSTNET": "0", "WAGO_SIM_IN_DOCKER": "1"}, []) is False # --------------------------------------------------------------------------- # Agent config + startup rules # --------------------------------------------------------------------------- class TestAgentConfig: def test_valid_config(self, tmp_path): cfg_file = tmp_path / "agent.yaml" cfg_file.write_text( "socket: /tmp/x.sock\nwhitelist: [eth0, eth1]\naudit_path: /tmp/a.log\n") cfg = load_agent_config(str(cfg_file)) assert cfg["socket"] == "/tmp/x.sock" assert cfg["whitelist"] == ["eth0", "eth1"] def test_missing_file(self): with pytest.raises(HostNetError): load_agent_config("nope/missing.yaml") def test_empty_whitelist_refused_at_startup(self, tmp_path): # a config without a whitelist must not start the agent cfg_file = tmp_path / "agent.yaml" cfg_file.write_text("socket: /tmp/x.sock\nwhitelist: []\n") with pytest.raises(HostNetError): HostNetAgent(backend=FakeBackend(), whitelist=[]) def test_bad_yaml(self, tmp_path): cfg_file = tmp_path / "agent.yaml" cfg_file.write_text("socket: [unclosed\n") with pytest.raises(HostNetError): load_agent_config(str(cfg_file)) # --------------------------------------------------------------------------- # Web API (spec §15.3 surface) # --------------------------------------------------------------------------- class TestHostnetAPI: async def test_disabled_routes_are_404(self, plain_client): # FR-15.2.1/2: no page, no API surface when the gates are off r = await plain_client.get("/api/hostnet") assert r.status_code == 404 r = await plain_client.get("/api/hostnet/audit") assert r.status_code == 404 r = await plain_client.post("/api/hostnet/interfaces/eth0/addresses", json={"address": "10.1.1.1", "prefix": 24}) assert r.status_code == 404 async def test_status_and_list(self, hostnet_client): r = await hostnet_client.get("/api/hostnet") assert r.status_code == 200 data = r.json() assert data["status"]["available"] is True assert data["status"]["agent"]["whitelist"] == ["eth0", "eth1"] names = {i["name"] for i in data["interfaces"]} assert names == {"eth0", "eth1", "lo"} async def test_add_then_remove_via_api(self, hostnet_client): r = await hostnet_client.post( "/api/hostnet/interfaces/eth0/addresses", json={"address": "192.168.1.33", "prefix": 24}) assert r.status_code == 200, r.text assert r.json()["address"] == "192.168.1.33" r = await hostnet_client.delete( "/api/hostnet/interfaces/eth0/addresses/192.168.1.33") assert r.status_code == 200 async def test_change_via_api(self, hostnet_client): r = await hostnet_client.put( "/api/hostnet/interfaces/eth0/addresses", json={"old": "192.168.1.10", "new": "192.168.1.20", "prefix": 24}) assert r.status_code == 200 r = await hostnet_client.get("/api/hostnet") eth0 = next(i for i in r.json()["interfaces"] if i["name"] == "eth0") assert {"address": "192.168.1.20", "prefix": 24} in eth0["addresses"] async def test_restore_via_api(self, hostnet_client): await hostnet_client.post( "/api/hostnet/interfaces/eth0/addresses", json={"address": "192.168.1.33", "prefix": 24}) r = await hostnet_client.post("/api/hostnet/interfaces/eth0/restore") assert r.status_code == 200 r = await hostnet_client.get("/api/hostnet") eth0 = next(i for i in r.json()["interfaces"] if i["name"] == "eth0") assert all(a["address"] != "192.168.1.33" for a in eth0["addresses"]) async def test_bad_input_400(self, hostnet_client): r = await hostnet_client.post( "/api/hostnet/interfaces/eth0/addresses", json={"address": "banana", "prefix": 24}) assert r.status_code == 400 assert "invalid IPv4" in r.json()["detail"] async def test_non_whitelisted_403(self, hostnet_client): r = await hostnet_client.post( "/api/hostnet/interfaces/lo/addresses", json={"address": "127.0.0.2", "prefix": 8}) assert r.status_code == 403 assert "whitelist" in r.json()["detail"] async def test_audit_view(self, hostnet_client): await hostnet_client.post( "/api/hostnet/interfaces/eth0/addresses", json={"address": "192.168.1.33", "prefix": 24}) r = await hostnet_client.get("/api/hostnet/audit") assert r.status_code == 200 assert "192.168.1.33" in "\n".join(r.json()["lines"]) async def test_operator_in_audit_trail(self, hostnet_client): await hostnet_client.post( "/api/hostnet/interfaces/eth0/addresses", json={"address": "10.77.0.1", "prefix": 24}) r = await hostnet_client.get("/api/hostnet/audit") # the logged-in user (admin) is named in every mutation line assert "by admin" in "\n".join(r.json()["lines"]) class TestHostnetReadonly: @pytest.fixture async def ro_client(self, agent): from wago_sim.app import build_app from wago_sim.modules import DI_8, DO_8 import httpx sim = build_app(modules=[DI_8, DO_8]) service = HostNetService(InProcessClient(agent)) webui = build_webui(sim, user="admin", password="simu", readonly=True, hostnet_service=service) transport = httpx.ASGITransport(app=webui) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: await c.post("/login", json={"user": "admin", "password": "simu"}) yield c async def test_read_status_allowed(self, ro_client): r = await ro_client.get("/api/hostnet") assert r.status_code == 200 assert r.json()["status"]["available"] is True async def test_mutations_403_in_readonly(self, ro_client): r = await ro_client.post( "/api/hostnet/interfaces/eth0/addresses", json={"address": "192.168.1.33", "prefix": 24}) assert r.status_code == 403 r = await ro_client.delete( "/api/hostnet/interfaces/eth0/addresses/192.168.1.10") assert r.status_code == 403 r = await ro_client.put( "/api/hostnet/interfaces/eth0/addresses", json={"old": "192.168.1.10", "new": "1.2.3.4", "prefix": 24}) assert r.status_code == 403 r = await ro_client.post("/api/hostnet/interfaces/eth0/restore") assert r.status_code == 403 class TestUnavailableAgent: async def test_unavailable_reported_not_raised(self): """FR-15.2.8: agent down -> available:false, no 5xx, banner text.""" from wago_sim.app import build_app from wago_sim.modules import DI_8, DO_8 import httpx class DeadClient: """Agent down: every op raises HostNetUnavailable (FR-15.2.8).""" def health(self): raise HostNetUnavailable("socket not found") def list(self): raise HostNetUnavailable("socket not found") def add_address(self, iface, address, prefix, operator=""): raise HostNetUnavailable("socket not found") def remove_address(self, iface, address, operator=""): raise HostNetUnavailable("socket not found") def change_address(self, iface, old, new, prefix, operator=""): raise HostNetUnavailable("socket not found") def restore(self, iface, operator=""): raise HostNetUnavailable("socket not found") def audit(self, limit=50): raise HostNetUnavailable("socket not found") sim = build_app(modules=[DI_8, DO_8]) webui = build_webui(sim, user="admin", password="simu", hostnet_service=HostNetService(DeadClient())) transport = httpx.ASGITransport(app=webui) async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: await c.post("/login", json={"user": "admin", "password": "simu"}) r = await c.get("/api/hostnet") assert r.status_code == 200 data = r.json() assert data["status"]["available"] is False assert "socket not found" in data["status"]["error"] assert data["interfaces"] is None # mutations report 503 with the same message r = await c.post("/api/hostnet/interfaces/eth0/addresses", json={"address": "1.2.3.4", "prefix": 24}) assert r.status_code == 503 async def test_hostnet_client_without_socket(self, tmp_path): from wago_sim.hostnet import HostNetClient client = HostNetClient(socket_path=str(tmp_path / "missing.sock")) with pytest.raises(HostNetUnavailable): client.health() class TestIproute2ParseFallback: """The `ip -o` output parsers — the fallback read path.""" def test_addr_parse(self, monkeypatch): monkeypatch.setenv("PATH", "/nonexistent") # `ip` never found from wago_sim.hostnet import Iproute2Backend backend = Iproute2Backend.__new__(Iproute2Backend) backend._ipr = None sample = ("2: eth0 inet 192.168.1.5/24 brd 192.168.1.255 " "scope global eth0\n" "3: eth1 inet 10.0.0.9/8 brd 10.255.255.255 " "scope global eth1\n" "4: docker0 inet 172.17.0.1/16 brd 172.17.255.255 " "scope global docker0\n") monkeypatch.setattr(backend, "_ip_cmd", lambda cmd: sample) addrs = backend._addr_list() assert ("eth0", "192.168.1.5", 24) in addrs assert ("eth1", "10.0.0.9", 8) in addrs def test_link_parse(self, monkeypatch): monkeypatch.setenv("PATH", "/nonexistent") from wago_sim.hostnet import Iproute2Backend backend = Iproute2Backend.__new__(Iproute2Backend) backend._ipr = None sample = ("1: lo: mtu 65536 qdisc noqueue\n" "2: eth0: mtu 1500\n" "3: eth9: mtu 1500\n") monkeypatch.setattr(backend, "_ip_cmd", lambda cmd: sample) links = backend._link_list() assert ("lo", True) in links assert ("eth0", True) in links assert ("eth9", False) in links