1 files changed, 346 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..1391722
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_node_ws_auth.py
@@ -0,0 +1,346 @@
+"""
+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")
+    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_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")
+    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")
+    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")
+    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")
+    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")
+    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")
+    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")
+    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")
+    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)")