1 files changed, 172 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_node_ws_auth.py b/packages/meshbay-hub/tests/test_node_ws_auth.py
new file mode 100644
index 0000000..6db95d7
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_node_ws_auth.py
@@ -0,0 +1,172 @@
+"""
+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]]