| 1 | import { execFileSync } from "node:child_process"; |
| 2 | import { createHash } from "node:crypto"; |
| 3 | import { readFileSync } from "node:fs"; |
| 4 | import http from "node:http"; |
| 5 | import net from "node:net"; |
| 6 | import path from "node:path"; |
| 7 | import tls from "node:tls"; |
| 8 | import { afterEach, beforeEach, describe, expect, it } from "vitest"; |
| 9 | import { startRelay, type Relay } from "../src/relay.js"; |
| 10 | import { tmp } from "./helpers.js"; |
| 11 | |
| 12 | /** |
| 13 | * The registry relay against a fake registry: blob uploads arrive in chunks no bigger |
| 14 | * than the limit, everything else passes through, and other hosts are tunnelled. |
| 15 | */ |
| 16 | |
| 17 | const HOST = "registry.test"; |
| 18 | const sha = (b: Buffer) => `sha256:${createHash("sha256").update(b).digest("hex")}`; |
| 19 | |
| 20 | interface Upload { |
| 21 | data: Buffer[]; |
| 22 | size: number; |
| 23 | } |
| 24 | |
| 25 | /** Just enough of the OCI upload API to see what the relay sends. */ |
| 26 | function fakeRegistry() { |
| 27 | const uploads = new Map<string, Upload>(); |
| 28 | const blobs = new Map<string, Buffer>(); |
| 29 | const requests: Array<{ method: string; url: string; length: number; range?: string }> = []; |
| 30 | let next = 0; |
| 31 | const server = http.createServer(async (req, res) => { |
| 32 | const parts: Buffer[] = []; |
| 33 | for await (const p of req) parts.push(p as Buffer); |
| 34 | const body = Buffer.concat(parts); |
| 35 | const url = new URL(req.url!, "http://x"); |
| 36 | requests.push({ method: req.method!, url: req.url!, length: body.length, range: req.headers["content-range"] as string | undefined }); |
| 37 | if (req.headers.authorization !== "Basic dTpw") return res.writeHead(401).end(); |
| 38 | const head = (u: Upload, id: string) => ({ location: `/v2/o/img/blobs/uploads/${id}`, range: `0-${Math.max(0, u.size - 1)}`, "diffinery-upload-size": String(u.size) }); |
| 39 | let m: RegExpExecArray | null; |
| 40 | if (req.method === "POST" && url.pathname === "/v2/o/img/blobs/uploads/") { |
| 41 | const id = `u${++next}`; |
| 42 | uploads.set(id, { data: [], size: 0 }); |
| 43 | return res.writeHead(202, head(uploads.get(id)!, id)).end(); |
| 44 | } |
| 45 | if ((m = /^\/v2\/o\/img\/blobs\/uploads\/(\w+)$/.exec(url.pathname))) { |
| 46 | const u = uploads.get(m[1]!); |
| 47 | if (!u) return res.writeHead(404).end(); |
| 48 | if (req.method === "GET") return res.writeHead(204, head(u, m[1]!)).end(); |
| 49 | const cr = req.headers["content-range"] as string | undefined; |
| 50 | if (cr && Number(cr.split("-")[0]) !== u.size) return res.writeHead(416, head(u, m[1]!)).end(); |
| 51 | if (body.length > 1024 * 1024) return res.writeHead(413).end("too large"); |
| 52 | u.data.push(body); |
| 53 | u.size += body.length; |
| 54 | if (req.method === "PATCH") return res.writeHead(202, head(u, m[1]!)).end(); |
| 55 | const all = Buffer.concat(u.data); |
| 56 | if (sha(all) !== url.searchParams.get("digest")) return res.writeHead(400).end("digest"); |
| 57 | blobs.set(sha(all), all); |
| 58 | return res.writeHead(201, { location: `/v2/o/img/blobs/${sha(all)}`, "docker-content-digest": sha(all) }).end(); |
| 59 | } |
| 60 | if ((m = /^\/v2\/o\/img\/blobs\/(sha256:\w+)$/.exec(url.pathname))) { |
| 61 | const b = blobs.get(m[1]!); |
| 62 | return b ? res.writeHead(200, { "content-length": String(b.length) }).end(b) : res.writeHead(404).end(); |
| 63 | } |
| 64 | res.writeHead(404).end(); |
| 65 | }); |
| 66 | return { server, uploads, blobs, requests }; |
| 67 | } |
| 68 | |
| 69 | let reg: ReturnType<typeof fakeRegistry>; |
| 70 | let relay: Relay; |
| 71 | let cert: string; |
| 72 | let logged: string[]; |
| 73 | |
| 74 | beforeEach(async () => { |
| 75 | const dir = tmp(); |
| 76 | execFileSync("openssl", ["req", "-x509", "-newkey", "ec", "-pkeyopt", "ec_paramgen_curve:prime256v1", "-nodes", "-days", "1", "-subj", `/CN=${HOST}`, "-addext", `subjectAltName=DNS:${HOST}`, "-keyout", path.join(dir, "k.pem"), "-out", path.join(dir, "c.pem")], { stdio: "ignore" }); |
| 77 | cert = readFileSync(path.join(dir, "c.pem"), "utf8"); |
| 78 | reg = fakeRegistry(); |
| 79 | await new Promise<void>((r) => reg.server.listen(0, "127.0.0.1", () => r())); |
| 80 | const upstream = `http://127.0.0.1:${(reg.server.address() as net.AddressInfo).port}`; |
| 81 | logged = []; |
| 82 | relay = await startRelay({ port: 0, host: HOST, upstream, cert, key: readFileSync(path.join(dir, "k.pem"), "utf8"), chunkSize: 1024 * 1024, log: (l) => void logged.push(l) }); |
| 83 | }); |
| 84 | |
| 85 | afterEach(async () => { |
| 86 | await relay.close(); |
| 87 | reg.server.closeAllConnections(); |
| 88 | await new Promise((r) => reg.server.close(r)); |
| 89 | }); |
| 90 | |
| 91 | /** An HTTPS request to the registry host through the relay, as the Docker daemon makes it. */ |
| 92 | function viaRelay(method: string, url: string, body?: Buffer, headers: Record<string, string> = {}): Promise<{ status: number; headers: http.IncomingHttpHeaders; body: Buffer }> { |
| 93 | return new Promise((resolve, reject) => { |
| 94 | const connect = http.request({ host: "127.0.0.1", port: relay.port, method: "CONNECT", path: `${HOST}:443` }); |
| 95 | connect.on("error", reject); |
| 96 | connect.on("connect", (_res, socket) => { |
| 97 | const secure = tls.connect({ socket, servername: HOST, ca: cert }); |
| 98 | const req = http.request({ method, path: url, headers: { host: HOST, authorization: "Basic dTpw", ...headers }, createConnection: () => secure }, (res) => { |
| 99 | const parts: Buffer[] = []; |
| 100 | res.on("data", (d: Buffer) => parts.push(d)); |
| 101 | res.on("end", () => { |
| 102 | secure.destroy(); |
| 103 | resolve({ status: res.statusCode!, headers: res.headers, body: Buffer.concat(parts) }); |
| 104 | }); |
| 105 | }); |
| 106 | req.on("error", reject); |
| 107 | // Stream the body in pieces, with no Content-Length, like docker does. |
| 108 | if (body) for (let i = 0; i < body.length; i += 300_000) req.write(body.subarray(i, i + 300_000)); |
| 109 | req.end(); |
| 110 | }); |
| 111 | connect.end(); |
| 112 | }); |
| 113 | } |
| 114 | |
| 115 | describe("registry relay", () => { |
| 116 | it("splits a single-PUT upload (containerd) into PATCH chunks under the limit", async () => { |
| 117 | const blob = Buffer.alloc(2_600_000, 7); |
| 118 | const start = await viaRelay("POST", "/v2/o/img/blobs/uploads/"); |
| 119 | expect(start.status).toBe(202); |
| 120 | const done = await viaRelay("PUT", `${start.headers.location}?digest=${sha(blob)}`, blob); |
| 121 | expect(done.status).toBe(201); |
| 122 | expect(done.headers["docker-content-digest"]).toBe(sha(blob)); |
| 123 | expect(reg.blobs.get(sha(blob))?.equals(blob)).toBe(true); |
| 124 | const sent = reg.requests.filter((r) => r.method === "PATCH" || r.method === "PUT"); |
| 125 | expect(sent.map((r) => [r.method, r.length])).toEqual([ |
| 126 | ["PATCH", 1048576], |
| 127 | ["PATCH", 1048576], |
| 128 | ["PUT", 502848], |
| 129 | ]); |
| 130 | expect(sent[1]!.range).toBe("1048576-2097151"); |
| 131 | // Its log (shown when a docker push fails) says how long the registry took to answer. |
| 132 | expect(logged.at(-1)).toMatch(/^relay: upload \/v2\/o\/img\/blobs\/uploads\/\S+: 2600000 bytes in 3 requests over [\d.]+s; the final PUT answered 201 after [\d.]+s$/); |
| 133 | }); |
| 134 | |
| 135 | it("answers a streamed PATCH (docker) like the registry would, then completes with an empty PUT", async () => { |
| 136 | const blob = Buffer.from(Array.from({ length: 2_300_000 }, (_, i) => i % 251)); |
| 137 | const start = await viaRelay("POST", "/v2/o/img/blobs/uploads/"); |
| 138 | const patch = await viaRelay("PATCH", String(start.headers.location), blob); |
| 139 | expect(patch.status).toBe(202); |
| 140 | expect(patch.headers.range).toBe(`0-${blob.length - 1}`); |
| 141 | expect(reg.requests.filter((r) => r.method === "PATCH").length).toBe(3); |
| 142 | const put = await viaRelay("PUT", `${patch.headers.location}?digest=${sha(blob)}`); |
| 143 | expect(put.status).toBe(201); |
| 144 | const back = await viaRelay("GET", `/v2/o/img/blobs/${sha(blob)}`); |
| 145 | expect(back.status).toBe(200); |
| 146 | expect(back.body.equals(blob)).toBe(true); |
| 147 | }); |
| 148 | |
| 149 | it("passes the registry's errors back", async () => { |
| 150 | const start = await viaRelay("POST", "/v2/o/img/blobs/uploads/"); |
| 151 | const bad = await viaRelay("PUT", `${start.headers.location}?digest=${sha(Buffer.from("other"))}`, Buffer.from("content")); |
| 152 | expect(bad.status).toBe(400); |
| 153 | expect((await viaRelay("GET", "/v2/o/img/blobs/uploads/nope")).status).toBe(404); |
| 154 | expect((await viaRelay("GET", "/v2/o/img/blobs/uploads/u1", undefined, { authorization: "Basic bm86bm8=" })).status).toBe(401); |
| 155 | }); |
| 156 | |
| 157 | it("tunnels other hosts untouched", async () => { |
| 158 | const echo = net.createServer((s) => s.pipe(s)); |
| 159 | await new Promise<void>((r) => echo.listen(0, "127.0.0.1", () => r())); |
| 160 | const port = (echo.address() as net.AddressInfo).port; |
| 161 | const reply = await new Promise<string>((resolve, reject) => { |
| 162 | const connect = http.request({ host: "127.0.0.1", port: relay.port, method: "CONNECT", path: `127.0.0.1:${port}` }); |
| 163 | connect.on("error", reject); |
| 164 | connect.on("connect", (res, socket) => { |
| 165 | expect(res.statusCode).toBe(200); |
| 166 | socket.once("data", (d) => { |
| 167 | resolve(d.toString()); |
| 168 | socket.destroy(); |
| 169 | }); |
| 170 | socket.write("ping"); |
| 171 | }); |
| 172 | connect.end(); |
| 173 | }); |
| 174 | expect(reply).toBe("ping"); |
| 175 | echo.close(); |
| 176 | }); |
| 177 | }); |