43cce38 · 10efbf8
@@ -26,7 +26,7 @@
import {
import type { Database } from "bun:sqlite";
import type { ForgeEvent } from "./events.ts";
import { getAppByRepo } from "./apps.ts";
import { canAdmin, canEnterApp, getRepoById } from "./repos.ts";
import { canAdmin, canEnterApp, canSeeOwnerFacts, getRepoById } from "./repos.ts";
import type { User } from "./users.ts";
export type EventAudience = "readers" | "admins" | "actor" | "app";
@@ -159,3 +159,34 @@
export function audienceFilter(
    }
  };
}
/** What a run event says about spend and placement: the owner's own facts. */
const RUN_OWNER_FIELDS = ["costUsd", "turns", "placement", "node"] as const;
/**
 * An event as this reader may see it. A `run.*` payload carries its cost,
 * turns and node, which are owner facts (`canSeeOwnerFacts`) like the same
 * fields on a run; the rest of the event is unchanged. Webhooks do not pass
 * through here: they go to admins and keep the full payload.
 */
export function eventForReader(
  db: Database,
  viewer: User | undefined,
): (event: ForgeEvent) => ForgeEvent {
  const owners = new Map<string, boolean>();
  const isOwner = (repoId: string): boolean => {
    const known = owners.get(repoId);
    if (known !== undefined) return known;
    const repo = getRepoById(db, repoId);
    const yes = repo !== undefined && canSeeOwnerFacts(db, repo, viewer);
    owners.set(repoId, yes);
    return yes;
  };
  return (event) => {
    if (!event.kind.startsWith("run.")) return event;
      
        sentinelcrew
        25d ago
        
      
      Good: a shallow copy, so the event on the bus and the copy webhooks get keep their full payload. If a nested payload is ever added to runPayload, this still holds because only top-level keys are deleted.
    
    if (event.repoId !== null && isOwner(event.repoId)) return event;
    const payload = { ...event.payload };
    for (const field of RUN_OWNER_FIELDS) delete payload[field];
    return { ...event, payload };
  };
}
@@ -165,7 +165,7 @@
import {
} from "../actions/users.ts";
import { listEvents } from "../core/events.ts";
import { audienceFor } from "../core/audience.ts";
import { audienceFilter } from "../core/event_audience.ts";
import { audienceFilter, eventForReader } from "../core/event_audience.ts";
import {
  forkParentFor,
  getReposByIds,
@@ -2279,7 +2279,7 @@
export function apiRoutes(router: Router<Handler>): void {
    return json(
      listEvents(ctx.app.db, { repoId: repo.id }).filter(
        audienceFilter(ctx.app.db, ctx.auth.user),
      ),
      ).map(eventForReader(ctx.app.db, ctx.auth.user)),
    );
  });
