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