diff options
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/api/federation.py | 124 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_federation.py | 179 |
2 files changed, 277 insertions, 26 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/federation.py b/packages/meshbay-hub/src/meshbay_hub/api/federation.py index 7f1262d..b1e0e30 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/federation.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/federation.py @@ -27,7 +27,7 @@ import uuid import jwt from fastapi import APIRouter, Depends, HTTPException, Header from pydantic import BaseModel -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from meshbay_common import MHP_VERSION @@ -41,6 +41,17 @@ 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.""" @@ -57,9 +68,15 @@ def _issue_mhp_token(target_hub_id: str) -> str: async def _verify_mhp_token( - token: str, db: AsyncSession, expected_aud: str | None = None, + token: str, db: AsyncSession, *, single_use: bool = False, ) -> dict: - """Verify a JWT from a peer hub using DB-stored public key.""" + """ + 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") @@ -67,15 +84,22 @@ async def _verify_mhp_token( 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_hub_pem.encode(), algorithms=["EdDSA"], - options=options, + 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 @@ -142,30 +166,50 @@ async def receive_directory( db: AsyncSession = Depends(get_db), ): try: - await _verify_mhp_token(authorization.removeprefix("Bearer "), db) + 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: - 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 + 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=g["id"], - name=g.get("name", ""), - source_hub=body.hub_id, - join_policy=g.get("join_policy", "invite"), - )) + id=gid, name=name, source_hub=sender, join_policy=jp)) 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} + log.info("Persisted %d groups from hub %s", count, sender[:16]) + return {"accepted": count, "from_hub": sender} # ── Revocation propagation ──────────────────────────────────────────────────── @@ -179,15 +223,43 @@ async def receive_revocation( 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: - await _verify_mhp_token(authorization.removeprefix("Bearer "), db) + payload = await _verify_mhp_token( + authorization.removeprefix("Bearer "), db, single_use=True) except Exception as e: raise HTTPException(status_code=401, detail=str(e)) - 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} + 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) ─────────────────────────────────────────────────── diff --git a/packages/meshbay-hub/tests/test_federation.py b/packages/meshbay-hub/tests/test_federation.py new file mode 100644 index 0000000..6035b0c --- /dev/null +++ b/packages/meshbay-hub/tests/test_federation.py @@ -0,0 +1,179 @@ +""" +MHP federation — what a registered peer hub may and may not do. + +A peer is trusted enough to advertise its own public groups into our directory +and to withdraw them. It is not trusted to speak for a third hub, to shadow a +local group, to revoke our users, or to replay a state-changing request. +""" + +import base64 +import hashlib +import time +import uuid + +import jwt +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_hub.api.deps import set_admin_usernames + + +def _auth_key(password: str, username: str) -> str: + salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest() + return base64.b64encode( + hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode() + + +async def _admin(client, username="root"): + pw = "a-long-enough-passphrase" + await client.post("/v1/users/register", json={ + "username": username, "email": f"{username}@example.com", + "auth_key": _auth_key(pw, username)}) + set_admin_usernames([username]) + r = await client.post("/v1/users/login", json={ + "username": username, "auth_key": _auth_key(pw, username)}) + return {"Authorization": f"Bearer {r.json()['access_token']}"} + + +class Peer: + def __init__(self, hub_id: str): + self.hub_id = hub_id + self._sk = Ed25519PrivateKey.generate() + self.pk_pem = self._sk.public_key().public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo).decode() + + def _sk_pem(self) -> bytes: + return self._sk.private_bytes( + serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, + serialization.NoEncryption()) + + def envelope(self, jti: str | None = None) -> str: + now = int(time.time()) + return jwt.encode( + {"iss": self.hub_id, "sub": self.hub_id, + "jti": jti or str(uuid.uuid4()), "iat": now, "exp": now + 300}, + self._sk_pem(), algorithm="EdDSA") + + def revocation(self, target: str, target_id: str) -> str: + return jwt.encode( + {"type": "revocation", "target": target, "target_id": target_id, + "iss": self.hub_id, "iat": int(time.time())}, + self._sk_pem(), algorithm="EdDSA") + + def header(self, **kw) -> dict: + return {"Authorization": f"Bearer {self.envelope(**kw)}"} + + +async def _register_peer(client, admin, peer: Peer): + r = await client.post("/mhp/peers", headers=admin, json={ + "hub_id": peer.hub_id, "hub_url": f"https://{peer.hub_id}", + "pk_hub_pem": peer.pk_pem}) + assert r.status_code == 201, r.text + + +# ── receive_directory ────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_unknown_peer_is_refused(client): + stranger = Peer("nobody.example") + r = await client.post("/mhp/directory", headers=stranger.header(), + json={"hub_id": "nobody.example", "groups": []}) + assert r.status_code == 401 + + +@pytest.mark.asyncio +async def test_source_hub_is_the_signer_not_the_body(client): + admin = await _admin(client) + peer = Peer("peer-a.example") + await _register_peer(client, admin, peer) + + r = await client.post("/mhp/directory", headers=peer.header(), json={ + "hub_id": "peer-b.example", # claims to relay another hub + "groups": [{"id": "g-1", "name": "Shared", "join_policy": "open"}]}) + assert r.status_code == 202 + + listing = (await client.get("/v1/groups")).json()["groups"] + row = next(g for g in listing if g["id"] == "g-1") + assert row["source"] == "peer-a.example" # the signer, not "peer-b.example" + + +@pytest.mark.asyncio +async def test_a_federated_id_cannot_shadow_a_local_group(client): + admin = await _admin(client) + peer = Peer("peer-a.example") + await _register_peer(client, admin, peer) + + owner = await _admin(client, "owner") + r = await client.post("/v1/groups", headers=owner, json={ + "name": "mine", "visibility": "public", "join_policy": "open"}) + local_id = r.json()["group_id"] + + r = await client.post("/mhp/directory", headers=peer.header(), json={ + "hub_id": peer.hub_id, + "groups": [{"id": local_id, "name": "evil twin", "join_policy": "open"}]}) + assert r.status_code == 202 + assert r.json()["accepted"] == 0 + + +@pytest.mark.asyncio +async def test_a_state_changing_token_cannot_be_replayed(client): + admin = await _admin(client) + peer = Peer("peer-a.example") + await _register_peer(client, admin, peer) + + env = peer.envelope(jti="fixed-jti") + h = {"Authorization": f"Bearer {env}"} + body = {"hub_id": peer.hub_id, + "groups": [{"id": "g-9", "name": "Once", "join_policy": "open"}]} + + assert (await client.post("/mhp/directory", headers=h, json=body)).status_code == 202 + assert (await client.post("/mhp/directory", headers=h, json=body)).status_code == 401 + + +# ── receive_revocation ───────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_a_peer_may_withdraw_its_own_group(client): + admin = await _admin(client) + peer = Peer("peer-a.example") + await _register_peer(client, admin, peer) + + await client.post("/mhp/directory", headers=peer.header(), json={ + "hub_id": peer.hub_id, + "groups": [{"id": "g-77", "name": "Bye", "join_policy": "open"}]}) + assert any(g["id"] == "g-77" for g in (await client.get("/v1/groups")).json()["groups"]) + + r = await client.post("/mhp/revoke", headers=peer.header(), + json={"token": peer.revocation("group", "g-77")}) + assert r.status_code == 202 and r.json()["pruned"] == 1 + assert not any(g["id"] == "g-77" for g in (await client.get("/v1/groups")).json()["groups"]) + + +@pytest.mark.asyncio +async def test_a_peer_cannot_withdraw_another_hubs_group(client): + admin = await _admin(client) + a, b = Peer("peer-a.example"), Peer("peer-b.example") + await _register_peer(client, admin, a) + await _register_peer(client, admin, b) + + await client.post("/mhp/directory", headers=a.header(), json={ + "hub_id": a.hub_id, + "groups": [{"id": "g-a", "name": "A's", "join_policy": "open"}]}) + + # b signs a revocation for a's group and presents it under b's envelope. + r = await client.post("/mhp/revoke", headers=b.header(), + json={"token": b.revocation("group", "g-a")}) + assert r.status_code == 202 and r.json()["pruned"] == 0 + assert any(g["id"] == "g-a" for g in (await client.get("/v1/groups")).json()["groups"]) + + +@pytest.mark.asyncio +async def test_federation_cannot_revoke_a_user(client): + admin = await _admin(client) + peer = Peer("peer-a.example") + await _register_peer(client, admin, peer) + + r = await client.post("/mhp/revoke", headers=peer.header(), + json={"token": peer.revocation("user", "some-user-id")}) + assert r.status_code == 202 and r.json()["pruned"] == 0 |