Fetch refinements

.github/workflows/test.yml

@@ -44,4 +44,6 @@ jobs:
       - run: pnpm build
       - name: Test native HTTP code path
         run: pnpm test-native
+      - name: Test fetch HTTP code path
+        run: pnpm test-fetch
 

README.md

@@ -641,7 +641,7 @@ const proxy = createProxyServer({
 
 - **ca**: Optionally override the trusted CA certificates. This is passed to https.request.
 
-- **fetchOptions**: Enable fetch API for HTTP/2 support. Set to `true` for defaults, or provide custom configuration:
+- **fetchOptions**: Enable fetch API for HTTP/2 support. Provide an object of type `FetchOptions` for custom configuration:
   - `requestOptions`: Additional fetch request options (e.g., undici Agent with `allowH2: true` for HTTP/2 as dispatcher)
   - `onBeforeRequest`: Async callback called before making the fetch request
   - `onAfterResponse`: Async callback called after receiving the fetch response
@@ -665,7 +665,7 @@ The following table shows which configuration options are compatible with differ
 |--------|-------------------|---------------|--------|
 | `target` | ✅ | ✅ | Core option, works in both paths |
 | `forward` | ✅ | ✅ | Core option, works in both paths |
-| `agent` | ✅ | ❌ | Native agents only, use `fetch.dispatcher` instead |
+| `agent` | ✅ | ❌ | Native agents only |
 | `ssl` | ✅ | ✅ | HTTPS server configuration |
 | `ws` | ✅ | ❌ | WebSocket proxying uses native path only |
 | `xfwd` | ✅ | ✅ | X-Forwarded headers |
@@ -674,8 +674,8 @@ The following table shows which configuration options are compatible with differ
 | `prependPath` | ✅ | ✅ | Path manipulation |
 | `ignorePath` | ✅ | ✅ | Path manipulation |
 | `localAddress` | ✅ | ✅ | Local interface binding |
-| `changeOrigin` | ✅ | ✅ | Host header rewriting |
-| `preserveHeaderKeyCase` | ✅ | ✅ | Header case preservation |
+| `changeOrigin` | ✅ | ❌ | Host header rewriting |
+| `preserveHeaderKeyCase` | ✅ | ❌ | Header case preservation |
 | `auth` | ✅ | ✅ | Basic authentication |
 | `hostRewrite` | ✅ | ✅ | Redirect hostname rewriting |
 | `autoRewrite` | ✅ | ✅ | Automatic redirect rewriting |
@@ -693,7 +693,7 @@ The following table shows which configuration options are compatible with differ
 | `fetch` | ❌ | ✅ | Fetch-specific configuration |
 
 **Notes:**
-- ¹ `secure` is not directly supported in the fetch path. Instead, use `fetch.dispatcher` with `{connect: {rejectUnauthorized: false}}` to disable SSL certificate verification (e.g., for self-signed certificates).
+- ¹ `secure` is not directly supported in the fetch path. Instead, use a custom dispatcher with `{rejectUnauthorized: false}` to disable SSL certificate verification (e.g., for self-signed certificates).
 
 **Code Path Selection:**
 - **Native Path**: Used by default, supports HTTP/1.1 and WebSockets

lib/http-proxy/index.ts

@@ -97,9 +97,14 @@ export interface ServerOptions {
    * This is passed to https.request.
    */
   ca?: string;
-  /** Enable using fetch for proxy requests. Set to true for defaults, or provide custom configuration. */
-  fetchOptions?: FetchOptions;
-  fetch?: typeof fetch;
+  /** Optional fetch implementation to use instead of global fetch, use this to activate fetch-based proxying,
+   * for example to proxy HTTP/2 requests
+  */
+ fetch?: typeof fetch;
+  /** Optional configuration object for fetch-based proxy requests. 
+   * Use this to customize fetch request and response handling. 
+   * For custom fetch implementations, use the `fetch` property.*/
+ fetchOptions?: FetchOptions;
 }
 export interface FetchOptions {
   /** Fetch request options */

lib/http-proxy/passes/web-incoming.ts

@@ -69,7 +69,7 @@ export function XHeaders(req: Request, _res: Response, options: ServerOptions) {
       (req.headers["x-forwarded-" + header] || "") + (req.headers["x-forwarded-" + header] ? "," : "") + values[header];
   }
 
-  req.headers["x-forwarded-host"] = req.headers["x-forwarded-host"] || req.headers["host"] || "";
+  req.headers["x-forwarded-host"] = req.headers["x-forwarded-host"] || req.headers["host"] || req.headers[":authority"] || "";
 }
 
 // Does the actual proxying. If `forward` is enabled fires up
@@ -208,6 +208,12 @@ async function stream2(
 ) {
   // Helper function to handle errors consistently throughout the fetch path
   const handleError = (err: Error, target?: ProxyTargetUrl) => {
+    const e = err as any;
+    // Copy code from cause if available and missing on err
+    if (e.code === undefined && e.cause?.code) {
+      e.code = e.cause.code;
+    }
+
     if (cb) {
       cb(err, req, res, target);
     } else {
@@ -227,26 +233,72 @@ async function stream2(
   });
 
   const customFetch = options.fetch || fetch;
-
   const fetchOptions = options.fetchOptions ?? {} as FetchOptions;
 
+  const controller = new AbortController();
+  const { signal } = controller;
 
-  if (options.forward) {
-    const outgoingOptions = common.setupOutgoing(options.ssl || {}, options, req, "forward");
+  if (options.proxyTimeout) {
+    setTimeout(() => {
+      controller.abort();
+    }, options.proxyTimeout);
+  }
 
+  // Ensure we abort proxy if request is aborted
+  res.on("close", () => {
+    const aborted = !res.writableFinished;
+    if (aborted) {
+      controller.abort();
+    }
+  });
+
+  const prepareRequest = (outgoing: common.Outgoing) => {
     const requestOptions: RequestInit = {
-      method: outgoingOptions.method,
+      method: outgoing.method,
+      signal,
+      ...fetchOptions.requestOptions,
     };
 
+    const headers = new Headers(fetchOptions.requestOptions?.headers);
+
+    if (!fetchOptions.requestOptions?.headers && outgoing.headers) {
+      for (const [key, value] of Object.entries(outgoing.headers)) {
+        if (typeof key === "string") {
+          if (Array.isArray(value)) {
+            for (const v of value) {
+              headers.append(key, v as string);
+            }
+          } else if (value != null) {
+            headers.append(key, value as string);
+          }
+        }
+      }
+    }
+
+    if (options.auth) {
+      headers.set("authorization", `Basic ${Buffer.from(options.auth).toString("base64")}`);
+    }
+
+    requestOptions.headers = headers;
 
-    // Handle request body
     if (options.buffer) {
       requestOptions.body = options.buffer as Stream.Readable;
     } else if (req.method !== "GET" && req.method !== "HEAD") {
       requestOptions.body = req;
       requestOptions.duplex = "half";
     }
 
+    return requestOptions;
+  };
+
+  if (options.forward) {
+    const outgoingOptions = common.setupOutgoing(options.ssl || {}, options, req, "forward");
+    const requestOptions = prepareRequest(outgoingOptions);
+    let targetUrl = new URL(outgoingOptions.url).origin + outgoingOptions.path;
+    if (targetUrl.startsWith("ws")) {
+      targetUrl = targetUrl.replace("ws", "http");
+    }
+
     // Call onBeforeRequest callback before making the forward request
     if (fetchOptions.onBeforeRequest) {
       try {
@@ -258,7 +310,7 @@ async function stream2(
     }
 
     try {
-      const result = await customFetch(new URL(outgoingOptions.url).origin + outgoingOptions.path, requestOptions);
+      const result = await customFetch(targetUrl, requestOptions);
 
       // Call onAfterResponse callback for forward requests (though they typically don't expect responses)
       if (fetchOptions.onAfterResponse) {
@@ -270,6 +322,16 @@ async function stream2(
         }
       }
     } catch (err) {
+      if ((err as Error).name === "AbortError") {
+        // Handle aborts (timeout or client disconnect)
+        if (options.proxyTimeout && signal.aborted) {
+          const proxyTimeoutErr = new Error("Proxy timeout");
+          (proxyTimeoutErr as any).code = "ECONNRESET";
+          handleError(proxyTimeoutErr, options.forward);
+        }
+        // If aborted by client (res.close), we might not want to emit an error or maybe just log it
+        return;
+      }
       handleError(err as Error, options.forward);
     }
 
@@ -279,30 +341,10 @@ async function stream2(
   }
 
   const outgoingOptions = common.setupOutgoing(options.ssl || {}, options, req);
-
-  // Remove symbols from headers
-  const requestOptions: RequestInit = {
-    method: outgoingOptions.method,
-    headers: Object.fromEntries(
-      Object.entries(outgoingOptions.headers || {}).filter(([key, _value]) => {
-        return typeof key === "string";
-      }),
-    ) as RequestInit["headers"],
-    ...fetchOptions.requestOptions,
-  };
-
-  if (options.auth) {
-    requestOptions.headers = {
-      ...requestOptions.headers,
-      authorization: `Basic ${Buffer.from(options.auth).toString("base64")}`,
-    };
-  }
-
-  if (options.buffer) {
-    requestOptions.body = options.buffer as Stream.Readable;
-  } else if (req.method !== "GET" && req.method !== "HEAD") {
-    requestOptions.body = req;
-    requestOptions.duplex = "half";
+  const requestOptions = prepareRequest(outgoingOptions);
+  let targetUrl = new URL(outgoingOptions.url).origin + outgoingOptions.path;
+  if (targetUrl.startsWith("ws")) {
+    targetUrl = targetUrl.replace("ws", "http");
   }
 
   // Call onBeforeRequest callback before making the request
@@ -316,7 +358,7 @@ async function stream2(
   }
 
   try {
-    const response = await customFetch(new URL(outgoingOptions.url).origin + outgoingOptions.path, requestOptions);
+    const response = await customFetch(targetUrl, requestOptions);
 
     // Call onAfterResponse callback after receiving the response
     if (fetchOptions.onAfterResponse) {
@@ -385,6 +427,14 @@ async function stream2(
       server?.emit("end", req, res, fakeProxyRes);
     }
   } catch (err) {
+    if ((err as Error).name === "AbortError") {
+      if (options.proxyTimeout && signal.aborted) {
+        const proxyTimeoutErr = new Error("Proxy timeout");
+        (proxyTimeoutErr as any).code = "ECONNRESET";
+        handleError(proxyTimeoutErr, options.target);
+      }
+      return;
+    }
     handleError(err as Error, options.target);
   }
 }

lib/test/http/customFetch.test.ts

@@ -1,5 +1,5 @@
 /*
-pnpm test proxy-http2-to-http2.test.ts
+pnpm test customFetch.test.ts
 
 */
 

lib/test/http/fetch-timeout.test.ts

@@ -0,0 +1,62 @@
+
+import * as http from "node:http";
+import * as httpProxy from "../..";
+import getPort from "../get-port";
+import { describe, it, expect, beforeAll, afterAll } from "vitest";
+import { fetch } from "undici";
+
+describe("Fetch Proxy Timeout", () => {
+  let ports: Record<"http" | "proxy", number>;
+  beforeAll(async () => {
+    ports = { http: await getPort(), proxy: await getPort() };
+  });
+
+  const servers: Record<string, any> = {};
+
+  it("Create the target HTTP server that hangs", async () => {
+    servers.http = http
+      .createServer((_req, _res) => {
+        // Do nothing, let it hang
+      })
+      .listen(ports.http);
+  });
+
+  it("Create the proxy server with fetch and timeout", async () => {
+    servers.proxy = httpProxy
+      .createServer({
+        target: `http://localhost:${ports.http}`,
+        fetch: fetch as any, // Enable fetch path
+        proxyTimeout: 500, // 500ms timeout
+      })
+      .listen(ports.proxy);
+  });
+
+  it("should timeout the request and emit error", async () => {
+    return new Promise<void>((resolve, reject) => {
+      const timeout = setTimeout(() => {
+        reject(new Error("Test timed out"));
+      }, 2000);
+
+      servers.proxy.once('error', (err: Error, _req: any, res: any) => {
+        clearTimeout(timeout);
+        try {
+          expect(err).toBeTruthy();
+          expect(err.message).toBe("Proxy timeout");
+          res.statusCode = 504;
+          res.end("Gateway Timeout");
+          resolve();
+        } catch (e) {
+          reject(e);
+        }
+      });
+
+      fetch(`http://localhost:${ports.proxy}`).catch(() => {
+         // Ignore client side fetch error, we care about server side error emission
+      });
+    });
+  });
+
+  afterAll(async () => {
+    Object.values(servers).map((x: any) => x?.close());
+  });
+});

lib/test/http/xfwd-http2.test.ts

@@ -0,0 +1,61 @@
+
+import * as http from "node:http";
+import * as httpProxy from "../..";
+import getPort from "../get-port";
+import { join } from "node:path";
+import { readFile } from "node:fs/promises";
+import { describe, it, expect, beforeAll, afterAll } from "vitest";
+import { Agent, fetch } from "undici";
+
+const TestAgent = new Agent({ allowH2: true, connect: { rejectUnauthorized: false } });
+
+const fixturesDir = join(__dirname, "..", "fixtures");
+
+describe("X-Forwarded-Host with HTTP/2", () => {
+  let ports: Record<"http" | "proxy", number>;
+  beforeAll(async () => {
+    ports = { http: await getPort(), proxy: await getPort() };
+  });
+
+  const servers: any = {};
+
+  it("Create the target HTTP server", async () => {
+    servers.http = http
+      .createServer((req, res) => {
+        res.writeHead(200, { "Content-Type": "application/json" });
+        res.write(JSON.stringify(req.headers));
+        res.end();
+      })
+      .listen(ports.http);
+  });
+
+  it("Create the HTTPS proxy server with xfwd", async () => {
+    servers.proxy = httpProxy
+      .createServer({
+        target: {
+          host: "localhost",
+          port: ports.http,
+        },
+        ssl: {
+          key: await readFile(join(fixturesDir, "agent2-key.pem"), "utf8"),
+          cert: await readFile(join(fixturesDir, "agent2-cert.pem"), "utf8"),
+        },
+        xfwd: true,
+      })
+      .listen(ports.proxy);
+  });
+
+  it("should pass x-forwarded-host when using HTTP/2", async () => {
+    const res = await fetch(`https://localhost:${ports.proxy}`, { dispatcher: TestAgent });
+    const headers = await res.json() as any;
+
+    // In HTTP/2, :authority is used instead of Host.
+    // The proxy should map :authority to x-forwarded-host if Host is missing.
+    expect(headers["x-forwarded-host"]).toBe(`localhost:${ports.proxy}`);
+  });
+
+  afterAll(async () => {
+    // cleans up
+    Object.values(servers).map((x: any) => x?.close());
+  });
+});

lib/test/lib/http-proxy-passes-web-incoming.test.ts

@@ -119,7 +119,11 @@ describe("#createProxyServer.web() using own http server", () => {
       const source = http.createServer((req, res) => {
         res.end();
         expect(req.method).toEqual("GET");
-        expect(req.headers.host?.split(":")[1]).toEqual(`${ports["8081"]}`);
+        if (process.env.FORCE_FETCH_PATH === "true") {
+          expect(req.headers.host?.split(":")[1]).toEqual(`${ports["8080"]}`);
+        } else {
+          expect(req.headers.host?.split(":")[1]).toEqual(`${ports["8081"]}`);
+        }
       });
 
       proxyServer.listen(ports["8081"]);