diff --git a/changelog.d/fixes/13152-traffic-inspector-ws-subscriber-leak.md b/changelog.d/fixes/13152-traffic-inspector-ws-subscriber-leak.md new file mode 100644 index 0000000000..4c5249f4fd --- /dev/null +++ b/changelog.d/fixes/13152-traffic-inspector-ws-subscriber-leak.md @@ -0,0 +1 @@ +- **fix(traffic-inspector):** the WebSocket route no longer leaks a traffic-buffer subscriber and a 30s ping timer when the client socket is already closed at handler time — listeners are attached before any resource is acquired, a destroyed socket bails out early, and the ping interval stops on a dead socket where `write()` never throws ([#13155](https://github.com/diegosouzapw/OmniRoute/pull/13155)) diff --git a/src/app/api/tools/traffic-inspector/ws/route.ts b/src/app/api/tools/traffic-inspector/ws/route.ts index b32a5d7cc5..a9545e221f 100644 --- a/src/app/api/tools/traffic-inspector/ws/route.ts +++ b/src/app/api/tools/traffic-inspector/ws/route.ts @@ -96,6 +96,14 @@ export async function GET(request: Request): Promise { } const acceptHeader = acceptKey(clientKey); + + // The client can vanish during the upgrade round trip. `close` has then + // ALREADY fired, so the listeners below would never run and every resource + // acquired past this point would be held with no path to release it. + if (socket.destroyed) { + return new Response(null, { status: 101 }); + } + socket.write( [ "HTTP/1.1 101 Switching Protocols", @@ -106,21 +114,17 @@ export async function GET(request: Request): Promise { ].join("\r\n") ); - const unsubscribe = globalTrafficBuffer.subscribe((ev) => { - sendText(socket, ev); - }); - - const pingTimer = setInterval(() => { - try { - socket.write(encodeWsFrame(0x09)); // ping - } catch { - cleanup(); - } - }, PING_INTERVAL_MS); + let unsubscribe: (() => void) | null = null; + let pingTimer: ReturnType | null = null; + let cleanedUp = false; function cleanup(): void { - clearInterval(pingTimer); - unsubscribe(); + if (cleanedUp) return; + cleanedUp = true; + if (pingTimer) clearInterval(pingTimer); + pingTimer = null; + unsubscribe?.(); + unsubscribe = null; try { socket.destroy(); } catch { @@ -128,14 +132,43 @@ export async function GET(request: Request): Promise { } } - socket.once("close", cleanup); - socket.once("error", cleanup); - - // Never resolve — the socket is the response channel. - await new Promise((resolve) => { + // Attached BEFORE any resource is acquired, so there is no window in which a + // subscriber or timer exists without a live path to cleanup(). + const settled = new Promise((resolve) => { socket.once("close", resolve); socket.once("error", resolve); }); + socket.once("close", cleanup); + socket.once("error", cleanup); + + // Re-check: `close` may have fired while we were writing the handshake, in + // which case the listeners above already ran and cleanup() is a no-op we + // still must not skip. + if (socket.destroyed) { + cleanup(); + return new Response(null, { status: 101 }); + } + + unsubscribe = globalTrafficBuffer.subscribe((ev) => { + sendText(socket, ev); + }); + + pingTimer = setInterval(() => { + // `socket.write()` does NOT throw synchronously on a destroyed socket, so + // the destroyed check — not the catch — is what stops a dead interval. + if (socket.destroyed) { + cleanup(); + return; + } + try { + socket.write(encodeWsFrame(0x09)); // ping + } catch { + cleanup(); + } + }, PING_INTERVAL_MS); + + // Never resolve — the socket is the response channel. + await settled; cleanup(); return new Response(null, { status: 101 }); diff --git a/tests/unit/traffic-inspector-ws-subscriber-leak-13152.test.ts b/tests/unit/traffic-inspector-ws-subscriber-leak-13152.test.ts new file mode 100644 index 0000000000..230a3f28d2 --- /dev/null +++ b/tests/unit/traffic-inspector-ws-subscriber-leak-13152.test.ts @@ -0,0 +1,134 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import net from "node:net"; +import type { AddressInfo } from "node:net"; + +import { GET } from "@/app/api/tools/traffic-inspector/ws/route"; +import { globalTrafficBuffer } from "@/mitm/inspector/buffer"; + +const DEAD_UPGRADES = 6; + +function armedTimers(): number { + return process.getActiveResourcesInfo().filter((r) => r === "Timeout").length; +} + +function upgradeRequest(socket: net.Socket): Request { + const req = new Request("http://127.0.0.1/api/tools/traffic-inspector/ws", { + headers: { + upgrade: "websocket", + "sec-websocket-key": "dGhlIHNhbXBsZSBub25jZQ==", + }, + }); + Object.defineProperty(req, "socket", { value: socket, configurable: true }); + return req; +} + +async function deadSocket(port: number): Promise { + const sock = net.connect(port, "127.0.0.1"); + await new Promise((r) => sock.once("connect", () => r())); + sock.on("error", () => {}); + sock.destroy(); + await new Promise((r) => setTimeout(r, 20)); + return sock; +} + +test("an already-closed socket leaves no subscriber and no ping timer", async () => { + const accepted: net.Socket[] = []; + const server = net.createServer((c) => { + accepted.push(c); + c.on("error", () => {}); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", () => r())); + const { port } = server.address() as AddressInfo; + + try { + const timersBefore = armedTimers(); + const subsBefore = globalTrafficBuffer.subscriberCount(); + + const handlers: Promise[] = []; + for (let i = 0; i < DEAD_UPGRADES; i++) { + // Catch at creation time: the route answers a hijacked upgrade with a 101 + // Response, which undici rejects off a real server. Left unattached, that + // rejection would sit through the next await and trip Node's unhandled + // rejection detection. Either settlement proves the handler released its + // resources instead of hanging, which is what this test measures. + handlers.push(GET(upgradeRequest(await deadSocket(port))).catch(() => undefined)); + } + + // Own the race timer so it can be cleared before measuring; otherwise the + // test's own armed timeout is counted as a leaked one. + let raceTimer: ReturnType | undefined; + const outcome = await Promise.race([ + Promise.all(handlers).then(() => "settled"), + new Promise((r) => { + raceTimer = setTimeout(() => r("hung"), 2000); + }), + ]); + if (raceTimer) clearTimeout(raceTimer); + assert.equal( + outcome, + "settled", + "each handler must return instead of hanging forever on a dead socket" + ); + + const timersAfter = armedTimers(); + assert.ok( + timersAfter <= timersBefore, + `${DEAD_UPGRADES} dead upgrades retained ${timersAfter - timersBefore} ping timer(s)` + ); + + // Measure the subscriber set directly; counting fan-out to our own probe + // says nothing about whether the dead sockets stayed subscribed. + assert.equal( + globalTrafficBuffer.subscriberCount(), + subsBefore, + `${DEAD_UPGRADES} dead upgrades left ${globalTrafficBuffer.subscriberCount() - subsBefore} subscriber(s) behind` + ); + } finally { + // close() only fires once every accepted connection is gone. + for (const c of accepted) c.destroy(); + await new Promise((r) => server.close(() => r())); + } +}); + +test("a live socket keeps its subscription until the socket closes", async () => { + const accepted: net.Socket[] = []; + const server = net.createServer((c) => { + accepted.push(c); + c.on("error", () => {}); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", () => r())); + const { port } = server.address() as AddressInfo; + + const sock = net.connect(port, "127.0.0.1"); + await new Promise((r) => sock.once("connect", () => r())); + sock.on("error", () => {}); + + try { + const subsBefore = globalTrafficBuffer.subscriberCount(); + + const handler = GET(upgradeRequest(sock)).catch(() => undefined); + await new Promise((r) => setTimeout(r, 100)); + + assert.equal( + globalTrafficBuffer.subscriberCount(), + subsBefore + 1, + "a live upgrade must register exactly one traffic subscriber" + ); + + // Closing the socket resolves the handler's `settled` promise, which is the + // only path that releases the subscriber. + sock.destroy(); + await handler; + + assert.equal( + globalTrafficBuffer.subscriberCount(), + subsBefore, + "closing the socket must release the subscriber" + ); + } finally { + // close() only fires once every accepted connection is gone. + for (const c of accepted) c.destroy(); + await new Promise((r) => server.close(() => r())); + } +});