""" 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 func, select from sqlalchemy.ext.asyncio import AsyncSession from meshbay_common import MHP_VERSION from meshbay_hub import __version__, hub_settings from meshbay_hub.api.deps import require_admin 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 FederatedGroup, Group, HubPeer, User log = logging.getLogger(__name__) router = APIRouter(prefix="/mhp", tags=["federation"]) # One push may not dump the world, and one peer may not fill the table. MAX_FEDERATED_GROUPS_PER_PUSH = 500 MAX_FEDERATED_GROUPS_PER_PEER = 2000 # Seen `jti` values for the state-changing MHP endpoints, pruned lazily. The # sending side that would set an `aud` claim is unbuilt, so audience binding is # not available; this stops a captured POST /mhp/directory or /mhp/revoke from # being replayed inside the token's short TTL. GET /mhp/directory is idempotent # and not covered. _seen_mhp_jti: dict[str, float] = {} 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, "mhp": MHP_VERSION, }, _hub_sk_pem, algorithm="EdDSA") async def _verify_mhp_token( token: str, db: AsyncSession, *, single_use: bool = False, ) -> dict: """ Verify a JWT from a peer hub against its DB-stored public key and return the payload. `single_use=True` (the state-changing endpoints) additionally rejects a replayed `jti` within the token's lifetime. """ unverified = jwt.decode(token, options={"verify_signature": False}) sender_id = unverified.get("iss") peer = await db.get(HubPeer, sender_id) if not peer: raise PermissionError(f"Unknown hub: {sender_id!r}. Register as peer first.") decoded = jwt.decode( token, peer.pk_hub_pem.encode(), algorithms=["EdDSA"], options={"require": ["exp", "iss"]}, ) if single_use: now = time.time() for j, exp in list(_seen_mhp_jti.items()): if exp < now: _seen_mhp_jti.pop(j, None) jti = decoded.get("jti", "") if not jti or jti in _seen_mhp_jti: raise PermissionError("MHP token replay") _seen_mhp_jti[jti] = float(decoded.get("exp", now + 300)) 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(...), ): try: await _verify_mhp_token(authorization.removeprefix("Bearer "), db) except Exception as e: raise HTTPException(status_code=401, detail=str(e)) # A hub with public groups switched off advertises nothing to its peers — # the local directory is empty (groups.list_public_groups), and the exported # one has to match or peers keep showing groups this hub no longer serves. if not await hub_settings.public_groups_allowed(db): groups = [] else: 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), ): try: payload = await _verify_mhp_token( authorization.removeprefix("Bearer "), db, single_use=True) except Exception as e: raise HTTPException(status_code=401, detail=str(e)) # `source_hub` is the signer of this request, never `body.hub_id` — a peer # does not get to relay or spoof a third hub's groups into our directory. sender = payload["iss"] if len(body.groups) > MAX_FEDERATED_GROUPS_PER_PUSH: raise HTTPException(status_code=413, detail="Too many groups in one push") from datetime import datetime, timezone now = datetime.now(timezone.utc) have = await db.scalar( select(func.count()).select_from(FederatedGroup) .where(FederatedGroup.source_hub == sender)) or 0 count = 0 for g in body.groups: gid = str(g.get("id", ""))[:36] name = str(g.get("name", ""))[:128] jp = g.get("join_policy", "invite") if not gid or jp not in ("invite", "open"): continue # A federated id must never shadow a real local group. if await db.get(Group, gid): log.warning("Federated id %s collides with a local group — skipped", gid[:8]) continue row = await db.get(FederatedGroup, gid) if row: if row.source_hub != sender: continue # only the hub that advertised it may update it row.name = name or row.name row.join_policy = jp row.updated_at = now else: if have + count >= MAX_FEDERATED_GROUPS_PER_PEER: break db.add(FederatedGroup( id=gid, name=name, source_hub=sender, join_policy=jp)) count += 1 await db.commit() log.info("Persisted %d groups from hub %s", count, sender[:16]) return {"accepted": count, "from_hub": sender} # ── Revocation propagation ──────────────────────────────────────────────────── class RevocationPayload(BaseModel): token: str @router.post("/revoke", status_code=202) async def receive_revocation( body: RevocationPayload, authorization: str = Header(...), db: AsyncSession = Depends(get_db), ): """ Act on a revocation from a peer hub. This does **not** reach local nodes: nothing here hosts a federated group, and a local node would reject a token signed by another hub's key anyway (that path was a silent no-op). What a peer may legitimately revoke is a group *it advertised to us* — so this prunes our copy of the peer's directory. Users are per-hub; a peer does not get to revoke ours. """ try: payload = await _verify_mhp_token( authorization.removeprefix("Bearer "), db, single_use=True) except Exception as e: raise HTTPException(status_code=401, detail=str(e)) sender = payload["iss"] peer = await db.get(HubPeer, sender) try: inner = jwt.decode( body.token, peer.pk_hub_pem.encode(), algorithms=["EdDSA"], options={"verify_exp": False}) except Exception as e: raise HTTPException(status_code=400, detail=f"Bad revocation token: {e}") if inner.get("type") != "revocation" or inner.get("target") != "group": return {"pruned": 0, "note": "federation may only revoke groups it advertised"} target_id = inner.get("target_id", "") row = await db.get(FederatedGroup, target_id) pruned = 0 if row and row.source_hub == sender: await db.delete(row) await db.commit() pruned = 1 log.info("Federated group %s revoked by %s (pruned=%d)", target_id[:8], sender[:16], pruned) return {"pruned": pruned} # ── Peer management (admin) ─────────────────────────────────────────────────── class PeerRegisterRequest(BaseModel): hub_id: str hub_url: str pk_hub_pem: str @router.post("/peers", status_code=201) async def register_peer( body: PeerRegisterRequest, current_user: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): """Admin: register a trusted peer hub.""" existing = await db.get(HubPeer, body.hub_id) if existing: existing.hub_url = body.hub_url existing.pk_hub_pem = body.pk_hub_pem else: db.add(HubPeer( hub_id=body.hub_id, hub_url=body.hub_url, pk_hub_pem=body.pk_hub_pem, )) await db.commit() 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(require_admin), db: AsyncSession = Depends(get_db), ): """Admin: list registered peer hubs.""" result = await db.execute(select(HubPeer)) peers = result.scalars().all() return { "peers": [ { "hub_id": p.hub_id, "url": p.hub_url, "trusted_since": p.trusted_since.isoformat(), } for p in peers ] }