@@ -18,7 +18,7 @@
import { holdForCaller } from "./caller.ts";
import type { App } from "../app.ts";
import type { User } from "../core/users.ts";
import { listEvents, listEventsVisibleTo } from "../core/events.ts";
import { audienceFilter } from "../core/event_audience.ts";
import { audienceFilter, eventForReader } from "../core/event_audience.ts";
import { canRead, getRepoById } from "../core/repos.ts";
import { mayReadLogOfRun } from "../actions/runs.ts";
import { getRun } from "../core/runs.ts";
@@ -203,7 +203,12 @@
export function viewerStream(
    seen.set(repoId, ok);
    return ok;
  };
  const named = (event: ForgeEvent): Record<string, unknown> => {
  // Readable is not the same as meant-for: configuration events are the
  // admins', a person's own choices are theirs (`core/event_audience.ts`).
      
        sentinelcrew
        25d ago
        outdated
      
      redact is declared below this function (line 221). Declare it above named so the dependency reads top-down and named isn't one call away from a TDZ error.
    
  const meant = audienceFilter(app.db, viewer);
  const redact = eventForReader(app.db, viewer);
  const named = (raw: ForgeEvent): Record<string, unknown> => {
    const event = redact(raw);
    const repo =
      event.repoId === null ? undefined : getRepoById(app.db, event.repoId);
    return {
@@ -214,9 +219,6 @@
export function viewerStream(
      at: event.createdAt,
    };
  };
  // Readable is not the same as meant-for: configuration events are the
  // admins', a person's own choices are theirs (`core/event_audience.ts`).
  const meant = audienceFilter(app.db, viewer);
  const after = resumeFrom(request);
  return eventStream(request, (emit, _close, ready) => {
    if (after !== undefined) {
@@ -268,6 +270,7 @@
export function sseStream(
  viewer: User | undefined,
): Response {
  const meant = audienceFilter(app.db, viewer);
  const redact = eventForReader(app.db, viewer);
  const after = resumeFrom(request);
  return eventStream(request, (emit) => {
    // The event id doubles as the SSE id, so a reconnecting client resumes
@@ -278,10 +281,10 @@
export function sseStream(
        afterId: after,
        limit: 200,
      }).reverse())
        if (meant(event)) emit(event.kind, event, event.id);
        if (meant(event)) emit(event.kind, redact(event), event.id);
      
        sentinelcrew
        25d ago
        
      
      Untested: revert redact(event) to event here (replay of the per-repository stream) and nothing fails. Needs a stream test as a stranger with Last-Event-ID.
    
    return app.bus.subscribe((event: ForgeEvent) => {
      if (event.repoId === repoId && meant(event))
        emit(event.kind, event, event.id);
        emit(event.kind, redact(event), event.id);
      
        sentinelcrew
        25d ago
        
      
      Same for the live path: append a run.finished while a stranger is connected and assert that the frame has no cost, turns or node.
    
    });
  });
}
@@ -152,6 +152,37 @@
describe("who an event is for", () => {
    expect(admin).toContain("app.env_set");
  });
  test("a run event tells a stranger no cost, turns or node, and the owner all of it", async () => {
    app.bus.append({
      kind: "run.finished",
      repoId: repo.id,
      payload: {
        runId: "r1",
        state: "succeeded",
        costUsd: 1.25,
        turns: 7,
        placement: "job",
        node: "vx-secret-node",
      },
    });
    const read = (headers: Record<string, string>) =>
      fetch(`${base}/api/v1/repos/alice/open/events`, { headers })
        .then((r) => r.json() as Promise<{ kind: string; payload: Record<string, unknown> }[]>)
        .then((all) => all.find((e) => e.kind === "run.finished")!);
    for (const headers of [{}, { authorization: `Bearer ${bobToken}` }]) {
      const seen = await read(headers);
      expect(seen.payload["runId"]).toBe("r1");
      expect(seen.payload["state"]).toBe("succeeded");
      for (const f of ["costUsd", "turns", "placement", "node"])
        expect(f in seen.payload).toBe(false);
    }
    const owner = await read({ authorization: `Bearer ${aliceToken}` });
    expect(owner.payload["costUsd"]).toBe(1.25);
    expect(owner.payload["turns"]).toBe(7);
    expect(owner.payload["placement"]).toBe("job");
      
        sentinelcrew
        25d ago
        outdated
      
      Also assert owner.payload["placement"]).toBe("job"). It is one of the four fields, and the owner side never checks it.
    
    expect(owner.payload["node"]).toBe("vx-secret-node");
  });
  test("the activity rail on the front page says nothing of it to a visitor", async () => {
    const page = await fetch(`${base}/`, {
      headers: { accept: "text/html" },
@@ -217,4 +248,66 @@
describe("who an event is for", () => {
      expect(admin).toContain("hooks.example.test");
    }
  }, 20_000);
  test("the streams tell a stranger no run cost, turns or node, in replay and live, and the owner all of it", async () => {
    const run = (runId: string) =>
      app.bus.append({
        kind: "run.finished",
        repoId: repo.id,
        payload: {
          runId,
          state: "succeeded",
          costUsd: 3.5,
          turns: 9,
          placement: "job",
          node: "vx-stream-node",
        },
      });
    const before = listEvents(app.db, { repoId: repo.id, limit: 500 })
      .map((e) => e.id)
      .sort()[0]!;
    run("replayed-run");
    const read = async (path: string, headers: Record<string, string>) => {
      const controller = new AbortController();
      const res = await fetch(`${base}${path}`, {
        headers: {
          accept: "text/event-stream",
          "last-event-id": before,
          ...headers,
        },
        signal: controller.signal,
      });
      const reader = res.body!.getReader();
      let text = "";
      const pump = (async () => {
        try {
          for (;;) {
            const { value, done } = await reader.read();
            if (done) break;
            text += new TextDecoder().decode(value);
          }
        } catch {
          // aborted
        }
      })();
      await Bun.sleep(150);
      run("live-run");
      await Bun.sleep(300);
      controller.abort();
      await pump;
      return text;
    };
    for (const path of ["/alice/open/stream", "/events"]) {
      const stranger = await read(path, {});
      expect(stranger).toContain("replayed-run");
      expect(stranger).toContain("live-run");
      for (const f of ["costUsd", "turns", "placement", "vx-stream-node"])
        expect(stranger).not.toContain(f);
      const owner = await read(path, { authorization: `Bearer ${aliceToken}` });
      expect(owner).toContain("replayed-run");
      expect(owner).toContain("live-run");
      for (const f of ["costUsd", "turns", "placement", "vx-stream-node"])
        expect(owner).toContain(f);
    }
  }, 20_000);
});
		@@ -26,7 +26,7 @@ import {
26	26	import type { Database } from "bun:sqlite";
27	27	import type { ForgeEvent } from "./events.ts";
28	28	import { getAppByRepo } from "./apps.ts";
29		import { canAdmin, canEnterApp, getRepoById } from "./repos.ts";
	29	import { canAdmin, canEnterApp, canSeeOwnerFacts, getRepoById } from "./repos.ts";
30	30	import type { User } from "./users.ts";
31	31
32	32	export type EventAudience = "readers" \| "admins" \| "actor" \| "app";
		@@ -159,3 +159,34 @@ export function audienceFilter(
159	159	}
160	160	};
161	161	}
	162
	163	/** What a run event says about spend and placement: the owner's own facts. */
	164	const RUN_OWNER_FIELDS = ["costUsd", "turns", "placement", "node"] as const;
	165
	166	/**
	167	* An event as this reader may see it. A `run.*` payload carries its cost,
	168	* turns and node, which are owner facts (`canSeeOwnerFacts`) like the same
	169	* fields on a run; the rest of the event is unchanged. Webhooks do not pass
	170	* through here: they go to admins and keep the full payload.
	171	*/
	172	export function eventForReader(
	173	db: Database,
	174	viewer: User \| undefined,
	175	): (event: ForgeEvent) => ForgeEvent {
	176	const owners = new Map<string, boolean>();
	177	const isOwner = (repoId: string): boolean => {
	178	const known = owners.get(repoId);
	179	if (known !== undefined) return known;
	180	const repo = getRepoById(db, repoId);
	181	const yes = repo !== undefined && canSeeOwnerFacts(db, repo, viewer);
	182	owners.set(repoId, yes);
	183	return yes;
	184	};
	185	return (event) => {
	186	if (!event.kind.startsWith("run.")) return event;
		sentinelcrew 25d ago Good: a shallow copy, so the event on the bus and the copy webhooks get keep their full payload. If a nested payload is ever added to `runPayload`, this still holds because only top-level keys are deleted.
	187	if (event.repoId !== null && isOwner(event.repoId)) return event;
	188	const payload = { ...event.payload };
	189	for (const field of RUN_OWNER_FIELDS) delete payload[field];
	190	return { ...event, payload };
	191	};
	192	}
		@@ -165,7 +165,7 @@ import {
165	165	} from "../actions/users.ts";
166	166	import { listEvents } from "../core/events.ts";
167	167	import { audienceFor } from "../core/audience.ts";
168		import { audienceFilter } from "../core/event_audience.ts";
	168	import { audienceFilter, eventForReader } from "../core/event_audience.ts";
169	169	import {
170	170	forkParentFor,
171	171	getReposByIds,
		@@ -2279,7 +2279,7 @@ export function apiRoutes(router: Router<Handler>): void {
2279	2279	return json(
2280	2280	listEvents(ctx.app.db, { repoId: repo.id }).filter(
2281	2281	audienceFilter(ctx.app.db, ctx.auth.user),
2282		),
	2282	).map(eventForReader(ctx.app.db, ctx.auth.user)),
2283	2283	);
2284	2284	});
2285	2285
		@@ -18,7 +18,7 @@ import { holdForCaller } from "./caller.ts";
18	18	import type { App } from "../app.ts";
19	19	import type { User } from "../core/users.ts";
20	20	import { listEvents, listEventsVisibleTo } from "../core/events.ts";
21		import { audienceFilter } from "../core/event_audience.ts";
	21	import { audienceFilter, eventForReader } from "../core/event_audience.ts";
22	22	import { canRead, getRepoById } from "../core/repos.ts";
23	23	import { mayReadLogOfRun } from "../actions/runs.ts";
24	24	import { getRun } from "../core/runs.ts";
		@@ -203,7 +203,12 @@ export function viewerStream(
203	203	seen.set(repoId, ok);
204	204	return ok;
205	205	};
206		const named = (event: ForgeEvent): Record<string, unknown> => {
	206	// Readable is not the same as meant-for: configuration events are the
	207	// admins', a person's own choices are theirs (`core/event_audience.ts`).
		sentinelcrew 25d ago outdated `redact` is declared below this function (line 221). Declare it above `named` so the dependency reads top-down and `named` isn't one call away from a TDZ error.
	208	const meant = audienceFilter(app.db, viewer);
	209	const redact = eventForReader(app.db, viewer);
	210	const named = (raw: ForgeEvent): Record<string, unknown> => {
	211	const event = redact(raw);
207	212	const repo =
208	213	event.repoId === null ? undefined : getRepoById(app.db, event.repoId);
209	214	return {
		@@ -214,9 +219,6 @@ export function viewerStream(
214	219	at: event.createdAt,
215	220	};
216	221	};
217		// Readable is not the same as meant-for: configuration events are the
218		// admins', a person's own choices are theirs (`core/event_audience.ts`).
219		const meant = audienceFilter(app.db, viewer);
220	222	const after = resumeFrom(request);
221	223	return eventStream(request, (emit, _close, ready) => {
222	224	if (after !== undefined) {
		@@ -268,6 +270,7 @@ export function sseStream(
268	270	viewer: User \| undefined,
269	271	): Response {
270	272	const meant = audienceFilter(app.db, viewer);
	273	const redact = eventForReader(app.db, viewer);
271	274	const after = resumeFrom(request);
272	275	return eventStream(request, (emit) => {
273	276	// The event id doubles as the SSE id, so a reconnecting client resumes
		@@ -278,10 +281,10 @@ export function sseStream(
278	281	afterId: after,
279	282	limit: 200,
280	283	}).reverse())
281		if (meant(event)) emit(event.kind, event, event.id);
	284	if (meant(event)) emit(event.kind, redact(event), event.id);
		sentinelcrew 25d ago Untested: revert `redact(event)` to `event` here (replay of the per-repository stream) and nothing fails. Needs a stream test as a stranger with `Last-Event-ID`.
282	285	return app.bus.subscribe((event: ForgeEvent) => {
283	286	if (event.repoId === repoId && meant(event))
284		emit(event.kind, event, event.id);
	287	emit(event.kind, redact(event), event.id);
		sentinelcrew 25d ago Same for the live path: append a `run.finished` while a stranger is connected and assert that the frame has no cost, turns or node.
285	288	});
286	289	});
287	290	}
		@@ -152,6 +152,37 @@ describe("who an event is for", () => {
152	152	expect(admin).toContain("app.env_set");
153	153	});
154	154
	155	test("a run event tells a stranger no cost, turns or node, and the owner all of it", async () => {
	156	app.bus.append({
	157	kind: "run.finished",
	158	repoId: repo.id,
	159	payload: {
	160	runId: "r1",
	161	state: "succeeded",
	162	costUsd: 1.25,
	163	turns: 7,
	164	placement: "job",
	165	node: "vx-secret-node",
	166	},
	167	});
	168	const read = (headers: Record<string, string>) =>
	169	fetch(`${base}/api/v1/repos/alice/open/events`, { headers })
	170	.then((r) => r.json() as Promise<{ kind: string; payload: Record<string, unknown> }[]>)
	171	.then((all) => all.find((e) => e.kind === "run.finished")!);
	172	for (const headers of [{}, { authorization: `Bearer ${bobToken}` }]) {
	173	const seen = await read(headers);
	174	expect(seen.payload["runId"]).toBe("r1");
	175	expect(seen.payload["state"]).toBe("succeeded");
	176	for (const f of ["costUsd", "turns", "placement", "node"])
	177	expect(f in seen.payload).toBe(false);
	178	}
	179	const owner = await read({ authorization: `Bearer ${aliceToken}` });
	180	expect(owner.payload["costUsd"]).toBe(1.25);
	181	expect(owner.payload["turns"]).toBe(7);
	182	expect(owner.payload["placement"]).toBe("job");
		sentinelcrew 25d ago outdated Also assert `owner.payload["placement"]).toBe("job")`. It is one of the four fields, and the owner side never checks it.
	183	expect(owner.payload["node"]).toBe("vx-secret-node");
	184	});
	185
155	186	test("the activity rail on the front page says nothing of it to a visitor", async () => {
156	187	const page = await fetch(`${base}/`, {
157	188	headers: { accept: "text/html" },
		@@ -217,4 +248,66 @@ describe("who an event is for", () => {
217	248	expect(admin).toContain("hooks.example.test");
218	249	}
219	250	}, 20_000);
	251
	252	test("the streams tell a stranger no run cost, turns or node, in replay and live, and the owner all of it", async () => {
	253	const run = (runId: string) =>
	254	app.bus.append({
	255	kind: "run.finished",
	256	repoId: repo.id,
	257	payload: {
	258	runId,
	259	state: "succeeded",
	260	costUsd: 3.5,
	261	turns: 9,
	262	placement: "job",
	263	node: "vx-stream-node",
	264	},
	265	});
	266	const before = listEvents(app.db, { repoId: repo.id, limit: 500 })
	267	.map((e) => e.id)
	268	.sort()[0]!;
	269	run("replayed-run");
	270	const read = async (path: string, headers: Record<string, string>) => {
	271	const controller = new AbortController();
	272	const res = await fetch(`${base}${path}`, {
	273	headers: {
	274	accept: "text/event-stream",
	275	"last-event-id": before,
	276	...headers,
	277	},
	278	signal: controller.signal,
	279	});
	280	const reader = res.body!.getReader();
	281	let text = "";
	282	const pump = (async () => {
	283	try {
	284	for (;;) {
	285	const { value, done } = await reader.read();
	286	if (done) break;
	287	text += new TextDecoder().decode(value);
	288	}
	289	} catch {
	290	// aborted
	291	}
	292	})();
	293	await Bun.sleep(150);
	294	run("live-run");
	295	await Bun.sleep(300);
	296	controller.abort();
	297	await pump;
	298	return text;
	299	};
	300	for (const path of ["/alice/open/stream", "/events"]) {
	301	const stranger = await read(path, {});
	302	expect(stranger).toContain("replayed-run");
	303	expect(stranger).toContain("live-run");
	304	for (const f of ["costUsd", "turns", "placement", "vx-stream-node"])
	305	expect(stranger).not.toContain(f);
	306	const owner = await read(path, { authorization: `Bearer ${aliceToken}` });
	307	expect(owner).toContain("replayed-run");
	308	expect(owner).toContain("live-run");
	309	for (const f of ["costUsd", "turns", "placement", "vx-stream-node"])
	310	expect(owner).toContain(f);
	311	}
	312	}, 20_000);
220	313	});