diff --git a/apps/api/src/db/migrations/1785714869_optio_actions_workspace_id.sql b/apps/api/src/db/migrations/1785714869_optio_actions_workspace_id.sql new file mode 100644 index 00000000..ed5d358a --- /dev/null +++ b/apps/api/src/db/migrations/1785714869_optio_actions_workspace_id.sql @@ -0,0 +1,14 @@ +-- Scope the Optio action audit trail to a workspace. +-- +-- The unified activity feed (GET /api/activity) previously returned every +-- tenant's actions to any authenticated user, and the logged `params` jsonb +-- could carry other tenants' resource ids (and, before the write-time +-- allowlist, request-body values). Actions are now stamped with the caller's +-- workspace so the feed can filter by tenant. +-- +-- Existing rows keep workspace_id = NULL. NULL is treated as an +-- operator/legacy action with no tenant context and is only surfaced to +-- admins by the activity feed (deny-by-default) — never to members/viewers. + +ALTER TABLE "optio_actions" ADD COLUMN "workspace_id" uuid;--> statement-breakpoint +CREATE INDEX "optio_actions_workspace_id_idx" ON "optio_actions" ("workspace_id"); diff --git a/apps/api/src/db/migrations/meta/_journal.json b/apps/api/src/db/migrations/meta/_journal.json index 85a71615..a350b8bd 100644 --- a/apps/api/src/db/migrations/meta/_journal.json +++ b/apps/api/src/db/migrations/meta/_journal.json @@ -554,6 +554,13 @@ "when": 1784588891000, "tag": "1784588891_backfill_task_workspace_id", "breakpoints": true + }, + { + "idx": 79, + "version": "7", + "when": 1785714869000, + "tag": "1785714869_optio_actions_workspace_id", + "breakpoints": true } ] } diff --git a/apps/api/src/db/schema.ts b/apps/api/src/db/schema.ts index 8b71f29b..3924f2b9 100644 --- a/apps/api/src/db/schema.ts +++ b/apps/api/src/db/schema.ts @@ -945,9 +945,12 @@ export const optioActions = pgTable( "optio_actions", { id: uuid("id").primaryKey().defaultRandom(), + // Nullable: NULL rows are operator/legacy actions with no tenant context and + // are only surfaced to admins by the activity feed (deny-by-default). + workspaceId: uuid("workspace_id"), userId: uuid("user_id").references(() => users.id), action: text("action").notNull(), // tool name e.g. "retry_task", "bulk_cancel_active" - params: jsonb("params").$type>(), // sanitized tool call parameters + params: jsonb("params").$type>(), // allowlisted, non-secret tool call parameters result: jsonb("result").$type>(), // outcome: affected IDs, error, etc. success: boolean("success").notNull(), conversationSnippet: text("conversation_snippet"), // user message that triggered this @@ -957,6 +960,7 @@ export const optioActions = pgTable( index("optio_actions_user_id_idx").on(table.userId), index("optio_actions_action_idx").on(table.action), index("optio_actions_created_at_idx").on(table.createdAt.desc()), + index("optio_actions_workspace_id_idx").on(table.workspaceId), ], ); diff --git a/apps/api/src/routes/activity.test.ts b/apps/api/src/routes/activity.test.ts index f8c3c052..b8b8c037 100644 --- a/apps/api/src/routes/activity.test.ts +++ b/apps/api/src/routes/activity.test.ts @@ -1,5 +1,7 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; import type { FastifyInstance } from "fastify"; +import type { SQL } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; import { buildRouteTestApp } from "../test-utils/build-route-test-app.js"; const mockDbExecute = vi.fn(); @@ -152,3 +154,129 @@ describe("GET /api/activity", () => { expect(res.json().error).toBe("Failed to fetch activity feed"); }); }); + +describe("GET /api/activity — workspace isolation & param whitelisting", () => { + const dialect = new PgDialect(); + + /** Compile the SQL fragment handed to the mocked db.execute into text + params. */ + function compile(call: unknown): { text: string; params: unknown[] } { + const q = dialect.sqlToQuery(call as SQL); + return { text: q.sql, params: q.params }; + } + + beforeEach(() => { + vi.clearAllMocks(); + // requireRole("member") must actually enforce — never let a stray env + // disable auth for these assertions. + delete process.env.OPTIO_AUTH_DISABLED; + }); + + it("scopes a member strictly to their own workspace (no null-workspace rows)", async () => { + mockDbExecute + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([{ total: 0 }]) + .mockResolvedValueOnce([]); + + const app = await buildRouteTestApp((await import("./activity.js")).activityRoutes, { + user: { id: "u-A", workspaceId: "ws-A", workspaceRole: "member" }, + }); + const res = await app.inject({ method: "GET", url: "/api/activity?type=action" }); + expect(res.statusCode).toBe(200); + + const { text, params } = compile(mockDbExecute.mock.calls[0][0]); + // Query is scoped by workspace_id and bound to the caller's workspace only. + expect(text).toContain("workspace_id"); + expect(params).toContain("ws-A"); + // Members get strict equality — NOT the admin `OR workspace_id IS NULL` branch. + expect(text.toLowerCase()).not.toContain("is null"); + }); + + it("lets an admin additionally see legacy null-workspace rows", async () => { + mockDbExecute + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([{ total: 0 }]) + .mockResolvedValueOnce([]); + + const app = await buildRouteTestApp((await import("./activity.js")).activityRoutes, { + user: { id: "u-A", workspaceId: "ws-A", workspaceRole: "admin" }, + }); + const res = await app.inject({ method: "GET", url: "/api/activity?type=action" }); + expect(res.statusCode).toBe(200); + + const { text, params } = compile(mockDbExecute.mock.calls[0][0]); + expect(params).toContain("ws-A"); + // Admins see own-workspace + operator/legacy (null) rows. + expect(text.toLowerCase()).toContain("workspace_id is null"); + }); + + it("does not leak another workspace's data: bound workspace is the caller's", async () => { + mockDbExecute + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([{ total: 0 }]) + .mockResolvedValueOnce([]); + + const app = await buildRouteTestApp((await import("./activity.js")).activityRoutes, { + user: { id: "u-A", workspaceId: "ws-A", workspaceRole: "member" }, + }); + await app.inject({ method: "GET", url: "/api/activity?type=action" }); + + const { params } = compile(mockDbExecute.mock.calls[0][0]); + // The only workspace ever bound is the caller's — workspace B is unreachable. + expect(params).toContain("ws-A"); + expect(params).not.toContain("ws-B"); + }); + + it("strips secret-bearing params from the response, keeping only allowlisted keys", async () => { + const now = new Date().toISOString(); + mockDbExecute + .mockResolvedValueOnce([ + { + id: "a1", + type: "action", + timestamp: now, + user_id: "u-A", + user_display_name: "Alice", + user_avatar_url: null, + action: "connection.update", + resource_type: "connection", + resource_id: "c1", + summary: "connection.update succeeded", + // Legacy row written before write-time filtering: full jsonb with secrets. + details: { + connectionId: "c1", + name: "prod-db", + apiToken: "SHOULD_NOT_LEAK", + password: "hunter2", + config: { url: "postgres://user:hunter2@db/app" }, + crossTenantResourceId: "ws-B-secret-id", + }, + }, + ]) + .mockResolvedValueOnce([{ total: 1 }]) + .mockResolvedValueOnce([{ type: "action", cnt: 1 }]); + + const app = await buildRouteTestApp((await import("./activity.js")).activityRoutes, { + user: { id: "u-A", workspaceId: "ws-A", workspaceRole: "admin" }, + }); + const res = await app.inject({ method: "GET", url: "/api/activity" }); + expect(res.statusCode).toBe(200); + + const details = res.json().items[0].details; + expect(details).toEqual({ connectionId: "c1", name: "prod-db" }); + // The whole serialized response must not carry the secret values. + const raw = res.payload; + expect(raw).not.toContain("SHOULD_NOT_LEAK"); + expect(raw).not.toContain("hunter2"); + expect(raw).not.toContain("crossTenantResourceId"); + }); + + it("forbids viewers (read-only) from reading the audit feed", async () => { + const app = await buildRouteTestApp((await import("./activity.js")).activityRoutes, { + user: { id: "u-v", workspaceId: "ws-A", workspaceRole: "viewer" }, + }); + const res = await app.inject({ method: "GET", url: "/api/activity" }); + expect(res.statusCode).toBe(403); + // Never touched the database. + expect(mockDbExecute).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/api/src/routes/activity.ts b/apps/api/src/routes/activity.ts index 13f8dcee..0087ad74 100644 --- a/apps/api/src/routes/activity.ts +++ b/apps/api/src/routes/activity.ts @@ -1,8 +1,20 @@ import type { FastifyInstance } from "fastify"; import type { ZodTypeProvider } from "fastify-type-provider-zod"; import { z } from "zod"; -import { sql } from "drizzle-orm"; +import { sql, type SQL } from "drizzle-orm"; import { db } from "../db/client.js"; +import { requireRole } from "../plugins/auth.js"; +import { ACTION_PARAM_ALLOWLIST } from "../services/optio-action-service.js"; + +/** Keep only allowlisted, non-secret keys from an action's stored params. */ +function whitelistActionDetails(details: unknown): Record | null { + if (!details || typeof details !== "object" || Array.isArray(details)) return null; + const clean: Record = {}; + for (const [k, v] of Object.entries(details as Record)) { + if (ACTION_PARAM_ALLOWLIST.has(k)) clean[k] = v; + } + return clean; +} const activityQuerySchema = z .object({ @@ -69,17 +81,23 @@ export async function activityRoutes(rawApp: FastifyInstance) { app.get( "/api/activity", { + // Action details can reveal resource ids and operation history; viewers + // are read-only observers and must not see the audit trail. + preHandler: [requireRole("member")], schema: { operationId: "getActivityFeed", summary: "Get unified workspace activity feed", description: "Merges user actions, task state transitions, auth events, and " + "infrastructure events into a single chronologically sorted feed. " + - "Supports filtering by type, user, and resource type.", + "Scoped to the caller's workspace. Auth/infrastructure events (which " + + "have no tenant column) and legacy null-workspace actions are only " + + "shown to admins. Supports filtering by type, user, and resource type.", tags: ["System"], querystring: activityQuerySchema, response: { 200: ActivityResponseSchema, + 403: z.object({ error: z.string() }), 500: z.object({ error: z.string() }), }, }, @@ -88,103 +106,115 @@ export async function activityRoutes(rawApp: FastifyInstance) { const { days, type, userId, resourceType, limit, offset } = req.query; const since = new Date(Date.now() - days * 86_400_000).toISOString(); + // Workspace scoping. Auth-disabled dev mode leaves req.user undefined — + // treat that as an operator with a full, unscoped view (local dev only). + const wsId = req.user?.workspaceId ?? null; + const isOperator = !req.user; + const isAdmin = isOperator || req.user?.workspaceRole === "admin"; + + // Restrict a tenant-scoped source (`col`) to the caller's workspace. + // Admins additionally see legacy/operator rows with a null workspace; + // members see strictly their own workspace. Operators (auth disabled) + // get no filter at all. + const scopeTo = (col: SQL): SQL | null => { + if (isOperator) return null; + return isAdmin ? sql`(${col} = ${wsId} OR ${col} IS NULL)` : sql`${col} = ${wsId}`; + }; + try { - // Build individual CTEs for each event source, applying filters - const parts: string[] = []; + // Build individual sub-selects for each event source, applying filters. + const parts: SQL[] = []; const typeFilters = type ? [type] : ["action", "task_event", "auth_event", "infra_event"]; if (typeFilters.includes("action")) { - const actionWhere = [`oa.created_at >= '${since}'`]; - if (userId) actionWhere.push(`oa.user_id = '${userId}'`); - if (resourceType) actionWhere.push(`split_part(oa.action, '.', 1) = '${resourceType}'`); - parts.push(` + const conds: SQL[] = [sql`oa.created_at >= ${since}`]; + const wsCond = scopeTo(sql`oa.workspace_id`); + if (wsCond) conds.push(wsCond); + if (userId) conds.push(sql`oa.user_id = ${userId}`); + if (resourceType) conds.push(sql`split_part(oa.action, '.', 1) = ${resourceType}`); + parts.push(sql` SELECT - oa.id::text, + oa.id::text AS id, 'action' AS type, oa.created_at AS timestamp, - oa.user_id, + oa.user_id AS user_id, u.display_name AS user_display_name, u.avatar_url AS user_avatar_url, - oa.action, + oa.action AS action, split_part(oa.action, '.', 1) AS resource_type, - COALESCE((oa.params->>'id')::text, (oa.params->>'taskId')::text, (oa.params->>'repoId')::text) AS resource_id, + COALESCE(oa.params->>'id', oa.params->>'taskId', oa.params->>'repoId') AS resource_id, oa.action || ' ' || CASE WHEN oa.success THEN 'succeeded' ELSE 'failed' END AS summary, oa.params AS details FROM optio_actions oa LEFT JOIN users u ON oa.user_id = u.id - WHERE ${actionWhere.join(" AND ")} + WHERE ${sql.join(conds, sql` AND `)} `); } - if (typeFilters.includes("task_event")) { - const teWhere = [`te.created_at >= '${since}'`]; - if (userId) teWhere.push(`te.user_id = '${userId}'`); - if (resourceType && resourceType !== "task") { - // task_events are always about tasks, skip if filtering for other types - } else { - parts.push(` - SELECT - te.id::text, - 'task_event' AS type, - te.created_at AS timestamp, - te.user_id, - u.display_name AS user_display_name, - u.avatar_url AS user_avatar_url, - 'task:' || COALESCE(te.from_state, 'new') || '→' || te.to_state AS action, - 'task' AS resource_type, - te.task_id::text AS resource_id, - 'Task transitioned to ' || te.to_state || ' via ' || te.trigger AS summary, - jsonb_build_object('fromState', te.from_state, 'toState', te.to_state, 'trigger', te.trigger) AS details - FROM task_events te - LEFT JOIN users u ON te.user_id = u.id - WHERE ${teWhere.join(" AND ")} - `); - } + // task_events are always about tasks; skip when filtering other types. + if (typeFilters.includes("task_event") && (!resourceType || resourceType === "task")) { + const conds: SQL[] = [sql`te.created_at >= ${since}`]; + const wsCond = scopeTo(sql`t.workspace_id`); + if (wsCond) conds.push(wsCond); + if (userId) conds.push(sql`te.user_id = ${userId}`); + parts.push(sql` + SELECT + te.id::text AS id, + 'task_event' AS type, + te.created_at AS timestamp, + te.user_id AS user_id, + u.display_name AS user_display_name, + u.avatar_url AS user_avatar_url, + 'task:' || COALESCE(te.from_state, 'new') || '→' || te.to_state AS action, + 'task' AS resource_type, + te.task_id::text AS resource_id, + 'Task transitioned to ' || te.to_state || ' via ' || te.trigger AS summary, + jsonb_build_object('fromState', te.from_state, 'toState', te.to_state, 'trigger', te.trigger) AS details + FROM task_events te + JOIN tasks t ON te.task_id = t.id + LEFT JOIN users u ON te.user_id = u.id + WHERE ${sql.join(conds, sql` AND `)} + `); } - if (typeFilters.includes("auth_event") && !resourceType) { - const aeWhere = [`ae.created_at >= '${since}'`]; - // auth_events have no userId, skip if filtering by user - if (!userId) { - parts.push(` - SELECT - ae.id::text, - 'auth_event' AS type, - ae.created_at AS timestamp, - NULL::uuid AS user_id, - NULL AS user_display_name, - NULL AS user_avatar_url, - 'auth:' || ae.token_type || '_failed' AS action, - 'auth' AS resource_type, - NULL AS resource_id, - ae.token_type || ' auth failed: ' || ae.error_message AS summary, - jsonb_build_object('tokenType', ae.token_type, 'error', ae.error_message) AS details - FROM auth_events ae - WHERE ${aeWhere.join(" AND ")} - `); - } + // auth_events and pod_health_events have no tenant column — they are + // deployment-global. Only admins (and dev operators) may see them. + if (typeFilters.includes("auth_event") && !resourceType && !userId && isAdmin) { + parts.push(sql` + SELECT + ae.id::text AS id, + 'auth_event' AS type, + ae.created_at AS timestamp, + NULL::uuid AS user_id, + NULL AS user_display_name, + NULL AS user_avatar_url, + 'auth:' || ae.token_type || '_failed' AS action, + 'auth' AS resource_type, + NULL AS resource_id, + ae.token_type || ' auth failed: ' || ae.error_message AS summary, + jsonb_build_object('tokenType', ae.token_type, 'error', ae.error_message) AS details + FROM auth_events ae + WHERE ae.created_at >= ${since} + `); } - if (typeFilters.includes("infra_event") && !resourceType) { - const ieWhere = [`phe.created_at >= '${since}'`]; - if (!userId) { - parts.push(` - SELECT - phe.id::text, - 'infra_event' AS type, - phe.created_at AS timestamp, - NULL::uuid AS user_id, - NULL AS user_display_name, - NULL AS user_avatar_url, - 'pod:' || phe.event_type AS action, - 'pod' AS resource_type, - phe.repo_pod_id::text AS resource_id, - 'Pod ' || COALESCE(phe.pod_name, 'unknown') || ' ' || phe.event_type AS summary, - jsonb_build_object('eventType', phe.event_type, 'podName', phe.pod_name, 'message', phe.message) AS details - FROM pod_health_events phe - WHERE ${ieWhere.join(" AND ")} - `); - } + if (typeFilters.includes("infra_event") && !resourceType && !userId && isAdmin) { + parts.push(sql` + SELECT + phe.id::text AS id, + 'infra_event' AS type, + phe.created_at AS timestamp, + NULL::uuid AS user_id, + NULL AS user_display_name, + NULL AS user_avatar_url, + 'pod:' || phe.event_type AS action, + 'pod' AS resource_type, + phe.repo_pod_id::text AS resource_id, + 'Pod ' || COALESCE(phe.pod_name, 'unknown') || ' ' || phe.event_type AS summary, + jsonb_build_object('eventType', phe.event_type, 'podName', phe.pod_name, 'message', phe.message) AS details + FROM pod_health_events phe + WHERE phe.created_at >= ${since} + `); } if (parts.length === 0) { @@ -195,27 +225,21 @@ export async function activityRoutes(rawApp: FastifyInstance) { }); } - const unionQuery = parts.join(" UNION ALL "); + const unionQuery = sql.join(parts, sql` UNION ALL `); // Get paginated results const [rows, countRows, statsRows] = await Promise.all([ - db.execute( - sql.raw(` + db.execute(sql` SELECT * FROM (${unionQuery}) AS activity ORDER BY timestamp DESC LIMIT ${limit} OFFSET ${offset} `), - ), - db.execute( - sql.raw(` + db.execute(sql` SELECT count(*)::int AS total FROM (${unionQuery}) AS activity `), - ), - db.execute( - sql.raw(` + db.execute(sql` SELECT type, count(*)::int AS cnt FROM (${unionQuery}) AS activity GROUP BY type `), - ), ]); const total = (countRows[0] as any)?.total ?? 0; @@ -243,7 +267,12 @@ export async function activityRoutes(rawApp: FastifyInstance) { resourceType: row.resource_type, resourceId: row.resource_id ?? null, summary: row.summary, - details: row.details ?? null, + // Action rows carry user-supplied params — reduce to the non-secret + // allowlist so legacy rows (written before write-time filtering) + // can't leak their full jsonb. Other sources build fixed, safe + // detail objects and pass through unchanged. + details: + row.type === "action" ? whitelistActionDetails(row.details) : (row.details ?? null), })); reply.send({ items, total, stats }); diff --git a/apps/api/src/routes/bulk.ts b/apps/api/src/routes/bulk.ts index 2f27fed9..56e6227d 100644 --- a/apps/api/src/routes/bulk.ts +++ b/apps/api/src/routes/bulk.ts @@ -71,6 +71,7 @@ export async function bulkRoutes(rawApp: FastifyInstance) { } } logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.bulk_retry", params: {}, @@ -126,6 +127,7 @@ export async function bulkRoutes(rawApp: FastifyInstance) { } } logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.bulk_cancel", params: {}, diff --git a/apps/api/src/routes/mcp-servers.ts b/apps/api/src/routes/mcp-servers.ts index 932c151e..602997a3 100644 --- a/apps/api/src/routes/mcp-servers.ts +++ b/apps/api/src/routes/mcp-servers.ts @@ -102,6 +102,7 @@ export async function mcpServerRoutes(rawApp: FastifyInstance) { const workspaceId = req.user?.workspaceId ?? null; const server = await mcpService.createMcpServer(req.body, workspaceId); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "mcp_server.create", params: { name: req.body.name, command: req.body.command }, @@ -135,6 +136,7 @@ export async function mcpServerRoutes(rawApp: FastifyInstance) { } const server = await mcpService.updateMcpServer(id, req.body); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "mcp_server.update", params: { mcpServerId: id, ...req.body }, @@ -167,6 +169,7 @@ export async function mcpServerRoutes(rawApp: FastifyInstance) { } await mcpService.deleteMcpServer(id); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "mcp_server.delete", params: { mcpServerId: id }, @@ -226,6 +229,7 @@ export async function mcpServerRoutes(rawApp: FastifyInstance) { workspaceId, ); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "mcp_server.create", params: { name: req.body.name, repoId: id }, diff --git a/apps/api/src/routes/optio-settings.ts b/apps/api/src/routes/optio-settings.ts index 9137c9f0..62605549 100644 --- a/apps/api/src/routes/optio-settings.ts +++ b/apps/api/src/routes/optio-settings.ts @@ -87,6 +87,7 @@ export async function optioSettingsRoutes(rawApp: FastifyInstance) { const workspaceId = req.user?.workspaceId ?? null; const settings = await optioSettingsService.upsertSettings(body, workspaceId); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "settings.update", params: { ...body }, diff --git a/apps/api/src/routes/pr-reviews.ts b/apps/api/src/routes/pr-reviews.ts index d2eb03dd..91e4d331 100644 --- a/apps/api/src/routes/pr-reviews.ts +++ b/apps/api/src/routes/pr-reviews.ts @@ -210,6 +210,7 @@ export async function prReviewRoutes(rawApp: FastifyInstance) { origin: "manual", }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "pr_review.launch", params: { prUrl: req.body.prUrl }, diff --git a/apps/api/src/routes/repos.ts b/apps/api/src/routes/repos.ts index 9e367ce8..aadb6b09 100644 --- a/apps/api/src/routes/repos.ts +++ b/apps/api/src/routes/repos.ts @@ -226,6 +226,7 @@ export async function repoRoutes(rawApp: FastifyInstance) { } logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "repo.create", params: { repoUrl: body.repoUrl, fullName: body.fullName }, @@ -319,6 +320,7 @@ export async function repoRoutes(rawApp: FastifyInstance) { const repo = await repoService.updateRepo(id, body); if (!repo) return reply.status(404).send({ error: "Repo not found" }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "repo.update", params: { repoId: id, ...body }, @@ -354,6 +356,7 @@ export async function repoRoutes(rawApp: FastifyInstance) { } await repoService.deleteRepo(id); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "repo.delete", params: { repoId: id, repoUrl: existing.repoUrl }, diff --git a/apps/api/src/routes/secrets.ts b/apps/api/src/routes/secrets.ts index 12e83420..51711881 100644 --- a/apps/api/src/routes/secrets.ts +++ b/apps/api/src/routes/secrets.ts @@ -189,6 +189,7 @@ export async function secretRoutes(rawApp: FastifyInstance) { } logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "secret.upsert", params: { name: input.name, scope: effectiveScope }, @@ -230,6 +231,7 @@ export async function secretRoutes(rawApp: FastifyInstance) { await secretService.deleteSecret(name, scope, workspaceId, effectiveUserId); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "secret.delete", params: { name }, diff --git a/apps/api/src/routes/sessions.ts b/apps/api/src/routes/sessions.ts index 1c7e121f..12ebb86a 100644 --- a/apps/api/src/routes/sessions.ts +++ b/apps/api/src/routes/sessions.ts @@ -265,6 +265,7 @@ export async function sessionRoutes(rawApp: FastifyInstance) { workspaceId: req.user?.workspaceId ?? null, }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "session.create", params: { repoUrl: input.repoUrl }, @@ -305,6 +306,7 @@ export async function sessionRoutes(rawApp: FastifyInstance) { try { const updated = await sessionService.endSession(id); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "session.end", params: { sessionId: id }, diff --git a/apps/api/src/routes/task-configs.ts b/apps/api/src/routes/task-configs.ts index 343094c6..52139c36 100644 --- a/apps/api/src/routes/task-configs.ts +++ b/apps/api/src/routes/task-configs.ts @@ -107,6 +107,7 @@ export async function taskConfigRoutes(rawApp: FastifyInstance) { createdBy: req.user?.id ?? null, }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task_config.create", params: { name: input.name }, @@ -175,6 +176,7 @@ export async function taskConfigRoutes(rawApp: FastifyInstance) { const taskConfig = await taskConfigService.updateTaskConfig(id, req.body); if (!taskConfig) return reply.status(404).send({ error: "Task config not found" }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: req.body.enabled !== undefined @@ -218,6 +220,7 @@ export async function taskConfigRoutes(rawApp: FastifyInstance) { await taskConfigService.deleteTaskConfig(id); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task_config.delete", params: { taskConfigId: id }, @@ -353,6 +356,7 @@ export async function taskConfigRoutes(rawApp: FastifyInstance) { enabled: input.enabled, }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task_config_trigger.create", params: { taskConfigId: id, type: input.type }, @@ -415,6 +419,7 @@ export async function taskConfigRoutes(rawApp: FastifyInstance) { const updated = await taskConfigService.updateTaskConfigTrigger(triggerId, req.body); if (!updated) return reply.status(404).send({ error: "Trigger not found" }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task_config_trigger.update", params: { taskConfigId: id, triggerId }, @@ -462,6 +467,7 @@ export async function taskConfigRoutes(rawApp: FastifyInstance) { await taskConfigService.deleteTaskConfigTrigger(triggerId); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task_config_trigger.delete", params: { taskConfigId: id, triggerId }, @@ -501,6 +507,7 @@ export async function taskConfigRoutes(rawApp: FastifyInstance) { try { const task = await taskConfigService.instantiateTask(id); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task_config.run", params: { taskConfigId: id }, diff --git a/apps/api/src/routes/tasks-unified.ts b/apps/api/src/routes/tasks-unified.ts index 2a3c65f7..a82c0536 100644 --- a/apps/api/src/routes/tasks-unified.ts +++ b/apps/api/src/routes/tasks-unified.ts @@ -157,6 +157,7 @@ export async function tasksUnifiedRoutes(rawApp: FastifyInstance) { params: body.params ?? undefined, }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.run", params: { type: parent.type, parentId: parent.data.id }, @@ -171,6 +172,7 @@ export async function tasksUnifiedRoutes(rawApp: FastifyInstance) { params: body.params ?? undefined, }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.run", params: { type: parent.type, parentId: parent.data.id }, diff --git a/apps/api/src/routes/tasks.ts b/apps/api/src/routes/tasks.ts index fa0e43cc..e75d2f2e 100644 --- a/apps/api/src/routes/tasks.ts +++ b/apps/api/src/routes/tasks.ts @@ -443,6 +443,7 @@ export async function taskRoutes(rawApp: FastifyInstance) { workspaceId: req.user?.workspaceId ?? undefined, }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.create", params: { type, workflowId: workflow.id, name }, @@ -482,6 +483,7 @@ export async function taskRoutes(rawApp: FastifyInstance) { createdBy: req.user?.id ?? null, }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.create", params: { type, taskConfigId: row.id, name }, @@ -528,6 +530,7 @@ export async function taskRoutes(rawApp: FastifyInstance) { workspaceId: req.user?.workspaceId ?? null, }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.create", params: { taskId: task.id, title: taskInput.title, repoUrl: taskInput.repoUrl }, @@ -612,6 +615,7 @@ export async function taskRoutes(rawApp: FastifyInstance) { req.user?.id, ); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.cancel", params: { taskId: id }, @@ -666,6 +670,7 @@ export async function taskRoutes(rawApp: FastifyInstance) { }, ); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.retry", params: { taskId: id }, @@ -722,6 +727,7 @@ export async function taskRoutes(rawApp: FastifyInstance) { }, ); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.force_redo", params: { taskId: id }, @@ -973,6 +979,7 @@ export async function taskRoutes(rawApp: FastifyInstance) { const { launchReview } = await import("../services/review-service.js"); const reviewTaskId = await launchReview(id); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.review", params: { taskId: id }, @@ -1041,6 +1048,7 @@ export async function taskRoutes(rawApp: FastifyInstance) { const task = await taskService.getTask(id); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.run_now", params: { taskId: id }, @@ -1091,6 +1099,7 @@ export async function taskRoutes(rawApp: FastifyInstance) { .where(eq(tasks.id, body.taskIds[i])); } logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "task.reorder", params: { taskIds: body.taskIds }, diff --git a/apps/api/src/routes/webhooks.ts b/apps/api/src/routes/webhooks.ts index dd80e450..b9744bef 100644 --- a/apps/api/src/routes/webhooks.ts +++ b/apps/api/src/routes/webhooks.ts @@ -145,6 +145,7 @@ export async function webhookRoutes(rawApp: FastifyInstance) { const workspaceId = req.user?.workspaceId ?? null; const webhook = await webhookService.createWebhook(req.body, req.user?.id, workspaceId); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "webhook.create", params: { url: req.body.url, events: req.body.events }, @@ -185,6 +186,7 @@ export async function webhookRoutes(rawApp: FastifyInstance) { const updated = await webhookService.updateWebhook(id, req.body); if (!updated) return reply.status(404).send({ error: "Webhook not found" }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "webhook.update", params: { webhookId: id, ...req.body }, @@ -221,6 +223,7 @@ export async function webhookRoutes(rawApp: FastifyInstance) { const deleted = await webhookService.deleteWebhook(id); if (!deleted) return reply.status(404).send({ error: "Webhook not found" }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "webhook.delete", params: { webhookId: id }, diff --git a/apps/api/src/routes/workflow-triggers.ts b/apps/api/src/routes/workflow-triggers.ts index b545dec0..59ad4d88 100644 --- a/apps/api/src/routes/workflow-triggers.ts +++ b/apps/api/src/routes/workflow-triggers.ts @@ -158,6 +158,7 @@ export async function workflowTriggerRoutes(rawApp: FastifyInstance) { enabled: input.enabled, }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "workflow_trigger.create", params: { workflowId: id, type: input.type }, @@ -227,6 +228,7 @@ export async function workflowTriggerRoutes(rawApp: FastifyInstance) { const trigger = await triggerService.updateTrigger(triggerId, input); if (!trigger) return reply.status(404).send({ error: "Trigger not found" }); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "workflow_trigger.update", params: { workflowId: id, triggerId }, @@ -276,6 +278,7 @@ export async function workflowTriggerRoutes(rawApp: FastifyInstance) { await triggerService.deleteTrigger(triggerId); logAction({ + workspaceId: req.user?.workspaceId ?? null, userId: req.user?.id, action: "workflow_trigger.delete", params: { workflowId: id, triggerId }, diff --git a/apps/api/src/services/optio-action-service.test.ts b/apps/api/src/services/optio-action-service.test.ts index aafed07b..23d08ee0 100644 --- a/apps/api/src/services/optio-action-service.test.ts +++ b/apps/api/src/services/optio-action-service.test.ts @@ -70,53 +70,61 @@ describe("logAction", () => { expect(result).toEqual(mockAction); }); - it("sanitizes sensitive fields in params", async () => { - const mockAction = { - id: "action-2", - action: "create_task", - params: { title: "test", apiToken: "[REDACTED]" }, - success: true, - createdAt: new Date(), - }; - const mockReturning = vi.fn().mockResolvedValue([mockAction]); + it("persists the caller's workspaceId", async () => { + const mockReturning = vi.fn().mockResolvedValue([{ id: "a" }]); const mockValues = vi.fn().mockReturnValue({ returning: mockReturning }); (db.insert as any).mockReturnValue({ values: mockValues }); await logAction({ - action: "create_task", - params: { title: "test", apiToken: "secret-value-123" }, + workspaceId: "ws-42", + action: "task.create", + params: { taskId: "t1" }, success: true, }); - // Verify the values call had the sanitized params const insertedValues = mockValues.mock.calls[0][0]; - expect(insertedValues.params.apiToken).toBe("[REDACTED]"); - expect(insertedValues.params.title).toBe("test"); + expect(insertedValues.workspaceId).toBe("ws-42"); + }); + + it("defaults workspaceId to null when omitted (operator/legacy)", async () => { + const mockReturning = vi.fn().mockResolvedValue([{ id: "a" }]); + const mockValues = vi.fn().mockReturnValue({ returning: mockReturning }); + (db.insert as any).mockReturnValue({ values: mockValues }); + + await logAction({ action: "task.create", params: { taskId: "t1" }, success: true }); + + const insertedValues = mockValues.mock.calls[0][0]; + expect(insertedValues.workspaceId).toBeNull(); }); - it("sanitizes various sensitive key patterns", async () => { + it("keeps only allowlisted params and drops everything else", async () => { const mockReturning = vi.fn().mockResolvedValue([{ id: "a" }]); const mockValues = vi.fn().mockReturnValue({ returning: mockReturning }); (db.insert as any).mockReturnValue({ values: mockValues }); await logAction({ - action: "test", + action: "connection.update", params: { + connectionId: "c1", // allowlisted id — kept + name: "prod-db", // allowlisted label — kept + // Everything below is NOT allowlisted and must be dropped, not stored: + config: { url: "postgres://user:hunter2@db/app" }, + apiToken: "secret-value-123", password: "pass123", - SECRET_KEY: "key123", - authHeader: "Bearer xyz", - normalField: "keep-this", - credential: "cred", + headers: { authorization: "Bearer xyz" }, + arbitraryField: "from ...req.body spread", }, success: true, }); const insertedValues = mockValues.mock.calls[0][0]; - expect(insertedValues.params.password).toBe("[REDACTED]"); - expect(insertedValues.params.SECRET_KEY).toBe("[REDACTED]"); - expect(insertedValues.params.authHeader).toBe("[REDACTED]"); - expect(insertedValues.params.normalField).toBe("keep-this"); - expect(insertedValues.params.credential).toBe("[REDACTED]"); + expect(insertedValues.params).toEqual({ connectionId: "c1", name: "prod-db" }); + // Explicitly assert secret-bearing keys are absent (not merely redacted). + expect(insertedValues.params).not.toHaveProperty("config"); + expect(insertedValues.params).not.toHaveProperty("apiToken"); + expect(insertedValues.params).not.toHaveProperty("password"); + expect(insertedValues.params).not.toHaveProperty("headers"); + expect(insertedValues.params).not.toHaveProperty("arbitraryField"); }); it("handles null params", async () => { diff --git a/apps/api/src/services/optio-action-service.ts b/apps/api/src/services/optio-action-service.ts index 1f44097b..848fc6a7 100644 --- a/apps/api/src/services/optio-action-service.ts +++ b/apps/api/src/services/optio-action-service.ts @@ -4,20 +4,73 @@ import { optioActions, users } from "../db/schema.js"; import type { OptioAction } from "@optio/shared"; import { publishEvent } from "./event-bus.js"; -// ── Sensitive key patterns to strip from params ───────────────────────────── +// ── Params allowlist ──────────────────────────────────────────────────────── +// +// The audit trail must never carry secrets or other tenants' data. Callers +// historically spread whole request bodies into `params` (e.g. `...req.body`), +// which could persist plaintext credentials, connection configs, webhook +// signing secrets, and cross-tenant ids. Rather than blocklist by key name +// (fragile — misses nested values and unexpected key spellings), we keep an +// explicit allowlist of non-secret, low-cardinality fields: resource ids, +// human-readable names/types, and a few scalar flags. Everything else is +// dropped at write time. The same allowlist is applied at read time in +// `routes/activity.ts` so legacy rows written before this change can't leak +// their full `params` either. + +export const ACTION_PARAM_ALLOWLIST: ReadonlySet = new Set([ + // Identifiers (safe: opaque ids scoped by the workspace filter) + "id", + "ids", + "taskId", + "taskIds", + "taskConfigId", + "configId", + "parentTaskId", + "repoId", + "workflowId", + "workflowRunId", + "runId", + "connectionId", + "assignmentId", + "mcpServerId", + "webhookId", + "triggerId", + "prReviewId", + "sessionId", + "agentId", + "providerSlug", + // Human-readable descriptors (labels, not secret values) + "name", + "slug", + "title", + "type", + "kind", + "scope", + "agentType", + "fullName", + "repoUrl", + "prUrl", + "events", + // Non-secret scalar flags / counts + "count", + "enabled", + "priority", + "status", + "state", +]); -const SENSITIVE_KEYS = /token|secret|password|key|credential|auth/i; - -/** Remove sensitive fields from a params object before persisting. */ -function sanitizeParams( +/** + * Keep only allowlisted, non-secret fields from a params object. Nested + * objects/arrays under non-allowlisted keys are dropped entirely, so secrets + * buried inside request bodies never reach the audit table. + */ +export function filterParams( params: Record | null | undefined, ): Record | null { if (!params) return null; const clean: Record = {}; for (const [k, v] of Object.entries(params)) { - if (SENSITIVE_KEYS.test(k)) { - clean[k] = "[REDACTED]"; - } else { + if (ACTION_PARAM_ALLOWLIST.has(k) && v !== undefined) { clean[k] = v; } } @@ -28,6 +81,8 @@ function sanitizeParams( export interface LogActionInput { userId?: string; + /** Tenant the action belongs to. Null = operator/legacy (admin-only visibility). */ + workspaceId?: string | null; action: string; params?: Record | null; result?: Record | null; @@ -36,16 +91,17 @@ export interface LogActionInput { } /** - * Record an Optio agent action in the audit trail. - * Params are sanitized to strip any sensitive values before storage. + * Record an Optio agent action in the audit trail. Params are reduced to an + * explicit non-secret allowlist before storage (see `filterParams`). */ export async function logAction(input: LogActionInput): Promise { const [row] = await db .insert(optioActions) .values({ + workspaceId: input.workspaceId ?? null, userId: input.userId, action: input.action, - params: sanitizeParams(input.params), + params: filterParams(input.params), result: input.result ?? null, success: input.success, conversationSnippet: input.conversationSnippet ?? null, diff --git a/apps/api/src/ws/persistent-agent-stream.ts b/apps/api/src/ws/persistent-agent-stream.ts index c5da1792..972a2b4f 100644 --- a/apps/api/src/ws/persistent-agent-stream.ts +++ b/apps/api/src/ws/persistent-agent-stream.ts @@ -7,6 +7,7 @@ import type { FastifyInstance } from "fastify"; import { z } from "zod"; import { createSubscriber } from "../services/event-bus.js"; import { authenticateWs } from "./ws-auth.js"; +import { assertWorkspace } from "./ws-authz.js"; import { getPersistentAgentUnscoped, listPersistentAgentTurns, @@ -40,6 +41,12 @@ export async function persistentAgentStreamWs(app: FastifyInstance) { return; } + // Enforce workspace isolation before streaming the agent's turn output. + if (!assertWorkspace(socket, user.workspaceId, agent.workspaceId)) { + releaseConnection(clientIp); + return; + } + // Catch-up: send the most recent turn's logs so reconnecting clients see // recent activity without scrolling history. try { diff --git a/apps/api/src/ws/pr-review-log-stream.ts b/apps/api/src/ws/pr-review-log-stream.ts index 9a301703..1e3a56e9 100644 --- a/apps/api/src/ws/pr-review-log-stream.ts +++ b/apps/api/src/ws/pr-review-log-stream.ts @@ -3,6 +3,7 @@ import { z } from "zod"; import { eq } from "drizzle-orm"; import { createSubscriber } from "../services/event-bus.js"; import { authenticateWs } from "./ws-auth.js"; +import { assertWorkspace } from "./ws-authz.js"; import { getPrReview, getLatestRun } from "../services/pr-review-service.js"; import { db } from "../db/client.js"; import { taskLogs } from "../db/schema.js"; @@ -36,8 +37,8 @@ export async function prReviewLogStreamWs(app: FastifyInstance) { releaseConnection(clientIp); return; } - if (user.workspaceId && review.workspaceId && review.workspaceId !== user.workspaceId) { - socket.close(4403, "Access denied"); + // Enforce workspace isolation before streaming the review's run output. + if (!assertWorkspace(socket, user.workspaceId, review.workspaceId)) { releaseConnection(clientIp); return; } diff --git a/apps/api/src/ws/workflow-run-log-stream.ts b/apps/api/src/ws/workflow-run-log-stream.ts index 0080e660..7f9bc802 100644 --- a/apps/api/src/ws/workflow-run-log-stream.ts +++ b/apps/api/src/ws/workflow-run-log-stream.ts @@ -2,7 +2,8 @@ import type { FastifyInstance } from "fastify"; import { z } from "zod"; import { createSubscriber } from "../services/event-bus.js"; import { authenticateWs } from "./ws-auth.js"; -import { getWorkflowRun, getWorkflowRunLogs } from "../services/workflow-service.js"; +import { assertWorkspace } from "./ws-authz.js"; +import { getWorkflow, getWorkflowRun, getWorkflowRunLogs } from "../services/workflow-service.js"; import { getClientIp, trackConnection, @@ -35,6 +36,13 @@ export async function workflowRunLogStreamWs(app: FastifyInstance) { return; } + // Enforce workspace isolation: a run's tenant is its parent workflow's. + const workflow = await getWorkflow(run.workflowId); + if (!assertWorkspace(socket, user.workspaceId, workflow?.workspaceId)) { + releaseConnection(clientIp); + return; + } + // Send catch-up: recent logs so reconnecting clients don't miss data try { const recentLogs = await getWorkflowRunLogs(workflowRunId, { limit: 50 }); diff --git a/apps/api/src/ws/ws-authz.test.ts b/apps/api/src/ws/ws-authz.test.ts new file mode 100644 index 00000000..40e206c9 --- /dev/null +++ b/apps/api/src/ws/ws-authz.test.ts @@ -0,0 +1,157 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +// ── Mocks shared by the handler-level tests ────────────────────────────────── + +const mockAuthenticateWs = vi.fn(); +vi.mock("./ws-auth.js", () => ({ + authenticateWs: (...a: unknown[]) => mockAuthenticateWs(...a), +})); + +const mockGetWorkflowRun = vi.fn(); +const mockGetWorkflow = vi.fn(); +const mockGetWorkflowRunLogs = vi.fn(); +vi.mock("../services/workflow-service.js", () => ({ + getWorkflowRun: (...a: unknown[]) => mockGetWorkflowRun(...a), + getWorkflow: (...a: unknown[]) => mockGetWorkflow(...a), + getWorkflowRunLogs: (...a: unknown[]) => mockGetWorkflowRunLogs(...a), +})); + +const mockGetPersistentAgent = vi.fn(); +const mockListTurns = vi.fn(); +const mockListTurnLogs = vi.fn(); +vi.mock("../services/persistent-agent-service.js", () => ({ + getPersistentAgentUnscoped: (...a: unknown[]) => mockGetPersistentAgent(...a), + listPersistentAgentTurns: (...a: unknown[]) => mockListTurns(...a), + listTurnLogs: (...a: unknown[]) => mockListTurnLogs(...a), +})); + +const mockCreateSubscriber = vi.fn(); +vi.mock("../services/event-bus.js", () => ({ + createSubscriber: (...a: unknown[]) => mockCreateSubscriber(...a), +})); + +// ws-limits: always admit the connection; track release calls. +const mockReleaseConnection = vi.fn(); +vi.mock("./ws-limits.js", () => ({ + getClientIp: () => "1.2.3.4", + trackConnection: () => true, + releaseConnection: (...a: unknown[]) => mockReleaseConnection(...a), + WS_CLOSE_CONNECTION_LIMIT: 4408, +})); + +import { assertWorkspace, WS_CLOSE_FORBIDDEN } from "./ws-authz.js"; +import { workflowRunLogStreamWs } from "./workflow-run-log-stream.js"; +import { persistentAgentStreamWs } from "./persistent-agent-stream.js"; + +function mockSocket() { + return { close: vi.fn(), send: vi.fn(), on: vi.fn() }; +} + +function fakeSubscriber() { + return { + subscribe: vi.fn(), + unsubscribe: vi.fn(), + disconnect: vi.fn(), + on: vi.fn(), + }; +} + +/** Register a websocket handler against a fake app and hand back the closure. */ +async function captureHandler( + register: (app: unknown) => Promise, +): Promise<(socket: unknown, req: unknown) => Promise> { + let handler!: (socket: unknown, req: unknown) => Promise; + const fakeApp = { + get: (_path: string, _opts: unknown, h: (socket: unknown, req: unknown) => Promise) => { + handler = h; + }, + }; + await register(fakeApp); + return handler; +} + +describe("assertWorkspace", () => { + it("passes when workspaces match", () => { + const socket = mockSocket(); + expect(assertWorkspace(socket, "ws-A", "ws-A")).toBe(true); + expect(socket.close).not.toHaveBeenCalled(); + }); + + it("passes when both are null (auth-disabled dev + null-workspace resource)", () => { + const socket = mockSocket(); + expect(assertWorkspace(socket, null, null)).toBe(true); + // undefined is normalized to null as well + expect(assertWorkspace(socket, undefined, null)).toBe(true); + expect(assertWorkspace(socket, null, undefined)).toBe(true); + expect(socket.close).not.toHaveBeenCalled(); + }); + + it("closes 4403 when the caller is in a different workspace than the resource", () => { + const socket = mockSocket(); + expect(assertWorkspace(socket, "ws-A", "ws-B")).toBe(false); + expect(socket.close).toHaveBeenCalledWith(WS_CLOSE_FORBIDDEN, "Access denied"); + }); + + it("closes 4403 for a scoped caller against a legacy null-workspace resource", () => { + const socket = mockSocket(); + expect(assertWorkspace(socket, "ws-A", null)).toBe(false); + expect(socket.close).toHaveBeenCalledWith(4403, "Access denied"); + }); +}); + +describe("workflow-run log stream: workspace enforcement", () => { + beforeEach(() => { + vi.clearAllMocks(); + mockCreateSubscriber.mockReturnValue(fakeSubscriber()); + mockGetWorkflowRunLogs.mockResolvedValue([]); + }); + + it("closes 4403 and never subscribes when the run belongs to another workspace", async () => { + mockAuthenticateWs.mockResolvedValue({ id: "u", workspaceId: "ws-A" }); + mockGetWorkflowRun.mockResolvedValue({ id: "run-1", workflowId: "wf-1" }); + mockGetWorkflow.mockResolvedValue({ id: "wf-1", workspaceId: "ws-B" }); + + const handler = await captureHandler(workflowRunLogStreamWs as never); + const socket = mockSocket(); + await handler(socket, { headers: {}, params: { workflowRunId: "run-1" } }); + + expect(socket.close).toHaveBeenCalledWith(4403, "Access denied"); + expect(mockCreateSubscriber).not.toHaveBeenCalled(); + expect(mockReleaseConnection).toHaveBeenCalled(); + }); + + it("streams when the run is in the caller's workspace", async () => { + mockAuthenticateWs.mockResolvedValue({ id: "u", workspaceId: "ws-A" }); + mockGetWorkflowRun.mockResolvedValue({ id: "run-1", workflowId: "wf-1" }); + mockGetWorkflow.mockResolvedValue({ id: "wf-1", workspaceId: "ws-A" }); + + const handler = await captureHandler(workflowRunLogStreamWs as never); + const socket = mockSocket(); + await handler(socket, { headers: {}, params: { workflowRunId: "run-1" } }); + + expect(socket.close).not.toHaveBeenCalledWith(4403, "Access denied"); + expect(mockCreateSubscriber).toHaveBeenCalled(); + }); +}); + +describe("persistent-agent event stream: workspace enforcement", () => { + beforeEach(() => { + vi.clearAllMocks(); + mockCreateSubscriber.mockReturnValue(fakeSubscriber()); + mockListTurns.mockResolvedValue([]); + mockListTurnLogs.mockResolvedValue([]); + }); + + it("closes 4403 and never subscribes for a cross-workspace agent", async () => { + mockAuthenticateWs.mockResolvedValue({ id: "u", workspaceId: "ws-A" }); + mockGetPersistentAgent.mockResolvedValue({ id: "ag-1", slug: "bot", workspaceId: "ws-B" }); + + const handler = await captureHandler(persistentAgentStreamWs as never); + const socket = mockSocket(); + await handler(socket, { headers: {}, params: { agentId: "ag-1" } }); + + expect(socket.close).toHaveBeenCalledWith(4403, "Access denied"); + expect(mockCreateSubscriber).not.toHaveBeenCalled(); + expect(mockReleaseConnection).toHaveBeenCalled(); + }); +}); diff --git a/apps/api/src/ws/ws-authz.ts b/apps/api/src/ws/ws-authz.ts new file mode 100644 index 00000000..16584b87 --- /dev/null +++ b/apps/api/src/ws/ws-authz.ts @@ -0,0 +1,43 @@ +/** + * Shared workspace authorization for WebSocket log/event streams. + * + * Log-stream sockets (`/ws/workflow-runs/:id/logs`, `/ws/persistent-agents/:id/events`, + * `/ws/pr-reviews/:id/logs`, ...) authenticate the caller but must ALSO confirm + * the resolved resource belongs to the caller's workspace before streaming any + * output — otherwise a client could tail another tenant's agent logs live. + */ + +/** Minimal WebSocket surface — avoids depending on @types/ws. */ +interface WsSocket { + close(code?: number, reason?: string): void; +} + +/** WebSocket close code used for a cross-workspace access denial. */ +export const WS_CLOSE_FORBIDDEN = 4403; + +/** + * Assert that the socket's user and the resolved resource share a workspace. + * + * Workspaces are null-normalized before comparison, so: + * - a `null` user workspace (auth-disabled synthetic dev user) matches a + * `null` resource workspace (local dev resources) and passes, and + * - a scoped user (`workspace A`) is denied access to a resource in + * `workspace B` or to a legacy null-workspace resource. + * + * On mismatch the socket is closed with code 4403 and `false` is returned; + * callers should stop and release the connection. Returns `true` when access + * is allowed. + */ +export function assertWorkspace( + socket: WsSocket, + userWorkspaceId: string | null | undefined, + resourceWorkspaceId: string | null | undefined, +): boolean { + const user = userWorkspaceId ?? null; + const resource = resourceWorkspaceId ?? null; + if (user !== resource) { + socket.close(WS_CLOSE_FORBIDDEN, "Access denied"); + return false; + } + return true; +}