diff --git a/packages/db/src/schema/cloud_upstreams.ts b/packages/db/src/schema/cloud_upstreams.ts index c7e2160b4da..93b4341f7e4 100644 --- a/packages/db/src/schema/cloud_upstreams.ts +++ b/packages/db/src/schema/cloud_upstreams.ts @@ -10,10 +10,12 @@ export const cloudUpstreamConnections = pgTable( sourceInstanceId: text("source_instance_id").notNull(), sourceInstanceFingerprint: text("source_instance_fingerprint").notNull(), sourcePublicKey: text("source_public_key").notNull(), + // Stored through the Cloud Upstream service as an encrypted credential envelope. privateKeyPem: text("private_key_pem").notNull(), tokenStatus: text("token_status").notNull(), scopes: text("scopes").array().notNull().default([]), authorizedGlobalUserId: text("authorized_global_user_id"), + // Stored through the Cloud Upstream service as an encrypted credential envelope. accessToken: text("access_token"), tokenId: text("token_id"), tokenExpiresAt: timestamp("token_expires_at", { withTimezone: true }), diff --git a/server/src/__tests__/cloud-upstreams.test.ts b/server/src/__tests__/cloud-upstreams.test.ts index bb7e84d92bb..34175af25b4 100644 --- a/server/src/__tests__/cloud-upstreams.test.ts +++ b/server/src/__tests__/cloud-upstreams.test.ts @@ -1,7 +1,28 @@ -import { describe, expect, it } from "vitest"; +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { companies, cloudUpstreamConnections, cloudUpstreamRuns, createDb } from "@paperclipai/db"; import { HttpError } from "../errors.js"; -import { cloudUpstreamRemoteFailureReport } from "../services/cloud-upstreams.js"; +import { + cloudUpstreamRemoteFailureReport, + cloudUpstreamService, + reconcileCloudUpstreamRunsOnStartup, + sealCloudUpstreamCredential, + unsealCloudUpstreamCredential, +} from "../services/cloud-upstreams.js"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres cloud upstream tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} describe("cloud upstream remote failures", () => { it("preserves the cloud response body and message on run reports", () => { @@ -24,3 +45,207 @@ describe("cloud upstream remote failures", () => { }); }); }); + +describe("cloud upstream credential storage", () => { + const previousMasterKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY; + + afterEach(() => { + if (previousMasterKey === undefined) { + delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; + } else { + process.env.PAPERCLIP_SECRETS_MASTER_KEY = previousMasterKey; + } + }); + + it("stores new credentials as encrypted envelopes and preserves legacy plaintext reads", async () => { + process.env.PAPERCLIP_SECRETS_MASTER_KEY = "12345678901234567890123456789012"; + const sealed = await sealCloudUpstreamCredential("cloud-access-token"); + + expect(sealed).toMatch(/^paperclip-cloud-credential:/); + expect(sealed).not.toContain("cloud-access-token"); + await expect(unsealCloudUpstreamCredential(sealed)).resolves.toBe("cloud-access-token"); + await expect(unsealCloudUpstreamCredential("legacy-plaintext-token")).resolves.toBe("legacy-plaintext-token"); + }); +}); + +describeEmbeddedPostgres("cloud upstream persistence", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + const previousMasterKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY; + + beforeAll(async () => { + process.env.PAPERCLIP_SECRETS_MASTER_KEY = "12345678901234567890123456789012"; + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-cloud-upstreams-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + vi.restoreAllMocks(); + await db.delete(cloudUpstreamRuns); + await db.delete(cloudUpstreamConnections); + await db.delete(companies); + }); + + afterAll(async () => { + if (previousMasterKey === undefined) { + delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; + } else { + process.env.PAPERCLIP_SECRETS_MASTER_KEY = previousMasterKey; + } + await tempDb?.cleanup(); + }); + + it("encrypts stored upstream credentials while keeping connection flows usable", async () => { + const companyId = randomUUID(); + await seedCompany(companyId); + const tokenUrl = "https://cloud.example.test/oauth/token"; + vi.spyOn(globalThis, "fetch").mockImplementation(async (input, init) => { + const url = String(input); + if (url.startsWith("https://cloud.example.test/.well-known/paperclip-upstream")) { + return jsonResponse({ + product: "Paperclip Cloud", + stack: { + id: "stack-1", + companyId: "cloud-company-1", + origin: "https://cloud.example.test", + primaryHost: "cloud.example.test", + }, + transfer: { + supportedSchemaMajor: 1, + maxChunkBytes: 8192, + }, + auth: { + scopes: ["upstream_import:write"], + pkce: { + authorizeUrl: "https://cloud.example.test/oauth/authorize", + tokenUrl, + }, + }, + }); + } + if (url === tokenUrl && init?.method === "POST") { + const payload = JSON.parse(String(init.body)); + expect(payload.codeVerifier).toEqual(expect.any(String)); + expect(payload.codeVerifier).not.toContain("paperclip-cloud-credential:"); + return jsonResponse({ + accessToken: "cloud-access-token", + token: { + id: "token-1", + expiresAt: "2026-05-22T13:00:00.000Z", + globalUserId: "user-1", + }, + }); + } + throw new Error(`Unexpected fetch: ${url}`); + }); + + const service = cloudUpstreamService(db, { instanceId: "test" }); + const started = await service.startConnect({ + companyId, + remoteUrl: "https://cloud.example.test", + redirectUri: "http://localhost:3100/callback", + }); + await service.finishConnect({ + pendingConnectionId: started.pendingConnectionId, + code: "auth-code", + state: new URL(started.authorizationUrl).searchParams.get("state") ?? "", + }); + + const [row] = await db.select().from(cloudUpstreamConnections); + expect(row.privateKeyPem).toMatch(/^paperclip-cloud-credential:/); + expect(row.privateKeyPem).not.toContain("BEGIN PRIVATE KEY"); + expect(row.accessToken).toMatch(/^paperclip-cloud-credential:/); + expect(row.accessToken).not.toContain("cloud-access-token"); + }); + + it("marks orphaned running runs failed during startup reconciliation", async () => { + const companyId = randomUUID(); + const connectionId = randomUUID(); + const runningRunId = randomUUID(); + const succeededRunId = randomUUID(); + const reconciledAt = new Date("2026-05-22T13:00:00.000Z"); + await seedCompany(companyId); + await db.insert(cloudUpstreamConnections).values({ + id: connectionId, + companyId, + remoteUrl: "https://cloud.example.test", + sourceInstanceId: "source-1", + sourceInstanceFingerprint: "sha256:test", + sourcePublicKey: "public-key", + privateKeyPem: "legacy-private-key", + tokenStatus: "connected", + scopes: ["upstream_import:write"], + authorizedGlobalUserId: "user-1", + accessToken: "legacy-token", + tokenId: "token-1", + targetStackId: "stack-1", + targetCompanyId: "cloud-company-1", + targetOrigin: "https://cloud.example.test", + targetPrimaryHost: "cloud.example.test", + targetProduct: "Paperclip Cloud", + targetSchemaMajor: 1, + targetMaxChunkBytes: 8192, + }); + await db.insert(cloudUpstreamRuns).values([ + cloudRunRow({ id: runningRunId, connectionId, companyId, status: "running" }), + cloudRunRow({ id: succeededRunId, connectionId, companyId, status: "succeeded", completedAt: reconciledAt }), + ]); + + await expect(reconcileCloudUpstreamRunsOnStartup(db, reconciledAt)).resolves.toEqual({ reconciled: 1 }); + + const rows = await db.select().from(cloudUpstreamRuns); + const running = rows.find((row) => row.id === runningRunId); + const succeeded = rows.find((row) => row.id === succeededRunId); + expect(running?.status).toBe("failed"); + expect(running?.completedAt?.toISOString()).toBe(reconciledAt.toISOString()); + expect(running?.events.at(-1)?.message).toContain("server startup"); + expect(running?.report).toMatchObject({ + error: "orphaned_running_run", + reconciledAt: reconciledAt.toISOString(), + }); + expect(succeeded?.status).toBe("succeeded"); + }); + + async function seedCompany(companyId: string) { + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + } +}); + +function jsonResponse(body: unknown): Response { + return new Response(JSON.stringify(body), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); +} + +function cloudRunRow(input: { + id: string; + connectionId: string; + companyId: string; + status: string; + completedAt?: Date; +}) { + return { + id: input.id, + connectionId: input.connectionId, + companyId: input.companyId, + status: input.status, + activeStep: "push", + progressPercent: input.status === "running" ? 45 : 100, + dryRun: false, + summary: [], + warnings: [], + conflicts: [], + events: [], + report: {}, + idempotencyKey: `key-${input.id}`, + manifestHash: `sha256:${input.id.replace(/-/g, "")}`, + targetUrl: "https://cloud.example.test", + completedAt: input.completedAt, + }; +} diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index f1023a91533..3cdab7c4ac8 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -166,6 +166,7 @@ vi.mock("../services/index.js", () => ({ }, })), })), + reconcileCloudUpstreamRunsOnStartup: vi.fn(async () => ({ reconciled: 0 })), reconcilePersistedRuntimeServicesOnStartup: vi.fn(async () => ({ reconciled: 0 })), routineService: vi.fn(() => ({ tickScheduledTriggers: vi.fn(async () => ({ triggered: 0 })), diff --git a/server/src/index.ts b/server/src/index.ts index bb89a46e567..d9c52d27d65 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -33,6 +33,7 @@ import { backfillPrincipalAccessCompatibility, heartbeatService, instanceSettingsService, + reconcileCloudUpstreamRunsOnStartup, reconcilePersistedRuntimeServicesOnStartup, routineService, } from "./services/index.js"; @@ -699,6 +700,19 @@ export async function startServer(): Promise { .catch((err) => { logger.error({ err }, "startup reconciliation of persisted runtime services failed"); }); + + void reconcileCloudUpstreamRunsOnStartup(db as any) + .then((result) => { + if (result.reconciled > 0) { + logger.warn( + { reconciled: result.reconciled }, + "reconciled cloud upstream runs from a previous server process", + ); + } + }) + .catch((err) => { + logger.error({ err }, "startup reconciliation of cloud upstream runs failed"); + }); if (config.heartbeatSchedulerEnabled) { const heartbeat = heartbeatService(db as any, { pluginWorkerManager }); diff --git a/server/src/routes/cloud-upstreams.ts b/server/src/routes/cloud-upstreams.ts index 5c9c24bd08f..085afe470db 100644 --- a/server/src/routes/cloud-upstreams.ts +++ b/server/src/routes/cloud-upstreams.ts @@ -45,7 +45,7 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = { router.post("/cloud-upstreams/:connectionId/push-runs/preview", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); - res.json(await service.preview(req.params.connectionId)); + res.json(await service.preview(req.params.connectionId, stringBody(req.body, "companyId"))); }); router.post("/cloud-upstreams/:connectionId/push-runs", async (req, res) => { @@ -53,6 +53,7 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = { await assertEnabled(); res.json(await service.createRun({ connectionId: req.params.connectionId, + companyId: stringBody(req.body, "companyId"), retryOfRunId: optionalString(req.body?.retryOfRunId), })); }); @@ -60,13 +61,13 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = { router.get("/cloud-upstreams/:connectionId/push-runs/:runId", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); - res.json(await service.readRun(req.params.connectionId, req.params.runId)); + res.json(await service.readRun(req.params.connectionId, req.params.runId, stringQuery(req.query.companyId, "companyId"))); }); router.post("/cloud-upstreams/:connectionId/push-runs/:runId/cancel", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); - res.json(await service.cancelRun(req.params.connectionId, req.params.runId)); + res.json(await service.cancelRun(req.params.connectionId, req.params.runId, stringBody(req.body, "companyId"))); }); router.post("/cloud-upstreams/:connectionId/push-runs/:runId/activation", async (req, res) => { @@ -75,6 +76,7 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = { res.json(await service.activateRunEntities({ connectionId: req.params.connectionId, runId: req.params.runId, + companyId: stringBody(req.body, "companyId"), entityType: activationEntityTypeBody(req.body), })); }); diff --git a/server/src/services/cloud-upstreams.ts b/server/src/services/cloud-upstreams.ts index cb4683e8788..0861513f6cd 100644 --- a/server/src/services/cloud-upstreams.ts +++ b/server/src/services/cloud-upstreams.ts @@ -30,6 +30,7 @@ import { } from "@paperclipai/db"; import { badRequest, HttpError, notFound } from "../errors.js"; import { companyPortabilityService } from "./company-portability.js"; +import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; const DEFAULT_SCOPES = ["upstream_import:preview", "upstream_import:write", "upstream_import:read"]; const TRANSFER_SCHEMA = { @@ -39,6 +40,9 @@ const TRANSFER_SCHEMA = { minor: 0, } as const; const DEFAULT_MAX_ENTITIES_PER_CHUNK = 100; +const DISCOVERY_FETCH_TIMEOUT_MS = 30_000; +const REMOTE_FETCH_TIMEOUT_MS = 120_000; +const CLOUD_CREDENTIAL_PREFIX = "paperclip-cloud-credential:"; type NormalizedSha256 = `sha256:${string}`; @@ -75,7 +79,6 @@ type LocalUpstreamExportChunk = { totalChunks: number; byteLength: number; sha256: NormalizedSha256; - manifestHash: NormalizedSha256; payload: { entityKeys: SourceEntityKey[]; }; @@ -100,9 +103,8 @@ type UpstreamTransferManifest = { idempotencyKey: string; generatedAt: string; entityCount: number; - perEntityTypeCounts: Record; entities: UpstreamTransferEntityRecord[]; - chunks: Array>; + chunks: Array & { manifestHash: NormalizedSha256 }>; warnings: UpstreamTransferWarning[]; featureFlags: string[]; manifestHash: NormalizedSha256; @@ -133,7 +135,8 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = .select() .from(cloudUpstreamRuns) .where(eq(cloudUpstreamRuns.companyId, companyId)) - .orderBy(desc(cloudUpstreamRuns.createdAt)), + .orderBy(desc(cloudUpstreamRuns.createdAt)) + .limit(50), ]); return { connections: connectionRows.map(connectionFromRow), @@ -170,7 +173,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = sourceInstanceId, sourceInstanceFingerprint, sourcePublicKey, - privateKeyPem: privateKey.export({ type: "pkcs8", format: "pem" }).toString(), + privateKeyPem: await sealCloudUpstreamCredential(privateKey.export({ type: "pkcs8", format: "pem" }).toString()), tokenStatus: "pending", scopes: scopesFromDiscovery(discovery), targetStackId: target.stackId, @@ -183,7 +186,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = targetSchemaMajor: target.schemaMajor, targetMaxChunkBytes: target.maxChunkBytes, pendingState: state, - pendingCodeVerifier: codeVerifier, + pendingCodeVerifier: await sealCloudUpstreamCredential(codeVerifier), pendingRedirectUri: input.redirectUri, pendingTokenUrl: tokenUrlFromDiscovery(discovery), }).returning(); @@ -221,7 +224,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = grantType: "authorization_code", code: input.code, redirectUri: pending.pendingRedirectUri, - codeVerifier: pending.pendingCodeVerifier, + codeVerifier: await unsealCloudUpstreamCredential(pending.pendingCodeVerifier), }); const accessToken = stringField(tokenResponse, "accessToken"); const token = objectField(tokenResponse, "token"); @@ -231,7 +234,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = .set({ tokenStatus: "connected", authorizedGlobalUserId: optionalString(token.globalUserId), - accessToken, + accessToken: await sealCloudUpstreamCredential(accessToken), tokenId: optionalString(token.id), tokenExpiresAt: expiresAt ? new Date(expiresAt) : null, pendingState: null, @@ -246,8 +249,8 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = return connectionFromRow(updated); }, - preview: async (connectionId: string): Promise => { - const connection = await getConnectionRow(connectionId); + preview: async (connectionId: string, companyId: string): Promise => { + const connection = await getConnectionRow(connectionId, companyId); const basePreview = await localPreview(connection); if (!basePreview.schemaCompatible || connection.tokenStatus !== "connected") { return basePreview; @@ -271,8 +274,8 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = }; }, - createRun: async (input: { connectionId: string; retryOfRunId?: string | null }): Promise => { - const connection = await getConnectionRow(input.connectionId); + createRun: async (input: { connectionId: string; companyId: string; retryOfRunId?: string | null }): Promise => { + const connection = await getConnectionRow(input.connectionId, input.companyId); if (connection.tokenStatus !== "connected") { throw badRequest("Cloud upstream connection is not connected"); } @@ -402,15 +405,15 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = } }, - readRun: async (connectionId: string, runId: string): Promise => { - const row = await getRunRow(connectionId, runId); + readRun: async (connectionId: string, runId: string, companyId: string): Promise => { + const row = await getRunRow(connectionId, runId, companyId); return runFromRow(row); }, - cancelRun: async (connectionId: string, runId: string): Promise => { - const row = await getRunRow(connectionId, runId); + cancelRun: async (connectionId: string, runId: string, companyId: string): Promise => { + const row = await getRunRow(connectionId, runId, companyId); if (row.status !== "running") return runFromRow(row); - const connection = await getConnectionRow(connectionId); + const connection = await getConnectionRow(connectionId, companyId); if (row.remoteRunId) { await remotePost(connection, `/api/upstream-import-runs/${encodeURIComponent(row.remoteRunId)}/cancel`, {}).catch(() => null); } @@ -429,9 +432,10 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = activateRunEntities: async (input: { connectionId: string; runId: string; + companyId: string; entityType: CloudUpstreamActivationEntityType; }): Promise => { - const row = await getRunRow(input.connectionId, input.runId); + const row = await getRunRow(input.connectionId, input.runId, input.companyId); assertActivationEntityType(input.entityType); if (row.status !== "succeeded") { throw badRequest("Only succeeded cloud upstream runs can activate imported entities"); @@ -470,21 +474,27 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = if (!row) throw notFound("Company was not found"); } - async function getConnectionRow(connectionId: string): Promise { + async function getConnectionRow(connectionId: string, companyId?: string): Promise { const row = await db .select() .from(cloudUpstreamConnections) - .where(eq(cloudUpstreamConnections.id, connectionId)) + .where(companyId + ? and(eq(cloudUpstreamConnections.id, connectionId), eq(cloudUpstreamConnections.companyId, companyId)) + : eq(cloudUpstreamConnections.id, connectionId)) .then((rows) => rows[0]); if (!row) throw notFound("Cloud upstream connection was not found"); return row; } - async function getRunRow(connectionId: string, runId: string): Promise { + async function getRunRow(connectionId: string, runId: string, companyId: string): Promise { const row = await db .select() .from(cloudUpstreamRuns) - .where(and(eq(cloudUpstreamRuns.id, runId), eq(cloudUpstreamRuns.connectionId, connectionId))) + .where(and( + eq(cloudUpstreamRuns.id, runId), + eq(cloudUpstreamRuns.connectionId, connectionId), + eq(cloudUpstreamRuns.companyId, companyId), + )) .then((rows) => rows[0]); if (!row) throw notFound("Cloud upstream run was not found"); return row; @@ -596,13 +606,53 @@ async function fetchDiscovery(remoteUrl: string): Promise; } +export async function reconcileCloudUpstreamRunsOnStartup(db: Db, now = new Date()): Promise<{ reconciled: number }> { + const runningRows = await db + .select() + .from(cloudUpstreamRuns) + .where(eq(cloudUpstreamRuns.status, "running")); + if (runningRows.length === 0) return { reconciled: 0 }; + + for (const row of runningRows) { + const report = asRecord(row.report); + await db + .update(cloudUpstreamRuns) + .set({ + status: "failed", + activeStep: row.activeStep, + progressPercent: 100, + completedAt: now, + updatedAt: now, + events: [ + ...safeRunEvents(row.events), + event( + now.toISOString(), + cloudUpstreamStep(row.activeStep), + "failed", + "Marked failed on server startup because the previous process stopped while the cloud upstream run was in progress.", + ), + ], + report: { + ...report, + error: optionalString(report.error) ?? "orphaned_running_run", + errorMessage: optionalString(report.errorMessage) + ?? "The server restarted while this cloud upstream run was running, so Paperclip marked it failed instead of leaving it stuck.", + reconciledAt: now.toISOString(), + }, + }) + .where(eq(cloudUpstreamRuns.id, row.id)); + } + + return { reconciled: runningRows.length }; +} + function firstPathSegment(pathname: string): string | null { const segment = pathname.split("/").find(Boolean); return segment && segment.toLowerCase() !== "dashboard" ? segment : null; @@ -816,34 +866,23 @@ function buildLocalUpstreamExportBundle(input: { idempotencyKey: input.idempotencyKey, generatedAt: new Date(0).toISOString(), entityCount: entities.length, - perEntityTypeCounts: countEntityTypesForManifest(entities), entities: entities.map((entity) => entity.record), - chunks: chunksWithoutManifestHash.map(({ payload: _payload, manifestHash: _manifestHash, ...chunk }) => chunk), + chunks: chunksWithoutManifestHash.map(({ payload: _payload, ...chunk }) => chunk), warnings: input.warnings ?? [], featureFlags: (input.featureFlags ?? ["cloud_sync"]).slice().sort(), }; const manifestHash = normalizedContentHash(manifestWithoutHash); - const chunks = chunksWithoutManifestHash.map((chunk) => ({ ...chunk, manifestHash })); return { manifest: { ...manifestWithoutHash, - chunks: chunks.map(({ payload: _payload, ...chunk }) => chunk), + chunks: chunksWithoutManifestHash.map(({ payload: _payload, ...chunk }) => ({ ...chunk, manifestHash })), manifestHash, }, entities, - chunks, + chunks: chunksWithoutManifestHash, }; } -function countEntityTypesForManifest(entities: LocalUpstreamExportEntity[]): Record { - const counts: Record = {}; - for (const entity of entities) { - const entityType = entity.record.key.sourceEntityType; - counts[entityType] = (counts[entityType] ?? 0) + 1; - } - return counts; -} - function buildLocalChunks(entities: LocalUpstreamExportEntity[], maxEntitiesPerChunk: number): LocalUpstreamExportChunk[] { if (!Number.isInteger(maxEntitiesPerChunk) || maxEntitiesPerChunk < 1) { throw new Error("maxEntitiesPerChunk must be a positive integer"); @@ -864,36 +903,37 @@ function buildLocalChunks(entities: LocalUpstreamExportEntity[], maxEntitiesPerC totalChunks: groups.length, byteLength: Buffer.byteLength(canonicalJson(payload)), sha256: normalizedContentHash(payload), - manifestHash: "sha256:" as NormalizedSha256, payload, }; }); } async function remoteGet(connection: ConnectionRow, path: string): Promise { - const response = await fetch(`${connection.targetOrigin}${path}`, { + const response = await fetchWithTimeout(`${connection.targetOrigin}${path}`, { method: "GET", - headers: proofHeaders(connection, "GET", path), - }); + headers: await proofHeaders(connection, "GET", path), + }, REMOTE_FETCH_TIMEOUT_MS); return parseRemoteResponse(response); } async function remotePost(connection: ConnectionRow, path: string, body: unknown): Promise { - const response = await fetch(`${connection.targetOrigin}${path}`, { + const response = await fetchWithTimeout(`${connection.targetOrigin}${path}`, { method: "POST", headers: { "Content-Type": "application/json", - ...proofHeaders(connection, "POST", path), + ...await proofHeaders(connection, "POST", path), }, body: JSON.stringify(body), - }); + }, REMOTE_FETCH_TIMEOUT_MS); return parseRemoteResponse(response); } -function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch: string): Record { +async function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch: string): Promise> { if (!connection.accessToken || !connection.tokenId) { throw badRequest("Cloud upstream connection is missing an import token"); } + const accessToken = await unsealCloudUpstreamCredential(connection.accessToken); + const privateKeyPem = await unsealCloudUpstreamCredential(connection.privateKeyPem); const timestamp = new Date().toISOString(); const nonce = crypto.randomUUID(); const payload = [ @@ -906,14 +946,14 @@ function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch: nonce, ].join("\n"); return { - Authorization: `Bearer ${connection.accessToken}`, + Authorization: `Bearer ${accessToken}`, "X-Paperclip-Upstream-Source-Instance-Id": connection.sourceInstanceId, "X-Paperclip-Upstream-Proof-Timestamp": timestamp, "X-Paperclip-Upstream-Proof-Nonce": nonce, "X-Paperclip-Upstream-Proof-Signature": sign( null, Buffer.from(payload, "utf8"), - connection.privateKeyPem, + privateKeyPem, ).toString("base64url"), }; } @@ -930,6 +970,32 @@ async function parseRemoteResponse(response: Response): Promise { return parsed; } +async function fetchWithTimeout(input: RequestInfo | URL, init: RequestInit | undefined, timeoutMs: number): Promise { + return fetch(input, { + ...init, + signal: AbortSignal.timeout(timeoutMs), + }); +} + +export async function sealCloudUpstreamCredential(value: string): Promise { + const prepared = await localEncryptedProvider.createSecret({ value }); + return `${CLOUD_CREDENTIAL_PREFIX}${JSON.stringify(prepared.material)}`; +} + +export async function unsealCloudUpstreamCredential(value: string): Promise { + if (!value.startsWith(CLOUD_CREDENTIAL_PREFIX)) return value; + const encoded = value.slice(CLOUD_CREDENTIAL_PREFIX.length); + const parsed = safeParseJson(encoded); + const material = optionalObject(parsed); + if (!material) { + throw badRequest("Invalid encrypted cloud upstream credential material"); + } + return localEncryptedProvider.resolveVersion({ + material, + externalRef: null, + }); +} + export function cloudUpstreamRemoteFailureReport(error: unknown): { error: string; errorMessage?: string; @@ -957,11 +1023,11 @@ function remoteErrorBody(details: unknown): { error?: string; message?: string } } async function postJson(url: string, body: unknown): Promise { - const response = await fetch(url, { + const response = await fetchWithTimeout(url, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(body), - }); + }, DISCOVERY_FETCH_TIMEOUT_MS); const payload = await response.json().catch(() => null); if (!response.ok) { throw badRequest((payload as { error?: string } | null)?.error ?? `Cloud upstream request failed: ${response.status}`); @@ -1026,6 +1092,10 @@ function eventsFromRemote(value: unknown): CloudUpstreamRunEvent[] { }); } +function safeRunEvents(value: unknown): CloudUpstreamRunEvent[] { + return Array.isArray(value) ? value as CloudUpstreamRunEvent[] : []; +} + function assertActivationEntityType(value: string): asserts value is CloudUpstreamActivationEntityType { if (value !== "agents" && value !== "routines" && value !== "monitors") { throw badRequest("entityType must be agents, routines, or monitors"); diff --git a/server/src/services/index.ts b/server/src/services/index.ts index a12dec9ca26..18857b7ca93 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -59,7 +59,7 @@ export type { } from "./authorization.js"; export { boardAuthService } from "./board-auth.js"; export { instanceSettingsService } from "./instance-settings.js"; -export { cloudUpstreamService } from "./cloud-upstreams.js"; +export { cloudUpstreamService, reconcileCloudUpstreamRunsOnStartup } from "./cloud-upstreams.js"; export { companyPortabilityService } from "./company-portability.js"; export { environmentService } from "./environments.js"; export { executionWorkspaceService } from "./execution-workspaces.js"; diff --git a/ui/src/api/cloudUpstreams.ts b/ui/src/api/cloudUpstreams.ts index 1a19fdabcb2..48ddff5ced6 100644 --- a/ui/src/api/cloudUpstreams.ts +++ b/ui/src/api/cloudUpstreams.ts @@ -15,23 +15,23 @@ export const cloudUpstreamsApi = { api.post("/cloud-upstreams/connect/start", input), finishConnect: (input: { pendingConnectionId: string; code: string; state: string }) => api.post("/cloud-upstreams/connect/finish", input), - preview: (connectionId: string) => - api.post(`/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/preview`, {}), - createRun: (connectionId: string, input?: { retryOfRunId?: string | null }) => + preview: (connectionId: string, input: { companyId: string }) => + api.post(`/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/preview`, input), + createRun: (connectionId: string, input: { companyId: string; retryOfRunId?: string | null }) => api.post(`/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs`, input ?? {}), - getRun: (connectionId: string, runId: string) => + getRun: (connectionId: string, runId: string, companyId: string) => api.get( - `/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/${encodeURIComponent(runId)}`, + `/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/${encodeURIComponent(runId)}?companyId=${encodeURIComponent(companyId)}`, ), - cancelRun: (connectionId: string, runId: string) => + cancelRun: (connectionId: string, runId: string, input: { companyId: string }) => api.post( `/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/${encodeURIComponent(runId)}/cancel`, - {}, + input, ), activateEntities: ( connectionId: string, runId: string, - input: { entityType: CloudUpstreamActivationEntityType }, + input: { companyId: string; entityType: CloudUpstreamActivationEntityType }, ) => api.post( `/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/${encodeURIComponent(runId)}/activation`, diff --git a/ui/src/pages/CloudUpstream.test.tsx b/ui/src/pages/CloudUpstream.test.tsx index 551f6ee8e7f..cb6d0c0a835 100644 --- a/ui/src/pages/CloudUpstream.test.tsx +++ b/ui/src/pages/CloudUpstream.test.tsx @@ -142,7 +142,7 @@ describe("CloudUpstream", () => { expect(mockCloudUpstreamsApi.activateEntities).toHaveBeenCalledWith( "connection-1", "run-1", - { entityType: "agents" }, + { companyId: "company-1", entityType: "agents" }, ); await act(async () => { @@ -260,6 +260,39 @@ describe("CloudUpstream", () => { } }); + it("does not retry the OAuth callback finish mutation after an error", async () => { + mockLocationState.pathname = "/PAP/company/settings/cloud-upstream"; + mockLocationState.search = "?code=cb-code&state=cb-state"; + mockCloudUpstreamsApi.list.mockResolvedValue({ connections: [], runs: [] }); + mockCloudUpstreamsApi.finishConnect.mockRejectedValue(new Error("state expired")); + window.localStorage.setItem("paperclip-cloud-upstream-pending-connection", "pending-1"); + + try { + const root = createRoot(container); + const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } }); + + await act(async () => { + root.render( + + + , + ); + }); + await flushReact(); + await flushReact(); + await flushReact(); + + expect(mockCloudUpstreamsApi.finishConnect).toHaveBeenCalledTimes(1); + expect(container.textContent).toContain("state expired"); + + await act(async () => { + root.unmount(); + }); + } finally { + window.localStorage.removeItem("paperclip-cloud-upstream-pending-connection"); + } + }); + it("keeps retry only for failed or cancelled runs", async () => { mockCloudUpstreamsApi.list.mockResolvedValue(stateWithRun(buildRun({ status: "failed" }))); const root = createRoot(container); diff --git a/ui/src/pages/CloudUpstream.tsx b/ui/src/pages/CloudUpstream.tsx index 0e7dee0aa72..1b23a0e545a 100644 --- a/ui/src/pages/CloudUpstream.tsx +++ b/ui/src/pages/CloudUpstream.tsx @@ -119,16 +119,22 @@ export function CloudUpstream() { }, onError: (error) => setActionError(error instanceof Error ? error.message : "Failed to finish connection."), }); + const { + mutate: finishConnect, + isError: finishConnectFailed, + isPending: finishConnectPending, + isSuccess: finishConnectSucceeded, + } = finishMutation; useEffect(() => { - if (!cloudSyncEnabled || !code || !state || finishMutation.isPending || finishMutation.isSuccess) return; + if (!cloudSyncEnabled || !code || !state || finishConnectPending || finishConnectSucceeded || finishConnectFailed) return; const pendingConnectionId = localStorage.getItem(PENDING_CONNECTION_KEY); if (!pendingConnectionId) { setActionError("No pending cloud upstream connection was found. Start the connection again."); return; } - finishMutation.mutate({ pendingConnectionId, code, state }); - }, [cloudSyncEnabled, code, state, finishMutation]); + finishConnect({ pendingConnectionId, code, state }); + }, [cloudSyncEnabled, code, finishConnect, finishConnectFailed, finishConnectPending, finishConnectSucceeded, state]); useEffect(() => { if (callbackError) { @@ -152,7 +158,8 @@ export function CloudUpstream() { }); const previewMutation = useMutation({ - mutationFn: (connectionId: string) => cloudUpstreamsApi.preview(connectionId), + mutationFn: (input: { connectionId: string; companyId: string }) => + cloudUpstreamsApi.preview(input.connectionId, { companyId: input.companyId }), onSuccess: (nextPreview) => { setPreview(nextPreview); setActionError(null); @@ -161,8 +168,11 @@ export function CloudUpstream() { }); const runMutation = useMutation({ - mutationFn: (input: { connectionId: string; retryOfRunId?: string | null }) => - cloudUpstreamsApi.createRun(input.connectionId, { retryOfRunId: input.retryOfRunId ?? null }), + mutationFn: (input: { connectionId: string; companyId: string; retryOfRunId?: string | null }) => + cloudUpstreamsApi.createRun(input.connectionId, { + companyId: input.companyId, + retryOfRunId: input.retryOfRunId ?? null, + }), onSuccess: async (run) => { setActiveRun(run); setNotice(run.status === "succeeded" @@ -175,7 +185,10 @@ export function CloudUpstream() { }); const activationMutation = useMutation({ mutationFn: (input: { run: CloudUpstreamRun; entityType: CloudUpstreamActivationEntityType }) => - cloudUpstreamsApi.activateEntities(input.run.connectionId, input.run.id, { entityType: input.entityType }), + cloudUpstreamsApi.activateEntities(input.run.connectionId, input.run.id, { + companyId: input.run.companyId, + entityType: input.entityType, + }), onSuccess: async (run) => { setActiveRun(run); setNotice("Activation checklist updated."); @@ -271,7 +284,7 @@ export function CloudUpstream() {