diff options
Diffstat (limited to 'packages/meshbay-node')
4 files changed, 230 insertions, 28 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 7dedad8..fa8c0ee 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -26,6 +26,7 @@ import asyncio import base64 import json import logging +import os import signal import sys from pathlib import Path @@ -130,6 +131,11 @@ class NodeDaemon: # 2. Start admin UI early (so operator can copy node key before hub login) self._state["pk_node_ed25519"] = keys.pk_ed25519_b64 self._state["config"] = self._config + # Per-run token for the local admin UI (11.5.3). Not a password: it keeps + # other local processes and rebound browser pages out of an API that can + # re-initialise group keys. + ui_token = base64.urlsafe_b64encode(os.urandom(18)).decode().rstrip("=") + self._state["ui_token"] = ui_token from meshbay_node.ui import create_ui_app ui_app = create_ui_app(self._state) ui_cfg = uvicorn.Config( @@ -140,7 +146,8 @@ class NodeDaemon: ) ui_server = uvicorn.Server(ui_cfg) self._tasks.append(asyncio.create_task(ui_server.serve())) - log.info("Admin UI at http://localhost:%d", self._config.node.ui_port) + log.info("Admin UI at http://127.0.0.1:%d/?t=%s", + self._config.node.ui_port, ui_token) # 3. Hub connection (Ed25519 auth — retries until node key is linked) hub_cfg = HubConfig( 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) diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index d2c3429..21e4445 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -9,7 +9,7 @@ FastAPI app providing: - API endpoints for all data (JSON) Served only on 127.0.0.1 — not exposed to the network. -No authentication required (localhost only). +Gated by a per-run session token (11.5.3) — printed at daemon startup. """ import base64 @@ -37,6 +37,28 @@ def create_ui_app(state: dict) -> FastAPI: ) @app.middleware("http") + async def _require_session_token(request, call_next): + """ + Gate the admin UI behind a per-run token (11.5.3). + + "localhost only" is weaker than it sounds: any process on the machine can + reach it, and a page in the operator's browser can reach it too via DNS + rebinding. Since this API can re-initialise a group's GEK and read the + audit log, an unauthenticated loopback service is a privilege boundary + waiting to be crossed. The token is printed at startup and accepted as + ?t= or the X-MeshBay-Token header. + """ + from fastapi.responses import PlainTextResponse + + token = state.get("ui_token") + if token: + supplied = (request.query_params.get("t") + or request.headers.get("X-MeshBay-Token")) + if supplied != token: + return PlainTextResponse("Forbidden", status_code=403) + return await call_next(request) + + @app.middleware("http") async def _security_headers(request, call_next): """ Defence in depth behind the escaping fixes for H2. This UI is unauthenticated @@ -340,7 +362,7 @@ def create_ui_app(state: dict) -> FastAPI: @app.get("/audit", response_class=HTMLResponse) async def audit_page(): - return _render_audit_page() + return _render_audit_page(state.get("ui_token", "")) return app @@ -356,6 +378,7 @@ def _fmt_size(n: int) -> str: def _render_page(state: dict) -> str: + token_js = json.dumps(state.get("ui_token", "")) status = state.get("status", "starting") indexes = state.get("indexes", {}) groups_ctx = state.get("groups_ctx", {}) @@ -552,13 +575,14 @@ def _render_page(state: dict) -> str: </div> </div> <script> +const TOKEN = {token_js}; async function initGEK(groupId) {{ const btn = document.getElementById('gek-btn-' + groupId.slice(0,8)); const status = document.getElementById('gek-status-' + groupId.slice(0,8)); if (btn) btn.disabled = true; if (status) status.textContent = 'Initializing...'; try {{ - const resp = await fetch('/api/groups/' + groupId + '/gek', {{ method: 'POST' }}); + const resp = await fetch('/api/groups/' + groupId + '/gek?t=' + TOKEN, {{ method: 'POST' }}); const data = await resp.json(); if (resp.ok) {{ if (status) status.textContent = 'GEK initialized — wrapped for ' @@ -582,8 +606,11 @@ setTimeout(()=>location.reload(), 10000); </html>""" -def _render_audit_page() -> str: - return """<!DOCTYPE html> +def _render_audit_page(token: str = "") -> str: + return _AUDIT_HTML.replace("__TOKEN__", json.dumps(token)) + + +_AUDIT_HTML = """<!DOCTYPE html> <html lang="en"> <head> <meta charset="utf-8"> @@ -619,7 +646,7 @@ def _render_audit_page() -> str: <body> <div class="container"> <h1>Audit Log</h1> - <nav><a href="/">Dashboard</a><a href="/audit">Audit Log</a></nav> + <nav><a id="navHome" href="/">Dashboard</a><a id="navAudit" href="/audit">Audit Log</a></nav> <div class="filters"> <select id="eventFilter"> @@ -648,10 +675,11 @@ def _render_audit_page() -> str: </table> </div> <script> +const TOKEN = __TOKEN__; async function load() { const ev = document.getElementById('eventFilter').value; const limit = document.getElementById('limitSelect').value; - let url = '/api/audit?limit=' + limit; + let url = '/api/audit?limit=' + limit + (TOKEN ? '&t=' + TOKEN : ''); if (ev) url += '&event=' + ev; const r = await fetch(url); const data = await r.json(); @@ -677,6 +705,10 @@ async function load() { return tr; })); } +for (const [id, href] of [['navHome','/'],['navAudit','/audit']]) { + const el = document.getElementById(id); + if (el && TOKEN) el.href = href + '?t=' + TOKEN; +} document.getElementById('eventFilter').onchange = load; document.getElementById('limitSelect').onchange = load; let debounceTimer; diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index 9552848..a6d69c6 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -10,6 +10,7 @@ If one of these starts failing, a fix has been reverted. Do not "fix" the test. """ import base64 +import struct from pathlib import Path import pytest @@ -465,6 +466,80 @@ def test_peer_errors_do_not_leak_internals(): assert '"detail": str(e)' not in source +def test_pre_handshake_message_budget_is_small(): + """ + H6: the frame limit was a flat 64 MB applied before authentication, so an + unauthenticated peer could announce a huge frame and dribble bytes into it. + """ + from meshbay_node.transport.webrtc_server import ( + MAX_MSG, PRE_HANDSHAKE_MAX_MSG, _DataChannelBuffer, + ) + assert PRE_HANDSHAKE_MAX_MSG <= 1024 * 1024 + assert PRE_HANDSHAKE_MAX_MSG < MAX_MSG + + buf = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG) + buf.feed(struct.pack(">I", PRE_HANDSHAKE_MAX_MSG + 1) + b"x") + with pytest.raises(ValueError): + list(buf.messages()) + + +def test_stream_segment_is_not_synchronous(): + """ + H6: _do_stream_segment ran subprocess.run(timeout=30) inside the event loop, + stalling every peer on the node for up to thirty seconds per request. + + Asserts the property (the worker is a coroutine, ffmpeg is spawned through + asyncio) rather than grepping for "subprocess.run" — which also matches the + comment that documents the old behaviour. + """ + import ast + import inspect + from meshbay_node.transport.webrtc_server import WebRTCPeerSession + + assert inspect.iscoroutinefunction(WebRTCPeerSession._do_stream_segment_async) + + source = (Path(__file__).parent.parent / "src" / "meshbay_node" + / "transport" / "webrtc_server.py").read_text() + tree = ast.parse(source) + blocking = [ + node for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "run" + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "subprocess" + ] + assert not blocking, "blocking subprocess.run() in the event loop" + assert "_transcode_sem" in source, "ffmpeg spawns must be capped" + + +def test_pre_proof_fetches_are_bounded(): + """C4: the pre-proof bundle window is a disclosure surface; bound it.""" + from meshbay_node.transport.webrtc_server import MAX_PRE_PROOF_FETCHES + assert 0 < MAX_PRE_PROOF_FETCHES <= 10 + + +def test_node_admin_ui_requires_token(): + """ + 11.5.3: "localhost only" is not authentication. Any local process — or a + rebound browser page — could re-initialise a group's GEK and read the audit log. + """ + from fastapi.testclient import TestClient + from meshbay_node.ui.app import create_ui_app + + app = create_ui_app({"status": "running", "groups_ctx": {}, + "indexes": {}, "ui_token": "secret-token"}) + client = TestClient(app) + + assert client.get("/api/status").status_code == 403 + assert client.get("/api/status?t=wrong").status_code == 403 + assert client.get("/api/config?t=wrong").status_code == 403 + assert client.get("/api/status?t=secret-token").status_code == 200 + assert client.get( + "/api/status", headers={"X-MeshBay-Token": "secret-token"} + ).status_code == 200 + + def test_admin_ui_escapes_filenames(tmp_path): """ H2: filenames are chosen by any group member and were rendered into the |