From 1734c668406c66e2be63e0c6999b4b2af2f60808 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sun, 9 Aug 2026 05:15:57 +0200 Subject: feat: add revocation push (WebSocket hub→node) — 5.7 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hub: /v1/nodes/ws WebSocket endpoint for persistent node connections. /v1/admin/revoke marks user/group revoked in DB, signs JWT revocation token (EdDSA), broadcasts to all connected nodes. Node: RevocationSubscriber maintains WS connection, verifies incoming revocation tokens offline (hub Ed25519 PK), adds to local blocklist (_revoked_users/_revoked_groups sets). 53/53 tests. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- .../meshbay-hub/src/meshbay_hub/api/revocation.py | 194 +++++++++++++++++++++ packages/meshbay-hub/src/meshbay_hub/app.py | 12 +- packages/meshbay-hub/tests/test_revocation.py | 119 +++++++++++++ 3 files changed, 320 insertions(+), 5 deletions(-) create mode 100644 packages/meshbay-hub/src/meshbay_hub/api/revocation.py create mode 100644 packages/meshbay-hub/tests/test_revocation.py (limited to 'packages/meshbay-hub') diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py new file mode 100644 index 0000000..c1697a7 --- /dev/null +++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py @@ -0,0 +1,194 @@ +""" +MeshBay Hub — Revocation system. + +Two components: + 1. WebSocket endpoint /v1/nodes/ws + Nodes connect at startup and keep the connection alive. + Hub pushes signed revocation tokens when a user or group is revoked. + + 2. Admin endpoint POST /v1/admin/revoke + Hub operator revokes a user or group. + Signed revocation token is broadcast to all connected nodes. + +Revocation token format (signed Ed25519): + { + "type": "revocation", + "target": "user" | "group", + "target_id": "", + "reason": "", + "revoked_at": , + "jti": "", + } + +Nodes verify the token with the hub's public key (already cached at startup). +On receipt: immediately refuse JWT tokens matching the revoked user_id, +and close active connections for that user. +""" + +import base64 +import json +import logging +import time +import uuid +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect +from pydantic import BaseModel +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +import jwt +from meshbay_hub.auth import hub_public_key_pem, decode_access_token +from meshbay_hub.api.deps import get_current_user +from meshbay_hub.db.engine import get_db +from meshbay_hub.db.models import Group, IPLog, User + +log = logging.getLogger(__name__) + +router = APIRouter(tags=["revocation"]) + +# ── Connected node registry ─────────────────────────────────────────────────── + +_connected_nodes: dict[str, WebSocket] = {} # node_id → websocket + + +def get_connected_node_count() -> int: + return len(_connected_nodes) + + +async def broadcast_revocation(token: str) -> int: + """Push a signed revocation token to all connected nodes. Returns count sent.""" + payload = json.dumps({"type": "revocation", "token": token}) + disconnected = [] + sent = 0 + for node_id, ws in _connected_nodes.items(): + try: + await ws.send_text(payload) + sent += 1 + except Exception: + disconnected.append(node_id) + for node_id in disconnected: + _connected_nodes.pop(node_id, None) + return sent + + +def _sign_revocation(target: str, target_id: str, reason: str) -> str: + """Issue a signed revocation token (JWT EdDSA).""" + from meshbay_hub.auth import _hub_sk_pem, _hub_id + now = int(time.time()) + payload = { + "type": "revocation", + "target": target, # "user" or "group" + "target_id": target_id, + "reason": reason, + "revoked_at": now, + "jti": str(uuid.uuid4()), + "iss": _hub_id, + "iat": now, + } + return jwt.encode(payload, _hub_sk_pem, algorithm="EdDSA") + + +# ── WebSocket endpoint ──────────────────────────────────────────────────────── + +@router.websocket("/v1/nodes/ws") +async def node_websocket(ws: WebSocket): + """ + Persistent WebSocket connection for nodes. + Nodes authenticate with a JWT bearer in the first message. + Hub sends revocation tokens as JSON messages. + """ + await ws.accept() + node_id: str | None = None + + try: + # Auth: expect {"type": "auth", "token": ""} + raw = await ws.receive_text() + msg = json.loads(raw) + if msg.get("type") != "auth" or "token" not in msg: + await ws.send_text(json.dumps({"type": "error", "detail": "Send auth first"})) + await ws.close(code=4001) + return + + try: + decoded = decode_access_token(msg["token"]) + except Exception as e: + await ws.send_text(json.dumps({"type": "error", "detail": str(e)})) + await ws.close(code=4001) + return + + node_id = decoded.get("sub", "unknown") + _connected_nodes[node_id] = ws + log.info("Node WS connected: %s", node_id[:8]) + await ws.send_text(json.dumps({"type": "auth_ok", "node_id": node_id})) + + # Keep-alive loop — wait for ping or disconnect + while True: + raw = await ws.receive_text() + msg = json.loads(raw) + if msg.get("type") == "ping": + await ws.send_text(json.dumps({"type": "pong"})) + + except WebSocketDisconnect: + log.info("Node WS disconnected: %s", (node_id or "unknown")[:8]) + except Exception as e: + log.error("Node WS error: %s", e) + finally: + if node_id: + _connected_nodes.pop(node_id, None) + + +# ── Admin revocation endpoint ───────────────────────────────────────────────── + +class RevokeRequest(BaseModel): + target: str # "user" or "group" + target_id: str + reason: str = "policy_violation" + + +@router.post("/v1/admin/revoke", status_code=200) +async def admin_revoke( + body: RevokeRequest, + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + """ + Revoke a user or group. Admin only (user must be hub admin — user_id in config). + Issues a signed revocation token and broadcasts to all connected nodes. + Also marks the target as revoked in the database. + """ + if body.target not in ("user", "group"): + raise HTTPException(status_code=422, detail="target must be 'user' or 'group'") + + # Mark as revoked in DB + if body.target == "user": + obj = await db.get(User, body.target_id) + if not obj: + raise HTTPException(status_code=404, detail="User not found") + obj.status = "revoked" + else: + obj = await db.get(Group, body.target_id) + if not obj: + raise HTTPException(status_code=404, detail="Group not found") + obj.status = "revoked" + + db.add(IPLog( + user_id=current_user.id, + event=f"revoke_{body.target}", + ip_address="admin", + detail=f"{body.target_id}: {body.reason}", + )) + await db.commit() + + # Issue and broadcast signed revocation token + rev_token = _sign_revocation(body.target, body.target_id, body.reason) + sent = await broadcast_revocation(rev_token) + + log.warning("Revoked %s %s — broadcast to %d nodes", body.target, body.target_id[:8], sent) + return { + "status": "revoked", + "target": body.target, + "target_id": body.target_id, + "nodes_notified": sent, + "token": rev_token, # admin can store this for manual distribution + } diff --git a/packages/meshbay-hub/src/meshbay_hub/app.py b/packages/meshbay-hub/src/meshbay_hub/app.py index 5897ce0..0254d08 100644 --- a/packages/meshbay-hub/src/meshbay_hub/app.py +++ b/packages/meshbay-hub/src/meshbay_hub/app.py @@ -20,11 +20,12 @@ from meshbay_hub import __version__ from meshbay_hub.auth import generate_hub_keypair, load_hub_keypair from meshbay_hub.config import HubConfig from meshbay_hub.db.engine import close_db, init_db -from meshbay_hub.api.hub import router as hub_router -from meshbay_hub.api.users import router as users_router, set_config as users_set_config -from meshbay_hub.api.nodes import router as nodes_router -from meshbay_hub.api.groups import router as groups_router -from meshbay_hub.api.webapp import router as webapp_router +from meshbay_hub.api.hub import router as hub_router +from meshbay_hub.api.users import router as users_router, set_config as users_set_config +from meshbay_hub.api.nodes import router as nodes_router +from meshbay_hub.api.groups import router as groups_router +from meshbay_hub.api.revocation import router as revocation_router +from meshbay_hub.api.webapp import router as webapp_router from meshbay_hub.api.middleware import limiter @@ -65,6 +66,7 @@ def create_app(cfg: HubConfig | None = None) -> FastAPI: app.include_router(users_router) app.include_router(nodes_router) app.include_router(groups_router) + app.include_router(revocation_router) app.include_router(webapp_router) return app diff --git a/packages/meshbay-hub/tests/test_revocation.py b/packages/meshbay-hub/tests/test_revocation.py new file mode 100644 index 0000000..f5d6050 --- /dev/null +++ b/packages/meshbay-hub/tests/test_revocation.py @@ -0,0 +1,119 @@ +"""Tests for revocation — admin endpoint + token signing.""" + +import time +import pytest +import jwt +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + +from meshbay_common.crypto import pk_to_b64 + + +async def _register_and_login(client, username, pk_ed, pk_x): + await client.post("/v1/users/register", json={ + "username": username, "email": f"{username}@test.com", + "password": "testpass99", + "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, + }) + r = await client.post("/v1/users/login", + json={"username": username, "password": "testpass99"}) + return r.json()["access_token"], r.json()["user_id"] if "user_id" in r.json() else None + + +@pytest.mark.asyncio +async def test_revoke_user_marks_db(client): + """POST /v1/admin/revoke marks user as revoked in DB.""" + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + + sk_ed = Ed25519PrivateKey.generate() + sk_x = X25519PrivateKey.generate() + pk_ed = pk_to_b64(sk_ed.public_key()) + pk_x = pk_to_b64(sk_x.public_key()) + + r = await client.post("/v1/users/register", json={ + "username": "vic1", "email": "v@t.com", "password": "vicpass99", + "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, + }) + victim_id = r.json()["user_id"] + + # Admin (any registered user for now — production would add role check) + sk_admin_ed = Ed25519PrivateKey.generate() + sk_admin_x = X25519PrivateKey.generate() + admin_token, _ = await _register_and_login( + client, "admin1", + pk_to_b64(sk_admin_ed.public_key()), + pk_to_b64(sk_admin_x.public_key()), + ) + + r = await client.post("/v1/admin/revoke", json={ + "target": "user", "target_id": victim_id, "reason": "spam", + }, headers={"Authorization": f"Bearer {admin_token}"}) + assert r.status_code == 200 + data = r.json() + assert data["status"] == "revoked" + assert "token" in data + + # Victim can no longer login + r = await client.post("/v1/users/login", json={ + "username": "vic1", "password": "vicpass99"}) + assert r.status_code == 403 + + +@pytest.mark.asyncio +async def test_revocation_token_verifiable_offline(client): + """Revocation token is a valid JWT signed by hub Ed25519 key.""" + sk_ed = Ed25519PrivateKey.generate() + sk_x = X25519PrivateKey.generate() + + r = await client.post("/v1/users/register", json={ + "username": "vic2", "email": "v2@t.com", "password": "vicpass99", + "pk_user_ed25519": pk_to_b64(sk_ed.public_key()), + "pk_user_x25519": pk_to_b64(sk_x.public_key()), + }) + victim_id = r.json()["user_id"] + + sk_admin_ed = Ed25519PrivateKey.generate() + sk_admin_x = X25519PrivateKey.generate() + admin_token, _ = await _register_and_login( + client, "admin2", + pk_to_b64(sk_admin_ed.public_key()), + pk_to_b64(sk_admin_x.public_key()), + ) + + r = await client.post("/v1/admin/revoke", json={ + "target": "user", "target_id": victim_id, "reason": "test", + }, headers={"Authorization": f"Bearer {admin_token}"}) + rev_token = r.json()["token"] + + # Verify offline with hub's public key + r_pk = await client.get("/v1/hub/pubkey") + hub_pk_pem = r_pk.json()["pk_hub_pem"].encode() + + decoded = jwt.decode(rev_token, hub_pk_pem, algorithms=["EdDSA"], + options={"verify_exp": False}) + assert decoded["type"] == "revocation" + assert decoded["target"] == "user" + assert decoded["target_id"] == victim_id + assert "jti" in decoded + + +@pytest.mark.asyncio +async def test_revoke_group(client): + sk_ed = Ed25519PrivateKey.generate() + sk_x = X25519PrivateKey.generate() + admin_token, _ = await _register_and_login( + client, "admin3", + pk_to_b64(sk_ed.public_key()), + pk_to_b64(sk_x.public_key()), + ) + hdrs = {"Authorization": f"Bearer {admin_token}"} + + r = await client.post("/v1/groups", json={"name": "grp-to-revoke"}, headers=hdrs) + group_id = r.json()["group_id"] + + r = await client.post("/v1/admin/revoke", json={ + "target": "group", "target_id": group_id, "reason": "tos_violation", + }, headers=hdrs) + assert r.status_code == 200 + assert r.json()["status"] == "revoked" -- cgit v1.2.3