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.

**Federation is switched off in the code** (`federation.FEDERATION_ENABLED`),
so every route answers 503 as the hub ships. These tests open it for their own
duration and exercise the protocol underneath, which is what will be wanted the
day it is re-opened — and they are also the reason it is closed. They pass, and
they passed while two hubs could not complete one authenticated request between
them: a second implementation of a peer proves the protocol and nothing about
two machines, which is the same sentence this repo already writes about a
second implementation of the client.

The one test that runs with the gate as it ships is the first one below.
"""

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


@pytest.fixture(autouse=True)
def _federation_open(monkeypatch):
    """Open the gate for the protocol tests, and only for them."""
    from meshbay_hub.api import federation
    monkeypatch.setattr(federation, "FEDERATION_ENABLED", True)


async def test_every_mhp_route_is_closed_as_the_hub_ships(client, monkeypatch):
    """The gate itself, with the flag as it ships.

    Four of these six take no authentication of their own — the MHP token is
    the authentication — and two of them write. A refusal that has to be
    written into each handler is one somebody adds a route without; this is a
    dependency on the router, so a route added later is closed before it is
    written.
    """
    from meshbay_hub.api import federation
    monkeypatch.setattr(federation, "FEDERATION_ENABLED", False)

    for method, path in (("get", "/mhp/info"),
                         ("get", "/mhp/directory"),
                         ("post", "/mhp/directory"),
                         ("post", "/mhp/revoke"),
                         ("get", "/mhp/peers"),
                         ("post", "/mhp/peers")):
        r = await getattr(client, method)(path, **({} if method == "get" else {"json": {}}))
        assert r.status_code == 503, f"{method.upper()} {path}: {r.status_code}"
        assert "not enabled" in r.text


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