diff --git a/README.md b/README.md index cc760232c..abf1ad339 100644 --- a/README.md +++ b/README.md @@ -474,10 +474,10 @@ What the check leaves alone: allowlist entries that **are** IP literals (allow-l - **mTLS upstreams** - only the in-sandbox client holds the client certificate, so the proxy cannot re-originate the connection on its behalf. - **Certificate-pinning clients** - clients that verify the upstream's identity themselves (custom CAs, SAN pinning) and reject the MITM certificate. - `network.tlsTerminate.extraCaCertPaths` - Paths to PEM CA certificate files appended to that trust bundle, after the MITM CA and the host's regular roots. Excluded (non-terminated) hosts are verified by the client inside the sandbox, and the trust env vars SRT sets (`SSL_CERT_FILE`, `GIT_SSL_CAINFO`, ...) _replace_ each tool's own trust configuration, so a site-local root (e.g. an internal mTLS CA) must be in the bundle or those hosts can never be verified. Only the `CERTIFICATE` blocks of each file are copied into the bundle (anything else, e.g. a private key in a combined PEM, is never exposed to the sandbox); files that are missing, unreadable, or contain no PEM `CERTIFICATE` block are skipped, so it is safe to list paths that exist on only some hosts. -- `network.tlsTerminate.maxTunnels` - At most this many CONNECT tunnels are TLS-terminated at once (integer, 1 to 65536, default 256). Past it a CONNECT is answered `503` with `X-Proxy-Error: too-many-tunnels`. A tunnel holds its slot from its CONNECT until it closes, so idle keep-alive tunnels count. The slot is taken before the client's first bytes are seen, so at the cap a non-TLS CONNECT (such as SSH) to a host that would be terminated is refused too; one that turns out not to carry TLS frees its slot. Hosts in `excludeDomains` do not count. +- `network.tlsTerminate.maxTunnels` - At most this many CONNECT tunnels are TLS-terminated at once (integer, 1 to 65536, default 256). Past it a CONNECT is answered `503` with `X-Proxy-Error: too-many-tunnels`, and the refusal is recorded as a violation, once per command until half the slots are free again. A tunnel holds its slot from its CONNECT until it closes, so idle keep-alive tunnels count until they are [closed](#tls-termination). The slot is taken before the client's first bytes are seen, so at the cap a non-TLS CONNECT (such as SSH) to a host that would be terminated is refused too; one that turns out not to carry TLS frees its slot. Hosts in `excludeDomains` do not count. - `network.tlsTerminate.handshakeTimeoutMs` - A tunnel to be terminated must finish its TLS handshake within this many milliseconds of its CONNECT (integer, 100 to 600000, default 10000), or it is closed and its slot freed. A CONNECT whose client sends nothing is closed at the deadline too. A tunnel past its handshake is never timed out by it. -These two are read when the proxy starts; `updateConfig` does not change them. +These two are read at each CONNECT, so `updateConfig` changes them for the tunnels that follow. ```json { @@ -858,7 +858,7 @@ tlsTerminate needs Node, or Bun 1.4 or later (this is Bun 1.3.13) A program built with `bun build --compile` runs on the Bun that built it, so build it with Bun 1.4 or later. Without `tlsTerminate`, nothing changes on any runtime. Should a tunnel still reach the proxy on such a runtime, it is closed rather than left hanging. -**Limits.** At most `maxTunnels` tunnels (default 256) are terminated at once, and each must finish its handshake within `handshakeTimeoutMs` (default 10 s) of its CONNECT; see [Network Configuration](#network-configuration). Leaf certificates share one key per CA, and the proxy keeps those of the 256 most recently used host names. +**Limits.** At most `maxTunnels` tunnels (default 256) are terminated at once, and each must finish its handshake within `handshakeTimeoutMs` (default 10 s) of its CONNECT; see [Network Configuration](#network-configuration). Past its handshake, an idle tunnel (no request on it, nothing queued for its client) is closed and its slot freed: 10 s after its ClientHello if no whole request head has come, and 6 s after a response if none has followed, which is 1 s later than the `Keep-Alive: timeout=5` that responses carry; a client that does not answer the close is cut off as long again later. A tunnel whose request is not all in or is held back for an upstream that takes no more of it, or whose client does not take what is queued for it, is closed 300 to 360 s after anything last moved. A client that goes on reading, slowly or in bursts, loses nothing; that it reads shows as the kernel's send buffer frees up, up to a megabyte or two at a time, so one that takes less than about 5 KB/s of a response larger than the kernel's buffers hold counts as one that has stopped. A whole request may wait for its response without limit. Leaf certificates share one key per CA, and the proxy keeps those of the 256 most recently used host names. **Memory held for slow peers.** A response to a client that is not reading waits once about 1 MiB is queued for that client, or once all tunnels together have 256 MiB queued, on Node and Bun alike. A request body to an upstream that is not reading is held back by the runtime under Node. Under Bun, the proxy does it itself (it pauses a tunnel once 4 MiB of its request body is buffered, or once the shared 256 MiB is used), but that only works on a connection the proxy is handed by its host. The proxy `SandboxManager` runs accepts its own connections, and Bun keeps reading such a tunnel while it is paused. So under Bun an upload to an upstream that has stopped reading is buffered in the SRT process without bound. This is a limitation of the runtime. Use Node where a sandboxed process may upload more than the host can hold. @@ -866,6 +866,7 @@ A program built with `bun build --compile` runs on the Bun that built it, so bui - The tunnel cap: past `maxTunnels`, a CONNECT is answered `503` (`X-Proxy-Error: too-many-tunnels`), whether or not it would have carried TLS. - The handshake deadline: a tunnel that has not finished its TLS handshake within `handshakeTimeoutMs` of its CONNECT is closed. +- Idle tunnels are closed 6 s after a response, as before, and 10 s after a ClientHello that no request follows (before: about 12 s under Bun, 90 s under Node). An upload is cut once it has not moved for 300 to 360 s, no longer 300 s after it began; so is a response of which the client takes nothing (before: never). - The server name passed to `filterRequest` (`sni`) is lower-cased. - With `requireHostMatch`, the leaf certificate is always the CONNECT target's, whatever server name the client asks for (a name that does not match gets `421`, as before). - A ClientHello whose server name cannot be read as a plain DNS name (letters, digits and inner hyphens in dot-separated labels, with no trailing dot; for example one with an underscore) gets the CONNECT target's leaf certificate instead of one minted for that name. With `requireHostMatch`, its requests are answered `421`. diff --git a/docs/srt-proxy.md b/docs/srt-proxy.md index 828b9a943..51924b339 100644 --- a/docs/srt-proxy.md +++ b/docs/srt-proxy.md @@ -284,17 +284,19 @@ addresses and the private ranges (`10/8`, `172.16/12`, `192.168/16`, ### Limits and backpressure The memory and CPU a client can make the proxy spend are bounded by the -limits below. Slots are not: there is no request-head deadline and no idle -deadline, so once its TLS handshake is done a client can hold a tunnel open -indefinitely by sending nothing or half a request head, and it can do the -same on a plain connection; plain connections are not capped in number. The -proxy serves a single sandboxed client, so a client that exhausts its slots -denies service only to itself. +limits below, and so is the time a tunnel holds its slot with no request +moving on it. A plain connection has no request-head deadline and no idle +deadline, so a client can hold one open indefinitely by sending nothing or +half a request head; plain connections are not capped in number. The proxy +serves a single sandboxed client, so a client that exhausts its slots denies +service only to itself. | Resource | Limit | | ------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------- | | TLS-terminated tunnels at once | 256; past it, 503 before the 200 | | Time to finish a tunnel's TLS handshake | 10 s after the CONNECT; then the tunnel is closed and its slot freed | +| A tunnel with no request on it and nothing queued for its client | closed 10 s after its ClientHello, or 6 s after its last response (`Keep-Alive: timeout=5`, and 1 s) | +| A request not all in or held back, a response its client leaves | the tunnel is closed 300 to 360 s after anything last moved. A whole request may wait for its response | | Request head | 2 MiB | | Request body the decider sees (`need_body`) | `max`, at most 32 MiB | | Request bodies held for the decider at once | 64 MiB, counting each at its Content-Length or else at `max`; past it, 503 `decider_unavailable` | @@ -303,8 +305,8 @@ denies service only to itself. | Per connection | 1 MiB queued for the client; under Bun, 4 MiB of request body (Node's own backpressure bounds it otherwise) | | Leaf certificates | the 256 most recently used names; all share one key, so a new name costs a signature, not a key generation | -The tunnel cap and the handshake deadline are fixed in srt-proxy. SRT's full -sandbox mode sets them with `network.tlsTerminate.maxTunnels` and +These are fixed in srt-proxy. SRT's full sandbox mode sets the tunnel cap and +the handshake deadline with `network.tlsTerminate.maxTunnels` and `handshakeTimeoutMs` (see the [README](../README.md#tls-termination)). A slow upstream holds back the client's upload, and a slow client holds back diff --git a/eslint.config.js b/eslint.config.js index b42d1f472..821860cec 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -21,6 +21,7 @@ export default [ projectService: { allowDefaultProject: [ 'eslint.config.js', + 'test/fixtures/tunnel-ends.mjs', 'test/utils/which-node-test.mjs', 'vendor/build-common.ts', 'vendor/seccomp/build.ts', diff --git a/src/sandbox/emitted-connection.ts b/src/sandbox/emitted-connection.ts index 6d71eb115..13a2eedc9 100644 --- a/src/sandbox/emitted-connection.ts +++ b/src/sandbox/emitted-connection.ts @@ -197,12 +197,15 @@ export function gateSocketOnBufferedBody( } const release = (): void => { socket.off('data', check) + socket.off('close', release) if (budget) adjustByteBudget(budget, -counted) counted = 0 } socket.on('data', check) req.once('end', release) req.once('close', release) + // Not every runtime closes the request of a socket that closes. + socket.once('close', release) } const endsGracefully = new WeakSet() diff --git a/src/sandbox/http-proxy.ts b/src/sandbox/http-proxy.ts index 740881e38..ae4246037 100644 --- a/src/sandbox/http-proxy.ts +++ b/src/sandbox/http-proxy.ts @@ -40,6 +40,7 @@ const ALLOWLIST_DENY = [ import { peekForClientHello, terminateAndForward, + type TunnelTimeouts, } from './tls-terminate-proxy.js' import { prepareBodySubstitution, @@ -341,7 +342,9 @@ export interface HttpProxyServerOptions { * terminated is accepted, before its first bytes are seen, and held until * the tunnel closes (an idle keep-alive tunnel holds one too), or until * the tunnel turns out not to carry TLS. So at the cap, a CONNECT that - * would have been tunnelled opaquely as non-TLS is refused too. + * would have been tunnelled opaquely as non-TLS is refused too. The + * refusal is logged and passed to onFilterRequestDenied with method CONNECT, + * once per command until half the slots are free again. * SandboxManager sets it from network.tlsTerminate.maxTunnels. */ maxTerminatedTunnels?: number @@ -349,11 +352,23 @@ export interface HttpProxyServerOptions { /** * A tunnel to be TLS-terminated must finish its TLS handshake this long * after its CONNECT (default 10 s), or it is closed and its slot freed. - * A tunnel that finished its handshake is not timed out afterwards. + * Past its handshake, tunnelTimeouts apply. * SandboxManager sets it from network.tlsTerminate.handshakeTimeoutMs. */ tlsHandshakeTimeoutMs?: number + /** + * When a TLS-terminated tunnel is closed, and its slot freed, past its + * handshake. Idle, with nothing queued for its client: with no whole + * request head yet (default 10 s), or none since its last response (6 s: + * the 5 s that `Keep-Alive: timeout` tells the client, and a margin of + * 1 s). Under way, with a request that is not all in or is held back for + * its upstream, or bytes queued for its client: once nothing has moved for + * 300 s. A whole request that waits for its response is not timed out. + * Primarily a test seam. + */ + tunnelTimeouts?: Partial + /** * Bytes all of this proxy's connections together may hold for slow * parties: request bodies waiting for an upstream, and responses waiting @@ -395,6 +410,8 @@ export function createHttpProxyServer(options: HttpProxyServerOptions): Server { : {}, ) let terminatedTunnels = 0 + /** The commands told of the tunnel cap in this burst of refusals. */ + const toldOfCap = new Set() const byteBudget = options.byteBudget ?? createByteBudget(options.maxBufferedBytes ?? 256 << 20) @@ -709,7 +726,29 @@ export function createHttpProxyServer(options: HttpProxyServerOptions): Server { // only terminate if it is one. Non-TLS falls through to the opaque // tunnel below — i.e. base-sandbox behaviour, hostname-allowlisted // but not content-inspected (same as the SOCKS path). - if (terminatedTunnels >= (options.maxTerminatedTunnels ?? 256)) { + const maxTunnels = options.maxTerminatedTunnels ?? 256 + if (terminatedTunnels >= maxTunnels) { + // Of a refused CONNECT clients show the status at most, so that + // the limit is the sandbox's own is said where refusals are. + if (!toldOfCap.has(auth.encodedCommand)) { + // The client picks the name, so the names kept are bounded. + if (toldOfCap.size >= 100) toldOfCap.clear() + toldOfCap.add(auth.encodedCommand) + const reason = + `the sandbox's limit of ${maxTunnels} TLS-terminated tunnels at ` + + `once is reached; the limit is network.tlsTerminate.maxTunnels ` + + `in the sandbox's configuration` + logForDebugging( + `[proxy] CONNECT ${hostname}:${port} refused: ${reason}`, + { level: 'warn' }, + ) + options.onFilterRequestDenied?.({ + method: 'CONNECT', + url: formatAuthority(hostname, port, 0), + reason, + encodedCommand: auth.encodedCommand, + }) + } const mark = markFor('too_many_tunnels', 'too-many-tunnels')! const [name, value] = typeof mark === 'string' @@ -728,6 +767,7 @@ export function createHttpProxyServer(options: HttpProxyServerOptions): Server { if (!slotHeld) return slotHeld = false terminatedTunnels-- + if (terminatedTunnels * 2 <= maxTunnels) toldOfCap.clear() } let handshakeDone = (): boolean => false const deadline = setTimeout(() => { @@ -776,6 +816,7 @@ export function createHttpProxyServer(options: HttpProxyServerOptions): Server { denyHeader: options.denyHeader, requestTargetAsSpelled: options.requestTargetAsSpelled, byteBudget, + timeouts: options.tunnelTimeouts, onEstablished: () => clearTimeout(deadline), onHandshakeProbe: isDone => { handshakeDone = isDone diff --git a/src/sandbox/mux-proxy.ts b/src/sandbox/mux-proxy.ts index 766a39406..6c69d1173 100644 --- a/src/sandbox/mux-proxy.ts +++ b/src/sandbox/mux-proxy.ts @@ -7,6 +7,7 @@ import { unlink } from 'node:fs/promises' import { logForDebugging } from '../utils/debug.js' import { getPlatform } from '../utils/platform.js' import { listenInRange } from './listen-in-range.js' +import { closeAfterFlush } from './tls-terminate-proxy.js' /** * First-byte values that select the SOCKS handler. SOCKS5's greeting is @@ -138,7 +139,8 @@ export function createMuxProxyServer(opts: MuxProxyOptions): MuxProxyServer { }) client.on('error', () => upstream.destroy()) client.once('close', () => upstream.destroy()) - upstream.once('close', () => client.destroy()) + // What is still queued for the client then is the end of an answer. + upstream.once('close', () => closeAfterFlush(client, () => {})) client.pipe(upstream) upstream.pipe(client) } diff --git a/src/sandbox/sandbox-config.ts b/src/sandbox/sandbox-config.ts index e74d69c0c..4ddcef481 100644 --- a/src/sandbox/sandbox-config.ts +++ b/src/sandbox/sandbox-config.ts @@ -879,13 +879,16 @@ export const NetworkConfigSchema = z.object({ .describe( 'At most this many CONNECT tunnels are TLS-terminated at once ' + '(default 256); past it a CONNECT is answered 503 with ' + - '"X-Proxy-Error: too-many-tunnels". A tunnel holds its slot from ' + - 'its CONNECT until it closes, so idle keep-alive tunnels count; ' + + '"X-Proxy-Error: too-many-tunnels", and the refusal is recorded ' + + 'as a violation, once per command until half the slots are free ' + + 'again. A tunnel holds its slot from its CONNECT until it ' + + 'closes, so idle keep-alive tunnels count until they are closed ' + + '(6 s after a response); ' + 'one that turns out not to carry TLS frees it. The slot is taken ' + 'before the first bytes are seen, so at the cap a non-TLS CONNECT ' + '(SSH, say) to a host that would be terminated is refused too. ' + - 'Hosts in excludeDomains do not count. Read when the proxy ' + - 'starts; updateConfig does not change it.', + 'Hosts in excludeDomains do not count. Read at each CONNECT, so ' + + 'updateConfig changes it for the tunnels that follow.', ), handshakeTimeoutMs: z .number() @@ -897,8 +900,8 @@ export const NetworkConfigSchema = z.object({ 'A tunnel to be TLS-terminated must finish its TLS handshake ' + 'within this many milliseconds of its CONNECT (default 10000), ' + 'or it is closed and its slot freed. A tunnel past its handshake ' + - 'is never timed out by it. Read when the proxy starts; ' + - 'updateConfig does not change it.', + 'is never timed out by it. Read at each CONNECT, so ' + + 'updateConfig changes it for the tunnels that follow.', ), }) .refine(o => !o.caCertPath === !o.caKeyPath, { diff --git a/src/sandbox/sandbox-manager.ts b/src/sandbox/sandbox-manager.ts index 2bdda4b9c..9c54a2ee6 100644 --- a/src/sandbox/sandbox-manager.ts +++ b/src/sandbox/sandbox-manager.ts @@ -744,11 +744,18 @@ async function startMuxProxyServer( stripResponseHeaders: config?.network.stripResponseHeaders, refuseOpaqueTunnels: config?.network.refuseOpaqueTunnels, requireHostMatch: config?.network.requireHostMatch, - maxTerminatedTunnels: config?.network.tlsTerminate?.maxTunnels, - tlsHandshakeTimeoutMs: config?.network.tlsTerminate?.handshakeTimeoutMs, + // Read at each CONNECT, so that updateConfig changes them. + get maxTerminatedTunnels() { + return config?.network.tlsTerminate?.maxTunnels + }, + get tlsHandshakeTimeoutMs() { + return config?.network.tlsTerminate?.handshakeTimeoutMs + }, onFilterRequestDenied: ({ method, url, reason, encodedCommand }) => { + // A refused CONNECT comes with its authority (`host:port`), not a URL. + const target = method === 'CONNECT' ? url : redactUrlForViolation(url) recordProxyViolation( - `deny http-request ${method} ${redactUrlForViolation(url)} (${reason})`, + `deny http-request ${method} ${target} (${reason})`, encodedCommand, ) }, diff --git a/src/sandbox/tls-terminate-proxy.ts b/src/sandbox/tls-terminate-proxy.ts index 4758e4786..a0ce8acd5 100644 --- a/src/sandbox/tls-terminate-proxy.ts +++ b/src/sandbox/tls-terminate-proxy.ts @@ -11,7 +11,12 @@ import { Agent as HttpsAgent, request as httpsRequest } from 'node:https' import { createServer as createHttpServer } from 'node:http' -import type { ClientRequest, IncomingMessage, ServerResponse } from 'node:http' +import type { + ClientRequest, + IncomingMessage, + Server, + ServerResponse, +} from 'node:http' import type { LookupFunction, Socket } from 'node:net' import { checkServerIdentity, TLSSocket } from 'node:tls' import type { TLSSocketOptions } from 'node:tls' @@ -261,6 +266,28 @@ const MAX_QUEUED_FOR_CLIENT = 1 << 20 */ const MAX_BUFFERED_REQUEST_BODY = 4 << 20 +/** How long a tunnel may stay as it is before boundTunnel closes it. */ +export type TunnelTimeouts = { + /** From the ClientHello to the first request's whole head. */ + firstRequestMs: number + /** What `Keep-Alive: timeout` says: from a response to the next request. */ + keepAliveMs: number + /** How much later than it says an idle tunnel is ended, for a client that takes it at its word. */ + keepAliveMarginMs: number + /** With a request or a response under way that does not move. */ + stalledMs: number +} + +export const TUNNEL_TIMEOUTS: TunnelTimeouts = { + firstRequestMs: 10_000, + keepAliveMs: 5_000, + keepAliveMarginMs: 1_000, + stalledMs: 300_000, +} + +/** A tunnel with something under way is looked at this many times in a timeout. */ +const LOOKS = 5 + export type TerminateTarget = { /** * Canonical (see canonicalizeHost) CONNECT target: what the allowlist @@ -308,6 +335,8 @@ export type TerminateTarget = { onHandshakeProbe?: (isDone: () => boolean) => void /** Shared across a proxy's tunnels: bytes they may hold for slow parties (see ByteBudget). */ byteBudget?: ByteBudget + /** In place of TUNNEL_TIMEOUTS (see HttpProxyServerOptions.tunnelTimeouts). */ + timeouts?: Partial } /** @@ -340,12 +369,14 @@ export function terminateAndForward( // The tunnel's one connection is its client's, so the server name in its // ClientHello is the tunnel's (read below, before the handshake). let serverName: ServerName = { kind: 'absent' } + const timeouts = { ...TUNNEL_TIMEOUTS, ...target.timeouts } // Never listens: the TLS socket below is handed to it. - const inner = createHttpServer( - target.maxHeaderSize !== undefined + const inner = createHttpServer({ + keepAliveTimeout: timeouts.keepAliveMs, + ...(target.maxHeaderSize !== undefined ? { maxHeaderSize: target.maxHeaderSize } - : {}, - ) + : {}), + }) // Needs an http.Server that serves a connection handed to it (Node, Bun // 1.4 and later: such a server has a 'connection' listener of its own). @@ -496,7 +527,9 @@ export function terminateAndForward( tls.once('data', onEstablished) // Neither event comes for a client that finishes the handshake and then // stays quiet where 'secure' is not said, so a deadline asks instead. - target.onHandshakeProbe?.(() => established || isTlsHandshakeDone(tls)) + const handshakeDone = (): boolean => established || isTlsHandshakeDone(tls) + target.onHandshakeProbe?.(handshakeDone) + boundTunnel(inner, tls, outbound, socket, timeouts, handshakeDone) tls.once('secure', () => { onEstablished() // Where the runtime reports the name its TLS layer took, it must be @@ -551,6 +584,109 @@ export function terminateAndForward( } } +/** + * Close a tunnel that would otherwise stay open, and hold its slot, for as + * long as its client or its upstream leaves it so. Its server never listens, + * so of a server's connection timeouts the runtime applies keepAliveTimeout + * alone (Node) or none (Bun). + * + * - Idle, that is no response owed (no request yet, between two, part of a + * head) and nothing queued for the client: ended with close_notify and a + * FIN, and destroyed if it is still there, and idle, as long again. + * - Under way, that is a request not all in, the tunnel paused for an + * upstream that takes no more of one (so that not even its client's + * leaving is seen), or bytes queued for a client: destroyed once nothing + * has moved for `stalledMs`. A client that goes on reading loses nothing. + * That it reads is seen when the kernel takes more, which it does once a + * good part of its send buffer is free: up to a megabyte or two at a time. + * + * A whole request that waits for its response is left alone: the tunnel is + * read meanwhile, so it closes when its client does. + * + * @internal Exported for its test. + */ +export function boundTunnel( + inner: Server, + tls: TLSSocket, + outbound: Writable, + socket: Duplex, + { firstRequestMs, keepAliveMs, keepAliveMarginMs, stalledMs }: TunnelTimeouts, + handshakeDone: () => boolean, +): void { + // What is on its way to the client and still in this process. No layer + // counts it all: a response is done, and Bun's TLS socket empty and + // finished, with a megabyte in the stream under it. (Under Node two of + // the three are one: all that is asked is whether it is zero, or changes.) + const queued = (): number => + tls.writableLength + outbound.writableLength + socket.writableLength + + let owed = 0 + let idle: ReturnType + const endWhenIdleFor = (ms: number): void => { + const busy = queued() > 0 + idle = setTimeout( + () => { + // Not idle for all of `ms`: a handshake under way has a deadline of + // its own, and what is queued is the client's to take. + if (!handshakeDone() || busy || queued() > 0) return endWhenIdleFor(ms) + // Ended, and still here: not every client answers an end, and not + // every runtime closes the socket under a TLS socket it ends. + if (tls.writableEnded) return void socket.destroy() + logForDebugging(`[tls-terminate] tunnel idle for ${ms} ms; closed`) + tls.end() + endWhenIdleFor(ms) + }, + busy ? ms / LOOKS : ms, + ) + idle.unref?.() + } + endWhenIdleFor(firstRequestMs) + // Node's own keep-alive timeout destroys the socket, whatever is queued. + inner.on('timeout', () => {}) + + let receiving: IncomingMessage | undefined + // A tunnel moves when bytes are read, when it is taken up again after a + // pause (what is read then, a runtime may have read ahead and counted), + // and when bytes written to it leave the queue. + let resumed = 0 + tls.on('resume', () => resumed++) + let moved = 0 + let still = 0 + const stalled = setInterval(() => { + const before = moved + moved = tls.bytesRead + resumed + tls.bytesWritten - queued() + const underWay = + tls.isPaused() || receiving?.complete === false || queued() > 0 + still = underWay && moved === before ? still + 1 : 0 + if (still < LOOKS) return + logForDebugging( + `[tls-terminate] nothing has moved for ${stalledMs} ms; tunnel closed`, + ) + socket.destroy() + }, stalledMs / LOOKS) + stalled.unref?.() + + inner.on('request', (req, res) => { + owed++ + clearTimeout(idle) + receiving = req + // Once the answer is out nothing reads the request on (an upstream that + // answered or failed early took part of it), and a runtime's server + // leaves alone a request that was being read. The rest is dropped, so + // that the tunnel is read on: to the next request, or its client's end. + res.once('finish', () => req.unpipe().resume()) + res.once('close', () => { + if (--owed === 0 && !socket.destroyed) { + endWhenIdleFor(keepAliveMs + keepAliveMarginMs) + } + }) + }) + socket.once('close', () => { + clearTimeout(idle) + clearInterval(stalled) + }) +} + /** * Read on until `head` holds the whole first TLS record (the ClientHello, * at most 16 KiB of it), the client stops sending, or 5 s pass. Pull-mode @@ -672,11 +808,14 @@ function hostNameEntry(body: Buffer): ServerName { * the client ends its side, or after a grace period. Until then what the * client sends is read and dropped, since unread input when the socket * closes turns the close into a reset, and a reset discards what the - * client has not read yet. A backstop covers a client that stops reading. + * client has not read yet. A client that has not taken it all in + * `stalledMs` is cut: what is queued leaves the queue all at once, so one + * that reads slowly cannot be told from one that has stopped. */ export function closeAfterFlush( socket: Duplex, done: (how: string) => void, + stalledMs = TUNNEL_TIMEOUTS.stalledMs, ): void { if (socket.destroyed) { done('the client, earlier') @@ -688,13 +827,16 @@ export function closeAfterFlush( const close = (how: string): void => { if (settled) return settled = true - clearTimeout(backstop) + clearTimeout(stalled) clearTimeout(grace) if (!socket.destroyed) socket.destroy() done(how) } - const backstop = setTimeout(() => close('the 30 s backstop'), 30_000) - backstop.unref?.() + const stalled = setTimeout( + () => close(`not flushed in ${stalledMs} ms`), + stalledMs, + ) + stalled.unref?.() let grace: ReturnType | undefined const onFinish = (): void => { finished = true diff --git a/test/fixtures/tunnel-ends.mjs b/test/fixtures/tunnel-ends.mjs new file mode 100644 index 000000000..13a5a3560 --- /dev/null +++ b/test/fixtures/tunnel-ends.mjs @@ -0,0 +1,610 @@ +/** + * The scenarios of tls-terminate-tunnel-ends.test.ts, run as a child under + * Node and under Bun against dist/. Each scenario has a proxy with one + * tunnel slot and an upstream of its own, and says how its tunnel ended. + * argv[2] is `accepted` (the proxy listens) or `handed` (it is given each + * connection with emit('connection')); argv[3], if given, names the one + * scenario to run. Prints one JSON object. (`exit with a tunnel open` prints + * nothing: it opens a tunnel and has nothing more to do.) + */ +import { readFileSync } from 'node:fs' +import { IncomingMessage } from 'node:http' +import { connect, createServer } from 'node:net' +import { Duplex } from 'node:stream' +import { fileURLToPath } from 'node:url' +import { + connect as tlsConnect, + createServer as createTlsServer, +} from 'node:tls' + +const dist = f => new URL(`../../dist/sandbox/${f}.js`, import.meta.url).href +const { createHttpProxyServer } = await import(dist('http-proxy')) +const { createByteBudget } = await import(dist('emitted-connection')) +const { createMitmCA } = await import(dist('mitm-ca')) +const { mintLeafCert } = await import(dist('mitm-leaf')) + +// Whether a server drains a request that was being read when its response +// ends differs by runtime and version. Here none does. +IncomingMessage.prototype._dump = function () {} + +const HANDED = process.argv[2] === 'handed' +const MiB = 1 << 20 +/** More than a proxy that stops reading, and the kernel, take in. */ +const UPLOAD = 24 * MiB +/** A tunnel still open this long after it should have closed counts as open. */ +const GIVE_UP_MS = 3_000 +/** A timeout that plays no part in a scenario. */ +const NEVER = 600_000 +const NO_TIMEOUTS = { + firstRequestMs: NEVER, + keepAliveMs: NEVER, + keepAliveMarginMs: 0, + stalledMs: NEVER, +} +const sleep = ms => new Promise(r => setTimeout(r, ms)) + +const pem = f => fileURLToPath(new URL(`tls-terminate/${f}`, import.meta.url)) +const caPem = readFileSync(pem('ca.crt'), 'utf8') +const mitmCA = createMitmCA({ + caCertPath: pem('ca.crt'), + caKeyPath: pem('ca.key'), +}) +const leaf = mintLeafCert(mitmCA, '127.0.0.1') +const listen = server => + new Promise(r => + server.listen(0, '127.0.0.1', () => r(server.address().port)), + ) +/** Nothing listens there. */ +const CLOSED = '127.0.0.1:1' + +/** + * A TLS upstream. It reads a request's head; a GET is answered by `onGet` + * (default: 200 at once), and a POST is left to `onPost`, paused, with how + * much of its body came with the head. + */ +async function startUpstream({ onPost, onGet } = {}) { + const open = new Set() + const answer = (s, status, body = 'ok') => + s.write( + `HTTP/1.1 ${status} X\r\nContent-Length: ${body.length}\r\n\r\n${body}`, + ) + const server = createTlsServer( + { cert: leaf.certPem, key: leaf.keyPem }, + s => { + open.add(s) + s.on('error', () => {}) + s.once('close', () => open.delete(s)) + let seen = Buffer.alloc(0) + const onHead = d => { + seen = Buffer.concat([seen, d]) + const end = seen.indexOf('\r\n\r\n') + if (end < 0) return + const rest = seen.length - end - 4 + const isPost = seen.toString('latin1', 0, 4) === 'POST' + seen = Buffer.alloc(0) + if (!isPost) return void (onGet ?? answer)(s, 200) + s.off('data', onHead) + s.pause() + onPost(s, answer, rest) + } + s.on('data', onHead) + }, + ) + return { + target: `127.0.0.1:${await listen(server)}`, + /** The connections the proxy still holds (a paused one must read to see its end). */ + async stillOpen() { + for (const s of open) s.resume() + await sleep(100) + return open.size + }, + } +} + +/** An `onGet` that answers with `size` bytes (a multiple of 64 KiB), as fast as they are taken. */ +const longAnswer = size => s => { + s.write(`HTTP/1.1 200 X\r\nContent-Length: ${size}\r\n\r\n`) + const chunk = Buffer.alloc(64 << 10, 7) + let sent = 0 + const pump = () => { + while (sent < size) { + sent += chunk.length + if (!s.write(chunk)) return void s.once('drain', pump) + } + } + pump() +} + +/** + * A tunnel through a proxy with one slot in front of `target`, its handshake + * done. `halfOpen`: the client's socket does not end because the proxy's did. + */ +async function openTunnel({ + target, + timeouts, + trustUpstream = true, + halfOpen = false, + ...options +}) { + const budget = createByteBudget(256 * MiB) + const proxy = createHttpProxyServer({ + filter: () => true, + filterRequest: () => ({ action: 'allow' }), + mitmCA, + tlsTerminateUpstreamCA: trustUpstream ? caPem : undefined, + maxTerminatedTunnels: 1, + tunnelTimeouts: timeouts, + byteBudget: budget, + ...options, + }) + const front = HANDED + ? createServer(s => { + s.on('error', () => {}) + proxy.emit('connection', s) + }) + : proxy + const port = await listen(front) + /** The status of the answer to a CONNECT, and its socket. */ + const connectTo = (allowHalfOpen = false) => + new Promise(resolve => { + const raw = connect({ port, host: '127.0.0.1', allowHalfOpen }, () => + raw.write(`CONNECT ${target} HTTP/1.1\r\nHost: ${target}\r\n\r\n`), + ) + raw.on('error', () => {}) + raw.once('data', d => + resolve({ status: Number(String(d).split(' ')[1]), raw }), + ) + }) + + const { raw } = await connectTo(halfOpen) + // The TLS client runs on a stream over the socket, so that what the proxy + // sends can be counted. + let got = 0 + let mute = false + const wire = new Duplex({ + read() {}, + write: (chunk, _encoding, callback) => + mute ? callback() : raw.write(chunk, callback), + final(callback) { + if (!mute) raw.end() + callback() + }, + }) + raw.on('data', d => { + got += d.length + wire.push(d) + }) + raw.once('end', () => wire.push(null)) + const closed = new Promise(r => raw.once('close', () => r(Date.now()))) + const t = tlsConnect({ + socket: wire, + ca: caPem, + checkServerIdentity: () => undefined, + }) + t.on('error', () => {}) + await new Promise(r => t.once('secureConnect', r)) + let text = '' + // The bytes of a long answer's body are counted, not kept, once its head + // is in. + let long = false + let body + let lastByteAt = Date.now() + t.on('data', d => { + lastByteAt = Date.now() + if (body !== undefined) return void (body += d.length) + text += d.toString('latin1') + const head = text.indexOf('\r\n\r\n') + if (!long || head < 0) return + body = text.length - head - 4 + text = '' + }) + + /** The next whole answer (status 0: none came). */ + const answer = async () => { + for (const until = Date.now() + 2 * GIVE_UP_MS; ; await sleep(5)) { + const m = /^(HTTP\/1\.1 (\d+)[^]*?\r\n\r\n)/.exec(text) + const length = Number(/content-length: (\d+)/i.exec(m?.[1])?.[1] ?? 0) + if (m && text.length >= m[1].length + length) { + text = text.slice(m[1].length + length) + return { status: Number(m[2]), head: m[1] } + } + if (raw.destroyed || Date.now() > until) return { status: 0, head: '' } + } + } + return { + answer, + write: bytes => void t.write(bytes), + get() { + t.write(`GET / HTTP/1.1\r\nHost: ${target}\r\n\r\n`) + return answer() + }, + /** + * Send a POST of `size`. Resolves true once it has all gone out, false + * once nothing of it has for 200 ms. + */ + async post(size, { host = target, path = '/' } = {}) { + t.write( + `POST ${path} HTTP/1.1\r\nHost: ${host}\r\nContent-Length: ${size}\r\n\r\n`, + ) + const chunk = Buffer.alloc(64 << 10, 7) + let sent = 0 + let out = false + const pump = () => { + while (sent < size && !t.destroyed) { + sent += chunk.length + const last = sent >= size ? () => (out = true) : undefined + if (!t.write(chunk, last)) return void t.once('drain', pump) + } + } + pump() + for (let seen = -1; !out && seen !== raw.bytesWritten; await sleep(200)) { + seen = raw.bytesWritten + } + return out + }, + /** Ask for a long answer; `headers` go with the request. */ + askForLong(headers = '') { + long = true + t.write(`GET / HTTP/1.1\r\nHost: ${target}\r\n${headers}\r\n`) + }, + /** Take nothing more from the socket, or take it up again. */ + stopReading: () => void raw.pause(), + readOn: () => void raw.resume(), + /** How much of the first `size` bytes of a long answer never came. */ + async missingOf(size) { + const missing = () => Math.max(0, size - (body ?? 0)) + const until = Date.now() + 20_000 + while (missing() > 0 && !raw.destroyed && Date.now() < until) { + await sleep(5) + } + return missing() + }, + leave: () => void t.end(), + /** From now on the client sends nothing, not even its end. */ + fallSilent: () => void (mute = true), + /** When the tunnel's slot is free again, from the proxy's last byte. */ + async slotFree() { + for (const until = Date.now() + GIVE_UP_MS; Date.now() < until; ) { + const next = await connectTo() + next.raw.destroy() + if (next.status === 200) return Date.now() - lastByteAt + await sleep(25) + } + return 'held' + }, + /** Let neither the proxy's listener nor the client's socket hold the process. */ + unref() { + front.unref() + raw.unref() + }, + /** + * How the tunnel ends: when, from the proxy's last byte or from `since`; + * whether the proxy said so first (a close_notify); whether its slot + * and the bytes counted for it are free again. + */ + async end(since = lastByteAt) { + const before = got + const closedAt = await Promise.race([closed, sleep(GIVE_UP_MS)]) + raw.destroy() + await sleep(50) + const next = await connectTo() + next.raw.destroy() + return { + closedAfterMs: closedAt === undefined ? 'open' : closedAt - since, + closeNotify: got > before, + nextConnect: next.status, + heldBytes: budget.used, + } + }, + } +} + +/** An upload answered before it is all in; `then` is what its client does next. */ +async function answeredEarly(then, { host, path, ...options } = {}) { + const up = await startUpstream({ + onPost: (s, answer) => setTimeout(() => answer(s, 413, 'early'), 200), + }) + const c = await openTunnel({ + target: up.target, + timeouts: NO_TIMEOUTS, + ...options, + }) + const sent = c.post(UPLOAD, { host, path }) + const { status } = await c.answer() + const more = await then(c, await sent) + return { status, ...more, ...(await c.end()), upstream: await up.stillOpen() } +} +const leaves = c => c.leave() + +/** Timed to a few hundred ms: run together, with nothing heavy beside them. */ +const light = { + async 'idle after a response'() { + const up = await startUpstream() + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, keepAliveMs: 1_000, keepAliveMarginMs: 500 }, + }) + const { status, head } = await c.get() + const advertised = /keep-alive: ([^\r]*)/i.exec(head)?.[1] + return { + status, + advertised, + ...(await c.end()), + upstream: await up.stillOpen(), + } + }, + + async 'asked again when the advertised time is up'() { + const up = await startUpstream() + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, keepAliveMs: 1_000, keepAliveMarginMs: 700 }, + }) + await c.get() + await sleep(1_100) + return { status: (await c.get()).status } + }, + + // Node's server has a keep-alive timeout of its own, 1 s past the time + // advertised. + async 'a margin longer than the runtime has'() { + const up = await startUpstream() + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, keepAliveMs: 200, keepAliveMarginMs: 1_800 }, + }) + await c.get() + return c.end() + }, + + async 'idle, and the client answers no end'() { + const up = await startUpstream() + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, keepAliveMs: 500 }, + halfOpen: true, + }) + await c.get() + c.fallSilent() + return { slotFreeAfterMs: await c.slotFree() } + }, + + async 'the default Keep-Alive'() { + const up = await startUpstream() + const c = await openTunnel({ target: up.target }) + const { head } = await c.get() + c.leave() + return { advertised: /keep-alive: ([^\r]*)/i.exec(head)?.[1] } + }, + + async 'each response starts the idle time anew'() { + const up = await startUpstream() + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, keepAliveMs: 800 }, + }) + const statuses = [] + for (let i = 0; i < 3; i++) { + if (i) await sleep(500) + statuses.push((await c.get()).status) + } + return { statuses, ...(await c.end()) } + }, + + async 'no request after the handshake'() { + const c = await openTunnel({ + target: CLOSED, + timeouts: { ...NO_TIMEOUTS, firstRequestMs: 800 }, + }) + return c.end(Date.now()) + }, + + async 'half a head'() { + const c = await openTunnel({ + target: CLOSED, + timeouts: { ...NO_TIMEOUTS, firstRequestMs: 800 }, + }) + c.write('GET / HTTP/1.1\r\nX-Half') + return c.end(Date.now()) + }, + + async 'half a head after a response'() { + const up = await startUpstream() + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, keepAliveMs: 800 }, + }) + const { status } = await c.get() + c.write('GET / HTTP/1.1\r\nX-Half') + return { status, ...(await c.end()) } + }, + + async 'half a body'() { + const up = await startUpstream({ onPost: s => s.resume() }) + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, stalledMs: 500 }, + }) + c.write( + `POST / HTTP/1.1\r\nHost: ${up.target}\r\nContent-Length: 10\r\n\r\nhalf`, + ) + return { ...(await c.end(Date.now())), upstream: await up.stillOpen() } + }, + + async 'a body that trickles in'() { + const up = await startUpstream({ + onPost(s, answer, rest) { + let taken = rest + s.on('data', d => (taken += d.length) >= 8 && answer(s, 200)) + s.resume() + }, + }) + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, stalledMs: 700 }, + }) + c.write( + `POST / HTTP/1.1\r\nHost: ${up.target}\r\nContent-Length: 8\r\n\r\n`, + ) + for (let i = 0; i < 8; i++) { + await sleep(250) + c.write('.') + } + return { status: (await c.answer()).status } + }, + + async 'an answer that takes longer than every timeout'() { + const up = await startUpstream({ + onGet: s => + setTimeout( + () => s.write('HTTP/1.1 200 X\r\nContent-Length: 0\r\n\r\n'), + 1_200, + ), + }) + const c = await openTunnel({ + target: up.target, + timeouts: { firstRequestMs: 300, keepAliveMs: 300, stalledMs: 300 }, + }) + return { status: (await c.get()).status } + }, +} + +/** Much to carry: run one at a time. */ +const heavy = { + // What the kernel takes for a client that reads nothing differs, and only + // an answer up to a megabyte longer than that is all out, and yet queued. + async 'readers that take nothing for a while, then everything'() { + const cases = [] + for (let size = MiB; size <= 6 * MiB; size += MiB / 2) { + cases.push([size, ''], [size, 'Connection: close\r\n']) + } + const ends = await Promise.all( + cases.map(async ([size, headers]) => { + const up = await startUpstream({ onGet: longAnswer(size) }) + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, keepAliveMs: 200 }, + }) + c.stopReading() + c.askForLong(headers) + await sleep(1_000) + c.readOn() + return [await c.missingOf(size), (await c.end()).closedAfterMs] + }), + ) + // All of it came, and then the tunnel was closed as any idle one is. + return { missing: ends.filter(([n, closed]) => n > 0 || closed === 'open') } + }, + + 'answered early, the client leaves': () => answeredEarly(leaves), + 'upstream port closed, the client leaves': () => + answeredEarly(leaves, { target: CLOSED }), + 'upstream certificate untrusted, the client leaves': () => + answeredEarly(leaves, { trustUpstream: false }), + 'refused for its Host, the client leaves': () => + answeredEarly(leaves, { requireHostMatch: true, host: 'other.test' }), + 'refused for its target, the client leaves': () => + answeredEarly(leaves, { path: '*' }), + 'answered early, the client stays': () => + answeredEarly(() => {}, { timeouts: { ...NO_TIMEOUTS, keepAliveMs: 600 } }), + 'answered early, the client sends on and asks again': () => + answeredEarly(async (c, allSent) => { + const second = allSent && (await c.get()).status + c.leave() + return { second } + }), + + async 'a reader that takes an answer in bursts'() { + const up = await startUpstream({ onGet: longAnswer(UPLOAD) }) + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, keepAliveMs: 200, stalledMs: 1_000 }, + }) + c.askForLong() + // 6 MiB, then nothing for 400 ms, and so on: longer in all than + // stalledMs, never that long at a time. + for (let next = 6 * MiB; next < UPLOAD; next += 6 * MiB) { + await c.missingOf(next) + c.stopReading() + await sleep(400) + c.readOn() + } + return { missing: await c.missingOf(UPLOAD) } + }, + + async 'a client that never reads its answer'() { + const up = await startUpstream({ onGet: longAnswer(UPLOAD) }) + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, stalledMs: 500 }, + }) + c.stopReading() + c.askForLong() + return { + slotFreeAfterMs: await c.slotFree(), + upstream: await up.stillOpen(), + } + }, + + async 'an upload the upstream stops taking'() { + const up = await startUpstream({ onPost() {} }) + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, stalledMs: 500 }, + }) + const since = Date.now() + await c.post(2 * UPLOAD) + return { ...(await c.end(since)), upstream: await up.stillOpen() } + }, + + async 'an upload the upstream takes in bursts'() { + const size = 2 * UPLOAD + const up = await startUpstream({ + // 6 MiB (more than the proxy holds back, so that it reads on), then + // nothing for 300 ms, and so on: far longer in all than stalledMs, + // never that long at a time. + onPost(s, answer, rest) { + let taken = rest + let burst = 0 + s.on('data', d => { + taken += d.length + burst += d.length + if (taken >= size) return void answer(s, 200) + if (burst < 6 * MiB) return + burst = 0 + s.pause() + setTimeout(() => s.resume(), 300) + }) + s.resume() + }, + }) + const c = await openTunnel({ + target: up.target, + timeouts: { ...NO_TIMEOUTS, stalledMs: 700 }, + }) + void c.post(size) + return { status: (await c.answer()).status } + }, +} + +if (process.argv[3] === 'exit with a tunnel open') { + const c = await openTunnel({ + target: CLOSED, + timeouts: NO_TIMEOUTS, + }) + c.unref() +} else { + const only = ([name]) => + process.argv[3] === undefined || name === process.argv[3] + const results = Object.fromEntries( + await Promise.all( + Object.entries(light) + .filter(only) + .map(async ([name, run]) => [name, await run()]), + ), + ) + for (const [name, run] of Object.entries(heavy).filter(only)) { + results[name] = await run() + } + process.stdout.write(JSON.stringify(results), () => process.exit(0)) +} diff --git a/test/sandbox/byte-budget.test.ts b/test/sandbox/byte-budget.test.ts index 1da715635..5ecadb1dd 100644 --- a/test/sandbox/byte-budget.test.ts +++ b/test/sandbox/byte-budget.test.ts @@ -1,7 +1,9 @@ import { expect, test } from 'bun:test' +import { EventEmitter } from 'node:events' import { adjustByteBudget, createByteBudget, + gateSocketOnBufferedBody, } from '../../src/sandbox/emitted-connection.js' test('a drop in a byte budget wakes each waiter once and empties the set', () => { @@ -57,3 +59,24 @@ test('many add and release cycles leave a byte budget empty', () => { expect(budget.used).toBe(0) expect(budget.waiters.size).toBe(0) }) + +test.each(['end', 'close'])( + "what a gated request holds is given back at its %s, or its socket's close, and the socket is let go of", + event => { + for (const from of ['req', 'socket'] as const) { + const budget = createByteBudget(1000) + const socket = Object.assign(new EventEmitter(), { pause() {} }) + const req = Object.assign(new EventEmitter(), { + socket, + readableLength: 10, + }) + gateSocketOnBufferedBody(req as never, 100, budget) + socket.emit('data') + expect(budget.used).toBe(10) + if (from === 'req') req.emit(event) + else socket.emit('close') + expect(budget.used).toBe(0) + expect(socket.eventNames()).toEqual([]) + } + }, +) diff --git a/test/sandbox/close-after-flush.test.ts b/test/sandbox/close-after-flush.test.ts index 2a540c916..0a2738405 100644 --- a/test/sandbox/close-after-flush.test.ts +++ b/test/sandbox/close-after-flush.test.ts @@ -54,4 +54,25 @@ describe('closeAfterFlush', () => { expect(read).toBe(size) expect(await how).toBe('the client ending after the flush') }, 60_000) + + test('a client that takes nothing is cut after the time allowed, not before', async () => { + const began = Date.now() + const how = await new Promise(closedBy => { + const server = createServer((s: Socket) => { + s.on('error', () => {}) + // More than the kernel takes for a client that reads nothing. + s.write(Buffer.alloc(32 << 20, 0x61)) + closeAfterFlush(s, closedBy, 400) + }) + cleanup.push(() => server.close()) + server.listen(0, '127.0.0.1', () => { + const c = connect((server.address() as AddressInfo).port, '127.0.0.1') + c.on('error', () => {}) + c.pause() + cleanup.push(() => c.destroy()) + }) + }) + expect(how).toBe('not flushed in 400 ms') + expect(Date.now() - began).toBeGreaterThanOrEqual(400) + }, 5_000) }) diff --git a/test/sandbox/mux-proxy.test.ts b/test/sandbox/mux-proxy.test.ts index c1be703a0..f795cff0d 100644 --- a/test/sandbox/mux-proxy.test.ts +++ b/test/sandbox/mux-proxy.test.ts @@ -100,6 +100,26 @@ describe('mux-proxy first-byte dispatch', () => { expect(text).toContain('http-connect:x:443') }) + it('ends a client whose backend has closed without a reset, which would cost it what it has not read yet', async () => { + const errors: unknown[] = [] + const hadError = await new Promise(resolve => { + const c = connect({ port, host: '127.0.0.1', allowHalfOpen: true }, () => + c.write('CONNECT example.com:443 HTTP/1.1\r\n\r\n'), + ) + c.on('error', err => errors.push(err)) + c.resume() + // Bytes sent to a socket that is closed are answered with a reset, + // and the next write fails. + c.once('end', () => { + c.write('late') + setTimeout(() => c.write('later'), 100) + setTimeout(() => c.end(), 200) + }) + c.once('close', resolve) + }) + expect([errors, hadError]).toEqual([[], false]) + }) + it('routes a TLS ClientHello first byte (0x16) to HTTP, not SOCKS', async () => { // Not a valid HTTP request — the backend will 400 it — but the point // is it must NOT be treated as SOCKS. diff --git a/test/sandbox/tls-terminate-runtime.test.ts b/test/sandbox/tls-terminate-runtime.test.ts index 8db0469b5..6441d2b81 100644 --- a/test/sandbox/tls-terminate-runtime.test.ts +++ b/test/sandbox/tls-terminate-runtime.test.ts @@ -5,6 +5,10 @@ import { createHttpProxyServer } from '../../src/sandbox/http-proxy.js' import { createMitmCA, generateCa } from '../../src/sandbox/mitm-ca.js' import { SandboxManager } from '../../src/sandbox/sandbox-manager.js' import type { SandboxRuntimeConfig } from '../../src/sandbox/sandbox-config.js' +import { + encodeSandboxedCommand, + proxyUsernameFor, +} from '../../src/sandbox/sandbox-utils.js' import { SERVES_EMITTED_CONNECTIONS } from '../helpers/emitted-connections.js' // TLS termination runs in this process and needs an http.Server that @@ -86,10 +90,13 @@ describe('a runtime that cannot terminate TLS in-process fails at start-up', () ) }) -/** A CONNECT to a.test:443 through the manager's proxy: the reply head and the socket. */ -function connectThroughManager(): Promise<{ head: string; sock: Socket }> { +/** A CONNECT to a.test:443 through the manager's proxy, as `command` if given: the reply head and the socket. */ +function connectThroughManager( + command?: string, +): Promise<{ head: string; sock: Socket }> { const port = SandboxManager.getProxyPort()! - const auth = Buffer.from(`srt:${SandboxManager.getProxyAuthToken()!}`) + const user = proxyUsernameFor(command && encodeSandboxedCommand(command)) + const auth = Buffer.from(`${user}:${SandboxManager.getProxyAuthToken()!}`) return new Promise(resolve => { const sock = connect(port, '127.0.0.1', () => sock.write( @@ -135,5 +142,44 @@ describe.skipIf(!SERVES_EMITTED_CONNECTIONS)( const third = await connectThroughManager() expect(third.head).toStartWith('HTTP/1.1 200') }, 20_000) + + test("the cap's 503 is recorded as a violation of the command it refused", async () => { + cleanup.push(() => SandboxManager.reset()) + await SandboxManager.initialize( + baseConfig({ tlsTerminate: { maxTunnels: 1 } }), + ) + await connectThroughManager('holds the slot') + const refused = await connectThroughManager('curl https://a.test/') + expect(refused.head).toStartWith('HTTP/1.1 503') + expect( + SandboxManager.getSandboxViolationStore() + .getViolationsForCommand('curl https://a.test/') + .map(v => v.line), + ).toEqual([ + "deny http-request CONNECT a.test:443 (the sandbox's limit of 1 " + + 'TLS-terminated tunnels at once is reached; ' + + 'the limit is network.tlsTerminate.maxTunnels in the ' + + "sandbox's configuration)", + ]) + }, 20_000) + + test('updateConfig changes both for the CONNECTs that follow', async () => { + cleanup.push(() => SandboxManager.reset()) + await SandboxManager.initialize( + baseConfig({ tlsTerminate: { maxTunnels: 1 } }), + ) + await connectThroughManager() + expect((await connectThroughManager()).head).toStartWith('HTTP/1.1 503') + SandboxManager.updateConfig( + baseConfig({ + tlsTerminate: { maxTunnels: 2, handshakeTimeoutMs: 300 }, + }), + ) + const second = await connectThroughManager() + expect(second.head).toStartWith('HTTP/1.1 200') + const openedAt = Date.now() + await new Promise(r => second.sock.once('close', r)) + expect(Date.now() - openedAt).toBeLessThan(5_000) + }, 20_000) }, ) diff --git a/test/sandbox/tls-terminate-tunnel-ends.test.ts b/test/sandbox/tls-terminate-tunnel-ends.test.ts new file mode 100644 index 000000000..0f5a7e218 --- /dev/null +++ b/test/sandbox/tls-terminate-tunnel-ends.test.ts @@ -0,0 +1,417 @@ +import { afterEach, beforeAll, describe, expect, test } from 'bun:test' +import { spawn } from 'node:child_process' +import { EventEmitter, once } from 'node:events' +import { join } from 'node:path' +import { + boundTunnel, + TUNNEL_TIMEOUTS, +} from '../../src/sandbox/tls-terminate-proxy.js' +import { SERVES_EMITTED_CONNECTIONS } from '../helpers/emitted-connections.js' + +// Every way a TLS-terminated tunnel can end frees its slot, its upstream +// connection and the bytes counted for it within a bounded time, whatever +// its client or its upstream does or leaves undone. +// +// The runtimes differ in what their servers do for a connection handed to +// them, so the scenarios (test/fixtures/tunnel-ends.mjs) run as children, +// under Node and under Bun, against dist/. + +const FIXTURE = join(import.meta.dir, '..', 'fixtures', 'tunnel-ends.mjs') + +type Outcome = { + status?: number + second?: number | false + statuses?: number[] + advertised?: string + closedAfterMs: number | 'open' + closeNotify: boolean + nextConnect: number + heldBytes: number + upstream?: number + slotFreeAfterMs?: number | 'held' + missing?: number | unknown[] +} + +const RUNS = [ + { name: 'Node', bin: 'node', how: 'accepted' }, + ...(SERVES_EMITTED_CONNECTIONS + ? [ + { name: 'Bun', bin: process.execPath, how: 'accepted' }, + { name: 'Bun', bin: process.execPath, how: 'handed' }, + ] + : []), +] + +/** The children run side by side, from the first that is asked for. */ +let runs: Array>> | undefined +const outcomeOf = (i: number): Promise> => + (runs ??= RUNS.map(({ bin, how }) => run(bin, how)))[i]! + +function run(bin: string, how: string): Promise> { + return new Promise((resolve, reject) => { + const child = spawn(bin, [FIXTURE, how], { + stdio: ['ignore', 'pipe', 'inherit'], + }) + let out = '' + child.stdout.on('data', (d: Buffer) => (out += d.toString())) + child.once('error', reject) + child.once('close', () => + resolve(JSON.parse(out) as Record), + ) + }) +} + +/** Timers do not fire early, but the two ends do not start timing at once. */ +const EARLY_MS = 100 +/** A busy machine. Well under the fixture's GIVE_UP_MS. */ +const LATE_MS = 1_500 + +/** The tunnel closed `ms` (to `lastMs`) on, and nothing of it is left. */ +function expectClosed(o: Outcome, ms: number, lastMs = ms): void { + expect(o.closedAfterMs).toBeNumber() + expect(o.closedAfterMs as number).toBeGreaterThanOrEqual(ms - EARLY_MS) + expect(o.closedAfterMs as number).toBeLessThan(lastMs + LATE_MS) + expect(o).toMatchObject({ nextConnect: 200, heldBytes: 0 }) + if (o.upstream !== undefined) expect(o.upstream).toBe(0) +} + +test('the default timeouts', () => { + expect(TUNNEL_TIMEOUTS).toEqual({ + firstRequestMs: 10_000, + keepAliveMs: 5_000, + keepAliveMarginMs: 1_000, + stalledMs: 300_000, + }) +}) + +// Under Bun a tunnel's socket does not hold the process once the listener +// is unref'd (under Node it does), so the tunnel's timers must not either. +test.skipIf(!SERVES_EMITTED_CONNECTIONS)( + 'an open tunnel does not keep a process from ending', + async () => { + const child = spawn( + process.execPath, + [FIXTURE, 'accepted', 'exit with a tunnel open'], + { stdio: 'inherit' }, + ) + const giveUp = setTimeout(() => child.kill('SIGKILL'), 5_000) + const [code] = (await once(child, 'exit')) as [number | null] + clearTimeout(giveUp) + expect(code).toBe(0) + }, + 10_000, +) + +// What no client can be made to time, or Bun 1.4.2 to serve (a second +// request while the first is owed its response), with stand-ins for the parts. +describe('with stand-ins', () => { + const sleep = (ms: number): Promise => + new Promise(r => setTimeout(r, ms)) + const part = (more: object = {}): EventEmitter => + Object.assign(new EventEmitter(), more) + const request = (): EventEmitter => { + const req = part({ complete: true, unpipe: () => req, resume: () => req }) + return req + } + const sockets: EventEmitter[] = [] + afterEach(() => { + for (const socket of sockets.splice(0)) socket.emit('close') + }) + /** + * A tunnel idle after `idleMs` and stalled after `stalledMs`, its parts, + * and what is done to it. + */ + function tunnel({ + idleMs = 40, + paused = false, + handshakeDone = (): boolean => true, + stalledMs = 60_000, + } = {}) { + const done: string[] = [] + const inner = part() + const socket = part({ + writableLength: 0, + destroyed: false, + destroy: () => done.push('destroy'), + }) + sockets.push(socket) + const outbound = part({ writableLength: 0 }) + const tls = part({ + bytesRead: 7, + bytesWritten: 0, + writableLength: 0, + isPaused: () => paused, + writableEnded: false, + end: () => + Object.assign(tls, { writableEnded: true }) && done.push('end'), + }) + boundTunnel( + inner as never, + tls as never, + outbound as never, + socket as never, + { + firstRequestMs: idleMs, + keepAliveMs: idleMs - 10, + keepAliveMarginMs: 10, + stalledMs, + }, + handshakeDone, + ) + /** A request, whole, and its response; `answered()` closes that. */ + const ask = (): (() => void) => { + const res = part() + inner.emit('request', request(), res) + return () => void res.emit('close') + } + return { done, socket, outbound, tls, ask } + } + + test('a whole request whose last bytes paused its tunnel: closed once it has not moved', async () => { + const { done, socket, ask } = tunnel({ paused: true, stalledMs: 40 }) + ask() + await sleep(130) + expect(done[0]).toBe('destroy') + // And left alone from then on. + socket.emit('close') + const calls = done.length + await sleep(100) + expect(done).toHaveLength(calls) + }) + + test('a handshake under way is left to its own deadline', async () => { + let shaken = false + const { done } = tunnel({ handshakeDone: () => shaken }) + await sleep(100) + expect(done).toEqual([]) + shaken = true + await sleep(40) + expect(done).toEqual(['end']) + }) + + test('idle only once no response is owed', async () => { + const { done, ask } = tunnel() + const [first, second] = [ask(), ask()] + first() + await sleep(100) + expect(done).toEqual([]) + second() + await sleep(60) + expect(done).toEqual(['end']) + // And no longer there as long again after. + await sleep(40) + expect(done).toEqual(['end', 'destroy']) + }) + + test.each(['tls', 'outbound', 'socket'] as const)( + 'idle only a whole time after the last byte queued in %s has gone, and not destroyed before', + async layer => { + const t = tunnel() + const queue = (writableLength: number): void => + void Object.assign(t[layer], { writableLength }) + queue(1) + t.ask()() + await sleep(120) + expect(t.done).toEqual([]) + queue(0) + await sleep(35) + expect(t.done).toEqual([]) + await sleep(65) + expect(t.done[0]).toBe('end') + // The same once it is ended. + const ended = tunnel() + ended.ask()() + await sleep(60) + expect(ended.done).toEqual(['end']) + Object.assign(ended[layer], { writableLength: 24 }) + await sleep(120) + expect(ended.done).toEqual(['end']) + Object.assign(ended[layer], { writableLength: 0 }) + await sleep(100) + expect(ended.done).toEqual(['end', 'destroy']) + }, + ) + + test('with bytes queued when the response ends, idle is counted from soon after the last has gone', async () => { + const t = tunnel({ idleMs: 500 }) + Object.assign(t.outbound, { writableLength: 1 }) + t.ask()() + await sleep(50) + Object.assign(t.outbound, { writableLength: 0 }) + await sleep(450) + expect(t.done).toEqual([]) + // Seen gone within a fifth of the time, so ended by 650 ms. + await sleep(300) + expect(t.done).toEqual(['end']) + }) + + test('bytes queued for a client that takes none: closed once nothing has moved; one that takes some is left alone', async () => { + const bounded = (): ReturnType => tunnel({ stalledMs: 40 }) + const [stuck, slow, fed] = [bounded(), bounded(), bounded()] + for (const t of [stuck, slow, fed]) { + t.ask() + Object.assign(t.outbound, { writableLength: 1_000 }) + } + for (let i = 1; i <= 15; i++) { + await sleep(10) + // The queue shrinks; or stays as long, with as much written as taken. + Object.assign(slow.outbound, { writableLength: 1_000 - i }) + Object.assign(fed.tls, { bytesWritten: i }) + } + expect([stuck.done[0], slow.done, fed.done]).toEqual(['destroy', [], []]) + }) + + test('a stalled tunnel is closed between one and one and a fifth times the time allowed', async () => { + const { done, tls, ask } = tunnel({ paused: true, stalledMs: 1_000 }) + ask() + await sleep(20) + Object.assign(tls, { bytesRead: 8 }) + await sleep(900) + expect(done).toEqual([]) + await sleep(480) + expect(done[0]).toBe('destroy') + }) + + test('a tunnel that has closed is not ended', async () => { + const before = tunnel() + before.socket.emit('close') + const after = tunnel() + const answered = after.ask() + Object.assign(after.socket, { destroyed: true }).emit('close') + answered() + await sleep(100) + expect([before.done, after.done]).toEqual([[], []]) + }) +}) + +describe.each(RUNS.map((r, i) => ({ ...r, i })))( + '$name, connections $how', + ({ i }) => { + let outcome: Record + beforeAll(async () => { + outcome = await outcomeOf(i) + }, 120_000) + + test('idle after a response: ended a margin later than the time it advertised', () => { + const o = outcome['idle after a response']! + expect(o).toMatchObject({ status: 200, advertised: 'timeout=1' }) + expectClosed(o, 1_500) + expect(o.closeNotify).toBe(true) + }) + + test('asked again when the advertised time is up: served', () => { + expect( + outcome['asked again when the advertised time is up']!.status, + ).toBe(200) + }) + + test("ended by its own time alone, not the runtime's", () => { + const o = outcome['a margin longer than the runtime has']! + expectClosed(o, 2_000) + expect(o.closeNotify).toBe(true) + }) + + test('idle, and its client answers no end: closed as long again after the end', () => { + const ms = outcome['idle, and the client answers no end']!.slotFreeAfterMs + expect(ms).toBeNumber() + expect(ms as number).toBeGreaterThanOrEqual(500 - EARLY_MS) + expect(ms as number).toBeLessThan(1_000 + LATE_MS) + }) + + test('the time advertised by default is the default timeout', () => { + expect(outcome['the default Keep-Alive']!.advertised).toBe( + `timeout=${TUNNEL_TIMEOUTS.keepAliveMs / 1000}`, + ) + }) + + test('each response starts the idle time anew', () => { + const o = outcome['each response starts the idle time anew']! + expect(o.statuses).toEqual([200, 200, 200]) + expectClosed(o, 800) + expect(o.closeNotify).toBe(true) + }) + + // Timed from the end of the handshake; the proxy's time runs from its start. + test.each(['no request after the handshake', 'half a head'])('%s', name => { + expectClosed(outcome[name]!, 500, 800) + }) + + test('half a head after a response', () => { + const o = outcome['half a head after a response']! + expectClosed(o, 800) + expect(o.closeNotify).toBe(true) + }) + + test('half a body: closed once it has not moved for the time allowed', () => { + expectClosed(outcome['half a body']!, 500, 1_000) + }) + + test('a body that trickles in arrives', () => { + expect(outcome['a body that trickles in']!.status).toBe(200) + }) + + test('a whole request may wait for its answer longer than every timeout', () => { + expect( + outcome['an answer that takes longer than every timeout']!.status, + ).toBe(200) + }) + + // No timeout plays a part: the proxy reads on, and so sees its client go. + test.each([ + ['answered early', 413], + ['upstream port closed', 502], + ['upstream certificate untrusted', 502], + ['refused for its Host', 421], + ['refused for its target', 400], + ])('an upload, %s, whose client leaves', (what, status) => { + const o = outcome[`${what}, the client leaves`]! + expect(o.status).toBe(status) + expectClosed(o, EARLY_MS) + }) + + test('an upload answered early, whose client stays: idle from the answer on', () => { + const o = outcome['answered early, the client stays']! + // (The client, busy sending, may see the answer late.) + expectClosed(o, EARLY_MS, 600) + expect(o.closeNotify).toBe(true) + }) + + test('an upload answered early, whose client sends it all: the next request is served', () => { + const o = outcome['answered early, the client sends on and asks again']! + expect(o).toMatchObject({ status: 413, second: 200, heldBytes: 0 }) + }) + + test('readers that take nothing for a while, then everything, lose nothing', () => { + expect( + outcome['readers that take nothing for a while, then everything']! + .missing, + ).toEqual([]) + }) + + test('a reader that takes an answer in bursts loses nothing', () => { + expect(outcome['a reader that takes an answer in bursts']!.missing).toBe( + 0, + ) + }) + + test('a client that never reads its answer: closed once nothing has moved for the time allowed', () => { + const o = outcome['a client that never reads its answer']! + expect(o.slotFreeAfterMs).toBeNumber() + expect(o.slotFreeAfterMs as number).toBeGreaterThanOrEqual(500 - EARLY_MS) + expect(o.slotFreeAfterMs as number).toBeLessThan(1_000 + LATE_MS) + expect(o.upstream).toBe(0) + }) + + test('an upload the upstream stops taking: closed once it has not moved for the time allowed', () => { + const o = outcome['an upload the upstream stops taking']! + expectClosed(o, 500, 1_000) + expect(o.closeNotify).toBe(false) + }) + + test('an upload the upstream takes in bursts arrives', () => { + expect(outcome['an upload the upstream takes in bursts']!.status).toBe( + 200, + ) + }) + }, +) diff --git a/test/sandbox/tunnel-cap-refusal.test.ts b/test/sandbox/tunnel-cap-refusal.test.ts new file mode 100644 index 000000000..b9173b1e4 --- /dev/null +++ b/test/sandbox/tunnel-cap-refusal.test.ts @@ -0,0 +1,124 @@ +import { afterEach, expect } from 'bun:test' +import { connect, type AddressInfo, type Socket } from 'node:net' +import { createHttpProxyServer } from '../../src/sandbox/http-proxy.js' +import { createMitmCA, generateCa } from '../../src/sandbox/mitm-ca.js' +import { withCapturedWarnings } from '../helpers/captured-warnings.js' +import { testWithTls as test } from '../helpers/emitted-connections.js' + +// Of the 503 at the tunnel cap a client shows the status at most. That the +// limit is the sandbox's own is logged and passed on as a refusal is, once +// per command and burst of refusals. + +const TOKEN = 'sekrit' +const cleanup: Array<() => void | Promise> = [] +afterEach(async () => { + while (cleanup.length) await cleanup.pop()!() +}) + +type Denial = { + method: string + url: string + reason: string + encodedCommand?: string +} + +async function proxyWithSlots(maxTerminatedTunnels: number) { + const ca = generateCa() + const denials: Denial[] = [] + const proxy = createHttpProxyServer({ + filter: () => true, + mitmCA: createMitmCA({ caCertPem: ca.certPem, caKeyPem: ca.keyPem }), + maxTerminatedTunnels, + proxyAuthToken: TOKEN, + onFilterRequestDenied: info => denials.push(info), + }) + await new Promise(r => proxy.listen(0, '127.0.0.1', r)) + cleanup.push(() => new Promise(r => proxy.close(() => r()))) + const port = (proxy.address() as AddressInfo).port + + /** CONNECT as `command` (none: the bare user name); the status and the socket. */ + const open = (command?: string): Promise<{ status: number; sock: Socket }> => + new Promise(resolve => { + const user = command === undefined ? 'srt' : `srt.${command}` + const basic = Buffer.from(`${user}:${TOKEN}`).toString('base64') + const sock = connect(port, '127.0.0.1', () => + sock.write( + 'CONNECT a.test:443 HTTP/1.1\r\nHost: a.test:443\r\n' + + `Proxy-Authorization: Basic ${basic}\r\n\r\n`, + ), + ) + sock.on('error', () => {}) + cleanup.push(() => void sock.destroy()) + sock.once('data', (d: Buffer) => + resolve({ status: Number(d.toString('latin1').split(' ')[1]), sock }), + ) + }) + const held: Socket[] = [] + return { + denials, + /** Take slots until none is free. */ + async fill() { + while (held.length < maxTerminatedTunnels) { + const { status, sock } = await open('holder') + expect(status).toBe(200) + held.push(sock) + } + }, + async free(slots: number) { + for (const sock of held.splice(0, slots)) sock.destroy() + await new Promise(r => setTimeout(r, 100)) + }, + /** A CONNECT at the cap; who has been told of the cap so far. */ + async refused(command?: string) { + expect((await open(command)).status).toBe(503) + return denials.map(d => d.encodedCommand) + }, + } +} + +test('the first refusal of each command is logged and passed on, naming the limit and its option', async () => { + const p = await proxyWithSlots(2) + await p.fill() + const { warnings } = await withCapturedWarnings(async () => { + expect(await p.refused('A')).toEqual(['A']) + expect(await p.refused('A')).toEqual(['A']) + expect(await p.refused('B')).toEqual(['A', 'B']) + expect(await p.refused()).toEqual(['A', 'B', undefined]) + expect(await p.refused('B')).toEqual(['A', 'B', undefined]) + }) + const reason = + "the sandbox's limit of 2 TLS-terminated tunnels at once is reached; " + + "the limit is network.tlsTerminate.maxTunnels in the sandbox's configuration" + expect(p.denials[0]).toEqual({ + method: 'CONNECT', + url: 'a.test:443', + reason, + encodedCommand: 'A', + }) + expect(warnings).toEqual( + Array(3).fill( + `[SandboxDebug] [proxy] CONNECT a.test:443 refused: ${reason}`, + ), + ) +}) + +test('a burst ends once half the slots are free again, not before', async () => { + const p = await proxyWithSlots(4) + await p.fill() + expect(await p.refused('A')).toEqual(['A']) + await p.free(1) + await p.fill() + expect(await p.refused('A')).toEqual(['A']) + await p.free(2) + await p.fill() + expect(await p.refused('A')).toEqual(['A', 'A']) +}) + +test('the names kept are bounded: past a hundred they are forgotten', async () => { + const p = await proxyWithSlots(1) + await p.fill() + for (let i = 0; i < 100; i++) await p.refused(`c${i}`) + expect(await p.refused('c0')).toHaveLength(100) + expect(await p.refused('one-more')).toHaveLength(101) + expect(await p.refused('c0')).toHaveLength(102) +})