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

"""
Phase 11.5 security regression tests — node WebSocket registration (finding C2).

The hub relays every WebRTC offer for a node to whoever holds that node's entry in
`_connected_nodes`. That registration used to be established from a client-supplied
`node_id` with no ownership check, so any registered user could take over a victim
node's signaling and become the endpoint browsers connect to.

These exercise `_authorize_node_ws` directly rather than through a socket: it is the
function that makes the authorization decision, and the hub test harness uses
ASGITransport, which has no WebSocket support.
"""

import base64

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey

from meshbay_common.crypto import pk_to_b64


async def _make_user(client, username: str) -> dict:
    """Register + log in a user, returning ids, token and keys."""
    sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate()
    pk_ed, pk_x = pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key())

    r = await client.post("/v1/users/register", json={
        "username": username,
        "email": f"{username}@example.test",
        "auth_key": base64.b64encode(b"k" * 32).decode(),
        "pk_user_ed25519": pk_ed,
        "pk_user_x25519": pk_x,
    })
    assert r.status_code == 201, r.text
    user_id = r.json()["user_id"]

    r = await client.post("/v1/users/login", json={
        "username": username,
        "auth_key": base64.b64encode(b"k" * 32).decode(),
    })
    assert r.status_code == 200, r.text
    return {"user_id": user_id, "token": r.json()["access_token"],
            "pk_ed": pk_ed, "sk_ed": sk_ed}


