diff --git a/docs/observers.md b/docs/observers.md index d6c5534c9..f3f5eedbc 100644 --- a/docs/observers.md +++ b/docs/observers.md @@ -275,10 +275,21 @@ Logic: **throws** on a mismatch. This server-side check is what guarantees a gatekeeper only receives a verifier minted by its own vendor; filtering account choices in the client is only a user-interface convenience. - - If any `addObserver` **throws** (or `getVerifier` throws on vendor mismatch), the user is not - (or no longer) allowed: best-effort `removeObserver(record.observerId)` on the gatekeepers - added in *this* pass, do **not** persist the working record, and deny the open with a clear - message. + - If any `addObserver` **throws** (or `getVerifier` throws on vendor mismatch, or returns null + for a disconnected account), the user is not (or no longer) allowed. Every such failure goes + through one `fail()` path that synchronously scrubs the failed gatekeeper from the + *persisted* record, so commit-time re-checks (`assertCollaboratorStillVerified`) fail closed + immediately, and the user is offered a bounded number of re-prompts to repair (e.g. + re-authenticate an expired account). On terminal failure the open is denied with a message + naming each refused binding, and the registrations are handled by which kind of verification + this was: a *first-ever* verification (no record at call start) best-effort-removes the + registrations it added or invalidated and persists no record — that collaborator was never + admitted and their minted id would otherwise linger unresolvable — while a *re-verification* + deliberately keeps them, because the registration is what preserves forward exclusion for + the collaborator's still-live sessions (coverage was already scrubbed, so nothing vouches + for them; the id keeps resolving so `excludeObservers` keeps naming them). One exception: if + a racing teardown deleted the record mid-call, the just-re-asserted registrations reference + an id no record resolves, so the full in-scope set is removed. 6. **Persist the observer record** (with merged `accountChoices` and `observerId`) only after all `addObserver` calls succeed. Storing/creating the record is the canonical moment the user @@ -329,7 +340,14 @@ observation proceed is when the named observer has *already lost access* in the For each id in `description.excludeObservers`: 1. Map the opaque `observerId` → `profileId` via the `observers.byObserverId` index. If there is - no record, the id is not an active observer → ignore it. + no record, the id is not an active observer → ignore it — **unless** the id belongs to a + first-time verification still in flight: `ensureObserver` registers a freshly minted id with + gatekeepers (`addObserver`) before the record is persisted, and that window spans awaits + (sibling verifier RPCs, even the configuration modal). The overseer tracks such ids in an + in-memory pending map and **blocks** an observation naming one (fail closed, with a distinct + "collaborator currently being verified" message) rather than reading it as unknown — otherwise + the observation would proceed and the collaborator be admitted moments later with the data + already in chat history. 2. Check sharing-graph reachability for that `profileId` (`SharingManager.getEffectiveRole` / `computeEffectiveRoles`). - **Still authorized → throw**, blocking the observation (degrade to per-observation @@ -340,6 +358,15 @@ For each id in `description.excludeObservers`: to observe; if they ever regain access they reconfigure from scratch (Step 3). 3. If, after evaluating all excluded ids, none are still-authorized, allow the observation. +Additionally, `authorizeObservation` fails **every** observation closed — not just those naming +excluded observers — whenever a revocation's restart is still pending +(`#revocationRestartPending`, set synchronously with the sever in `tearDownLostObservers`): the +DO abort that actually ends the removed user's live sessions runs only after the awaited teardown +and listing-refresh phases, and once the teardown's `removeObserver` fan-out de-registers the +observer, gatekeepers stop naming them in `excludeObservers` at all — so no per-gate check can +see the removed user, and only blocking everything covers their still-live sessions. See +`OverseerImpl.scheduleRevocationRestart`. + This is the runtime counterpart of `addObserver`: `addObserver` covers observers configured *after* data was read; `excludeObservers` covers data read *after* observers were configured. Persisting the observation record itself is unchanged; we only gate it. @@ -357,9 +384,12 @@ methods wrapping `SharingManager` mutations (`removeCollaborator`, `revokeShareL downgrades — see the matching methods on `OverseerClientInterface` and `SharingManager`): - After a mutation, use the returned `AffectedCollaborator[]` to find users who **lost access**. - For each who is now unreachable, if they have an observer record: best-effort - `removeObserver(record.observerId)` on **all** gatekeeper facets, then delete the observer - record. + For each who is now unreachable, if they have an observer record: delete the observer record, + then best-effort `removeObserver(record.observerId)` on **all** gatekeeper facets. Any + non-empty affected set also sets `#revocationRestartPending` synchronously with the sever, so + `authorizeObservation` fails every observation closed until the revocation restart disconnects + the affected users' still-live sessions (the awaited fan-out here is part of why that restart + is not immediate). - For a **`build` → `use` downgrade**, optionally `removeObserver` (and drop the corresponding `accountChoices` entries) for the now-out-of-scope bindings (those without a `bindingName`). Safe to defer — an over-broad observer set only ever errs toward stricter future checks — but @@ -406,7 +436,14 @@ already in the JSDoc in `gatekeeper.ts`; add anything missing there rather than can open at all (`overseer.ts:2770`). Observer checks only matter when sharing is allowed. 5. **Owner adds a new binding after sharing** — existing observers see an incremental modal for just the new binding on their next open, and may be denied if they lack access to the new - resource (inherent to the security model). + resource (inherent to the security model). Until that next open, their already-live sessions + watch the new connection's (non-restricted) observations unverified — the accepted residual of + verifying at open time. A connection added *while a collaborator's verification is parked* on + an await (the modal, verifier RPCs) is part of this same residual, not a bypass: the committed + record simply lacks an entry for it, and every consumer fails closed on absence — + `assertCollaboratorStillVerified` recomputes the in-scope set live against the persisted + record, and the next open re-verifies the uncovered binding — while their live session watches + like any other until re-open. 6. **Performance** — `ensureObserver` does one `getVerifier` + one `addObserver` per in-scope gatekeeper per open. Parallelize with `Promise.all` and pipe the verifier promise straight into `addObserver`. Expensive gatekeepers cache on their side. @@ -428,6 +465,9 @@ already in the JSDoc in `gatekeeper.ts`; add anything missing there rather than opens do not (record covers them) but still re-run `addObserver`. - a thrown `addObserver` denies the open and triggers best-effort `removeObserver` rollback on bindings added in the same pass, and does not persist the record. + - a failure against an *already-covered* binding scrubs that binding from the persisted record, + so commit-time re-checks (`assertCollaboratorStillVerified`) fail closed after a revocation + (edge case 3) instead of trusting coverage the live check just refused. - missing account → binding reported as a need to the callback; callback rejection denies open. - **`authorizeObservation` exclusion:** observation naming a still-authorized observer throws; observation naming an observer who lost access proceeds and deletes that observer record (+ diff --git a/packages/integration-tests/__tests__/external-message-verification.test.ts b/packages/integration-tests/__tests__/external-message-verification.test.ts new file mode 100644 index 000000000..66aa3eafa --- /dev/null +++ b/packages/integration-tests/__tests__/external-message-verification.test.ts @@ -0,0 +1,205 @@ +// Tests for the external-message authorization gate (authorizeCollaborator in overseer.ts): +// receiveExternalMessage() must hold a collaborator to the same observer verification open() +// applies -- non-interactively, since this path has no way to prompt for account configuration -- +// and must deny an insufficient role *before* verification runs. +// +// These live in their own file -- with their own harness, like every suite here -- so the suite +// stays self-contained as the observer suites around it grow. + +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import type { RpcStub } from "capnweb"; +import type { AuthenticatedApi, PublicApi } from "@gadgets/workshop-shared/api"; +import type { + SubmitExternalMessageResult, +} from "@gadgets/workshop-shared/external-message-gateway"; +import { + startTestGatekeeperHarness, TEST_GATEKEEPER_WORKER, TEST_VENDOR_ID, type Harness, +} from "../src/harness.js"; +import { + accountLabel, connect, listConnectedAccounts, MAX_OBSERVER_PROMPTS, nextUsernames, + ObserverConfigRecorder, signUp, stubFor, waitFor, type ConnectedAccount, +} from "../src/rpc-client.js"; +import { NetworkInterceptor } from "../src/network-interceptor.js"; + +// Reason text shaped like what a gatekeeper actually reports on a settled denial. Its appearance +// in the gateway's reply below is what proves a live verification round trip happened. +const DENIED_REASON = "You do not have access to this thing."; + +let harness: Harness; +let interceptor: NetworkInterceptor; + +beforeAll(async () => { + interceptor = new NetworkInterceptor(); + interceptor.install(); + harness = await startTestGatekeeperHarness(); +}); + +afterAll(async () => { + const unmocked = interceptor.getUnmockedCalls(); + await harness?.server.close(); + interceptor.uninstall(); + interceptor.reset(); + expect(unmocked).toEqual([]); +}); + +async function withSession(body: (api: RpcStub) => Promise): Promise { + const publicApi = connect(harness.url); + try { + return await body(publicApi); + } finally { + publicApi[Symbol.dispose](); + } +} + +function thingUrl(name: string): string { + return `https://gadgets-test.example/things/${name}`; +} + +async function provisionAccount(api: RpcStub): Promise { + await api.provisionAmbientAccount(TEST_VENDOR_ID); + return waitFor("the test account to be provisioned", async () => { + const accounts = await listConnectedAccounts(api); + return accounts.find(a => a.vendorId === TEST_VENDOR_ID) ?? null; + }); +} + +/** + * Submit an external chat message as `callerEmail`, through the fixture worker's control surface + * (and so through the Workshop's real ExternalMessageGateway entrypoint). + */ +async function submitExternalMessage(input: { + callerEmail: string; gadgetKey: string; prompt: string; +}): Promise { + const res = await harness.fetchWorker( + TEST_GATEKEEPER_WORKER, "http://gatekeeper-test.test/control/submit-external-message", + { method: "POST", body: JSON.stringify({ + chatKey: `chat-${input.gadgetKey}`, messageKey: crypto.randomUUID(), + gadgetTitle: input.gadgetKey, ...input }) }); + if (res.status !== 200) { + throw new Error(`submit-external-message failed with ${res.status}: ${await res.text()}`); + } + return await res.json() as SubmitExternalMessageResult; +} + +/** Tell the gatekeeper what to do the next time it's asked to admit `label` as an observer. */ +async function setVerifyOutcome( + label: string, outcome: { allow: true } | { allow: false; reason: string }): Promise { + const res = await harness.fetchWorker( + TEST_GATEKEEPER_WORKER, "http://gatekeeper-test.test/control/verify-outcome", + { method: "POST", body: JSON.stringify({ label, ...outcome }) }); + if (res.status !== 204) { + throw new Error(`Setting the verify outcome failed with ${res.status}: ${await res.text()}`); + } +} + +/** The workspace id behind an external gadgetKey -- the DO id the gateway derives from it. */ +async function externalGadgetId(gadgetKey: string): Promise { + const res = await harness.fetchWorker( + TEST_GATEKEEPER_WORKER, "http://gatekeeper-test.test/control/external-gadget-id", + { method: "POST", body: JSON.stringify({ gadgetKey }) }); + if (res.status !== 200) { + throw new Error(`external-gadget-id failed with ${res.status}: ${await res.text()}`); + } + return (await res.json() as { gadgetId: string }).gadgetId; +} + +describe("external-message verification", () => { + it.concurrent("the external-message path verifies collaborators like open() does", async () => { + await withSession(async publicApi => { + const [alice, bob, carol] = nextUsernames("alice", "bob", "carol"); + const aliceApi = await signUp(publicApi, alice); + const aliceAccount = await provisionAccount(aliceApi); + const gadgetKey = `external-${crypto.randomUUID()}`; + + // Alice creates the workspace through the external channel. No test user has an AI model, + // so a submission that passes the authorization gate is rejected with the model message -- + // which is what tells "passed the gate" apart from a gate denial below. + await expect(submitExternalMessage({ callerEmail: alice, gadgetKey, prompt: "hello" })) + .resolves.toMatchObject({ + accepted: false, message: expect.stringMatching(/AI model/i) }); + + // Wire the workspace up over the web API: connect a Thing (an account-requiring connection, + // so collaborators must be observer-verified against it) and add Bob. + const gadgetId = await externalGadgetId(gadgetKey); + using overseer = await aliceApi.openGadget(gadgetId); + const gatekeeper = await overseer.newGatekeeper(aliceAccount.id, thingUrl("external")); + if (!gatekeeper) throw new Error("Failed to create the test connection"); + + const bobApi = await signUp(publicApi, bob); + const bobAccount = await provisionAccount(bobApi); + await overseer.addCollaborator(bob, "build"); + + // A stranger is turned away by role, before verification is ever attempted. + await signUp(publicApi, carol); + await expect(submitExternalMessage({ callerEmail: carol, gadgetKey, prompt: "hi" })) + .resolves.toMatchObject({ + accepted: false, message: expect.stringMatching(/do not have access/i) }); + + // Bob has build access but has never opened, so he was never observer-verified -- and this + // path has no configuration channel to fix that. The agent's reply could surface anything + // the workspace has already read, so the external path must refuse him rather than fall + // through to the model check. + await expect(submitExternalMessage({ callerEmail: bob, gadgetKey, prompt: "hi" })) + .resolves.toMatchObject({ + accepted: false, message: expect.stringMatching(/could not be verified/i) }); + + // Opening the workspace verifies him; the same submission now passes the gate and fails + // only on the missing AI model, exactly like the owner's did. + const callback = stubFor( + new ObserverConfigRecorder().alwaysChoose(bobAccount.id, MAX_OBSERVER_PROMPTS)); + try { + (await bobApi.openGadget(gadgetId, undefined, callback))[Symbol.dispose](); + } finally { + callback[Symbol.dispose](); + } + await expect(submitExternalMessage({ callerEmail: bob, gadgetKey, prompt: "hi" })) + .resolves.toMatchObject({ + accepted: false, message: expect.stringMatching(/AI model/i) }); + + // The gatekeeper now revokes Bob's underlying access. His persisted observer record is + // untouched, so only a live addObserver re-verification on this submission can notice -- + // and the gatekeeper's own refusal reason appearing in the reply is the proof that round + // trip happened, since nothing persisted in the Workshop contains it. An implementation + // that merely checked the record would keep accepting him here. + await setVerifyOutcome(accountLabel(bobAccount), { allow: false, reason: DENIED_REASON }); + const revoked = await submitExternalMessage({ callerEmail: bob, gadgetKey, prompt: "hi" }); + if (revoked.accepted) throw new Error("The revoked submission was accepted"); + expect(revoked.message).toMatch(/could not be verified/i); + expect(revoked.message).toContain(DENIED_REASON); + }); + }); + + it.concurrent("the external-message path denies a use collaborator by role, not verification", + async () => { + await withSession(async publicApi => { + const [alice, dave] = nextUsernames("alice", "dave"); + const aliceApi = await signUp(publicApi, alice); + const aliceAccount = await provisionAccount(aliceApi); + const gadgetKey = `external-use-${crypto.randomUUID()}`; + + // Alice creates the workspace through the external channel (the AI-model rejection means + // her submission passed the gate), then binds its connection to a gadget so it falls in + // "use" verification scope. + await expect(submitExternalMessage({ callerEmail: alice, gadgetKey, prompt: "hello" })) + .resolves.toMatchObject({ + accepted: false, message: expect.stringMatching(/AI model/i) }); + const gadgetId = await externalGadgetId(gadgetKey); + using overseer = await aliceApi.openGadget(gadgetId); + const gatekeeper = await overseer.newGatekeeper(aliceAccount.id, thingUrl("external-use")); + if (!gatekeeper) throw new Error("Failed to create the test connection"); + using gadget = await overseer.createGadget("Test Gadget", undefined, "TEST_GADGET"); + await gadget.bind("TEST_THING", await gatekeeper.getId()); + + // Dave is in verification scope and unverified, but this path can never grant a "use" + // collaborator agent access, so his role is checked before verification runs: he gets the + // plain denial, not a verification failure he has no reason to go fix. + await signUp(publicApi, dave); + if (!await overseer.addCollaborator(dave, "use")) { + throw new Error(`Failed to share the gadget with ${dave}`); + } + await expect(submitExternalMessage({ callerEmail: dave, gadgetKey, prompt: "hi" })) + .resolves.toMatchObject({ + accepted: false, message: expect.stringMatching(/do not have access/i) }); + }); + }); +}); diff --git a/packages/integration-tests/fixtures/gatekeeper-test/src/env.d.ts b/packages/integration-tests/fixtures/gatekeeper-test/src/env.d.ts index b5b372a78..8a9715af0 100644 --- a/packages/integration-tests/fixtures/gatekeeper-test/src/env.d.ts +++ b/packages/integration-tests/fixtures/gatekeeper-test/src/env.d.ts @@ -12,6 +12,17 @@ declare namespace Cloudflare { // Storage classes exposed as DO namespaces on ctx.exports. durableNamespaces: "TestGatekeeper" | "TestControl"; } + + interface Env { + // The Workshop's external-message gateway entrypoint (see wrangler.jsonc). The contract + // interface is not entrypoint-branded (the shipping class implements it), so brand it here to + // satisfy Fetcher's constraint. + WORKSHOP_EXTERNAL_MESSAGES: Fetcher< + import("@gadgets/workshop-shared/external-message-gateway").ExternalMessageGateway & + Rpc.WorkerEntrypointBranded>; + // The Workshop's Overseer DO namespace (see wrangler.jsonc); used only to derive ids. + WORKSHOP_OVERSEER: DurableObjectNamespace; + } } interface ExecutionContext { diff --git a/packages/integration-tests/fixtures/gatekeeper-test/src/test-gatekeeper.ts b/packages/integration-tests/fixtures/gatekeeper-test/src/test-gatekeeper.ts index a279323c6..f9ffc75cc 100644 --- a/packages/integration-tests/fixtures/gatekeeper-test/src/test-gatekeeper.ts +++ b/packages/integration-tests/fixtures/gatekeeper-test/src/test-gatekeeper.ts @@ -20,12 +20,15 @@ // is one control knob here, `allow`, and the reason string is what carries the distinction to the // user. Tests exercise both narratives by choosing reason text. -import { DurableObject, WorkerEntrypoint, type RpcStub } from "cloudflare:workers"; +import { DurableObject, RpcTarget, WorkerEntrypoint, type RpcStub } from "cloudflare:workers"; import type { AccountDescription, ActionKind, ApprovalQueue, Gatekeeper, GatekeeperConnectCallback, GatekeeperUser, GatekeeperUserVerifier, ResourceDescription, ResourceConfiguratorFrame, SupportedResource, VendorDescription, } from "@gadgets/workshop-shared/gatekeeper"; +import type { + ChatGatewayRpcTarget, GadgetResponse, +} from "@gadgets/workshop-shared/external-message-gateway"; // Nothing but classes and the default handler may be exported from a Worker entry module: workerd // treats every named export as an entrypoint and rejects anything that isn't one. @@ -217,8 +220,44 @@ export class TestVerifier // --------------------------------------------------------------------------- // Gatekeeper (one per bound resource, running as a facet under the gadget's Overseer) -/** No operations: these tests never open a gadget's session, only verify observers. */ -export type TestSession = Record; +/** + * A live session against a Test Thing, opened via `GatekeeperClient.openSession()`. + * + * The two methods exist so tests can drive the overseer's observation/action policy through the + * same `ApprovalQueue` funnel a shipping gatekeeper uses: `readThing()` records an observation, + * and `doThing()` submits an action. + */ +export class TestSession extends RpcTarget { + #queue: RpcStub; + #title: string; + + constructor(queue: RpcStub, title: string) { + super(); + this.#queue = queue; + this.#title = title; + } + + async readThing(): Promise { + await this.#queue.authorizeObservation({ + title: `Read ${this.#title}`, + description: `The test read ${this.#title}.`, + }); + return `the contents of ${this.#title}`; + } + + async doThing(): Promise { + await this.#queue.submitAction(0, { + title: `Poke ${this.#title}`, + description: `The test poked ${this.#title}.`, + implementsRevert: false, + }); + } + + /** The session owns the queue stub dup'd in startSession(); release it with the session. */ + [Symbol.dispose]() { + this.#queue[Symbol.dispose](); + } +} export class TestGatekeeper extends DurableObject implements Gatekeeper { @@ -251,8 +290,15 @@ export class TestGatekeeper return []; } - async startSession(_approvalQueue: RpcStub): Promise { - return {}; + async startSession(approvalQueue: RpcStub): Promise { + // The session calls the queue after startSession() returns, so it owns a duplicate. + let queue = approvalQueue.dup(); + try { + return new TestSession(queue, (await this.describe()).title); + } catch (err) { + queue[Symbol.dispose]?.(); + throw err; + } } /** @@ -309,8 +355,16 @@ function isNonEmptyString(value: unknown): value is string { return typeof value === "string" && value.length > 0; } +/** + * Discards Gadget responses. The control endpoint below only asserts on the submission result, + * and the rejection paths under test return before any response is produced. + */ +class DevNullChatGateway extends RpcTarget implements ChatGatewayRpcTarget { + async onGadgetResponse(_response: GadgetResponse): Promise {} +} + export default { - async fetch(req: Request, _env: Cloudflare.Env, ctx: ExecutionContext): Promise { + async fetch(req: Request, env: Cloudflare.Env, ctx: ExecutionContext): Promise { const url = new URL(req.url); let body: unknown; @@ -348,6 +402,38 @@ export default { return Response.json({ count: await control(ctx.exports).getAmbientVerificationCount(label) }); } + // Submit an external chat message through the Workshop's ExternalMessageGateway entrypoint, + // the way a chat-integration worker would, so tests can drive receiveExternalMessage(). + // Body: {"callerEmail", "gadgetKey", "chatKey", "messageKey", "gadgetTitle", "prompt"} + // -> SubmitExternalMessageResult + if (url.pathname === "/control/submit-external-message" && req.method === "POST") { + const fields = + ["callerEmail", "gadgetKey", "chatKey", "messageKey", "gadgetTitle", "prompt"] as const; + const input = {} as Record<(typeof fields)[number], string>; + for (const field of fields) { + const value = (body as Record)[field]; + if (!isNonEmptyString(value)) return badRequest(`\`${field}\` must be a non-empty string`); + input[field] = value; + } + // The instance becomes a stub when it crosses the RPC boundary; the parameter type can only + // name the stub side of that. + const chatGatewayRpcTarget = + new DevNullChatGateway() as unknown as RpcStub; + return Response.json(await env.WORKSHOP_EXTERNAL_MESSAGES.submitExternalMessage( + { ...input, chatGatewayRpcTarget })); + } + + // Map an external gadgetKey to the Overseer id the gateway targets -- the DO named + // ":", where "test" is the `source` prop on WORKSHOP_EXTERNAL_MESSAGES -- + // so a test can open the same workspace over the web API, which addresses by DO id string. + // Body: {"gadgetKey": "..."} -> {"gadgetId": "..."} + if (url.pathname === "/control/external-gadget-id" && req.method === "POST") { + const { gadgetKey } = body as Record; + if (!isNonEmptyString(gadgetKey)) return badRequest("`gadgetKey` must be a non-empty string"); + return Response.json( + { gadgetId: env.WORKSHOP_OVERSEER.idFromName(`test:${gadgetKey}`).toString() }); + } + // Make this Worker issue a subrequest, so a test can prove that Worker-originated fetches really // do route through the interceptor rather than out to the internet. // diff --git a/packages/integration-tests/fixtures/gatekeeper-test/wrangler.jsonc b/packages/integration-tests/fixtures/gatekeeper-test/wrangler.jsonc index d4ca3ff2d..96167a8a4 100644 --- a/packages/integration-tests/fixtures/gatekeeper-test/wrangler.jsonc +++ b/packages/integration-tests/fixtures/gatekeeper-test/wrangler.jsonc @@ -12,7 +12,32 @@ "compatibility_date": "2026-02-02", "compatibility_flags": ["experimental", "allow_irrevocable_stub_storage"], - // DO classes are reached via ctx.exports; no durable_objects binding needed. + // Lets the control surface submit external chat messages through the Workshop's gateway + // entrypoint the way a real chat-integration worker (bound with its own `source` prop) would. + // The harness always boots workshop-backend as the primary worker, so the name resolves. + "services": [ + { + "binding": "WORKSHOP_EXTERNAL_MESSAGES", + "service": "workshop-backend", + "entrypoint": "ExternalMessageGateway", + "props": { "source": "test" } + } + ], + + // The Workshop's Overseer namespace, so the control surface can derive the DO id behind an + // external gadgetKey -- the same name-derived id the gateway targets -- for tests to open the + // workspace over the web API. The binding only derives ids; it never reaches an instance. + "durable_objects": { + "bindings": [ + { + "name": "WORKSHOP_OVERSEER", + "class_name": "OverseerDurableObject", + "script_name": "workshop-backend" + } + ] + }, + + // This worker's own DO classes are reached via ctx.exports; no durable_objects binding needed. "migrations": [ { "tag": "v0", diff --git a/packages/integration-tests/src/harness.ts b/packages/integration-tests/src/harness.ts index 199facdeb..da1df1514 100644 --- a/packages/integration-tests/src/harness.ts +++ b/packages/integration-tests/src/harness.ts @@ -76,6 +76,12 @@ function readWorkerConfig(dir: string): WorkerConfig { const config = parsed.data; config.build = { ...config.build, cwd: dir }; config.main = join(dir, config.main); + + // Local-dev var files (.dev.vars/.env at the harness root) must not leak into tests: a + // developer's local settings (say CF_AI_GATEWAY_*) would make suites behave differently on + // their machine than in CI -- up to sending real AI traffic. Declaring an empty required-secrets + // list makes wrangler exclude every such key that is not already a config var. + config.secrets = { required: [] }; return config; } diff --git a/packages/integration-tests/src/rpc-client.ts b/packages/integration-tests/src/rpc-client.ts index 83ec8da96..68911da2a 100644 --- a/packages/integration-tests/src/rpc-client.ts +++ b/packages/integration-tests/src/rpc-client.ts @@ -77,6 +77,14 @@ export async function signUp( return (await api.authenticate(token)) as unknown as RpcStub; } +/** Log back into an account created by signUp(), e.g. from a fresh connection. */ +export async function logIn( + api: RpcStub, username: string): Promise> { + const token = await api.login(username, passwordHashFor(username)); + if (!token) throw new Error(`Login failed for "${username}"`); + return (await api.authenticate(token)) as unknown as RpcStub; +} + export type ConnectedAccount = { id: number; vendorId: string; @@ -138,10 +146,12 @@ export const MAX_OBSERVER_PROMPTS = 2; */ export class ObserverConfigRecorder extends RpcTarget implements ObserverConfigCallback { readonly calls: ObserverBindingNeed[][] = []; - #responses: ((needs: ObserverBindingNeed[]) => ObserverAccountChoice[])[] = []; + #responses: ((needs: ObserverBindingNeed[]) + => ObserverAccountChoice[] | Promise)[] = []; /** Queue one response. The nth configure() call is answered by the nth queued responder. */ - respondWith(responder: (needs: ObserverBindingNeed[]) => ObserverAccountChoice[]): this { + respondWith(responder: (needs: ObserverBindingNeed[]) + => ObserverAccountChoice[] | Promise): this { this.#responses.push(responder); return this; } diff --git a/packages/workshop-backend/__tests__/external-message-staleness.test.ts b/packages/workshop-backend/__tests__/external-message-staleness.test.ts new file mode 100644 index 000000000..5f02cef50 --- /dev/null +++ b/packages/workshop-backend/__tests__/external-message-staleness.test.ts @@ -0,0 +1,247 @@ +// receiveExternalMessage's entry gate (authorizeCollaborator) is separated from the prompt +// commit by real await windows -- the owner's registration, the caller's context RPC, message +// preparation -- in which a concurrent verification's fail() can scrub the caller's coverage, +// a sharing change can sever the caller's role (tearing down their observer record), or a new +// connection can widen the scope they were never verified against. +// The agent then runs over the unfiltered chat tail and its reply leaves the Workshop, so the +// authorization must be re-asserted synchronously with *every* write the submission justifies: +// newChat runs the registration's assertStillAuthorized as the first statement of the +// transaction that writes the prompt, and sendChatMessage runs it just before +// materializeChatChanges (its first write, which has non-transactional side effects and so cannot +// move inside the transaction; no awaits separate the check from the transaction). A stale caller +// therefore commits nothing -- no chat, no message (not even a materialized "changes" one), no +// response target, no agent turn. +// +// Runs against a real OverseerDurableObject (the TEST_OVERSEER binding, like +// observer-serialization.test.ts); the gatekeeper facet and the caller's User DO are the fakes. +// A unit test rather than an integration test because the staleness must land deterministically +// inside the context-RPC window, which a fake getExternalMessageChatContext controls exactly. + +import { describe, expect, it } from "vitest"; +import { env, RpcStub as NativeRpcStub } from "cloudflare:workers"; +import { runInDurableObject } from "cloudflare:test"; +import type { OverseerDurableObject } from "../src/overseer.js"; + +declare module "cloudflare:workers" { + interface ProvidedEnv { + TEST_OVERSEER: DurableObjectNamespace; + } +} + +const OWNER = "alice"; + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + let promise = new Promise(r => { resolve = r; }); + return { promise, resolve }; +} + +function seedGatekeeper(impl: any, id: number): void { + impl.storage.gatekeepers.put({ + id, + resourceTitle: `Connection ${id}`, + class: {} as any, + creationSpec: { + type: "gatekeeper", + vendorId: "testvendor", + resourceUrl: `https://example.com/${id}`, + typeUrlPattern: "https://*", + }, + }); +} + +// Seeds an owned workspace with connection 1 and bob as a confirmed "build" collaborator holding +// a covering observer record, and fakes bob's User DO so the context RPC parks until the test +// releases it -- the window under test. startAgent is a spy: the assertion target is that it +// never runs for a denied submission. registerExternalMessageResponseTarget is a spy too, since +// the real one persists the gateway stub into DO storage and a stub minted inside the test +// context is not storable ("RpcStub cannot be serialized in this context") -- what matters here +// is whether it ran at all, and it runs inside the same transaction the re-check aborts. +function setup(instance: OverseerDurableObject): { + impl: any; + startAgentCalls: number; + registrations: number; + releaseContext: () => void; +} { + let impl = (instance as unknown as { impl: any }).impl; + impl.ownerProfileId = OWNER; + impl.ownerId = "owner-do-id"; + seedGatekeeper(impl, 1); + impl.storage.collaborators.put({ + profile: { id: "bob", name: "Bob" }, + addedBy: [{ type: "user", sharer: OWNER, created: new Date(), role: "build" }], + }); + impl.storage.observers.put( + { profileId: "bob", observerId: "obs-b", accountChoices: { 1: 10 } }); + impl.getGatekeeperFacet = () => ({ addObserver: async () => {} }); + + let state = { impl, startAgentCalls: 0, registrations: 0, releaseContext: () => {} }; + impl.startAgent = () => { state.startAgentCalls++; }; + impl.registerExternalMessageResponseTarget = () => { state.registrations++; }; + + let held = deferred(); + state.releaseContext = held.resolve; + let fakeCaller = { + id: { toString: () => "caller-do-id" }, + whoamiIfExists: async () => ({ type: "user", id: "bob", name: "Bob" }), + getVerifier: async () => ({}), + getExternalMessageChatContext: async () => { + await held.promise; + return { + profile: { type: "user", id: "bob", name: "Bob" }, + aiModel: { + profile: { type: "agent", id: "test-model", name: "Test Model" }, + config: { provider: "anthropic" }, + }, + }; + }, + }; + impl.users = { + getByName: () => fakeCaller, + // The best-effort lastActive bump resolves the owner's DO through these; give it an inert + // target so the control case doesn't log a spurious bump failure. + idFromString: (id: string) => id, + get: () => ({ setGadgetLastActive: async () => {} }), + }; + return state; +} + +function submitInput(key: string) { + return { + callerEmail: "bob@example.com", + externalChatKey: `ext-${key}`, + idempotencyKey: `idem-${key}`, + prompt: "Hello agent", + chatGatewayRpcTarget: new NativeRpcStub({ deliverResponse: async () => {} }) as any, + title: "My Workspace", + }; +} + +const tick = () => new Promise(resolve => setTimeout(resolve, 0)); + +function expectNothingCommitted( + state: { impl: any; startAgentCalls: number; registrations: number }, key: string): void { + expect([...state.impl.storage.chatMeta.list()]).toHaveLength(0); + expect([...state.impl.storage.chats.list()]).toHaveLength(0); + expect(state.registrations).toBe(0); + expect(state.impl.storage.externalChats.get(`ext-${key}`)).toBeUndefined(); + expect(state.startAgentCalls).toBe(0); +} + +describe("receiveExternalMessage's commit-time authorization re-check", () => { + it("denies when a concurrent verification failure scrubbed coverage mid-flight", async () => { + let stub = env.TEST_OVERSEER.getByName("external-staleness-coverage-scrub"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let state = setup(instance); + let result = instance.receiveExternalMessage(submitInput("scrub")); + await tick(); + + // A sibling verification's fail() scrubs the failed gatekeeper from the persisted record, + // synchronously, exactly as ensureObserver does. + let record = state.impl.storage.observers.get("bob"); + delete record.accountChoices[1]; + state.impl.storage.observers.put(record); + + state.releaseContext(); + await expect(result).resolves.toMatchObject({ accepted: false }); + expect((await result).message).toMatch(/could not be verified/); + // The transaction aborted before anything landed, and the agent never started. + expectNothingCommitted(state, "scrub"); + }); + }); + + it("denies when the caller's role was severed mid-flight", async () => { + let stub = env.TEST_OVERSEER.getByName("external-staleness-role-severed"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let state = setup(instance); + let result = instance.receiveExternalMessage(submitInput("severed")); + await tick(); + + state.impl.storage.collaborators.delete("bob"); + + state.releaseContext(); + await expect(result).resolves.toMatchObject({ accepted: false }); + expect((await result).message).toMatch(/could not be verified/); + expectNothingCommitted(state, "severed"); + }); + }); + + it("denies when a connection added mid-flight widened the unverified scope", async () => { + let stub = env.TEST_OVERSEER.getByName("external-staleness-scope-widened"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let state = setup(instance); + let result = instance.receiveExternalMessage(submitInput("widened")); + await tick(); + + // The re-check recomputes the scope live, so the new connection -- which bob was never + // verified against -- fails closed. + seedGatekeeper(state.impl, 2); + + state.releaseContext(); + await expect(result).resolves.toMatchObject({ accepted: false }); + expect((await result).message).toMatch(/could not be verified/); + expectNothingCommitted(state, "widened"); + }); + }); + + it("denies the existing-chat path without committing the message", async () => { + let stub = env.TEST_OVERSEER.getByName("external-staleness-existing-chat"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let state = setup(instance); + // A prior conversation already exists for this external chat key -- with a live change row, + // so that sendChatMessage's materializeChatChanges has something to write. Without the row + // this case would pass vacuously: materialization no-ops, which would hide it running + // *before* the authorization re-check and durably writing a "changes" message, retiring the + // row, and setting hasProposedChanges on a denied submission. + let started = new Date(); + state.impl.storage.chatMeta.put( + { id: 7, title: "Existing chat", started, lastActive: started }); + state.impl.storage.externalChats.put({ externalChatKey: "ext-existing", chatId: 7 }); + state.impl.storage.chatChanges.put({ + chatId: 7, generation: 0, revision: 1, timestamp: started, + author: { type: "user", id: "bob", name: "Bob" }, + change: {}, source: "user", + }); + + let result = instance.receiveExternalMessage(submitInput("existing")); + await tick(); + + let record = state.impl.storage.observers.get("bob"); + delete record.accountChoices[1]; + state.impl.storage.observers.put(record); + + state.releaseContext(); + await expect(result).resolves.toMatchObject({ accepted: false }); + expect((await result).message).toMatch(/could not be verified/); + // The chat survives but gained no message -- not even a materialized "changes" message -- + // no response target, no agent turn; the change row is still live and the chat's meta + // untouched. + expect([...state.impl.storage.chats.list()]).toHaveLength(0); + let rows = [...state.impl.storage.chatChanges.list()]; + expect(rows).toHaveLength(1); + expect(rows[0].retired).toBeUndefined(); + expect(state.impl.storage.chatMeta.get(7).hasProposedChanges).toBeUndefined(); + expect(state.registrations).toBe(0); + expect(state.startAgentCalls).toBe(0); + }); + }); + + it("accepts and commits when authorization stays intact", async () => { + let stub = env.TEST_OVERSEER.getByName("external-staleness-control"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let state = setup(instance); + let result = instance.receiveExternalMessage(submitInput("ok")); + await tick(); + + state.releaseContext(); + let outcome = await result; + expect(outcome.accepted).toBe(true); + + let messages = [...state.impl.storage.chats.list()]; + expect(messages).toHaveLength(1); + expect(messages[0].message).toBe("Hello agent"); + expect(state.registrations).toBe(1); + expect(state.startAgentCalls).toBe(1); + }); + }); +}); diff --git a/packages/workshop-backend/__tests__/keyless-commit-gate.test.ts b/packages/workshop-backend/__tests__/keyless-commit-gate.test.ts new file mode 100644 index 000000000..7b323e988 --- /dev/null +++ b/packages/workshop-backend/__tests__/keyless-commit-gate.test.ts @@ -0,0 +1,217 @@ +// authorizeCollaborator's commit gate must re-check the caller's live role for *every* +// verification. A keyless open parks in ensureObserver across real await windows (verifier RPCs, +// even the configuration modal), and a removal landing there used to be caught only by the +// revocation restart -- which fires after the teardown/listing phases, so a verification +// resolving inside that window slipped through: step 6's blind observers.put *resurrected* the +// record tearDownLostObservers had just deleted (record and account choices were loaded +// pre-park), leaving a removed user with coverage that a later re-grant would trust without +// re-verification, and the open returned a full stale-role capability besides. The gate now +// denies at commit time, and the post-verification role re-derivation caps a mid-park downgrade +// at the live role. +// +// Runs against a real OverseerDurableObject (the TEST_OVERSEER binding, like +// observer-serialization.test.ts); the gatekeeper facet and the client's User DO are the fakes. +// Bob is a *returning* collaborator (persisted covering record), so the open parks inside step +// 5's addObserver -- no configuration modal is involved. +// +// The first describe drives authorizeCollaborator directly. The second drives the production +// open() entry point, which carries the same gate inline (it cannot use authorizeCollaborator +// yet -- see that method's doc comment): a keyless open() must deny a mid-park removal without +// resurrecting the record, and a mid-park downgrade must hand back the restricted "use" +// capability rather than the full interface the stale role selected. + +import { describe, expect, it } from "vitest"; +import { env, RpcStub as NativeRpcStub } from "cloudflare:workers"; +import { runInDurableObject } from "cloudflare:test"; +import type { OverseerDurableObject } from "../src/overseer.js"; + +declare module "cloudflare:workers" { + interface ProvidedEnv { + TEST_OVERSEER: DurableObjectNamespace; + } +} + +const OWNER = "alice"; + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + let promise = new Promise(r => { resolve = r; }); + return { promise, resolve }; +} + +const tick = () => new Promise(resolve => setTimeout(resolve, 0)); + +// Seeds bob as a confirmed "build" collaborator with a covering observer record and parks the +// gatekeeper facet's addObserver on a deferred. +function seedParkedReturningBob(instance: OverseerDurableObject): { + impl: any; + release: () => void; + events: string[]; +} { + let impl = (instance as unknown as { impl: any }).impl; + impl.ownerProfileId = OWNER; + impl.storage.gatekeepers.put({ + id: 1, + resourceTitle: "Connection 1", + class: {} as any, + creationSpec: { + type: "gatekeeper", + vendorId: "testvendor", + resourceUrl: "https://example.com/1", + typeUrlPattern: "https://*", + }, + }); + impl.storage.collaborators.put({ + profile: { id: "bob", name: "Bob" }, + addedBy: [{ type: "user", sharer: OWNER, created: new Date(), role: "build" }], + }); + impl.storage.observers.put( + { profileId: "bob", observerId: "obs-b", accountChoices: { 1: 10 } }); + + // Ordered log of the fake gatekeeper's calls: an addObserver entry is recorded when the call + // *completes* (after the park), so the log shows whether a rollback's removeObserver could + // have lost to a still-in-flight registration. + let events: string[] = []; + let held = deferred(); + impl.getGatekeeperFacet = () => ({ + addObserver: async (id: string) => { await held.promise; events.push(`add:${id}`); }, + removeObserver: async (id: string) => { events.push(`remove:${id}`); }, + }); + return { impl, release: held.resolve, events }; +} + +// Starts bob's keyless open through authorizeCollaborator directly. +function startParkedKeylessOpen(instance: OverseerDurableObject): { + impl: any; + open: Promise; + release: () => void; + events: string[]; +} { + let { impl, release, events } = seedParkedReturningBob(instance); + let open = impl.authorizeCollaborator("bob", { getVerifier: async () => ({}) } as any, {}); + return { impl, open, release, events }; +} + +// Starts bob's keyless open through the production open() entry point. The parts of open() +// that would cross workers are faked: the caller's User DO (whoami/getVerifier/listing writes), +// ambient capsule reconciliation, and the session fan-outs the returned capability joins. +function startParkedProductionOpen(instance: OverseerDurableObject): { + impl: any; + open: Promise; + release: () => void; +} { + let { impl, release } = seedParkedReturningBob(instance); + impl.ownerId = "owner-do-id"; + impl.users = { + idFromString: (id: string) => id, + get: () => ({ + whoami: async () => ({ id: "bob", name: "Bob" }), + getVerifier: async () => ({}), + recordSharedGadgetOpen: async () => {}, + }), + }; + impl.ensureAmbientCapsules = async () => {}; + impl.syncOutputsTo = async () => {}; + impl.joinPresence = () => () => {}; + impl.joinOutputsFanout = () => () => {}; + + let notifyClosed = new NativeRpcStub<() => void>(() => {}); + let open = instance.open("bob-user-id", "bob", notifyClosed); + return { impl, open, release }; +} + +describe("the commit gate on keyless opens", () => { + it("denies a mid-verification removal and does not resurrect the torn-down record", + async () => { + let stub = env.TEST_OVERSEER.getByName("keyless-commit-gate-removal"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let { impl, open, release, events } = startParkedKeylessOpen(instance); + await tick(); + + // The owner removes bob while his re-verification is parked: the sever and the observer + // teardown run exactly as removeCollaborator drives them. The revocation restart would + // eventually kill this DO, but the parked verification can resolve before it lands. + let record = impl.storage.collaborators.get("bob"); + impl.storage.collaborators.delete("bob"); + await impl.tearDownLostObservers( + [{ profile: record.profile, addedBy: record.addedBy, oldRole: "build", newRole: null }]); + expect(impl.storage.observers.get("bob")).toBeUndefined(); + + release(); + // Pre-fix the keyless path had no commit gate: the open resolved "build"... + await expect(open).rejects.toThrow(/revoked while it was being verified/); + // ...and step 6's put resurrected the record the teardown just deleted, coverage a later + // re-grant would then trust without re-verification. + expect(impl.storage.observers.get("bob")).toBeUndefined(); + // The teardown's removeObserver ran while the re-assertion was still parked, so on its own + // it left the re-asserted registration behind, orphaned (no record resolves obs-b anymore). + // The rollback must issue another removeObserver *after* the parked addObserver completed. + expect(events).toEqual(["remove:obs-b", "add:obs-b", "remove:obs-b"]); + }); + }); + + it("caps a mid-verification downgrade at the live role", async () => { + let stub = env.TEST_OVERSEER.getByName("keyless-commit-gate-downgrade"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let { impl, open, release } = startParkedKeylessOpen(instance); + await tick(); + + // The owner downgrades bob's edge to "use" while his verification is parked. The gate + // passes (his role is still non-null, and verification at the pre-park "build" scope + // covers the narrower "use" scope), but the capability handed out must be the live role. + let record = impl.storage.collaborators.get("bob"); + record.addedBy[0].role = "use"; + impl.storage.collaborators.put(record); + + release(); + await expect(open).resolves.toBe("use"); + expect(impl.storage.observers.get("bob")).toBeDefined(); + }); + }); +}); + +describe("the commit gate on production open()", () => { + it("denies a mid-verification removal and does not resurrect the torn-down record", + async () => { + let stub = env.TEST_OVERSEER.getByName("production-open-gate-removal"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let { impl, open, release } = startParkedProductionOpen(instance); + open.catch(() => {}); // asserted below; don't let the park window reject unhandled + await tick(); + + // The owner removes bob while his re-verification is parked, exactly as removeCollaborator + // drives it. Pre-fix, open() ran ensureObserver with no commit gate at all. + let record = impl.storage.collaborators.get("bob"); + impl.storage.collaborators.delete("bob"); + await impl.tearDownLostObservers( + [{ profile: record.profile, addedBy: record.addedBy, oldRole: "build", newRole: null }]); + expect(impl.storage.observers.get("bob")).toBeUndefined(); + + release(); + await expect(open).rejects.toThrow(/revoked while it was being verified/); + // Step 6's put must not have resurrected the record the teardown just deleted. + expect(impl.storage.observers.get("bob")).toBeUndefined(); + }); + }); + + it("hands a mid-verification downgrade the restricted capability", async () => { + let stub = env.TEST_OVERSEER.getByName("production-open-gate-downgrade"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let { impl, open, release } = startParkedProductionOpen(instance); + await tick(); + + // The owner downgrades bob to "use" while his verification is parked. The gate passes + // (his role is still non-null), but the capability selected must follow the live role: + // pre-fix the stale "build" yielded the full OverseerClientInterface. + let record = impl.storage.collaborators.get("bob"); + record.addedBy[0].role = "use"; + impl.storage.collaborators.put(record); + + release(); + let client = await open; + expect(client.constructor.name).toBe("UseOverseerInterface"); + expect(impl.storage.observers.get("bob")).toBeDefined(); + client[Symbol.dispose](); + }); + }); +}); diff --git a/packages/workshop-backend/__tests__/observer-scope-prune.test.ts b/packages/workshop-backend/__tests__/observer-scope-prune.test.ts new file mode 100644 index 000000000..11810f938 --- /dev/null +++ b/packages/workshop-backend/__tests__/observer-scope-prune.test.ts @@ -0,0 +1,123 @@ +// ensureObserver must prune out-of-scope account choices from the observer record at every open, +// restoring the invariant commit-time re-checks (assertCollaboratorStillVerified) rest on: "entry +// present => verified at this collaborator's most recent open". Without the prune, a "use" +// collaborator opening while a connection is unbound from every gadget verifies nothing against +// it, yet their stale entry survives; rebinding the connection keeps the same gatekeeper id (only +// gadget binding edges change), so the re-check would trust coverage that the collaborator's +// opens during the unbound window never re-verified. +// +// Runs against a real OverseerDurableObject (the TEST_OVERSEER binding, like +// observer-serialization.test.ts) so ensureObserver's storage is real; the gatekeeper facet and +// the client's User DO are the only fakes. + +import { describe, expect, it } from "vitest"; +import { env } from "cloudflare:workers"; +import { runInDurableObject } from "cloudflare:test"; +import type { OverseerDurableObject } from "../src/overseer.js"; + +declare module "cloudflare:workers" { + interface ProvidedEnv { + TEST_OVERSEER: DurableObjectNamespace; + } +} + +function seedGatekeepers(impl: any): void { + for (let id of [1, 2]) { + impl.storage.gatekeepers.put({ + id, + resourceTitle: `Connection ${id}`, + class: {} as any, + creationSpec: { + type: "gatekeeper", + vendorId: "testvendor", + resourceUrl: `https://example.com/${id}`, + typeUrlPattern: "https://*", + }, + }); + } +} + +// A gadget that binds only gatekeeper 1, leaving gatekeeper 2 out of "use" scope. +function seedGadgetBindingGk1(impl: any): void { + impl.storage.gadgets.put({ + id: 100, + title: "G", + created: new Date(), + bindingName: "G", + bindings: { DB: { target: 1 } }, + }); +} + +// A client User DO that always has the account and always mints a verifier. +const fakeClientUser = { + getVerifier: async () => ({}), +} as any; + +describe("ensureObserver out-of-scope coverage pruning", () => { + it("prunes an unbound gatekeeper's entry at a use-role open", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-scope-prune-use"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + seedGadgetBindingGk1(impl); + impl.ownerProfileId = "owner"; + impl.storage.observers.put( + { profileId: "alice", observerId: "obs-1", accountChoices: { 1: 10, 2: 20 } }); + + let verified: number[] = []; + impl.getGatekeeperFacet = (id: number) => ({ + addObserver: async () => { verified.push(id); }, + }); + + await impl.ensureObserver("alice", fakeClientUser, "use"); + + // Gatekeeper 2 is outside "use" scope: its stale entry is gone, and nothing re-verified it. + expect(verified).toEqual([1]); + expect(impl.storage.observers.get("alice").accountChoices).toEqual({ 1: 10 }); + }); + }); + + it("prunes everything at an empty-scope open, keeping the record", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-scope-prune-empty"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + // No gadgets at all: a "use" collaborator's verification scope is empty. + impl.ownerProfileId = "owner"; + impl.storage.observers.put( + { profileId: "alice", observerId: "obs-1", accountChoices: { 1: 10, 2: 20 } }); + + // No configureCb: the open must still resolve (nothing in scope to configure), and it must + // still prune -- this is exactly the everything-unbound open the fix exists for. + await impl.ensureObserver("alice", fakeClientUser, "use"); + + let record = impl.storage.observers.get("alice"); + expect(record).toBeDefined(); + expect(record.accountChoices).toEqual({}); + }); + }); + + it("keeps unbound gatekeepers' entries at a build-role open", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-scope-prune-build"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + seedGadgetBindingGk1(impl); + impl.ownerProfileId = "owner"; + impl.storage.observers.put( + { profileId: "alice", observerId: "obs-1", accountChoices: { 1: 10, 2: 20 } }); + + let verified: number[] = []; + impl.getGatekeeperFacet = (id: number) => ({ + addObserver: async () => { verified.push(id); }, + }); + + // "build" scope is every account-requiring gatekeeper regardless of gadget bindings, so + // both entries are in scope and nothing may be pruned (guards against over-pruning). + await impl.ensureObserver("alice", fakeClientUser, "build"); + + expect(verified.toSorted()).toEqual([1, 2]); + expect(impl.storage.observers.get("alice").accountChoices).toEqual({ 1: 10, 2: 20 }); + }); + }); +}); diff --git a/packages/workshop-backend/__tests__/observer-serialization.test.ts b/packages/workshop-backend/__tests__/observer-serialization.test.ts new file mode 100644 index 000000000..31656fe9a --- /dev/null +++ b/packages/workshop-backend/__tests__/observer-serialization.test.ts @@ -0,0 +1,475 @@ +// ensureObserver must serialize per profile: its body awaits verifier RPCs and the configuration +// modal (unbounded), and DO input gates don't cover those awaits, so two concurrent opens for one +// profile would otherwise interleave -- most visibly, two concurrent *first* opens would each mint +// their own observerId and register both with the gatekeepers, while the last-written record +// forgets the other id ever existed (leaving it registered but unremovable). +// +// Runs against a real OverseerDurableObject (the TEST_OVERSEER binding, like +// git-migration-do.test.ts) so ensureObserver's private state is real; the gatekeeper facet and +// the client's User DO are the only fakes. + +import { describe, expect, it } from "vitest"; +import { env } from "cloudflare:workers"; +import { runInDurableObject } from "cloudflare:test"; +import type { OverseerDurableObject } from "../src/overseer.js"; + +declare module "cloudflare:workers" { + interface ProvidedEnv { + TEST_OVERSEER: DurableObjectNamespace; + } +} + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + let promise = new Promise(r => { resolve = r; }); + return { promise, resolve }; +} + +const tick = () => new Promise(resolve => setTimeout(resolve, 0)); + +function seedGatekeepers(impl: any): void { + for (let id of [1, 2]) { + impl.storage.gatekeepers.put({ + id, + resourceTitle: `Connection ${id}`, + class: {} as any, + creationSpec: { + type: "gatekeeper", + vendorId: "testvendor", + resourceUrl: `https://example.com/${id}`, + typeUrlPattern: "https://*", + }, + }); + } +} + +// A client User DO that always has the account and always mints a verifier. +const fakeClientUser = { + getVerifier: async () => ({}), +} as any; + +describe("ensureObserver per-profile serialization", () => { + it("gives two concurrent first opens one shared observerId", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-serialization-first-opens"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + + let registered: string[] = []; + impl.getGatekeeperFacet = () => ({ + addObserver: async (observerId: string) => { registered.push(observerId); }, + }); + + // Open A parks inside the configuration modal -- the unbounded window the serialization + // exists for -- while open B arrives with its own (competing) account choices. + let held = deferred(); + let configureA = { + configure: async () => { + await held.promise; + return [{ gatekeeperId: 1, accountId: 10 }, { gatekeeperId: 2, accountId: 20 }]; + }, + } as any; + let configureB = { + configure: async () => + [{ gatekeeperId: 1, accountId: 11 }, { gatekeeperId: 2, accountId: 21 }], + } as any; + + let openA = impl.ensureObserver("alice", fakeClientUser, "build", configureA); + await tick(); + let openB = impl.ensureObserver("alice", fakeClientUser, "build", configureB); + await tick(); + + // B must not have verified anything while A is still parked in its modal. + expect(registered).toHaveLength(0); + + held.resolve(); + await Promise.all([openA, openB]); + + // A registered both gatekeepers, then B re-verified both -- all under one id, which is the + // id the persisted record carries. Without serialization, B minted a second id while A was + // parked, and whichever record was written last orphaned the other id inside the + // gatekeepers. + expect(registered).toHaveLength(4); + expect(new Set(registered).size).toBe(1); + + let record = impl.storage.observers.get("alice"); + expect(record.observerId).toBe(registered[0]); + // B found A's committed record and re-verified A's choices rather than asking again. + expect(record.accountChoices).toEqual({ 1: 10, 2: 20 }); + }); + }); + + it("runs a queued open normally after the open ahead of it rejects", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-serialization-rejected-predecessor"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + + let registered: string[] = []; + impl.getGatekeeperFacet = () => ({ + addObserver: async (observerId: string) => { registered.push(observerId); }, + removeObserver: async () => {}, + }); + + // Open A parks in its modal, then the user cancels -- configure throws, so A registers + // nothing and persists nothing. Open B is already queued behind it: the chain must hand + // over to B anyway (release() runs in a finally and the link is a promise that never + // rejects), not stay poisoned or deadlocked by A's failure. + let held = deferred(); + let configureA = { + configure: async () => { + await held.promise; + throw new Error("cancelled"); + }, + } as any; + let configureB = { + configure: async () => + [{ gatekeeperId: 1, accountId: 11 }, { gatekeeperId: 2, accountId: 21 }], + } as any; + + let openA = impl.ensureObserver("alice", fakeClientUser, "build", configureA); + await tick(); + let openB = impl.ensureObserver("alice", fakeClientUser, "build", configureB); + await tick(); + + // B is still parked behind A; nothing has been verified yet. + expect(registered).toHaveLength(0); + + held.resolve(); + await expect(openA).rejects.toThrow(); + await openB; + + // B ran as an ordinary first open: one fresh id, both gatekeepers registered under it, and + // the persisted record carries B's own choices (A never committed any). + expect(registered).toHaveLength(2); + expect(new Set(registered).size).toBe(1); + let record = impl.storage.observers.get("alice"); + expect(record.observerId).toBe(registered[0]); + expect(record.accountChoices).toEqual({ 1: 11, 2: 21 }); + }); + }); + + it("a failed check's coverage scrub survives a concurrent open", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-serialization-scrub"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + // Already-configured coverage for both gatekeepers, as a previous successful open left it. + impl.storage.observers.put( + { profileId: "alice", observerId: "obs-1", accountChoices: { 1: 10, 2: 20 } }); + + // Gatekeeper 1's first re-verification (open A's) parks, then succeeds; its second (open + // B's) refuses -- the provider revoked access between the two. + let held = deferred(); + let gk1Calls = 0; + impl.getGatekeeperFacet = (id: number) => ({ + addObserver: async () => { + if (id === 1 && ++gk1Calls === 2) throw new Error("access revoked upstream"); + if (id === 1) await held.promise; + }, + removeObserver: async () => {}, + }); + + let openA = impl.ensureObserver("alice", fakeClientUser, "build"); + await tick(); + // B's re-prompt offer is declined, as a client with no way to repair would. + let openB = impl.ensureObserver("alice", fakeClientUser, "build", { + configure: async () => { throw new Error("cancelled"); }, + } as any); + await tick(); + held.resolve(); + + await expect(openA).resolves.toBeUndefined(); + await expect(openB).rejects.toThrow(); + + // B's failure scrubbed gatekeeper 1 from persisted coverage, and A's success -- which ran + // strictly before B under the per-profile lock -- cannot have resurrected it. Without the + // lock, A's final put lands after B's scrub and restores coverage the live check just + // refused, which assertCollaboratorStillVerified would then trust. + let record = impl.storage.observers.get("alice"); + expect(1 in record.accountChoices).toBe(false); + expect(record.accountChoices[2]).toBe(20); + }); + }); + + it("a getVerifier rejection scrubs that gatekeeper's persisted coverage", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-serialization-getverifier-rejection"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + // Already-configured coverage for both gatekeepers, as a previous successful open left it. + impl.storage.observers.put( + { profileId: "alice", observerId: "obs-1", accountChoices: { 1: 10, 2: 20 } }); + + let removed: number[] = []; + impl.getGatekeeperFacet = (id: number) => ({ + addObserver: async () => {}, + removeObserver: async () => { removed.push(id); }, + }); + + // Gatekeeper 1's verifier never materializes: the client's User DO *rejects* (the + // deterministic vendor-mismatch throw, or any cross-worker transport failure) rather than + // returning null. + let failingClientUser = { + getVerifier: async (accountId: number) => { + if (accountId === 10) throw new Error("account is for a different vendor"); + return {}; + }, + describeConnectedAccount: async () => null, + } as any; + + // No repair channel, so the failure is terminal -- and descriptive, not the raw RPC error. + await expect(impl.ensureObserver("alice", failingClientUser, "build")) + .rejects.toThrow(/could not confirm/); + + // The rejection went through fail(): gatekeeper 1's persisted coverage is scrubbed -- so + // assertCollaboratorStillVerified stops admitting this collaborator's older live sessions' + // external-message writes -- while gatekeeper 2's survives. The gatekeeper-side + // registration is deliberately kept (this was a re-verification of an admitted observer, + // not a first open): it preserves forward exclusion for alice's still-live sessions, and + // the next successful open's addObserver overwrites it. + let record = impl.storage.observers.get("alice"); + expect(1 in record.accountChoices).toBe(false); + expect(record.accountChoices[2]).toBe(20); + expect(removed).toEqual([]); + }); + }); + + it("a failed re-verification keeps an admitted observer's gatekeeper registrations", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-serialization-reverify-keeps"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + // Alice was admitted by a previous successful open: her record covers both gatekeepers, + // and (implicitly) her sessions may still be live. + impl.storage.observers.put( + { profileId: "alice", observerId: "obs-1", accountChoices: { 1: 10, 2: 20 } }); + + let removed: number[] = []; + impl.getGatekeeperFacet = (id: number) => ({ + addObserver: async () => { + if (id === 1) throw new Error("access revoked upstream"); + }, + removeObserver: async () => { removed.push(id); }, + }); + + // No repair channel, so gatekeeper 1's refusal is terminal. + await expect(impl.ensureObserver("alice", fakeClientUser, "build")) + .rejects.toThrow(/could not confirm/); + + // Coverage for the refused gatekeeper is scrubbed (assertCollaboratorStillVerified fails + // closed on her external-message writes), but the registrations stay put: tearing them + // down would drop alice from excludeObservers while her sessions -- which a failed + // re-verification does not restart -- keep receiving later observations. + expect(removed).toEqual([]); + let record = impl.storage.observers.get("alice"); + expect(1 in record.accountChoices).toBe(false); + expect(record.accountChoices[2]).toBe(20); + expect(record.observerId).toBe("obs-1"); + }); + }); + + it("keeps distinct profiles concurrent", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-serialization-distinct-profiles"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + impl.getGatekeeperFacet = () => ({ addObserver: async () => {} }); + + // Alice parks in her modal; Bob's open must complete anyway. + let held = deferred(); + let configureAlice = { + configure: async () => { + await held.promise; + return [{ gatekeeperId: 1, accountId: 10 }, { gatekeeperId: 2, accountId: 20 }]; + }, + } as any; + let configureBob = { + configure: async () => + [{ gatekeeperId: 1, accountId: 30 }, { gatekeeperId: 2, accountId: 40 }], + } as any; + + let openAlice = impl.ensureObserver("alice", fakeClientUser, "build", configureAlice); + await tick(); + await impl.ensureObserver("bob", fakeClientUser, "build", configureBob); + expect(impl.storage.observers.get("bob")).toBeDefined(); + expect(impl.storage.observers.get("alice")).toBeUndefined(); + + held.resolve(); + await openAlice; + expect(impl.storage.observers.get("alice")).toBeDefined(); + }); + }); +}); + +// A first-time verification registers its freshly minted observerId with gatekeepers before the +// observer record is persisted, so byObserverId cannot resolve the id for the duration of the +// awaits in between (sibling RPCs, the configuration modal). #enforceExcludeObservers must fail +// closed on such an id (via #pendingObserverIds) rather than read it as "not an active observer" +// and let an excluded observation through moments before the collaborator is admitted. +describe("excludeObservers naming a mid-registration observer", () => { + const observation = (excludeObservers: string[]) => + ({ title: "t", description: "d", excludeObservers }); + + it("blocks while the first-time verification is in flight", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-pending-exclusion-block"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + impl.ownerProfileId = "owner"; + + // Gatekeeper 1 accepts the registration immediately (capturing the minted id); gatekeeper 2 + // parks, holding the open in the window where the id is gatekeeper-visible but unpersisted. + let held = deferred(); + let captured: string | undefined; + impl.getGatekeeperFacet = (id: number) => ({ + addObserver: async (observerId: string) => { + captured = observerId; + if (id === 2) await held.promise; + }, + }); + + let open = impl.ensureObserver("alice", fakeClientUser, "build", { + configure: async () => + [{ gatekeeperId: 1, accountId: 10 }, { gatekeeperId: 2, accountId: 20 }], + } as any); + await tick(); + expect(captured).toBeDefined(); + expect(impl.storage.observers.get("alice")).toBeUndefined(); + + // Excluding the mid-registration id fails closed with the distinct message, while a + // genuinely unknown id stays inert. + await expect(impl.authorizeObservation(1, observation([captured!]), { from: "user" })) + .rejects.toThrow(/currently being verified/); + await expect( + impl.authorizeObservation(1, observation(["not-an-observer"]), { from: "user" })) + .resolves.toBeUndefined(); + + held.resolve(); + await open; + }); + }); + + it("becomes inert when the verification fails", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-pending-exclusion-failure"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + impl.ownerProfileId = "owner"; + + let captured: string | undefined; + let removed: number[] = []; + impl.getGatekeeperFacet = (id: number) => ({ + addObserver: async (observerId: string) => { + captured = observerId; + if (id === 2) throw new Error("access refused upstream"); + }, + removeObserver: async () => { removed.push(id); }, + }); + + // The re-prompt offer after gatekeeper 2's refusal is declined, making the failure terminal. + let configured = false; + await expect(impl.ensureObserver("alice", fakeClientUser, "build", { + configure: async () => { + if (configured) throw new Error("cancelled"); + configured = true; + return [{ gatekeeperId: 1, accountId: 10 }, { gatekeeperId: 2, accountId: 20 }]; + }, + } as any)).rejects.toThrow(); + + // A *first-ever* verification failure rolls back both the accepted registration + // (gatekeeper 1, newlyAdded) and the refused one (gatekeeper 2, invalidated): alice was + // never admitted, so nothing preserves forward exclusion, and the minted id would linger + // unresolvable inside the gatekeepers. This is the boundary of the keep-on-re-verification + // rule above, which applies only once a record exists. + expect(removed.toSorted()).toEqual([1, 2]); + + // The finally cleaned the pending map and no record was persisted, so the id is + // unresolvable and correctly inert: that collaborator was never admitted. + expect(captured).toBeDefined(); + expect(impl.storage.observers.get("alice")).toBeUndefined(); + await expect(impl.authorizeObservation(1, observation([captured!]), { from: "user" })) + .resolves.toBeUndefined(); + }); + }); + + it("tears down a lost observer by observer id, tolerating duplicate ids and a mid-teardown " + + "re-verification", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-excluded-teardown-by-id"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + impl.ownerProfileId = "owner"; + // Bob lost access (no collaborator record) but his observer record lingers -- the state + // #decideExcludeObservers resolves to "lost" and #tearDownExcludedObservers cleans up. + impl.storage.observers.put( + { profileId: "bob", observerId: "obs-old", accountChoices: { 1: 10 } }); + + // Park the teardown's removeObserver fan-out. The ids cross the RPC boundary from + // gatekeeper code, so nothing guarantees uniqueness: a duplicate used to enter "lost" + // twice, and the second iteration's delete-by-profileId ran from a snapshot staled by the + // first iteration's await. + let held = deferred(); + let removed: string[] = []; + impl.getGatekeeperFacet = () => ({ + addObserver: async () => {}, + removeObserver: async (id: string) => { removed.push(id); await held.promise; }, + }); + let observing = impl.authorizeObservation( + 1, observation(["obs-old", "obs-old"]), { from: "user" }); + await tick(); + + // Mid-park, bob is re-granted and a fresh verification completes, minting a replacement + // record under a new observerId. + impl.storage.collaborators.put({ + profile: { type: "user", id: "bob", name: "Bob" }, + addedBy: [{ type: "user", sharer: "owner", created: new Date(), role: "build" }], + }); + impl.storage.observers.put( + { profileId: "bob", observerId: "obs-new", accountChoices: { 1: 10 } }); + + held.resolve(); + await expect(observing).resolves.toBeUndefined(); + + // Pre-fix the duplicate's second iteration deleted the replacement record by profileId, + // after which exclusions naming obs-new silently no-oped (fail-open) until bob's next + // open. The teardown must delete only the record it snapshotted. + expect(impl.storage.observers.get("bob")?.observerId).toBe("obs-new"); + await expect(impl.authorizeObservation(1, observation(["obs-new"]), { from: "user" })) + .rejects.toThrow(/current collaborator/); + // The snapshotted id was still de-registered from the gatekeepers, exactly once per + // gatekeeper (the duplicate deduped). + expect(removed.toSorted()).toEqual(["obs-old", "obs-old"]); + }); + }); + + it("hands off seamlessly to the persisted index on success", async () => { + let stub = env.TEST_OVERSEER.getByName("observer-pending-exclusion-success"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let impl = (instance as unknown as { impl: any }).impl; + seedGatekeepers(impl); + impl.ownerProfileId = "owner"; + // Alice is a reachable collaborator (shared directly by the owner). + impl.storage.collaborators.put({ + profile: { type: "user", id: "alice", name: "Alice" }, + addedBy: [{ type: "user", sharer: "owner", created: new Date(), role: "build" }], + }); + + let captured: string | undefined; + impl.getGatekeeperFacet = () => ({ + addObserver: async (observerId: string) => { captured = observerId; }, + }); + + await impl.ensureObserver("alice", fakeClientUser, "build", { + configure: async () => + [{ gatekeeperId: 1, accountId: 10 }, { gatekeeperId: 2, accountId: 20 }], + } as any); + + // The record now carries the id the gatekeepers saw, and exclusion resolves it through the + // index to the still-authorized collaborator -- the pre-existing block, not the pending one. + expect(impl.storage.observers.get("alice")?.observerId).toBe(captured); + await expect(impl.authorizeObservation(1, observation([captured!]), { from: "user" })) + .rejects.toThrow(/current collaborator/); + }); + }); +}); diff --git a/packages/workshop-backend/__tests__/revocation-restart-window.test.ts b/packages/workshop-backend/__tests__/revocation-restart-window.test.ts new file mode 100644 index 000000000..2709426a4 --- /dev/null +++ b/packages/workshop-backend/__tests__/revocation-restart-window.test.ts @@ -0,0 +1,157 @@ +// A revocation's DO abort (scheduleRevocationRestart) is what actually ends the removed +// collaborator's live sessions, but it runs only after two awaited RPC phases -- +// tearDownLostObservers (serial removeObserver fan-out per collaborator) and +// refreshAffectedCollaboratorListings (chunked cross-DO round trips) -- a window that scales with +// collaborator and gatekeeper count, not the ~100ms the abort's own delay suggests. Inside that +// window the removed user still watches the session fan-out, yet no per-observation check can +// see them: their record was already deleted (so an exclusion naming their id reads as "not an +// active observer"), and once the teardown's removeObserver fan-out completes, gatekeepers stop +// naming them at all. authorizeObservation must instead fail every observation closed until the +// restart lands, via the in-memory #revocationRestartPending flag tearDownLostObservers sets +// synchronously with the sever. +// +// Runs against a real OverseerDurableObject (the TEST_OVERSEER binding, like +// observer-serialization.test.ts); the gatekeeper facet is the fake, and the teardown's +// removeObserver is parked on a deferred so the tests occupy the window deterministically. No +// real DO abort ever fires here (scheduleRevocationRestart is not called). + +import { describe, expect, it } from "vitest"; +import { env } from "cloudflare:workers"; +import { runInDurableObject } from "cloudflare:test"; +import type { OverseerDurableObject } from "../src/overseer.js"; + +declare module "cloudflare:workers" { + interface ProvidedEnv { + TEST_OVERSEER: DurableObjectNamespace; + } +} + +const OWNER = "alice"; + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + let promise = new Promise(r => { resolve = r; }); + return { promise, resolve }; +} + +const tick = () => new Promise(resolve => setTimeout(resolve, 0)); + +function seedGatekeeper(impl: any, id: number): void { + impl.storage.gatekeepers.put({ + id, + resourceTitle: `Connection ${id}`, + class: {} as any, + creationSpec: { + type: "gatekeeper", + vendorId: "testvendor", + resourceUrl: `https://example.com/${id}`, + typeUrlPattern: "https://*", + }, + }); +} + +// Seeds a workspace with connection 1 and bob as a confirmed collaborator holding a covering +// observer record, and parks the gatekeeper facet's removeObserver on a deferred the test +// releases -- the teardown window under test. +function setup(instance: OverseerDurableObject): { + impl: any; + releaseTeardown: () => void; + removed: string[]; +} { + let impl = (instance as unknown as { impl: any }).impl; + impl.ownerProfileId = OWNER; + seedGatekeeper(impl, 1); + impl.storage.collaborators.put({ + profile: { id: "bob", name: "Bob" }, + addedBy: [{ type: "user", sharer: OWNER, created: new Date(), role: "build" }], + }); + impl.storage.observers.put( + { profileId: "bob", observerId: "obs-b", accountChoices: { 1: 10 } }); + + let held = deferred(); + let removed: string[] = []; + impl.getGatekeeperFacet = () => ({ + addObserver: async () => {}, + removeObserver: async (id: string) => { removed.push(id); await held.promise; }, + }); + return { impl, releaseTeardown: held.resolve, removed }; +} + +// Severs bob's edge and starts the teardown in the same synchronous block, exactly as +// removeCollaborator does (SharingManager.removeCollaborator is synchronous, and the handler +// awaits tearDownLostObservers immediately after). +function severBob(impl: any): Promise { + let record = impl.storage.collaborators.get("bob"); + impl.storage.collaborators.delete("bob"); + return impl.tearDownLostObservers( + [{ profile: record.profile, addedBy: record.addedBy, oldRole: "build", newRole: null }]); +} + +const excludedObservation = (excludeObservers: string[]) => + ({ title: "Read a thing", description: "The test read a thing.", excludeObservers }); + +describe("observation gates during the revocation-restart window", () => { + it("fails an excluded observation closed while the teardown is parked", async () => { + let stub = env.TEST_OVERSEER.getByName("revocation-window-excluded"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let { impl, releaseTeardown } = setup(instance); + let teardown = severBob(impl); + await tick(); + + // Bob's record was deleted synchronously at the sever, so byObserverId no longer resolves + // obs-b: pre-fix the id read as "not an active observer" and the observation naming bob was + // admitted -- into chat history his still-live session watches. + await expect(impl.authorizeObservation(1, excludedObservation(["obs-b"]), { from: "user" })) + .rejects.toThrow(/just revoked/); + + releaseTeardown(); + await teardown; + + // The flag is never cleared: the restart is what ends the window, and if it were somehow + // lost, staying blocked is the safe direction. + await expect(impl.authorizeObservation(1, excludedObservation(["obs-b"]), { from: "user" })) + .rejects.toThrow(/just revoked/); + }); + }); + + it("fails a plain observation (no exclusions) closed until the restart", async () => { + let stub = env.TEST_OVERSEER.getByName("revocation-window-plain"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let { impl, releaseTeardown } = setup(instance); + let teardown = severBob(impl); + await tick(); + + // No excludeObservers at all: pre-fix only the exclusion gate consulted the flag, so this + // observation was admitted into history bob's still-live session watches. + await expect(impl.authorizeObservation( + 1, { title: "Read a thing", description: "The test read a thing." }, { from: "user" })) + .rejects.toThrow(/just revoked/); + + releaseTeardown(); + await teardown; + + // The window the finding names: the teardown's removeObserver fan-out has completed, so the + // gatekeeper no longer knows obs-b and subsequent observations arrive with no exclusion + // naming bob. Only the every-observation check can cover it. + await expect(impl.authorizeObservation( + 1, { title: "Read a thing", description: "The test read a thing." }, { from: "user" })) + .rejects.toThrow(/just revoked/); + }); + }); + + it("a no-op sharing change does not block observations", async () => { + let stub = env.TEST_OVERSEER.getByName("revocation-window-noop"); + await runInDurableObject(stub, async (instance: OverseerDurableObject) => { + let { impl } = setup(instance); + // A removal that affected nobody (e.g. severing an edge nobody relied on) skips the restart, + // so it must not block observations either -- same predicate as the restart's. + await impl.tearDownLostObservers([]); + + // The named id is unknown, so this is an ordinary admitted observation (naming obs-b would + // block for the wrong reason: bob stays authorized in this test's setup). + await expect(impl.authorizeObservation( + 1, excludedObservation(["not-an-observer"]), { from: "user" })) + .resolves.toBeUndefined(); + }); + }); +}); diff --git a/packages/workshop-backend/src/overseer.ts b/packages/workshop-backend/src/overseer.ts index 3b0723f3e..0812a4cf6 100644 --- a/packages/workshop-backend/src/overseer.ts +++ b/packages/workshop-backend/src/overseer.ts @@ -41,7 +41,7 @@ import type { ProductAnalyticsConnectionType, ProductAnalyticsGadgetInput } from import { checkUsageAndBalance } from "./ai-gateway-billing/limits/usage-checker"; import { completeAgentCatalogSnapshot, normalizeAgentCatalog } from "./agent-catalog"; import { refreshCachedBalance } from "./ai-gateway-billing/cloudflare/connection-service"; -import { SharingManager, SharingCaller, CollaboratorRecord, ShareKeyRecord } from "./sharing"; +import { SharingManager, SharingCaller, CollaboratorRecord, ShareKeyRecord, roleRank } from "./sharing"; import { AutoApprovalDrainer } from "./auto-approval"; import { collectSlashCommands, invokeSlashCommand } from "./slash-commands"; import { createWorkshopLogger, obsContext, traced } from "./observability"; @@ -761,6 +761,16 @@ type ExternalMessageRecord = { type ExternalMessageResponseTargetRegistration = { idempotencyKey: string; chatGatewayRpcTarget: NativeRpcStub; + // Re-asserts the submitting caller's authorization, strictly after newChat/sendChatMessage's + // internal awaits and synchronously with every write the submission justifies, so a caller + // whose access went stale in the window since receiveExternalMessage's entry gate commits + // nothing (see assertCollaboratorStillVerified). newChat runs it as the first statement of the + // transaction that commits the prompt and registers the response target; sendChatMessage runs + // it just before materializeChatChanges -- its first write, which cannot move inside the + // transaction (non-transactional side effects) -- with no awaits between the check and the + // transaction, so the one check covers the whole synchronous write sequence. Absent for the + // owner, whose access cannot go stale. + assertStillAuthorized?: () => void; }; type ExternalMessageResponseTargetRegistrationDecision = @@ -4356,8 +4366,29 @@ class OverseerImpl implements AgentHooks { async authorizeObservation(gatekeeperId: number, description: ObservationDescription, caller: GatekeeperCaller): Promise { + // House rule (cf. addCollaborator): every check runs in one synchronous block with the writes + // it justifies -- the prohibitAllSharing latch and the action record. The only await before + // that block is the memoized sharing manager. Previously the excluded observers' cross-worker + // teardown was awaited between the exclusion check and the action record, so a re-grant + // landing in that window admitted an observation naming a collaborator who was authorized + // again by the time it was recorded. The one genuinely-async step -- tearing down excluded + // observers' gatekeeper registrations -- is deferred to after the writes. + let sharing = await this.getSharingManager(); + + // A revocation whose restart hasn't landed yet leaves the removed user's sessions live and + // watching while their observer record is already gone -- and once the teardown's + // removeObserver fan-out completes, gatekeepers no longer know their observer id, so + // subsequent observations arrive with no exclusion naming them. No per-observation check can + // see the removed user then, so every observation fails closed until the restart. Checked + // before the prohibitAllSharing latch: a blocked observation must not latch. + if (this.#revocationRestartPending) { + throw new Error( + "This observation was blocked because a collaborator's access to this workspace was " + + "just revoked and their sessions have not yet been disconnected."); + } + if (description.prohibitAllSharing) { - if ((await this.getSharingManager()).hasAnyShares()) { + if (sharing.hasAnyShares()) { throw new Error( "This observation was blocked because it contains sensitive data that must only be " + "shown to the account owner, but this workspace is shared with other users. Try again " + @@ -4371,9 +4402,11 @@ class OverseerImpl implements AgentHooks { // v1 has no per-thread hiding, the only way to let such an observation proceed is if the named // observer has already lost access in the sharing graph. If any named observer is still // authorized, we cannot prevent them from seeing it, so we block the observation. See - // observers-implementation-plan.md §5 Step 5. + // observers-implementation-plan.md §5 Step 5. The decision is synchronous; the losers' + // teardown is deferred past the writes below. + let lostObservers: ObserverRecord[] = []; if (description.excludeObservers && description.excludeObservers.length > 0) { - await this.#enforceExcludeObservers(description.excludeObservers); + lostObservers = this.#decideExcludeObservers(description.excludeObservers, sharing); } let actionId = this.storage.nextActionId.get(); @@ -4395,6 +4428,11 @@ class OverseerImpl implements AgentHooks { this.storage.actions.put(record); this.#associateAction(caller, actionId); + + // Awaited rather than handed to waitUntil: ApprovalQueueImpl returns this promise to the + // gatekeeper worker, so an admitted observation implies the excluded observers' teardown ran + // before any data flows. It never throws (removeObserver is best-effort). + await this.#tearDownExcludedObservers(lostObservers); } async getChatAttachmentData(chatId: number, id: string): Promise { @@ -4490,20 +4528,34 @@ class OverseerImpl implements AgentHooks { }); } - // Enforce an observation's `excludeObservers`. For each named opaque observerId: + // Decide an observation's `excludeObservers`. For each named opaque observerId: // - Map it back to a profileId via the byObserverId index. An unknown id is not an active - // observer (e.g. already torn down), so it is ignored. + // observer (e.g. already torn down), so it is ignored -- unless a first-time verification + // registered it with a gatekeeper but has not yet persisted its record + // (#pendingObserverIds), in which case the profile may be admitted moments later and we + // fail closed. // - If that profileId is still authorized in the sharing graph, we cannot guarantee they won't // see the observation (v1 has no per-thread hiding), so we throw to block it. - // - If that profileId is no longer authorized, we allow the observation for them and delete - // their observer record (best-effort removeObserver on all gatekeepers). They are no longer - // set up to observe; if they regain access they reconfigure from scratch (Step 3). - // If no named observer is still authorized, the observation is allowed. - async #enforceExcludeObservers(observerIds: string[]): Promise { - let sharing = await this.getSharingManager(); - - // Observers who are still authorized block the observation outright. - for (let observerId of observerIds) { + // - If that profileId is no longer authorized, we allow the observation for them and return + // their record: they are no longer set up to observe, so the caller tears them down + // (#tearDownExcludedObservers); if they regain access they reconfigure from scratch (Step 3). + // Deliberately synchronous (the sharing manager is a parameter) so authorizeObservation can + // decide and record in one synchronous block, deferring only the teardown. + #decideExcludeObservers(observerIds: string[], sharing: SharingManager): ObserverRecord[] { + // No #revocationRestartPending check here: authorizeObservation fails *every* observation + // closed across the revocation window before this runs, which also covers the unknown-id + // `continue` below admitting an observation naming a just-torn-down observer. + let lost: ObserverRecord[] = []; + // Deduped: the ids cross the RPC boundary from gatekeeper code, so nothing guarantees + // uniqueness, and a duplicate would push the same record into `lost` twice -- two teardown + // iterations for one observer, the second running from a snapshot the first's awaited + // fan-out staled (see #tearDownExcludedObservers). + for (let observerId of new Set(observerIds)) { + if (this.#pendingObserverIds.has(observerId)) { + throw new Error( + "This observation was blocked because it contains data that a collaborator currently " + + "being verified may not be permitted to see."); + } let observer = this.storage.observers.byObserverId.get(observerId); if (!observer) continue; // not an active observer -> ignore @@ -4512,16 +4564,28 @@ class OverseerImpl implements AgentHooks { "This observation was blocked because it contains data that a current collaborator " + "is not permitted to see."); } + lost.push(observer); } + return lost; + } - // No still-authorized observer was named. Tear down any named observers who have already lost - // access, since they are no longer set up to observe. + // Tear down excluded observers #decideExcludeObservers found to have already lost access: delete + // each record and best-effort removeObserver on all gatekeepers. Never throws. + async #tearDownExcludedObservers(observers: ObserverRecord[]): Promise { + if (observers.length === 0) return; let gatekeeperIds = [...this.storage.gatekeepers.list()].map(gk => gk.id); - for (let observerId of observerIds) { - let observer = this.storage.observers.byObserverId.get(observerId); - if (!observer) continue; - this.storage.observers.delete(observer.profileId); - await this.#removeObserverFromGatekeepers(observerId, gatekeeperIds); + for (let observer of observers) { + // Each earlier iteration's awaited fan-out is a yield in which the profile may have been + // re-granted and re-verified, minting a replacement record under a new observerId; deleting + // by profileId from this method's (snapshotted) list would then remove the *replacement*, + // silently no-oping every exclusion that names the new id until the user's next open. So + // re-read and delete only the record actually snapshotted; the snapshotted id is still + // de-registered unconditionally (removeObserver is idempotent and the id is dead either + // way). + if (this.storage.observers.get(observer.profileId)?.observerId === observer.observerId) { + this.storage.observers.delete(observer.profileId); + } + await this.#removeObserverFromGatekeepers(observer.observerId, gatekeeperIds); } } @@ -4907,6 +4971,13 @@ class OverseerImpl implements AgentHooks { // the owner, who is also connected and will be disconnected) before their connection drops. // Without the delay their own removeCollaborator()/revokeShareLink() call might reject with a // connection error even though it succeeded. + // + // Note the abort lands well after the sever, not just this method's own ~100ms delay: the + // handlers first await tearDownLostObservers (a serial removeObserver fan-out per lost + // collaborator) and refreshAffectedCollaboratorListings (chunked cross-DO round trips), so the + // window scales with collaborator and gatekeeper count. The removed users' sessions stay live + // and watching throughout; #revocationRestartPending is what keeps authorizeObservation + // failing closed across it. async scheduleRevocationRestart(): Promise { await this.ctx.storage.sync(); await scheduler.wait(100); @@ -5199,6 +5270,11 @@ class OverseerImpl implements AgentHooks { let chatId!: number; let timestamp = this.getChatTimestamp(); this.ctx.storage.transactionSync(() => { + // Re-assert an external caller's authorization atomically with the writes it justifies: + // every await above (message preparation included) was a window in which it may have gone + // stale, and a throw here aborts the transaction -- no chat, no message, no response + // target -- before startAgent below can run over the chat tail. + responseTargetRegistration?.assertStillAuthorized?.(); chatId = this.nextChatId(); let meta: AiChatMetadata = { id: chatId, @@ -5277,6 +5353,15 @@ class OverseerImpl implements AgentHooks { message, (canonicalAttachments?.length ?? 0) > 0); let meta = this.assertChatNotActive(chatId, true); + // Re-assert an external caller's authorization before *any* write this submission justifies + // -- which on this path starts with materializeChatChanges, not the transaction below: the + // materialization durably writes a "changes" chat message, retires and prunes change rows, + // and sets hasProposedChanges, none of which a stale caller may trigger. It stays outside + // the transaction because it has non-transactional side effects (proposedChangesChanged's + // facet abort, the TTL prune), so the check is hoisted instead; everything from here through + // the transactionSync is one synchronous block (no awaits), so this single check covers the + // whole write sequence -- same invariant as newChat, differently shaped. + responseTargetRegistration?.assertStillAuthorized?.(); let result = this.materializeChatChanges(chatId, meta); if (result) meta = result.meta; meta.lastActive = this.getChatTimestamp(); @@ -7701,6 +7786,32 @@ class OverseerImpl implements AgentHooks { return this.#inScopeGatekeepers(role).map(observerBindingNeed); } + // Re-assert, synchronously, what authorizeCollaborator established for `profileId` when it + // admitted them: an effective role of at least `requireRole`, and -- when their live + // verification scope is nonempty -- a persisted observer record covering every in-scope + // gatekeeper. This mirrors ensureObserver's success invariant exactly: on success it persists a + // record whose accountChoices cover the whole scope, and an empty scope persists no record at + // all (hence the nonempty guard -- no spurious failures). The scope is recomputed live, so a + // gatekeeper *added* since the entry gate fails closed. For entry points whose entry gate is + // separated from the write it justifies by real await windows (receiveExternalMessage), in + // which a sharing change may have severed the caller's role and torn down their observer + // record. The sharing manager is a parameter, not an internal await, so the caller can run this + // inside the same synchronous block as the write (the house rule -- cf. authorizeObservation). + assertCollaboratorStillVerified( + profileId: string, requireRole: CollaboratorRole, sharing: SharingManager): void { + let role = sharing.getEffectiveRole(profileId); + if (!role || roleRank(role) < roleRank(requireRole)) { + throw new Error("You no longer have access to this workspace."); + } + let inScope = this.#inScopeGatekeepers(role); + if (inScope.length === 0) return; + let record = this.storage.observers.get(profileId); + if (!record || inScope.some(gk => !(gk.id in record.accountChoices))) { + throw new Error( + "Your verification for the data this workspace has read is no longer valid."); + } + } + // Best-effort `removeObserver(observerId)` across the given gatekeeper ids. Never throws; logs // and continues on error. An orphaned observer entry only ever causes superfluous future checks, // never a data leak (the leak-relevant gate is authorizeObservation, which keys off the live @@ -7717,13 +7828,29 @@ class OverseerImpl implements AgentHooks { })); } + // Whether a sharing change that removed or downgraded someone has happened this DO session: + // the revocation restart (scheduleRevocationRestart's abort) is coming, but it lands only + // after the awaited teardown and listing-refresh phases below, so the affected users' live + // sessions keep watching the fan-out in the meantime. Once the teardown de-registers the + // observer, gatekeepers stop naming them in excludeObservers, so no per-gate check can see the + // removed user; authorizeObservation therefore consults this flag for *every* observation and + // fails closed across the window. + // In-memory is the right scope, mirroring #pendingObserverIds: the abort destroys the flag + // with the DO, and it is deliberately never cleared -- if the restart is somehow lost, staying + // blocked is the safe direction (the next observation-free reconnect gets a fresh DO anyway). + #revocationRestartPending = false; + // Tear down observer records for collaborators who lost access as a result of a sharing change. // For each affected collaborator who is now fully unauthorized (newRole === null) and has an - // observer record: best-effort removeObserver on all gatekeeper facets, then delete the record. + // observer record: delete the record, then best-effort removeObserver on all gatekeeper facets. // All calls are best-effort -- an orphaned observer entry only causes superfluous future checks, // never a data leak (the leak-relevant gate is authorizeObservation, keyed off the live sharing // graph). See observers-implementation-plan.md §5 Step 6. async tearDownLostObservers(affected: AffectedCollaborator[]): Promise { + // Both callers (removeCollaborator, revokeShareLink) enter here synchronously after the + // sever, and restart on exactly this predicate (downgrades included -- a downgraded "use" + // session must not keep watching at "build" width either). + if (affected.length > 0) this.#revocationRestartPending = true; let gatekeeperIds = [...this.storage.gatekeepers.list()].map(gk => gk.id); for (let entry of affected) { if (entry.newRole !== null) continue; // downgraded but still has access -> keep record @@ -7762,26 +7889,174 @@ class OverseerImpl implements AgentHooks { } } + // The authorization gate every non-owner entry point must pass through: resolve the caller's + // effective role, then verify them as an observer of everything this workspace has read. + // Returns null for no access; verification failures throw. A caller that requires at least + // `requireRole` (e.g. receiveExternalMessage needs "build") passes it so an insufficient role + // is denied *before* verification runs -- otherwise the caller would be verified (real + // addObserver calls, a persisted observer record) only to be turned away, or worse, told to fix + // a verification failure that can never grant them access. `configureCb` is forwarded to + // ensureObserver to prompt for unconfigured account choices; without it, verification is + // non-interactive and an unconfigured binding denies access. + // + // Verification may park unboundedly (verifier RPCs, the configuration modal), and a removal + // landing in that window is otherwise caught only by the revocation restart, which fires after + // the awaited teardown/listing phases -- so the effective role is re-checked when the + // verification commits (the commit gate below) and re-derived before the capability is + // returned. + // + // Today receiveExternalMessage() is the only caller: open() still runs the same role + // resolution, ensureObserver call, and commit gate inline, interleaved with share-key + // redemption, and migrating it onto this gate is deliberately left to the redemption rework + // that has to restructure that path anyway. + async authorizeCollaborator( + profileId: string, + clientUser: DurableObjectStub, + opts: { + configureCb?: RpcStub; + requireRole?: CollaboratorRole; + } = {}): Promise { + let sharing = await this.getSharingManager(); + let role = sharing.getEffectiveRole(profileId); + if (!role || (opts.requireRole && roleRank(role) < roleRank(opts.requireRole))) return null; + + // Hand ensureObserver a commit gate: a live-role re-check run synchronously with the write + // that persists the observer record. Verification parks across real await windows, in which + // the caller may be removed: the removal's teardown deletes the caller's observer record, and + // without the gate step 6's put would *resurrect* it (record and account choices were loaded + // pre-park) -- coverage a later re-grant would trust without re-verification -- and the open + // would hand out a full stale-role capability for however long the revocation restart takes + // to land. A denial throws *inside* ensureObserver's try, taking the same rollback as any + // other verification failure; success runs the gate and the record persist back-to-back + // inside the per-profile verification lock. + let commitGate = () => { + if (!sharing.getEffectiveRole(profileId)) { + throw new Error( + "Your access to this workspace was revoked while it was being verified."); + } + }; + + await this.ensureObserver(profileId, clientUser, role, opts.configureCb, commitGate); + + // Re-derive the role from the live graph. A revocation landing mid-verification is expected + // to be denied by the commit gate above, before anything persists; this re-check is the + // residual guard for a change landing in the await gaps *after* the gate ran -- the role + // collapses to null and the caller rejects. + let confirmed = sharing.getEffectiveRole(profileId); + if (!confirmed || + (opts.requireRole && roleRank(confirmed) < roleRank(opts.requireRole))) { + return null; + } + // Decreases pass through (verification at the wider pre-park role covers the narrower live + // scope, and the capability handed out must not exceed the live role), but an increase + // -- say an owner grant of "build" landing while verification waited on the configuration + // modal -- must not ride out on this open: ensureObserver verified the caller at `role`, and + // a wider role widens the gatekeeper scope that verification must cover. The raise takes + // effect at the caller's next open, which verifies at the wider scope. + return roleRank(confirmed) < roleRank(role) ? confirmed : role; + } + + // In-flight verification per profile (see ensureObserver). Entries are removed when their + // verification settles; the map is small (one entry per concurrently-opening collaborator). + #observerVerification = new Map>(); + + // Observer ids a first-time verification has registered with at least one gatekeeper but whose + // record is not yet persisted, so byObserverId cannot resolve them (observerId -> profileId). + // Consulted by #decideExcludeObservers, which otherwise reads such an id as "not an active + // observer" and lets an excluded observation through -- the collaborator is then admitted + // moments later with the data already in chat history. In-memory is the right scope: a DO + // restart kills the in-flight open, and its gatekeeper-side registration then references an id + // no record will ever carry, so ignoring it is correct. + #pendingObserverIds = new Map(); + // Bring a non-owner `profileId` into compliance as an observer for their `role`, so that they may // open the Gadget. May invoke `configureCb` to ask the user to choose connected accounts for // gatekeeper bindings they haven't configured yet. Re-runs `addObserver` (re-verification) for // already-configured bindings on every open, catching revocation of the user's underlying // resource access promptly. Returns when fully verified; throws to deny access. // + // Serialized per profile: the body loads the observer record, awaits verifier RPCs (and possibly + // the configuration modal, which parks on user input indefinitely), and persists the record at + // the end. Input gates don't cover those awaits, so two concurrent opens for one profile would + // otherwise race -- the second open's final put would resurrect coverage the first open's + // failure just scrubbed, and two concurrent *first* opens would each mint their own observerId, + // registering the losing id with gatekeepers while the winning record forgets it ever existed. + // A promise chain rather than blockConcurrencyWhile, which would freeze the whole DO for the + // duration (unbounded, given the modal) -- same pattern as #preparingChatMessages and the + // Google gatekeeper's credential lock. Serializing only per profile keeps distinct + // collaborators' opens concurrent. + // + // `commitGate`, when given, runs synchronously at each success exit -- immediately before the + // step-6 record persist, or at the nothing-to-verify early return -- inside the per-profile + // lock. A throw denies the verification and takes the same rollback path as any other failure, + // so the caller can piggyback its own commit-time checks on the record persist's synchronous + // block. Running the gate *inside* the lock matters: a gate invoked after this method returned + // would run after the lock's release, so a queued sibling verification for the same profile + // would start against no record and mint a second observerId, breaking the shared-id invariant. + // // See observers-implementation-plan.md §5 Step 3. async ensureObserver( profileId: string, clientUser: DurableObjectStub, role: CollaboratorRole, - configureCb?: RpcStub): Promise { - // 1. Select in-scope gatekeepers. If none require an account, there is nothing to verify and - // no observer record is needed (built-in gatekeepers never name observers in - // excludeObservers). + configureCb?: RpcStub, + commitGate?: () => void): Promise { + let previous = this.#observerVerification.get(profileId) ?? Promise.resolve(); + let release!: () => void; + let current = new Promise(resolve => { release = resolve; }); + this.#observerVerification.set(profileId, current); + await previous; + try { + await this.#ensureObserverLocked(profileId, clientUser, role, configureCb, commitGate); + } finally { + release(); + if (this.#observerVerification.get(profileId) === current) { + this.#observerVerification.delete(profileId); + } + } + } + + async #ensureObserverLocked( + profileId: string, + clientUser: DurableObjectStub, + role: CollaboratorRole, + configureCb?: RpcStub, + commitGate?: () => void): Promise { + // 1. Select in-scope gatekeepers. If none require an account, there is nothing to verify + // (built-in gatekeepers never name observers in excludeObservers). let inScope = this.#inScopeGatekeepers(role); - if (inScope.length === 0) return; - // 2. Load any existing observer record, and build a working copy of its account choices. + // 2. Load any existing observer record, and prune every account choice for a gatekeeper now + // outside this collaborator's verification scope. This restores the invariant commit-time + // re-checks (assertCollaboratorStillVerified) rest on: entry present => verified at this + // collaborator's most recent open. Without the prune, a "use" collaborator opening while a + // connection is unbound from every gadget verifies nothing against it, yet their stale + // entry survives to be trusted the moment the connection is rebound (same gatekeeper id -- + // only gadget binding edges changed). The prune must run even when the remaining scope is + // empty -- that's exactly the everything-unbound open. The registration itself is left + // with the gatekeeper (no removeObserver): keeping it preserves forward exclusion via + // byObserverId, and the record stays even if its accountChoices empties, since the + // observerId remains referenced. let record = this.storage.observers.get(profileId); + if (record) { + let inScopeIds = new Set(inScope.map(gk => gk.id)); + let pruned = false; + for (let key of Object.keys(record.accountChoices)) { + if (!inScopeIds.has(Number(key))) { + delete record.accountChoices[Number(key)]; + pruned = true; + } + } + if (pruned) this.storage.observers.put(record); + } + if (inScope.length === 0) { + // Nothing to verify, but this is still a success exit: the caller's commit gate must still + // run. Nothing has been minted at this point, so a throw here has no rollback to do. + commitGate?.(); + return; + } + + // Build a working copy of the (pruned) account choices. let accountChoices: {[gatekeeperId: number]: number} = {...record?.accountChoices}; // Gatekeeper ids whose account choice came from the persisted record (vs. configured during @@ -7791,8 +8066,17 @@ class OverseerImpl implements AgentHooks { inScope.filter(gk => gk.id in accountChoices).map(gk => gk.id)); let observerId = record?.observerId ?? crypto.randomUUID(); + // A freshly minted id becomes visible to gatekeepers at the first addObserver below, but + // resolvable via byObserverId only at the step-6 put -- hold it in #pendingObserverIds across + // that window so #decideExcludeObservers can fail closed on it. The commit gate, the put, + // and the finally's delete run synchronously back-to-back (the gate is synchronous by + // contract), so there is no gap where neither the map nor the index resolves the id. + if (!record) this.#pendingObserverIds.set(observerId, profileId); // Gatekeepers we successfully registered the observer with during this call. let newlyAdded = new Set(); + // Gatekeepers that refused (or whose account was gone) during this call -- see fail() below, + // which scrubs each from the persisted observer record as the failure is determined. + let invalidated = new Set(); // Failures from the previous pass, keyed by gatekeeper id: an already-configured binding whose // chosen account was disconnected, or which the gatekeeper refused. @@ -7883,25 +8167,44 @@ class OverseerImpl implements AgentHooks { let fail = (reason: string, err?: unknown) => { failures.set(gk.id, {accountId, reason}); + // Commit-time re-checks (assertCollaboratorStillVerified) read the *persisted* + // record from other turns, so until this gatekeeper is scrubbed from it, the record + // keeps vouching for the collaborator's still-live sessions -- admitting their + // external-message writes -- even though the live check just refused them. Scrub it + // synchronously with the failure determination (the record is re-read because the + // awaits since load may have let a concurrent open update it; get/put are synchronous + // in this single-threaded DO, so nothing lands between the check and the write). + // Scoped to the failed gatekeeper: coverage elsewhere stays intact, and a repaired + // pass re-persists full coverage at step 6. + invalidated.add(gk.id); + let persisted = this.storage.observers.get(profileId); + if (persisted && gk.id in persisted.accountChoices) { + delete persisted.accountChoices[gk.id]; + this.storage.observers.put(persisted); + } this.logger.warn("observer verification failed", { event: "gatekeeper.observer.verify.failed", gatekeeperId: gk.id, vendorId, accountId, observerId, error: err, }); }; - let verifier = await clientUser.getVerifier(accountId, vendorId); - if (!verifier) { - // Account gone -> the overseer authors the reason. (Wrong vendor throws above.) - fail("This account is no longer connected."); - return; - } - try { + let verifier = await clientUser.getVerifier(accountId, vendorId); + if (!verifier) { + // Account gone -> the overseer authors the reason. (Wrong vendor throws above.) + fail("This account is no longer connected."); + return; + } await this.getGatekeeperFacet(gk.id).addObserver(observerId, verifier); if (!preConfigured.has(gk.id)) newlyAdded.add(gk.id); } catch (err) { // Either a settled denial or an operational failure (expired credentials, upstream - // outage). Treat every failure as repairable and let the user try again. + // outage) -- whether from resolving the verifier or from the gatekeeper's + // addObserver. Treat every failure as repairable and let the user try again. + // getVerifier sits inside this try so its rejection (the wrong-vendor throw, or a + // cross-worker transport failure) scrubs the persisted coverage like any other + // refusal -- and so these callbacks never reject, which keeps the terminal catch's + // newlyAdded/invalidated snapshot from missing late-finishing siblings. fail(stringifyError(err), err); } })); @@ -7935,16 +8238,41 @@ class OverseerImpl implements AgentHooks { // All in-scope bindings verified successfully. break; } + + // 6. Run the caller's commit gate, then persist the observer record, only after all + // addObserver calls succeed. The two run in one synchronous block: a gate denial throws + // into the catch below and rolls back like any other verification failure, and nothing + // can land between a passing gate and the put. Creating/updating the record is the + // canonical moment the user becomes a configured observer. + commitGate?.(); + this.storage.observers.put({profileId, observerId, accountChoices}); } catch (err) { - // Best-effort remove all the observers that were newly-added since we didn't persist the - // user's observer record. - await this.#removeObserverFromGatekeepers(observerId, [...newlyAdded]); + // Roll back gatekeeper registrations only for a *first-ever* verification (no persisted + // record at call start -- the same discriminator as #pendingObserverIds): that collaborator + // was never admitted, has no live session, and once the pending entry is dropped in the + // finally their minted id would linger unresolvable inside the gatekeepers. For a + // re-verification failure the registrations are deliberately kept: coverage was already + // scrubbed synchronously in fail() (so assertCollaboratorStillVerified fails closed on + // their external-message writes), while the registration is what preserves forward + // exclusion -- byObserverId keeps resolving the id, so prepareObservation keeps naming + // this observer in excludeObservers for their still-live sessions (a failed + // re-verification does not end sessions; only scheduleRevocationRestart does). A + // kept-but-stale registration is fail-closed (it can only add exclusion names) and + // self-heals: the next successful open's addObserver overwrites the verifier. + if (!record) { + await this.#removeObserverFromGatekeepers( + observerId, [...new Set([...newlyAdded, ...invalidated])]); + } else if (this.storage.observers.get(profileId)?.observerId !== observerId) { + // A teardown racing this call's parked awaits deleted the record this call anchored on; + // the registrations just re-asserted reference an id no record resolves (the kept- + // registration rationale above needs byObserverId to keep resolving it), so remove them + // all. Issued after step 5 settled, so this cannot lose to an in-flight addObserver. + await this.#removeObserverFromGatekeepers(observerId, inScope.map(gk => gk.id)); + } throw err; + } finally { + if (!record) this.#pendingObserverIds.delete(observerId); } - - // 6. Persist the observer record only after all addObserver calls succeed. Creating/updating - // the record is the canonical moment the user becomes a configured observer. - this.storage.observers.put({profileId, observerId, accountChoices}); } // Render the observer verification failures as one line per binding, naming the connection and the @@ -8270,7 +8598,33 @@ export class OverseerDurableObject extends DurableObject { // gatekeepers, configuring their connected accounts if needed. This runs only after a valid // role is confirmed, so it never reveals gatekeeper or resource metadata to an unauthorized // user. The prohibitAllSharing short-circuit above still wins -- lockdown takes precedence. - await this.impl.ensureObserver(profileId, clientUser, role, configureObservers); + // + // Verification parks unboundedly (verifier RPCs, the configuration modal), and a removal + // landing in that window is otherwise caught only by the revocation restart, which fires + // after the awaited teardown/listing phases -- so the live role is re-checked when the + // verification commits (the same gate authorizeCollaborator uses; without it, step 6's put + // would resurrect the observer record the removal's teardown just deleted), and re-derived + // below before the capability is selected. redeemShareKey above wrote a live edge before + // the role read, so the live-graph re-check is the whole story here. + let commitGate = () => { + if (!sharing.getEffectiveRole(profileId)) { + throw new Error( + "Your access to this workspace was revoked while it was being verified."); + } + }; + await this.impl.ensureObserver( + profileId, clientUser, role, configureObservers, commitGate); + + // Re-derive the role from the live graph: the gate denies a removal that landed before the + // record persisted; this covers a change landing in the await gaps after it. A mid-park + // downgrade caps the capability handed out (a stale "build" must not yield the full + // interface), while an increase must not ride out on this open either -- verification ran + // at the narrower scope (cf. authorizeCollaborator). + let confirmed = sharing.getEffectiveRole(profileId); + if (!confirmed) { + throw createOpenGadgetError(OPEN_GADGET_ERROR_CODES.workspaceAccessDenied); + } + role = roleRank(confirmed) < roleRank(role) ? confirmed : role; // Fire-and-forget a call to the collaborator's user DO so the gadget appears on // (or is refreshed on) their home page. @@ -8343,7 +8697,25 @@ export class OverseerDurableObject extends DurableObject { ownerId = callerId; } - // Caller must be the owner or a build collaborator. + // Caller must be the owner or a build collaborator. The agent's reply can surface anything + // the workspace has already read (chat history, gadget storage), so a collaborator passes the + // same authorization gate as open() -- but non-interactively: with no way to configure + // accounts here, an unverified caller is sent to open the workspace, which is where + // verification happens. Requiring "build" up front means a "use" collaborator gets the plain + // denial below rather than being verified (or told to fix a verification failure) for access + // this path can never grant them. The gate is then re-asserted synchronously with the prompt + // commit (assertStillAuthorized below): the awaits between here and sendChatMessage/newChat + // are windows in which a sharing change can strip the caller's access (or a new connection + // widen the scope they were verified against), and the reply must not leave the Workshop on + // a check that went stale. + let unverifiedDenial = (err: unknown): SubmitExternalMessageResult => ({ + accepted: false, + message: "Your access to the data this workspace has read could not be verified. Open " + + "the workspace in your browser to verify your access, then try again. " + + `(${stringifyError(err)})`, + }); + let verificationWentStale = false; + let assertStillAuthorized: (() => void) | undefined; if (ownerId !== callerId) { if (this.impl.storage.prohibitAllSharing.get()) { return { @@ -8351,13 +8723,33 @@ export class OverseerDurableObject extends DurableObject { message: "This workspace has sharing disabled, so only its owner can access it.", }; } - let role = (await this.impl.getSharingManager()).getEffectiveRole(callerProfile.id); + let role: CollaboratorRole | null; + try { + role = await this.impl.authorizeCollaborator( + callerProfile.id, caller, {requireRole: "build"}); + } catch (err) { + return unverifiedDenial(err); + } if (role !== "build") { return { accepted: false, message: "You do not have access to interact with this workspace through its agent.", }; } + // The manager is resolved here (memoized; authorizeCollaborator just used it) so the + // commit-time re-assertion itself is fully synchronous. The flag discriminates the gate's + // own throw from ordinary submit errors, so the catch around the submit below can answer + // with the verification denial rather than rethrowing. + let sharing = await this.impl.getSharingManager(); + let profileId = callerProfile.id; + assertStillAuthorized = () => { + try { + this.impl.assertCollaboratorStillVerified(profileId, "build", sharing); + } catch (err) { + verificationWentStale = true; + throw err; + } + }; } // Complete pending registration in the owner's UserDO. @@ -8400,29 +8792,35 @@ export class OverseerDurableObject extends DurableObject { let responseTargetRegistration: ExternalMessageResponseTargetRegistration = { idempotencyKey: input.idempotencyKey, chatGatewayRpcTarget: input.chatGatewayRpcTarget, + assertStillAuthorized, }; let chatId: number; - if (externalChat) { - await this.impl.sendChatMessage( - caller, - userContext, - externalChat.chatId, - input.prompt, - undefined, - undefined, - responseTargetRegistration, - ); - chatId = externalChat.chatId; - } else { - chatId = await this.impl.newChat( - caller, - userContext, - input.prompt, - undefined, - undefined, - responseTargetRegistration, - input.externalChatKey, - ); + try { + if (externalChat) { + await this.impl.sendChatMessage( + caller, + userContext, + externalChat.chatId, + input.prompt, + undefined, + undefined, + responseTargetRegistration, + ); + chatId = externalChat.chatId; + } else { + chatId = await this.impl.newChat( + caller, + userContext, + input.prompt, + undefined, + undefined, + responseTargetRegistration, + input.externalChatKey, + ); + } + } catch (err) { + if (verificationWentStale) return unverifiedDenial(err); + throw err; } return { accepted: true, chatPath: `/workspace/${this.ctx.id.toString()}?chat=${chatId}` }; diff --git a/packages/workshop-backend/src/sharing.ts b/packages/workshop-backend/src/sharing.ts index ff0cb9e18..f11123922 100644 --- a/packages/workshop-backend/src/sharing.ts +++ b/packages/workshop-backend/src/sharing.ts @@ -25,8 +25,12 @@ import { AiChatAuthorInfo, CollaboratorInfo, PermissionEdge, CollaboratorRole, A from "@gadgets/workshop-shared/api"; import { Collection, NonUniqueIndex } from "@gadgets/typed-storage"; -// Roles are totally ordered: build > use. Higher rank means strictly more access. -function roleRank(role: CollaboratorRole): number { +/** + * Roles are totally ordered: build > use. Higher rank means strictly more access. Exported so + * role comparisons elsewhere (e.g. the Overseer's `requireRole` floor) rank rather than + * string-compare, which stays correct if a role is ever added between the two. + */ +export function roleRank(role: CollaboratorRole): number { return role === "build" ? 2 : 1; }