""" 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} async def _announce_node(client, user: dict) -> str: 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 == 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"], user["pk_ed"], 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") 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") 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") 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") 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") 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") 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_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") 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]]