fix: bound what one member can cost the others

An availability review, prompted by the group claim above: a participant
supplies input — who else bears the cost? Six answers where the cost fell on
someone other than the sender, and none of them needs an attacker.

  AV3  `chat_notify` carried a `group_id` the hub believed, so any connected
       node could write a notification to every member of any group on the
       hub, carrying a display string of its choosing, with its account
       having no relation to that group. This is the group claim again, two
       hundred lines further down the same socket. Gated on what the node is
       registered for, and metered: the fan-out is one write per member. The
       budget expires by time rather than on disconnect, or reconnecting
       would refill it and a node token is good for an hour.

  AV4  A swarm source named its own `endpoint` as free text documented as
       "ip:port", so an account could publish a third party's address — H6's
       `peer_ip` defect, never applied here. Nothing dials a swarm source
       today, which is the only reason it was not already a reflection
       primitive. It is a transport and a port now, never a host, and the
       number of hashes one account may claim is bounded: rows were keyed
       (hash, account) with no cap at all.

  AV5  `handle_webrtc_answer` resolved any pending `peer_id` from any node's
       socket. The answer is the SDP a browser then connects to. That this
       had not happened rested on a uuid4 being unguessable.

  AV6  `relay_register` had no authentication of any kind: it compared
       `pk_relay` against the approved value, which is a *public* key, so
       anyone who could read it could rewrite where the hub tells nodes to
       send relayed traffic. The module docstring promised signed JWTs and
       `jwt` was imported and never used.

  AV7  The node held unlimited peer connections and kept one that never
       completed a handshake for the life of the daemon. H6 bounded what one
       unauthenticated peer costs; the hub's cap is three offers in flight
       per *account*, a limit on each caller and not on the machine, so an
       operator's exposure grew with the size of their groups.

  AV8  `invite-notify` put a request-supplied `group_name` into the subject
       of an email the hub sends under its own domain, to any account, with
       no rate limit. The name comes from the group row now.

The tests are two accounts each, in one file that says why: a one-member test
proves a one-member property, and every finding here needed a second person
to exist at all. Each was checked against the unfixed code. Two did not
survive that check and were rewritten — one re-enacted the disconnect path
instead of running it (hence `forget_node`), the other called the reaper
itself and would have passed with the call removed from `handle_offer`.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01T4YmK41VsEURWFdop4EEeT

2 files changed, 196 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index a0776d2..29d6e7f 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -192,6 +192,18 @@ MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024   # 4 GB per file
 # Budget for an unauthenticated peer: enough for a handshake and a bundle fetch,
 # nowhere near enough to be a memory-exhaustion primitive (H6).
 PRE_HANDSHAKE_MAX_MSG = 64 * 1024
+
+# How many peer connections this node holds at once, and how long one may stay
+# without completing the MNP handshake. The budget above bounds what *one*
+# unauthenticated peer costs; these bound how many there may be and how long
+# each lasts, which is the other half and was missing. The hub caps three
+# offers in flight per account — a limit on each caller, not on this machine —
+# so the cost to an operator grew with the number of people in their groups.
+# Sized to be unreachable in ordinary use: a browser holds one connection per
+# open group, and a handshake unfinished after a minute is not going to finish.
+MAX_PEER_SESSIONS = 64
+UNAUTHENTICATED_SESSION_TIMEOUT = 60   # seconds
+
 # ffmpeg is spawned per stream request; without a cap any member can fork-bomb
 # the node by requesting many streams at once (H6).
 #
@@ -5882,6 +5894,7 @@ class WebRTCTransport:
         from meshbay_node.config import DEFAULT_STUN_SERVERS
         self._stun = stun_servers or list(DEFAULT_STUN_SERVERS)
         self._sessions: dict[str, WebRTCPeerSession] = {}
