diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index ecfb81fb92b..8b614c04b44 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2905,6 +2905,12 @@ Hidden browser tabs suspend the company live-events connection and transcript log reads. Returning to a visible tab refreshes active queries once and resumes transcript reads from their retained offsets. A queued live-event invalidation that flushes after the tab hides marks data stale without starting a refetch. + +Live-event delivery and workspace-operation run checks select only the stored +company, run scope, and task binding. Event-list access also selects the run ID. +These checks do not load run context, runner configuration, results, or output. +Every delivery and request still checks current permissions. Grant revocation, +deleted tasks, missing runs, and cross-company run IDs remain fail-closed. The developer-server health poll also stops in hidden tabs. Workspace detail responses share concurrent Git inspections and reuse their diff --git a/server/src/__tests__/agent-live-run-routes.test.ts b/server/src/__tests__/agent-live-run-routes.test.ts index 0fafbdc7849..94f6d525268 100644 --- a/server/src/__tests__/agent-live-run-routes.test.ts +++ b/server/src/__tests__/agent-live-run-routes.test.ts @@ -14,6 +14,7 @@ const mockHeartbeatService = vi.hoisted(() => ({ decorateActiveRunStatus: vi.fn(), getRunIssueSummary: vi.fn(), getActiveRunIssueSummaryForAgent: vi.fn(), + getRunReadAccess: vi.fn(), getRunLogAccess: vi.fn(), readLog: vi.fn(), wakeup: vi.fn(), @@ -363,6 +364,12 @@ describe("agent live run routes", () => { null, ); mockHeartbeatService.buildRunOutputSilence.mockResolvedValue(null); + mockHeartbeatService.getRunReadAccess.mockResolvedValue({ + id: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + companyId: "company-1", + scopeKind: "company", + issueId: null, + }); mockHeartbeatService.getRunLogAccess.mockResolvedValue({ id: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", companyId: "company-1", @@ -442,6 +449,7 @@ describe("agent live run routes", () => { expect(response.body).toEqual({ error: "Invalid heartbeat run ID" }); } expect(mockHeartbeatService.getRun).not.toHaveBeenCalled(); + expect(mockHeartbeatService.getRunReadAccess).not.toHaveBeenCalled(); expect(mockHeartbeatService.getRunLogAccess).not.toHaveBeenCalled(); expect(mockHeartbeatService.readLog).not.toHaveBeenCalled(); expect(mockProviderTraceStore.inspect).not.toHaveBeenCalled(); @@ -458,9 +466,11 @@ describe("agent live run routes", () => { expect(mockHeartbeatService.getRunLogAccess).toHaveBeenCalledWith(runId); }); - it.each(["get", "log"])("keeps missing and cross-company %s lookups indistinguishable", async (kind) => { - const lookup = kind === "log" ? mockHeartbeatService.getRunLogAccess : mockHeartbeatService.getRun; - const suffix = kind === "log" ? "/log" : ""; + it.each(["get", "log", "events"])("keeps missing and cross-company %s lookups indistinguishable", async (kind) => { + const lookup = kind === "events" + ? mockHeartbeatService.getRunReadAccess + : kind === "log" ? mockHeartbeatService.getRunLogAccess : mockHeartbeatService.getRun; + const suffix = kind === "get" ? "" : `/${kind}`; const app = await createApp({}, { type: "board", userId: "test-user", source: "session", companyIds: ["company-1"], }); diff --git a/server/src/__tests__/heartbeat-run-privacy-routes.test.ts b/server/src/__tests__/heartbeat-run-privacy-routes.test.ts index cb1fa213108..03914cadeda 100644 --- a/server/src/__tests__/heartbeat-run-privacy-routes.test.ts +++ b/server/src/__tests__/heartbeat-run-privacy-routes.test.ts @@ -1,5 +1,6 @@ import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; +import { drizzle } from "drizzle-orm/postgres-js"; import express from "express"; import request from "supertest"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; @@ -20,6 +21,7 @@ import { toolConnections, workspaceOperations, } from "@paperclipai/db"; +import * as schema from "@paperclipai/db/schema/index"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -28,7 +30,7 @@ import { errorHandler } from "../middleware/index.js"; import { activityRoutes } from "../routes/activity.js"; import { agentRoutes } from "../routes/agents.js"; import { authorizationService } from "../services/authorization.js"; -import { getHeartbeatRunReadDecisions } from "../services/heartbeat-run-privacy.js"; +import { canActorReadWorkspaceOperation, getHeartbeatRunReadDecisions } from "../services/heartbeat-run-privacy.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -219,7 +221,7 @@ describeEmbeddedPostgres("heartbeat run privacy routes", { concurrent: false }, }; } - function createApp(companyId: string, agentId: string) { + function createApp(companyId: string, agentId: string, sourceDb = db) { const app = express(); app.use(express.json()); app.use((req, _res, next) => { @@ -231,8 +233,8 @@ describeEmbeddedPostgres("heartbeat run privacy routes", { concurrent: false }, }; next(); }); - app.use("/api", agentRoutes(db)); - app.use("/api", activityRoutes(db)); + app.use("/api", agentRoutes(sourceDb)); + app.use("/api", activityRoutes(sourceDb)); app.use(errorHandler); return app; } @@ -508,6 +510,31 @@ describeEmbeddedPostgres("heartbeat run privacy routes", { concurrent: false }, expect(await getHeartbeatRunReadDecisions(db, access, { ...actor, agentId: fixture.otherAgentId }, [binding])).toEqual([false]); }); + it("keeps event and operation authorization reads narrow for large runs", async () => { + const fixture = await seedFixture(); + await db.update(heartbeatRuns).set({ + contextSnapshot: { issueId: fixture.issueId, prompt: "large-private-context".repeat(65_536) }, + runnerProfileJson: { instructions: "large-private-profile".repeat(65_536) }, + }).where(eq(heartbeatRuns.id, fixture.privateRunId)); + const queries: string[] = []; + const observedDb = drizzle(db.$client, { schema, logger: { logQuery(query) { queries.push(query); } } }) as typeof db; + const app = createApp(fixture.companyId, fixture.grantedAgentId, observedDb); + const response = await request(app).get(`/api/heartbeat-runs/${fixture.privateRunId}/events`); + expect(response.status).toBe(200); + expect(response.body[0].message).toBe("private@example.test"); + expect(queries.some(query => query.startsWith('select "id", "company_id", "scope_kind", "issue_id" from "heartbeat_runs"'))).toBe(true); + expect(queries.filter(query => query.startsWith("select ")).join("\n")).not.toMatch(/"runner_profile_json"|"stdout_excerpt"/); + queries.length = 0; + const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, fixture.operationId)); + expect(await canActorReadWorkspaceOperation(observedDb, authorizationService(observedDb), { + type: "agent", companyId: fixture.companyId, agentId: fixture.grantedAgentId, + }, operation)).toBe(true); + expect(queries.some(query => query.startsWith('select "company_id", "scope_kind", "issue_id" from "heartbeat_runs"'))).toBe(true); + expect(queries.filter(query => query.startsWith("select ")).join("\n")).not.toMatch(/"runner_profile_json"|"stdout_excerpt"/); + await db.update(issueAccessGrants).set({ revokedAt: new Date() }).where(eq(issueAccessGrants.subjectId, fixture.grantedAgentId)); + expect((await request(app).get(`/api/heartbeat-runs/${fixture.privateRunId}/events`)).status).toBe(404); + }); + it("rechecks a revoked grant on the next history request", async () => { const fixture = await seedFixture(); const duplicateId = randomUUID(); diff --git a/server/src/__tests__/live-events-run-read.test.ts b/server/src/__tests__/live-events-run-read.test.ts new file mode 100644 index 00000000000..76704f2f63c --- /dev/null +++ b/server/src/__tests__/live-events-run-read.test.ts @@ -0,0 +1,124 @@ +import { randomUUID } from "node:crypto"; +import { once } from "node:events"; +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import { eq } from "drizzle-orm"; +import { drizzle } from "drizzle-orm/postgres-js"; +import { WebSocket } from "ws"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + agents, authUsers, companies, companyMemberships, createDb, + heartbeatRuns, issueAccessGrants, issues, type Db, +} from "@paperclipai/db"; +import * as schema from "@paperclipai/db/schema/index"; +import { setupLiveEventsWebSocketServer } from "../realtime/live-events-ws.js"; +import { publishLiveEvent } from "../services/live-events.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +describe("live-event run read boundaries", () => { + let temporary: Awaited>; + let db: Db; + const previousPrivacy = process.env.PAPERCLIP_ISSUE_PRIVACY_MODE; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("live-event-run-reads-"); + db = createDb(temporary.connectionString); + process.env.PAPERCLIP_ISSUE_PRIVACY_MODE = "enforce"; + }, 120_000); + + afterAll(async () => { + if (previousPrivacy === undefined) delete process.env.PAPERCLIP_ISSUE_PRIVACY_MODE; + else process.env.PAPERCLIP_ISSUE_PRIVACY_MODE = previousPrivacy; + await temporary?.cleanup(); + }); + + it("reads only run bindings and reauthorizes every delivery after revocation or deletion", async () => { + const companyId = randomUUID(), foreignCompanyId = randomUUID(); + const owner = randomUUID(), viewer = randomUUID(); + await db.insert(companies).values([ + { id: companyId, name: "Run reads", issuePrefix: "RUNREAD" }, + { id: foreignCompanyId, name: "Other company", issuePrefix: "OTHERREAD" }, + ]); + await db.insert(authUsers).values([owner, viewer].map(id => ({ + id, name: "Fixture viewer", email: `${id}@example.test`, createdAt: new Date(), updatedAt: new Date(), + }))); + await db.insert(companyMemberships).values([owner, viewer].map(principalId => ({ + companyId, principalType: "user", principalId, membershipRole: "operator", status: "active", + }))); + const [agent] = await db.insert(agents).values({ companyId, name: "Fixture agent", adapterType: "process" }).returning(); + const [foreignAgent] = await db.insert(agents).values({ companyId: foreignCompanyId, name: "Other agent", adapterType: "process" }).returning(); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Private task", visibility: "private", privacyRootIssueId: issueId, responsibleUserId: owner }); + const [grant] = await db.insert(issueAccessGrants).values({ issueId, subjectType: "user", subjectId: viewer, source: "explicit" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ + companyId, agentId: agent.id, scopeKind: "issue", issueId, status: "running", + contextSnapshot: { issueId, prompt: "private-context-".repeat(65_536) }, + runnerProfileJson: { instructions: "private-profile-".repeat(65_536) }, + stdoutExcerpt: "private-output-".repeat(65_536), + }).returning({ id: heartbeatRuns.id }); + const [maintenance] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, scopeKind: "company" }).returning({ id: heartbeatRuns.id }); + const [foreign] = await db.insert(heartbeatRuns).values({ companyId: foreignCompanyId, agentId: foreignAgent.id, scopeKind: "company" }).returning({ id: heartbeatRuns.id }); + const reads: string[] = []; + let failNextRunRead = false; + const observedDb = drizzle(db.$client, { schema, logger: { logQuery(query) { + if (query.startsWith("select ") && query.includes('from "heartbeat_runs"')) { + reads.push(query); + if (failNextRunRead) { failNextRunRead = false; throw new Error("fixture run lookup unavailable"); } + } + } } }) as Db; + const server = createServer(); + const wss = setupLiveEventsWebSocketServer(server, observedDb, { + deploymentMode: "authenticated", + resolveCloudActor: async () => ({ userId: viewer, companyIds: [companyId] }), + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const socket = new WebSocket(`ws://127.0.0.1:${(server.address() as AddressInfo).port}/api/companies/${companyId}/events/ws`); + const messages: Array<{ type: string; payload: Record }> = []; + socket.on("message", bytes => messages.push(JSON.parse(bytes.toString()))); + try { + await once(socket, "open"); + async function deliver(runId: string) { + const start = messages.length; + const barrier = randomUUID(); + publishLiveEvent({ companyId, type: "heartbeat.run.status", payload: { runId, status: "running" } }); + publishLiveEvent({ companyId, type: "agent.status", payload: { agentId: barrier, status: "idle" } }); + await expect.poll(() => messages.some(message => message.payload.agentId === barrier)).toBe(true); + return messages.slice(start).filter(message => message.type === "heartbeat.run.status"); + } + + expect(await deliver(run.id)).toMatchObject([{ payload: { runId: run.id, status: "running" } }]); + await db.update(issueAccessGrants).set({ revokedAt: new Date() }).where(eq(issueAccessGrants.id, grant.id)); + expect(await deliver(run.id)).toEqual([]); + expect(await deliver(maintenance.id)).toHaveLength(1); + expect(await deliver(foreign.id)).toEqual([]); + expect(await deliver(randomUUID())).toEqual([]); + failNextRunRead = true; + expect(await deliver(maintenance.id)).toEqual([]); + expect(await deliver(maintenance.id)).toHaveLength(1); + await db.update(issueAccessGrants).set({ revokedAt: null }).where(eq(issueAccessGrants.id, grant.id)); + expect(await deliver(run.id)).toHaveLength(1); + await db.delete(issues).where(eq(issues.id, issueId)); + expect(await deliver(run.id)).toEqual([]); + + expect(reads).toHaveLength(9); + for (const query of reads) { + expect(query).toMatch(/^select "company_id", "scope_kind", "issue_id" from "heartbeat_runs"/); + expect(query).not.toMatch(/context_snapshot|runner_profile_json|stdout_excerpt|result_json/); + } + expect(JSON.stringify(messages)).not.toMatch(/private-context|private-profile|private-output/); + } finally { + socket.terminate(); + await Promise.all([...wss.clients].map(client => { + const closed = new Promise(resolve => client.on("close", resolve)); + client.terminate(); + return closed; + })); + await new Promise((resolve, reject) => { + (wss as typeof wss & { close(callback: (error?: Error) => void): void }) + .close(error => error ? reject(error) : resolve()); + }); + if (server.listening) await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + } + }); +}); diff --git a/server/src/realtime/live-events-ws.ts b/server/src/realtime/live-events-ws.ts index 86a88c2d53b..6f299bfe672 100644 --- a/server/src/realtime/live-events-ws.ts +++ b/server/src/realtime/live-events-ws.ts @@ -323,7 +323,11 @@ export function setupLiveEventsWebSocketServer( const runId = typeof payload.runId === "string" ? payload.runId : payload.entityType === "heartbeat_run" && typeof payload.entityId === "string" ? payload.entityId : null; if (runId) { - const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, event.companyId))); + const [run] = await db.select({ + companyId: heartbeatRuns.companyId, + scopeKind: heartbeatRuns.scopeKind, + issueId: heartbeatRuns.issueId, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, event.companyId))); if (!run || !(await canActorReadHeartbeatRun(db, access, context!.actor, run))) return null; } if (event.type.startsWith("heartbeat.run.")) return runId ? event : null; diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index a2986d558ea..f2edb56ac0d 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -7509,7 +7509,7 @@ export function agentRoutes( router.get("/heartbeat-runs/:runId/events", async (req, res) => { const runId = readHeartbeatRunId(req); - const run = await getAccessibleResource(req, res, heartbeat.getRun(runId), "Heartbeat run not found"); + const run = await getAccessibleResource(req, res, heartbeat.getRunReadAccess(runId), "Heartbeat run not found"); if (!run) return; if (!(await assertRunTelemetryReadAllowed(req, res, run.companyId))) return; if (!(await assertRunReadAllowed(req, res, run))) return; diff --git a/server/src/services/heartbeat-run-privacy.ts b/server/src/services/heartbeat-run-privacy.ts index 49b225c8943..17b61f4cbbd 100644 --- a/server/src/services/heartbeat-run-privacy.ts +++ b/server/src/services/heartbeat-run-privacy.ts @@ -182,7 +182,11 @@ export async function canActorReadWorkspaceOperation( } for (const runId of runIds) { if (!/^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/i.test(runId)) return false; - const [run] = await db.select().from(heartbeatRuns) + const [run] = await db.select({ + companyId: heartbeatRuns.companyId, + scopeKind: heartbeatRuns.scopeKind, + issueId: heartbeatRuns.issueId, + }).from(heartbeatRuns) .where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, operation.companyId))).limit(1); if (!run || !(await canActorReadHeartbeatRun(db, access, actor, run))) return false; } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 81672288f4e..53092b6f8e9 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -819,6 +819,7 @@ export function heartbeatService( clearTaskSessions, resolveExplicitResumeSessionOverride, hasResolvablePriorSessionWorkspaceForWake, + getRunReadAccess, getRunLogAccess, listRuns, getRuntimeStateWithSessions, @@ -3976,6 +3977,7 @@ export function heartbeatService( recordRuntimeProgress: recordCurrentHeartbeatRunRuntimeProgress, sweepExpiredRuntimeStatuses: sweepExpiredHeartbeatRunRuntimeStatuses, + getRunReadAccess, getRunLogAccess, getRuntimeState: getRuntimeStateWithSessions, diff --git a/server/src/services/heartbeat/run-state.test.ts b/server/src/services/heartbeat/run-state.test.ts index 2191ca1959d..087931f01e8 100644 --- a/server/src/services/heartbeat/run-state.test.ts +++ b/server/src/services/heartbeat/run-state.test.ts @@ -91,6 +91,22 @@ describe("heartbeat run state module", () => { expect(raw.execute).not.toHaveBeenCalled(); expect(raw.columns[0].resultJson).toBe(heartbeatRuns.resultJson); }); + + it("loads run read bindings without large fields or an encoding probe", async () => { + const binding = { id: "run", companyId: "company", scopeKind: "issue", issueId: "issue" }; + const reader = readingDatabase("SQL_ASCII", [binding], []); + const state = extracted.createHeartbeatRunState(reader.db); + expect(await state.getRunReadAccess("run")).toEqual(binding); + expect(await state.getRunReadAccess("missing")).toBeNull(); + expect(reader.execute).not.toHaveBeenCalled(); + expect(reader.columns).toEqual(Array(2).fill({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + scopeKind: heartbeatRuns.scopeKind, + issueId: heartbeatRuns.issueId, + })); + expect(reader.scopes).toEqual([["run"], ["missing"]]); + }); }); const postgresSupport = await getEmbeddedPostgresTestSupport(); diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts index e979572e222..a6f302ab64a 100644 --- a/server/src/services/heartbeat/run-state.ts +++ b/server/src/services/heartbeat/run-state.ts @@ -341,11 +341,15 @@ const heartbeatRunSqlAsciiSafeColumns = { stderrExcerpt: sql`NULL`.as("stderrExcerpt"), } as const; -const heartbeatRunLogAccessColumns = { +const heartbeatRunReadAccessColumns = { id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, scopeKind: heartbeatRuns.scopeKind, issueId: heartbeatRuns.issueId, +} as const; + +const heartbeatRunLogAccessColumns = { + ...heartbeatRunReadAccessColumns, logStore: heartbeatRuns.logStore, logRef: heartbeatRuns.logRef, } as const; @@ -1115,6 +1119,14 @@ export function createHeartbeatRunState(db: Db) { .then((rows) => rows[0] ?? null); } + async function getRunReadAccess(runId: string) { + return db + .select(heartbeatRunReadAccessColumns) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + } + async function getRunLogAccess(runId: string) { return db .select(heartbeatRunLogAccessColumns) @@ -1672,6 +1684,7 @@ export function createHeartbeatRunState(db: Db) { clearTaskSessions, resolveExplicitResumeSessionOverride, hasResolvablePriorSessionWorkspaceForWake, + getRunReadAccess, getRunLogAccess, listRuns: async (