3 files changed, 150 insertions, 4 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
index 0f1ddba..2e3323c 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
@@ -33,7 +33,7 @@ import time
 import uuid
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect
+from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect
 from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -274,12 +274,28 @@ class IncomingRequest(BaseModel):
 async def notify_incoming(
     node_id: str,
     body: IncomingRequest,
+    request: Request,
     current_user: User = Depends(get_current_user),
 ):
     """
     Signal a node that a client wants to connect (NAT punch coordination).
     Hub forwards the request via WebSocket; node punches NAT and replies punch_ready.
+
+    Finding H6: peer_ip was taken verbatim, so any authenticated user could make an
+    arbitrary node emit UDP packets to an address of their choosing — a small
+    reflection primitive using someone else's machine. The probe target must now be
+    the caller's own source address.
     """
+    from meshbay_hub.api.netutil import client_ip
+
+    caller_ip = client_ip(request)
+    if body.peer_ip != caller_ip:
+        raise HTTPException(
+            status_code=403,
+            detail="peer_ip must match the requesting address")
+    if not (1 <= body.peer_port <= 65535):
+        raise HTTPException(status_code=422, detail="Invalid peer_port")
+
     ws = _connected_nodes.get(node_id)
     if not ws:
         raise HTTPException(status_code=404, detail="Node not connected")
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
index bd343c9..8f84163 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
@@ -17,11 +17,15 @@ import json
 import logging
 import uuid
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub.api.deps import get_current_user
-from meshbay_hub.db.models import User
+from meshbay_hub.api.middleware import limiter
+from meshbay_hub.db.engine import get_db
+from meshbay_hub.db.models import Group, GroupMember, User
 
 log = logging.getLogger(__name__)
 
@@ -41,25 +45,66 @@ class WebRTCOfferResponse(BaseModel):
     peer_id: str
 
 
+MAX_SDP_BYTES = 16 * 1024          # an SDP offer is ~2 KB
+MAX_PENDING_PER_USER = 3           # concurrent in-flight offers per account
+
+_pending_per_user: dict[str, int] = {}
+
+
 @router.post("/{node_id}/webrtc/offer", response_model=WebRTCOfferResponse)
+@limiter.limit("30/minute")
 async def webrtc_offer(
     node_id: str,
     body: WebRTCOfferRequest,
+    request: Request,
     current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
 ):
     """
     Browser sends WebRTC SDP offer for a node. Hub relays via WebSocket.
     Returns the node's SDP answer once received.
+
+    Finding H6: this was reachable by any authenticated user, for any node, with no
+    rate limit and no membership check. Each call makes the node allocate an
+    aiortc RTCPeerConnection and gather ICE, so it was a remote resource-exhaustion
+    primitive against an arbitrary third party's machine.
+
+    Finding H4: it also ignored group status, so "suspend a group" did not stop new
+    connections from being brokered to nodes hosting it.
     """
-    from meshbay_hub.api.revocation import _connected_nodes
+    from meshbay_hub.api.revocation import _connected_nodes, _node_groups
+
+    if len(body.sdp) > MAX_SDP_BYTES:
+        raise HTTPException(status_code=413, detail="SDP too large")
 
     ws = _connected_nodes.get(node_id)
     if not ws:
         raise HTTPException(status_code=404, detail="Node not connected")
 
+    # The caller must share at least one active group with the target node.
+    node_group_ids = set(_node_groups.get(node_id, []))
+    if node_group_ids:
+        result = await db.execute(
+            select(GroupMember.group_id).where(
+                GroupMember.user_id == current_user.id,
+                GroupMember.group_id.in_(node_group_ids),
+            ))
+        shared = [gid for (gid,) in result.all()]
+        if not shared:
+            raise HTTPException(status_code=403, detail="Not a member of any group on this node")
+
+        active = await db.execute(
+            select(Group.id).where(Group.id.in_(shared), Group.status == "active"))
+        if not active.first():
+            raise HTTPException(status_code=403, detail="Group is not active")
+
+    if _pending_per_user.get(current_user.id, 0) >= MAX_PENDING_PER_USER:
+        raise HTTPException(status_code=429, detail="Too many pending connections")
+
     peer_id = str(uuid.uuid4())
     answer_future: asyncio.Future = asyncio.get_event_loop().create_future()
     _webrtc_answers[peer_id] = answer_future
+    _pending_per_user[current_user.id] = _pending_per_user.get(current_user.id, 0) + 1
 
     try:
         await ws.send_text(json.dumps({
@@ -83,6 +128,11 @@ async def webrtc_offer(
         )
     finally:
         _webrtc_answers.pop(peer_id, None)
+        remaining = _pending_per_user.get(current_user.id, 1) - 1
+        if remaining > 0:
+            _pending_per_user[current_user.id] = remaining
+        else:
+            _pending_per_user.pop(current_user.id, None)
 
 
 def handle_webrtc_answer(msg: dict) -> None:
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