Add receiver-side resource limits for untrusted peers (#184)

Deserializing messages from an untrusted peer could exhaust CPU, memory,
or the stack:

- A long [bigint,...] digit string fed straight to BigInt() blocks the
  event loop, because BigInt()'s decimal parse is superlinear.
- A deeply nested message could overflow the stack during evaluation.
- An arbitrarily large message was JSON-parsed with no up-front size bound.

Add three local, receiver-side guards in the deserialization path:

- Cap bigint digit length and reject non-numeric strings before BigInt().
- Bound nesting depth in Evaluator.evaluateImpl, mirroring the existing
  send-side depth limit in Devaluator.
- Reject oversized messages in the session read loop before JSON.parse.

Limits have safe exported defaults (DEFAULT_LIMITS) and are overridable per
session via a new RpcLimits-typed 'limits' field on RpcSessionOptions,
surfaced to the Evaluator through the Importer interface so the standalone
deserialize() path stays protected by the defaults. Limits are purely local
(the protocol has no negotiation step); exceeding one throws, which aborts
the session via the existing abort path.

Author: ndisidore

.changeset/untrusted-resource-limits.md

@@ -0,0 +1,5 @@
+---
+"capnweb": patch
+---
+
+Add receiver-side resource limits to guard against untrusted-peer resource exhaustion (#184). Deserialization now caps the length of `bigint` values (and rejects non-numeric digit strings before calling `BigInt()`), bounds message nesting depth, and rejects oversized incoming messages before parsing. The limits are local, receiver-side decisions with safe defaults (`DEFAULT_LIMITS`), and can be overridden per session via the new `limits` field on `RpcSessionOptions`. Exceeding a limit aborts the session, reusing the existing abort path.

__tests__/limits.test.ts

@@ -0,0 +1,236 @@
+// Copyright (c) 2025 Cloudflare, Inc.
+// Licensed under the MIT license found in the LICENSE.txt file or at:
+//     https://opensource.org/license/mit
+
+// Tests for the receive-side resource limits that guard against untrusted-peer resource
+// exhaustion (issue #184): bigint length cap + numeric validation, message nesting depth, and
+// max message size. See `RpcLimits` / `DEFAULT_LIMITS` in src/serialize.ts.
+
+import { expect, it, describe } from "vitest"
+import { deserialize, serialize, RpcSession, RpcTransport, RpcTarget, DEFAULT_LIMITS,
+         type RpcLimits } from "../src/index.js"
+
+// Builds the wire string for a value nested inside `depth` escaped-array layers. Each escaped
+// array `[[ inner ]]` adds one level of recursion to the deserializer (evaluateImpl recurses into
+// its single element). The innermost value is the number 1.
+function nestedEscapedArrays(depth: number): string {
+  let s = "1";
+  for (let i = 0; i < depth; i++) {
+    s = `[[${s}]]`;
+  }
+  return s;
+}
+
+// A one-shot transport that feeds a single pre-baked frame to the session and then blocks, while
+// capturing everything the session sends back. Lets us assert exactly how the session reacts to a
+// hostile incoming message.
+class ScriptedTransport implements RpcTransport {
+  sent: string[] = [];
+  aborted = false;
+  private toReceive: string[];
+  private blockForever = new Promise<string>(() => {});
+
+  constructor(frames: string[]) {
+    this.toReceive = [...frames];
+  }
+
+  async send(message: string): Promise<void> {
+    this.sent.push(message);
+  }
+
+  async receive(): Promise<string> {
+    let next = this.toReceive.shift();
+    return next === undefined ? this.blockForever : next;
+  }
+
+  abort(reason: any): void {
+    this.aborted = true;
+  }
+
+  // Resolves once the session has sent an ["abort", ...] frame, or rejects after a spin budget.
+  async waitForAbort(): Promise<string> {
+    for (let i = 0; i < 200; i++) {
+      let abortFrame = this.sent.find(m => m.startsWith('["abort"'));
+      if (abortFrame !== undefined) return abortFrame;
+      await Promise.resolve();
+    }
+    throw new Error(`session did not abort; sent: ${JSON.stringify(this.sent)}`);
+  }
+
+  async expectNoAbort(): Promise<void> {
+    for (let i = 0; i < 50; i++) await Promise.resolve();
+    let abortFrame = this.sent.find(m => m.startsWith('["abort"'));
+    expect(abortFrame, "session unexpectedly aborted").toBeUndefined();
+    expect(this.aborted, "transport unexpectedly aborted").toBe(false);
+  }
+}
+
+// Drives a server session that processes a single ["push", expr] frame, returning the transport so
+// the caller can inspect the reaction.
+function pushFrame(expr: string, limits?: Partial<RpcLimits>): ScriptedTransport {
+  let transport = new ScriptedTransport([`["push",${expr}]`]);
+  new RpcSession(transport, new EchoTarget(), limits ? { limits } : undefined);
+  return transport;
+}
+
+class EchoTarget extends RpcTarget {
+  echo(value: unknown) { return value; }
+}
+
+describe("bigint deserialization limits", () => {
+  it("accepts a bigint at the default digit limit", () => {
+    let digits = "9".repeat(DEFAULT_LIMITS.maxBigIntDigits);
+    let value = deserialize(`["bigint","${digits}"]`);
+    expect(value).toBe(BigInt(digits));
+  });
+
+  it("rejects a bigint over the default digit limit", () => {
+    let digits = "9".repeat(DEFAULT_LIMITS.maxBigIntDigits + 1);
+    expect(() => deserialize(`["bigint","${digits}"]`))
+        .toThrowError(/bigint exceeds maximum length/);
+  });
+
+  it("rejects a non-numeric bigint string before calling BigInt()", () => {
+    // These are all forms BigInt() would otherwise accept (hex/binary) or choke on; the numeric
+    // pre-validation rejects anything that isn't an optional sign plus decimal digits.
+    expect(() => deserialize('["bigint","0x10"]')).toThrowError(/invalid characters/);
+    expect(() => deserialize('["bigint","12n"]')).toThrowError(/invalid characters/);
+    expect(() => deserialize('["bigint","1 2"]')).toThrowError(/invalid characters/);
+    expect(() => deserialize('["bigint",""]')).toThrowError(/invalid characters/);
+    expect(() => deserialize('["bigint","abc"]')).toThrowError(/invalid characters/);
+  });
+
+  it("accepts a normal negative bigint", () => {
+    expect(deserialize('["bigint","-123"]')).toBe(-123n);
+  });
+
+  it("honors a per-session override of maxBigIntDigits", async () => {
+    let underLimit = `["bigint","${"9".repeat(4)}"]`;
+    let overLimit = `["bigint","${"9".repeat(5)}"]`;
+
+    // With an override of 4 digits, a 4-digit bigint is fine but a 5-digit one aborts the session.
+    await pushFrame(underLimit, { maxBigIntDigits: 4 }).expectNoAbort();
+
+    let aborted = await pushFrame(overLimit, { maxBigIntDigits: 4 }).waitForAbort();
+    expect(aborted).toMatch(/bigint exceeds maximum length/);
+
+    // The same 5-digit bigint is well under the default limit, so it does NOT abort by default.
+    await pushFrame(overLimit).expectNoAbort();
+  });
+});
+
+describe("message depth limits", () => {
+  it("accepts nesting just under the default depth limit", () => {
+    // The top value sits at depth 0, so maxDepth-1 escaped-array layers stay within the limit.
+    let value = deserialize(nestedEscapedArrays(DEFAULT_LIMITS.maxDepth - 1));
+    expect(value).toBeInstanceOf(Array);
+  });
+
+  it("rejects nesting over the default depth limit", () => {
+    expect(() => deserialize(nestedEscapedArrays(DEFAULT_LIMITS.maxDepth + 5)))
+        .toThrowError(/maximum allowed message depth/);
+  });
+
+  it("honors a per-session override of maxDepth", async () => {
+    // Override depth to a small value; nesting beyond it aborts the session.
+    let deep = nestedEscapedArrays(10);
+    let aborted = await pushFrame(deep, { maxDepth: 4 }).waitForAbort();
+    expect(aborted).toMatch(/maximum allowed message depth/);
+
+    // The same depth is well within the default, so by default it does not abort.
+    await pushFrame(deep).expectNoAbort();
+  });
+});
+
+describe("message size limits", () => {
+  it("aborts the session on a message over the size limit", async () => {
+    // The pushed expression is a long-but-valid JSON string literal, so the resulting frame is
+    // valid JSON and well over the 64-char override. The size check runs (and aborts) before
+    // JSON.parse, so this exercises the size guard specifically, not a parse error.
+    let aborted = await pushFrame(`"${"a".repeat(200)}"`, { maxMessageSize: 64 }).waitForAbort();
+    expect(aborted).toMatch(/exceeds maximum size/);
+  });
+
+  it("does not abort on a message under the size limit", async () => {
+    await pushFrame('"small"', { maxMessageSize: 1024 }).expectNoAbort();
+  });
+});
+
+describe("limits backwards compatibility", () => {
+  it("DEFAULT_LIMITS has the documented values", () => {
+    expect(DEFAULT_LIMITS).toStrictEqual({
+      maxBigIntDigits: 16384,
+      maxDepth: 64,
+      maxMessageSize: 32 * 1024 * 1024,
+    });
+  });
+
+  it("standalone deserialize round-trips normal values unchanged with no config", () => {
+    expect(deserialize('["bigint","123"]')).toBe(123n);
+    expect(deserialize(serialize(12345678901234567890n))).toBe(12345678901234567890n);
+    expect(deserialize('[[[[123,456]]]]')).toStrictEqual([[123, 456]]);
+    expect(deserialize('{"foo":{"bar":123}}')).toStrictEqual({foo: {bar: 123}});
+  });
+
+  it("a session constructed with no options behaves exactly as before", async () => {
+    // A legitimate bigint, normal nesting, and a normal call all flow through a default session
+    // without aborting.
+    using harness = new SessionPair(new EchoTarget());
+
+    let stub = harness.stub as any;
+    expect(await stub.echo(123n)).toBe(123n);
+    expect(await stub.echo([[1, 2, 3]])).toStrictEqual([[1, 2, 3]]);
+    expect(await stub.echo({a: {b: {c: 1}}})).toStrictEqual({a: {b: {c: 1}}});
+
+    await harness.dispose();
+  });
+
+  it("a session constructed with an empty options object behaves identically", async () => {
+    using harness = new SessionPair(new EchoTarget(), {});
+    let stub = harness.stub as any;
+    expect(await stub.echo(999n)).toBe(999n);
+    await harness.dispose();
+  });
+});
+
+// A minimal in-memory bidirectional session pair, sufficient for the round-trip checks above.
+class PairTransport implements RpcTransport {
+  partner!: PairTransport;
+  private queue: string[] = [];
+  private waiter?: () => void;
+
+  async send(message: string): Promise<void> {
+    this.partner.queue.push(message);
+    this.partner.waiter?.();
+    this.partner.waiter = undefined;
+  }
+
+  async receive(): Promise<string> {
+    while (this.queue.length === 0) {
+      await new Promise<void>(resolve => { this.waiter = resolve; });
+    }
+    return this.queue.shift()!;
+  }
+}
+
+class SessionPair {
+  private clientTransport = new PairTransport();
+  private serverTransport = new PairTransport();
+  private client: RpcSession;
+  private server: RpcSession;
+  stub: ReturnType<RpcSession["getRemoteMain"]>;
+
+  constructor(target: RpcTarget, options?: { limits?: Partial<RpcLimits> }) {
+    this.clientTransport.partner = this.serverTransport;
+    this.serverTransport.partner = this.clientTransport;
+    this.client = new RpcSession(this.clientTransport);
+    this.server = new RpcSession(this.serverTransport, target, options);
+    this.stub = this.client.getRemoteMain();
+  }
+
+  async dispose() {
+    for (let i = 0; i < 16; i++) await Promise.resolve();
+  }
+
+  [Symbol.dispose]() {}
+}

src/index.ts

@@ -5,6 +5,7 @@
 import { RpcTarget as RpcTargetImpl, RpcStub as RpcStubImpl, RpcPromise as RpcPromiseImpl } from "./core.js";
 import { serialize, deserialize } from "./serialize.js";
 import { RpcTransport, RpcSession as RpcSessionImpl, RpcSessionOptions } from "./rpc.js";
+import { RpcLimits, DEFAULT_LIMITS } from "./serialize.js";
 import { RpcTargetBranded, RpcCompatible, Stub, Stubify, __RPC_TARGET_BRAND } from "./types.js";
 import { newWebSocketRpcSession as newWebSocketRpcSessionImpl,
          newWorkersWebSocketRpcResponse } from "./websocket.js";
@@ -19,8 +20,8 @@ forceInitStreams();
 
 // Re-export public API types.
 export { serialize, deserialize, newWorkersWebSocketRpcResponse, newHttpBatchRpcResponse,
-         nodeHttpBatchRpcResponse };
-export type { RpcTransport, RpcSessionOptions, RpcCompatible };
+         nodeHttpBatchRpcResponse, DEFAULT_LIMITS };
+export type { RpcTransport, RpcSessionOptions, RpcCompatible, RpcLimits };
 
 // Hack the type system to make RpcStub's types work nicely!
 /**
@@ -143,17 +144,18 @@ export let newMessagePortRpcSession:<T extends RpcCompatible<T> = Empty>
  * credentials as parameters and returns the authorized API), then cross-origin requests should
  * be safe.
  */
-export async function newWorkersRpcResponse(request: Request, localMain: any) {
+export async function newWorkersRpcResponse(
+    request: Request, localMain: any, options?: RpcSessionOptions) {
   if (request.method === "POST") {
-    let response = await newHttpBatchRpcResponse(request, localMain);
+    let response = await newHttpBatchRpcResponse(request, localMain, options);
     // Since we're exposing the same API over WebSocket, too, and WebSocket always allows
     // cross-origin requests, the API necessarily must be safe for cross-origin use (e.g. because
     // it uses in-band authorization, as recommended in the readme). So, we might as well allow
     // batch requests to be made cross-origin as well.
     response.headers.set("Access-Control-Allow-Origin", "*");
     return response;
   } else if (request.headers.get("Upgrade")?.toLowerCase() === "websocket") {
-    return newWorkersWebSocketRpcResponse(request, localMain);
+    return newWorkersWebSocketRpcResponse(request, localMain, options);
   } else {
     return new Response("This endpoint only accepts POST or WebSocket requests.", { status: 400 });
   }

src/map.ts

@@ -3,7 +3,7 @@
 //     https://opensource.org/license/mit
 
 import { StubHook, PropertyPath, RpcPayload, RpcStub, RpcPromise, withCallInterceptor, ErrorStubHook, mapImpl, PayloadStubHook, unwrapStubAndPath, unwrapStubNoProperties } from "./core.js";
-import { Devaluator, Exporter, Importer, ExportId, ImportId, Evaluator } from "./serialize.js";
+import { Devaluator, Exporter, Importer, ExportId, ImportId, Evaluator, RpcLimits, DEFAULT_LIMITS } from "./serialize.js";
 
 let currentMapBuilder: MapBuilder | undefined;
 
@@ -296,6 +296,14 @@ class MapApplicator implements Importer {
   getPipeReadable(exportId: ExportId): never {
     throw new Error("A mapper function cannot use pipe readables.");
   }
+
+  getLimits(): RpcLimits {
+    // A mapper's instructions arrived inside a ["remap"] message that was itself deserialized
+    // under the session's limits, so the default floor is sufficient protection here without
+    // threading per-session overrides through the several layers between the session and the
+    // mapper.
+    return DEFAULT_LIMITS;
+  }
 }
 
 function applyMapToElement(input: unknown, parent: object | undefined, owner: RpcPayload | null,

src/rpc.ts

@@ -3,7 +3,7 @@
 //     https://opensource.org/license/mit
 
 import { StubHook, RpcPayload, RpcStub, PropertyPath, PayloadStubHook, ErrorStubHook, RpcTarget, unwrapStubAndPath, streamImpl } from "./core.js";
-import { Devaluator, Evaluator, ExportId, ImportId, Exporter, Importer, serialize } from "./serialize.js";
+import { Devaluator, Evaluator, ExportId, ImportId, Exporter, Importer, serialize, RpcLimits, DEFAULT_LIMITS } from "./serialize.js";
 
 /**
  * Interface for an RPC transport, which is a simple bidirectional message stream. Implement this
@@ -316,6 +316,17 @@ export type RpcSessionOptions = {
    * to serialize the error with the stack omitted.
    */
   onSendError?: (error: Error) => Error | void;
+
+  /**
+   * Overrides for the resource limits enforced while deserializing messages from the peer. Any
+   * field left unset falls back to `DEFAULT_LIMITS`. These guard against resource-exhaustion
+   * attacks from untrusted peers; see `RpcLimits` for the meaning and defaults of each field.
+   *
+   * Limits are a purely local, receiver-side decision -- the protocol has no negotiation step, so
+   * the peer never learns these values. A message that exceeds a limit is rejected, aborting the
+   * session.
+   */
+  limits?: Partial<RpcLimits>;
 };
 
 class RpcSessionImpl implements Importer, Exporter {
@@ -340,8 +351,14 @@ class RpcSessionImpl implements Importer, Exporter {
   // may be deleted from the middle (hence leaving the array sparse).
   onBrokenCallbacks: ((error: any) => void)[] = [];
 
+  // Resource limits enforced on incoming messages, resolved once from the defaults plus any
+  // per-session overrides.
+  private limits: RpcLimits;
+
   constructor(private transport: RpcTransport, mainHook: StubHook,
       private options: RpcSessionOptions) {
+    this.limits = { ...DEFAULT_LIMITS, ...options.limits };
+
     // Export zero is automatically the bootstrap object.
     this.exports.push({hook: mainHook, refcount: 1});
 
@@ -534,6 +551,10 @@ class RpcSessionImpl implements Importer, Exporter {
     return readable;
   }
 
+  getLimits(): RpcLimits {
+    return this.limits;
+  }
+
   createPipe(readable: ReadableStream, readableHook: StubHook): ImportId {
     if (this.abortReason) throw this.abortReason;
 
@@ -746,6 +767,14 @@ class RpcSessionImpl implements Importer, Exporter {
         }
       }
 
+      // Bound the up-front allocation of a single message before parsing it. This is a local,
+      // receiver-side guard against an untrusted peer sending an enormous message; a throw here
+      // propagates out of readLoop and aborts the session.
+      if (msgText.length > this.limits.maxMessageSize) {
+        throw new TypeError(
+            `Incoming message exceeds maximum size of ${this.limits.maxMessageSize} characters.`);
+      }
+
       let msg = JSON.parse(msgText);
       if (this.abortReason) break;  // check again before processing