diff options
Diffstat (limited to 'packages/meshbay-node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 60 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_peer_session_limits.py | 136 |
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 |