diff --git a/apps/api/src/proxy/__tests__/directForward.test.ts b/apps/api/src/proxy/__tests__/directForward.test.ts new file mode 100644 index 0000000..f1f9484 --- /dev/null +++ b/apps/api/src/proxy/__tests__/directForward.test.ts @@ -0,0 +1,199 @@ +import { afterAll, describe, expect, test } from "bun:test" +import { gzipSync } from "node:zlib" +import { directForwardHttp } from "../directForward" + +const fullBody = Buffer.from("0123456789ABCDEF") + +const chunked = (...chunks: Uint8Array[]): ReadableStream => + new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk) + controller.close() + }, + }) + +const fetchFixture = (req: Request): Response => { + const { pathname } = new URL(req.url) + if (pathname === "/chunked-html") + return new Response( + chunked(Buffer.from("Normal page"), Buffer.from("

small response

")), + { headers: { "Content-Type": "text/html; charset=utf-8" } }, + ) + if (pathname === "/chunked-challenge") + return new Response( + chunked(Buffer.from('Just a moment...
')), + { status: 503, headers: { "Content-Type": "text/html; charset=utf-8" } }, + ) + if (pathname === "/gzip-challenge") { + const body = gzipSync('Just a moment...
') + return new Response(body, { + status: 503, + headers: { "Content-Encoding": "gzip", "Content-Type": "text/html; charset=utf-8" }, + }) + } + if (pathname === "/video") + return new Response(chunked(Buffer.from([0, 1, 2, 3]), Buffer.from([4, 5, 6, 7])), { + headers: { "Content-Type": "video/mp4" }, + }) + if (pathname === "/fixed-video") + return new Response(Buffer.from([0, 1, 2, 3, 4, 5, 6, 7]), { + headers: { "Content-Type": "video/mp4" }, + }) + + const match = /^bytes=(\d+)-(\d+)$/.exec(req.headers.get("range") ?? "") + if (!match) + return new Response(fullBody, { + headers: { "Content-Type": "application/octet-stream" }, + }) + + const start = Number(match[1]) + const end = Number(match[2]) + const body = fullBody.subarray(start, end + 1) + return new Response(body, { + status: 206, + headers: { + "Content-Range": `bytes ${start}-${end}/${fullBody.length}`, + "Content-Type": "application/octet-stream", + }, + }) +} + +const createTestServer = () => { + for (let attempt = 0; attempt < 20; attempt++) { + const port = 30_000 + ((process.pid + attempt * 997) % 20_000) + try { + return Bun.serve({ fetch: fetchFixture, hostname: "127.0.0.1", port }) + } catch (error) { + if (!(error instanceof Error) || !error.message.includes("port")) throw error + } + } + throw new Error("failed to bind direct-forward test server") +} + +const server = createTestServer() +const baseUrl = `http://127.0.0.1:${server.port}` + +afterAll(() => server.stop(true)) + +describe("directForwardHttp — Range / 206 Partial Content", () => { + test("forwards Range request header and gets 206 Partial Content back", async () => { + const result = await directForwardHttp({ + url: `${baseUrl}/file.bin`, + method: "GET", + headers: { Range: "bytes=4-9" }, + }) + expect(result.mode).toBe("buffer") + if (result.mode !== "buffer") return + expect(result.status).toBe(206) + expect(result.body.toString("latin1")).toBe("456789") + expect(result.contentLength).toBe(6) + expect(result.headers["content-range"]).toBe("bytes 4-9/16") + }) + + test("forwards multiple range types (single-byte suffix)", async () => { + const result = await directForwardHttp({ + url: `${baseUrl}/file.bin`, + method: "GET", + headers: { Range: "bytes=15-15" }, + }) + expect(result.mode).toBe("buffer") + if (result.mode !== "buffer") return + expect(result.status).toBe(206) + expect(result.body.toString("latin1")).toBe("F") + expect(result.contentLength).toBe(1) + expect(result.headers["content-range"]).toBe("bytes 15-15/16") + }) + + test("no Range header → 200 + full body (Range pass-through, not injection)", async () => { + const result = await directForwardHttp({ + url: `${baseUrl}/file.bin`, + method: "GET", + headers: {}, + }) + expect(result.mode).toBe("buffer") + if (result.mode !== "buffer") return + expect(result.status).toBe(200) + expect(result.body.length).toBe(fullBody.length) + expect(result.body.toString("latin1")).toBe(fullBody.toString("latin1")) + expect(result.headers["content-range"]).toBeUndefined() + }) + + test("preserves Content-Range through the proxy without re-computing", async () => { + const result = await directForwardHttp({ + url: `${baseUrl}/file.bin`, + method: "GET", + headers: { Range: "bytes=0-3" }, + }) + expect(result.mode).toBe("buffer") + if (result.mode !== "buffer") return + expect(result.headers["content-range"]).toBe("bytes 0-3/16") + expect(result.headers["content-length"]).toBe("4") + expect(result.body.length).toBe(4) + }) +}) + +describe("directForwardHttp — buffered by default", () => { + test("buffers and de-chunks small HTML instead of treating it as a stream", async () => { + const result = await directForwardHttp({ + url: `${baseUrl}/chunked-html`, + method: "GET", + headers: {}, + }) + + expect(result.mode).toBe("buffer") + if (result.mode !== "buffer") return + expect(result.body.toString()).toBe("Normal page

small response

") + expect(result.challengeDetected).toBe(false) + }) + + test("detects a challenge in a chunked HTML response", async () => { + const result = await directForwardHttp({ + url: `${baseUrl}/chunked-challenge`, + method: "GET", + headers: {}, + }) + + expect(result.mode).toBe("buffer") + if (result.mode !== "buffer") return + expect(result.challengeDetected).toBe(true) + expect(result.body.toString()).toContain("Just a moment") + }) + + test("detects a challenge in a compressed HTML response", async () => { + const result = await directForwardHttp({ + url: `${baseUrl}/gzip-challenge`, + method: "GET", + headers: {}, + }) + + expect(result.mode).toBe("buffer") + if (result.mode !== "buffer") return + expect(result.challengeDetected).toBe(true) + expect(result.headers["content-encoding"]).toBe("gzip") + }) + + test("streams explicit video responses", async () => { + const result = await directForwardHttp({ + url: `${baseUrl}/video`, + method: "GET", + headers: {}, + }) + + expect(result.mode).toBe("stream") + if (result.mode !== "stream") return + result.socket.destroy() + }) + + test("streams video even when Content-Length is known", async () => { + const result = await directForwardHttp({ + url: `${baseUrl}/fixed-video`, + method: "GET", + headers: {}, + }) + + expect(result.mode).toBe("stream") + if (result.mode !== "stream") return + expect(result.contentLength).toBe(8) + result.socket.destroy() + }) +}) diff --git a/apps/api/src/proxy/__tests__/streaming.test.ts b/apps/api/src/proxy/__tests__/streaming.test.ts new file mode 100644 index 0000000..ae343b0 --- /dev/null +++ b/apps/api/src/proxy/__tests__/streaming.test.ts @@ -0,0 +1,103 @@ +import { describe, expect, test } from "bun:test" +import { STREAM_THRESHOLD, shouldStream } from "../streaming" + +describe("shouldStream", () => { + describe("size threshold", () => { + test("streams responses >= 8 MiB", () => { + expect(shouldStream("https://example.com/file.bin", STREAM_THRESHOLD, "application/octet-stream")).toMatchObject({ + stream: true, + reason: "size-threshold", + }) + expect(shouldStream("https://example.com/file.bin", STREAM_THRESHOLD + 1, "text/plain")).toMatchObject({ + stream: true, + reason: "size-threshold", + }) + }) + + test("buffers responses below 8 MiB", () => { + const d = shouldStream("https://example.com/data.json", 1024, "application/json") + expect(d).toEqual({ stream: false, reason: "default-buffer" }) + }) + }) + + describe("content-type", () => { + test("streams video/*", () => { + expect(shouldStream("https://x/y", 1024, "video/mp4")).toMatchObject({ + stream: true, + reason: "video-content-type", + }) + expect(shouldStream("https://x/y", 1024, "video/webm; codecs=vp9")).toMatchObject({ + stream: true, + reason: "video-content-type", + }) + }) + + test("streams audio/*", () => { + expect(shouldStream("https://x/y", 1024, "audio/mpeg")).toMatchObject({ + stream: true, + reason: "audio-content-type", + }) + expect(shouldStream("https://x/y", 1024, "audio/ogg; codecs=opus")).toMatchObject({ + stream: true, + reason: "audio-content-type", + }) + }) + + test("unknown length + binary content-type → stream (safe default)", () => { + expect(shouldStream("https://x/y", undefined, "application/octet-stream")).toMatchObject({ + stream: true, + reason: "unknown-length-binary", + }) + expect(shouldStream("https://x/y", undefined, "application/pdf")).toMatchObject({ + stream: true, + reason: "unknown-length-binary", + }) + }) + + test("unknown length + text content-type → buffer (challenge detection possible)", () => { + const d = shouldStream("https://x/y", undefined, "text/html") + expect(d).toEqual({ stream: false, reason: "default-buffer" }) + }) + }) + + describe("URL extension", () => { + test.each([ + "https://cdn.example/movie.mp4", + "https://cdn.example/clip.mkv?token=xyz", + "https://cdn.example/episode.webm", + "https://cdn.example/track.mp3", + "https://cdn.example/manifest.m3u8", + "https://cdn.example/segment.ts", + "https://dl.example/installer.dmg", + "https://dl/example/archive.zip", + "https://dl/example/disk.iso", + "https://dl/example/setup.exe", + "https://dl/example/font.woff2", + "https://dl/example/document.pdf", + ])("streams binary extension %s", (url) => { + expect(shouldStream(url, 1024, "application/octet-stream").stream).toBe(true) + }) + + test("buffers HTML/JSON even with no extension", () => { + expect(shouldStream("https://api.example/data", 1024, "application/json").stream).toBe(false) + expect(shouldStream("https://example.com/page", 1024, "text/html").stream).toBe(false) + }) + }) + + describe("edge cases", () => { + test("missing content-type and length → buffer (text-y assumption)", () => { + expect(shouldStream("https://example.com/").stream).toBe(false) + expect(shouldStream("https://example.com/", undefined, undefined).stream).toBe(false) + }) + + test("empty content-type with explicit length → size decides", () => { + expect(shouldStream("https://example.com/", 100, "").stream).toBe(false) + expect(shouldStream("https://example.com/", STREAM_THRESHOLD + 1, "").stream).toBe(true) + }) + + test("content-type with charset parameter is parsed correctly", () => { + // "text/html; charset=utf-8" → base "text/html" → not a video/audio type + expect(shouldStream("https://example.com/", 1024, "text/html; charset=utf-8").stream).toBe(false) + }) + }) +}) diff --git a/apps/api/src/proxy/directForward.ts b/apps/api/src/proxy/directForward.ts new file mode 100644 index 0000000..4223d9d --- /dev/null +++ b/apps/api/src/proxy/directForward.ts @@ -0,0 +1,491 @@ +// Tier 0 — direct TCP/TLS forward to upstream. +// +// Why: the MITM proxy at :8192 used to spin up a browser for every single +// request, including Netflix/YouTube/banks that don't need CF bypass. Tier 0 +// makes those requests near-direct: open a socket to upstream, write the +// client's request, read the response. If the response looks like a challenge, +// escalate to the existing browser-tier pipeline (`scrape()` from @trawl/tiers). +// +// Two entry points: +// - directForwardHttp() → plain HTTP (no client TLS) +// - directForwardHttps() → upstream is HTTPS (open new TLS socket to upstream) +// +// Both return a `ForwardResult` telling the caller whether to buffer (small +// response, do challenge detection) or stream (large/binary response, pipe +// through to the client untouched). + +import net from "node:net" +import tls from "node:tls" +import { brotliDecompressSync, gunzipSync, inflateSync } from "node:zlib" +import { detectChallengeType, isChallengeWall } from "@trawl/tiers" +import { shouldStream } from "./streaming" + +export interface ForwardResultBuffered { + mode: "buffer" + status: number + headers: Record + contentType: string + contentLength?: number + body: Buffer + challengeDetected: boolean +} + +export interface ForwardResultStream { + mode: "stream" + status: number + headers: Record + contentType: string + contentLength?: number + // The live upstream socket. The caller must pipe this to the client socket + // and close it when the upstream ends. + socket: net.Socket + // Body bytes the caller should write to the client BEFORE piping the socket — + // these arrived in the same TCP segment as the response headers, so the upstream + // socket hasn't seen them yet. Writing them first preserves byte ordering. + prefix?: Buffer +} + +export interface ForwardResultError { + mode: "error" + error: Error +} + +export type ForwardResult = ForwardResultBuffered | ForwardResultStream | ForwardResultError + +const MAX_HEADER_BYTES = 64 * 1024 + +export interface DirectForwardHttpOpts { + url: string + method: string + headers: Record + body?: Buffer + // If true, do NOT challenge-detect the response — just buffer and return. + // Used by the caller when challengeCache already says this hostname is "cf". + skipChallengeDetection?: boolean + // Socket timeout for the upstream connection. Default 30s. + timeoutMs?: number +} + +export async function directForwardHttp(opts: DirectForwardHttpOpts): Promise { + const parsed = new URL(opts.url) + if (parsed.protocol !== "http:") { + return { mode: "error", error: new Error(`directForwardHttp: expected http://, got ${parsed.protocol}`) } + } + const port = parsed.port ? Number(parsed.port) : 80 + const host = parsed.hostname + + // Construct an unconnected socket; the callback below performs the single + // connect attempt. Calling connect() on a socket returned by + // createConnection() races two connection attempts under Bun. + const socket = new net.Socket() + // Bun panics on socket.write() if no handlers are attached ("No handlers set + // on Socket"). Attach an error handler immediately so writes are safe; the + // real error handling is set up after socket.connect. + socket.on("error", () => {}) + return new Promise((resolve) => { + let settled = false + const fail = (err: Error) => { + if (settled) return + settled = true + socket.destroy() + resolve({ mode: "error", error: err }) + } + + const timeout = opts.timeoutMs ?? 30_000 + socket.setTimeout(timeout) + socket.once("timeout", () => fail(new Error(`upstream timeout after ${timeout}ms`))) + socket.once("error", fail) + + socket.connect(port, host, () => { + const authority = parsed.port ? `${host}:${parsed.port}` : host + const path = `${parsed.pathname}${parsed.search}` + writeRequest(socket, requestHead(opts.method, path, authority, opts.headers, opts.body), opts.body) + + readHttpResponse(socket, opts.url, opts.skipChallengeDetection ?? false) + .then((result) => { + if (settled) return + settled = true + resolve(result) + }) + .catch(fail) + }) + }) +} + +export interface DirectForwardHttpsOpts { + host: string + port: number + method: string + path: string + headers: Record + body?: Buffer + skipChallengeDetection?: boolean + timeoutMs?: number + servername?: string +} + +const requestHead = ( + method: string, + path: string, + host: string, + headers: Record, + body?: Buffer, +): string => { + const forwarded: Record = { Host: host, ...headers } + delete forwarded.host + if (body?.length && forwarded["content-length"] === undefined) forwarded["Content-Length"] = String(body.length) + + return `${method} ${path} HTTP/1.1\r\n${Object.entries(forwarded) + .map(([name, value]) => `${name}: ${value}`) + .join("\r\n")}\r\n\r\n` +} + +const writeRequest = (socket: net.Socket, head: string, body?: Buffer): void => { + socket.write(head) + if (body?.length) socket.write(body) +} + +export async function directForwardHttps(opts: DirectForwardHttpsOpts): Promise { + const socket = tls.connect({ + host: opts.host, + port: opts.port, + servername: opts.servername ?? opts.host, + minVersion: "TLSv1.2", + }) + // Same Bun-safety as directForwardHttp: attach a no-op error handler so + // subsequent socket.write() calls don't panic before readUpTo attaches the + // real handlers. + socket.on("error", () => {}) + return new Promise((resolve) => { + let settled = false + const fail = (err: Error) => { + if (settled) return + settled = true + socket.destroy() + resolve({ mode: "error", error: err }) + } + + const timeout = opts.timeoutMs ?? 30_000 + socket.setTimeout(timeout) + socket.once("timeout", () => fail(new Error(`upstream TLS timeout after ${timeout}ms`))) + socket.once("error", fail) + + socket.once("secureConnect", () => { + const host = opts.port === 443 ? opts.host : `${opts.host}:${opts.port}` + writeRequest(socket, requestHead(opts.method, opts.path, host, opts.headers, opts.body), opts.body) + + readHttpResponse(socket, `https://${opts.host}${opts.path}`, opts.skipChallengeDetection ?? false) + .then((result) => { + if (settled) return + settled = true + resolve(result) + }) + .catch(fail) + }) + }) +} + +// Read HTTP/1.1 response off the socket. Splits into header + body. Returns +// either a buffered result (with challenge detection done) or a stream result +// (caller takes over the socket for piping). +async function readHttpResponse( + socket: net.Socket, + url: string, + skipChallengeDetection: boolean, +): Promise { + const headerBuf = await readUpTo(socket, MAX_HEADER_BYTES, "\r\n\r\n") + if (!headerBuf.found) { + socket.destroy() + return { mode: "error", error: new Error("upstream response headers exceeded 64 KiB") } + } + + const headerText = headerBuf.text + const lines = headerText.split("\r\n") + const statusLine = lines[0] ?? "" + const statusMatch = statusLine.match(/^HTTP\/1\.[01] (\d{3})(?: (.*))?$/i) + if (!statusMatch) { + socket.destroy() + return { mode: "error", error: new Error(`upstream returned non-HTTP status line: ${statusLine}`) } + } + const status = Number(statusMatch[1]) + + const headers: Record = {} + for (const line of lines.slice(1)) { + const idx = line.indexOf(":") + if (idx <= 0) continue + const name = line.slice(0, idx).trim().toLowerCase() + const value = line.slice(idx + 1).trim() + if (!name) continue + // Keep first occurrence for multi-value headers; Set-Cookie is the + // common case where only the first makes it through. Caller can read + // raw lines via the set-cookie header if needed. + if (headers[name] === undefined) headers[name] = value + } + + const contentType = headers["content-type"] ?? "application/octet-stream" + const rawLen = headers["content-length"] + const contentLength = rawLen ? Number(rawLen) : undefined + const streamDecision = shouldStream(url, contentLength, contentType) + + const isChunked = (headers["transfer-encoding"] ?? "").toLowerCase().includes("chunked") + if (streamDecision.stream) { + const prefix = headerBuf.leftover?.length ? headerBuf.leftover : undefined + return { + mode: "stream", + status, + headers, + contentType, + contentLength, + socket, + prefix, + } + } + + if (isChunked || contentLength === undefined) { + // Ordinary HTML/JSON/text responses are buffered to completion. This is + // required both for valid chunk decoding and challenge detection. + const initial = headerBuf.leftover ?? Buffer.alloc(0) + const remaining = isChunked ? await readCompleteChunkedBody(socket, initial) : await drainSocket(socket) + socket.destroy() + const wireBody = isChunked ? remaining : Buffer.concat([initial, remaining]) + let body: Buffer + try { + body = isChunked ? decodeChunkedBody(wireBody) : wireBody + } catch (err) { + return { mode: "error", error: err instanceof Error ? err : new Error(String(err)) } + } + const previewText = decodeForInspection(body, headers["content-encoding"]) + const challengeType = detectChallengeType(previewText, headers) + const challengeDetected = !skipChallengeDetection && isChallengeWall(status, body.length, challengeType) + return { + mode: "buffer", + status, + headers, + contentType, + contentLength: body.length, + body, + challengeDetected, + } + } + + // Buffer the body up to Content-Length. Use leftover bytes from the header + // read first (they were received in the same TCP segment as the headers). + const body = Buffer.alloc(contentLength) + let offset = 0 + if (headerBuf.leftover && headerBuf.leftover.length > 0) { + const toCopy = Math.min(headerBuf.leftover.length, contentLength) + headerBuf.leftover.copy(body, 0, 0, toCopy) + offset = toCopy + if (headerBuf.leftover.length > toCopy) { + // Upstream sent more body bytes than Content-Length declared. + socket.destroy() + return { + mode: "buffer", + status, + headers, + contentType, + contentLength, + body: body.subarray(0, offset), + challengeDetected: false, + } + } + } + while (offset < contentLength) { + const { chunk } = await readChunk(socket) + if (!chunk) { + socket.destroy() + return { + mode: "error", + error: new Error(`upstream closed before Content-Length was satisfied (${offset}/${contentLength})`), + } + } + const toCopy = Math.min(chunk.length, contentLength - offset) + chunk.copy(body, offset, 0, toCopy) + offset += toCopy + if (chunk.length > toCopy) { + // Upstream sent more than Content-Length declared. Stop reading, close. + socket.destroy() + break + } + } + socket.destroy() + + // Challenge detection on the buffered body. Bounded preview keeps this cheap. + const previewText = decodeForInspection(body.subarray(0, offset), headers["content-encoding"]) + const challengeType = detectChallengeType(previewText, headers) + const challengeDetected = !skipChallengeDetection && isChallengeWall(status, body.length, challengeType) + + return { + mode: "buffer", + status, + headers, + contentType, + contentLength, + body: body.subarray(0, offset), + challengeDetected, + } +} + +// Read bytes from the socket until we find `delimiter` or hit `maxBytes`. +async function readUpTo( + socket: net.Socket, + maxBytes: number, + delimiter: string, +): Promise<{ found: boolean; text: string; leftover?: Buffer }> { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = [] + let total = 0 + + const onData = (chunk: Buffer) => { + chunks.push(chunk) + total += chunk.length + const buf = Buffer.concat(chunks, total) + const idx = buf.indexOf(delimiter) + if (idx >= 0) { + // Stop flowing before handing control back. Otherwise a small response + // can emit its body/end between this callback and the caller attaching + // the body listeners. + socket.pause() + socket.off("data", onData) + socket.off("error", onError) + const endIdx = idx + delimiter.length + const leftover = endIdx < buf.length ? buf.subarray(endIdx) : undefined + resolve({ found: true, text: buf.subarray(0, endIdx).toString("latin1"), leftover }) + return + } + if (total > maxBytes) { + socket.off("data", onData) + socket.off("error", onError) + resolve({ found: false, text: buf.toString("latin1") }) + } + } + const onError = (err: Error) => { + socket.off("data", onData) + socket.off("error", onError) + reject(err) + } + + socket.on("data", onData) + socket.once("error", onError) + }) +} + +async function readChunk(socket: net.Socket) { + return new Promise<{ chunk?: Buffer }>((resolve) => { + const onData = (chunk: Buffer) => { + socket.off("data", onData) + socket.off("end", onEnd) + socket.off("error", onError) + resolve({ chunk }) + } + const onEnd = () => { + socket.off("data", onData) + socket.off("end", onEnd) + socket.off("error", onError) + resolve({}) + } + const onError = () => { + socket.off("data", onData) + socket.off("end", onEnd) + socket.off("error", onError) + resolve({}) + } + socket.once("data", onData) + socket.once("end", onEnd) + socket.once("error", onError) + socket.resume() + }) +} + +// Drain everything remaining on the socket into a single Buffer. Used after +// challenge detection to capture the full challenge HTML before escalation. +async function drainSocket(socket: net.Socket): Promise { + const chunks: Buffer[] = [] + return new Promise((resolve) => { + const onData = (chunk: Buffer) => chunks.push(chunk) + const finish = () => { + socket.off("data", onData) + socket.off("end", onFinish) + socket.off("error", onFinish) + resolve(Buffer.concat(chunks)) + } + const onFinish = () => finish() + socket.on("data", onData) + socket.once("end", onFinish) + socket.once("error", onFinish) + socket.resume() + }) +} + +async function readCompleteChunkedBody(socket: net.Socket, initial: Buffer): Promise { + try { + decodeChunkedBody(initial) + return initial + } catch { + // The first packet commonly contains only part of the chunked body. + } + + return new Promise((resolve) => { + const chunks = initial.length ? [initial] : [] + let total = initial.length + const finish = () => { + socket.off("data", onData) + socket.off("end", onEnd) + socket.off("error", onEnd) + resolve(Buffer.concat(chunks, total)) + } + const onData = (chunk: Buffer) => { + chunks.push(chunk) + total += chunk.length + try { + decodeChunkedBody(Buffer.concat(chunks, total)) + finish() + } catch { + // Keep reading until the terminating zero-size chunk arrives. + } + } + const onEnd = () => finish() + socket.on("data", onData) + socket.once("end", onEnd) + socket.once("error", onEnd) + socket.resume() + }) +} + +function decodeChunkedBody(wire: Buffer): Buffer { + const chunks: Buffer[] = [] + let offset = 0 + + while (offset < wire.length) { + const lineEnd = wire.indexOf("\r\n", offset) + if (lineEnd < 0) throw new Error("invalid chunked response: missing chunk-size terminator") + const sizeText = wire.subarray(offset, lineEnd).toString("ascii").split(";", 1)[0]?.trim() ?? "" + const size = Number.parseInt(sizeText, 16) + if (!Number.isFinite(size) || size < 0) throw new Error(`invalid chunked response size: ${sizeText}`) + offset = lineEnd + 2 + + if (size === 0) return Buffer.concat(chunks) + if (offset + size + 2 > wire.length) throw new Error("invalid chunked response: truncated chunk") + chunks.push(wire.subarray(offset, offset + size)) + offset += size + if (wire.subarray(offset, offset + 2).toString("ascii") !== "\r\n") { + throw new Error("invalid chunked response: missing chunk terminator") + } + offset += 2 + } + + throw new Error("invalid chunked response: missing final chunk") +} + +function decodeForInspection(body: Buffer, contentEncoding?: string): string { + let decoded = body + try { + const encoding = contentEncoding?.split(",", 1)[0]?.trim().toLowerCase() + if (encoding === "gzip" || encoding === "x-gzip") decoded = gunzipSync(body) + else if (encoding === "deflate") decoded = inflateSync(body) + else if (encoding === "br") decoded = brotliDecompressSync(body) + } catch { + // If an upstream mislabeled or truncated the encoding, inspect the raw bytes. + } + return decoded.toString("utf8", 0, Math.min(decoded.length, 4096)) +} diff --git a/apps/api/src/proxy/streaming.ts b/apps/api/src/proxy/streaming.ts new file mode 100644 index 0000000..ec45b10 --- /dev/null +++ b/apps/api/src/proxy/streaming.ts @@ -0,0 +1,70 @@ +// Adaptive buffer-vs-stream decision for the MITM proxy. +// +// Goal: keep memory footprint low for big binaries (videos, archives, ISO images) +// while still letting small JSON/HTML/text responses round-trip through the +// challenge detector in a single Buffer — which is required because Tier 0 +// (direct forward) only knows whether the response is CF-challenged after it +// has the body to inspect. +// +// Decision precedence: +// 1. Unknown Content-Length + binary-looking Content-Type → stream (safer). +// 2. Content-Length >= STREAM_THRESHOLD → stream. +// 3. Content-Type starts with video/ or audio/ → stream. +// 4. URL extension matches a known binary container → stream. +// 5. Otherwise → buffer. + +export const STREAM_THRESHOLD = 8 * 1024 * 1024 // 8 MiB + +// URL extensions that are almost always large binaries worth streaming. +// Audio/video and HLS manifests are streamed even below the size threshold +// because chunked transport is the whole point of those protocols. +const STREAM_EXTENSIONS = + /\.(mp4|mkv|webm|mov|avi|flv|wmv|m4v|mp3|flac|wav|ogg|opus|aac|m4a|aif|ts|m3u8|mpd|hls|exe|zip|dmg|iso|img|tar|gz|tgz|bz2|xz|7z|rar|pkg|deb|rpm|apk|ipa|msi|pdf|woff2?|ttf|otf|eot)(?:\?|$|#)/i + +export interface StreamDecision { + stream: boolean + reason: + | "size-threshold" + | "video-content-type" + | "audio-content-type" + | "binary-extension" + | "unknown-length-binary" + | "default-buffer" +} + +export function shouldStream(url: string, contentLength?: number, contentType?: string): StreamDecision { + const ct = (contentType ?? "").toLowerCase() + const ctBase = ct.split(";")[0]?.trim() ?? "" + + // 1. Unknown length with binary-looking type → stream to avoid buffering + // multi-gigabyte downloads into RAM. + if ((contentLength === undefined || Number.isNaN(contentLength)) && ctBase) { + if ( + ctBase.startsWith("video/") || + ctBase.startsWith("audio/") || + ctBase === "application/octet-stream" || + ctBase === "application/pdf" || + ctBase.startsWith("application/zip") || + ctBase === "application/x-7z-compressed" || + ctBase === "application/x-rar-compressed" || + ctBase === "application/x-tar" || + ctBase === "application/x-bittorrent" + ) { + return { stream: true, reason: "unknown-length-binary" } + } + } + + // 2. Size threshold. + if (typeof contentLength === "number" && contentLength >= STREAM_THRESHOLD) { + return { stream: true, reason: "size-threshold" } + } + + // 3/4. Video/audio content-type. + if (ctBase.startsWith("video/")) return { stream: true, reason: "video-content-type" } + if (ctBase.startsWith("audio/")) return { stream: true, reason: "audio-content-type" } + + // 5. Known binary extension in URL. + if (STREAM_EXTENSIONS.test(url)) return { stream: true, reason: "binary-extension" } + + return { stream: false, reason: "default-buffer" } +}