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 | 128 |
1 files changed, 108 insertions, 20 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 43e6fdc..190d580 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -64,6 +64,16 @@ MAX_MSG = 64 * 1024 * 1024 # node sovereignty and defeated the delete authorization (overwrite a file, become its # recorded uploader, then delete it legitimately). 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 +# ffmpeg is spawned per stream request; without a cap any member can fork-bomb +# the node by requesting many streams at once (H6). +MAX_CONCURRENT_TRANSCODES = 2 +# Bundle fetches are served in the pre-proof window (C4). Bounded and audited +# until the native client removes remote keypair bundles entirely. +MAX_PRE_PROOF_FETCHES = 4 UPLOAD_DIR_NAME = ".uploads" # Conservative allowlist: also what keeps markup out of filenames, which the node admin # UI used to render unescaped (finding H2). @@ -137,10 +147,18 @@ def _pack(obj: dict) -> bytes: class _DataChannelBuffer: - """Accumulate DataChannel messages and extract length-prefixed msgpack.""" + """ + Accumulate DataChannel messages and extract length-prefixed msgpack. + + Finding H6: the limit was a flat 64 MB applied even before the handshake, so an + unauthenticated peer could announce a 64 MB frame and dribble bytes into it, + holding that much memory per connection. Until a peer has proved GEK + possession it gets a small budget; the large one is for file uploads. + """ - def __init__(self): + def __init__(self, max_message: int = MAX_MSG): self._buf = bytearray() + self.max_message = max_message def feed(self, data: bytes): self._buf.extend(data) @@ -148,7 +166,7 @@ class _DataChannelBuffer: def messages(self): while len(self._buf) >= 4: length = struct.unpack(">I", self._buf[:4])[0] - if length > MAX_MSG: + if length > self.max_message: raise ValueError(f"Message too large: {length}") if len(self._buf) < 4 + length: break @@ -180,7 +198,8 @@ class WebRTCPeerSession: self._pc = pc self._ctx = node_ctx self._channel: RTCDataChannel | None = None - self._buffer = _DataChannelBuffer() + self._buffer = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG) + self._pre_proof_fetches = 0 self._user_id: str | None = None self._group_id: str | None = None self._peer_id: str = peer_id @@ -210,10 +229,24 @@ class WebRTCPeerSession: self._do_handshake(msg) elif mtype == MNP.HANDSHAKE_RESPONSE: self._do_handshake_response(msg) - elif mtype == MNP.GEK_BUNDLE_FETCH and self._gek_challenge is not None: - asyncio.ensure_future(self._do_gek_bundle_fetch()) - elif mtype == MNP.KEYPAIR_BUNDLE_FETCH and self._gek_challenge is not None: - asyncio.ensure_future(self._do_keypair_bundle_fetch()) + elif mtype in (MNP.GEK_BUNDLE_FETCH, MNP.KEYPAIR_BUNDLE_FETCH) \ + and self._gek_challenge is not None: + # Served before the GEK proof by necessity: the client needs its + # wrapped bundle in order to compute the proof. That window is a + # disclosure surface (C4) — a hub that forges a JWT reaches it — so + # it is bounded and audited here, and closed properly when clients + # stop storing keypair bundles on other people's nodes. + self._pre_proof_fetches += 1 + if self._pre_proof_fetches > MAX_PRE_PROOF_FETCHES: + self._audit_auth_failed( + getattr(self, "_pending_group", ""), "pre-proof fetch flood") + self._send({"type": "error", "detail": "Too many requests"}) + return + self._audit_pre_proof_fetch(mtype) + if mtype == MNP.GEK_BUNDLE_FETCH: + asyncio.ensure_future(self._do_gek_bundle_fetch()) + else: + asyncio.ensure_future(self._do_keypair_bundle_fetch()) elif self._user_id is None: self._send({"type": "error", "detail": "Handshake required"}) elif mtype == MNP.INDEX_SYNC: @@ -357,6 +390,9 @@ class WebRTCPeerSession: self._complete_handshake() def _complete_handshake(self) -> None: + # Authenticated peers may send large frames (file uploads); unauthenticated + # ones may not (H6). + self._buffer.max_message = MAX_MSG self._user_id = self._pending_sub self._group_id = self._pending_group self._username = self._pending_username @@ -496,6 +532,20 @@ class WebRTCPeerSession: "detail": "keypair_bundle_stored", }) + def _audit_pre_proof_fetch(self, mtype: str) -> None: + """Record bundle access made before the GEK proof (C4).""" + audit = self._ctx.get("audit_store") + if not audit: + return + self._remote_ip = self._remote_ip or _get_remote_ip(self._pc) + asyncio.ensure_future(audit.log_event( + user_id=getattr(self, "_pending_sub", "unknown"), + event="pre_proof_fetch", + ip=self._remote_ip, + group_id=getattr(self, "_pending_group", "") or "", + detail=mtype, + )) + def _audit_auth_failed(self, group_id: str, reason: str) -> None: audit = self._ctx.get("audit_store") if audit: @@ -574,6 +624,17 @@ class WebRTCPeerSession: self._audit("file_download", entry.name) def _do_stream_segment(self, msg: dict) -> None: + asyncio.ensure_future(self._do_stream_segment_async(msg)) + + async def _do_stream_segment_async(self, msg: dict) -> None: + """ + Legacy HLS segment extraction (superseded by stream_req/MSE). + + Finding H6: this ran subprocess.run(..., timeout=30) directly inside the + event loop, so a single request stalled the whole daemon — every peer, + every group — for up to thirty seconds. Now async and under the same + transcode semaphore as _stream_video. + """ ctx = self._group_ctx() file_id = msg["file_id"] segment_index = msg["segment_index"] @@ -589,21 +650,34 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not on disk"}) return - import subprocess + sem = self._ctx.get("_transcode_sem") + if sem is None: + sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES) + self._ctx["_transcode_sem"] = sem + try: - result = subprocess.run( - ["ffmpeg", "-hide_banner", "-loglevel", "error", - "-ss", str(segment_index * segment_duration), - "-i", str(file_path), - "-t", str(segment_duration), - "-c:v", "copy", "-c:a", "copy", - "-f", "mpegts", "pipe:1"], - capture_output=True, timeout=30, - ) - if result.returncode != 0 or not result.stdout: + async with sem: + proc = await asyncio.create_subprocess_exec( + "ffmpeg", "-hide_banner", "-loglevel", "error", + "-ss", str(segment_index * segment_duration), + "-i", str(file_path), + "-t", str(segment_duration), + "-c:v", "copy", "-c:a", "copy", + "-f", "mpegts", "pipe:1", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + try: + stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + self._send({"type": "error", "detail": "Segment extraction timed out"}) + return + if proc.returncode != 0 or not stdout: self._send({"type": "error", "detail": "Segment extraction failed"}) return - segment_data = result.stdout + segment_data = stdout except Exception: self._send({"type": "error", "detail": "Segment extraction failed"}) return @@ -963,6 +1037,20 @@ class WebRTCPeerSession: async def _stream_video(self, msg: dict) -> None: """Stream a video file as fMP4 segments via MSE-compatible output.""" + # One ffmpeg per request with no cap lets any member exhaust the node's + # CPU and process table (H6). The semaphore lives on the transport context + # so it is shared across all peers, not per-session. + sem = self._ctx.get("_transcode_sem") + if sem is None: + sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES) + self._ctx["_transcode_sem"] = sem + if sem.locked() and sem._value <= 0: + self._send({"type": "error", "detail": "Server busy, retry shortly"}) + return + async with sem: + await self._stream_video_inner(msg) + + async def _stream_video_inner(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg.get("file_id", "") entry = ctx["index"].get_entry(file_id) |