Allow restricted data writes to its own restricted gatekeeper

docs/observers.md

@@ -34,7 +34,8 @@ The mechanism is a per-user, gatekeeper-mediated check — "this data may be sha
 people who *also* have access to it". (Maximally sensitive data gets an extra layer: an
 observation marked **`containsRestrictedData`**
 (`ObservationDescription.containsRestrictedData` in `packages/workshop-shared/src/gatekeeper.ts`)
-latches the workspace into a restricted mode — no actions, no web fetches — and is admitted only
+latches the workspace into a restricted mode — no web fetches, and actions only back to the
+connections that produced the restricted data, each manually approved — and is admitted only
 if every current collaborator has been verified against the gatekeeper producing it; see the
 coverage guard, `#assertSensitiveObservationCoverage`, in `overseer.ts` and edge case 4 below.)
 
@@ -441,7 +442,8 @@ already in the JSDoc in `gatekeeper.ts`; add anything missing there rather than
    on any collaborator regardless of role). At open() time, `ensureObserver` re-verifies each
    collaborator against every in-scope gatekeeper, which is what admits (or refuses) them for
    sensitive data. The flag also latches the workspace into a restricted mode that blocks
-   actions and web fetches.
+   web fetches and limits actions to the connections that produced the restricted data, each
+   requiring manual approval.
    `use` scope is *live* binding state, with a transition case in each direction. Adding a
    binding grows it, and edge case 5 covers the interim. Unbinding shrinks it with no guard:
    a formerly-bound producer drops out of `use` verification scope, so its sensitive reads
@@ -604,7 +606,8 @@ its resource types.
 
 - **A — Private-only.** Non-owner observers are refused: `addObserver()` unconditionally throws.
   For data that must additionally never leak back out, the `containsRestrictedData` restricted
-  mode (no actions, no web fetches) is available separately; combined with strategy A it makes
+  mode (actions only to the connections that produced the data, each manually approved; no web
+  fetches) is available separately; combined with strategy A it makes
   the workspace effectively private once sensitive data is observed.
   `getVerifier()` must still exist (the overseer mints one on every open) but is never consulted.
 

packages/integration-tests/__tests__/auto-approval-policy.test.ts

