Test WireGuard status API

app.py

@@ -1213,6 +1213,67 @@ def _ssh(host, cmd, timeout=600):
         return 124, "", f"SSH command timed out after {timeout} seconds"
 
 
+def _wireguard_status_command() -> str:
+    script = '''import json, subprocess
+
+def run(args):
+    proc = subprocess.run(args, text=True, capture_output=True, timeout=15)
+    if proc.returncode != 0:
+        raise RuntimeError((proc.stderr or proc.stdout).strip() or "command failed")
+    return proc.stdout.strip()
+
+result = {"interface": "wg0", "addresses": [], "listen_port": None, "service_active": False, "service_enabled": False, "routes": [], "peers": []}
+result["service_active"] = subprocess.run(["systemctl", "is-active", "--quiet", "wg-quick@wg0"]).returncode == 0
+result["service_enabled"] = subprocess.run(["systemctl", "is-enabled", "--quiet", "wg-quick@wg0"]).returncode == 0
+try:
+    addr_data = json.loads(run(["ip", "-j", "address", "show", "dev", "wg0"]))
+    for item in addr_data:
+        for address in item.get("addr_info", []):
+            result["addresses"].append(address["local"] + "/" + str(address["prefixlen"]))
+    result["routes"] = json.loads(run(["ip", "-j", "route", "show", "dev", "wg0"]))
+    rows = run(["wg", "show", "wg0", "dump"]).splitlines()
+    if rows:
+        interface = rows[0].split("\\t")
+        result["listen_port"] = int(interface[2])
+    for raw in rows[1:]:
+        fields = raw.split("\\t")
+        result["peers"].append({
+            "public_key": fields[0],
+            "endpoint": None if fields[2] == "(none)" else fields[2],
+            "allowed_ips": [] if fields[3] == "(none)" else fields[3].split(","),
+            "latest_handshake": int(fields[4]),
+            "rx_bytes": int(fields[5]),
+            "tx_bytes": int(fields[6]),
+            "persistent_keepalive": int(fields[7]),
+        })
+except Exception as exc:
+    result["error"] = str(exc)[:300]
+print(json.dumps(result))
+'''
+    encoded = base64.b64encode(script.encode()).decode()
+    return f'python3 -c "import base64;exec(base64.b64decode(\'{encoded}\'))"'
+
+
+@app.get("/network/wireguard/{host}")
+async def network_wireguard_status(host: str, _=Depends(_verify)):
+    """Return redacted WireGuard state without private or preshared keys."""
+    if not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,62}", host):
+        raise HTTPException(400, "Invalid host name")
+    inventory = await asyncio.to_thread(_find_inventory_host, host)
+    if not inventory:
+        raise HTTPException(404, f"Host {host} not found")
+    target = f'{inventory["user"]}@{inventory["ip"]}'
+    rc, out, err = await asyncio.to_thread(_ssh, target, _wireguard_status_command(), 30)
+    if rc != 0:
+        raise HTTPException(502, (err or out).strip()[-500:] or "WireGuard status failed")
+    try:
+        result = json.loads(out)
+    except json.JSONDecodeError as exc:
+        raise HTTPException(502, "WireGuard status returned invalid JSON") from exc
+    _audit(f"/network/wireguard/{host}", "GET", 200, "redacted live status")
+    return {"host": host, **result}
+
+
 SYSCTL_AUDIT_KEYS = (
     "net.core.default_qdisc",
     "net.core.rmem_default",

tests/test_app.py

@@ -46,6 +46,52 @@ def test_health_exposes_current_version():
     assert response.json()["version"] == app.VERSION == "2.3.5"
 
 
+def test_wireguard_status_returns_redacted_live_state(monkeypatch):
+    payload = {
+        "interface": "wg0",
+        "addresses": ["10.11.12.1/32"],
+        "listen_port": 37888,
+        "service_active": True,
+        "service_enabled": True,
+        "routes": [{"dst": "10.11.12.3", "prefsrc": "10.11.12.1"}],
+        "peers": [{
+            "public_key": "peer-public-key",
+            "endpoint": "203.0.113.9:51820",
+            "allowed_ips": ["10.11.12.3/32"],
+            "latest_handshake": 123,
+            "rx_bytes": 456,
+            "tx_bytes": 789,
+            "persistent_keepalive": 25,
+        }],
+    }
+    monkeypatch.setattr(app, "_find_inventory_host", lambda host: {"user": "debian", "ip": "141.94.237.199"})
+    monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=30: (0, json.dumps(payload), ""))
+
+    with TestClient(app.app) as client:
+        response = client.get(
+            "/network/wireguard/guck-vps",
+            headers={"Authorization": "Bearer test-token"},
+        )
+
+    assert response.status_code == 200
+    assert response.json()["host"] == "guck-vps"
+    assert response.json()["peers"][0]["allowed_ips"] == ["10.11.12.3/32"]
+    assert "private" not in response.text.lower()
+
+
+def test_wireguard_status_rejects_unknown_host_without_ssh(monkeypatch):
+    monkeypatch.setattr(app, "_find_inventory_host", lambda host: None)
+    monkeypatch.setattr(app, "_ssh", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("SSH must not run")))
+
+    with TestClient(app.app) as client:
+        response = client.get(
+            "/network/wireguard/does-not-exist",
+            headers={"Authorization": "Bearer test-token"},
+        )
+
+    assert response.status_code == 404
+
+
 def test_tts_generate_returns_cloned_wav(monkeypatch):
     captured = {}