""" 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 asyncio 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, require_admin 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 _node_groups: dict[str, list[str]] = {} # node_id → [group_id, ...] _punch_events: dict[str, asyncio.Event] = {} # node_id → signaling event def get_connected_node_count() -> int: return len(_connected_nodes) def get_online_nodes_for_group(group_id: str) -> list[str]: return [nid for nid, gids in _node_groups.items() if group_id in gids] 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 _handle_chat_notify(group_id: str, sender_name: str, sender_user_id: str) -> None: """Node informs hub that a chat message was posted — create notifications for offline members.""" if not group_id: return try: from meshbay_hub.db.engine import get_session_factory from meshbay_hub.db.models import GroupMember, Group from meshbay_hub.api.notifications import create_notification async with get_session_factory()() as db: group = await db.get(Group, group_id) if not group: return result = await db.execute( select(GroupMember.user_id).where(GroupMember.group_id == group_id) ) member_ids = [r[0] for r in result.all()] for uid in member_ids: if uid == sender_user_id: continue await create_notification( db, uid, "chat_message", f"{sender_name or 'Someone'} posted in {group.name}", link=f"#/group/{group_id}", ) await db.commit() except Exception as e: log.warning("Chat notify failed: %s", e) 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 = msg.get("node_id") or decoded.get("sub", "unknown") _connected_nodes[node_id] = ws group_ids = msg.get("group_ids", []) if group_ids: _node_groups[node_id] = group_ids log.info("Node WS connected: %s (groups=%d)", node_id[:8], len(group_ids)) await ws.send_text(json.dumps({"type": "auth_ok", "node_id": node_id})) # Message loop — handle ping, punch_ready, etc. while True: raw = await ws.receive_text() msg = json.loads(raw) if msg.get("type") == "ping": await ws.send_text(json.dumps({"type": "pong"})) elif msg.get("type") == "punch_ready": event = _punch_events.get(node_id) if event: event.set() elif msg.get("type") == "webrtc_answer": from meshbay_hub.api.signaling import handle_webrtc_answer handle_webrtc_answer(msg) elif msg.get("type") == "chat_notify": asyncio.ensure_future(_handle_chat_notify( msg.get("group_id", ""), msg.get("sender_name", ""), decoded.get("sub", ""), )) 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) _node_groups.pop(node_id, None) # ── Admin revocation endpoint ───────────────────────────────────────────────── class IncomingRequest(BaseModel): peer_ip: str peer_port: int @router.post("/v1/nodes/{node_id}/incoming", status_code=200) async def notify_incoming( node_id: str, body: IncomingRequest, current_user: User = Depends(get_current_user), ): """ Signal a node that a client wants to connect (NAT punch coordination). Hub forwards the request via WebSocket; node punches NAT and replies punch_ready. """ ws = _connected_nodes.get(node_id) if not ws: raise HTTPException(status_code=404, detail="Node not connected") event = asyncio.Event() _punch_events[node_id] = event await ws.send_text(json.dumps({ "type": "client_incoming", "peer_ip": body.peer_ip, "peer_port": body.peer_port, })) try: await asyncio.wait_for(event.wait(), timeout=5.0) except asyncio.TimeoutError: raise HTTPException(status_code=504, detail="Node did not respond in time") finally: _punch_events.pop(node_id, None) return {"status": "ready", "node_id": node_id} 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(require_admin), 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 }