@@ -0,0 +1,164 @@
+// Tests for the auto-approval policy gates around sensitive data and operator warnings.
+//
+// Auto-approval requires the author's per-action `autoApprovable` verdict AND a user-enabled rule
+// for the action's kind. Two things must additionally force manual approval no matter what:
+// `operatorWarnings` on the action (a warning exists to be read by a human), and the
+// restricted-data latch -- even when the rule was enabled before the data was read. (The
+// web-fetch restriction has no client-reachable surface, so it is not asserted here.)
+//
+// The fixture gatekeeper's session drives this through the real ApprovalQueue funnel: `doThing()`
+// submits a "poke" action with the given verdict/warnings, and `applyAction` succeeds, so the
+// drain's submit -> auto-approve -> apply round trip is the real one.
+
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import type { RpcStub } from "capnweb";
+import type {
+  ActionLogEntry, AuthenticatedApi, Overseer, PublicApi,
+} from "@gadgets/workshop-shared/api";
+import { startTestGatekeeperHarness, TEST_VENDOR_ID, type Harness } from "../src/harness.js";
+import {
+  connect, listConnectedAccounts, nextUsernames, signUp, waitFor, type ConnectedAccount,
+} from "../src/rpc-client.js";
+import { NetworkInterceptor } from "../src/network-interceptor.js";
+
+const POKE = { tag: "poke", label: "Pokes" };
+
+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<T>(body: (api: RpcStub<PublicApi>) => Promise<T>): Promise<T> {
+  const publicApi = connect(harness.url);
+  try {
+    return await body(publicApi);
+  } finally {
+    publicApi[Symbol.dispose]();
+  }
+}
+
+async function provisionAccount(api: RpcStub<AuthenticatedApi>): Promise<ConnectedAccount> {
+  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;
+  });
+}
+
+type Workspace = {
+  overseer: RpcStub<Overseer>;
+  session: any;
+  gatekeeperId: number;
+};
+
+async function newWorkspace(publicApi: RpcStub<PublicApi>, thingName: string): Promise<Workspace> {
+  const [alice] = nextUsernames("alice");
+  const aliceApi = await signUp(publicApi, alice);
+  const account = await provisionAccount(aliceApi);
+  const overseer = await aliceApi.newGadget();
+  const gatekeeper = await overseer.newGatekeeper(
+      account.id, `https://gadgets-test.example/things/${thingName}`);
+  if (!gatekeeper) throw new Error("Failed to create the test connection");
+  return {
+    overseer,
+    session: await gatekeeper.openSession(),
+    gatekeeperId: await gatekeeper.getId(),
+  };
+}
+
+async function listPokes(ws: Workspace): Promise<Array<ActionLogEntry & { type: "action" }>> {
+  const actions = await ws.overseer.listActions();
+  return actions.filter(
+      (a): a is ActionLogEntry & { type: "action" } =>
+          a.type === "action" && a.gatekeeperId === ws.gatekeeperId);
+}
+
+// The drain runs via ctx.waitUntil after submit, so "did not auto-approve" needs a settle window.
+// One further RPC round trip plus a beat is far beyond the drain's synchronous storage work.
+async function settle(ws: Workspace): Promise<void> {
+  await ws.overseer.listActions();
+  await new Promise(resolve => setTimeout(resolve, 300));
+}
+
+describe("auto-approval policy", () => {
+  it.concurrent("auto-approves a rule-enabled, author-approvable action", async () => {
+    await withSession(async publicApi => {
+      const ws = await newWorkspace(publicApi, "auto-happy");
+      await ws.overseer.setAutoApprovedActionKind(ws.gatekeeperId, POKE);
+      await ws.session.doThing({ autoApprovable: true });
+
+      const applied = await waitFor("the poke to be auto-approved", async () => {
+        const [poke] = await listPokes(ws);
+        return poke?.state === "approved" ? poke : null;
+      });
+      expect(applied.autoApproved).toBe(true);
+    });
+  });
+
+  it.concurrent("a warned action holds the queue until a human approves it", async () => {
+    await withSession(async publicApi => {
+      const ws = await newWorkspace(publicApi, "warned");
+      await ws.overseer.setAutoApprovedActionKind(ws.gatekeeperId, POKE);
+
+      // The warned action is fully rule-covered and author-approved -- only the warning stands
+      // between it and auto-application. A clean eligible action behind it must not be applied
+      // either (the drain never skips ahead of a manual gate).
+      await ws.session.doThing({ autoApprovable: true, warnings: ["Cross-account data risk."] });
+      await ws.session.doThing({ autoApprovable: true });
+      await settle(ws);
+
+      const pokes = await listPokes(ws);
+      expect(pokes.map(p => p.state)).toEqual(["pending", "pending"]);
+      expect(pokes[0].description.operatorWarnings).toEqual(["Cross-account data risk."]);
+
+      // A human approving the warned action clears the gate; the one behind it then auto-applies.
+      await ws.overseer.approveAction(pokes[0].id);
+      const drained = await waitFor("the queued poke to auto-apply", async () => {
+        const [first, second] = await listPokes(ws);
+        return first.state === "approved" && second?.state === "approved" ? [first, second] : null;
+      });
+      expect(drained[0].autoApproved).toBeFalsy();
+      expect(drained[1].autoApproved).toBe(true);
+    });
+  });
+
+  it.concurrent("a pre-latch auto-approval rule stops firing once the workspace reads sensitive data",
+      async () => {
+    await withSession(async publicApi => {
+      const ws = await newWorkspace(publicApi, "pre-latch");
+      await ws.overseer.setAutoApprovedActionKind(ws.gatekeeperId, POKE);
+      await ws.session.readThing(true);
+
+      // Writes-to-self still pend, but the rule the user enabled before the latch must not fire.
+      await ws.session.doThing({ autoApprovable: true });
+      await settle(ws);
+      const [poke] = await listPokes(ws);
+      expect(poke.state).toBe("pending");
+
+      // The rule surface refuses a latched workspace outright.
+      await expect(ws.overseer.setAutoApprovedActionKind(ws.gatekeeperId, POKE))
+          .rejects.toThrow(/cannot be auto-approved/i);
+      await expect(ws.overseer.listPreApprovableActions()).resolves.toEqual([]);
+
+      // Manual approval still works: the human is the intended path.
+      await ws.overseer.approveAction(poke.id);
+      const approved = await waitFor("the poke to be applied after manual approval", async () => {
+        const [fresh] = await listPokes(ws);
+        return fresh.state === "approved" ? fresh : null;
+      });
+      expect(approved.autoApproved).toBeFalsy();
+    });
+  });
+});

packages/integration-tests/__tests__/sensitive-observations.test.ts

