1 files changed, 80 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
index 6db95d7..9ec251d 100644
--- a/packages/meshbay-hub/tests/test_node_ws_auth.py
+++ b/packages/meshbay-hub/tests/test_node_ws_auth.py
@@ -150,6 +150,86 @@ async def test_ws_group_claims_cannot_widen_beyond_membership(client):
 
 
 @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