Cap the total time of a self-hosted proxy request

infra/cors-worker/README.md

@@ -178,6 +178,11 @@ allowlist). `OPTIONS` answers preflight.
   leaves no window for DNS to change between check and connect, and each
   redirect hop goes through the same lookup.
 - **Size cap.** Responses advertising more than 25 MiB are refused with `413`.
+  The Node shell also cuts off a chunked body once it passes 25 MiB.
+- **Timeouts (Node shell).** An upstream connection is dropped after 30 s of
+  inactivity, and the whole exchange (request to the body's end) after 120 s,
+  so an upstream that trickles bytes can't hold a connection open forever.
+  (On Cloudflare, the platform enforces its own limits.)
 - **Log hygiene.** Only errors are logged, and only the hostname — never the
   full `?url=` (which can carry API keys). The old per-request logging was also
   needlessly expensive.

infra/cors-worker/guarded-fetch.js

@@ -24,6 +24,9 @@ import { Readable } from "node:stream";
 import { isBlockedAddress } from "./ssrf-policy.js";
 
 const DEFAULT_TIMEOUT_MS = 30_000;
+// Inactivity alone can be dodged by an upstream sending a byte every few
+// seconds, which would hold a socket (and a client) open indefinitely.
+const DEFAULT_DEADLINE_MS = 120_000;
 // Matches core's MAX_BYTES. Core can only refuse an *advertised* oversize
 // length; this also cuts off a chunked body that keeps growing.
 const DEFAULT_MAX_BYTES = 25 * 1024 * 1024;
@@ -80,12 +83,14 @@ export function createGuardedLookup({ lookup = dns.lookup, isBlocked = isBlocked
  *
  * `lookup` (a `dns.lookup`-shaped resolver) and `isBlocked` are injectable for
  * tests; `timeoutMs` bounds socket inactivity, before and during the body;
+ * `deadlineMs` bounds the whole exchange, from request to the body's end;
  * `maxBytes` cuts the body off (the stream errors) once it exceeds the cap.
  */
 export function createGuardedFetch({
   lookup = dns.lookup,
   isBlocked = isBlockedAddress,
   timeoutMs = DEFAULT_TIMEOUT_MS,
+  deadlineMs = DEFAULT_DEADLINE_MS,
   maxBytes = DEFAULT_MAX_BYTES,
 } = {}) {
   const guardedLookup = createGuardedLookup({ lookup, isBlocked });
@@ -109,8 +114,21 @@ export function createGuardedFetch({
       const request = client.request(url, { method, headers, agent, lookup: guardedLookup });
 
       request.setTimeout(timeoutMs, () => request.destroy(new Error("upstream timeout")));
-      request.on("error", reject);
+      // Before the response, the deadline fails the request; after it, the
+      // body stream, so the reason reaches whoever is reading it.
+      let inFlight = request;
+      const deadline = setTimeout(
+        () => inFlight.destroy(new Error("upstream deadline timeout")),
+        deadlineMs
+      );
+      deadline.unref();
+      request.on("error", (error) => {
+        clearTimeout(deadline);
+        reject(error);
+      });
       request.on("response", (upstream) => {
+        inFlight = upstream;
+        upstream.on("close", () => clearTimeout(deadline));
         try {
           const responseHeaders = new Headers();
           const raw = upstream.rawHeaders;

infra/cors-worker/guarded-fetch.test.mjs

@@ -142,6 +142,45 @@ test("times out a stalled upstream", async () => {
   }
 });
 
+test("a total deadline ends an upstream that trickles to dodge the inactivity timeout", async () => {
+  let timer;
+  const server = await startServer((req, res) => {
+    res.writeHead(200, { "content-type": "text/plain" });
+    timer = setInterval(() => res.write("."), 20); // never idle long enough to time out
+  });
+  try {
+    const { lookup } = fakeLookup({ "drip.test": ["127.0.0.1"] });
+    const guardedFetch = createGuardedFetch({
+      lookup,
+      isBlocked: allowLoopbackForTest,
+      timeoutMs: 1000,
+      deadlineMs: 150,
+    });
+    const response = await guardedFetch(`http://drip.test:${server.port}/`);
+    assert.equal(response.status, 200);
+    await assert.rejects(response.text(), /deadline/);
+  } finally {
+    clearInterval(timer);
+    await server.close();
+  }
+});
+
+test("the deadline also covers an upstream that never answers", async () => {
+  const server = await startServer(() => {});
+  try {
+    const { lookup } = fakeLookup({ "slow.test": ["127.0.0.1"] });
+    const guardedFetch = createGuardedFetch({
+      lookup,
+      isBlocked: allowLoopbackForTest,
+      timeoutMs: 5000,
+      deadlineMs: 50,
+    });
+    await assert.rejects(guardedFetch(`http://slow.test:${server.port}/`), /deadline/);
+  } finally {
+    await server.close();
+  }
+});
+
 test("cuts off a body that grows past the byte cap without advertising a length", async () => {
   const server = await startServer((req, res) => {
     res.writeHead(200, { "content-type": "application/octet-stream" }); // chunked