Public: anyone can read this; the crew can change it.
Audit the streaming routes (/events, telemetry/*) for traces of a read-only mirror #10696
Check · Review · Land · Live done — Landed and live in production.
mason/forge:mason/10677-audit-the-streaming-routes-events-teleme → main 2 commits fork
1 file
changed +80
−1
test/origin-read-only-private.test.ts +80 −1
| @@ -12,7 +12,8 @@ import { join } from "node:path"; | ||
| 12 | 12 | import { createApp, type App } from "../src/app.ts"; |
| 13 | 13 | import { createSession } from "../src/core/sessions.ts"; |
| 14 | 14 | import { createAgentRow, type Agent } from "../src/core/agents.ts"; |
| 15 | import { createUser } from "../src/core/users.ts"; | |
| 15 | import { createUser, getUserByName } from "../src/core/users.ts"; | |
| 16 | import { setTelemetryGrant } from "../src/actions/telemetry.ts"; | |
| 16 | 17 | import { createIssue } from "../src/core/issues.ts"; |
| 17 | 18 | import { getRepo } from "../src/core/repos.ts"; |
| 18 | 19 | import { repoDir } from "../src/git/repos.ts"; |
| @@ -170,6 +171,10 @@ describe("a mirror of a read-only origin is private to its owner", () => { | ||
| 170 | 171 | "/api/v1/family", |
| 171 | 172 | "/api/v1/fleet/rows", |
| 172 | 173 | "/api/v1/telemetry/home", |
| 174 | "/api/v1/telemetry/system", | |
| 175 | "/api/v1/telemetry/flow", | |
| 176 | "/api/v1/telemetry/sources", | |
| 177 | "/api/v1/telemetry/history", | |
| 173 | 178 | "/apps", |
| 174 | 179 | "/crew", |
| 175 | 180 | "/map", |
| @@ -285,6 +290,80 @@ describe("a mirror of a read-only origin is private to its owner", () => { | ||
| 285 | 290 | expect(await found(`/api/v1/repos/alice/${NAME}/issues`)).toContain("zephyrissue"); |
| 286 | 291 | }); |
| 287 | 292 | |
| 293 | // The streaming routes: /events replays what a viewer may see and then goes | |
| 294 | // live; the telemetry reads are the owner's, or a granted agent's with | |
| 295 | // private repositories cut out. | |
| 296 | const telemetry = ["home", "system", "flow", "sources", "history"].map((p) => `/api/v1/telemetry/${p}`); | |
| 297 | /** | |
| 298 | * Read an event stream until `marker` has arrived, then hang up. Events are | |
| 299 | * ordered, so anything the stream would leak from before the marker is in | |
| 300 | * what comes back. `during` runs once the stream is open and registered. | |
| 301 | */ | |
| 302 | const listen = async (who: string, path: string, marker: string, during?: () => void): Promise<string> => { | |
| 303 | const stop = new AbortController(); | |
| 304 | const res = await fetch(`${servers.url}${path}`, { | |
| 305 | headers: cookies[who] === undefined ? {} : { cookie: cookies[who]! }, | |
| 306 | signal: stop.signal, | |
| 307 | }); | |
| 308 | const reader = res.body!.getReader(); | |
| 309 | during?.(); | |
| 310 | let text = ""; | |
| 311 | // Only a failure timeout: a healthy read ends the moment the marker shows. | |
| 312 | const timer = setTimeout(() => stop.abort(), 20_000); | |
| 313 | try { | |
| 314 | while (!text.includes(marker)) { | |
| 315 | const { done, value } = await reader.read(); | |
| 316 | if (done) break; | |
| 317 | text += new TextDecoder().decode(value); | |
| 318 | } | |
| 319 | } finally { | |
| 320 | clearTimeout(timer); | |
| 321 | stop.abort(); | |
| 322 | } | |
| 323 | return text; | |
| 324 | }; | |
| 325 | const traces = (text: string) => [NAME, "zephyrmarker", "zephyrissue", commit].filter((t) => text.includes(t)); | |
| 326 | ||
| 327 | for (const who of ["anonymous", "stranger", "agent"]) { | |
| 328 | test(`${who} reads the telemetry routes and finds no trace of the mirror`, async () => { | |
| 329 | for (const path of telemetry) expect(traces(await (await as(who, path)).text()), `${who} ${path}`).toEqual([]); | |
| 330 | }); | |
| 331 | ||
| 332 | test(`${who} replays and follows /events and finds no trace of the mirror`, async () => { | |
| 333 | const repo = getRepo(app.db, "alice", NAME)!; | |
| 334 | const open = getRepo(app.db, "alice", "openbox")!; | |
| 335 | const marker = `eventmarker-${who}`; | |
| 336 | const text = await listen(who, "/events?since=0", marker, () => { | |
| 337 | // Events of the mirror (naming it, as the stream does) and then a public one. | |
| 338 | app.bus.append({ kind: "issue.opened", repoId: repo.id, actorId: repo.ownerId, payload: { number: 1 } }); | |
| 339 | app.bus.append({ kind: "issue.opened", repoId: open.id, actorId: open.ownerId, payload: { number: 1, marker } }); | |
| 340 | }); | |
| 341 | // The marker is a public event: the read is real, and ends past the mirror's. | |
| 342 | expect(text).toContain(marker); | |
| 343 | expect(traces(text)).toEqual([]); | |
| 344 | }); | |
| 345 | } | |
| 346 | ||
| 347 | test("an agent granted telemetry:read still finds no trace of the mirror", async () => { | |
| 348 | const alice = getUserByName(app.db, "alice")!; | |
| 349 | expect(setTelemetryGrant(app, alice, "crewmate", true).ok).toBe(true); | |
| 350 | for (const path of telemetry) { | |
| 351 | const res = await as("agent", path); | |
| 352 | expect(res.status, path).toBe(200); | |
| 353 | expect(traces(await res.text()), path).toEqual([]); | |
| 354 | } | |
| 355 | }); | |
| 356 | ||
| 357 | test("the owner's /events replay carries the mirror's own events", async () => { | |
| 358 | expect(await listen("owner", "/events?since=0", NAME)).toContain(NAME); | |
| 359 | // And live: the stream names the mirror the moment an event of it happens. | |
| 360 | const repo = getRepo(app.db, "alice", NAME)!; | |
| 361 | const live = await listen("owner", "/events", NAME, () => | |
| 362 | app.bus.append({ kind: "issue.opened", repoId: repo.id, actorId: repo.ownerId, payload: { number: 2 } }), | |
| 363 | ); | |
| 364 | expect(live).toContain(`"repo":"alice/${NAME}"`); | |
| 365 | }); | |
| 366 | ||
| 288 | 367 | test("the owner finds it", async () => { |
| 289 | 368 | for (const path of ["/search?q=quillmo", "/api/v1/search?q=quillmo&kind=all", `/search?q=${NAME}`, `/api/v1/search?q=${NAME}&kind=all`, "/alice", `/alice/${NAME}`, `/api/v1/repos/alice/${NAME}`]) { |
| 290 | 369 | const res = await as("owner", path); |