/** * HTTP_PROXY listener for the Traffic Inspector. * * Accepts HTTP_PROXY=http://127.0.0.1:8080 style upstream traffic. Two paths: * * 1. HTTP direct (non-CONNECT): the proxy reads the request body, forwards * it via `fetch()`, captures the response, and records the full exchange. * 2. CONNECT (TLS tunnel): the proxy opens a raw TCP bridge so HTTPS still * works, but only metadata (host:port) is captured — bodies stay opaque. * The buffer entry carries a `note` field explaining why. * * `EADDRINUSE` during `listen()` rejects the returned promise so callers can * surface a clean error to the user. See master-plan §3.12 + plan 12 §4.2.6. */ import http from "node:http"; import net from "node:net"; import { randomUUID } from "node:crypto"; import { sanitizeErrorMessage } from "@omniroute/open-sse/utils/error"; import { sanitizeHeaders } from "../sanitizeHeaders.ts"; import { maskSecret } from "../maskSecrets.ts"; import { globalTrafficBuffer } from "./buffer.ts"; import type { InterceptedRequest } from "./types.ts"; const DEFAULT_PORT = parseEnvNumber(process.env.INSPECTOR_HTTP_PROXY_PORT, 8080); function parseEnvNumber(value: string | undefined, fallback: number): number { if (!value) return fallback; const parsed = Number(value); if (!Number.isFinite(parsed) || parsed <= 0) return fallback; return parsed; } export interface HttpProxyServerHandle { port: number; server: http.Server; stop(): Promise; } /** * Build a sanitized Headers object suitable for an upstream `fetch()`. * Drops hop-by-hop fields via the existing denylist and coerces array values. */ function buildFetchHeaders(raw: http.IncomingHttpHeaders): Record { // sanitizeHeaders applies upstream denylist + masks Authorization for buffer // logging; here we want denylist only (so the upstream still sees the auth // header). Reuse sanitizeHeaders with a separate masking pass for the buffer. const out: Record = {}; for (const [name, value] of Object.entries(raw)) { if (value === undefined || value === null) continue; const lower = name.toLowerCase(); // Skip hop-by-hop / framing — same names sanitizeHeaders also drops. if ( lower === "host" || lower === "connection" || lower === "keep-alive" || lower === "proxy-authenticate" || lower === "proxy-authorization" || lower === "te" || lower === "trailer" || lower === "transfer-encoding" || lower === "upgrade" || lower === "content-length" ) { continue; } out[lower] = Array.isArray(value) ? value.join(", ") : String(value); } return out; } function safeUrl(rawUrl: string | undefined, hostHeader: string | undefined): URL | null { if (!rawUrl) return null; try { if (/^https?:\/\//i.test(rawUrl)) return new URL(rawUrl); if (hostHeader) return new URL(`http://${hostHeader}${rawUrl}`); } catch { return null; } return null; } async function readBody(req: http.IncomingMessage): Promise { const chunks: Buffer[] = []; for await (const chunk of req) { chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); } return Buffer.concat(chunks); } function handleHttp(req: http.IncomingMessage, res: http.ServerResponse): void { const startedAt = performance.now(); const intercepted: InterceptedRequest = { id: randomUUID(), source: "http-proxy", timestamp: new Date().toISOString(), method: req.method ?? "GET", host: req.headers.host ?? "", path: req.url ?? "/", requestHeaders: sanitizeHeaders(req.headers), requestBody: null, requestSize: 0, responseHeaders: {}, responseBody: null, responseSize: 0, status: "in-flight", }; globalTrafficBuffer.push(intercepted); void (async () => { try { const target = safeUrl(req.url, req.headers.host); if (!target) { throw new Error("Invalid request URL"); } intercepted.host = target.host; intercepted.path = target.pathname + target.search; const body = await readBody(req); intercepted.requestSize = body.length; intercepted.requestBody = body.length > 0 ? maskSecret(body.toString("utf8")) : null; const upstreamHeaders = buildFetchHeaders(req.headers); const upstream = await fetch(target.toString(), { method: req.method ?? "GET", headers: upstreamHeaders, body: body.length > 0 ? body : undefined, redirect: "manual", }); const respBuf = Buffer.from(await upstream.arrayBuffer()); const totalLatencyMs = performance.now() - startedAt; intercepted.responseHeaders = sanitizeHeaders( Object.fromEntries(upstream.headers) as Record ); intercepted.responseBody = maskSecret(respBuf.toString("utf8")); intercepted.responseSize = respBuf.length; intercepted.status = upstream.status; intercepted.totalLatencyMs = totalLatencyMs; intercepted.upstreamLatencyMs = totalLatencyMs; intercepted.proxyLatencyMs = 0; const safeRespHeaders: Record = {}; upstream.headers.forEach((value, key) => { if (key.toLowerCase() === "content-length") return; if (key.toLowerCase() === "transfer-encoding") return; safeRespHeaders[key] = value; }); res.writeHead(upstream.status, safeRespHeaders); res.end(respBuf); globalTrafficBuffer.update(intercepted.id, intercepted); } catch (err) { intercepted.status = "error"; intercepted.error = sanitizeErrorMessage(err); intercepted.totalLatencyMs = performance.now() - startedAt; globalTrafficBuffer.update(intercepted.id, intercepted); if (!res.headersSent) { res.writeHead(502, { "content-type": "text/plain" }); res.end("Bad Gateway"); } else { res.end(); } } })(); } function handleConnect( req: http.IncomingMessage, clientSocket: net.Socket, head: Buffer ): void { const target = req.url ?? ""; const [host, rawPort] = target.split(":"); const port = Number(rawPort) || 443; const intercepted: InterceptedRequest = { id: randomUUID(), source: "http-proxy", timestamp: new Date().toISOString(), method: "CONNECT", host, path: `:${port}`, requestHeaders: sanitizeHeaders(req.headers), requestBody: null, requestSize: 0, responseHeaders: {}, responseBody: null, responseSize: 0, status: "in-flight", note: "TLS tunnel — for body capture, redirect host via Custom Hosts mode", }; globalTrafficBuffer.push(intercepted); const targetSocket = net.connect(port, host); const finalize = (status: number | "error", err?: unknown): void => { intercepted.status = status; if (err !== undefined) intercepted.error = sanitizeErrorMessage(err); globalTrafficBuffer.update(intercepted.id, intercepted); }; targetSocket.once("connect", () => { clientSocket.write("HTTP/1.1 200 Connection Established\r\n\r\n"); if (head && head.length > 0) targetSocket.write(head); targetSocket.pipe(clientSocket); clientSocket.pipe(targetSocket); finalize(200); }); const onError = (err: unknown): void => { finalize("error", err); try { clientSocket.end(); } catch { // socket already closed } try { targetSocket.destroy(); } catch { // already destroyed } }; targetSocket.once("error", onError); clientSocket.once("error", onError); } /** * Start the HTTP_PROXY listener. Resolves with a handle once `listening` has * fired; rejects with `Error.code === "EADDRINUSE"` (and similar) when the * bind fails so callers can surface a clean error. */ export function startHttpProxyServer(port: number = DEFAULT_PORT): Promise { return new Promise((resolve, reject) => { const server = http.createServer(); server.on("request", (req, res) => handleHttp(req, res)); server.on("connect", (req, socket, head) => handleConnect(req, socket as net.Socket, head)); server.once("error", (err: NodeJS.ErrnoException) => { // Decorate with a code so callers can pattern-match without parsing strings. reject(Object.assign(err, { code: err.code ?? "ELISTEN" })); }); server.once("listening", () => { const addr = server.address(); const boundPort = typeof addr === "object" && addr ? addr.port : port; resolve({ port: boundPort, server, stop: () => new Promise((res) => { server.close(() => res()); }), }); }); server.listen(port, "127.0.0.1"); }); }