path: root/packages/meshbay-hub/tests/test_federation.py

"""
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, audience: str | None = None) -> str:
        """The shape `federation._issue_mhp_token` actually emits.

        This used to omit `aud`, and so tested a token production never sends:
        the verifier names an audience, and PyJWT refuses a token carrying
        `aud` when the caller names none — which meant MHP could not complete
        one authenticated request between two real hubs while this file stayed
        green. A fixture narrower than the real thing tests the fixture.
        """
        # Read from the module rather than written down here: the receiving
        # hub's id is what `aud` has to carry, and a copy of it in a test is a
        # second place for it to drift from.
        from meshbay_hub.auth import _hub_id

        now = int(time.time())
        return jwt.encode(
            {"iss": self.hub_id, "sub": self.hub_id, "aud": audience or _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


# ── The hub's own identity, and the audience it binds ────────────────────────
#
# MHP could not complete one authenticated request between two real hubs, and
# nothing said so: `_issue_mhp_token` sets `aud`, `_verify_mhp_token` named no
# audience, and PyJWT refuses a token carrying `aud` when the caller names
# none. This file stayed green because its envelopes were built by hand
# without one — a fixture narrower than what production emits tests the
# fixture. Underneath that, `federation.py` bound `_hub_id` and `_hub_sk_pem`
# at import, which is *before* `load_hub_keypair` runs: it signed with `None`
# and called itself `meshbay.org` whatever the instance was named.


def test_the_hub_issues_tokens_under_its_configured_identity(app):
    """Driven through the real issuer, which is the half no test called."""
    from meshbay_hub.api.federation import _issue_mhp_token
    from meshbay_hub.auth import hub_id

    assert hub_id() == "test-hub", "the fixture hub is not configured"

    token = jwt.decode(_issue_mhp_token("peer-hub"),
                       options={"verify_signature": False})
    assert token["iss"] == "test-hub", (
        "the hub announced an identity that is not its own")
    assert token["aud"] == "peer-hub"


@pytest.mark.asyncio
async def test_a_token_for_another_hub_is_refused(client):
    """What the audience binding is for: a token captured here must not be
    replayable at a peer, and one minted for a peer must not work here."""
    admin = await _admin(client)
    peer = Peer("peer-elsewhere.test")
    await _register_peer(client, admin, peer)

    r = await client.post(
        "/mhp/directory",
        headers={"Authorization":
                 f"Bearer {peer.envelope(audience='some-other-hub.test')}"},
        json={"hub_id": peer.hub_id, "groups": []})
    assert r.status_code == 401, r.text

    r = await client.post(
        "/mhp/directory", headers=peer.header(),
        json={"hub_id": peer.hub_id, "groups": []})
    assert r.status_code == 202, r.text