| 1 | /** |
| 2 | * Brokered secrets for a self-hosted runner's job. That job's traffic |
| 3 | * never passes `egress`, and a brokered secret's value is never sent to a |
| 4 | * computer g1t does not run, so the job's harness (crates/runner |
| 5 | * actions/broker.rs) passes each of its requests to a brokered secret's |
| 6 | * hosts that carries a stand-in to g1t, `POST /actions/jobs/{job}/broker`, |
| 7 | * which the API hands here as it is (apps/api broker.rs). This fills the |
| 8 | * stand-ins in as broker.ts does for g1t's own sandboxes, sends the |
| 9 | * request without following redirects, and answers with what came back, |
| 10 | * the values swapped back for their stand-ins in its headers and in any |
| 11 | * text body, whatever its size. |
| 12 | * |
| 13 | * It is not a proxy for anything else: a request that carries none of |
| 14 | * the job's stand-ins for its host, or is not HTTPS, is refused, and the |
| 15 | * job's log says so once per secret (or host) and job. The request comes |
| 16 | * with the job's current token, in `Authorization`, and only a running |
| 17 | * self-hosted job's token opens its stand-ins (`relayed_secrets`). |
| 18 | * |
| 19 | * The request to make: `x-g1t-broker-request`, base64 of |
| 20 | * `{ method, url, headers: [[name, value], …] }`, and the body. The |
| 21 | * answer: status 200 with `x-g1t-broker-response`, base64 of |
| 22 | * `{ status, headers }`, and the upstream body, decoded; or g1t's own |
| 23 | * refusal, with another status, a JSON `error` and no such header. |
| 24 | */ |
| 25 | |
| 26 | import type { BrokeredSecret, ServiceBinding } from "@g1t/contracts"; |
| 27 | |
| 28 | import { fill, misdirectedNote, newlyMisdirected, scrubHeaders, standInsIn } from "./broker.ts"; |
| 29 | import { normalizeHost } from "./egress.ts"; |
| 30 | |
| 31 | export const REQUEST_HEADER = "x-g1t-broker-request"; |
| 32 | export const ANSWER_HEADER = "x-g1t-broker-response"; |
| 33 | /** The largest request body relayed: a package or a release asset upload. */ |
| 34 | export const MAX_REQUEST_BYTES = 32 * 1024 * 1024; |
| 35 | /** The largest answer relayed. */ |
| 36 | export const MAX_ANSWER_BYTES = 1024 * 1024 * 1024; |
| 37 | /** How long one relayed request may take, its answer's body included. */ |
| 38 | export const TIME_LIMIT_MS = 10 * 60_000; |
| 39 | /** What may be relayed: no `CONNECT`, `TRACE` or anything unusual. */ |
| 40 | const METHODS = new Set(["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]); |
| 41 | /** |
| 42 | * Headers about one connection, not the request, which are not passed on |
| 43 | * either way; `host` comes from the URL, and lengths are worked out again. |
| 44 | */ |
| 45 | const HOP_BY_HOP = new Set([ |
| 46 | "connection", |
| 47 | "keep-alive", |
| 48 | "proxy-authenticate", |
| 49 | "proxy-authorization", |
| 50 | "proxy-connection", |
| 51 | "te", |
| 52 | "trailer", |
| 53 | "transfer-encoding", |
| 54 | "upgrade", |
| 55 | "host", |
| 56 | "content-length", |
| 57 | ]); |
| 58 | |
| 59 | /** The request a harness asked to have made. */ |
| 60 | export type Relayed = { method: string; url: string; headers: [string, string][] }; |
| 61 | |
| 62 | /** The request in `x-g1t-broker-request`, or null when it is not one. */ |
| 63 | export function readRelayed(header: string | null): Relayed | null { |
| 64 | if (!header) return null; |
| 65 | try { |
| 66 | const bytes = Uint8Array.from(atob(header.trim()), (c) => c.charCodeAt(0)); |
| 67 | const value = JSON.parse(new TextDecoder("utf-8", { fatal: true, ignoreBOM: false }).decode(bytes)) as Partial<Relayed>; |
| 68 | if (typeof value.method !== "string" || typeof value.url !== "string" || !Array.isArray(value.headers)) return null; |
| 69 | const headers = value.headers.filter( |
| 70 | (pair): pair is [string, string] => Array.isArray(pair) && pair.length === 2 && typeof pair[0] === "string" && typeof pair[1] === "string", |
| 71 | ); |
| 72 | if (headers.length !== value.headers.length) return null; |
| 73 | return { method: value.method.toUpperCase(), url: value.url, headers }; |
| 74 | } catch { |
| 75 | return null; |
| 76 | } |
| 77 | } |
| 78 | |
| 79 | function encode(value: unknown): string { |
| 80 | let binary = ""; |
| 81 | for (const byte of new TextEncoder().encode(JSON.stringify(value))) binary += String.fromCharCode(byte); |
| 82 | return btoa(binary); |
| 83 | } |
| 84 | |
| 85 | /** A request's headers as they go on: the connection's own left out. */ |
| 86 | export function passedHeaders(pairs: readonly [string, string][]): Headers { |
| 87 | const headers = new Headers(); |
| 88 | for (const [name, value] of pairs) { |
| 89 | if (!HOP_BY_HOP.has(name.toLowerCase())) headers.append(name, value); |
| 90 | } |
| 91 | return headers; |
| 92 | } |
| 93 | |
| 94 | /** An answer's headers as a list, each `set-cookie` on its own. */ |
| 95 | export function answerHeaders(headers: Headers): [string, string][] { |
| 96 | const out: [string, string][] = []; |
| 97 | headers.forEach((value, name) => { |
| 98 | const lower = name.toLowerCase(); |
| 99 | // The body is relayed decoded, so its encoding and length go too. |
| 100 | if (HOP_BY_HOP.has(lower) || lower === "content-encoding" || lower === "set-cookie") return; |
| 101 | out.push([name, value]); |
| 102 | }); |
| 103 | const cookies = (headers as Headers & { getSetCookie?: () => string[] }).getSetCookie?.() ?? []; |
| 104 | for (const cookie of cookies) out.push(["set-cookie", cookie]); |
| 105 | return out; |
| 106 | } |
| 107 | |
| 108 | /** A body read whole, or null when it is longer than `limit`. */ |
| 109 | export async function readCapped(body: ReadableStream<Uint8Array> | null, limit: number): Promise<Uint8Array | null> { |
| 110 | if (!body) return new Uint8Array(0); |
| 111 | const reader = body.getReader(); |
| 112 | const parts: Uint8Array[] = []; |
| 113 | let length = 0; |
| 114 | for (;;) { |
| 115 | const { done, value } = await reader.read(); |
| 116 | if (done) break; |
| 117 | length += value.byteLength; |
| 118 | if (length > limit) { |
| 119 | await reader.cancel().catch(() => undefined); |
| 120 | return null; |
| 121 | } |
| 122 | parts.push(value); |
| 123 | } |
| 124 | return concat(parts); |
| 125 | } |
| 126 | |
| 127 | function concat(parts: readonly Uint8Array[]): Uint8Array { |
| 128 | const out = new Uint8Array(parts.reduce((sum, part) => sum + part.byteLength, 0)); |
| 129 | let at = 0; |
| 130 | for (const part of parts) { |
| 131 | out.set(part, at); |
| 132 | at += part.byteLength; |
| 133 | } |
| 134 | return out; |
| 135 | } |
| 136 | |
| 137 | /** A stream that fails once more than `limit` bytes have passed. */ |
| 138 | export function capped(limit: number): TransformStream<Uint8Array, Uint8Array> { |
| 139 | let passed = 0; |
| 140 | return new TransformStream({ |
| 141 | transform(chunk, controller) { |
| 142 | passed += chunk.byteLength; |
| 143 | if (passed > limit) controller.error(new Error(`the answer is larger than ${limit} bytes`)); |
| 144 | else controller.enqueue(chunk); |
| 145 | }, |
| 146 | }); |
| 147 | } |
| 148 | |
| 149 | type Pair = { from: Uint8Array; to: Uint8Array }; |
| 150 | |
| 151 | function matchesAt(buffer: Uint8Array, at: number, pattern: Uint8Array): boolean { |
| 152 | if (at + pattern.length > buffer.length) return false; |
| 153 | for (let i = 0; i < pattern.length; i++) if (buffer[at + i] !== pattern[i]) return false; |
| 154 | return true; |
| 155 | } |
| 156 | |
| 157 | /** |
| 158 | * Swaps each pair's `from` for its `to` in `buffer`. Not `final`, the |
| 159 | * last bytes that could still begin a value are kept back as `rest`, to |
| 160 | * be read again with the next chunk, so a value split between two chunks |
| 161 | * is found too. |
| 162 | */ |
| 163 | export function replaceBytes(buffer: Uint8Array, pairs: readonly Pair[], final: boolean): { out: Uint8Array; rest: Uint8Array } { |
| 164 | const longest = pairs.reduce((most, pair) => Math.max(most, pair.from.length), 0); |
| 165 | const stop = final ? buffer.length : buffer.length - longest + 1; |
| 166 | const parts: Uint8Array[] = []; |
| 167 | let start = 0; |
| 168 | let i = 0; |
| 169 | while (i < stop) { |
| 170 | const pair = pairs.find((p) => p.from.length > 0 && buffer[i] === p.from[0] && matchesAt(buffer, i, p.from)); |
| 171 | if (pair) { |
| 172 | parts.push(buffer.subarray(start, i), pair.to); |
| 173 | i += pair.from.length; |
| 174 | start = i; |
| 175 | } else { |
| 176 | i++; |
| 177 | } |
| 178 | } |
| 179 | i = Math.max(i, start); |
| 180 | parts.push(buffer.subarray(start, Math.min(i, buffer.length))); |
| 181 | return { out: concat(parts), rest: buffer.slice(Math.min(i, buffer.length)) }; |
| 182 | } |
| 183 | |
| 184 | /** A stream with each filled-in value swapped back for its stand-in. */ |
| 185 | export function scrubStream(filled: readonly BrokeredSecret[]): TransformStream<Uint8Array, Uint8Array> { |
| 186 | const encoder = new TextEncoder(); |
| 187 | const pairs: Pair[] = filled |
| 188 | .filter((secret) => secret.value) |
| 189 | .map((secret) => ({ from: encoder.encode(secret.value), to: encoder.encode(secret.standIn) })) |
| 190 | .sort((a, b) => b.from.length - a.from.length); |
| 191 | let rest: Uint8Array = new Uint8Array(0); |
| 192 | return new TransformStream({ |
| 193 | transform(chunk, controller) { |
| 194 | const swapped = replaceBytes(concat([rest, chunk]), pairs, false); |
| 195 | rest = swapped.rest; |
| 196 | if (swapped.out.byteLength) controller.enqueue(swapped.out); |
| 197 | }, |
| 198 | flush(controller) { |
| 199 | const swapped = replaceBytes(rest, pairs, true); |
| 200 | if (swapped.out.byteLength) controller.enqueue(swapped.out); |
| 201 | }, |
| 202 | }); |
| 203 | } |
| 204 | |
| 205 | /** |
| 206 | * Whether an answer's body is scrubbed: text of any kind (JSON, XML, HTML, |
| 207 | * forms), and one that does not say. Archives, images and other binary |
| 208 | * bodies stream as they are. |
| 209 | */ |
| 210 | export function scrubbedType(headers: Headers): boolean { |
| 211 | const type = (headers.get("content-type") ?? "").toLowerCase().trim(); |
| 212 | return type === "" || /^(text\/|application\/(json|xml|javascript|x-www-form-urlencoded|[a-z0-9.+-]+\+(json|xml)))/.test(type); |
| 213 | } |
| 214 | |
| 215 | /** What the relay needs from around it: the job's stand-ins, its log, and the internet. */ |
| 216 | export type RelayDeps = { |
| 217 | /** A running self-hosted job's stand-ins, or null when the token does not open it. */ |
| 218 | secrets(job: string, token: string): Promise<BrokeredSecret[] | null>; |
| 219 | /** A line for the job's log. */ |
| 220 | note(job: string, token: string, text: string): Promise<void>; |
| 221 | /** Sends a request upstream (forward.ts). */ |
| 222 | send(request: Request): Promise<Response>; |
| 223 | }; |
| 224 | |
| 225 | function refused(status: number, code: string, message: string): Response { |
| 226 | return Response.json({ error: { code, message } }, { status }); |
| 227 | } |
| 228 | |
| 229 | /** The line a job's log gets for a request that carried none of its stand-ins for its host. */ |
| 230 | export function refusedNote(host: string): string { |
| 231 | return `##[warning]A request to ${normalizeHost(host)} was passed to g1t's broker without a stand-in for that host, so g1t refused it.`; |
| 232 | } |
| 233 | |
| 234 | /** What each job's log was told, in this isolate: each thing once. */ |
| 235 | const NOTED = new Map<string, string[]>(); |
| 236 | const NOTED_JOBS = 500; |
| 237 | |
| 238 | async function noteOnce(deps: RelayDeps, job: string, token: string, name: string, host: string, text: string): Promise<void> { |
| 239 | const noted = newlyMisdirected(NOTED.get(job) ?? [], name, host); |
| 240 | if (!noted) return; |
| 241 | if (NOTED.size >= NOTED_JOBS && !NOTED.has(job)) NOTED.delete(NOTED.keys().next().value as string); |
| 242 | NOTED.set(job, noted.seen); |
| 243 | try { |
| 244 | await deps.note(job, token, text); |
| 245 | } catch (error) { |
| 246 | console.log("broker note not kept", job, String(error)); |
| 247 | } |
| 248 | } |
| 249 | |
| 250 | type Outcome<T> = { ok: true; value: T } | { ok: false; error: { message: string } }; |
| 251 | |
| 252 | /** The relay's needs, met by the actions service and `forward`. */ |
| 253 | export function relayDeps(actions: ServiceBinding, send: (request: Request) => Promise<Response>): RelayDeps { |
| 254 | const call = (method: string, body: unknown) => |
| 255 | actions.fetch(`https://actions/rpc/${method}`, { |
| 256 | method: "POST", |
| 257 | headers: { "content-type": "application/json" }, |
| 258 | body: JSON.stringify(body), |
| 259 | }); |
| 260 | return { |
| 261 | async secrets(job, token) { |
| 262 | const response = await call("relayed_secrets", { job, token }); |
| 263 | if (!response.ok) throw new Error(`relayed_secrets answered ${response.status}`); |
| 264 | const answer = (await response.json()) as Outcome<BrokeredSecret[]>; |
| 265 | return answer.ok ? answer.value : null; |
| 266 | }, |
| 267 | async note(job, token, text) { |
| 268 | const response = await call("job_report", { job, token, report: { kind: "note", text } }); |
| 269 | if (!response.ok) throw new Error(`job_report answered ${response.status}`); |
| 270 | }, |
| 271 | send, |
| 272 | }; |
| 273 | } |
| 274 | |
| 275 | /** Relays one broker request (the module's comment). Never throws for what a job sent. */ |
| 276 | export async function relay(request: Request, deps: RelayDeps): Promise<Response> { |
| 277 | const job = request.headers.get("x-g1t-broker-job") ?? ""; |
| 278 | const token = /^bearer\s+(\S+)$/i.exec(request.headers.get("authorization") ?? "")?.[1] ?? ""; |
| 279 | if (!job || !token) return refused(401, "unauthenticated", "A job and its token are required."); |
| 280 | const relayed = readRelayed(request.headers.get(REQUEST_HEADER)); |
| 281 | if (!relayed || !METHODS.has(relayed.method)) return refused(400, "invalid", "The request to relay does not read."); |
| 282 | let target: URL; |
| 283 | try { |
| 284 | target = new URL(relayed.url); |
| 285 | } catch { |
| 286 | return refused(400, "invalid", "The request to relay has no URL."); |
| 287 | } |
| 288 | const secrets = await deps.secrets(job, token); |
| 289 | if (!secrets) return refused(401, "unauthenticated", "That job is not running on a self-hosted runner, or the token is not its."); |
| 290 | |
| 291 | const headers = passedHeaders(relayed.headers); |
| 292 | const carried = new Set(standInsIn(target.toString(), headers)); |
| 293 | const { url, headers: filledHeaders, filled, misdirected } = fill(target.toString(), headers, secrets); |
| 294 | for (const secret of misdirected) { |
| 295 | await noteOnce(deps, job, token, secret.name, target.host, misdirectedNote(secret.name, target.host)); |
| 296 | } |
| 297 | if (filled.length === 0) { |
| 298 | if (misdirected.length === 0) await noteOnce(deps, job, token, "", target.host, refusedNote(target.host)); |
| 299 | const why = carried.size === 0 ? "it carries none of this job's stand-ins" : "its stand-ins are not for this host, or it is not HTTPS"; |
| 300 | return refused(403, "forbidden", `g1t relays only a request with a brokered secret's stand-in, over HTTPS, to that secret's hosts: ${why}.`); |
| 301 | } |
| 302 | |
| 303 | const body = relayed.method === "GET" || relayed.method === "HEAD" ? null : await readCapped(request.body, MAX_REQUEST_BYTES); |
| 304 | if (body === null && relayed.method !== "GET" && relayed.method !== "HEAD") { |
| 305 | return refused(413, "too_large", `A relayed request's body is ${MAX_REQUEST_BYTES / 1024 / 1024} MB at most.`); |
| 306 | } |
| 307 | const upstream = await deps.send( |
| 308 | new Request(url, { |
| 309 | method: relayed.method, |
| 310 | headers: filledHeaders, |
| 311 | body: body && body.byteLength > 0 ? body : null, |
| 312 | redirect: "manual", |
| 313 | signal: AbortSignal.timeout(TIME_LIMIT_MS), |
| 314 | }), |
| 315 | ); |
| 316 | const length = Number(upstream.headers.get("content-length") ?? NaN); |
| 317 | if (Number.isFinite(length) && length > MAX_ANSWER_BYTES) { |
| 318 | await upstream.body?.cancel().catch(() => undefined); |
| 319 | return refused(502, "too_large", `The answer is larger than g1t relays (${MAX_ANSWER_BYTES / 1024 / 1024} MB).`); |
| 320 | } |
| 321 | const answer = encode({ status: upstream.status, headers: answerHeaders(scrubHeaders(upstream.headers, filled)) }); |
| 322 | let stream = upstream.body?.pipeThrough(capped(MAX_ANSWER_BYTES)) ?? null; |
| 323 | if (stream && scrubbedType(upstream.headers)) stream = stream.pipeThrough(scrubStream(filled)); |
| 324 | return new Response(stream, { |
| 325 | status: 200, |
| 326 | headers: { [ANSWER_HEADER]: answer, "content-type": "application/octet-stream", "cache-control": "no-store" }, |
| 327 | }); |
| 328 | } |