Merge branch 'main' into feat/live-ops-wall
This commit is contained in:
@@ -312,7 +312,8 @@ async def delete_node_client(node_id: str) -> None:
|
||||
async def reconcile_all() -> None:
|
||||
"""
|
||||
Rebuild dynsec state for every approved node from Firestore
|
||||
(node_keys is the source of truth). Called once at c2-core startup,
|
||||
(node_keys is the source of truth), and delete the login of every
|
||||
rejected node. Called once at c2-core startup,
|
||||
after ensure_roles_and_c2core_grant(). Self-heals a lost/corrupted
|
||||
dynamic-security.json (e.g. volume wiped, or a prior approve/reissue's
|
||||
dynsec publish silently failed to persist for some other reason) —
|
||||
@@ -320,6 +321,20 @@ async def reconcile_all() -> None:
|
||||
dynsec store would lock out every previously-approved node until
|
||||
someone noticed and manually re-approved each one.
|
||||
"""
|
||||
# server-26#195: revoke first. A node rejected before reject_node()
|
||||
# revoked anything (or whose revoke hit a dynsec failure) still has a
|
||||
# broker login in dynamic-security.json; startup is where that heals.
|
||||
# delete_node_client() treats "not found" as success, so this is a no-op
|
||||
# for every node already revoked.
|
||||
for node in await fstore.collection_list("nodes", approval_status="rejected"):
|
||||
node_id = node.get("node_id")
|
||||
if not node_id:
|
||||
continue
|
||||
try:
|
||||
await delete_node_client(node_id)
|
||||
except DynsecError as e:
|
||||
logger.error(f"dynsec reconcile: failed to revoke rejected node {node_id!r}: {e}")
|
||||
|
||||
nodes = await fstore.collection_list("nodes", approval_status="approved")
|
||||
if not nodes:
|
||||
return
|
||||
|
||||
@@ -119,11 +119,36 @@ async def delete_node(node_id: str, _: dict = Depends(require_admin_token)):
|
||||
|
||||
@router.post("/{node_id}/reject")
|
||||
async def reject_node(node_id: str, _: dict = Depends(require_admin_token)):
|
||||
"""Reject = revoke (server-26#195). Before this, reject only flipped
|
||||
``approval_status``, so an approved-then-rejected node kept its MQTT login
|
||||
(dynsec client), its ``/upload`` key and ``/systems`` read access (both
|
||||
keyed off ``node_keys``).
|
||||
|
||||
Order is fail-closed, the opposite of approve's: the status flips to
|
||||
``rejected`` FIRST, so even if the broker call fails, ``/upload`` (which
|
||||
now checks approval_status) and ``GET /nodes/{id}/credentials`` already
|
||||
refuse the node, and ``dynsec.reconcile_all()`` will no longer re-create
|
||||
its login on the next c2-core restart (it also deletes rejected nodes'
|
||||
logins there). Then the broker login is deleted, then the key. A dynsec
|
||||
failure returns 502 with node_keys left intact, and the admin retries:
|
||||
every step is idempotent. Re-approving later mints a fresh key, because
|
||||
node_keys is gone; the revoked key is never handed back.
|
||||
"""
|
||||
node = await fstore.doc_get("nodes", node_id)
|
||||
if not node:
|
||||
raise HTTPException(404, f"Node '{node_id}' not found.")
|
||||
await fstore.doc_update("nodes", node_id, {"approval_status": "rejected"})
|
||||
return {"ok": True}
|
||||
try:
|
||||
await dynsec.delete_node_client(node_id)
|
||||
except dynsec.DynsecError as e:
|
||||
logger.error(f"Reject {node_id!r}: dynsec deleteClient failed, node_keys NOT deleted: {e}")
|
||||
raise HTTPException(
|
||||
502,
|
||||
f"Node marked rejected, but its MQTT login could not be revoked: {e}. Retry reject.",
|
||||
)
|
||||
await fstore.doc_delete("node_keys", node_id)
|
||||
logger.info(f"Reject {node_id!r}: status rejected, MQTT login and API key revoked")
|
||||
return {"ok": True, "revoked": True}
|
||||
|
||||
|
||||
@router.post("/{node_id}/command")
|
||||
|
||||
@@ -47,6 +47,18 @@ async def upload_call_audio(
|
||||
f"(received prefix: {credentials.credentials[:8]}...)"
|
||||
)
|
||||
raise HTTPException(401, "Invalid node API key")
|
||||
# server-26#195: a valid key is not enough, the node must still be
|
||||
# approved. reject_node() deletes node_keys too, but this closes the
|
||||
# window where that delete failed or was never run (a node rejected before
|
||||
# reject revoked anything). Uncached read: revocation must bite on the
|
||||
# next upload, not after doc_get_cached's 5-minute TTL.
|
||||
node = await fstore.doc_get("nodes", node_id)
|
||||
if not node or node.get("approval_status") != "approved":
|
||||
logger.warning(
|
||||
f"Upload 403: node_id={node_id!r} has a valid key but approval_status="
|
||||
f"{(node or {}).get('approval_status')!r}"
|
||||
)
|
||||
raise HTTPException(403, "Node is not approved")
|
||||
|
||||
data = await file.read()
|
||||
if not data:
|
||||
@@ -63,10 +75,10 @@ async def upload_call_audio(
|
||||
# org_id is stamped defensively here too (not just in
|
||||
# mqtt_handler.py's call_start/call_end): key_doc above proves this
|
||||
# node_id is real and authenticated, so resolving org_id from the
|
||||
# node doc here covers a call whose Firestore doc was somehow
|
||||
# never written by call_start (the upload is otherwise the
|
||||
# authoritative record of which node this audio came from).
|
||||
node = await fstore.doc_get_cached("nodes", node_id)
|
||||
# node doc (fetched by the approval gate above) here covers a call
|
||||
# whose Firestore doc was somehow never written by call_start (the
|
||||
# upload is otherwise the authoritative record of which node this
|
||||
# audio came from).
|
||||
updates = {"audio_gcs_uri": gcs_uri}
|
||||
if node and node.get("org_id"):
|
||||
updates["org_id"] = node["org_id"]
|
||||
|
||||
@@ -0,0 +1,203 @@
|
||||
"""
|
||||
server-26#195 — reject must revoke: delete the node's dynsec (MQTT) login and
|
||||
its node_keys entry, /upload must refuse a node that is not approved, and
|
||||
dynsec.reconcile_all() must delete rejected nodes' logins on startup.
|
||||
"""
|
||||
import asyncio
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
from app.internal import dynsec
|
||||
from app.internal.auth import require_admin_token, require_service_or_firebase_token
|
||||
from app.routers import nodes, upload
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
|
||||
def setup_function():
|
||||
app.dependency_overrides[require_admin_token] = lambda: {"admin": True}
|
||||
app.dependency_overrides[require_service_or_firebase_token] = lambda: {"admin": True}
|
||||
|
||||
|
||||
def teardown_function():
|
||||
app.dependency_overrides.pop(require_admin_token, None)
|
||||
app.dependency_overrides.pop(require_service_or_firebase_token, None)
|
||||
|
||||
|
||||
def _fake_store(node, key_doc):
|
||||
store = {"nodes": node, "node_keys": key_doc}
|
||||
calls = []
|
||||
|
||||
async def doc_get(c, i):
|
||||
return store.get(c)
|
||||
|
||||
async def doc_update(c, i, data):
|
||||
calls.append(("doc_update", c, data))
|
||||
store[c] = {**(store[c] or {}), **data}
|
||||
|
||||
async def doc_delete(c, i):
|
||||
calls.append(("doc_delete", c))
|
||||
store[c] = None
|
||||
|
||||
async def doc_set(c, i, data, merge=False):
|
||||
store[c] = data
|
||||
|
||||
return store, calls, doc_get, doc_update, doc_delete, doc_set
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# reject_node
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_reject_revokes_mqtt_login_and_key_in_fail_closed_order():
|
||||
store, calls, doc_get, doc_update, doc_delete, _ = _fake_store(
|
||||
{"node_id": "n1", "approval_status": "approved"}, {"node_id": "n1", "api_key": "live-key"})
|
||||
|
||||
async def delete_client(node_id):
|
||||
calls.append(("dynsec_delete", node_id))
|
||||
|
||||
with patch.object(nodes.fstore, "doc_get", doc_get), \
|
||||
patch.object(nodes.fstore, "doc_update", doc_update), \
|
||||
patch.object(nodes.fstore, "doc_delete", doc_delete), \
|
||||
patch.object(nodes.dynsec, "delete_node_client", delete_client):
|
||||
resp = client.post("/nodes/n1/reject")
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == {"ok": True, "revoked": True}
|
||||
assert calls == [
|
||||
("doc_update", "nodes", {"approval_status": "rejected"}),
|
||||
("dynsec_delete", "n1"),
|
||||
("doc_delete", "node_keys"),
|
||||
]
|
||||
assert store["node_keys"] is None
|
||||
assert store["nodes"]["approval_status"] == "rejected"
|
||||
|
||||
|
||||
def test_reject_dynsec_failure_is_502_and_keeps_key_but_status_is_already_rejected():
|
||||
store, calls, doc_get, doc_update, doc_delete, _ = _fake_store(
|
||||
{"node_id": "n1", "approval_status": "approved"}, {"node_id": "n1", "api_key": "live-key"})
|
||||
|
||||
with patch.object(nodes.fstore, "doc_get", doc_get), \
|
||||
patch.object(nodes.fstore, "doc_update", doc_update), \
|
||||
patch.object(nodes.fstore, "doc_delete", doc_delete), \
|
||||
patch.object(nodes.dynsec, "delete_node_client",
|
||||
AsyncMock(side_effect=dynsec.DynsecError("timeout"))):
|
||||
resp = client.post("/nodes/n1/reject")
|
||||
|
||||
assert resp.status_code == 502
|
||||
assert "Retry reject" in resp.json()["detail"]
|
||||
assert store["nodes"]["approval_status"] == "rejected"
|
||||
assert store["node_keys"] == {"node_id": "n1", "api_key": "live-key"}
|
||||
assert ("doc_delete", "node_keys") not in calls
|
||||
|
||||
|
||||
def test_reject_unknown_node_is_404_and_touches_nothing():
|
||||
with patch.object(nodes.fstore, "doc_get", AsyncMock(return_value=None)), \
|
||||
patch.object(nodes.fstore, "doc_update", AsyncMock()) as doc_update, \
|
||||
patch.object(nodes.fstore, "doc_delete", AsyncMock()) as doc_delete, \
|
||||
patch.object(nodes.dynsec, "delete_node_client", AsyncMock()) as delete_client:
|
||||
resp = client.post("/nodes/ghost/reject")
|
||||
assert resp.status_code == 404
|
||||
doc_update.assert_not_awaited()
|
||||
doc_delete.assert_not_awaited()
|
||||
delete_client.assert_not_awaited()
|
||||
|
||||
|
||||
def test_reject_requires_admin():
|
||||
app.dependency_overrides.pop(require_admin_token, None)
|
||||
resp = client.post("/nodes/n1/reject")
|
||||
assert resp.status_code in (401, 403)
|
||||
|
||||
|
||||
def test_reapprove_after_reject_mints_a_new_key_never_the_revoked_one():
|
||||
store, _, doc_get, doc_update, doc_delete, doc_set = _fake_store(
|
||||
{"node_id": "n1", "approval_status": "approved"}, {"node_id": "n1", "api_key": "old-key"})
|
||||
|
||||
with patch.object(nodes.fstore, "doc_get", doc_get), \
|
||||
patch.object(nodes.fstore, "doc_update", doc_update), \
|
||||
patch.object(nodes.fstore, "doc_delete", doc_delete), \
|
||||
patch.object(nodes.fstore, "doc_set", doc_set), \
|
||||
patch.object(nodes.dynsec, "delete_node_client", AsyncMock()), \
|
||||
patch.object(nodes.dynsec, "upsert_node_client", AsyncMock()) as upsert, \
|
||||
patch.object(nodes.mqtt_handler, "publish_node_key", MagicMock()):
|
||||
assert client.post("/nodes/n1/reject").status_code == 200
|
||||
assert client.post("/nodes/n1/approve").status_code == 200
|
||||
|
||||
new_key = store["node_keys"]["api_key"]
|
||||
assert new_key != "old-key" and len(new_key) == 64
|
||||
upsert.assert_awaited_once_with("n1", new_key)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# /upload approval gate
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _upload(node_doc, key="live-key"):
|
||||
store = {"node_keys": {"node_id": "n1", "api_key": "live-key"}, "nodes": node_doc}
|
||||
|
||||
async def doc_get(c, i):
|
||||
return store.get(c)
|
||||
|
||||
with patch.object(upload.fstore, "doc_get", doc_get), \
|
||||
patch.object(upload, "upload_audio", AsyncMock(return_value=None)) as up:
|
||||
resp = client.post(
|
||||
"/upload",
|
||||
data={"call_id": "c1", "node_id": "n1"},
|
||||
files={"file": ("a.mp3", b"\x00\x01", "audio/mpeg")},
|
||||
headers={"Authorization": f"Bearer {key}"},
|
||||
)
|
||||
return resp, up
|
||||
|
||||
|
||||
def test_upload_refuses_valid_key_from_rejected_node():
|
||||
resp, up = _upload({"node_id": "n1", "approval_status": "rejected"})
|
||||
assert resp.status_code == 403
|
||||
up.assert_not_awaited()
|
||||
|
||||
|
||||
def test_upload_refuses_valid_key_from_pending_or_missing_node():
|
||||
for node_doc in ({"node_id": "n1", "approval_status": "pending"}, {"node_id": "n1"}, None):
|
||||
resp, up = _upload(node_doc)
|
||||
assert resp.status_code == 403, node_doc
|
||||
up.assert_not_awaited()
|
||||
|
||||
|
||||
def test_upload_accepts_approved_node():
|
||||
resp, up = _upload({"node_id": "n1", "approval_status": "approved"})
|
||||
assert resp.status_code == 200
|
||||
up.assert_awaited_once()
|
||||
|
||||
|
||||
def test_upload_wrong_key_still_401_before_approval_check():
|
||||
resp, up = _upload({"node_id": "n1", "approval_status": "approved"}, key="wrong")
|
||||
assert resp.status_code == 401
|
||||
up.assert_not_awaited()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# dynsec.reconcile_all
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_reconcile_deletes_rejected_logins_and_still_syncs_approved():
|
||||
async def collection_list(c, **filters):
|
||||
status = filters.get("approval_status")
|
||||
return {
|
||||
"rejected": [{"node_id": "bad1"}, {"node_id": "bad2"}],
|
||||
"approved": [{"node_id": "good"}],
|
||||
}[status]
|
||||
|
||||
async def doc_get(c, i):
|
||||
return {"node_id": i, "api_key": "k-" + i}
|
||||
|
||||
delete_client = AsyncMock(side_effect=[dynsec.DynsecError("boom"), None])
|
||||
with patch.object(dynsec.fstore, "collection_list", collection_list), \
|
||||
patch.object(dynsec.fstore, "doc_get", doc_get), \
|
||||
patch.object(dynsec, "delete_node_client", delete_client), \
|
||||
patch.object(dynsec, "upsert_node_client", AsyncMock()) as upsert:
|
||||
asyncio.run(dynsec.reconcile_all())
|
||||
|
||||
# one failed revoke does not stop the rest, nor the approved sync
|
||||
assert [c.args[0] for c in delete_client.await_args_list] == ["bad1", "bad2"]
|
||||
upsert.assert_awaited_once_with("good", "k-good")
|
||||
@@ -197,7 +197,7 @@ export default function NodeDetailPage() {
|
||||
}
|
||||
|
||||
async function handleReject() {
|
||||
if (!confirm("Reject this node? It will not be able to upload recordings.")) return;
|
||||
if (!confirm("Reject this node? Any MQTT login and API key it holds are revoked, so it cannot connect or upload. Re-approving later issues a new key.")) return;
|
||||
setApproving(true);
|
||||
try {
|
||||
await c2api.rejectNode(id);
|
||||
|
||||
Reference in New Issue
Block a user