fix(reef): bound inbox WebSocket frames (#108886)

extensions/reef/package.json

@@ -7,6 +7,7 @@
     "@noble/ciphers": "2.2.0",
     "@noble/curves": "2.2.0",
     "@noble/hashes": "2.2.0",
+    "ws": "8.21.0",
     "zod": "4.4.3"
   },
   "openclaw": {

extensions/reef/src/channel.ts

@@ -26,8 +26,8 @@ import { loadKeys, openStores, resolveStateDir, ReviewApprovalStore } from "./st
 import {
   ReefInboxConnection,
   ReefTransportClient,
-  type WebSocketLike,
   abortableSleep,
+  createReefWebSocket,
 } from "./transport.js";
 import { isReefPairingApprovalToken, openReefTrustStore } from "./trust-store.js";
 import type { ReefAccount, ReefIngressMessage } from "./types.js";
@@ -292,11 +292,10 @@ export const reefPlugin: ChannelPlugin<ReefAccount> = {
       };
       await reconcile();
       ctx.setStatus({ accountId: "default", running: true, connected: false });
-      const socketFactory = (url: string) => new WebSocket(url) as unknown as WebSocketLike;
       const inbox = new ReefInboxConnection(
         transport,
         (entries) => flow.processEntries(entries),
-        socketFactory,
+        createReefWebSocket,
         (state) => {
           if (ctx.abortSignal.aborted) {
             return;

extensions/reef/src/transport.test.ts

@@ -1,7 +1,14 @@
 import { createPublicKey, verify as verifySignature } from "node:crypto";
+import { once } from "node:events";
 import { describe, expect, it, vi } from "vitest";
+import { WebSocketServer } from "ws";
 import { canonicalBytes, fromBase64url, sha256Hex } from "../protocol/index.js";
-import { ReefRelayError, ReefTransportClient } from "./transport.js";
+import {
+  ReefInboxConnection,
+  ReefRelayError,
+  ReefTransportClient,
+  createReefWebSocket,
+} from "./transport.js";
 import type { ReefKeys, RelayFriend } from "./types.js";
 
 const ts = 1_752_300_000;
@@ -318,3 +325,91 @@ describe("ReefTransportClient response body bounds", () => {
     });
   });
 });
+
+const INBOX_WEBSOCKET_MAX_PAYLOAD_BYTES = 64 * 1024;
+
+function inboxFrameAtSize(bytes: number): string {
+  const prefix =
+    '{"type":"entry","entry":{"seq":1,"peer":"bob","id":"01ARZ3NDEKTSV4RRFFQ69G5FAV","kind":"receipt","receipt":{"pad":"';
+  const suffix = '"},"ts":1752300000}}';
+  return `${prefix}${"x".repeat(bytes - prefix.length - suffix.length)}${suffix}`;
+}
+
+async function deliverInboxFrame(frame: string): Promise<{
+  entries: unknown[];
+  states: string[];
+}> {
+  const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
+  await once(server, "listening");
+  const address = server.address();
+  if (!address || typeof address === "string") {
+    throw new Error("WebSocket test server did not bind a TCP port");
+  }
+  const entries: unknown[] = [];
+  const states: string[] = [];
+  const abort = new AbortController();
+  const timeout = setTimeout(() => abort.abort(), 2_000);
+  server.once("connection", (socket) => socket.send(frame));
+  const client = new ReefTransportClient(
+    `http://127.0.0.1:${address.port}`,
+    "alice",
+    keys,
+    async () => Response.json({ entries: [], cursor: 0 }),
+    () => ts,
+  );
+  const inbox = new ReefInboxConnection(
+    client,
+    async (received) => {
+      entries.push(...received);
+      abort.abort();
+    },
+    createReefWebSocket,
+    (state) => {
+      states.push(state);
+      if (state === "disconnected") {
+        abort.abort();
+      }
+    },
+  );
+
+  try {
+    await inbox.start(abort.signal);
+  } finally {
+    clearTimeout(timeout);
+    for (const socket of server.clients) {
+      socket.terminate();
+    }
+    await new Promise<void>((resolve, reject) => {
+      server.close((error) => {
+        if (error) {
+          reject(error);
+          return;
+        }
+        resolve();
+      });
+    });
+  }
+  return { entries, states };
+}
+
+describe("ReefInboxConnection response frame bounds", () => {
+  it("accepts a relay frame exactly at the payload limit", async () => {
+    const frame = inboxFrameAtSize(INBOX_WEBSOCKET_MAX_PAYLOAD_BYTES);
+
+    const result = await deliverInboxFrame(frame);
+
+    expect(Buffer.byteLength(frame)).toBe(INBOX_WEBSOCKET_MAX_PAYLOAD_BYTES);
+    expect(result.entries).toHaveLength(1);
+    expect(result.states).toContain("connected");
+  });
+
+  it("rejects a relay frame above the payload limit before dispatch", async () => {
+    const frame = inboxFrameAtSize(INBOX_WEBSOCKET_MAX_PAYLOAD_BYTES + 1);
+
+    const result = await deliverInboxFrame(frame);
+
+    expect(Buffer.byteLength(frame)).toBe(INBOX_WEBSOCKET_MAX_PAYLOAD_BYTES + 1);
+    expect(result.entries).toEqual([]);
+    expect(result.states).toEqual(["connected", "disconnected"]);
+  });
+});

extensions/reef/src/transport.ts

@@ -1,4 +1,5 @@
 import { readProviderJsonResponse } from "openclaw/plugin-sdk/provider-http";
+import WebSocket from "ws";
 import { sha256Hex, signDeviceRequest, utf8 } from "../protocol/index.js";
 import type { Envelope, SignedReceipt } from "../protocol/index.js";
 import type { InboxEntry, ReefKeys, RelayFriend } from "./types.js";
@@ -10,6 +11,9 @@ type FetchLike = typeof fetch;
 // force unbounded allocation through response.json().
 const REEF_RELAY_JSON_MAX_BYTES = 16 * 1024 * 1024;
 const REEF_RELAY_ERROR_JSON_MAX_BYTES = 64 * 1024;
+// Relay envelopes are capped at 48 KiB. Leave room for inbox metadata while
+// rejecting oversized or compressed frames before ws materializes the message.
+const REEF_RELAY_WEBSOCKET_MAX_PAYLOAD_BYTES = 64 * 1024;
 
 export class ReefRelayError extends Error {
   constructor(
@@ -186,6 +190,10 @@ export interface WebSocketLike {
   close(): void;
 }
 
+export function createReefWebSocket(url: string): WebSocketLike {
+  return new WebSocket(url, { maxPayload: REEF_RELAY_WEBSOCKET_MAX_PAYLOAD_BYTES });
+}
+
 export function abortableSleep(ms: number, signal?: AbortSignal): Promise<void> {
   return new Promise<void>((resolve) => {
     if (signal?.aborted) {

pnpm-lock.yaml

@@ -1596,6 +1596,9 @@ importers:
       '@noble/hashes':
         specifier: 2.2.0
         version: 2.2.0
+      ws:
+        specifier: 8.21.0
+        version: 8.21.0
       zod:
         specifier: 4.4.3
         version: 4.4.3