diff --git a/packages/secubox-toolbox/secubox_toolbox/api.py b/packages/secubox-toolbox/secubox_toolbox/api.py index c520a031..de3874c6 100644 --- a/packages/secubox-toolbox/secubox_toolbox/api.py +++ b/packages/secubox-toolbox/secubox_toolbox/api.py @@ -3980,6 +3980,248 @@ async def admin_sentinel_c2_allow(host: str = Form(...)) -> dict: return {"ok": sentinel_link.c2_allow(host)} +# ───────────────── /rlevel — per-peer wg-toolbox MITM R-level (#rlevel-per-peer, task 6) ───────────────── +# +# Modes, increasing intrusion: off(0) < passive(1) < active(2) < reel(3). +# Effective = forced ?? clamp(chosen, floor, reel) — mirrors cmd/sbxmitm/rlevel.go +# effective() and sbxmitm-policyctl's bash re-implementation, kept in lockstep +# here so the admin/self-service list can compute it WITHOUT asking the ctl a +# second time. No in-process root: every mutation is delegated to +# `sudo -n /usr/sbin/sbxmitm-policyctl` (mirrors the proxypac API's `_ctl` +# pattern) — this module only ever READS peer-rlevel.json (via `ctl list`) +# and wg-peers.json. +_RLEVEL_MODES = ("off", "passive", "active", "reel") +_RLEVEL_RANK = {m: i for i, m in enumerate(_RLEVEL_MODES)} +_RLEVEL_POLICYCTL = "/usr/sbin/sbxmitm-policyctl" +_RLEVEL_WG_PEERS = Path("/var/lib/secubox/toolbox/wg-peers.json") + + +def _rlevel_valid_mode(mode: str) -> bool: + return mode in _RLEVEL_RANK + + +def _rlevel_clamp(value: str, floor: str, ceiling: str = "reel") -> str: + r = _RLEVEL_RANK.get(value, _RLEVEL_RANK["passive"]) + lo = _RLEVEL_RANK.get(floor, _RLEVEL_RANK["passive"]) + hi = _RLEVEL_RANK.get(ceiling, _RLEVEL_RANK["reel"]) + r = max(lo, min(hi, r)) + return _RLEVEL_MODES[r] + + +def _rlevel_effective(chosen: str, forced: str | None, floor: str) -> str: + if forced: + return forced + return _rlevel_clamp(chosen, floor) + + +def _rlevel_ctl(args: list[str], timeout: int = 15): + """Delegate a privileged rlevel action to sbxmitm-policyctl via `sudo -n`. + Never raises — returns (returncode, stdout, stderr); a transport failure + (sudo missing, timeout, ctl absent) yields rc=1 so callers fail closed.""" + import subprocess + try: + p = subprocess.run( + ["sudo", "-n", _RLEVEL_POLICYCTL, *args], + capture_output=True, text=True, timeout=timeout, + ) + return p.returncode, p.stdout.strip(), p.stderr.strip() + except (OSError, subprocess.SubprocessError) as e: + return 1, "", str(e)[:200] + + +def _rlevel_load_wg_peers() -> dict: + """{pubkey: {ip, label, ...}} from wg-peers.json — best-effort, {} on error.""" + import json as _json + try: + return (_json.loads(_RLEVEL_WG_PEERS.read_text()).get("peers") or {}) + except Exception: + return {} + + +def _rlevel_load_policy() -> dict: + """Whole peer-rlevel.json document via `sbxmitm-policyctl list` (read-only — + the ctl owns the file; this module never touches it directly).""" + import json as _json + rc, out, _err = _rlevel_ctl(["list"]) + if rc != 0 or not out: + return {"defaults": {"mode": "passive", "floor": "passive"}, "peers": {}} + try: + doc = _json.loads(out) + except Exception: + return {"defaults": {"mode": "passive", "floor": "passive"}, "peers": {}} + doc.setdefault("defaults", {"mode": "passive", "floor": "passive"}) + doc.setdefault("peers", {}) + return doc + + +def _rlevel_wg_live(pubkeys: set) -> dict: + """Best-effort {pubkey: bool} — True if wg-toolbox reports a handshake in + the last 180s for that peer. Never raises; unavailable wg → {} (UI treats + missing as unknown, never as a false 'live').""" + import subprocess + live: dict = {} + if not pubkeys: + return live + try: + out = subprocess.run( + ["wg", "show", "wg-toolbox", "dump"], + capture_output=True, text=True, timeout=2, check=False, + ).stdout + except Exception: + return live + now = time.time() + for line in out.splitlines()[1:]: # skip the interface header line + cols = line.split("\t") + if len(cols) < 5: + continue + pk = cols[0] + if pk not in pubkeys: + continue + try: + hs = int(cols[4]) + except (ValueError, IndexError): + hs = 0 + live[pk] = bool(hs) and (now - hs) < 180 + return live + + +def _rlevel_peer_entry(doc: dict, pubkey: str) -> tuple[str, str | None, str]: + """(chosen, forced, floor) for pubkey, falling back to doc['defaults'].""" + defaults = doc.get("defaults") or {} + def_mode = defaults.get("mode", "passive") + def_floor = defaults.get("floor", "passive") + p = (doc.get("peers") or {}).get(pubkey) or {} + chosen = p.get("chosen") or def_mode + forced = p.get("forced") + floor = p.get("floor") or def_floor + return chosen, forced, floor + + +def _rlevel_peer_for_ip(ip: str | None) -> str | None: + """Resolve a wg-toolbox tunnel source IP to its pubkey — the self-service + auth identity. None if the IP is not a known wg-toolbox peer.""" + if not ip: + return None + for pk, meta in _rlevel_load_wg_peers().items(): + if meta.get("ip") == ip: + return pk + return None + + +@router.get("/rlevel/peers") +async def rlevel_peers() -> dict: + """Admin — every wg-toolbox peer's rlevel status (chosen/forced/floor/ + effective + best-effort live handshake). Read-only: sourced from + wg-peers.json (identity) + `sbxmitm-policyctl list` (policy).""" + wg_peers = _rlevel_load_wg_peers() + doc = _rlevel_load_policy() + live = _rlevel_wg_live(set(wg_peers.keys())) + peers = [] + for pk, meta in wg_peers.items(): + chosen, forced, floor = _rlevel_peer_entry(doc, pk) + peers.append({ + "pubkey": pk, + "label": meta.get("label"), + "chosen": chosen, + "forced": forced, + "floor": floor, + "effective": _rlevel_effective(chosen, forced, floor), + "live": live.get(pk), + }) + return {"peers": peers, "defaults": doc.get("defaults")} + + +@router.post("/rlevel/peer/{pubkey}") +async def rlevel_peer_set(pubkey: str, request: Request) -> dict: + """Admin — set a peer's floor and/or forced mode. Body: {"floor"?: mode, + "forced"?: mode|null}. Every write is delegated to sbxmitm-policyctl + (no in-process root).""" + try: + body = await request.json() + except Exception: + body = {} + if not isinstance(body, dict): + body = {} + floor = body.get("floor") + forced_present = "forced" in body + forced = body.get("forced") + + if floor is not None and not _rlevel_valid_mode(floor): + raise HTTPException(status_code=400, detail=f"invalid floor mode: {floor}") + if forced_present and forced is not None and not _rlevel_valid_mode(forced): + raise HTTPException(status_code=400, detail=f"invalid forced mode: {forced}") + + results = {} + if floor is not None: + rc, out, err = _rlevel_ctl(["set-floor", pubkey, floor]) + if rc != 0: + raise HTTPException(status_code=502, detail=err or "set-floor failed") + results["floor"] = floor + if forced_present: + mode = forced if forced is not None else "none" + rc, out, err = _rlevel_ctl(["force", pubkey, mode]) + if rc != 0: + raise HTTPException(status_code=502, detail=err or "force failed") + results["forced"] = forced + + if not results: + raise HTTPException(status_code=400, detail="nothing to set (expected floor and/or forced)") + return {"ok": True, "pubkey": pubkey, **results} + + +@router.get("/rlevel/me") +async def rlevel_me(request: Request) -> dict: + """Peer self-service — identity is the wg-toolbox tunnel source IP itself + (10.99.1.x), resolved to a pubkey via wg-peers.json. 403 if the source IP + is not a known wg-toolbox peer (tunnel-auth, no token/JWT involved).""" + ip = _client_ip(request) + pubkey = _rlevel_peer_for_ip(ip) + if not pubkey: + raise HTTPException(status_code=403, detail="source IP is not a known wg-toolbox peer") + doc = _rlevel_load_policy() + chosen, forced, floor = _rlevel_peer_entry(doc, pubkey) + return { + "chosen": chosen, + "forced": forced, + "floor": floor, + "effective": _rlevel_effective(chosen, forced, floor), + } + + +@router.post("/rlevel/me") +async def rlevel_me_set(request: Request) -> dict: + """Peer self-service — set own `chosen` mode, bounded: a peer can never + descend below its own floor nor lift a `forced` (409 either way). Delegates + to `sbxmitm-policyctl set-chosen` (which re-clamps server-side too).""" + ip = _client_ip(request) + pubkey = _rlevel_peer_for_ip(ip) + if not pubkey: + raise HTTPException(status_code=403, detail="source IP is not a known wg-toolbox peer") + + try: + body = await request.json() + except Exception: + body = {} + if not isinstance(body, dict): + body = {} + chosen = body.get("chosen") + if not chosen or not _rlevel_valid_mode(chosen): + raise HTTPException(status_code=400, detail=f"invalid chosen mode: {chosen}") + + doc = _rlevel_load_policy() + _cur_chosen, forced, floor = _rlevel_peer_entry(doc, pubkey) + if forced: + raise HTTPException(status_code=409, detail=f"mode is forced to '{forced}' by admin") + if _RLEVEL_RANK[chosen] < _RLEVEL_RANK[floor]: + raise HTTPException(status_code=409, detail=f"below floor '{floor}'") + + rc, out, err = _rlevel_ctl(["set-chosen", pubkey, chosen]) + if rc != 0: + raise HTTPException(status_code=502, detail=err or "set-chosen failed") + return {"ok": True, "chosen": chosen, "floor": floor, + "effective": _rlevel_effective(chosen, None, floor)} + + @router.get("/admin/filters/ui", response_class=HTMLResponse) async def admin_filters_ui() -> HTMLResponse: """#566 — minimal filter toggle panel for the toolbox WebUI.""" diff --git a/packages/secubox-toolbox/tests/test_rlevel_api.py b/packages/secubox-toolbox/tests/test_rlevel_api.py new file mode 100644 index 00000000..b3b168b0 --- /dev/null +++ b/packages/secubox-toolbox/tests/test_rlevel_api.py @@ -0,0 +1,228 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +"""API /rlevel — admin peer list + floor/force delegation, peer self-service +bounded by tunnel-IP identity (#rlevel-per-peer, task 6). + +No in-process root: every write goes through `_rlevel_ctl` which shells out +to `sudo -n sbxmitm-policyctl` (task 4). Tests stub `_rlevel_ctl` directly for +behavioral assertions, and stub `subprocess.run` once to prove the real +delegation command line is well-formed. +""" +import json + +from fastapi.testclient import TestClient + +from secubox_toolbox.app import app +from secubox_toolbox import api + +client = TestClient(app) + +PK1 = "PK1pubkeyBASE64aaaaaaaaaaaaaaaaaaaaaaaaaaaa=" +PK2 = "PK2pubkeyBASE64bbbbbbbbbbbbbbbbbbbbbbbbbbbb=" + + +def _write_wg_peers(tmp_path, monkeypatch, peers: dict): + p = tmp_path / "wg-peers.json" + p.write_text(json.dumps({"peers": peers})) + monkeypatch.setattr(api, "_RLEVEL_WG_PEERS", p) + return p + + +def _stub_ctl(monkeypatch, doc=None, rc=0): + """Stub _rlevel_ctl for 'list' calls; capture invocations.""" + calls = [] + doc = doc if doc is not None else {"defaults": {"mode": "passive", "floor": "passive"}, + "peers": {}} + + def fake(args, timeout=15): + calls.append(args) + if args and args[0] == "list": + return rc, json.dumps(doc), "" + return rc, "", "" if rc == 0 else "ctl error" + + monkeypatch.setattr(api, "_rlevel_ctl", fake) + return calls + + +# ── admin: GET /rlevel/peers ──────────────────────────────────────────── + +def test_admin_list_peers_computes_effective(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, { + PK1: {"ip": "10.99.1.5", "label": "phone-A"}, + PK2: {"ip": "10.99.1.6", "label": "phone-B"}, + }) + doc = { + "defaults": {"mode": "passive", "floor": "passive"}, + "peers": { + PK1: {"chosen": "active", "forced": None, "floor": "passive"}, + PK2: {"forced": "off", "chosen": "reel", "floor": "passive"}, + }, + } + _stub_ctl(monkeypatch, doc=doc) + + r = client.get("/rlevel/peers") + assert r.status_code == 200 + body = r.json() + by_pk = {p["pubkey"]: p for p in body["peers"]} + assert by_pk[PK1]["label"] == "phone-A" + assert by_pk[PK1]["effective"] == "active" # clamp(active, passive, reel) + assert by_pk[PK2]["effective"] == "off" # forced wins + assert by_pk[PK1]["floor"] == "passive" + + +def test_admin_list_peers_unknown_peer_falls_back_to_defaults(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {PK1: {"ip": "10.99.1.5", "label": "new"}}) + _stub_ctl(monkeypatch, doc={"defaults": {"mode": "active", "floor": "passive"}, "peers": {}}) + r = client.get("/rlevel/peers") + body = r.json() + assert body["peers"][0]["chosen"] == "active" + assert body["peers"][0]["effective"] == "active" + + +# ── admin: POST /rlevel/peer/{pubkey} ─────────────────────────────────── + +def test_admin_set_floor_delegates_to_real_ctl_command(monkeypatch): + """Structural: the sudo command line actually invokes sbxmitm-policyctl.""" + captured = {} + + class FakeProc: + returncode = 0 + stdout = "" + stderr = "" + + def fake_run(cmd, **kw): + captured["cmd"] = cmd + return FakeProc() + + monkeypatch.setattr("subprocess.run", fake_run) + r = client.post(f"/rlevel/peer/{PK1}", json={"floor": "active"}) + assert r.status_code == 200, r.text + assert "sudo" in captured["cmd"] + assert "-n" in captured["cmd"] + assert any("sbxmitm-policyctl" in c for c in captured["cmd"]) + assert "set-floor" in captured["cmd"] + assert PK1 in captured["cmd"] + assert "active" in captured["cmd"] + + +def test_admin_set_floor_and_force(monkeypatch): + calls = _stub_ctl(monkeypatch) + r = client.post(f"/rlevel/peer/{PK1}", json={"floor": "active", "forced": "reel"}) + assert r.status_code == 200 + assert ["set-floor", PK1, "active"] in calls + assert ["force", PK1, "reel"] in calls + + +def test_admin_force_none_clears(monkeypatch): + calls = _stub_ctl(monkeypatch) + r = client.post(f"/rlevel/peer/{PK1}", json={"forced": None}) + assert r.status_code == 200 + assert ["force", PK1, "none"] in calls + + +def test_admin_set_invalid_floor_mode_400(monkeypatch): + _stub_ctl(monkeypatch) + r = client.post(f"/rlevel/peer/{PK1}", json={"floor": "bogus"}) + assert r.status_code == 400 + + +def test_admin_set_empty_body_400(monkeypatch): + _stub_ctl(monkeypatch) + r = client.post(f"/rlevel/peer/{PK1}", json={}) + assert r.status_code == 400 + + +def test_admin_ctl_failure_surfaces_502(monkeypatch): + _stub_ctl(monkeypatch, rc=1) + r = client.post(f"/rlevel/peer/{PK1}", json={"floor": "active"}) + assert r.status_code == 502 + + +# ── peer: GET /rlevel/me ───────────────────────────────────────────────── + +def test_me_resolves_via_tunnel_ip(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {PK1: {"ip": "10.99.1.5", "label": "me"}}) + _stub_ctl(monkeypatch, doc={ + "defaults": {"mode": "passive", "floor": "passive"}, + "peers": {PK1: {"chosen": "active", "forced": None, "floor": "passive"}}, + }) + r = client.get("/rlevel/me", headers={"X-R3-Peer": "10.99.1.5"}) + assert r.status_code == 200 + body = r.json() + assert body["chosen"] == "active" + assert body["effective"] == "active" + assert body["forced"] is None + + +def test_me_unknown_ip_403(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {PK1: {"ip": "10.99.1.5", "label": "me"}}) + _stub_ctl(monkeypatch) + r = client.get("/rlevel/me", headers={"X-R3-Peer": "10.99.1.99"}) + assert r.status_code == 403 + + +def test_me_no_ip_at_all_403(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {}) + _stub_ctl(monkeypatch) + r = client.get("/rlevel/me") + assert r.status_code == 403 + + +# ── peer: POST /rlevel/me ──────────────────────────────────────────────── + +def test_post_me_valid_delegates_set_chosen(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {PK1: {"ip": "10.99.1.5", "label": "me"}}) + calls = _stub_ctl(monkeypatch, doc={ + "defaults": {"mode": "passive", "floor": "passive"}, + "peers": {PK1: {"chosen": "passive", "forced": None, "floor": "passive"}}, + }) + r = client.post("/rlevel/me", headers={"X-R3-Peer": "10.99.1.5"}, json={"chosen": "active"}) + assert r.status_code == 200, r.text + assert ["set-chosen", PK1, "active"] in calls + + +def test_post_me_below_floor_409(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {PK1: {"ip": "10.99.1.5", "label": "me"}}) + _stub_ctl(monkeypatch, doc={ + "defaults": {"mode": "passive", "floor": "passive"}, + "peers": {PK1: {"chosen": "active", "forced": None, "floor": "active"}}, + }) + r = client.post("/rlevel/me", headers={"X-R3-Peer": "10.99.1.5"}, json={"chosen": "off"}) + assert r.status_code == 409 + + +def test_post_me_cannot_lift_forced_409(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {PK1: {"ip": "10.99.1.5", "label": "me"}}) + _stub_ctl(monkeypatch, doc={ + "defaults": {"mode": "passive", "floor": "passive"}, + "peers": {PK1: {"chosen": "passive", "forced": "off", "floor": "passive"}}, + }) + r = client.post("/rlevel/me", headers={"X-R3-Peer": "10.99.1.5"}, json={"chosen": "reel"}) + assert r.status_code == 409 + + +def test_post_me_unknown_ip_403(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {PK1: {"ip": "10.99.1.5", "label": "me"}}) + _stub_ctl(monkeypatch) + r = client.post("/rlevel/me", headers={"X-R3-Peer": "10.99.1.77"}, json={"chosen": "reel"}) + assert r.status_code == 403 + + +def test_post_me_invalid_mode_400(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {PK1: {"ip": "10.99.1.5", "label": "me"}}) + _stub_ctl(monkeypatch, doc={ + "defaults": {"mode": "passive", "floor": "passive"}, + "peers": {PK1: {"chosen": "passive", "forced": None, "floor": "passive"}}, + }) + r = client.post("/rlevel/me", headers={"X-R3-Peer": "10.99.1.5"}, json={"chosen": "bogus"}) + assert r.status_code == 400 + + +def test_post_me_ctl_failure_surfaces_502(tmp_path, monkeypatch): + _write_wg_peers(tmp_path, monkeypatch, {PK1: {"ip": "10.99.1.5", "label": "me"}}) + _stub_ctl(monkeypatch, doc={ + "defaults": {"mode": "passive", "floor": "passive"}, + "peers": {PK1: {"chosen": "passive", "forced": None, "floor": "passive"}}, + }, rc=1) + r = client.post("/rlevel/me", headers={"X-R3-Peer": "10.99.1.5"}, json={"chosen": "active"}) + assert r.status_code == 502