| 1 | import http from "node:http"; |
| 2 | import https from "node:https"; |
| 3 | import net from "node:net"; |
| 4 | import tls from "node:tls"; |
| 5 | |
| 6 | /** |
| 7 | * Registry relay for hosted runners: an HTTP(S) proxy for the Docker daemon. |
| 8 | * |
| 9 | * Cloudflare refuses request bodies over 100 MB, and Docker sends each image layer in a |
| 10 | * single request, so a big layer could never reach the Diffinery registry directly. The |
| 11 | * Docker daemon on a hosted runner uses this relay as its HTTPS proxy. Connections to the |
| 12 | * Diffinery host are terminated here (the daemon trusts this relay's certificate for that |
| 13 | * host only, through /etc/docker/certs.d), and blob uploads are re-sent as a series of |
| 14 | * chunked PATCH requests with Content-Range, which the registry API supports. Everything |
| 15 | * else is passed through unchanged, and other hosts are tunnelled untouched. |
| 16 | */ |
| 17 | export interface RelayOptions { |
| 18 | /** Port to listen on (127.0.0.1). */ |
| 19 | port: number; |
| 20 | /** Host name the relay intercepts, e.g. diffinery.com. */ |
| 21 | host: string; |
| 22 | /** Where intercepted requests really go, e.g. https://diffinery.com. */ |
| 23 | upstream: string; |
| 24 | cert: string; |
| 25 | key: string; |
| 26 | /** Upload chunk size; must stay under Cloudflare's 100 MB limit. */ |
| 27 | chunkSize?: number; |
| 28 | log?: (line: string) => void; |
| 29 | } |
| 30 | |
| 31 | const HOP_BY_HOP = new Set(["connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "proxy-connection", "te", "trailer", "transfer-encoding", "upgrade", "host", "content-length"]); |
| 32 | const UPLOAD_PATH = /^\/v2\/.+\/blobs\/uploads\/[^/?]+(\?.*)?$/; |
| 33 | |
| 34 | export interface Relay { |
| 35 | port: number; |
| 36 | close(): Promise<void>; |
| 37 | } |
| 38 | |
| 39 | export async function startRelay(o: RelayOptions): Promise<Relay> { |
| 40 | const log = o.log ?? ((l: string) => process.stdout.write(`${l}\n`)); |
| 41 | const upstream = new URL(o.upstream); |
| 42 | const client = upstream.protocol === "https:" ? https : http; |
| 43 | const agent = new client.Agent({ keepAlive: true, maxSockets: 32 }); |
| 44 | const chunkSize = o.chunkSize ?? 64 * 1024 * 1024; |
| 45 | |
| 46 | /** One upstream request with a fully buffered (or empty) body. */ |
| 47 | function send(method: string, path: string, headers: http.OutgoingHttpHeaders, body?: Buffer): Promise<{ status: number; headers: http.IncomingHttpHeaders; body: Buffer }> { |
| 48 | return new Promise((resolve, reject) => { |
| 49 | const req = client.request({ protocol: upstream.protocol, hostname: upstream.hostname, port: upstream.port || undefined, method, path, headers: { ...headers, host: upstream.host, "content-length": String(body?.length ?? 0) }, agent }, (res) => { |
| 50 | const parts: Buffer[] = []; |
| 51 | res.on("data", (d: Buffer) => parts.push(d)); |
| 52 | res.on("end", () => resolve({ status: res.statusCode ?? 502, headers: res.headers, body: Buffer.concat(parts) })); |
| 53 | res.on("error", reject); |
| 54 | }); |
| 55 | req.on("error", reject); |
| 56 | req.end(body); |
| 57 | }); |
| 58 | } |
| 59 | |
| 60 | function forwardHeaders(req: http.IncomingMessage): http.OutgoingHttpHeaders { |
| 61 | const h: http.OutgoingHttpHeaders = {}; |
| 62 | for (const [k, v] of Object.entries(req.headers)) if (!HOP_BY_HOP.has(k) && v !== undefined) h[k] = v; |
| 63 | return h; |
| 64 | } |
| 65 | |
| 66 | /** Streams a request and its response straight through. */ |
| 67 | function passThrough(req: http.IncomingMessage, res: http.ServerResponse): void { |
| 68 | const headers = forwardHeaders(req); |
| 69 | if (req.headers["content-length"]) headers["content-length"] = req.headers["content-length"]; |
| 70 | const up = client.request({ protocol: upstream.protocol, hostname: upstream.hostname, port: upstream.port || undefined, method: req.method, path: req.url, headers: { ...headers, host: upstream.host }, agent }, (r) => { |
| 71 | const out: http.OutgoingHttpHeaders = {}; |
| 72 | for (const [k, v] of Object.entries(r.headers)) if (!HOP_BY_HOP.has(k) || k === "content-length") out[k] = v; |
| 73 | res.writeHead(r.statusCode ?? 502, out); |
| 74 | r.pipe(res); |
| 75 | }); |
| 76 | up.on("error", (e) => { |
| 77 | log(`relay: ${req.method} ${req.url} failed: ${e.message}`); |
| 78 | if (!res.headersSent) res.writeHead(502, { "content-type": "application/json" }); |
| 79 | res.end(JSON.stringify({ errors: [{ code: "UNKNOWN", message: `registry relay: ${e.message}` }] })); |
| 80 | }); |
| 81 | req.pipe(up); |
| 82 | } |
| 83 | |
| 84 | /** |
| 85 | * A blob upload (PATCH or PUT to an upload URL): sent on in chunks. Each full chunk |
| 86 | * goes as a PATCH with Content-Range; a PUT's remainder goes with the final PUT. |
| 87 | */ |
| 88 | async function chunkedUpload(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> { |
| 89 | const url = new URL(req.url!, "http://relay"); |
| 90 | const started = Date.now(); |
| 91 | let requests = 0; |
| 92 | const secs = (since: number) => `${((Date.now() - since) / 1000).toFixed(1)}s`; |
| 93 | const headers = forwardHeaders(req); |
| 94 | delete headers["content-range"]; |
| 95 | delete headers["content-type"]; |
| 96 | let location = url.pathname; |
| 97 | // Where the upload stands: earlier chunks may have come through another connection. |
| 98 | const status = await send("GET", location, { authorization: req.headers.authorization ?? "" }); |
| 99 | if (status.status !== 204) return reply(res, status); |
| 100 | let offset = uploadSize(status.headers); |
| 101 | const declaredStart = /^(?:bytes[ =])?(\d+)-/.exec(String(req.headers["content-range"] ?? "")); |
| 102 | if (declaredStart && Number(declaredStart[1]) !== offset) return reply(res, status, 416); |
| 103 | |
| 104 | // Pieces received since the last chunk went out. Joining them only when a chunk is full |
| 105 | // copies each byte once; growing one buffer per piece copied whole chunks thousands of times. |
| 106 | let pieces: Buffer[] = []; |
| 107 | let held = 0; |
| 108 | const take = (n: number): Buffer => { |
| 109 | const all = pieces.length === 1 ? pieces[0]! : Buffer.concat(pieces, held); |
| 110 | const rest = all.subarray(n); |
| 111 | pieces = rest.length ? [rest] : []; |
| 112 | held = rest.length; |
| 113 | return all.subarray(0, n); |
| 114 | }; |
| 115 | let last: Awaited<ReturnType<typeof send>> | null = null; |
| 116 | const flush = async (data: Buffer) => { |
| 117 | for (let attempt = 1; ; attempt++) { |
| 118 | try { |
| 119 | requests++; |
| 120 | const r = await send("PATCH", location, { ...headers, "content-type": "application/octet-stream", "content-range": `${offset}-${offset + data.length - 1}` }, data); |
| 121 | if (r.status !== 202) log(`relay: PATCH ${url.pathname} (${data.length} bytes at ${offset}) answered ${r.status} (attempt ${attempt})`); |
| 122 | if (r.status === 202) { |
| 123 | location = new URL(String(r.headers.location ?? location), "http://relay").pathname; |
| 124 | offset += data.length; |
| 125 | last = r; |
| 126 | return true; |
| 127 | } |
| 128 | if (r.status < 500 || attempt >= 3) { |
| 129 | last = r; |
| 130 | return false; |
| 131 | } |
| 132 | } catch (e) { |
| 133 | if (attempt >= 3) throw e; |
| 134 | } |
| 135 | // Retry from wherever the registry got to. |
| 136 | const s = await send("GET", location, { authorization: req.headers.authorization ?? "" }); |
| 137 | if (s.status === 204 && uploadSize(s.headers) === offset + data.length) { |
| 138 | offset += data.length; |
| 139 | last = s; |
| 140 | return true; |
| 141 | } |
| 142 | await new Promise((r) => setTimeout(r, 1000 * attempt)); |
| 143 | } |
| 144 | }; |
| 145 | for await (const piece of req as AsyncIterable<Buffer>) { |
| 146 | pieces.push(piece); |
| 147 | held += piece.length; |
| 148 | while (held >= chunkSize) if (!(await flush(take(chunkSize)))) return reply(res, last!); |
| 149 | } |
| 150 | const buf = take(held); |
| 151 | if (req.method === "PUT") { |
| 152 | const put = Date.now(); |
| 153 | const r = await send("PUT", `${location}${url.search}`, { ...headers, ...(buf.length ? { "content-type": "application/octet-stream", "content-range": `${offset}-${offset + buf.length - 1}` } : {}) }, buf); |
| 154 | log(`relay: upload ${url.pathname}: ${offset + buf.length} bytes in ${requests + 1} requests over ${secs(started)}; the final PUT answered ${r.status} after ${secs(put)}`); |
| 155 | return reply(res, r); |
| 156 | } |
| 157 | if (buf.length && !(await flush(buf))) return reply(res, last!); |
| 158 | if (!last) last = await send("GET", location, { authorization: req.headers.authorization ?? "" }); |
| 159 | // The client sent one PATCH; answer it like the registry would, covering everything so far. |
| 160 | return reply(res, { ...last, status: 202, headers: { ...last.headers, location, range: `0-${Math.max(0, offset - 1)}` } }, 202); |
| 161 | } |
| 162 | |
| 163 | function reply(res: http.ServerResponse, r: { status: number; headers: http.IncomingHttpHeaders; body: Buffer }, status = r.status): void { |
| 164 | const out: http.OutgoingHttpHeaders = {}; |
| 165 | for (const [k, v] of Object.entries(r.headers)) if (!HOP_BY_HOP.has(k)) out[k] = v; |
| 166 | out["content-length"] = String(r.body.length); |
| 167 | res.writeHead(status, out); |
| 168 | res.end(r.body); |
| 169 | } |
| 170 | |
| 171 | // Intercepted HTTPS traffic for the Diffinery host. |
| 172 | const inner = http.createServer((req, res) => { |
| 173 | const big = (req.method === "PATCH" || req.method === "PUT") && UPLOAD_PATH.test(req.url ?? ""); |
| 174 | if (!big) return passThrough(req, res); |
| 175 | chunkedUpload(req, res).catch((e: Error) => { |
| 176 | log(`relay: upload ${req.url} failed: ${e.message}`); |
| 177 | if (!res.headersSent) res.writeHead(502, { "content-type": "application/json" }); |
| 178 | res.end(JSON.stringify({ errors: [{ code: "BLOB_UPLOAD_INVALID", message: `registry relay: ${e.message}` }] })); |
| 179 | }); |
| 180 | }); |
| 181 | inner.requestTimeout = 0; |
| 182 | inner.headersTimeout = 120_000; |
| 183 | const secure = { cert: o.cert, key: o.key }; |
| 184 | |
| 185 | // The proxy itself: CONNECT tunnels, plus plain-HTTP proxying for completeness. |
| 186 | const proxy = http.createServer((req, res) => { |
| 187 | let target: URL; |
| 188 | try { |
| 189 | target = new URL(req.url ?? ""); |
| 190 | } catch { |
| 191 | res.writeHead(400).end("absolute URL required"); |
| 192 | return; |
| 193 | } |
| 194 | const up = http.request(target, { method: req.method, headers: { ...forwardHeaders(req), host: target.host, ...(req.headers["content-length"] ? { "content-length": req.headers["content-length"] } : {}) } }, (r) => { |
| 195 | res.writeHead(r.statusCode ?? 502, r.headers); |
| 196 | r.pipe(res); |
| 197 | }); |
| 198 | up.on("error", () => res.writeHead(502).end()); |
| 199 | req.pipe(up); |
| 200 | }); |
| 201 | proxy.requestTimeout = 0; |
| 202 | proxy.on("connect", (req: http.IncomingMessage, socket: net.Socket, head: Buffer) => { |
| 203 | const [host, portStr] = (req.url ?? "").split(":"); |
| 204 | const port = Number(portStr || 443); |
| 205 | socket.on("error", () => socket.destroy()); |
| 206 | if (host?.toLowerCase() === o.host.toLowerCase() && port === 443) { |
| 207 | socket.write("HTTP/1.1 200 Connection Established\r\n\r\n"); |
| 208 | if (head.length) socket.unshift(head); |
| 209 | const t = new tls.TLSSocket(socket, { isServer: true, ...secure, ALPNProtocols: ["http/1.1"] }); |
| 210 | t.on("error", () => t.destroy()); |
| 211 | inner.emit("connection", t); |
| 212 | return; |
| 213 | } |
| 214 | const out = net.connect(port, host!, () => { |
| 215 | socket.write("HTTP/1.1 200 Connection Established\r\n\r\n"); |
| 216 | if (head.length) out.write(head); |
| 217 | out.pipe(socket); |
| 218 | socket.pipe(out); |
| 219 | }); |
| 220 | out.on("error", () => { |
| 221 | if (socket.writable) socket.end("HTTP/1.1 502 Bad Gateway\r\n\r\n"); |
| 222 | socket.destroy(); |
| 223 | }); |
| 224 | }); |
| 225 | |
| 226 | await new Promise<void>((resolve, reject) => { |
| 227 | proxy.once("error", reject); |
| 228 | proxy.listen(o.port, "127.0.0.1", () => resolve()); |
| 229 | }); |
| 230 | const port = (proxy.address() as net.AddressInfo).port; |
| 231 | log(`registry relay for ${o.host} -> ${o.upstream} listening on 127.0.0.1:${port}`); |
| 232 | return { |
| 233 | port, |
| 234 | close: () => |
| 235 | new Promise<void>((resolve) => { |
| 236 | agent.destroy(); |
| 237 | inner.close(); |
| 238 | proxy.closeAllConnections(); |
| 239 | proxy.close(() => resolve()); |
| 240 | }), |
| 241 | }; |
| 242 | } |
| 243 | |
| 244 | /** Bytes the registry holds for an upload. */ |
| 245 | function uploadSize(h: http.IncomingHttpHeaders): number { |
| 246 | const exact = Number(h["diffinery-upload-size"]); |
| 247 | if (Number.isInteger(exact) && exact >= 0) return exact; |
| 248 | const m = /^(?:bytes=)?(\d+)-(\d+)$/.exec(String(h.range ?? "")); |
| 249 | return m && Number(m[2]) > 0 ? Number(m[2]) + 1 : 0; |
| 250 | } |