mirror of
https://github.com/germondai/trawl.git
synced 2026-08-17 12:11:23 +02:00
feat(proxy): add adaptive direct forwarding
This commit is contained in:
germondai
@@ -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<Uint8Array> =>
|
||||
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("<!doctype html><title>Normal page</title>"), Buffer.from("<p>small response</p>")),
|
||||
{ headers: { "Content-Type": "text/html; charset=utf-8" } },
|
||||
)
|
||||
if (pathname === "/chunked-challenge")
|
||||
return new Response(
|
||||
chunked(Buffer.from('<!doctype html><title>Just a moment...</title><div id="cf-browser-verification"></div>')),
|
||||
{ status: 503, headers: { "Content-Type": "text/html; charset=utf-8" } },
|
||||
)
|
||||
if (pathname === "/gzip-challenge") {
|
||||
const body = gzipSync('<!doctype html><title>Just a moment...</title><div id="cf-browser-verification"></div>')
|
||||
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("<!doctype html><title>Normal page</title><p>small response</p>")
|
||||
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()
|
||||
})
|
||||
})
|
||||
@@ -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)
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -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<string, string>
|
||||
contentType: string
|
||||
contentLength?: number
|
||||
body: Buffer
|
||||
challengeDetected: boolean
|
||||
}
|
||||
|
||||
export interface ForwardResultStream {
|
||||
mode: "stream"
|
||||
status: number
|
||||
headers: Record<string, string>
|
||||
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<string, string>
|
||||
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<ForwardResult> {
|
||||
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<ForwardResult>((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<string, string>
|
||||
body?: Buffer
|
||||
skipChallengeDetection?: boolean
|
||||
timeoutMs?: number
|
||||
servername?: string
|
||||
}
|
||||
|
||||
const requestHead = (
|
||||
method: string,
|
||||
path: string,
|
||||
host: string,
|
||||
headers: Record<string, string>,
|
||||
body?: Buffer,
|
||||
): string => {
|
||||
const forwarded: Record<string, string> = { 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<ForwardResult> {
|
||||
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<ForwardResult>((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<ForwardResult> {
|
||||
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<string, string> = {}
|
||||
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<Buffer> {
|
||||
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<Buffer> {
|
||||
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))
|
||||
}
|
||||
@@ -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" }
|
||||
}
|
||||
Reference in New Issue
Block a user