fix(daemon): close browser HTTP upstreams

Author: lefarcen

apps/daemon/src/browser-network-proxy.ts

@@ -1,5 +1,5 @@
-import http, { type IncomingMessage, type ServerResponse } from 'node:http';
-import net from 'node:net';
+import http, { type ClientRequest, type IncomingMessage, type ServerResponse } from 'node:http';
+import net, { type Socket } from 'node:net';
 import type { Duplex } from 'node:stream';
 
 import {
@@ -57,8 +57,18 @@ export async function createBrowserNetworkProxy(
   policy: BrowserNetworkPolicy = {},
 ): Promise<BrowserNetworkProxy> {
   const tunnels = new Set<Duplex>();
+  const httpRequests = new Set<ClientRequest>();
+  const httpSockets = new Set<Socket>();
+  let closing = false;
   const server = http.createServer((request, response) => {
-    void proxyHttpRequest(request, response, policy);
+    void proxyHttpRequest(
+      request,
+      response,
+      policy,
+      httpRequests,
+      httpSockets,
+      () => closing,
+    );
   });
 
   server.on('connect', (request, client, head) => {
@@ -84,6 +94,11 @@ export async function createBrowserNetworkProxy(
   return {
     port: address.port,
     close: async () => {
+      closing = true;
+      for (const request of httpRequests) request.destroy();
+      for (const socket of httpSockets) socket.destroy();
+      httpRequests.clear();
+      httpSockets.clear();
       for (const socket of tunnels) socket.destroy();
       tunnels.clear();
       if (!server.listening) return;
@@ -96,20 +111,53 @@ async function proxyHttpRequest(
   request: IncomingMessage,
   response: ServerResponse,
   policy: BrowserNetworkPolicy,
+  httpRequests: Set<ClientRequest>,
+  httpSockets: Set<Socket>,
+  isClosing: () => boolean,
 ): Promise<void> {
   try {
     const target = await resolveBrowserNetworkTarget(requestUrl(request).href, policy);
+    if (isClosing() || response.destroyed) {
+      response.destroy();
+      return;
+    }
     const upstream = http.request({
+      agent: false,
       family: target.family,
       headers: upstreamHeaders(request, target.url.host),
       host: target.address,
       method: request.method,
       path: `${target.url.pathname}${target.url.search}`,
       port: target.url.port ? Number(target.url.port) : 80,
     }, (upstreamResponse) => {
+      if (response.destroyed) {
+        upstreamResponse.destroy();
+        return;
+      }
       response.writeHead(upstreamResponse.statusCode ?? 502, upstreamResponse.headers);
       upstreamResponse.pipe(response);
+      upstreamResponse.once('error', (error) => {
+        if (!response.destroyed) response.destroy(error);
+      });
     });
+    httpRequests.add(upstream);
+    const destroyUpstream = () => upstream.destroy();
+    const forgetUpstream = () => {
+      httpRequests.delete(upstream);
+      request.off('aborted', destroyUpstream);
+      request.off('error', destroyUpstream);
+      response.off('close', destroyUpstream);
+      response.off('error', destroyUpstream);
+    };
+    upstream.once('socket', (socket) => {
+      httpSockets.add(socket);
+      socket.once('close', () => httpSockets.delete(socket));
+    });
+    upstream.once('close', forgetUpstream);
+    request.once('aborted', destroyUpstream);
+    request.once('error', destroyUpstream);
+    response.once('close', destroyUpstream);
+    response.once('error', destroyUpstream);
     upstream.once('error', (error) => {
       if (!response.headersSent) response.writeHead(502, { 'content-type': 'text/plain; charset=utf-8' });
       response.end(`browser proxy upstream failed: ${proxyErrorMessage(error)}\n`);

apps/daemon/tests/browser-session-security.test.ts

@@ -115,6 +115,71 @@ describe('Website Clone browser broker security boundary', () => {
     }
   });
 
+  it('tears down an ordinary HTTP upstream when the browser client disconnects', async () => {
+    const targetSockets = new Set<Socket>();
+    const target = createServer((_request, response) => {
+      response.writeHead(200);
+      response.write('partial response');
+    });
+    target.on('connection', (socket) => {
+      targetSockets.add(socket);
+      socket.on('error', () => undefined);
+      socket.once('close', () => targetSockets.delete(socket));
+    });
+    await new Promise<void>((resolve) => target.listen(0, '127.0.0.1', resolve));
+    const targetAddress = target.address();
+    if (!targetAddress || typeof targetAddress === 'string') throw new Error('target did not bind');
+    const proxy = await createBrowserNetworkProxy({ allowPrivateNetwork: true });
+    const client = openHttpStreamThroughProxy(proxy.port, targetAddress.port);
+
+    try {
+      await client.responseStarted;
+      expect(targetSockets.size).toBe(1);
+      client.request.destroy();
+      await waitFor(() => targetSockets.size === 0);
+    } finally {
+      client.request.destroy();
+      await proxy.close();
+      for (const socket of targetSockets) socket.destroy();
+      await new Promise<void>((resolve) => target.close(() => resolve()));
+    }
+  });
+
+  it('does not retain ordinary HTTP upstreams across repeated proxy lifecycles', async () => {
+    const targetSockets = new Set<Socket>();
+    const target = createServer((_request, response) => {
+      response.writeHead(200);
+      response.write('partial response');
+    });
+    target.on('connection', (socket) => {
+      targetSockets.add(socket);
+      socket.on('error', () => undefined);
+      socket.once('close', () => targetSockets.delete(socket));
+    });
+    await new Promise<void>((resolve) => target.listen(0, '127.0.0.1', resolve));
+    const targetAddress = target.address();
+    if (!targetAddress || typeof targetAddress === 'string') throw new Error('target did not bind');
+
+    try {
+      for (let iteration = 0; iteration < 3; iteration += 1) {
+        const proxy = await createBrowserNetworkProxy({ allowPrivateNetwork: true });
+        const client = openHttpStreamThroughProxy(proxy.port, targetAddress.port);
+        try {
+          await client.responseStarted;
+          expect(targetSockets.size).toBe(1);
+          await proxy.close();
+          await waitFor(() => targetSockets.size === 0);
+        } finally {
+          client.request.destroy();
+          await proxy.close();
+        }
+      }
+    } finally {
+      for (const socket of targetSockets) socket.destroy();
+      await new Promise<void>((resolve) => target.close(() => resolve()));
+    }
+  });
+
   it('exposes only the CDP methods required by the staged recon adapter', () => {
     expect(WEB_CLONE_CDP_METHODS).toEqual(new Set([
       'Emulation.setDeviceMetricsOverride',
@@ -179,6 +244,29 @@ async function connectTunnel(proxyPort: number, targetPort: number): Promise<Soc
   });
 }
 
+function openHttpStreamThroughProxy(proxyPort: number, targetPort: number): {
+  request: ReturnType<typeof request>;
+  responseStarted: Promise<void>;
+} {
+  let responseStartedResolve: (() => void) | undefined;
+  let responseStartedReject: ((error: Error) => void) | undefined;
+  const responseStarted = new Promise<void>((resolve, reject) => {
+    responseStartedResolve = resolve;
+    responseStartedReject = reject;
+  });
+  const outbound = request({
+    headers: { host: `127.0.0.1:${targetPort}` },
+    host: '127.0.0.1',
+    path: `http://127.0.0.1:${targetPort}/stream`,
+    port: proxyPort,
+  }, (response) => {
+    response.once('data', () => responseStartedResolve?.());
+  });
+  outbound.on('error', (error) => responseStartedReject?.(error));
+  outbound.end();
+  return { request: outbound, responseStarted };
+}
+
 async function waitFor(condition: () => boolean, timeoutMs = 1_000): Promise<void> {
   const startedAt = Date.now();
   while (!condition()) {