+        self._reapers: set[asyncio.Task] = set()
 
     def set_capacity(self, *, max_concurrent_streams: int | None = None,
                      max_concurrent_downloads: int | None = None,
@@ -5991,9 +6004,21 @@ class WebRTCTransport:
         config = RTCConfiguration(
             iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else []
         )
+        # Before anything is allocated. Every offer costs an RTCPeerConnection
+        # with its own DTLS and SCTP stacks, and nothing here used to bound how
+        # many a node would hold: the hub caps three in flight *per account*,
+        # which is a limit on each caller and not on this machine, so the cost
+        # grew with the number of members in the group. An operator's node must
+        # not be exhaustible by the people they invited.
+        if len(self._sessions) >= MAX_PEER_SESSIONS:
+            log.warning("Refusing WebRTC offer: %d peer sessions already open",
+                        len(self._sessions))
+            raise RuntimeError("Node is at its peer-connection limit")
+
         pc = RTCPeerConnection(configuration=config)
         session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id)
         self._sessions[peer_id] = session
+        self._reap_if_unauthenticated(peer_id)
 
         @pc.on("datachannel")
         def on_datachannel(channel: RTCDataChannel):
@@ -6048,12 +6073,47 @@ class WebRTCTransport:
             ", ".join(sorted(host_addrs)) or "none", srflx)
         return answer_sdp, []
 
+    def _reap_if_unauthenticated(self, peer_id: str) -> None:
+        """Close a session that never completes the handshake.
+
+        A peer that connects and then says nothing is indistinguishable from a
+        working one until it is asked to prove something, and it was never
+        asked: `connectionstatechange` reaps a connection that *fails*, and one
+        that succeeds and stays silent was held for the node's lifetime. That
+        is the cheapest way to spend someone else's memory — no GEK, no token,
+        no group, just an open connection. `_user_id` is set by the GEK proof
+        (`_do_handshake_response`), so it is the one honest test of whether
+        this peer ever became anybody.
+        """
+        async def reap() -> None:
+            try:
+                await asyncio.sleep(UNAUTHENTICATED_SESSION_TIMEOUT)
+                session = self._sessions.get(peer_id)
+                if session is not None and not session._user_id:
+                    log.warning("Closing peer %s: no handshake within %ds",
+                                peer_id[:8], UNAUTHENTICATED_SESSION_TIMEOUT)
+                    await self.close_peer(peer_id)
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                log.warning("Reaping peer %s failed: %s", peer_id[:8], e)
+
+        # Held in a set for the same reason every other task here is: asyncio
+        # keeps only a weak reference, and a reaper collected mid-sleep reaps
+        # nothing (see WebRTCPeerSession.__init__).
+        task = asyncio.ensure_future(reap())
+        self._reapers.add(task)
+        task.add_done_callback(self._reapers.discard)
+
     async def close_peer(self, peer_id: str) -> None:
         session = self._sessions.pop(peer_id, None)
         if session:
             await session.close()
 
     async def close_all(self) -> None:
+        for task in list(self._reapers):
+            task.cancel()
+        self._reapers.clear()
         for session in list(self._sessions.values()):
             await session.close()
         self._sessions.clear()
