diff options
Diffstat (limited to 'packages/meshbay-node/src')
5 files changed, 202 insertions, 57 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index b1ebd54..78ee873 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -36,7 +36,10 @@ url = "https://meshbay.org" username = "myusername" [node] -quic_port = 19010 # QUIC (MNP) — LAN, port-forwarded, hub-less direct access +# QUIC (MNP) direct path — LAN, port-forwarded, hub-less. Off by default: no +# client speaks QUIC yet, so leaving it on only opens a UDP port. +quic_enabled = false +quic_port = 19010 ui_port = 18000 # local control API — JSON, 127.0.0.1 only, token-gated # One-time codes. An invitation waits for someone to read their messages; an @@ -129,6 +132,13 @@ class HubConfig: @dataclass class NodeConfig: quic_port: int = 19010 + # The QUIC MNP listener. Off by default: no shipping client speaks QUIC yet + # (the browser and the desktop client use WebRTC; the hub-less `group://` + # sidecar is unbuilt), so starting it only opens a UDP port with nothing to + # reach it. Turn on for LAN / port-forwarded / hub-less direct access once a + # client for it exists. `punch_nat()` is a direct-connection helper, not a + # NAT-traversal stack — a peer behind NAT still needs the port forwarded. + quic_enabled: bool = False ui_port: int = 18000 # How long a one-time code stays usable. Invitations travel through a human # conversation and are answered days later; operator pairing happens during @@ -303,6 +313,7 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: # `port` (TCP+TLS) and `http_port` no longer exist — both listeners were removed # in Phase 11.5 (findings C1, C6). Regenerate node.toml with `meshbay-node init`. cfg.node.quic_port = nd.get("quic_port", cfg.node.quic_port) + cfg.node.quic_enabled = bool(nd.get("quic_enabled", cfg.node.quic_enabled)) cfg.node.ui_port = nd.get("ui_port", cfg.node.ui_port) cfg.node.invite_ttl_hours = int( nd.get("invite_ttl_hours", cfg.node.invite_ttl_hours)) @@ -371,6 +382,8 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: cfg.hub.username = user if port := os.environ.get("MESHBAY_QUIC_PORT"): cfg.node.quic_port = int(port) + if (qe := os.environ.get("MESHBAY_QUIC_ENABLED")) is not None: + cfg.node.quic_enabled = qe.strip().lower() in ("1", "true", "yes", "on") if streams := os.environ.get("MESHBAY_MAX_CONCURRENT_STREAMS"): cfg.node.max_concurrent_streams = _positive( streams, cfg.node.max_concurrent_streams, diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 056371a..8f9307c 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -537,7 +537,11 @@ class NodeDaemon: log.warning("WebRTC not available (aiortc not installed)") # 7. QUIC chunk server (LAN / port-forwarded / hub-less direct access) - if QUIC_AVAILABLE: + # + # Off unless `[node] quic_enabled = true`: no shipping client speaks + # QUIC (browser and desktop use WebRTC; the `group://` sidecar is + # unbuilt), so starting it by default only exposes a UDP port. + if QUIC_AVAILABLE and self._config.node.quic_enabled: self._quic_server = QuicChunkServer( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, @@ -553,6 +557,8 @@ class NodeDaemon: await self._quic_server.start() log.info("QUIC server on port %d (%d groups)", self._config.node.quic_port, len(groups_ctx)) + elif QUIC_AVAILABLE: + log.info("QUIC server disabled ([node] quic_enabled = false)") # 8. Hub WebSocket (signaling + revocations + WebRTC offers) async def on_webrtc_offer(sdp, peer_id, ice_candidates): @@ -1708,6 +1714,7 @@ def main() -> None: f'username = "{username}"', "", "[node]", + "quic_enabled = false # QUIC direct path; no client uses it yet", "quic_port = 19010", "ui_port = 18000", "", diff --git a/packages/meshbay-node/src/meshbay_node/linkpreview.py b/packages/meshbay-node/src/meshbay_node/linkpreview.py index 64067d1..b223aea 100644 --- a/packages/meshbay-node/src/meshbay_node/linkpreview.py +++ b/packages/meshbay-node/src/meshbay_node/linkpreview.py @@ -14,12 +14,15 @@ hub: Because the node makes an outbound request to an address a *member* chose, this is an SSRF surface. `safe_url()` is the gate: http(s) only, no -credentials, and the resolved address must be globally routable — no -loopback, private, link-local, multicast or reserved range. Redirects are -followed by hand so every hop is re-checked. Residual: a DNS name that -resolves clean here and to something internal microseconds later at connect -time (rebinding) — narrow, and closed properly by pinning the checked IP, -which is a follow-up. +credentials, the port restricted to the web set, and every resolved address +must be globally routable — no loopback, private, link-local, multicast or +reserved range. Redirects are followed by hand so every hop is re-checked, +and the address the connection actually landed on is re-checked against the +same rule (`_reject_if_rebound`), so a name that resolves clean and then to +something internal (rebinding) does not get its body read. A full pin — +connect to the validated literal, verify the certificate for the name — is +the remaining hardening. How many previews a member can trigger is +rate-limited by the caller (`_do_link_preview_request`). Nothing is stored durably: the caller keeps an in-memory TTL cache and the OG image rides the existing `media_cache` thumb store (same as a poster). @@ -42,9 +45,15 @@ _TIMEOUT = 5.0 _MAX_REDIRECTS = 3 _MAX_HTML_BYTES = 512 * 1024 _MAX_IMAGE_BYTES = 2 * 1024 * 1024 +_MAX_IMAGE_PIXELS = 40_000_000 # ~40 MP; an OG card image is a fraction of this _IMAGE_MAX_DIM = 600 _UA = "MeshBayBot/1.0 (+https://meshbay.org; link preview)" +# Ports a real OpenGraph-bearing page is served on. Everything else — SSH, mail, +# databases, caches, search, admin panels — is refused, so a member cannot aim +# the node at an arbitrary service even on a public host. +_ALLOWED_PORTS = frozenset({80, 443, 8080, 8443}) + class UnsafeURL(ValueError): """The URL points somewhere the node must not fetch from.""" @@ -75,6 +84,12 @@ def safe_url(url: str) -> str: host = parts.hostname if not host: raise UnsafeURL("no host") + try: + port = parts.port + except ValueError: + raise UnsafeURL("bad port") + if port is not None and port not in _ALLOWED_PORTS: + raise UnsafeURL(f"port {port}") # An IP literal is checked directly; a name is resolved and every answer # must be public — a hostname with one public and one 127.0.0.1 record # would otherwise be a way in. @@ -136,12 +151,32 @@ def _first(metas: dict[str, str], *keys: str) -> str | None: return None +def _reject_if_rebound(resp: httpx.Response) -> None: + """ + `safe_url` validated the name's addresses; this checks the one the + connection actually landed on, so a name that resolves clean and then to + something internal (DNS rebinding) does not get its body read. + + Best-effort: the `network_stream` extension is not present on every + transport (a MockTransport in tests has none), and its absence is not a + failure — the pre-check and the per-hop redirect re-check still stand. + """ + try: + stream = resp.extensions.get("network_stream") + addr = stream.get_extra_info("server_addr") if stream else None + except Exception: + return + if addr and not _addr_is_public(str(addr[0])): + raise UnsafeURL(f"connected to non-public address {addr[0]}") + + async def _get(client: httpx.AsyncClient, url: str) -> httpx.Response: """One GET with manual, re-validated redirects.""" current = safe_url(url) for _ in range(_MAX_REDIRECTS + 1): resp = await client.get(current, headers={"User-Agent": _UA}, follow_redirects=False) + _reject_if_rebound(resp) if resp.is_redirect and "location" in resp.headers: current = safe_url(urljoin(current, resp.headers["location"])) continue @@ -242,6 +277,10 @@ def _downscale(raw: bytes) -> bytes | None: return None try: with Image.open(BytesIO(raw)) as im: + # The header is parsed but the pixels are not decoded yet — refuse a + # decompression bomb before convert()/thumbnail() allocate for it. + if im.width * im.height > _MAX_IMAGE_PIXELS: + return None im = im.convert("RGB") im.thumbnail((_IMAGE_MAX_DIM, _IMAGE_MAX_DIM)) out = BytesIO() diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index ce6fe17..30c7daf 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -61,6 +61,15 @@ CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 ALPN = ["meshbay-mnp"] +# ffmpeg is spawned per STREAM_SEGMENT request, and `_extract_segment` runs +# `subprocess.run` synchronously — so without a bound, an authenticated peer can +# both fork-bomb the node and block its event loop for up to 30 s per request +# (finding M2c). Extraction now runs in a thread and passes through this +# semaphore. Small on purpose: the QUIC path has no shipping client yet, this is +# parity work with the WebRTC transcode cap. +_MAX_CONCURRENT_SEGMENTS = 4 +_segment_sem = asyncio.Semaphore(_MAX_CONCURRENT_SEGMENTS) + class Denylist: """ @@ -204,6 +213,9 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._nonce_client: bytes = b"" self._gek_challenge: bytes | None = None self._pending = None + # asyncio holds only a weak reference to a bare task, so a spawned + # handler still running can be collected mid-flight. Hold them. + self._tasks: set[asyncio.Task] = set() def quic_event_received(self, event: QuicEvent) -> None: if isinstance(event, StreamDataReceived): @@ -232,7 +244,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): elif mtype == MNP.FILE_REQUEST: self._do_file_request_sync(stream_id, msg) elif mtype == MNP.STREAM_SEGMENT: - self._do_stream_segment_sync(stream_id, msg) + self._spawn(self._do_stream_segment(stream_id, msg)) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message_sync(stream_id, msg) elif mtype == MNP.PING: @@ -256,11 +268,10 @@ class _MNPServerProtocol(QuicConnectionProtocol): checks could drift from the WebRTC path independently. All of that now comes from meshbay_common.handshake, shared with WebRTC. - NOT YET DONE — finding C6 remains open on this transport: there is still no - GEK proof here, so a forged or stolen token reaches the node and can inject - chat without holding the group key. The challenge/response and mutual node - proof (quic_binding() is written and unit-tested for exactly this) are the - remaining work in 11.5.4/5/6. + The GEK proof is enforced here too: `_do_handshake_response_sync` runs + the same challenge/response and mutual node proof, from the same shared + module, bound to the QUIC certificate hash. Finding C6 is closed on this + transport. """ try: peer = authorize_token( @@ -339,9 +350,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._user_id = peer.user_id self._group_id = peer.group_id - peers = self._ctx.get("_peers") - if peers is not None: - peers[self._user_id] = self + self._peer_registry()[self._user_id] = self transcript = handshake_transcript( ROLE_NODE, peer.group_id, self._nonce_client, self._gek_challenge, binding) @@ -365,6 +374,20 @@ class _MNPServerProtocol(QuicConnectionProtocol): return self._ctx["groups"][self._group_id] return self._ctx + def _spawn(self, coro) -> None: + task = asyncio.ensure_future(coro) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + def _peer_registry(self) -> dict: + """QUIC peers for THIS connection's group, keyed per group so a message + never crosses into another group on a multi-group node (findings M2b / + H1). Deliberately separate from the WebRTC registry that also lives in + the group context: the two transports' session objects have different + `_send` signatures, and cross-transport chat fan-out is not wired (no + QUIC client ships yet).""" + return self._group_ctx().setdefault("_quic_peers", {}) + def _do_index_sync_sync(self, stream_id: int) -> None: ctx = self._group_ctx() wire = ctx["index"].serialize() @@ -399,61 +422,83 @@ class _MNPServerProtocol(QuicConnectionProtocol): ) self._send(stream_id, chunk_data) - def _do_stream_segment_sync(self, stream_id: int, msg: dict) -> None: - """Extract and serve one HLS segment via ffmpeg.""" - ctx = self._group_ctx() - file_id = msg["file_id"] - segment_index = msg["segment_index"] - segment_duration = msg.get("segment_duration", 4) + async def _do_stream_segment(self, stream_id: int, msg: dict) -> None: + """ + Extract and serve one segment via ffmpeg — off the event loop and behind + a concurrency bound, so one request can neither stall the whole node nor + fork-bomb it (finding M2c). The WebRTC path has had both since Phase 11.5. + """ + try: + ctx = self._group_ctx() + file_id = msg["file_id"] + segment_index = msg["segment_index"] + segment_duration = msg.get("segment_duration", 4) - entry = ctx["index"].get_entry(file_id) - if not entry: - self._send(stream_id, {"type": "error", "detail": "File not found"}) - return + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send(stream_id, {"type": "error", "detail": "File not found"}) + return - file_path = entry_abs_path(ctx["roots"], entry) - if not file_path.exists(): - self._send(stream_id, {"type": "error", "detail": "File not on disk"}) - return + file_path = entry_abs_path(ctx["roots"], entry) + if not file_path.exists(): + self._send(stream_id, {"type": "error", "detail": "File not on disk"}) + return - start_time = segment_index * segment_duration - segment_data = _extract_segment(file_path, start_time, segment_duration) - if segment_data is None: - self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) - return + start_time = segment_index * segment_duration + loop = asyncio.get_event_loop() + async with _segment_sem: + segment_data = await loop.run_in_executor( + None, _extract_segment, file_path, start_time, segment_duration) + if segment_data is None: + self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) + return - self._send(stream_id, { - "type": MNP.STREAM_SEGMENT, - "v": MNP_VERSION, - "file_id": file_id, - "segment_index": segment_index, - "data_b64": base64.b64encode(segment_data).decode(), - "size": len(segment_data), - }) + self._send(stream_id, { + "type": MNP.STREAM_SEGMENT, + "v": MNP_VERSION, + "file_id": file_id, + "segment_index": segment_index, + "data_b64": base64.b64encode(segment_data).decode(), + "size": len(segment_data), + }) + except Exception as e: + log.error("stream_segment: %s", e) + self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) def _do_chat_message_sync(self, stream_id: int, msg: dict) -> None: - """Receive a chat message, store it, and broadcast to other connected peers.""" - chat_store = self._ctx.get("chat_store") + """ + Store a chat message and broadcast it to the rest of THIS group. + + `sender_id` is the authenticated session's, never the wire's — a peer + must not be able to post as someone else (NS6 / finding M2a). The store + and the peer set come from the group context, not a connection-global + one, so a message never crosses into another group on a multi-group node + (findings M2b / H1). The WebRTC path has done both since Phase 11.5. + """ + gctx = self._group_ctx() + payload = msg.get("payload", b"") + if isinstance(payload, str): + payload = payload.encode() + + chat_store = gctx.get("chat_store") if chat_store: - import asyncio - asyncio.ensure_future(chat_store.save_message( - sender_id=msg.get("sender_id", self._user_id), + self._spawn(chat_store.save_message( + sender_id=self._user_id, iteration=msg.get("iteration", 0), - payload=msg.get("payload", b"").encode() if isinstance(msg.get("payload"), str) else msg.get("payload", b""), + payload=payload, thread_id=msg.get("thread_id"), )) - peers = self._ctx.get("_peers", {}) broadcast = { "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, - "sender_id": msg.get("sender_id", self._user_id), + "sender_id": self._user_id, "iteration": msg.get("iteration", 0), "payload": msg.get("payload", ""), "thread_id": msg.get("thread_id"), "group_id": self._group_id or "", } - for uid, proto in peers.items(): + for uid, proto in list(self._peer_registry().items()): if uid != self._user_id and proto is not self: try: proto._send(0, broadcast) @@ -463,9 +508,10 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "ack", "v": MNP_VERSION}) def connection_lost(self, exc) -> None: - peers = self._ctx.get("_peers") - if peers and self._user_id: - peers.pop(self._user_id, None) + if self._user_id: + self._peer_registry().pop(self._user_id, None) + for task in list(self._tasks): + task.cancel() super().connection_lost(exc) def _send(self, stream_id: int, obj: dict) -> None: @@ -558,7 +604,7 @@ class QuicChunkServer: self._ctx["groups"] = groups self._denylist = denylist or Denylist() self._ctx["denylist"] = self._denylist - self._ctx["_peers"] = {} + # Peer sets are per group now — see _MNPServerProtocol._peer_registry(). self._host = host self._port = port self._cert_path = cert_path or Path.home() / ".config/meshbay/node_tls.crt" 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 003cd23..64ba75c 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -143,6 +143,16 @@ def _link_preview_cache_put(url: str, value: dict) -> None: _link_preview_cache.pop(oldest, None) _link_preview_cache[url] = (time.time(), value) + +# A member pasting a link is normal; a member — or a hub minting tokens for many +# accounts — firing hundreds is amplification/DoS and a way to make the node +# reach arbitrary hosts on demand (finding M3). Only a real outbound fetch is +# counted (a cache hit costs nothing), and the ceilings are generous enough that +# ordinary chat never meets them. +_LINK_PREVIEW_RATE_WINDOW = 60.0 +_LINK_PREVIEW_RATE_PER_CONN = 15 +_LINK_PREVIEW_RATE_NODE = 60 + # Upload limits (finding C5a). Uploads used to land directly in the shared root under # a name the client chose, overwriting whatever was already there — which both violated # node sovereignty and defeated the delete authorization (overwrite a file, become its @@ -3588,6 +3598,27 @@ class WebRTCPeerSession: ], }) + def _link_preview_rate_ok(self) -> bool: + """ + True when this preview fetch is within both the per-connection and the + node-wide window; records it when so, and both counts are trimmed to the + window on every call so the lists cannot grow without bound. + """ + now = time.monotonic() + w = _LINK_PREVIEW_RATE_WINDOW + mine = [t for t in getattr(self, "_link_preview_hits", []) if now - t < w] + node = [t for t in self._ctx.get("link_preview_hits", []) if now - t < w] + if (len(mine) >= _LINK_PREVIEW_RATE_PER_CONN + or len(node) >= _LINK_PREVIEW_RATE_NODE): + self._link_preview_hits = mine + self._ctx["link_preview_hits"] = node + return False + mine.append(now) + node.append(now) + self._link_preview_hits = mine + self._ctx["link_preview_hits"] = node + return True + async def _do_link_preview_request(self, msg: dict) -> None: """ Unfurl a URL a member pasted into chat (draft-v6 §2.7 enrichment rule: @@ -3608,6 +3639,15 @@ class WebRTCPeerSession: self._send({**cached, "type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION}) return + if not self._link_preview_rate_ok(): + # Same shape as any other miss — the client shows the bare link. A + # rate-limited result is not cached, so it is retried once the + # window clears rather than pinned as "no preview". + log.debug("link_preview_req: rate-limited (peer=%s)", self._peer_id) + self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION, + "url": key, "ok": False}) + return + resp: dict = {"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION, "url": key, "ok": False} try: |