async def _announce_node(client, user: dict) -> str:
    # Announce now requires proof of possession of the node key (M8).
    import time as _t
    ts = int(_t.time())
    msg = f"meshbay:node_announce:{user['user_id']}:{user['pk_ed']}:{ts}".encode()
    r = await client.post(
        "/v1/nodes/announce",
        json={
            "pk_node": user["pk_ed"], "endpoint_hint": "test",
            "timestamp": ts,
            "signature": base64.b64encode(user["sk_ed"].sign(msg)).decode(),
        },
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert r.status_code == 201, r.text
    return r.json()["node_id"]


def _node_token(user: dict) -> str:
    from meshbay_hub.auth import issue_access_token
    return issue_access_token(user["user_id"], scope="node")


@pytest.mark.asyncio
async def test_ws_rejects_user_scoped_token(client):
    """C2: a browser token must never be able to register as a node."""
    from meshbay_hub.api.revocation import _authorize_node_ws

    victim = await _make_user(client, "victim1_test")
    node_id = await _announce_node(client, victim)

    resolved, detail = await _authorize_node_ws(victim["token"], node_id, None)
    assert resolved is None
    assert "node-scoped" in detail.lower()


@pytest.mark.asyncio
async def test_ws_rejects_foreign_node_id(client):
    """
    C2: the impersonation itself. An attacker with a perfectly valid node-scoped
    token of their own must not be able to claim someone else's node_id.
    """
    from meshbay_hub.api.revocation import _authorize_node_ws

    victim = await _make_user(client, "victim2_test")
    attacker = await _make_user(client, "attacker2")
    victim_node = await _announce_node(client, victim)
    await _announce_node(client, attacker)

    resolved, detail = await _authorize_node_ws(
        _node_token(attacker), victim_node, None)
    assert resolved is None, "attacker hijacked the victim's node registration (C2)"
    assert "does not belong" in detail.lower()


@pytest.mark.asyncio
async def test_ws_rejects_unknown_node_id(client):
    """C2: an invented node_id must not register either."""
    from meshbay_hub.api.revocation import _authorize_node_ws

    user = await _make_user(client, "user3_test")
    resolved, _ = await _authorize_node_ws(_node_token(user), "no-such-node", None)
    assert resolved is None


@pytest.mark.asyncio
async def test_ws_rejects_missing_node_id(client):
    """C2: identity may not fall back to the token subject."""
    from meshbay_hub.api.revocation import _authorize_node_ws

    user = await _make_user(client, "user4_test")
    resolved, _ = await _authorize_node_ws(_node_token(user), "", None)
    assert resolved is None


@pytest.mark.asyncio
async def test_ws_accepts_own_node(client):
    """The legitimate path still works."""
    from meshbay_hub.api.revocation import _authorize_node_ws

    user = await _make_user(client, "owner5_test")
    node_id = await _announce_node(client, user)

    resolved, groups = await _authorize_node_ws(_node_token(user), node_id, None)
    assert resolved == node_id
    assert groups == []


@pytest.mark.asyncio
async def test_ws_group_claims_cannot_widen_beyond_membership(client):
    """
    C2: `group_ids` used to be taken verbatim, letting a node advertise itself as
    an online source for any group on the hub and attract clients to it.
    """
    from meshbay_hub.api.revocation import _authorize_node_ws

    user = await _make_user(client, "owner6_test")
    node_id = await _announce_node(client, user)

    r = await client.post(
        "/v1/groups",
        json={"name": "mine", "visibility": "private", "join_policy": "invite"},
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert r.status_code == 201, r.text
    own_group = r.json()["group_id"]

    resolved, groups = await _authorize_node_ws(
        _node_token(user), node_id, [own_group, "someone-elses-group"])

    assert resolved == node_id
    assert groups == [own_group], "node advertised a group it is not a member of"


@pytest.mark.asyncio
async def test_signaling_rejects_non_member(client):
    """
    H6/H4: POST /v1/nodes/{id}/webrtc/offer was reachable by any authenticated
    user for any node, with no membership check and no rate limit. Each call makes
    the target node allocate an aiortc PeerConnection and gather ICE, so it was a
    remote resource-exhaustion primitive against a third party's machine.
    """
    from meshbay_hub.api import revocation as rev

    owner = await _make_user(client, "owner8_test")
    outsider = await _make_user(client, "outsider8")
    node_id = await _announce_node(client, owner)

    r = await client.post(
        "/v1/groups",
        json={"name": "private-g", "visibility": "private", "join_policy": "invite"},
        headers={"Authorization": f"Bearer {owner['token']}"},
    )
    group_id = r.json()["group_id"]

    # Pretend the node is connected and hosting that group.
    class _FakeWS:
        async def send_text(self, _):
            raise AssertionError("offer relayed to node despite non-membership")

    rev._connected_nodes[node_id] = _FakeWS()
    rev._node_groups[node_id] = [group_id]
    try:
        resp = await client.post(
            f"/v1/nodes/{node_id}/webrtc/offer",
            json={"sdp": "v=0", "ice_candidates": []},
            headers={"Authorization": f"Bearer {outsider['token']}"},
        )
        assert resp.status_code == 403, resp.text
    finally:
        rev._connected_nodes.pop(node_id, None)
        rev._node_groups.pop(node_id, None)


@pytest.mark.asyncio
async def test_signaling_rejects_oversized_sdp(client):
    """H6: an SDP offer is ~2 KB; unbounded input is a memory amplifier."""
    user = await _make_user(client, "user9_test")
    resp = await client.post(
        "/v1/nodes/whatever/webrtc/offer",
        json={"sdp": "v=0" + ("x" * 200_000), "ice_candidates": []},
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert resp.status_code == 413


@pytest.mark.asyncio
async def test_incoming_rejects_foreign_peer_ip(client):
    """
    H6: peer_ip was taken verbatim, letting any user make an arbitrary node emit
    UDP packets to an address of their choosing — reflection via someone else's
    machine. The probe target must be the caller's own address.
    """
    from meshbay_hub.api import revocation as rev

    owner = await _make_user(client, "owner10_test")
    node_id = await _announce_node(client, owner)

    class _FakeWS:
        async def send_text(self, _):
            raise AssertionError("punch relayed with attacker-chosen peer_ip")

    rev._connected_nodes[node_id] = _FakeWS()
    try:
        resp = await client.post(
            f"/v1/nodes/{node_id}/incoming",
            json={"peer_ip": "198.51.100.7", "peer_port": 9999},
            headers={"Authorization": f"Bearer {owner['token']}"},
        )
        assert resp.status_code == 403, resp.text
    finally:
        rev._connected_nodes.pop(node_id, None)


@pytest.mark.asyncio
async def test_ws_node_may_narrow_its_group_set(client):
    """A node hosting a subset of the operator's groups may say so."""
    from meshbay_hub.api.revocation import _authorize_node_ws

    user = await _make_user(client, "owner7_test")
    node_id = await _announce_node(client, user)

    created = []
    for name in ("g-one", "g-two"):
        r = await client.post(
            "/v1/groups",
            json={"name": name, "visibility": "private", "join_policy": "invite"},
            headers={"Authorization": f"Bearer {user['token']}"},
        )
        created.append(r.json()["group_id"])

    resolved, groups = await _authorize_node_ws(
        _node_token(user), node_id, [created[0]])
    assert resolved == node_id
    assert groups == [created[0]]


# ── M8: announce proof of possession ─────────────────────────────────────────

def _announce_payload(user_id: str, sk, pk_b64: str, ts: int | None = None):
    import time as _t
    ts = ts if ts is not None else int(_t.time())
    msg = f"meshbay:node_announce:{user_id}:{pk_b64}:{ts}".encode()
    return {
        "pk_node": pk_b64,
        "endpoint_hint": "test",
        "timestamp": ts,
        "signature": base64.b64encode(sk.sign(msg)).decode(),
    }


@pytest.mark.asyncio
async def test_announce_requires_proof_of_possession(client):
    """
    M8: /v1/nodes/announce accepted any pk_node with no proof the announcer held
    the private key, so a user could announce a record carrying someone else's
    node key.
    """
    user = await _make_user(client, "ann1_test")
    r = await client.post(
        "/v1/nodes/announce",
        json={"pk_node": user["pk_ed"], "endpoint_hint": "test"},
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert r.status_code == 400, r.text


@pytest.mark.asyncio
async def test_announce_rejects_foreign_key(client):
    """M8: announcing someone else's public key must fail — no matching private key."""
    user = await _make_user(client, "ann2_test")
    victim_sk = Ed25519PrivateKey.generate()
    victim_pk = pk_to_b64(victim_sk.public_key())

    attacker_sk = Ed25519PrivateKey.generate()
    payload = _announce_payload(user["user_id"], attacker_sk, victim_pk)

    r = await client.post(
        "/v1/nodes/announce", json=payload,
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert r.status_code == 401, r.text


@pytest.mark.asyncio
async def test_announce_rejects_stale_timestamp(client):
    """M8: a captured announce must not be replayable later."""
    import time as _t
    user = await _make_user(client, "ann3_test")
    sk = Ed25519PrivateKey.generate()
    payload = _announce_payload(
        user["user_id"], sk, pk_to_b64(sk.public_key()), ts=int(_t.time()) - 3600)

    r = await client.post(
        "/v1/nodes/announce", json=payload,
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert r.status_code == 401, r.text


@pytest.mark.asyncio
async def test_announce_with_valid_proof_succeeds_and_is_idempotent(client):
    """The legitimate path works, and re-announcing updates rather than piling up rows."""
    user = await _make_user(client, "ann4_test")
    sk = Ed25519PrivateKey.generate()
    pk_b64 = pk_to_b64(sk.public_key())

    first = await client.post(
        "/v1/nodes/announce", json=_announce_payload(user["user_id"], sk, pk_b64),
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert first.status_code == 201, first.text

    second = await client.post(
        "/v1/nodes/announce", json=_announce_payload(user["user_id"], sk, pk_b64),
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert second.status_code == 201, second.text
    assert second.json()["node_id"] == first.json()["node_id"], (
        "re-announcing the same key must not create a second node record (M8)")


# ── An empty claim is not a claim on everything ──────────────────────────────
#
# 2026-09-11, found on a live deployment. `_claimable` used to read
# `set(claimed_groups or authorized)`, and the node omits `group_ids` entirely
# when it hosts nothing — so "I host no groups" was read as "I host all of
# yours". The node cannot serve any of them (no GEK, and its own handshake
# refuses them), but `/v1/groups/{id}/nodes` lists nodes in registration order,
# so whenever such a node won the reconnection race after a hub restart it
# became `nodes[0]` and the group stopped opening for every member.


@pytest.mark.asyncio
async def test_ws_absent_claim_registers_no_groups(client):
    """A node that declares nothing hosts nothing — it must not inherit the set."""
    from meshbay_hub.api.revocation import _authorize_node_ws

    user = await _make_user(client, "empty1_test")
    node_id = await _announce_node(client, user)
    for name in ("has-one", "has-two"):
        r = await client.post(
            "/v1/groups",
            json={"name": name, "visibility": "private", "join_policy": "invite"},
            headers={"Authorization": f"Bearer {user['token']}"},
        )
        assert r.status_code == 201, r.text

    resolved, groups = await _authorize_node_ws(_node_token(user), node_id, None)
    assert resolved == node_id
    assert groups == [], (
        "a node hosting nothing was registered as a host for its owner's groups")


@pytest.mark.asyncio
async def test_ws_explicit_empty_claim_registers_no_groups(client):
    """And the same when the node says so out loud, which it now does."""
    from meshbay_hub.api.revocation import _authorize_node_ws

    user = await _make_user(client, "empty2_test")
    node_id = await _announce_node(client, user)
    r = await client.post(
        "/v1/groups",
        json={"name": "lonely", "visibility": "private", "join_policy": "invite"},
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert r.status_code == 201, r.text

    resolved, groups = await _authorize_node_ws(_node_token(user), node_id, [])
    assert resolved == node_id
    assert groups == []


@pytest.mark.asyncio
async def test_empty_node_cannot_shadow_another_members_group(client):
    """
    The outage itself: two members, one group, and only one of them hosts it.
    The other's node — running, configured with nothing — must not appear as a
    source for that group, because it is the one clients would reach first.
    """
    from meshbay_hub.api.revocation import (
        _authorize_node_ws, _node_groups, get_online_nodes_for_group)

    host = await _make_user(client, "hoster_test")
    guest = await _make_user(client, "guest_test")
    host_node = await _announce_node(client, host)
    guest_node = await _announce_node(client, guest)

    r = await client.post(
        "/v1/groups",
        json={"name": "shared", "visibility": "private", "join_policy": "invite"},
        headers={"Authorization": f"Bearer {host['token']}"},
    )
    assert r.status_code == 201, r.text
    group_id = r.json()["group_id"]

    r = await client.post(
        f"/v1/groups/{group_id}/members/{'guest_test'}",
        headers={"Authorization": f"Bearer {host['token']}"},
    )
    assert r.status_code == 201, r.text

    # The guest's node registers first — the order that made this fatal.
    _, guest_groups = await _authorize_node_ws(_node_token(guest), guest_node, None)
    _, host_groups = await _authorize_node_ws(
        _node_token(host), host_node, [group_id])
    _node_groups[guest_node] = guest_groups
    _node_groups[host_node] = host_groups
    try:
        assert get_online_nodes_for_group(group_id) == [host_node], (
            "a member's empty node shadowed the node actually hosting the group")
    finally:
        _node_groups.pop(guest_node, None)
        _node_groups.pop(host_node, None)


@pytest.mark.asyncio
async def test_update_groups_is_held_to_the_same_ceiling(client):
    """
    `update_groups` assigned the message's list verbatim, so the C2 ceiling held
    at authentication could be stepped over one message later: a node had only
    to reload to claim any group on the hub. It now goes through the same gate,
    which is what this asserts — the socket loop itself needs a WebSocket the
    ASGI harness has not got (see this module's docstring).
    """
    from meshbay_hub.api.revocation import _authorized_groups, _claimable

    user = await _make_user(client, "reloader")
    r = await client.post(
        "/v1/groups",
        json={"name": "owned", "visibility": "private", "join_policy": "invite"},
        headers={"Authorization": f"Bearer {user['token']}"},
    )
    assert r.status_code == 201, r.text
    own = r.json()["group_id"]

    authorized = await _authorized_groups(user["user_id"])
    assert _claimable([own, "someone-elses-group"], authorized) == [own]
    assert _claimable([], authorized) == []
    assert _claimable(None, authorized) == []