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 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_private_key_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__)

# ── Federation is closed ──────────────────────────────────────────────────────
#
# **Every MHP endpoint refuses, and this flag is the only thing that decides it.**
#
# Not because the protocol is wrong, but because nothing has ever run it. Two
# hubs have never completed a single authenticated request between them: until
# 2026-09-12 `_issue_mhp_token` bound the hub's signing key at import — before
# `load_hub_keypair` runs — so this hub signed with `None` and called itself
# `meshbay.org` whatever it was named, while the verifier named no audience for
# the `aud` the issuer sets, which PyJWT refuses outright (**AV14**). Both were
# found by reading, and they were found because a test had written down, in its
# own words, that the real issuer "cannot be used from a test" and had signed
# its own tokens instead. What else is in here of that shape is not known, and
# the way to know is to stand up a second hub — not to leave the door open
# meanwhile.
#
# The surface being closed is worth naming: four of the six routes carry no
# authentication of their own (the MHP token *is* the authentication), two of
# those write — a directory push and a revocation — and every one of them is
# reachable by anyone who can reach the hub.
#
# **A constant and not a setting, deliberately.** A row in `hub_settings` and a
# switch in the admin panel would invite an operator to turn on a feature that
# has never worked between two machines. This takes an edit and a deploy, by
# somebody who has read this. The refusal is a stated 503 rather than a 404
# because a peer hub deserves a reason it can act on, which is §5.6's rule for
# the wire one level up.
#
# **To re-open it:** set this True, stand up a second hub, and run the exchange
# both ways. `test_federation.py` covers the protocol and proves nothing about
# two machines; §15.2 carries federation as not built until that has happened.
FEDERATION_ENABLED = False


def _federation_open() -> None:
    """Refuse every route on this router while federation is closed.

    A router dependency rather than a line in each handler: it covers the six
    routes that exist and every one anybody adds later. A gate you have to
    remember to write is the shape **C6** is the standing lesson about.
    """
    if not FEDERATION_ENABLED:
        raise HTTPException(status_code=503,
                            detail="Federation is not enabled on this hub")


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

# 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. This
# stops a captured POST /mhp/directory or /mhp/revoke from being replayed at
# *this* hub inside the token's short TTL; the `aud` check in
# `_verify_mhp_token` is what stops it being replayed at a different one. GET
# /mhp/directory is idempotent and not covered.
#
# The comment here used to say the sending side sets no `aud` claim, so
# audience binding was unavailable. `_issue_mhp_token` four lines below has
# always set one.
_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.

    Read through the accessors, not from names bound at import: this module
    held `_hub_sk_pem` as it was *before* startup, which is `None`, so this
    function could only ever have raised — and `_hub_id` as `meshbay.org`,
    whatever the instance was actually called.
    """
    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_private_key_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.")

    # `audience` is what makes the `aud` claim mean anything — and omitting it
    # did not merely leave the binding unenforced. PyJWT refuses a token that
    # carries `aud` when the caller names no audience, so **every token this
    # hub issues was rejected by every hub running this code**: MHP could not
    # complete a single authenticated request between two peers. Nothing
    # noticed because `test_federation.py` builds its envelopes by hand,
    # without an `aud`, so the suite exercised a token shape production never
    # emits — the fixture was not the thing.
    decoded = jwt.decode(
        token, peer.pk_hub_pem.encode(),
        algorithms=["EdDSA"],
        audience=hub_id(),
        options={"require": ["exp", "iss", "aud"]},
    )

    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
        ]
    }