From 515605423f9125b6d348ca0e717343f061ec5e5c Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:43:54 +0000 Subject: [PATCH] refactor(service): Split bodies, clients, run config, recording files, and the JSON scan into files Move code out of the four largest files of the service at their feature lines. The code does not change: a compare of the code lines before and after, without imports, comments, and `export`, differs only in the line breaks of three signatures that oxfmt wrapped. - `server.ts` -> `bodies.ts` (request bodies, their limits, and the shared budget) and `clients.ts` (the PROXY protocol header and the address of a client). - `service.ts` -> `run-config.ts` (the check of a run config, its limits, the tenant name, and the CI attributes). - `recorder.ts` -> `recording-file.ts` (`toRecording`, `fromRecording`, and the decoding of compressed bodies). - `secrets.ts` -> `json-strings.ts` (`scanStrings`, `decodeString`, and `parseJson`). The docs also had drifted from the code: - README said that an idle tunnel closes after 10 minutes. Between two requests, Node closes it after about 5 s (`keepAliveTimeout`), as a probe of a plain and a TLS tunnel shows. In a tunnel without TLS, Node removes the 10-minute limit after the first request. The README and the comment of `TUNNEL_IDLE_MS` now say so. - TELEMETRY.md: `roach.mode` can be `off`, and `roach.reason` is on the `roach.request` metric too. - README "Files" lists the new files. Co-Authored-By: David Cramer <23610+dcramer@users.noreply.github.com> --- README.md | 15 +- TELEMETRY.md | 4 +- packages/service/src/bodies.ts | 169 ++++++++++++++ packages/service/src/clients.ts | 127 +++++++++++ packages/service/src/json-strings.ts | 185 ++++++++++++++++ packages/service/src/recorder.ts | 143 +----------- packages/service/src/recording-file.ts | 151 +++++++++++++ packages/service/src/run-config.ts | 160 ++++++++++++++ packages/service/src/secrets.ts | 180 +-------------- packages/service/src/server.ts | 291 +------------------------ packages/service/src/service.ts | 155 +------------ packages/service/tests/service.test.ts | 2 +- 12 files changed, 843 insertions(+), 739 deletions(-) create mode 100644 packages/service/src/bodies.ts create mode 100644 packages/service/src/clients.ts create mode 100644 packages/service/src/json-strings.ts create mode 100644 packages/service/src/recording-file.ts create mode 100644 packages/service/src/run-config.ts diff --git a/README.md b/README.md index ba6aafc..ad4afc3 100644 --- a/README.md +++ b/README.md @@ -761,7 +761,11 @@ To see how well the cache works: whose body does not come in 300 s too. This is also true in a tunnel and with `proxyProtocol`. - A `CONNECT` tunnel that sends no byte for 10 s after the `200` closes. A - tunnel that sends and gets no byte for 10 minutes closes. + tunnel that sends and gets no byte for 10 minutes while a request waits + for its response closes. Between two requests, a tunnel closes after + about 5 s without a new request (the `keepAliveTimeout` of Node), and + the client opens a new one. A gap: in a tunnel without TLS, Node removes + the 10-minute limit after the first request. - The credentials that a run learns and the values that the sessions of a run without the token remember have limits. See [Config](#config) and [Changing values](#changing-values). @@ -1040,9 +1044,16 @@ In `packages/service/src/`: - `cli.ts`: the command line, which the image runs. - `server.ts`: sockets, HTTPS interception, the allow list, and the control API. +- `bodies.ts`: request bodies, their size limits, and the budget that the + bodies of runs without the token share. +- `clients.ts`: the PROXY protocol header and the address of a client. +- `run-config.ts`: the check of the run config of a client, with its + limits. - `errors.ts`: the log line and the client text of an unexpected error. - `addresses.ts`: the public addresses that the proxy can send requests to. - `recorder.ts`: modes, sessions, replay, and misses. +- `recording-file.ts`: a recording file from a live response, the response + of a replay, and the decoding of compressed bodies. - `request-key.ts`: the key of a request. - `key-worker.ts`: the worker process that makes the keys of runs without the token and redacts the text of their misses, with a time limit and a @@ -1055,4 +1066,6 @@ In `packages/service/src/`: - `streams.ts`: merges the deltas of a recorded model stream. - `usage.ts`: the token usage of a model response, for Sentry. - `secrets.ts`: redaction of credentials. +- `json-strings.ts`: the one-pass scan of the JSON strings of a body, for + `secrets.ts`. - `certificates.ts`: the certificate authority for HTTPS. diff --git a/TELEMETRY.md b/TELEMETRY.md index 35503a8..88f0078 100644 --- a/TELEMETRY.md +++ b/TELEMETRY.md @@ -66,11 +66,11 @@ For metrics, use `-f dataset=tracemetrics`, | Pivot | Meaning | Found In | | --------------------------- | ------------------------------------------ | --------------------------------- | | `roach.tenant` | repository of the run | metrics, spans, errors | -| `roach.mode` | `auto`, `replay`, `record` | metrics, spans, errors | +| `roach.mode` | `auto`, `replay`, `record`, `off` | metrics, spans, errors | | `roach.rule` | recording rule, such as `model` | metrics, spans | | `roach.result` | `replayed`, `live`, and others (see Words) | metrics, spans | | `roach.origin` | upstream origin | metrics, spans | -| `roach.reason` | why the proxy refused a request | `roach.request` | +| `roach.reason` | why the proxy refused a request | `roach.request` metric and span | | `roach.run.id` | `name` of the run, else the service id | spans, errors | | `roach.ci.run_id` | `GITHUB_RUN_ID` of the run | spans, errors | | `roach.ci.job` | `GITHUB_JOB` of the run | spans, errors | diff --git a/packages/service/src/bodies.ts b/packages/service/src/bodies.ts new file mode 100644 index 0000000..0c0f0e0 --- /dev/null +++ b/packages/service/src/bodies.ts @@ -0,0 +1,169 @@ +/** + * Request bodies of the proxy and the control API: their size limits, the + * budget of bytes that the bodies of many runs share (`BodyBudget`), and + * the time that a body has to come. + */ +import type http from "node:http"; + +/** + * A control API body over this size gets HTTP 413. Anyone can call + * `POST /runs`, so the service reads no more before it checks the token. + */ +const MAX_CONTROL_BODY_BYTES = 1024 * 1024; + +/** An error that answers a request with its own status. */ +const httpError = (status: number, message: string) => + Object.assign(new Error(message), { status }); + +/** + * The bytes of request bodies that the requests of many targets can hold + * at one time (`ProxyTarget.bodyBudget`). `used` counts the bodies of the + * requests that did not finish, and `clients` counts them for each client + * (`ProxyTarget.client`). One client can hold at most `perClient` bytes, + * so it cannot take all of `max`. A body that takes longer than `readMs` + * to come fails with HTTP 408 and gives its bytes back. + */ +export interface BodyBudget { + max: number; + perClient: number; + readMs: number; + used: number; + clients: Map; +} + +/** The bytes that one request holds of its `BodyBudget`. */ +export interface Held { + budget: BodyBudget | undefined; + client: string; + bytes: number; +} + +/** Hold `bytes` more of the budget. Returns `false` when they do not fit. */ +function hold(held: Held, bytes: number): boolean { + const { budget, client } = held; + if (!budget) return true; + const ofClient = budget.clients.get(client) ?? 0; + if (budget.used + bytes > budget.max || ofClient + bytes > budget.perClient) { + return false; + } + budget.used += bytes; + budget.clients.set(client, ofClient + bytes); + held.bytes += bytes; + return true; +} + +/** Give back all bytes that a request holds. */ +export function release(held: Held): void { + const { budget, client, bytes } = held; + held.bytes = 0; + if (!budget || bytes === 0) return; + budget.used -= bytes; + const left = (budget.clients.get(client) ?? 0) - bytes; + if (left > 0) budget.clients.set(client, left); + else budget.clients.delete(client); +} + +const busyError = () => + httpError( + 503, + "the service has too many large requests at this time. Send the request again later.", + ); + +/** + * Read a whole body. A body over `maxBytes` fails with HTTP 413. A body + * that does not fit in the budget of `held` fails with HTTP 503, and one + * that does not come in the `readMs` of the budget fails with HTTP 408. + * Only request bodies have a limit: the client sends them, and a large + * upstream response is not the fault of the client. + */ +export function readBody( + stream: http.IncomingMessage, + maxBytes = Infinity, + held: Held = { budget: undefined, client: "", bytes: 0 }, +): Promise { + const declared = Number(stream.headers["content-length"]); + if (declared > maxBytes) { + // Drop the body without a look at it, so it costs no memory. + stream.resume(); + return Promise.reject(httpError(413, `body is over ${maxBytes} bytes`)); + } + // Hold the bytes when they come, not the declared length. So a client + // that declares a large body and then sends nothing holds nothing. + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + let size = 0; + let failed = false; + const fail = (error: Error) => { + // Read the rest without keeping it, so the client gets the error. + failed = true; + chunks.length = 0; + clearTimeout(deadline); + reject(error); + }; + // Node gives a whole request 300 s (`HTTP_OPTIONS`). That is too long + // to hold a part of the budget, so these bodies have a shorter limit. + const readMs = held.budget?.readMs; + const deadline = + readMs === undefined + ? undefined + : setTimeout( + () => + fail( + httpError(408, `the body did not come in ${readMs / 1000} s`), + ), + readMs, + ); + stream.on("data", (chunk: Buffer) => { + if (failed) return; + size += chunk.length; + if (size > maxBytes) { + fail(httpError(413, `body is over ${maxBytes} bytes`)); + } else if (!hold(held, chunk.length)) { + fail(busyError()); + } else { + chunks.push(chunk); + } + }); + stream.on("end", () => { + clearTimeout(deadline); + resolve(Buffer.concat(chunks)); + }); + stream.on("error", (error: Error & { code?: string }) => { + clearTimeout(deadline); + // A client that closes its connection before its body came is not a + // failure of the proxy, so it does not go to Sentry. `method` is set + // only on a request, not on an upstream response. + const aborted = stream.method !== null && error.code === "ECONNRESET"; + reject(aborted ? abortedError() : error); + }); + }); +} + +/** The error of a request whose client closed it before its body came. */ +const abortedError = () => + Object.assign( + httpError(400, "the client closed the request before its body came"), + { aborted: true }, + ); + +/** + * Read a JSON object body of a control API call. A body over + * `MAX_CONTROL_BODY_BYTES` fails with HTTP 413 before it is parsed. Any + * other body that is not a JSON object fails with HTTP 400. + */ +export async function readJson(incoming: http.IncomingMessage): Promise { + const text = (await readBody(incoming, MAX_CONTROL_BODY_BYTES)).toString( + "utf8", + ); + let value: unknown; + try { + value = JSON.parse(text || "{}"); + } catch { + throw httpError(400, "body is not JSON"); + } + // Callers read fields, so `null`, a list, or a single value is a bad body. + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw httpError(400, "body is not a JSON object"); + } + return value as T; +} diff --git a/packages/service/src/clients.ts b/packages/service/src/clients.ts new file mode 100644 index 0000000..e4e1bda --- /dev/null +++ b/packages/service/src/clients.ts @@ -0,0 +1,127 @@ +/** + * The address of the client of a proxy connection, for the limit of runs + * for each client (`service.ts`). + * + * Behind a load balancer, a connection can start with a PROXY protocol v1 + * header (`proxyProtocol`). It gives the address of the client + * (`clientAddress()`). An IPv6 client is grouped by its /64 + * (`clientGroup()`). + */ +import type http from "node:http"; +import { isIP, type Socket } from "node:net"; + +/** + * A PROXY protocol v1 header, such as + * `PROXY TCP4 203.0.113.7 10.0.0.2 51234 443`. It is at most 107 bytes. + * The first bytes of a connection must come in `PROXY_HEADER_MS`. + */ +const PROXY_PREFIX = Buffer.from("PROXY "); + +const PROXY_HEADER = + /^PROXY (?:TCP[46] ([0-9A-Fa-f:.]+) [0-9A-Fa-f:.]+ \d{1,5} \d{1,5}|UNKNOWN[^\r\n]*)\r\n/; + +const MAX_PROXY_HEADER_BYTES = 107; + +const PROXY_HEADER_MS = 10_000; + +/** The address of the client of each connection with a PROXY header. */ +const proxiedClients = new WeakMap(); + +/** + * The address of the client of a request. With `proxyProtocol`, it comes + * from the PROXY header of the connection, when the connection has one. + * Else it is the address of the socket. + */ +export function clientAddress(incoming: http.IncomingMessage): string { + return ( + proxiedClients.get(incoming.socket) ?? incoming.socket.remoteAddress ?? "" + ); +} + +/** + * The group of a client address for the limit of runs for each client. An + * IPv6 client often has a whole /64, so the group of an IPv6 address is + * its /64, such as `2001:db8:0:1::/64`. An IPv4 address is its own group, + * also in each IPv6 form, such as `::ffff:192.0.2.1`, `::ffff:c000:201`, + * and `0:0:0:0:0:ffff:192.0.2.1`. + */ +export function clientGroup(address: string): string { + if (isIP(address) !== 6) return address; + const groups = ipv6Groups(address); + if ( + groups.slice(0, 5).every((group) => group === 0) && + groups[5] === 0xffff + ) { + const [high = 0, low = 0] = groups.slice(6); + return [high >> 8, high & 0xff, low >> 8, low & 0xff].join("."); + } + return `${groups + .slice(0, 4) + .map((group) => group.toString(16)) + .join(":")}::/64`; +} + +/** The eight 16-bit groups of an IPv6 address. */ +function ipv6Groups(address: string): number[] { + // The URL parser writes each IPv6 form with hex groups only, such as + // `::ffff:c000:201`. A zone, such as `%eth0`, is not part of it. + const [head = "", tail] = new URL( + `http://[${address.split("%")[0]}]`, + ).hostname + .slice(1, -1) + .split("::"); + const front = head ? head.split(":") : []; + const back = tail ? tail.split(":") : []; + const zeros = Array(8 - front.length - back.length).fill("0"); + return [...front, ...zeros, ...back].map((group) => + Number.parseInt(group, 16), + ); +} + +/** + * Read the PROXY protocol v1 header of a new connection, when it has one, + * and keep the address of its client. Then give the connection to + * `server`. A connection without the header keeps the address of its + * socket. A header that is not valid closes the connection. + * + * The load balancer writes its header first, so a client cannot put a + * header before it. Without a load balancer, a client can send its own + * header. That changes only the address for the limit of runs for each + * client. + */ +export function acceptProxied(socket: Socket, server: http.Server): void { + let read = Buffer.alloc(0); + const timer = setTimeout(() => socket.destroy(), PROXY_HEADER_MS); + const accept = (rest: Buffer) => { + socket.off("data", onData); + clearTimeout(timer); + socket.pause(); + if (rest.length > 0) socket.unshift(rest); + server.emit("connection", socket); + socket.resume(); + }; + const onData = (chunk: Buffer) => { + read = Buffer.concat([read, chunk]); + const prefix = Math.min(read.length, PROXY_PREFIX.length); + if (!read.subarray(0, prefix).equals(PROXY_PREFIX.subarray(0, prefix))) { + accept(read); + return; + } + const end = read.indexOf("\r\n"); + if (end < 0 || end + 2 > MAX_PROXY_HEADER_BYTES) { + if (end >= 0 || read.length >= MAX_PROXY_HEADER_BYTES) socket.destroy(); + return; + } + const match = PROXY_HEADER.exec(read.subarray(0, end + 2).toString()); + if (!match) { + socket.destroy(); + return; + } + // `UNKNOWN`, such as from a health check, gives no address. + if (match[1] && isIP(match[1])) proxiedClients.set(socket, match[1]); + accept(read.subarray(end + 2)); + }; + socket.on("error", () => socket.destroy()); + socket.on("close", () => clearTimeout(timer)); + socket.on("data", onData); +} diff --git a/packages/service/src/json-strings.ts b/packages/service/src/json-strings.ts new file mode 100644 index 0000000..c7b97bd --- /dev/null +++ b/packages/service/src/json-strings.ts @@ -0,0 +1,185 @@ +/** + * The JSON reads of the redaction of credentials (`secrets.ts`). + * + * `scanStrings()` reads JSON text in one pass and gives each string with + * its depth, whether it is a field name, and its escapes. `decodeString()` + * decodes one string. `parseJson()` parses a whole body. + */ + +/** + * The parsed value of a JSON text, or `undefined` when it is not JSON. A + * byte order mark at the start does not count. + */ +export const parseJson = (text: string): unknown => { + try { + return JSON.parse(text.startsWith("\uFEFF") ? text.slice(1) : text); + } catch { + return undefined; + } +}; + +const QUOTE = 0x22; + +const BACKSLASH = 0x5c; + +const LINE_BREAK = 0x0a; + +const COLON = 0x3a; + +/** `{` or `[`. A compare is faster than a set for each character. */ +export const isOpen = (code: number) => code === 0x7b || code === 0x5b; + +/** `}` or `]`. */ +const isClose = (code: number) => code === 0x7d || code === 0x5d; + +/** Space, tab, line feed, or carriage return. */ +export const isJsonSpace = (code: number) => + code === 0x20 || code === 0x09 || code === 0x0a || code === 0x0d; + +/** + * The escapes of a JSON string: it has none, all are valid, or one is not + * valid (or it has a control character), so `JSON.parse` would throw. + */ +export type Escapes = "none" | "valid" | "bad"; + +/** + * The escapes that are not `u` escapes, by the character after the + * backslash, and the character that each one gives. + */ +const SIMPLE_ESCAPES = new Map([ + [0x22, '"'], + [0x5c, "\\"], + [0x2f, "/"], + [0x62, "\b"], + [0x66, "\f"], + [0x6e, "\n"], + [0x72, "\r"], + [0x74, "\t"], +]); + +const U = 0x75; + +/** Tells whether the character at `at` of `text` is a hex digit. */ +function isHex(text: string, at: number): boolean { + const code = text.charCodeAt(at); + return ( + (code >= 0x30 && code <= 0x39) || + (code >= 0x41 && code <= 0x46) || + (code >= 0x61 && code <= 0x66) + ); +} + +/** Tells whether `text` has a valid `u` escape at `at` (its backslash). */ +const uEscape = (text: string, at: number) => + text.charCodeAt(at + 1) === U && + isHex(text, at + 2) && + isHex(text, at + 3) && + isHex(text, at + 4) && + isHex(text, at + 5); + +/** + * Call `visit` for each JSON string of `text`, in one pass: its start, its + * end after the closing quote, the depth of objects and arrays around it, + * whether it is a field name (a `:` follows it), its escapes (`Escapes`), + * and the length of its value. The length is right only when the escapes + * are valid. A string ends at a line break, because JSON does not allow + * one in a string. So a line that is not JSON, such as a header line in + * the text of a key, cannot make it read the next lines wrong. + */ +export function scanStrings( + text: string, + visit: ( + start: number, + end: number, + depth: number, + name: boolean, + escapes: Escapes, + length: number, + ) => void, +): void { + let depth = 0; + let at = 0; + while (at < text.length) { + const code = text.charCodeAt(at); + if (isOpen(code)) depth += 1; + else if (isClose(code)) depth -= 1; + if (code !== QUOTE) { + at += 1; + continue; + } + let end = at + 1; + let escapes: Escapes = "none"; + let length = 0; + while (end < text.length) { + const next = text.charCodeAt(end); + if (next === QUOTE) break; + if (next === LINE_BREAK) break; + length += 1; + if (next === BACKSLASH && text.charCodeAt(end + 1) !== LINE_BREAK) { + if (escapes === "none") escapes = "valid"; + // The hex digits of a `u` escape cannot be a quote or a line break. + if (uEscape(text, end)) end += 6; + else { + if (!SIMPLE_ESCAPES.has(text.charCodeAt(end + 1))) escapes = "bad"; + end += 2; + } + } else { + // JSON does not allow a control character, such as a tab, in a + // string. A header line can have one. + if (next < 0x20) escapes = "bad"; + end += 1; + } + } + if (text.charCodeAt(end) !== QUOTE) { + // Not a string. Each quote before `end` has a backslash before it, so + // no string can start there either. Go on at `end`, so each character + // is read once. + at = end; + continue; + } + end += 1; + let after = end; + while (isJsonSpace(text.charCodeAt(after))) after += 1; + visit(at, end, depth, text.charCodeAt(after) === COLON, escapes, length); + at = end; + } +} + +/** + * The value of the JSON string from `start` to `end` of `text`, or + * `undefined` when its escapes are not valid (`scanStrings`). It decodes + * the escapes itself, so it does not throw and it costs little for each + * string. A `u` escape gives one UTF-16 code unit, as in `JSON.parse`, so a + * lone surrogate stays. + */ +export function decodeString( + text: string, + start: number, + end: number, + escapes: Escapes, +): string | undefined { + if (escapes === "none") return text.slice(start + 1, end - 1); + if (escapes === "bad") return undefined; + let value = ""; + let from = start + 1; + let at = from; + while (at < end - 1) { + if (text.charCodeAt(at) !== BACKSLASH) { + at += 1; + continue; + } + value += text.slice(from, at); + const next = text.charCodeAt(at + 1); + if (next === U) { + value += String.fromCharCode( + Number.parseInt(text.slice(at + 2, at + 6), 16), + ); + at += 6; + } else { + value += SIMPLE_ESCAPES.get(next)!; + at += 2; + } + from = at; + } + return value + text.slice(from, end - 1); +} diff --git a/packages/service/src/recorder.ts b/packages/service/src/recorder.ts index 04216cb..76e7bb6 100644 --- a/packages/service/src/recorder.ts +++ b/packages/service/src/recorder.ts @@ -32,8 +32,6 @@ */ import { randomBytes } from "node:crypto"; import type { IncomingHttpHeaders } from "node:http"; -import { promisify } from "node:util"; -import { brotliDecompress, gunzip, inflate, zstdDecompress } from "node:zlib"; import { reportInternal } from "./errors.ts"; import { firstChange, sampleKey, toSample, withheldSample } from "./misses.ts"; import { @@ -42,13 +40,15 @@ import { type JobError, type JobOptions, } from "./key-worker.ts"; -import { keyRequest, keyTimeLimit, type KeyedRequest } from "./request-key.ts"; +import { keyRequest, keyTimeLimit } from "./request-key.ts"; import { - createSecrets, - redactRequestUrl, - toRedact, - type LearnLimit, -} from "./secrets.ts"; + decodeBody, + fromRecording, + RECORDED_HEADERS, + toRecording, + utf8Text, +} from "./recording-file.ts"; +import { createSecrets, toRedact, type LearnLimit } from "./secrets.ts"; import { recordingKey, RULE_NAME, @@ -56,11 +56,6 @@ import { type RecordingStore, type RequestSample, } from "./store.ts"; -import { - isEventStream, - mapStreamEvents, - mergeStreamDeltas, -} from "./streams.ts"; import { usageAttributes } from "./usage.ts"; import type { RecordingMiss, @@ -71,8 +66,6 @@ import type { } from "@sentry/roach/types"; import { canBeKnown, - fillValues, - templateValues, withTimeLimit, type RequestValues, } from "@sentry/roach/values"; @@ -175,66 +168,6 @@ const METRIC_OTHER_ORIGIN = "other"; const MAX_ENDED_SESSIONS = 1000; /** Open sessions of a run, such as one for each test worker. */ const MAX_OPEN_SESSIONS = 64; -/** Content types that a recording keeps as text. Other bodies are base64. */ -const TEXT_CONTENT_TYPE = /^text\/|json|xml|x-www-form-urlencoded/i; -/** Response headers that a recording keeps. Other headers change each run. */ -const RECORDED_HEADERS = ["content-type", "location"]; -const NULL_BODY_STATUSES = new Set([101, 204, 205, 304]); -/** - * The largest body that the recorder decodes from a content encoding. It - * is the largest request body that the proxy reads (`server.ts`), so a - * small compressed body cannot fill the memory of the service. - */ -const MAX_DECODED_BYTES = 64 * 1024 * 1024; -/** - * Decoders of the content encodings that the recorder can read. They run - * on the thread pool of zlib, so they do not block other requests. - */ -const DECODERS: Record< - string, - (body: Buffer, options: { maxOutputLength: number }) => Promise -> = { - gzip: promisify(gunzip), - "x-gzip": promisify(gunzip), - deflate: promisify(inflate), - br: promisify(brotliDecompress), - zstd: promisify(zstdDecompress), -}; - -/** - * The body of a response without its content encoding. The proxy asks for - * `identity`, but an upstream can encode the body all the same, and the - * recorder must decode it to find credentials in it. An empty body, such - * as of a 304, stays empty. Rejects when it does not know an encoding, - * when the body is not valid, or when the decoded body is larger than - * `MAX_DECODED_BYTES`. - */ -async function decodeBody({ headers, body }: ProxyResponse): Promise { - if (body.length === 0) return body; - const encodings = (headers["content-encoding"] ?? "") - .split(",") - .map((encoding) => encoding.trim().toLowerCase()) - .filter((encoding) => encoding !== "" && encoding !== "identity"); - let result = body; - for (const encoding of encodings.toReversed()) { - const decoder = DECODERS[encoding]; - if (!decoder) throw new Error(`unknown content encoding ${encoding}`); - result = await decoder(result, { maxOutputLength: MAX_DECODED_BYTES }); - } - return result; -} - -/** The text of `body` when it is valid UTF-8, or `undefined`. */ -function utf8Text(body: Buffer): string | undefined { - try { - return new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode( - body, - ); - } catch { - return undefined; - } -} - /** Statuses that are never recorded, because they are temporary. */ const isTemporaryStatus = (status: number) => status === 429 || status >= 500; @@ -252,64 +185,6 @@ function matches( ); } -/** - * The response as a recording keeps it. A stream gets one delta per content - * block, and each value of the request becomes a placeholder. A body that - * is not text gets `redactBytes` before it is encoded, because a redaction - * of the encoded text cannot find a credential in it. The URL has the - * value of each credential query parameter redacted, and of each - * parameter that holds the value of a credential field of the body - * (`redactRequestUrl`). - */ -function toRecording( - request: ProxyRequest, - rule: RecordingRule, - keyed: KeyedRequest, - session: string | undefined, - response: ProxyResponse, - redactBytes: (body: Buffer) => Buffer, -): Recording { - const contentType = response.headers["content-type"]; - const headers: Record = {}; - for (const name of RECORDED_HEADERS) { - const value = response.headers[name]; - if (typeof value === "string") headers[name] = value; - } - let body = ""; - let bodyEncoding: Recording["response"]["bodyEncoding"] = "base64"; - if (!TEXT_CONTENT_TYPE.test(contentType ?? "")) { - body = redactBytes(response.body).toString("base64"); - } else { - const text = response.body.toString("utf8"); - const template = (part: string) => templateValues(part, keyed.values); - body = isEventStream(contentType) - ? mapStreamEvents(mergeStreamDeltas(text), template) - : template(text); - bodyEncoding = "utf8"; - } - return { - session, - request: { - method: request.method, - url: redactRequestUrl(request, rule), - }, - response: { status: response.status, headers, body, bodyEncoding }, - }; -} - -/** The response of a recording, with the values of the current request. */ -function fromRecording( - { response }: Recording, - values: RequestValues, -): ProxyResponse { - const body = NULL_BODY_STATUSES.has(response.status) - ? Buffer.alloc(0) - : response.bodyEncoding === "utf8" - ? Buffer.from(fillValues(response.body, values)) - : Buffer.from(response.body, "base64"); - return { status: response.status, headers: response.headers, body }; -} - /** The config of a recorder. */ export interface RecorderConfig { mode: RecordingMode; @@ -829,7 +704,7 @@ export function createRecorder(config: RecorderConfig, store: RecordingStore) { /** * Do not record the response, because the recorder cannot redact it: * it cannot read the body, or the body decodes to more than - * `MAX_DECODED_BYTES`, or a minted credential stays in it. The client + * `MAX_DECODED_BYTES` of `recording-file.ts`, or a minted credential stays in it. The client * gets it as it came. */ const notRecorded = (why: string, error: unknown) => { diff --git a/packages/service/src/recording-file.ts b/packages/service/src/recording-file.ts new file mode 100644 index 0000000..3cb34db --- /dev/null +++ b/packages/service/src/recording-file.ts @@ -0,0 +1,151 @@ +/** + * The conversion between a live response and the recording file that the + * recorder writes (`recorder.ts`). + * + * `toRecording()` makes a recording from a live response: a text body + * when the response is text, and else base64. `fromRecording()` makes the + * response that a replay sends. `decodeBody()` reads a compressed body, + * with a limit on its decoded size. + */ +import { promisify } from "node:util"; +import { brotliDecompress, gunzip, inflate, zstdDecompress } from "node:zlib"; +import type { ProxyRequest, ProxyResponse } from "./recorder.ts"; +import type { KeyedRequest } from "./request-key.ts"; +import { redactRequestUrl } from "./secrets.ts"; +import type { Recording } from "./store.ts"; +import { + isEventStream, + mapStreamEvents, + mergeStreamDeltas, +} from "./streams.ts"; +import type { RecordingRule } from "@sentry/roach/types"; +import { + fillValues, + templateValues, + type RequestValues, +} from "@sentry/roach/values"; + +/** Content types that a recording keeps as text. Other bodies are base64. */ +const TEXT_CONTENT_TYPE = /^text\/|json|xml|x-www-form-urlencoded/i; + +/** Response headers that a recording keeps. Other headers change each run. */ +export const RECORDED_HEADERS = ["content-type", "location"]; + +const NULL_BODY_STATUSES = new Set([101, 204, 205, 304]); + +/** + * The largest body that the recorder decodes from a content encoding. It + * is the largest request body that the proxy reads (`server.ts`), so a + * small compressed body cannot fill the memory of the service. + */ +const MAX_DECODED_BYTES = 64 * 1024 * 1024; + +/** + * Decoders of the content encodings that the recorder can read. They run + * on the thread pool of zlib, so they do not block other requests. + */ +const DECODERS: Record< + string, + (body: Buffer, options: { maxOutputLength: number }) => Promise +> = { + gzip: promisify(gunzip), + "x-gzip": promisify(gunzip), + deflate: promisify(inflate), + br: promisify(brotliDecompress), + zstd: promisify(zstdDecompress), +}; + +/** + * The body of a response without its content encoding. The proxy asks for + * `identity`, but an upstream can encode the body all the same, and the + * recorder must decode it to find credentials in it. An empty body, such + * as of a 304, stays empty. Rejects when it does not know an encoding, + * when the body is not valid, or when the decoded body is larger than + * `MAX_DECODED_BYTES`. + */ +export async function decodeBody({ + headers, + body, +}: ProxyResponse): Promise { + if (body.length === 0) return body; + const encodings = (headers["content-encoding"] ?? "") + .split(",") + .map((encoding) => encoding.trim().toLowerCase()) + .filter((encoding) => encoding !== "" && encoding !== "identity"); + let result = body; + for (const encoding of encodings.toReversed()) { + const decoder = DECODERS[encoding]; + if (!decoder) throw new Error(`unknown content encoding ${encoding}`); + result = await decoder(result, { maxOutputLength: MAX_DECODED_BYTES }); + } + return result; +} + +/** The text of `body` when it is valid UTF-8, or `undefined`. */ +export function utf8Text(body: Buffer): string | undefined { + try { + return new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode( + body, + ); + } catch { + return undefined; + } +} + +/** + * The response as a recording keeps it. A stream gets one delta per content + * block, and each value of the request becomes a placeholder. A body that + * is not text gets `redactBytes` before it is encoded, because a redaction + * of the encoded text cannot find a credential in it. The URL has the + * value of each credential query parameter redacted, and of each + * parameter that holds the value of a credential field of the body + * (`redactRequestUrl`). + */ +export function toRecording( + request: ProxyRequest, + rule: RecordingRule, + keyed: KeyedRequest, + session: string | undefined, + response: ProxyResponse, + redactBytes: (body: Buffer) => Buffer, +): Recording { + const contentType = response.headers["content-type"]; + const headers: Record = {}; + for (const name of RECORDED_HEADERS) { + const value = response.headers[name]; + if (typeof value === "string") headers[name] = value; + } + let body = ""; + let bodyEncoding: Recording["response"]["bodyEncoding"] = "base64"; + if (!TEXT_CONTENT_TYPE.test(contentType ?? "")) { + body = redactBytes(response.body).toString("base64"); + } else { + const text = response.body.toString("utf8"); + const template = (part: string) => templateValues(part, keyed.values); + body = isEventStream(contentType) + ? mapStreamEvents(mergeStreamDeltas(text), template) + : template(text); + bodyEncoding = "utf8"; + } + return { + session, + request: { + method: request.method, + url: redactRequestUrl(request, rule), + }, + response: { status: response.status, headers, body, bodyEncoding }, + }; +} + +/** The response of a recording, with the values of the current request. */ +export function fromRecording( + { response }: Recording, + values: RequestValues, +): ProxyResponse { + const body = NULL_BODY_STATUSES.has(response.status) + ? Buffer.alloc(0) + : response.bodyEncoding === "utf8" + ? Buffer.from(fillValues(response.body, values)) + : Buffer.from(response.body, "base64"); + return { status: response.status, headers: response.headers, body }; +} diff --git a/packages/service/src/run-config.ts b/packages/service/src/run-config.ts new file mode 100644 index 0000000..304626b --- /dev/null +++ b/packages/service/src/run-config.ts @@ -0,0 +1,160 @@ +/** + * The check of the run config that a client sends to the service + * (`service.ts`), with its limits, the tenant name, and the span + * attributes of its CI fields. + */ +import { parseOrigins } from "./server.ts"; +import { RULE_NAME } from "./store.ts"; +import type { RecordingMode, RunCi, RunConfig } from "@sentry/roach/types"; +import { invalidValuePatterns } from "@sentry/roach/values"; + +const MODES = new Set(["auto", "off", "record", "replay"]); + +const MAX_RUN_NAME = 128; + +/** + * Limits of the config of a run. Anyone can start a run, so they keep the + * work of a run small: the service makes a certificate for each host that + * a run tunnels to, and runs each value pattern on each request. + */ +const MAX_RUN_ORIGINS = 32; +const MAX_RULES = 64; +const MAX_RULE_VALUES = 32; +const MAX_RULE_URLS = 32; +const MAX_PATTERN_LENGTH = 2048; + +/** The span attribute of each field of `RunConfig.ci`. */ +const CI_ATTRIBUTES: Record = { + runId: "roach.ci.run_id", + job: "roach.ci.job", + repo: "roach.ci.repo", + ref: "roach.ci.ref", +}; +const MAX_CI_VALUE = 256; + +/** Enough for the API keys and OIDC tokens of a CI job. */ +const MAX_RUN_SECRETS = 32; +const MAX_SECRET_LENGTH = 16 * 1024; + +/** + * A tenant name: `owner/repo`, with the characters of GitHub names. It is a + * path in the store, so each part starts with a letter, digit, or `_`, and + * there are always two parts. Then no tenant is in the directory of another. + */ +export const TENANT_NAME = + /^[A-Za-z0-9_][\w.-]{0,99}\/[A-Za-z0-9_][\w.-]{0,99}$/; + +/** Check the run config of a client. Returns the error, or `undefined`. */ +export function invalidRunConfig( + config: Partial, +): string | undefined { + if (!config.mode || !MODES.has(config.mode)) { + return "mode must be auto, off, record, or replay"; + } + if (config.name !== undefined) { + if (typeof config.name !== "string" || config.name.length > MAX_RUN_NAME) { + return `name must be a string of at most ${MAX_RUN_NAME} characters`; + } + } + if (config.secrets !== undefined) { + if ( + !Array.isArray(config.secrets) || + config.secrets.length > MAX_RUN_SECRETS || + !config.secrets.every( + (secret) => + typeof secret === "string" && secret.length <= MAX_SECRET_LENGTH, + ) + ) { + return `secrets must be a list of at most ${MAX_RUN_SECRETS} strings of at most ${MAX_SECRET_LENGTH} characters`; + } + } + if (config.ci !== undefined) { + // A newer client can send more fields. The service ignores them. + const ci = config.ci as Record | null; + if ( + typeof ci !== "object" || + ci === null || + !Object.keys(CI_ATTRIBUTES).every( + (field) => + ci[field] === undefined || + (typeof ci[field] === "string" && ci[field].length <= MAX_CI_VALUE), + ) + ) { + return `ci must be an object of strings of at most ${MAX_CI_VALUE} characters`; + } + } + if (!Array.isArray(config.rules) || config.rules.length > MAX_RULES) { + return `rules must be a list of at most ${MAX_RULES} rules`; + } + const names = new Set(); + for (const rule of config.rules) { + if (typeof rule?.name !== "string" || !RULE_NAME.test(rule.name)) { + return `rule name must match ${RULE_NAME}`; + } + if (names.has(rule.name)) return `rule ${rule.name} is not unique`; + names.add(rule.name); + if (typeof rule.match !== "object" || rule.match === null) { + return `rule ${rule.name} must have match`; + } + const { url } = rule.match; + if ( + url !== undefined && + typeof url !== "string" && + !( + Array.isArray(url) && + url.length > 0 && + url.length <= MAX_RULE_URLS && + url.every((prefix) => typeof prefix === "string") + ) + ) { + return `match.url of rule ${rule.name} must be a string or a list of 1 to ${MAX_RULE_URLS} strings`; + } + // A slow pattern fails only its own requests (`request-key.ts`), so + // the service takes any pattern that compiles. + if (rule.values !== undefined) { + if (typeof rule.values !== "object" || rule.values === null) { + return `values of rule ${rule.name} must be an object`; + } + const values = Object.entries(rule.values); + if (values.length > MAX_RULE_VALUES) { + return `rule ${rule.name} can have at most ${MAX_RULE_VALUES} values`; + } + for (const [name, source] of values) { + if (typeof source !== "string" || source.length > MAX_PATTERN_LENGTH) { + return `value ${name} of rule ${rule.name} must be a pattern of at most ${MAX_PATTERN_LENGTH} characters`; + } + } + } + } + const patternError = invalidValuePatterns( + config.rules.flatMap((rule) => (rule.values ? [rule.values] : [])), + ); + if (patternError) return patternError; + if ( + !Array.isArray(config.allow) || + config.allow.length === 0 || + config.allow.length > MAX_RUN_ORIGINS + ) { + return `allow must be a list of 1 to ${MAX_RUN_ORIGINS} origins`; + } + for (const origin of config.allow) { + // The client sends JSON, so check the type at run time. + if (typeof origin !== "string") return "allow must be a list of origins"; + try { + parseOrigins([origin.replace(/\/$/, "")]); + } catch { + return `${origin} is not an http or https origin`; + } + } + return undefined; +} + +/** The span attributes of the CI job of a run. */ +export function ciAttributes(ci: RunCi | undefined): Record { + return Object.fromEntries( + Object.entries(CI_ATTRIBUTES).flatMap(([field, name]) => { + const value = ci?.[field as keyof RunCi]; + return value ? [[name, value]] : []; + }), + ); +} diff --git a/packages/service/src/secrets.ts b/packages/service/src/secrets.ts index fb3c7f8..b3d23d1 100644 --- a/packages/service/src/secrets.ts +++ b/packages/service/src/secrets.ts @@ -10,7 +10,7 @@ * recording keeps. * * It parses a JSON body and walks it to find the fields. Then it replaces - * them in one pass over the JSON strings of the text (`scanStrings`). It + * them in one pass over the JSON strings of the text (`scanStrings` of `json-strings.ts`). It * does not run a regular expression on JSON text, because a long string * value can overflow the stack of the expression, and it does not run a * pass for each value, because a client can send many. So it finds a @@ -21,6 +21,14 @@ * or a copy of a value inside other text. */ import type { IncomingHttpHeaders } from "node:http"; +import { + parseJson, + isOpen, + isJsonSpace, + type Escapes, + scanStrings, + decodeString, +} from "./json-strings.ts"; /** * Request headers and query parameters that can carry credentials. The @@ -159,138 +167,7 @@ const isCredentialField: CredentialTest = (name, value) => { const isCredentialParam: CredentialTest = (name) => CREDENTIAL_HEADER.test(name); -/** - * The parsed value of a JSON text, or `undefined` when it is not JSON. A - * byte order mark at the start does not count. - */ -const parseJson = (text: string): unknown => { - try { - return JSON.parse(text.startsWith("\uFEFF") ? text.slice(1) : text); - } catch { - return undefined; - } -}; - const REDACTED_JSON = `"${REDACTED}"`; -const QUOTE = 0x22; -const BACKSLASH = 0x5c; -const LINE_BREAK = 0x0a; -const COLON = 0x3a; -/** `{` or `[`. A compare is faster than a set for each character. */ -const isOpen = (code: number) => code === 0x7b || code === 0x5b; -/** `}` or `]`. */ -const isClose = (code: number) => code === 0x7d || code === 0x5d; -/** Space, tab, line feed, or carriage return. */ -const isJsonSpace = (code: number) => - code === 0x20 || code === 0x09 || code === 0x0a || code === 0x0d; - -/** - * The escapes of a JSON string: it has none, all are valid, or one is not - * valid (or it has a control character), so `JSON.parse` would throw. - */ -type Escapes = "none" | "valid" | "bad"; -/** - * The escapes that are not `u` escapes, by the character after the - * backslash, and the character that each one gives. - */ -const SIMPLE_ESCAPES = new Map([ - [0x22, '"'], - [0x5c, "\\"], - [0x2f, "/"], - [0x62, "\b"], - [0x66, "\f"], - [0x6e, "\n"], - [0x72, "\r"], - [0x74, "\t"], -]); -const U = 0x75; - -/** Tells whether the character at `at` of `text` is a hex digit. */ -function isHex(text: string, at: number): boolean { - const code = text.charCodeAt(at); - return ( - (code >= 0x30 && code <= 0x39) || - (code >= 0x41 && code <= 0x46) || - (code >= 0x61 && code <= 0x66) - ); -} - -/** Tells whether `text` has a valid `u` escape at `at` (its backslash). */ -const uEscape = (text: string, at: number) => - text.charCodeAt(at + 1) === U && - isHex(text, at + 2) && - isHex(text, at + 3) && - isHex(text, at + 4) && - isHex(text, at + 5); - -/** - * Call `visit` for each JSON string of `text`, in one pass: its start, its - * end after the closing quote, the depth of objects and arrays around it, - * whether it is a field name (a `:` follows it), its escapes (`Escapes`), - * and the length of its value. The length is right only when the escapes - * are valid. A string ends at a line break, because JSON does not allow - * one in a string. So a line that is not JSON, such as a header line in - * the text of a key, cannot make it read the next lines wrong. - */ -function scanStrings( - text: string, - visit: ( - start: number, - end: number, - depth: number, - name: boolean, - escapes: Escapes, - length: number, - ) => void, -): void { - let depth = 0; - let at = 0; - while (at < text.length) { - const code = text.charCodeAt(at); - if (isOpen(code)) depth += 1; - else if (isClose(code)) depth -= 1; - if (code !== QUOTE) { - at += 1; - continue; - } - let end = at + 1; - let escapes: Escapes = "none"; - let length = 0; - while (end < text.length) { - const next = text.charCodeAt(end); - if (next === QUOTE) break; - if (next === LINE_BREAK) break; - length += 1; - if (next === BACKSLASH && text.charCodeAt(end + 1) !== LINE_BREAK) { - if (escapes === "none") escapes = "valid"; - // The hex digits of a `u` escape cannot be a quote or a line break. - if (uEscape(text, end)) end += 6; - else { - if (!SIMPLE_ESCAPES.has(text.charCodeAt(end + 1))) escapes = "bad"; - end += 2; - } - } else { - // JSON does not allow a control character, such as a tab, in a - // string. A header line can have one. - if (next < 0x20) escapes = "bad"; - end += 1; - } - } - if (text.charCodeAt(end) !== QUOTE) { - // Not a string. Each quote before `end` has a backslash before it, so - // no string can start there either. Go on at `end`, so each character - // is read once. - at = end; - continue; - } - end += 1; - let after = end; - while (isJsonSpace(text.charCodeAt(after))) after += 1; - visit(at, end, depth, text.charCodeAt(after) === COLON, escapes, length); - at = end; - } -} - /** * The kind of a field: a credential field at any value, or a bare `token` * that is a credential only with a credential prefix. @@ -805,45 +682,6 @@ const mintedValue = (json: Record, name: string) => { return typeof value === "string" && isMinted(name, value); }; -/** - * The value of the JSON string from `start` to `end` of `text`, or - * `undefined` when its escapes are not valid (`scanStrings`). It decodes - * the escapes itself, so it does not throw and it costs little for each - * string. A `u` escape gives one UTF-16 code unit, as in `JSON.parse`, so a - * lone surrogate stays. - */ -function decodeString( - text: string, - start: number, - end: number, - escapes: Escapes, -): string | undefined { - if (escapes === "none") return text.slice(start + 1, end - 1); - if (escapes === "bad") return undefined; - let value = ""; - let from = start + 1; - let at = from; - while (at < end - 1) { - if (text.charCodeAt(at) !== BACKSLASH) { - at += 1; - continue; - } - value += text.slice(from, at); - const next = text.charCodeAt(at + 1); - if (next === U) { - value += String.fromCharCode( - Number.parseInt(text.slice(at + 2, at + 6), 16), - ); - at += 6; - } else { - value += SIMPLE_ESCAPES.get(next)!; - at += 2; - } - from = at; - } - return value + text.slice(from, end - 1); -} - /** * The JSON text of a response body with the value of each minted field of * the object at the top replaced (`isMinted`), and those values. It diff --git a/packages/service/src/server.ts b/packages/service/src/server.ts index be4c59f..0dd2210 100644 --- a/packages/service/src/server.ts +++ b/packages/service/src/server.ts @@ -23,9 +23,8 @@ * `missed` (`replay` mode had no recording), or `passthrough` (no rule * matched). * - * Behind a load balancer, a connection can start with a PROXY protocol v1 - * header (`proxyProtocol`). It gives the address of the client - * (`clientAddress()`). + * Request bodies and their shared budget are in `bodies.ts`. The PROXY + * protocol header and the address of the client are in `clients.ts`. */ import { timingSafeEqual } from "node:crypto"; import http from "node:http"; @@ -42,6 +41,14 @@ import type { CertificateAuthority } from "./certificates.ts"; import { reportInternal } from "./errors.ts"; import type { ProxyRequest, Recorder, Telemetry } from "./recorder.ts"; import type { OpenedSession } from "@sentry/roach/types"; +import { + type BodyBudget, + type Held, + readBody, + readJson, + release, +} from "./bodies.ts"; +import { acceptProxied } from "./clients.ts"; /** Headers for one connection, which the proxy must not forward. */ const HOP_HEADERS = new Set([ @@ -60,16 +67,12 @@ const AUTHORITY = /^(\[[0-9a-f:.]+\]|[a-z0-9.-]+):(\d{1,5})$/i; * has a smaller `maxBodyBytes`. */ const MAX_BODY_BYTES = 64 * 1024 * 1024; -/** - * A control API body over this size gets HTTP 413. Anyone can call - * `POST /runs`, so the service reads no more before it checks the token. - */ -const MAX_CONTROL_BODY_BYTES = 1024 * 1024; /** A tunnel that sends no byte this long after its `200` closes. */ const TUNNEL_FIRST_BYTE_MS = 10_000; /** * A tunnel that sends and gets no byte this long closes. A live model - * call can take minutes before its first byte, so this is long. + * call can take minutes before its first byte, so this is long. Between + * two requests, Node closes the tunnel sooner (`keepAliveTimeout`, 5 s). */ const TUNNEL_IDLE_MS = 10 * 60 * 1000; /** @@ -81,75 +84,6 @@ const HTTP_OPTIONS: http.ServerOptions = { headersTimeout: 10_000, connectionsCheckingInterval: 1000, }; -/** - * A PROXY protocol v1 header, such as - * `PROXY TCP4 203.0.113.7 10.0.0.2 51234 443`. It is at most 107 bytes. - * The first bytes of a connection must come in `PROXY_HEADER_MS`. - */ -const PROXY_PREFIX = Buffer.from("PROXY "); -const PROXY_HEADER = - /^PROXY (?:TCP[46] ([0-9A-Fa-f:.]+) [0-9A-Fa-f:.]+ \d{1,5} \d{1,5}|UNKNOWN[^\r\n]*)\r\n/; -const MAX_PROXY_HEADER_BYTES = 107; -const PROXY_HEADER_MS = 10_000; - -/** The address of the client of each connection with a PROXY header. */ -const proxiedClients = new WeakMap(); - -/** - * The address of the client of a request. With `proxyProtocol`, it comes - * from the PROXY header of the connection, when the connection has one. - * Else it is the address of the socket. - */ -export function clientAddress(incoming: http.IncomingMessage): string { - return ( - proxiedClients.get(incoming.socket) ?? incoming.socket.remoteAddress ?? "" - ); -} - -/** - * The group of a client address for the limit of runs for each client. An - * IPv6 client often has a whole /64, so the group of an IPv6 address is - * its /64, such as `2001:db8:0:1::/64`. An IPv4 address is its own group, - * also in each IPv6 form, such as `::ffff:192.0.2.1`, `::ffff:c000:201`, - * and `0:0:0:0:0:ffff:192.0.2.1`. - */ -export function clientGroup(address: string): string { - if (isIP(address) !== 6) return address; - const groups = ipv6Groups(address); - if ( - groups.slice(0, 5).every((group) => group === 0) && - groups[5] === 0xffff - ) { - const [high = 0, low = 0] = groups.slice(6); - return [high >> 8, high & 0xff, low >> 8, low & 0xff].join("."); - } - return `${groups - .slice(0, 4) - .map((group) => group.toString(16)) - .join(":")}::/64`; -} - -/** The eight 16-bit groups of an IPv6 address. */ -function ipv6Groups(address: string): number[] { - // The URL parser writes each IPv6 form with hex groups only, such as - // `::ffff:c000:201`. A zone, such as `%eth0`, is not part of it. - const [head = "", tail] = new URL( - `http://[${address.split("%")[0]}]`, - ).hostname - .slice(1, -1) - .split("::"); - const front = head ? head.split(":") : []; - const back = tail ? tail.split(":") : []; - const zeros = Array(8 - front.length - back.length).fill("0"); - return [...front, ...zeros, ...back].map((group) => - Number.parseInt(group, 16), - ); -} - -/** An error that answers a request with its own status. */ -const httpError = (status: number, message: string) => - Object.assign(new Error(message), { status }); - /** Parse the allowed origins, by origin. A value that is not one throws. */ export function parseOrigins(values: string[]): Map { return new Map( @@ -172,159 +106,6 @@ function authorityOf(origin: URL): string { return `${origin.hostname}:${port}`; } -/** - * The bytes of request bodies that the requests of many targets can hold - * at one time (`ProxyTarget.bodyBudget`). `used` counts the bodies of the - * requests that did not finish, and `clients` counts them for each client - * (`ProxyTarget.client`). One client can hold at most `perClient` bytes, - * so it cannot take all of `max`. A body that takes longer than `readMs` - * to come fails with HTTP 408 and gives its bytes back. - */ -export interface BodyBudget { - max: number; - perClient: number; - readMs: number; - used: number; - clients: Map; -} - -/** The bytes that one request holds of its `BodyBudget`. */ -interface Held { - budget: BodyBudget | undefined; - client: string; - bytes: number; -} - -/** Hold `bytes` more of the budget. Returns `false` when they do not fit. */ -function hold(held: Held, bytes: number): boolean { - const { budget, client } = held; - if (!budget) return true; - const ofClient = budget.clients.get(client) ?? 0; - if (budget.used + bytes > budget.max || ofClient + bytes > budget.perClient) { - return false; - } - budget.used += bytes; - budget.clients.set(client, ofClient + bytes); - held.bytes += bytes; - return true; -} - -/** Give back all bytes that a request holds. */ -function release(held: Held): void { - const { budget, client, bytes } = held; - held.bytes = 0; - if (!budget || bytes === 0) return; - budget.used -= bytes; - const left = (budget.clients.get(client) ?? 0) - bytes; - if (left > 0) budget.clients.set(client, left); - else budget.clients.delete(client); -} - -const busyError = () => - httpError( - 503, - "the service has too many large requests at this time. Send the request again later.", - ); - -/** - * Read a whole body. A body over `maxBytes` fails with HTTP 413. A body - * that does not fit in the budget of `held` fails with HTTP 503, and one - * that does not come in the `readMs` of the budget fails with HTTP 408. - * Only request bodies have a limit: the client sends them, and a large - * upstream response is not the fault of the client. - */ -function readBody( - stream: http.IncomingMessage, - maxBytes = Infinity, - held: Held = { budget: undefined, client: "", bytes: 0 }, -): Promise { - const declared = Number(stream.headers["content-length"]); - if (declared > maxBytes) { - // Drop the body without a look at it, so it costs no memory. - stream.resume(); - return Promise.reject(httpError(413, `body is over ${maxBytes} bytes`)); - } - // Hold the bytes when they come, not the declared length. So a client - // that declares a large body and then sends nothing holds nothing. - return new Promise((resolve, reject) => { - const chunks: Buffer[] = []; - let size = 0; - let failed = false; - const fail = (error: Error) => { - // Read the rest without keeping it, so the client gets the error. - failed = true; - chunks.length = 0; - clearTimeout(deadline); - reject(error); - }; - // Node gives a whole request 300 s (`HTTP_OPTIONS`). That is too long - // to hold a part of the budget, so these bodies have a shorter limit. - const readMs = held.budget?.readMs; - const deadline = - readMs === undefined - ? undefined - : setTimeout( - () => - fail( - httpError(408, `the body did not come in ${readMs / 1000} s`), - ), - readMs, - ); - stream.on("data", (chunk: Buffer) => { - if (failed) return; - size += chunk.length; - if (size > maxBytes) { - fail(httpError(413, `body is over ${maxBytes} bytes`)); - } else if (!hold(held, chunk.length)) { - fail(busyError()); - } else { - chunks.push(chunk); - } - }); - stream.on("end", () => { - clearTimeout(deadline); - resolve(Buffer.concat(chunks)); - }); - stream.on("error", (error: Error & { code?: string }) => { - clearTimeout(deadline); - // A client that closes its connection before its body came is not a - // failure of the proxy, so it does not go to Sentry. `method` is set - // only on a request, not on an upstream response. - const aborted = stream.method !== null && error.code === "ECONNRESET"; - reject(aborted ? abortedError() : error); - }); - }); -} - -/** The error of a request whose client closed it before its body came. */ -const abortedError = () => - Object.assign( - httpError(400, "the client closed the request before its body came"), - { aborted: true }, - ); - -/** - * Read a JSON object body of a control API call. A body over - * `MAX_CONTROL_BODY_BYTES` fails with HTTP 413 before it is parsed. Any - * other body that is not a JSON object fails with HTTP 400. - */ -export async function readJson(incoming: http.IncomingMessage): Promise { - const text = (await readBody(incoming, MAX_CONTROL_BODY_BYTES)).toString( - "utf8", - ); - let value: unknown; - try { - value = JSON.parse(text || "{}"); - } catch { - throw httpError(400, "body is not JSON"); - } - // Callers read fields, so `null`, a list, or a single value is a bad body. - if (typeof value !== "object" || value === null || Array.isArray(value)) { - throw httpError(400, "body is not a JSON object"); - } - return value as T; -} - /** Compare tokens in constant time. */ export function sameToken( actual: string | undefined, @@ -615,54 +396,6 @@ function refuse(status: 403 | 407, outgoing: http.ServerResponse): void { ); } -/** - * Read the PROXY protocol v1 header of a new connection, when it has one, - * and keep the address of its client. Then give the connection to - * `server`. A connection without the header keeps the address of its - * socket. A header that is not valid closes the connection. - * - * The load balancer writes its header first, so a client cannot put a - * header before it. Without a load balancer, a client can send its own - * header. That changes only the address for the limit of runs for each - * client. - */ -function acceptProxied(socket: Socket, server: http.Server): void { - let read = Buffer.alloc(0); - const timer = setTimeout(() => socket.destroy(), PROXY_HEADER_MS); - const accept = (rest: Buffer) => { - socket.off("data", onData); - clearTimeout(timer); - socket.pause(); - if (rest.length > 0) socket.unshift(rest); - server.emit("connection", socket); - socket.resume(); - }; - const onData = (chunk: Buffer) => { - read = Buffer.concat([read, chunk]); - const prefix = Math.min(read.length, PROXY_PREFIX.length); - if (!read.subarray(0, prefix).equals(PROXY_PREFIX.subarray(0, prefix))) { - accept(read); - return; - } - const end = read.indexOf("\r\n"); - if (end < 0 || end + 2 > MAX_PROXY_HEADER_BYTES) { - if (end >= 0 || read.length >= MAX_PROXY_HEADER_BYTES) socket.destroy(); - return; - } - const match = PROXY_HEADER.exec(read.subarray(0, end + 2).toString()); - if (!match) { - socket.destroy(); - return; - } - // `UNKNOWN`, such as from a health check, gives no address. - if (match[1] && isIP(match[1])) proxiedClients.set(socket, match[1]); - accept(read.subarray(end + 2)); - }; - socket.on("error", () => socket.destroy()); - socket.on("close", () => clearTimeout(timer)); - socket.on("data", onData); -} - /** Proxy one request through its target, and answer errors with 502. */ function serveProxied( { target, session }: { target: ProxyTarget; session: string | undefined }, diff --git a/packages/service/src/service.ts b/packages/service/src/service.ts index 798217c..17b5fb1 100644 --- a/packages/service/src/service.ts +++ b/packages/service/src/service.ts @@ -48,29 +48,25 @@ import { type WorkLimit, } from "./recorder.ts"; import { createFileStore } from "./recordings.ts"; +import { readJson, type BodyBudget } from "./bodies.ts"; +import { clientAddress, clientGroup } from "./clients.ts"; import { - clientAddress, - clientGroup, controlRecorder, listenProxy, parseOrigins, - readJson, sameToken, sendJson, - type BodyBudget, type ProxyTarget, type TargetResult, } from "./server.ts"; -import { RULE_NAME, type RecordingStore } from "./store.ts"; +import { ciAttributes, invalidRunConfig, TENANT_NAME } from "./run-config.ts"; +import type { RecordingStore } from "./store.ts"; import { CONTROL_PATH, - type RecordingMode, type RecordingStats, type RemoteRun, - type RunCi, type RunConfig, } from "@sentry/roach/types"; -import { invalidValuePatterns } from "@sentry/roach/values"; /** The configuration of the service. The operator owns it. */ export interface RoachServiceConfig { @@ -134,7 +130,6 @@ interface Run extends ProxyTarget { lastUsed: number; } -const MODES = new Set(["auto", "off", "record", "replay"]); /** The service gets little traffic, so it keeps all traces. */ const TRACES_SAMPLE_RATE = 1; /** A run that is open longer than this ends as failed, such as a dead job. */ @@ -214,28 +209,6 @@ const WRITE_WORK_LIMIT: WorkLimit = { totalMs: Infinity, untrusted: false, }; -const MAX_RUN_NAME = 128; -/** - * Limits of the config of a run. Anyone can start a run, so they keep the - * work of a run small: the service makes a certificate for each host that - * a run tunnels to, and runs each value pattern on each request. - */ -const MAX_RUN_ORIGINS = 32; -const MAX_RULES = 64; -const MAX_RULE_VALUES = 32; -const MAX_RULE_URLS = 32; -const MAX_PATTERN_LENGTH = 2048; -/** The span attribute of each field of `RunConfig.ci`. */ -const CI_ATTRIBUTES: Record = { - runId: "roach.ci.run_id", - job: "roach.ci.job", - repo: "roach.ci.repo", - ref: "roach.ci.ref", -}; -const MAX_CI_VALUE = 256; -/** Enough for the API keys and OIDC tokens of a CI job. */ -const MAX_RUN_SECRETS = 32; -const MAX_SECRET_LENGTH = 16 * 1024; /** * The credentials that a run learns from its requests (`secrets.ts`). A * run without a token gets a small limit, because anyone can start one. @@ -244,13 +217,6 @@ const MAX_SECRET_LENGTH = 16 * 1024; */ const PUBLIC_LEARN_LIMIT = { values: 1024, length: 1024 * 1024 }; const WRITE_LEARN_LIMIT = { values: 16 * 1024, length: 16 * 1024 * 1024 }; -/** - * A tenant name: `owner/repo`, with the characters of GitHub names. It is a - * path in the store, so each part starts with a letter, digit, or `_`, and - * there are always two parts. Then no tenant is in the directory of another. - */ -const TENANT_NAME = /^[A-Za-z0-9_][\w.-]{0,99}\/[A-Za-z0-9_][\w.-]{0,99}$/; - const sha256 = (value: string) => createHash("sha256").update(value).digest(); /** Check that `authorization` has the write token, in constant time. */ @@ -293,119 +259,6 @@ function proxyUrl( return url.href.replace(/\/$/, ""); } -/** Check the run config of a client. Returns the error, or `undefined`. */ -function invalidRunConfig(config: Partial): string | undefined { - if (!config.mode || !MODES.has(config.mode)) { - return "mode must be auto, off, record, or replay"; - } - if (config.name !== undefined) { - if (typeof config.name !== "string" || config.name.length > MAX_RUN_NAME) { - return `name must be a string of at most ${MAX_RUN_NAME} characters`; - } - } - if (config.secrets !== undefined) { - if ( - !Array.isArray(config.secrets) || - config.secrets.length > MAX_RUN_SECRETS || - !config.secrets.every( - (secret) => - typeof secret === "string" && secret.length <= MAX_SECRET_LENGTH, - ) - ) { - return `secrets must be a list of at most ${MAX_RUN_SECRETS} strings of at most ${MAX_SECRET_LENGTH} characters`; - } - } - if (config.ci !== undefined) { - // A newer client can send more fields. The service ignores them. - const ci = config.ci as Record | null; - if ( - typeof ci !== "object" || - ci === null || - !Object.keys(CI_ATTRIBUTES).every( - (field) => - ci[field] === undefined || - (typeof ci[field] === "string" && ci[field].length <= MAX_CI_VALUE), - ) - ) { - return `ci must be an object of strings of at most ${MAX_CI_VALUE} characters`; - } - } - if (!Array.isArray(config.rules) || config.rules.length > MAX_RULES) { - return `rules must be a list of at most ${MAX_RULES} rules`; - } - const names = new Set(); - for (const rule of config.rules) { - if (typeof rule?.name !== "string" || !RULE_NAME.test(rule.name)) { - return `rule name must match ${RULE_NAME}`; - } - if (names.has(rule.name)) return `rule ${rule.name} is not unique`; - names.add(rule.name); - if (typeof rule.match !== "object" || rule.match === null) { - return `rule ${rule.name} must have match`; - } - const { url } = rule.match; - if ( - url !== undefined && - typeof url !== "string" && - !( - Array.isArray(url) && - url.length > 0 && - url.length <= MAX_RULE_URLS && - url.every((prefix) => typeof prefix === "string") - ) - ) { - return `match.url of rule ${rule.name} must be a string or a list of 1 to ${MAX_RULE_URLS} strings`; - } - // A slow pattern fails only its own requests (`request-key.ts`), so - // the service takes any pattern that compiles. - if (rule.values !== undefined) { - if (typeof rule.values !== "object" || rule.values === null) { - return `values of rule ${rule.name} must be an object`; - } - const values = Object.entries(rule.values); - if (values.length > MAX_RULE_VALUES) { - return `rule ${rule.name} can have at most ${MAX_RULE_VALUES} values`; - } - for (const [name, source] of values) { - if (typeof source !== "string" || source.length > MAX_PATTERN_LENGTH) { - return `value ${name} of rule ${rule.name} must be a pattern of at most ${MAX_PATTERN_LENGTH} characters`; - } - } - } - } - const patternError = invalidValuePatterns( - config.rules.flatMap((rule) => (rule.values ? [rule.values] : [])), - ); - if (patternError) return patternError; - if ( - !Array.isArray(config.allow) || - config.allow.length === 0 || - config.allow.length > MAX_RUN_ORIGINS - ) { - return `allow must be a list of 1 to ${MAX_RUN_ORIGINS} origins`; - } - for (const origin of config.allow) { - // The client sends JSON, so check the type at run time. - if (typeof origin !== "string") return "allow must be a list of origins"; - try { - parseOrigins([origin.replace(/\/$/, "")]); - } catch { - return `${origin} is not an http or https origin`; - } - } - return undefined; -} - -/** The span attributes of the CI job of a run. */ -function ciAttributes(ci: RunCi | undefined): Record { - return Object.fromEntries( - Object.entries(CI_ATTRIBUTES).flatMap(([field, name]) => { - const value = ci?.[field as keyof RunCi]; - return value ? [[name, value]] : []; - }), - ); -} - /** * Send an unexpected error to Sentry, with `tags`. Returns the id of its * event, or `undefined` without a DSN. diff --git a/packages/service/tests/service.test.ts b/packages/service/tests/service.test.ts index ece17c8..c86f8dc 100644 --- a/packages/service/tests/service.test.ts +++ b/packages/service/tests/service.test.ts @@ -32,7 +32,7 @@ import { import { describeRecordingMisses } from "@sentry/roach/report"; import { keyInWorker } from "../src/key-worker.ts"; import { keyRequest } from "../src/request-key.ts"; -import { clientGroup } from "../src/server.ts"; +import { clientGroup } from "../src/clients.ts"; import { startRoachService, type RoachService } from "../src/service.ts"; import type { RemoteRun, RunConfig } from "@sentry/roach/types"; import { VALUE_PATTERNS } from "@sentry/roach/values";