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

8 files changed, 343 insertions, 59 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:
diff --git a/packages/meshbay-node/tests/test_link_preview_request.py b/packages/meshbay-node/tests/test_link_preview_request.py
index d975f91..fe7dec6 100644
--- a/packages/meshbay-node/tests/test_link_preview_request.py
+++ b/packages/meshbay-node/tests/test_link_preview_request.py
@@ -34,9 +34,10 @@ def _clear_cache():
     webrtc_server._link_preview_cache.clear()
 
 
-def _session(media_cache):
+def _session(media_cache, ctx=None):
     s = WebRTCPeerSession.__new__(WebRTCPeerSession)
-    s._ctx = {"media_cache": media_cache}
+    s._ctx = ctx if ctx is not None else {"media_cache": media_cache}
+    s._peer_id = "t"
     s.sent = []
     s._send = s.sent.append
     return s
@@ -79,6 +80,51 @@ async def test_unfurlable_failure_is_ok_false(media_cache, monkeypatch):
     assert "image_thumb_hash" not in resp
 
 
+async def test_rate_limit_per_connection(media_cache, monkeypatch):
+    """A member firing many previews is bounded; over the ceiling the reply is
+    a plain `ok: false` (bare link) and no outbound fetch is made."""
+    monkeypatch.setattr(webrtc_server, "_LINK_PREVIEW_RATE_PER_CONN", 3)
+    calls = {"n": 0}
+
+    async def counting_preview(url, **k):
+        calls["n"] += 1
+        return {"url": url, "title": "x", "description": None,
+                "site_name": None, "image_url": None}
+    monkeypatch.setattr(linkpreview, "fetch_preview", counting_preview)
+
+    s = _session(media_cache)
+    for i in range(3):
+        await s._do_link_preview_request({"url": f"https://example.com/{i}"})
+    assert calls["n"] == 3
+    assert all(r["ok"] for r in s.sent)
+
+    await s._do_link_preview_request({"url": "https://example.com/over"})
+    assert calls["n"] == 3                 # not fetched
+    assert s.sent[-1]["ok"] is False
+
+
+async def test_rate_limit_is_node_wide(media_cache, monkeypatch):
+    """Two connections share the node-wide ceiling."""
+    monkeypatch.setattr(webrtc_server, "_LINK_PREVIEW_RATE_PER_CONN", 100)
+    monkeypatch.setattr(webrtc_server, "_LINK_PREVIEW_RATE_NODE", 2)
+    calls = {"n": 0}
+
+    async def counting_preview(url, **k):
+        calls["n"] += 1
+        return {"url": url, "title": "x", "description": None,
+                "site_name": None, "image_url": None}
+    monkeypatch.setattr(linkpreview, "fetch_preview", counting_preview)
+
+    ctx = {"media_cache": media_cache}
+    a, b = _session(media_cache, ctx), _session(media_cache, ctx)
+    await a._do_link_preview_request({"url": "https://example.com/a"})
+    await b._do_link_preview_request({"url": "https://example.com/b"})
+    await b._do_link_preview_request({"url": "https://example.com/c"})
+
+    assert calls["n"] == 2
+    assert b.sent[-1]["ok"] is False
+
+
 async def test_second_request_for_the_same_url_is_served_from_cache(media_cache, monkeypatch):
     calls = {"n": 0}
 
diff --git a/packages/meshbay-node/tests/test_linkpreview.py b/packages/meshbay-node/tests/test_linkpreview.py
index 0dd951a..a14b173 100644
--- a/packages/meshbay-node/tests/test_linkpreview.py
+++ b/packages/meshbay-node/tests/test_linkpreview.py
@@ -48,6 +48,30 @@ def test_safe_url_refuses(url):
         safe_url(url)
 
 
