feat: enforce session authorization and revoke stale sockets

packages/control-plane/README.md

@@ -57,7 +57,7 @@ The control plane provides:
 
 | Endpoint                        | Method    | Description                    |
 | ------------------------------- | --------- | ------------------------------ |
-| `/sessions`                     | GET       | List user's sessions           |
+| `/sessions`                     | GET       | List workspace sessions        |
 | `/sessions`                     | POST      | Create new session             |
 | `/sessions/:id`                 | GET       | Get canonical session snapshot |
 | `/sessions/:id`                 | DELETE    | Delete session                 |
@@ -67,7 +67,7 @@ The control plane provides:
 | `/sessions/:id/ws`              | WebSocket | Real-time connection           |
 | `/sessions/:id/events`          | GET       | Paginated events               |
 | `/sessions/:id/artifacts`       | GET       | List artifacts                 |
-| `/sessions/:id/participants`    | GET/POST  | Manage participants            |
+| `/sessions/:id/participants`    | GET       | List runtime participants      |
 | `/sessions/:id/messages`        | GET       | List messages                  |
 | `/sessions/:id/pr`              | POST      | Create pull request            |
 | `/sessions/:id/scm-credentials` | POST      | Broker sandbox git credentials |

packages/control-plane/src/auth/identity-enforcement.ts

@@ -20,12 +20,7 @@ import { error, json, type RequestContext } from "../routes/shared";
 const logger = createLogger("identity-enforcement");
 
 /** The route families that consume caller-supplied identity. */
