path: root/packages/meshbay-hub/src/meshbay_hub/api/revocation.py

"""
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":  "<user_id or group_id>",
    "reason":     "<reason string>",
    "revoked_at": <unix timestamp>,
    "jti":        "<uuid4>",
  }

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 ────────────────────────────────────────────────────────

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)


@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": "<jwt>"}
        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
    }