diff --git a/packages/meshbay-node/tests/test_peer_session_limits.py b/packages/meshbay-node/tests/test_peer_session_limits.py
new file mode 100644
index 0000000..64961ad
--- /dev/null
+++ b/packages/meshbay-node/tests/test_peer_session_limits.py
@@ -0,0 +1,136 @@
+"""
+Availability, on the node: what the people you invited can cost you.
+
+The message budget for an unauthenticated peer (`PRE_HANDSHAKE_MAX_MSG`, H6)
+bounds what *one* of them spends. Nothing bounded how many there could be, or
+how long one could stay without ever proving anything — and the hub's cap is
+three offers in flight **per account**, which is a limit on each caller rather
+than on this machine, so an operator's exposure grew with the number of people
+in their groups. A member who never meant any harm — a tab left open through a
+sleep, a client retrying a connection it cannot complete — arrives here the
+same way as one who does.
+"""
+
+import asyncio
+from unittest.mock import MagicMock
+
+import pytest
+
+from meshbay_node.transport import webrtc_server as ws_mod
+from meshbay_node.transport.webrtc_server import (
+    MAX_PEER_SESSIONS, UNAUTHENTICATED_SESSION_TIMEOUT, WebRTCTransport)
+
+
+def _transport() -> WebRTCTransport:
+    return WebRTCTransport(sk_node=MagicMock(), hub_pk_pem=b"", gek=None,
+                           roots=None, index=None)
+
+
+class _FakeSession:
+    """Stands in for a peer that connected and then said nothing."""
+
+    def __init__(self, user_id=None):
+        self._user_id = user_id
+        self.closed = False
+
+    async def close(self):
+        self.closed = True
+
+
+@pytest.mark.asyncio
+async def test_the_node_refuses_more_peers_than_it_will_hold():
+    """The refusal comes before an RTCPeerConnection is allocated, or the cap
+    would be counting the thing it is meant to prevent."""
+    tp = _transport()
+    for i in range(MAX_PEER_SESSIONS):
+        tp._sessions[f"peer-{i}"] = _FakeSession(user_id="someone")
+
+    with pytest.raises(RuntimeError, match="peer-connection limit"):
+        await tp.handle_offer("v=0", "one-too-many")
+
+    assert "one-too-many" not in tp._sessions
+
+
+@pytest.mark.asyncio
+async def test_a_peer_that_never_handshakes_is_closed(monkeypatch):
+    """`connectionstatechange` reaps a connection that *fails*. One that
+    succeeds and stays silent was held for the life of the daemon."""
+    monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05)
+    tp = _transport()
+    session = _FakeSession(user_id=None)
+    tp._sessions["quiet"] = session
+
+    tp._reap_if_unauthenticated("quiet")
+    await asyncio.sleep(0.2)
+
+    assert session.closed, "a peer that never proved anything was kept"
+    assert "quiet" not in tp._sessions
+
+
+@pytest.mark.asyncio
+async def test_a_peer_that_handshaked_is_left_alone(monkeypatch):
+    """`_user_id` is set by the GEK proof, and is the one honest test of
+    whether this peer ever became anybody."""
+    monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05)
+    tp = _transport()
+    session = _FakeSession(user_id=None)
+    tp._sessions["real"] = session
+
+    tp._reap_if_unauthenticated("real")
+    session._user_id = "a-real-member"        # the handshake completes
+    await asyncio.sleep(0.2)
+
+    assert not session.closed, "a member who completed the handshake was cut off"
+    assert tp._sessions["real"] is session
+
+
+@pytest.mark.asyncio
+async def test_handle_offer_arms_the_reaper(monkeypatch):
+    """
+    The seam, driven rather than described. The three tests above call
+    `_reap_if_unauthenticated` themselves, so every one of them would still
+    pass with the call removed from `handle_offer` and no peer reaped at all —
+    which is the defect, not the helper. aiortc is stubbed because a real
+    RTCPeerConnection wants a real SDP; everything else here is the shipped
+    code path.
+    """
+    class _FakePC:
+        def __init__(self, *a, **kw):
+            self.localDescription = MagicMock(sdp="v=0\r\n")
+            self.remoteDescription = None
+
+        def on(self, _event):
+            return lambda fn: fn
+
+        async def setRemoteDescription(self, _d): pass
+        async def createAnswer(self): return MagicMock()
+        async def setLocalDescription(self, _d): pass
+        async def close(self): pass
+
+    monkeypatch.setattr(ws_mod, "RTCPeerConnection", _FakePC)
+    monkeypatch.setattr(ws_mod, "RTCSessionDescription",
+                        lambda **kw: MagicMock(**kw))
+
+    tp = _transport()
+    await tp.handle_offer("v=0\r\n", "fresh-peer")
+    try:
+        assert tp._reapers, (
+            "handle_offer allocated a peer session and armed nothing to "
+            "close it if the handshake never comes")
+    finally:
+        await tp.close_all()
+
+
+@pytest.mark.asyncio
+async def test_the_reaper_is_held_and_cancelled_with_the_transport():
+    """asyncio keeps only a weak reference to a task, and a reaper collected
+    mid-sleep reaps nothing — the same trap as every other task in this file."""
+    tp = _transport()
+    tp._sessions["held"] = _FakeSession()
+    tp._reap_if_unauthenticated("held")
+    assert tp._reapers, "the reaper was fired and forgotten"
+
+    await tp.close_all()
+    assert not tp._reapers
+    # The timeout is a real duration, not something a test has to wait out.
+    assert UNAUTHENTICATED_SESSION_TIMEOUT >= 30