fix(browser): bound CDP discovery response size and lifetime

src/browser/CdpEndpoint.ts

@@ -1,4 +1,4 @@
-import { request } from "node:http";
+import { request, type ClientRequest, type IncomingMessage } from "node:http";
 
 import { z } from "zod";
 
@@ -29,6 +29,17 @@ const endpointTargetSchema = z.object({
 });
 const endpointTargetsSchema = z.array(endpointTargetSchema);
 
+/** Per-response bounds for loopback CDP discovery documents. */
+const CDP_DISCOVERY_RESPONSE_LIMITS = {
+  bytes: 8 * 1024 * 1024,
+  timeoutMs: 10_000,
+} as const;
+
+interface CdpDiscoveryResponseLimits {
+  readonly bytes: number;
+  readonly timeoutMs: number;
+}
+
 /** Validated direct CDP WebSocket bound to one discovered page target. */
 interface CdpPageWebSocketEndpoint {
   readonly scope: "page";
@@ -291,29 +302,71 @@ export const readCdpJson = async (
   url: URL,
   operation: BrowserObservationOperation,
   signal?: AbortSignal,
+  limits: CdpDiscoveryResponseLimits = CDP_DISCOVERY_RESPONSE_LIMITS,
 ): Promise<unknown> =>
   await new Promise((resolve, reject) => {
     if (signal?.aborted === true) {
       reject(new AnalysisCancelledError(operation));
       return;
     }
-    const request_ = request(
+    let response: IncomingMessage | undefined;
+    let request_: ClientRequest | undefined;
+    let timer: NodeJS.Timeout | undefined;
+    let settled = false;
+    const removeAbort = (): void =>
+      signal?.removeEventListener("abort", onAbort);
+    const finish = (error?: Error, value?: unknown): void => {
+      if (settled) return;
+      settled = true;
+      if (timer !== undefined) clearTimeout(timer);
+      removeAbort();
+      if (error !== undefined) {
+        reject(error);
+        response?.destroy();
+        request_?.destroy();
+      } else resolve(value);
+    };
+    const onAbort = (): void => {
+      if (signal?.aborted === true)
+        finish(new AnalysisCancelledError(operation));
+    };
+    request_ = request(
       url,
-      { method: "GET", signal, headers: { Accept: "application/json" } },
-      (response) => {
-        if (response.statusCode !== 200) {
-          response.resume();
-          reject(
+      { method: "GET", headers: { Accept: "application/json" } },
+      (received) => {
+        response = received;
+        if (settled) {
+          received.destroy();
+          return;
+        }
+        if (received.statusCode !== 200) {
+          finish(
             new BrowserObservationError(operation, "invalid_endpoint_response"),
           );
           return;
         }
         const chunks: Buffer[] = [];
-        response.on("data", (chunk: Buffer) => chunks.push(chunk));
-        response.on("end", () => {
-          const parsed = safeParseJson(Buffer.concat(chunks).toString("utf8"));
+        let bytes = 0;
+        received.on("data", (chunk: Buffer) => {
+          if (settled) return;
+          bytes += chunk.byteLength;
+          if (bytes > limits.bytes) {
+            finish(
+              new BrowserObservationError(operation, "payload_limit", {
+                detail: `CDP discovery response exceeds its ${String(limits.bytes)} byte limit.`,
+              }),
+            );
+            return;
+          }
+          chunks.push(chunk);
+        });
+        received.on("end", () => {
+          if (settled) return;
+          const parsed = safeParseJson(
+            Buffer.concat(chunks, bytes).toString("utf8"),
+          );
           if (!parsed.ok) {
-            reject(
+            finish(
               new BrowserObservationError(
                 operation,
                 "invalid_endpoint_response",
@@ -322,15 +375,41 @@ export const readCdpJson = async (
             );
             return;
           }
-          resolve(parsed.value);
+          finish(undefined, parsed.value);
         });
-        response.on("error", (cause: unknown) =>
-          reject(endpointFailure(cause, operation, signal)),
+        received.on("aborted", () =>
+          finish(
+            new BrowserObservationError(operation, "invalid_endpoint_response"),
+          ),
+        );
+        received.on("close", () => {
+          if (!settled && !received.complete)
+            finish(
+              new BrowserObservationError(
+                operation,
+                "invalid_endpoint_response",
+              ),
+            );
+        });
+        received.on("error", (cause: unknown) =>
+          finish(endpointFailure(cause, operation, signal)),
         );
       },
     );
     request_.on("error", (cause: unknown) =>
-      reject(endpointFailure(cause, operation, signal)),
+      finish(endpointFailure(cause, operation, signal)),
+    );
+    signal?.addEventListener("abort", onAbort, { once: true });
+    onAbort();
+    if (settled) return;
+    timer = setTimeout(
+      () =>
+        finish(
+          new BrowserObservationError(operation, "timeout", {
+            detail: `CDP discovery response exceeded its ${String(limits.timeoutMs)} ms deadline.`,
+          }),
+        ),
+      limits.timeoutMs,
     );
     request_.end();
   });

tests/boundary/browser/cdpEndpointResponseBounds.test.ts

@@ -0,0 +1,158 @@
+import { createServer } from "node:http";
+import type { RequestListener, Server } from "node:http";
+
+import { afterEach, describe, expect, it } from "vitest";
+
+import { readCdpJson } from "../../../src/browser/CdpEndpoint.js";
+
+const servers: Server[] = [];
+
+afterEach(async () => {
+  await Promise.all(
+    servers.splice(0).map(
+      (server) =>
+        new Promise<void>((resolve, reject) => {
+          server.closeAllConnections();
+          server.close((error) =>
+            error === undefined ? resolve() : reject(error),
+          );
+        }),
+    ),
+  );
+});
+
+describe("CDP discovery response bounds", () => {
+  it("accepts a complete JSON response exactly at the byte limit", async () => {
+    const body = JSON.stringify("x".repeat(14));
+    expect(Buffer.byteLength(body)).toBe(16);
+    const server = await startServer((_request, response) =>
+      response.end(body),
+    );
+
+    await expect(
+      readCdpJson(server.url, "list_browser_targets", undefined, {
+        bytes: 16,
+        timeoutMs: 1_000,
+      }),
+    ).resolves.toBe("x".repeat(14));
+  });
+
+  it("rejects and closes an oversized chunked response without parsing a prefix", async () => {
+    let responseClosed: (() => void) | undefined;
+    const closed = new Promise<void>((resolve) => {
+      responseClosed = resolve;
+    });
+    const server = await startServer((_request, response) => {
+      response.once("close", () => responseClosed?.());
+      response.write(JSON.stringify("x".repeat(15)));
+    });
+
+    await expect(
+      readCdpJson(server.url, "list_browser_targets", undefined, {
+        bytes: 16,
+        timeoutMs: 1_000,
+      }),
+    ).rejects.toMatchObject({
+      _tag: "BrowserObservationError",
+      reason: "payload_limit",
+    });
+    await closed;
+  });
+
+  it("ends a response that exceeds the total elapsed deadline", async () => {
+    let responseClosed: (() => void) | undefined;
+    let requestStarted: (() => void) | undefined;
+    const closed = new Promise<void>((resolve) => {
+      responseClosed = resolve;
+    });
+    const started = new Promise<void>((resolve) => {
+      requestStarted = resolve;
+    });
+    const server = await startServer((_request, response) => {
+      response.once("close", () => responseClosed?.());
+      requestStarted?.();
+    });
+    const pending = readCdpJson(server.url, "list_browser_targets", undefined, {
+      bytes: 16,
+      timeoutMs: 50,
+    });
+    const timeoutAssertion = expect(pending).rejects.toMatchObject({
+      _tag: "BrowserObservationError",
+      reason: "timeout",
+    });
+    await started;
+
+    await timeoutAssertion;
+    await closed;
+  });
+
+  it("preserves caller cancellation and closes the response", async () => {
+    let responseClosed: (() => void) | undefined;
+    let requestStarted: (() => void) | undefined;
+    const closed = new Promise<void>((resolve) => {
+      responseClosed = resolve;
+    });
+    const started = new Promise<void>((resolve) => {
+      requestStarted = resolve;
+    });
+    const server = await startServer((_request, response) => {
+      response.once("close", () => responseClosed?.());
+      requestStarted?.();
+    });
+    const controller = new AbortController();
+    const pending = readCdpJson(
+      server.url,
+      "list_browser_targets",
+      controller.signal,
+      {
+        bytes: 16,
+        timeoutMs: 1_000,
+      },
+    );
+    await started;
+    controller.abort();
+
+    await expect(pending).rejects.toMatchObject({
+      _tag: "AnalysisCancelledError",
+      operation: "list_browser_targets",
+    });
+    await closed;
+  });
+
+  it("destroys an unsuccessful HTTP response", async () => {
+    let responseClosed: (() => void) | undefined;
+    const closed = new Promise<void>((resolve) => {
+      responseClosed = resolve;
+    });
+    const server = await startServer((_request, response) => {
+      response.once("close", () => responseClosed?.());
+      response.writeHead(503).flushHeaders();
+    });
+
+    await expect(
+      readCdpJson(server.url, "list_browser_targets", undefined, {
+        bytes: 16,
+        timeoutMs: 1_000,
+      }),
+    ).rejects.toMatchObject({
+      _tag: "BrowserObservationError",
+      reason: "invalid_endpoint_response",
+    });
+    await closed;
+  });
+});
+
+const startServer = async (
+  handler: RequestListener,
+): Promise<{ readonly url: URL }> => {
+  const server = createServer(handler);
+  servers.push(server);
+  await new Promise<void>((resolve, reject) => {
+    server.once("error", reject);
+    server.listen(0, "127.0.0.1", resolve);
+  });
+  const address = server.address();
+  if (address === null || typeof address === "string")
+    throw new TypeError("Expected a TCP listener address");
+  return { url: new URL(`http://127.0.0.1:${String(address.port)}`) };
+};