-export type IdentityRoute =
-  | "session-create"
-  | "ws-token"
-  | "prompt"
-  | "session-lifecycle"
-  | "automation-create";
+type IdentityRoute = "session-create" | "ws-token" | "prompt" | "automation-create";
 
 const SPAWNING_FORBIDDEN_FIELDS = [
   "userId",
@@ -49,7 +44,6 @@ const FORBIDDEN_IDENTITY_FIELDS: Record<IdentityRoute, readonly string[]> = {
   "session-create": SPAWNING_FORBIDDEN_FIELDS,
   "ws-token": ["userId", "scmToken", "scmRefreshToken", "scmUserId"],
   prompt: ["authorId"],
-  "session-lifecycle": ["userId"],
   "automation-create": SPAWNING_FORBIDDEN_FIELDS,
 };
 
@@ -73,7 +67,7 @@ function requiresUserMessage(route: IdentityRoute): string | undefined {
 }
 
 /** Identity a verified principal implies for a consuming route. */
-export interface DerivedIdentity {
+interface DerivedIdentity {
   /** DO participant id: bare canonical id for users, `ns:id` for bot actors. */
   participantUserId: string | null;
   /** Canonical D1 users.id when the principal resolves to one. */
@@ -134,7 +128,7 @@ function isJsonObject(value: unknown): value is Record<string, unknown> {
   return typeof value === "object" && value !== null && !Array.isArray(value);
 }
 
-export type IdentityEnforcement<R extends IdentityRoute> =
+type IdentityEnforcement<R extends IdentityRoute> =
   | { rejection: Response; enforced?: undefined }
   | { rejection?: undefined; enforced: EnforcedIdentity<R> };
 

packages/control-plane/src/db/session-index.test.ts

@@ -750,43 +750,6 @@ describe("SessionIndexStore", () => {
       ]);
     });
 
-    it("trims and lowercases repo filters", async () => {
-      await store.create(makeSession({ id: "match", repoOwner: "Owner", repoName: "Repo" }));
-      await store.create(makeSession({ id: "other", repoOwner: "Other", repoName: "Repo" }));
-
-      const result = await store.list({ repoOwner: "  OWNER  ", repoName: "  REPO  " });
-
-      expect(result.sessions).toHaveLength(1);
-      expect(result.sessions[0].id).toBe("match");
-    });
-
-    it("matches sessions through secondary members, not just the scalar primary", async () => {
-      await store.create(
-        makeSession({
-          id: "multi",
-          repoOwner: "acme",
-          repoName: "frontend",
-          repositories: [
-            { repoOwner: "acme", repoName: "frontend", repoId: 1, baseBranch: "main" },
-            { repoOwner: "acme", repoName: "backend", repoId: 2, baseBranch: "main" },
-          ],
-        })
-      );
-      await store.create(makeSession({ id: "other", repoOwner: "acme", repoName: "unrelated" }));
-
-      const result = await store.list({ repoOwner: "acme", repoName: "backend" });
-
-      expect(result.sessions.map((s) => s.id)).toEqual(["multi"]);
-    });
-
-    it("falls back to the scalar columns for pre-feature sessions without member rows", async () => {
-      await store.create(makeSession({ id: "legacy", repoOwner: "acme", repoName: "app" }));
-
-      const result = await store.list({ repoOwner: "acme", repoName: "app" });
-
-      expect(result.sessions.map((s) => s.id)).toEqual(["legacy"]);
-    });
-
     it("supports multiple creator user ids", async () => {
       await store.create(makeSession({ id: "alice", userId: "alice", updatedAt: 1000 }));
       await store.create(makeSession({ id: "bob", userId: "bob", updatedAt: 3000 }));

packages/control-plane/src/db/session-index.ts

@@ -33,12 +33,6 @@ import { INACTIVE_SESSION_STATUS_SQL } from "@open-inspect/shared/types/session-
 import { readStateFromRow, unreadSql, type ViewerReadStateRow } from "./session-read-state";
 import type { SqlDatabase, SqlStatement } from "./sql-database";
 
-export type {
-  ListSessionInboxOptions,
-  ListSessionInboxResult,
-  ListSessionInboxSnapshotResult,
-} from "./session-inbox-store";
-
 const CHILD_ADMISSION_LEASE_TTL_MS = 5 * 60 * 1000;
 
 export interface ChildAdmissionLease {
@@ -60,8 +54,9 @@ const MAX_DESCENDANT_DEPTH = 10;
  * primary, mirrored into the scalar repo_owner/repo_name columns). Aliases
  * the shared wire type so Session.repositories and this share one shape.
  */
-export type SessionIndexRepository = SessionListRepository;
+type SessionIndexRepository = SessionListRepository;
 
+/** Persisted session metadata with optional viewer-specific read state. */
 export interface SessionEntry {
   id: string;
   title: string | null;
@@ -142,24 +137,24 @@ interface SessionModelProviderAuthRow {
   inherited_from_session_id: string | null;
 }
 
+/** Filters, pagination, and viewer read state for a session list query. */
 export interface ListSessionsOptions {
   status?: SessionStatus;
   excludeStatus?: SessionStatus;
   excludeAutomationLineage?: boolean;
-  repoOwner?: string;
-  repoName?: string;
   createdByUserIds?: readonly string[];
   limit?: number;
   offset?: number;
   viewerUserId?: string;
 }
 
+/** Paginated session index entries. */
 export interface ListSessionsResult {
   sessions: SessionEntry[];
   hasMore: boolean;
 }
 
-interface ViewerSessionRow extends SessionRow, ViewerReadStateRow {}
+type ViewerSessionRow = SessionRow & ViewerReadStateRow;
 
 function toEntry(row: SessionRow): SessionEntry {
   return {
@@ -236,6 +231,7 @@ function normalizeSessionRepositoryFields(session: SessionEntry): {
   };
 }
 
+/** D1-backed session index and viewer-specific list projection. */
 export class SessionIndexStore {
   constructor(private readonly db: SqlDatabase) {}
 
@@ -507,13 +503,12 @@ export class SessionIndexStore {
     return row !== null;
   }
 
+  /** List sessions with optional viewer-specific read state. */
   async list(options: ListSessionsOptions = {}): Promise<ListSessionsResult> {
     const {
       status,
       excludeStatus,
       excludeAutomationLineage,
-      repoOwner,
-      repoName,
       createdByUserIds,
       limit = DEFAULT_SESSION_LIST_LIMIT,
       offset = DEFAULT_SESSION_LIST_OFFSET,
@@ -541,39 +536,14 @@ export class SessionIndexStore {
       conditions.push("automation_id IS NULL AND spawn_source NOT IN ('automation', 'github-bot')");
     }
 
-    // Repo filters match against the membership table so a session is found
-    // through ANY member, not just the scalar primary mirror. The scalar arm
-    // is the fallback for pre-feature sessions without member rows.
-    const normalizedRepoOwner = normalizeRepoIdentifier(repoOwner);
-    const normalizedRepoName = normalizeRepoIdentifier(repoName);
-    if (normalizedRepoOwner || normalizedRepoName) {
-      const memberConditions: string[] = [];
-      const scalarConditions: string[] = [];
-      const repoFilterParams: unknown[] = [];
-      if (normalizedRepoOwner) {
-        memberConditions.push("sr.repo_owner = ?");
-        scalarConditions.push("repo_owner = ?");
-        repoFilterParams.push(normalizedRepoOwner);
-      }
-      if (normalizedRepoName) {
-        memberConditions.push("sr.repo_name = ?");
-        scalarConditions.push("repo_name = ?");
-        repoFilterParams.push(normalizedRepoName);
-      }
-      conditions.push(
-        `(EXISTS (SELECT 1 FROM session_repositories sr WHERE sr.session_id = sessions.id AND ${memberConditions.join(" AND ")}) OR (${scalarConditions.join(" AND ")}))`
-      );
-      params.push(...repoFilterParams, ...repoFilterParams);
-    }
-
     if (createdByUserIds?.length) {
       conditions.push(`user_id IN (${createdByUserIds.map(() => "?").join(", ")})`);
       params.push(...createdByUserIds);
     }
-
     const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : "";
 
     const pageSql = `SELECT * FROM sessions ${where} ORDER BY updated_at DESC LIMIT ? OFFSET ?`;
+    const pageParams = [...params, limit + 1, offset];
     const result = viewerUserId
       ? await this.db
           .prepare(
@@ -587,11 +557,11 @@ export class SessionIndexStore {
               AND read_state.user_id = viewer.id
              ORDER BY paged_sessions.updated_at DESC`
           )
-          .bind(...params, limit + 1, offset, viewerUserId)
+          .bind(...pageParams, viewerUserId)
           .all<ViewerSessionRow>()
       : await this.db
           .prepare(pageSql)
-          .bind(...params, limit + 1, offset)
+          .bind(...pageParams)
           .all<SessionRow>();
 
     const rows = result.results || [];
@@ -608,10 +578,12 @@ export class SessionIndexStore {
     };
   }
 
+  /** List one inbox category with viewer-specific read state. */
   async listInbox(options: ListSessionInboxOptions): Promise<ListSessionInboxResult> {
     return new SessionInboxStore(this.db).list(options);
   }
 
+  /** List the first page of every inbox category with viewer-specific read state. */
   async listInboxSnapshot(
     options: Omit<ListSessionInboxOptions, "category" | "cursor">
   ): Promise<ListSessionInboxSnapshotResult> {
@@ -657,11 +629,6 @@ export class SessionIndexStore {
     return (result.meta.changes ?? 0) > 0;
   }
 
-  /** Current single-tenant visibility boundary; future grants belong here. */
-  async getVisibleForUser(sessionId: string, _userId: string): Promise<SessionEntry | null> {
-    return this.get(sessionId);
-  }
-
   async updateReadState(
     userId: string,
     sessionId: string,

packages/control-plane/src/router.policy.test.ts

@@ -152,17 +152,21 @@ describe("route policy table", () => {
       allOf: [{ kind: "permission", permission: "sessions.read" }],
       service: { kind: "deny" },
     });
+    expect(routeFor("POST", "/sessions/session-1/ws-token")?.authorization).toMatchObject({
+      kind: "active-user",
+      allOf: [
+        { kind: "permission", permission: "sessions.read" },
+        { kind: "permission", permission: "sessions.collaborate" },
+        { kind: "permission", permission: "sessions.lifecycle" },
+      ],
+    });
     expect(routeFor("POST", "/sessions/session-1/stop")?.authorization).toMatchObject({
       service: { kind: "actor", actorlessGrants: [{ service: "linear-bot" }] },
     });
     expect(routeFor("GET", "/sessions/session-1/media/artifact-1")?.authorization).toMatchObject({
       service: { kind: "actor", actorlessGrants: [{ service: "slack-bot" }] },
     });
-    expect(routeFor("POST", "/sessions/session-1/participants")?.authorization).toEqual({
-      kind: "active-user",
-      allOf: [{ kind: "permission", permission: "sessions.collaborate" }],
-      service: { kind: "actor" },
-    });
+    expect(routeFor("POST", "/sessions/session-1/participants")).toBeUndefined();
     expect(routeFor("POST", "/sessions/parent/children")?.authorization).toMatchObject({
       kind: "active-user",
       allOf: [

packages/control-plane/src/routes/session-runtime-proxy.test.ts

@@ -342,12 +342,11 @@ describe("session runtime proxy routes", () => {
     expect(requests[0].method).toBe("POST");
     expect(new URL(requests[0].url).pathname).toBe(SessionInternalPaths.updateTitle);
     await expect(requests[0].json()).resolves.toEqual({
-      userId: "user-1",
       title: "New title",
     });
   });
 
-  it("forwards the verified service actor on title updates", async () => {
+  it("does not forward service actor identity on title updates", async () => {
     const requests: Request[] = [];
     const fetch = vi.fn(async (request: Request) => {
       requests.push(request);
@@ -380,7 +379,6 @@ describe("session runtime proxy routes", () => {
     expect(response.status).toBe(200);
     expect(fetch).toHaveBeenCalledOnce();
     await expect(requests[0].json()).resolves.toEqual({
-      userId: "slack:U0123",
       title: "New title",
     });
   });
@@ -437,26 +435,6 @@ describe("session runtime proxy routes", () => {
     await expect(response.json()).resolves.toEqual({ error: "Session not found" });
   });
 
-  it("rejects malformed add-participant JSON without forwarding to the runtime", async () => {
-    const fetch = vi.fn(async () => Response.json({ status: "ok" }));
-    const { handler, match } = getHandler("POST", "/sessions/session-1/participants");
-
-    const response = await handler(
-      new Request("https://test.local/sessions/session-1/participants", {
-        method: "POST",
-        headers: { "Content-Type": "application/json" },
-        body: "{",
-      }),
-      createEnv(fetch),
-      match,
-      createCtx()
-    );
-
-    expect(response.status).toBe(400);
-    await expect(response.json()).resolves.toEqual({ error: "Invalid JSON body" });
-    expect(fetch).not.toHaveBeenCalled();
-  });
-
   it("forwards the draft flag through the create-PR contract", async () => {
     const requests: Request[] = [];
     const fetch = vi.fn(async (request: Request) => {