@@ -3,8 +3,9 @@
 // A sensitive observation is blocked only while some *current collaborator* has not been verified
 // (via `addObserver`) against the gatekeeper producing it; sharing itself stays available, since
 // recipients are verified when they open. The observation also latches the workspace into a
-// restricted mode: once latched, the workspace may not perform actions (nor fetch from the web,
-// which has no client-reachable surface to assert here).
+// restricted mode: once latched, the workspace may only perform actions targeting the connections
+// that produced the sensitive data -- the writes-to-self carve-out -- and may not fetch from the
+// web (which has no client-reachable surface to assert here).
 //
 // The fixture gatekeeper's session drives all of this through the real ApprovalQueue funnel:
 // `readThing(true)` records a `containsRestrictedData` observation, `doThing()` submits an action.
@@ -136,7 +137,8 @@ async function bobOpens(gadgetId: string, bobApi: RpcStub<AuthenticatedApi>,
 }
 
 describe("sensitive observations", () => {
-  it.concurrent("latch restricted mode: actions are blocked and metadata reports it", async () => {
+  it.concurrent("latch restricted mode: only writes-to-self are allowed and metadata reports it",
+      async () => {
     await withSession(async publicApi => {
       const ws = await newWorkspace(publicApi, "latch");
 
@@ -147,13 +149,61 @@ describe("sensitive observations", () => {
       await expect(ws.session.readThing(true)).resolves.toContain("latch");
 
       expect((await ws.overseer.getMetadata()).containsRestrictedData).toBe(true);
-      await expect(ws.session.doThing()).rejects.toThrow(/prohibited from performing actions/i);
+
+      // Actions targeting the gatekeeper that produced the sensitive data still pend (the
+      // writes-to-self carve-out: the data came from there, so sending it back reveals nothing
+      // new). Actions on any other connection are blocked.
+      await expect(ws.session.doThing()).resolves.toBeUndefined();
+      const accounts = await listConnectedAccounts(ws.aliceApi);
+      const account = accounts.find(a => a.vendorId === TEST_VENDOR_ID)!;
+      const other = await ws.overseer.newGatekeeper(account.id, thingUrl("latch-other"));
+      if (!other) throw new Error("Failed to create the second test connection");
+      const otherSession: any = await other.openSession();
+      await expect(otherSession.doThing())
+          .rejects.toThrow(/only perform actions on those same connections/i);
+
       // Reads -- sensitive or not -- keep working.
       await expect(ws.session.readThing()).resolves.toContain("latch");
       await expect(ws.session.readThing(true)).resolves.toContain("latch");
     });
   });
 
+  it.concurrent("a pre-latch pending action on another connection cannot be approved after " +
+      "the latch", async () => {
+    await withSession(async publicApi => {
+      const ws = await newWorkspace(publicApi, "approve-after-latch");
+
+      // A second connection queues an action while the workspace is still unlatched, so
+      // submitAction's carve-out check admits it and it pends for manual approval.
+      const accounts = await listConnectedAccounts(ws.aliceApi);
+      const account = accounts.find(a => a.vendorId === TEST_VENDOR_ID)!;
+      const other = await ws.overseer.newGatekeeper(
+          account.id, thingUrl("approve-after-latch-other"));
+      if (!other) throw new Error("Failed to create the second test connection");
+      const otherGatekeeperId = await other.getId();
+      const otherSession: any = await other.openSession();
+      await expect(otherSession.doThing()).resolves.toBeUndefined();
+
+      // The first connection latches the workspace. The second connection is not a producer, so
+      // its still-pending action now violates the writes-to-self carve-out.
+      await expect(ws.session.readThing(true)).resolves.toContain("approve-after-latch");
+
+      const actions = await ws.overseer.listActions();
+      const pending = actions.find(
+          a => a.type === "action" && a.state === "pending" &&
+               a.gatekeeperId === otherGatekeeperId);
+      if (!pending) throw new Error("The pre-latch action is not pending");
+
+      // Approval is refused at the apply chokepoint; the action stays pending, and denying it --
+      // the way to unstick a suspended agent turn -- still works.
+      await expect(ws.overseer.approveAction(pending.id))
+          .rejects.toThrow(/only perform actions on those same connections/i);
+      await expect(ws.overseer.rejectAction(pending.id)).resolves.toBeUndefined();
+      const after = await ws.overseer.listActions();
+      expect(after.find(a => a.id === pending.id)?.state).toBe("rejected");
+    });
+  });
+
   it.concurrent("an unredeemed share link does not block a sensitive observation", async () => {
     await withSession(async publicApi => {
       const ws = await newWorkspace(publicApi, "unredeemed");
@@ -576,6 +626,27 @@ describe("sensitive observations", () => {
     });
   });
 
+  it.concurrent("a connection with pending approval requests cannot be removed", async () => {
+    await withSession(async publicApi => {
+      const ws = await newWorkspace(publicApi, "remove-pending");
+
+      // Queue an action; it pends for manual approval. Removing the connection now would delete
+      // the facet both approval and rejection resolve through, stranding the record forever.
+      await expect(ws.session.doThing()).resolves.toBeUndefined();
+      const gatekeeper = await ws.overseer.getGatekeeperById(ws.gatekeeperId);
+      await expect(gatekeeper.remove()).rejects.toThrow(/pending approval requests/i);
+      // The refused removal left the connection intact.
+      await expect(ws.session.readThing()).resolves.toContain("remove-pending");
+
+      // Denying the action resolves it, which unblocks the removal.
+      const actions = await ws.overseer.listActions();
+      const pending = actions.find(a => a.type === "action" && a.state === "pending");
+      if (!pending) throw new Error("The submitted action is not pending");
+      await ws.overseer.rejectAction(pending.id);
+      await expect(gatekeeper.remove()).resolves.toBeUndefined();
+    });
+  });
+
   it.concurrent("a latched connection cannot be removed while the workspace is shared",
       async () => {
     await withSession(async publicApi => {

packages/integration-tests/fixtures/gatekeeper-test/src/test-gatekeeper.ts

@@ -235,7 +235,8 @@ export class TestVerifier
  * 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
  * (optionally marked `containsRestrictedData`, to trip the sensitive-data coverage guard and the
- * restricted-mode latch), and `doThing()` submits an action (which restricted mode blocks).
+ * restricted-mode latch), and `doThing()` submits an action (which restricted mode allows only
+ * back to the producing connections, never auto-approved).
  */
 export class TestSession extends RpcTarget {
   #queue: RpcStub<ApprovalQueue>;
@@ -256,11 +257,17 @@ export class TestSession extends RpcTarget {
     return `the contents of ${this.#title}`;
   }
 
-  async doThing(): Promise<void> {
+  async doThing(opts?: { autoApprovable?: boolean; warnings?: string[] }): Promise<void> {
     await this.#queue.submitAction(0, {
       title: `Poke ${this.#title}`,
       description: `The test poked ${this.#title}.`,
       implementsRevert: false,
+      // Always tagged, so a test can enable an auto-approval rule for it; whether this specific
+      // poke is eligible is the per-action verdict below, exactly as a shipping gatekeeper would
+      // set it.
+      actionKind: POKE_ACTION_KIND,
+      ...(opts?.autoApprovable ? { autoApprovable: true } : {}),
+      ...(opts?.warnings ? { operatorWarnings: opts.warnings } : {}),
     });
   }
 
@@ -270,6 +277,8 @@ export class TestSession extends RpcTarget {
   }
 }
 
+const POKE_ACTION_KIND: ActionKind = { tag: "poke", label: "Pokes" };
+
 export class TestGatekeeper
     extends DurableObject<Cloudflare.Env, BindingProps> implements Gatekeeper<TestSession> {
   async describe(): Promise<ResourceDescription> {
@@ -298,7 +307,7 @@ export class TestGatekeeper
   }
 
   async getAutoApprovableActions(): Promise<ActionKind[]> {
-    return [];
+    return [POKE_ACTION_KIND];
   }
 
   async startSession(approvalQueue: RpcStub<ApprovalQueue>): Promise<TestSession> {
@@ -336,14 +345,18 @@ export class TestGatekeeper
     this.ctx.storage.kv.delete(`observer:${id}`);
   }
 
-  async applyAction(_action: number): Promise<void> {
-    throw new Error("The test gatekeeper submits no actions.");
+  /**
+   * Applying succeeds and leaves a record, so approval-path tests (manual and auto) can drive a
+   * submit -> approve -> apply round trip against real machinery.
+   */
+  async applyAction(action: number): Promise<void> {
+    this.ctx.storage.kv.put(`applied:${action}`, true);
   }
 
   async rejectAction(_action: number): Promise<void> {}
 
   async revertAction(_action: number): Promise<void> {
-    throw new Error("The test gatekeeper submits no actions.");
+    throw new Error("The test gatekeeper does not implement revert.");
   }
 }