router-ui: web management page for ports, devices, traffic, speedtest

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

scripts/router-ui-check.py

@@ -0,0 +1,143 @@
+"""Self-check for router-ui.py: TOML round-trips and the refusal rules.
+
+Run it against the same python the service uses:
+
+    nix shell .#nixosConfigurations.FredOS-Mediaserver.pkgs.python3Packages.tomlkit \
+      -c python3 scripts/router-ui-check.py
+
+The TOML round-trip cases are the ones that matter: a save must leave an
+unedited file byte-identical, and must never eat the parked/commented forwards
+in ports.toml.
+"""
+import difflib
+import importlib.util
+import os
+import sys
+import tempfile
+
+import tomlkit
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+REPO = os.path.join(HERE, "..")
+
+spec = importlib.util.spec_from_file_location("routerui", os.path.join(HERE, "router-ui.py"))
+r = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(r)
+
+
+def refuses(fn, *a):
+    try:
+        fn(*a)
+    except r.Refused as e:
+        return str(e)
+    raise AssertionError(f"{fn.__name__} accepted what it should have refused")
+
+
+def same(before, after, what):
+    assert before == after, what + ":\n" + "".join(
+        difflib.unified_diff(before.splitlines(1), after.splitlines(1), "before", "after"))
+
+
+# --- ports.toml round-trip ---------------------------------------------------
+src = open(os.path.join(REPO, "ports.toml")).read()
+doc = tomlkit.parse(src)
+items = r.forwards_from(doc)
+assert len(items) == 4, items
+assert items[3]["port"] == "25565-25600", items[3]
+
+# Saving without editing anything must not touch a single byte.
+r.check_forwards(items)
+r.edit_aot(doc, "forward", items)
+same(src, tomlkit.dumps(doc), "no-op forward save rewrote the file")
+
+# Add one, drop the Pelican range; every comment block must survive.
+doc2 = tomlkit.parse(src)
+items2 = r.forwards_from(doc2)
+items2.pop(3)
+items2.append({"_i": None, "name": "Test", "port": "9999", "protocol": "udp",
+               "dest": "10.0.0.5"})
+r.check_forwards(items2)
+r.edit_aot(doc2, "forward", items2)
+out = tomlkit.dumps(doc2)
+assert "7DTD voice/dynamic" in out, "parked blocks were lost"
+assert "WoW Classic realmd" in out, "parked blocks were lost"
+assert "Pelican game-server allocation range" in out, "inline comment was lost"
+assert "Pelican game servers" not in out.replace("# Pelican", ""), "deletion did not happen"
+assert 'name = "Test"' in out and "port = 9999" in out, out[-300:]
+back = tomlkit.parse(out)["forward"]
+assert len(back) == 4 and back[3]["protocol"] == "udp" and back[3]["dest"] == "10.0.0.5"
+assert [f["name"] for f in back[:3]] == ["HTTP", "HTTPS", "SSH"]
+
+# --- forward validation ------------------------------------------------------
+mk = lambda: r.forwards_from(tomlkit.parse(src))  # noqa: E731 - fresh fixture per case
+print(refuses(r.check_forwards, [f for f in mk() if f["name"] != "SSH"]))
+print(refuses(r.check_forwards, [f for f in mk() if f["name"] != "HTTPS"]))
+print(refuses(r.check_forwards, mk() + [{"_i": None, "name": "x", "port": "70000",
+                                         "protocol": "tcp", "dest": ""}]))
+print(refuses(r.check_forwards, mk() + [{"_i": None, "name": "x", "port": "9-8",
+                                         "protocol": "tcp", "dest": ""}]))
+print(refuses(r.check_forwards, mk() + [{"_i": None, "name": "x", "port": "80",
+                                         "protocol": "sctp", "dest": ""}]))
+print(refuses(r.check_forwards, mk() + [{"_i": None, "name": "", "port": "80",
+                                         "protocol": "tcp", "dest": ""}]))
+# A range that covers 22 and 443 satisfies the keep-me-reachable rule.
+r.check_forwards([{"_i": None, "name": "wide", "port": "20-500", "protocol": "both",
+                   "dest": ""}])
+
+# --- device validation -------------------------------------------------------
+def dev(**kw):
+    base = {"_i": None, "mac": "aa:bb:cc:dd:ee:01", "name": "thing", "ip": "",
+            "blocked": False, "note": ""}
+    return {**base, **kw}
+
+
+me, my_ip = "aa:bb:cc:dd:ee:99", "10.0.0.161"
+
+print(refuses(r.check_devices, [dev(ip="10.0.0.161")], me, my_ip))   # inside the pool
+print(refuses(r.check_devices, [dev(ip="10.0.0.1")], me, my_ip))     # the router itself
+print(refuses(r.check_devices, [dev(ip="192.168.1.5")], me, my_ip))  # off-LAN
+print(refuses(r.check_devices, [dev(mac="nope")], me, my_ip))
+print(refuses(r.check_devices, [dev(name="bad name")], me, my_ip))
+print(refuses(r.check_devices, [dev(), dev()], me, my_ip))           # duplicate MAC
+print(refuses(r.check_devices, [dev(ip="10.0.0.20"),
+                                dev(mac="aa:bb:cc:dd:ee:02", ip="10.0.0.20")], me, my_ip))
+print(refuses(r.check_devices, [dev(mac=me, blocked=True)], me, my_ip))
+print(refuses(r.check_devices, [dev(ip="10.0.0.30", blocked=True)], me, "10.0.0.30"))
+
+ok = [dev(ip="10.0.0.39", name="camera-bedroom")]
+r.check_devices(ok, me, my_ip)
+assert ok[0]["ip"] == "10.0.0.39" and "blocked" not in ok[0] and "note" not in ok[0]
+cleared = [dev(ip="")]
+r.check_devices(cleared, me, my_ip)
+assert "ip" not in cleared[0], "an empty reservation should drop the key entirely"
+
+# --- devices.toml round-trip -------------------------------------------------
+dsrc = open(os.path.join(REPO, "devices.toml")).read()
+ddoc = tomlkit.parse(dsrc)
+ditems = r.devices_from(ddoc)
+assert len(ditems) == 1 and ditems[0]["ip"] == "10.0.0.39"
+r.check_devices(ditems, me, my_ip)
+r.edit_aot(ddoc, "device", ditems)
+same(dsrc, tomlkit.dumps(ddoc), "no-op device save rewrote the file")
+
+ditems = r.devices_from(ddoc)
+ditems[0]["ip"] = ""
+ditems[0]["blocked"] = True
+r.check_devices(ditems, me, my_ip)
+r.edit_aot(ddoc, "device", ditems)
+entry = tomlkit.dumps(ddoc).split("[[device]]")[1]
+assert "ip = " not in entry, entry
+assert "blocked = true" in entry, entry
+
+# --- speedtest parsing (bytes/s -> Mbps, ns -> ms) ---------------------------
+sample = ('{"timestamp":"2026-08-15 12:00:55","servers":[{"name":"Preston",'
+          '"latency":18796260,"dl_speed":63607890.17,"ul_speed":8839605.08}]}\n')
+tmp = os.path.join(tempfile.mkdtemp(), "st.jsonl")
+open(tmp, "w").write(sample + "not json at all\n" + sample)
+r.SPEEDTEST_LOG = tmp
+got = r.speedtests()
+assert len(got) == 2, got
+assert got[0] == {"ts": "2026-08-15 12:00:55", "down": 508.9, "up": 70.7,
+                  "ping": 18.8, "server": "Preston"}, got[0]
+
+print("\nall checks passed", file=sys.stderr)