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

"""
MeshBay Hub — MHP (Mesh Bay Hub Protocol) federation endpoints.

Federation allows multiple hubs to exchange public group directories
and propagate revocations. Each hub explicitly chooses its peers
(no automatic discovery).

MHP endpoints:
  GET  /mhp/info          — hub identity and capabilities
  POST /mhp/directory     — receive a Mesh Directory from a peer hub
  POST /mhp/revoke        — receive a revocation from a peer hub
  GET  /mhp/directory     — export our public Mesh Directory
  POST /mhp/peers         — admin: register a trusted peer hub
  GET  /mhp/peers         — admin: list registered peers

Peer authentication: each request carries a JWT signed by the
sending hub's Ed25519 key. Receiving hub verifies with the sender's
cached public key (registered when adding the peer).

Protocol version: MHP 0.1
"""

import logging
import time
import uuid

import jwt
from fastapi import APIRouter, Depends, HTTPException, Header
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from meshbay_common import MHP_VERSION
from meshbay_hub import __version__
from meshbay_hub.api.deps import get_current_user
from meshbay_hub.auth import _hub_id, _hub_sk_pem, hub_public_key_pem
from meshbay_hub.db.engine import get_db
from meshbay_hub.db.models import Group, User
from sqlalchemy.ext.asyncio import AsyncSession

log = logging.getLogger(__name__)

router = APIRouter(prefix="/mhp", tags=["federation"])

# ── In-memory peer registry ───────────────────────────────────────────────────
# Production: move to DB table (peers: hub_id, hub_url, pk_hub_pem, trusted_since)

_peers: dict[str, dict] = {}   # hub_id → {url, pk_pem, trusted_since}


def _issue_mhp_token(target_hub_id: str) -> str:
    """Issue a short-lived JWT for authenticating to a peer hub."""
    now = int(time.time())
    return jwt.encode({
        "iss": _hub_id,
        "sub": _hub_id,
        "aud": target_hub_id,
        "jti": str(uuid.uuid4()),
        "iat": now,
        "exp": now + 300,   # 5 minute window
        "mhp": MHP_VERSION,
    }, _hub_sk_pem, algorithm="EdDSA")


def _verify_mhp_token(token: str, expected_aud: str | None = None) -> dict:
    """Verify a JWT from a peer hub."""
    # First decode without verification to get iss (sender hub_id)
    unverified = jwt.decode(token, options={"verify_signature": False})
    sender_id = unverified.get("iss")

    peer = _peers.get(sender_id)
    if not peer:
        raise PermissionError(f"Unknown hub: {sender_id!r}. Register as peer first.")

    options = {}
    if expected_aud:
        options["audience"] = expected_aud

    decoded = jwt.decode(
        token, peer["pk_pem"].encode(),
        algorithms=["EdDSA"],
        options=options,
    )
    return decoded


# ── Hub identity ──────────────────────────────────────────────────────────────

@router.get("/info")
async def mhp_info():
    """Return this hub's identity for peer registration."""
    return {
        "hub_id":      _hub_id,
        "mhp_version": MHP_VERSION,
        "hub_version": __version__,
        "pk_hub_pem":  hub_public_key_pem().decode(),
    }


# ── Directory exchange ────────────────────────────────────────────────────────

@router.get("/directory")
async def export_directory(
    db: AsyncSession = Depends(get_db),
    authorization: str = Header(...),
):
    """
    Export our public Mesh Directory to a peer hub.
    Auth: Bearer JWT signed by peer hub's key.
    """
    try:
        _verify_mhp_token(authorization.removeprefix("Bearer "))
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))

    result = await db.execute(
        select(Group).where(Group.visibility == "public", Group.status == "active"))
    groups = result.scalars().all()

    return {
        "hub_id":      _hub_id,
        "mhp_version": MHP_VERSION,
        "groups": [
            {
                "id":          g.id,
                "name":        g.name,
                "join_policy": g.join_policy,
                "created_at":  g.created_at.isoformat(),
                "hub_id":      _hub_id,
            }
            for g in groups
        ],
    }


class DirectoryPayload(BaseModel):
    hub_id:  str
    groups:  list[dict]


@router.post("/directory", status_code=202)
async def receive_directory(
    body: DirectoryPayload,
    authorization: str = Header(...),
    db: AsyncSession = Depends(get_db),
):
    """
    Receive a Mesh Directory update from a peer hub.
    Persists groups to federated_groups table for cross-hub search.
    """
    try:
        _verify_mhp_token(authorization.removeprefix("Bearer "))
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))

    from meshbay_hub.db.models import FederatedGroup
    from datetime import datetime, timezone
    now = datetime.now(timezone.utc)
    count = 0
    for g in body.groups:
        existing = await db.get(FederatedGroup, g["id"])
        if existing:
            existing.name        = g.get("name", existing.name)
            existing.join_policy = g.get("join_policy", existing.join_policy)
            existing.updated_at  = now
        else:
            db.add(FederatedGroup(
                id=g["id"],
                name=g.get("name", ""),
                source_hub=body.hub_id,
                join_policy=g.get("join_policy", "invite"),
            ))
        count += 1
    await db.commit()
    log.info("Persisted %d groups from hub %s", count, body.hub_id[:16])
    return {"accepted": count, "from_hub": body.hub_id}


# ── Revocation propagation ────────────────────────────────────────────────────

class RevocationPayload(BaseModel):
    token: str   # signed revocation JWT from originating hub


@router.post("/revoke", status_code=202)
async def receive_revocation(
    body: RevocationPayload,
    authorization: str = Header(...),
):
    """
    Receive a revocation from a peer hub. Verify and propagate to our nodes.
    """
    try:
        _verify_mhp_token(authorization.removeprefix("Bearer "))
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))

    # The revocation token is signed by the ORIGINATING hub's key (not the relaying hub)
    # For now: re-broadcast to our connected nodes
    from meshbay_hub.api.revocation import broadcast_revocation
    sent = await broadcast_revocation(body.token)
    log.info("Propagated revocation to %d local nodes", sent)
    return {"propagated_to": sent}


# ── Peer management (admin) ───────────────────────────────────────────────────

class PeerRegisterRequest(BaseModel):
    hub_id:    str
    hub_url:   str
    pk_hub_pem: str   # peer hub's Ed25519 public key PEM


@router.post("/peers", status_code=201)
async def register_peer(
    body: PeerRegisterRequest,
    current_user: User = Depends(get_current_user),
):
    """Admin: register a trusted peer hub. Manual step — no auto-discovery."""
    _peers[body.hub_id] = {
        "url":          body.hub_url,
        "pk_pem":       body.pk_hub_pem,
        "trusted_since": int(time.time()),
    }
    log.info("Peer registered: %s (%s)", body.hub_id, body.hub_url)
    return {"status": "registered", "hub_id": body.hub_id}


@router.get("/peers")
async def list_peers(current_user: User = Depends(get_current_user)):
    """Admin: list registered peer hubs."""
    return {
        "peers": [
            {"hub_id": hid, "url": p["url"], "trusted_since": p["trusted_since"]}
            for hid, p in _peers.items()
        ]
    }