+@pytest.mark.parametrize("url", [
+    "http://example.com:22/x",       # SSH
+    "http://example.com:3306/x",     # MySQL
+    "http://example.com:6379/x",     # Redis
+    "http://example.com:9200/x",     # Elasticsearch
+    "http://example.com:5000/x",     # a common internal admin port
+])
+def test_safe_url_refuses_non_web_ports(url, resolves_public):
+    with pytest.raises(UnsafeURL):
+        safe_url(url)
+
+
+@pytest.mark.parametrize("url", [
+    "http://example.com/x",          # implicit 80
+    "https://example.com/x",         # implicit 443
+    "http://example.com:80/x",
+    "https://example.com:443/x",
+    "http://example.com:8080/x",
+    "https://example.com:8443/x",
+])
+def test_safe_url_allows_the_web_ports(url, resolves_public):
+    assert safe_url(url) == url
+
+
 def test_safe_url_accepts_a_public_host(resolves_public):
     assert safe_url("https://example.com/some/page") == "https://example.com/some/page"
 
@@ -153,3 +177,19 @@ async def test_fetch_image_downscales(resolves_public):
     assert jpeg and jpeg[:2] == b"\xff\xd8"        # JPEG SOI
     with Image.open(BytesIO(jpeg)) as im:
         assert max(im.size) <= linkpreview._IMAGE_MAX_DIM
+
+
+async def test_fetch_image_refuses_a_decompression_bomb(resolves_public, monkeypatch):
+    from io import BytesIO
+
+    from PIL import Image
+    # A tiny file that reports enormous dimensions from its header alone.
+    monkeypatch.setattr(linkpreview, "_MAX_IMAGE_PIXELS", 1_000_000)
+    buf = BytesIO()
+    Image.new("RGB", (2000, 2000), (0, 0, 0)).save(buf, format="PNG")  # 4 MP > cap
+    bomb = buf.getvalue()
+
+    def handler(request):
+        return httpx.Response(200, headers={"content-type": "image/png"}, content=bomb)
+    async with _client(handler) as c:
+        assert await linkpreview.fetch_image("https://example.com/x.png", client=c) is None
diff --git a/packages/meshbay-node/tests/test_quic_enabled.py b/packages/meshbay-node/tests/test_quic_enabled.py
new file mode 100644
index 0000000..c0c1568
--- /dev/null
+++ b/packages/meshbay-node/tests/test_quic_enabled.py
@@ -0,0 +1,53 @@
+"""
+The QUIC MNP listener is off unless the operator turns it on.
+
+No shipping client speaks QUIC (browser and desktop use WebRTC; the hub-less
+`group://` sidecar is unbuilt), so a node that started it by default would only
+be exposing a UDP port. `daemon.py` gates `QuicChunkServer` on
+`self._config.node.quic_enabled`; these follow the value down the config path.
+"""
+
+import textwrap
+from pathlib import Path
+
+from meshbay_node.config import load_config
+
+
+def _cfg(tmp_path: Path, body: str):
+    p = tmp_path / "node.toml"
+    p.write_text(textwrap.dedent(body))
+    return load_config(p)
+
+
+def test_off_by_default(tmp_path):
+    cfg = _cfg(tmp_path, """
+        [node]
+        quic_port = 19010
+    """)
+    assert cfg.node.quic_enabled is False
+
+
+def test_the_operator_turns_it_on(tmp_path):
+    cfg = _cfg(tmp_path, """
+        [node]
+        quic_enabled = true
+    """)
+    assert cfg.node.quic_enabled is True
+
+
+def test_the_environment_can_force_it_on(tmp_path, monkeypatch):
+    monkeypatch.setenv("MESHBAY_QUIC_ENABLED", "1")
+    cfg = _cfg(tmp_path, """
+        [node]
+        quic_enabled = false
+    """)
+    assert cfg.node.quic_enabled is True
+
+
+def test_the_environment_can_force_it_off(tmp_path, monkeypatch):
+    monkeypatch.setenv("MESHBAY_QUIC_ENABLED", "false")
+    cfg = _cfg(tmp_path, """
+        [node]
+        quic_enabled = true
+    """)
+    assert cfg.node.quic_enabled is False