mirror of
https://github.com/CyberMind-FR/secubox-deb.git
synced 2026-07-29 18:36:55 +00:00
feat(toolbox): API /rlevel admin + peer self-service (tunnel-auth, ctl delegation)
Adds GET/POST /rlevel/peers, /rlevel/peer/{pubkey}, /rlevel/me to the toolbox
FastAPI router. Admin routes list every wg-toolbox peer's rlevel status
(chosen/forced/floor/effective + best-effort live handshake) and set
floor/forced. Peer self-service resolves identity from the wg tunnel source
IP (_client_ip -> wg-peers.json), 403s on an unknown IP, and 409s if the
requested chosen mode would drop below the peer's floor or lift an admin
forced mode. Every write delegates to `sudo -n sbxmitm-policyctl`
(set-floor/force/set-chosen) -- no in-process root, mirrors the proxypac
API's _ctl pattern.
Co-Authored-By: Gerald KERMA <devel@cybermind.fr>
This commit is contained in:
parent
e8a6058488
commit
add0fda3f6
|
|
@ -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."""
|
||||
|
|
|
|||
228
packages/secubox-toolbox/tests/test_rlevel_api.py
Normal file
228
packages/secubox-toolbox/tests/test_rlevel_api.py
Normal file
|
|
@ -0,0 +1,228 @@
|
|||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gérald Kerma <devel@cybermind.fr>
|
||||
"""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
|
||||
Loading…
Reference in New Issue
Block a user