aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src/meshbay_node/transport
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-01 21:51:25 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-01 21:51:25 +0200
commit799d87999c8324564dce5159191532e008dd93d2 (patch)
tree5ff1816f18625dfece9eb67fa06c7b25fdece4f8 /packages/meshbay-node/src/meshbay_node/transport
parent8a6294b0412a86f378c6e2e937c28de64a903c91 (diff)
parent1e6db7d23c70b7bd7e1422f09911b3645f0fb2e2 (diff)
downloadmeshbay-799d87999c8324564dce5159191532e008dd93d2.tar.gz
Merge branch 'fix/third-review-h1-h2-m1-m6'
Third security review (docs/third-review.md) plus its remediation. Fixed and verified: - H1 moderator could grant admin / hard-revoke → handler split by field - H2 unauthenticated 2-report global blocklist → auth + distinct reporters + rate limit + refused when public groups are off - M1 registration reCAPTCHA was inert → gate unconditional; the desktop client's CSP allows the widget - M2 QUIC chat/stream handlers lagged WebRTC → brought to parity; the QUIC listener is now off by default ([node] quic_enabled) - M3 link-preview SSRF gaps → rate limit + port allowlist + connect-address re-check + decompression-bomb guard - M4 federated peer over-trust → source bound to the signer, push capped, revocation prunes the peer's own entries, replay rejected - M5 no CSP / security headers on the SPA → middleware; verified against the live app with no violations Withdrawn: - M6 add_group_member accepting node tokens is deliberate (commit 0443cf8, the CLI invite flow). The "fix" broke that flow on the deployed hub and was reverted. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011pG75yGK3NthNfyjH74omG
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/quic_server.py144
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py40
2 files changed, 135 insertions, 49 deletions
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: