diff --git a/packages/db/src/migrations/0087_cloud_upstreams.sql b/packages/db/src/migrations/0087_cloud_upstreams.sql index caf14522551..4eb06a21374 100644 --- a/packages/db/src/migrations/0087_cloud_upstreams.sql +++ b/packages/db/src/migrations/0087_cloud_upstreams.sql @@ -5,11 +5,11 @@ CREATE TABLE IF NOT EXISTS "cloud_upstream_connections" ( "source_instance_id" text NOT NULL, "source_instance_fingerprint" text NOT NULL, "source_public_key" text NOT NULL, - "private_key_pem" text NOT NULL, + "private_key_material" jsonb NOT NULL, "token_status" text NOT NULL, "scopes" text[] DEFAULT '{}' NOT NULL, "authorized_global_user_id" text, - "access_token" text, + "access_token_material" jsonb, "token_id" text, "token_expires_at" timestamp with time zone, "target_stack_id" text NOT NULL, diff --git a/packages/db/src/schema/cloud_upstreams.ts b/packages/db/src/schema/cloud_upstreams.ts index c7e2160b4da..636e8b52c31 100644 --- a/packages/db/src/schema/cloud_upstreams.ts +++ b/packages/db/src/schema/cloud_upstreams.ts @@ -10,11 +10,11 @@ export const cloudUpstreamConnections = pgTable( sourceInstanceId: text("source_instance_id").notNull(), sourceInstanceFingerprint: text("source_instance_fingerprint").notNull(), sourcePublicKey: text("source_public_key").notNull(), - privateKeyPem: text("private_key_pem").notNull(), + privateKeyMaterial: jsonb("private_key_material").$type>().notNull(), tokenStatus: text("token_status").notNull(), scopes: text("scopes").array().notNull().default([]), authorizedGlobalUserId: text("authorized_global_user_id"), - accessToken: text("access_token"), + accessTokenMaterial: jsonb("access_token_material").$type>(), tokenId: text("token_id"), tokenExpiresAt: timestamp("token_expires_at", { withTimezone: true }), diff --git a/server/src/__tests__/cloud-upstreams-routes-authz.test.ts b/server/src/__tests__/cloud-upstreams-routes-authz.test.ts new file mode 100644 index 00000000000..3a59962b016 --- /dev/null +++ b/server/src/__tests__/cloud-upstreams-routes-authz.test.ts @@ -0,0 +1,144 @@ +import express from "express"; +import request from "supertest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mockCloudUpstreamService = vi.hoisted(() => ({ + list: vi.fn(), + readConnection: vi.fn(), + startConnect: vi.fn(), + finishConnect: vi.fn(), + preview: vi.fn(), + createRun: vi.fn(), + readRun: vi.fn(), + cancelRun: vi.fn(), + activateRunEntities: vi.fn(), +})); + +vi.mock("../routes/authz.js", async () => { + const { forbidden } = await vi.importActual("../errors.js"); + + function assertBoardOrgAccess(req: Express.Request) { + if (req.actor.type !== "board") { + throw forbidden("Board access required"); + } + } + + function assertCompanyAccess(req: Express.Request, companyId: string) { + assertBoardOrgAccess(req); + if (!req.actor.companyIds?.includes(companyId)) { + throw forbidden("User does not have access to this company"); + } + } + + return { + assertBoardOrgAccess, + assertCompanyAccess, + }; +}); + +vi.mock("../services/index.js", () => ({ + cloudUpstreamService: () => mockCloudUpstreamService, + instanceSettingsService: () => ({ + getExperimental: vi.fn(async () => ({ enableCloudSync: true })), + }), +})); + +let routeModules: + | Promise<[ + typeof import("../middleware/index.js"), + typeof import("../routes/cloud-upstreams.js"), + ]> + | null = null; + +async function loadRouteModules() { + routeModules ??= Promise.all([ + import("../middleware/index.js"), + import("../routes/cloud-upstreams.js"), + ]); + return routeModules; +} + +async function createApp(companyIds: string[]) { + const [{ errorHandler }, { cloudUpstreamRoutes }] = await loadRouteModules(); + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { + (req as any).actor = { + type: "board", + source: "session", + companyIds, + }; + next(); + }); + app.use("/api", cloudUpstreamRoutes({} as any)); + app.use(errorHandler); + return app; +} + +describe("cloud upstream route authorization", () => { + beforeEach(() => { + vi.clearAllMocks(); + mockCloudUpstreamService.list.mockResolvedValue({ connections: [], runs: [] }); + mockCloudUpstreamService.readConnection.mockResolvedValue({ + id: "connection-1", + companyId: "company-a", + remoteUrl: "https://cloud.example", + target: { + stackId: "stack-1", + stackSlug: null, + stackDisplayName: null, + companyId: "cloud-company", + primaryHost: "cloud.example", + origin: "https://cloud.example", + product: "Paperclip Cloud", + schemaMajor: 1, + maxChunkBytes: 1024, + }, + tokenStatus: "connected", + scopes: [], + authorizedGlobalUserId: null, + expiresAt: null, + createdAt: new Date(0).toISOString(), + updatedAt: new Date(0).toISOString(), + lastRunId: null, + }); + mockCloudUpstreamService.preview.mockResolvedValue({ + connectionId: "connection-1", + sourceCompanyId: "company-a", + target: { + stackId: "stack-1", + stackSlug: null, + stackDisplayName: null, + companyId: "cloud-company", + primaryHost: "cloud.example", + origin: "https://cloud.example", + product: "Paperclip Cloud", + schemaMajor: 1, + maxChunkBytes: 1024, + }, + schemaCompatible: true, + summary: [], + warnings: [], + conflicts: [], + generatedAt: new Date(0).toISOString(), + }); + }); + + it("rejects list requests for companies outside the board actor membership", async () => { + const app = await createApp(["company-a"]); + + const res = await request(app).get("/api/cloud-upstreams?companyId=company-b"); + + expect(res.status).toBe(403); + expect(mockCloudUpstreamService.list).not.toHaveBeenCalled(); + }); + + it("checks connection company access before previewing a connection-scoped endpoint", async () => { + const app = await createApp(["company-b"]); + + const res = await request(app).post("/api/cloud-upstreams/connection-1/push-runs/preview").send({}); + + expect(res.status).toBe(403); + expect(mockCloudUpstreamService.preview).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/__tests__/cloud-upstreams.test.ts b/server/src/__tests__/cloud-upstreams.test.ts index bb7e84d92bb..95b347f6798 100644 --- a/server/src/__tests__/cloud-upstreams.test.ts +++ b/server/src/__tests__/cloud-upstreams.test.ts @@ -18,6 +18,32 @@ describe("cloud upstream remote failures", () => { }); }); + it("redacts credential-shaped fields from cloud response details", () => { + const body = { + error: "bad_request", + message: "remote rejected request", + accessToken: "upt_secret", + nested: { + authorization: "Bearer upt_secret", + privateKeyPem: "-----BEGIN PRIVATE KEY-----", + }, + }; + + expect(cloudUpstreamRemoteFailureReport(new HttpError(400, "bad_request", body))).toEqual({ + error: "bad_request", + errorMessage: "remote rejected request", + details: { + error: "bad_request", + message: "remote rejected request", + accessToken: "[redacted]", + nested: { + authorization: "[redacted]", + privateKeyPem: "[redacted]", + }, + }, + }); + }); + it("falls back to the thrown error message for non-remote failures", () => { expect(cloudUpstreamRemoteFailureReport(new Error("network failed"))).toEqual({ error: "network failed", diff --git a/server/src/routes/cloud-upstreams.ts b/server/src/routes/cloud-upstreams.ts index 5c9c24bd08f..95c5fe7235e 100644 --- a/server/src/routes/cloud-upstreams.ts +++ b/server/src/routes/cloud-upstreams.ts @@ -1,7 +1,7 @@ import { Router } from "express"; import type { Db } from "@paperclipai/db"; import { badRequest, notFound } from "../errors.js"; -import { assertBoardOrgAccess } from "./authz.js"; +import { assertBoardOrgAccess, assertCompanyAccess } from "./authz.js"; import { cloudUpstreamService, instanceSettingsService } from "../services/index.js"; export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {}) { @@ -16,10 +16,16 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = { } } + async function assertConnectionAccess(req: Parameters[0], connectionId: string) { + const connection = await service.readConnection(connectionId); + assertCompanyAccess(req, connection.companyId); + } + router.get("/cloud-upstreams", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); const companyId = stringQuery(req.query.companyId, "companyId"); + assertCompanyAccess(req, companyId); res.json(await service.list(companyId)); }); @@ -27,6 +33,7 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = { assertBoardOrgAccess(req); await assertEnabled(); const companyId = stringBody(req.body, "companyId"); + assertCompanyAccess(req, companyId); const remoteUrl = stringBody(req.body, "remoteUrl"); const redirectUri = stringBody(req.body, "redirectUri"); res.json(await service.startConnect({ companyId, remoteUrl, redirectUri })); @@ -35,8 +42,10 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = { router.post("/cloud-upstreams/connect/finish", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); + const pendingConnectionId = stringBody(req.body, "pendingConnectionId"); + await assertConnectionAccess(req, pendingConnectionId); res.json(await service.finishConnect({ - pendingConnectionId: stringBody(req.body, "pendingConnectionId"), + pendingConnectionId, code: stringBody(req.body, "code"), state: stringBody(req.body, "state"), })); @@ -45,12 +54,14 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = { router.post("/cloud-upstreams/:connectionId/push-runs/preview", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); + await assertConnectionAccess(req, req.params.connectionId); res.json(await service.preview(req.params.connectionId)); }); router.post("/cloud-upstreams/:connectionId/push-runs", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); + await assertConnectionAccess(req, req.params.connectionId); res.json(await service.createRun({ connectionId: req.params.connectionId, retryOfRunId: optionalString(req.body?.retryOfRunId), @@ -60,18 +71,21 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = { router.get("/cloud-upstreams/:connectionId/push-runs/:runId", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); + await assertConnectionAccess(req, req.params.connectionId); res.json(await service.readRun(req.params.connectionId, req.params.runId)); }); router.post("/cloud-upstreams/:connectionId/push-runs/:runId/cancel", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); + await assertConnectionAccess(req, req.params.connectionId); res.json(await service.cancelRun(req.params.connectionId, req.params.runId)); }); router.post("/cloud-upstreams/:connectionId/push-runs/:runId/activation", async (req, res) => { assertBoardOrgAccess(req); await assertEnabled(); + await assertConnectionAccess(req, req.params.connectionId); res.json(await service.activateRunEntities({ connectionId: req.params.connectionId, runId: req.params.runId, diff --git a/server/src/services/cloud-upstreams.ts b/server/src/services/cloud-upstreams.ts index cb4683e8788..4043b61e1d3 100644 --- a/server/src/services/cloud-upstreams.ts +++ b/server/src/services/cloud-upstreams.ts @@ -1,5 +1,5 @@ import crypto, { sign } from "node:crypto"; -import { and, count, desc, eq } from "drizzle-orm"; +import { and, count, desc, eq, lt } from "drizzle-orm"; import type { CloudUpstreamConnectStartResponse, CloudUpstreamActivationDecision, @@ -29,9 +29,12 @@ import { routines, } from "@paperclipai/db"; import { badRequest, HttpError, notFound } from "../errors.js"; +import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; import { companyPortabilityService } from "./company-portability.js"; const DEFAULT_SCOPES = ["upstream_import:preview", "upstream_import:write", "upstream_import:read"]; +const RUN_HISTORY_LIMIT = 50; +const RUN_STALE_AFTER_MS = 30 * 60 * 1000; const TRANSFER_SCHEMA = { family: "paperclip-upstream-transfer", version: "1.0.0", @@ -123,6 +126,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = return { list: async (companyId: string): Promise => { + await recoverStaleRunningRuns(companyId); const [connectionRows, runRows] = await Promise.all([ db .select() @@ -133,7 +137,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(RUN_HISTORY_LIMIT), ]); return { connections: connectionRows.map(connectionFromRow), @@ -141,6 +146,10 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = }; }, + readConnection: async (connectionId: string): Promise => { + return connectionFromRow(await getConnectionRow(connectionId)); + }, + startConnect: async (input: { companyId: string; remoteUrl: string; @@ -158,6 +167,9 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = const codeChallenge = crypto.createHash("sha256").update(codeVerifier, "utf8").digest("base64url"); const { publicKey, privateKey } = crypto.generateKeyPairSync("ed25519"); const sourcePublicKey = publicKey.export({ type: "spki", format: "pem" }).toString(); + const privateKeyMaterial = await encryptCloudCredential( + privateKey.export({ type: "pkcs8", format: "pem" }).toString(), + ); const sourceInstanceFingerprint = `sha256:${crypto .createHash("sha256") .update(publicKey.export({ type: "spki", format: "der" })) @@ -170,7 +182,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = sourceInstanceId, sourceInstanceFingerprint, sourcePublicKey, - privateKeyPem: privateKey.export({ type: "pkcs8", format: "pem" }).toString(), + privateKeyMaterial, tokenStatus: "pending", scopes: scopesFromDiscovery(discovery), targetStackId: target.stackId, @@ -224,6 +236,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = codeVerifier: pending.pendingCodeVerifier, }); const accessToken = stringField(tokenResponse, "accessToken"); + const accessTokenMaterial = await encryptCloudCredential(accessToken); const token = objectField(tokenResponse, "token"); const expiresAt = optionalString(token.expiresAt) ?? optionalString(tokenResponse.expiresAt); const [updated] = await db @@ -231,7 +244,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = .set({ tokenStatus: "connected", authorizedGlobalUserId: optionalString(token.globalUserId), - accessToken, + accessTokenMaterial, tokenId: optionalString(token.id), tokenExpiresAt: expiresAt ? new Date(expiresAt) : null, pendingState: null, @@ -276,18 +289,11 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = if (connection.tokenStatus !== "connected") { throw badRequest("Cloud upstream connection is not connected"); } - const preview = await localPreview(connection); - if (!preview.schemaCompatible) { - throw badRequest("Cloud stack schema is not compatible with this local Paperclip version"); - } - - const bundle = await buildBundle(connection, "apply"); const runId = crypto.randomUUID(); const now = new Date(); const initialEvents = [ event(now.toISOString(), "connect", "completed", "Connected to the target Paperclip Cloud stack."), - event(now.toISOString(), "scan", "completed", "Scanned the local company inventory."), - event(now.toISOString(), "preview", "completed", "Generated the transfer manifest."), + event(now.toISOString(), "scan", "updated", "Queued local company inventory scan."), ...(input.retryOfRunId ? [event(now.toISOString(), "push", "retrying", `Retrying run ${input.retryOfRunId} with the same import ledger idempotency key.`)] : []), @@ -297,113 +303,31 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = connectionId: connection.id, companyId: connection.companyId, status: "running", - activeStep: "push", - progressPercent: 45, + activeStep: "scan", + progressPercent: 5, dryRun: false, retryOfRunId: input.retryOfRunId ?? null, - summary: preview.summary, - warnings: preview.warnings, - conflicts: preview.conflicts, + summary: [], + warnings: [], + conflicts: [], events: initialEvents, report: {}, - idempotencyKey: bundle.manifest.idempotencyKey, - manifestHash: bundle.manifest.manifestHash, + idempotencyKey: `pending:${runId}`, + manifestHash: `pending:${runId}`, targetUrl: connection.targetOrigin, createdAt: now, updatedAt: now, }).returning(); if (!created) throw badRequest("Failed to create cloud upstream run"); - try { - const remoteRun = await remotePost(connection, `/api/companies/${encodeURIComponent(connection.targetCompanyId)}/upstream-imports/runs`, { - mode: "apply", - manifest: bundle.manifest, - entities: bundle.entities, - }); - const remoteRunId = remoteRunIdFromResponse(remoteRun); - await updateRun(runId, { - remoteRunId, - activeStep: "push", - progressPercent: 60, - events: [ - ...initialEvents, - event(new Date().toISOString(), "push", "updated", "Created or resumed the cloud import ledger run."), - ], - }); - - for (const chunk of bundle.chunks) { - await remotePost(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/chunks`, chunk); - } - await updateRun(runId, { - activeStep: "verify", - progressPercent: 82, - events: [ - ...initialEvents, - event(new Date().toISOString(), "push", "completed", `Uploaded ${bundle.chunks.length} manifest chunk${bundle.chunks.length === 1 ? "" : "s"}.`), - ], - }); - - const applied = await remotePost(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/apply`, {}); - const remoteEvents = await remoteGet(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/events`).catch(() => null); - const completedAt = new Date(); - const finalEvents = [ - ...initialEvents, - event(completedAt.toISOString(), "push", "completed", "Pushed mapped objects without duplicate creation."), - event(completedAt.toISOString(), "verify", "completed", "Verified the cloud import ledger and generated a run report."), - event(completedAt.toISOString(), "activate", "completed", "Activation checklist is ready for manual unpause decisions."), - ...eventsFromRemote(remoteEvents), - ]; - const finalRun = await updateRun(runId, { - remoteRunId, - status: "succeeded", - activeStep: "activate", - progressPercent: 100, - warnings: mergeWarnings(preview.warnings, warningsFromRemote(applied)), - conflicts: conflictsFromRemote(applied), - events: finalEvents, - report: { - runId, - remoteRunId, - target: targetFromConnectionRow(connection), - manifestHash: bundle.manifest.manifestHash, - idempotencyKey: bundle.manifest.idempotencyKey, - retryOfRunId: input.retryOfRunId ?? null, - result: applied, - events: remoteEvents, - }, - completedAt, - }); - await db - .update(cloudUpstreamConnections) - .set({ lastRunId: finalRun.id, updatedAt: new Date() }) - .where(eq(cloudUpstreamConnections.id, connection.id)); - return finalRun; - } catch (error) { - const failedAt = new Date(); - const failure = cloudUpstreamRemoteFailureReport(error); - return updateRun(runId, { - status: "failed", - activeStep: "push", - progressPercent: 100, - events: [ - ...initialEvents, - event(failedAt.toISOString(), "push", "failed", failure.errorMessage ?? failure.error), - ], - report: { - runId, - target: targetFromConnectionRow(connection), - manifestHash: bundle.manifest.manifestHash, - idempotencyKey: bundle.manifest.idempotencyKey, - retryOfRunId: input.retryOfRunId ?? null, - ...failure, - }, - completedAt: failedAt, - }); - } + void runCloudPush({ connection, runId, retryOfRunId: input.retryOfRunId ?? null, initialEvents }).catch(() => undefined); + return runFromRow(created); }, readRun: async (connectionId: string, runId: string): Promise => { const row = await getRunRow(connectionId, runId); + const recovered = await recoverStaleRun(row); + if (recovered) return recovered; return runFromRow(row); }, @@ -500,6 +424,179 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = return runFromRow(updated); } + async function runCloudPush(input: { + connection: ConnectionRow; + runId: string; + retryOfRunId: string | null; + initialEvents: CloudUpstreamRunEvent[]; + }): Promise { + let preview: CloudUpstreamPreview | null = null; + let bundle: LocalUpstreamExportBundle | null = null; + try { + preview = await localPreview(input.connection); + if (!preview.schemaCompatible) { + throw badRequest("Cloud stack schema is not compatible with this local Paperclip version"); + } + await updateRun(input.runId, { + activeStep: "preview", + progressPercent: 25, + summary: preview.summary, + warnings: preview.warnings, + conflicts: preview.conflicts, + events: [ + ...input.initialEvents, + event(new Date().toISOString(), "scan", "completed", "Scanned the local company inventory."), + event(new Date().toISOString(), "preview", "updated", "Building the transfer manifest."), + ], + }); + + bundle = await buildBundle(input.connection, "apply"); + await updateRun(input.runId, { + activeStep: "push", + progressPercent: 45, + idempotencyKey: bundle.manifest.idempotencyKey, + manifestHash: bundle.manifest.manifestHash, + events: [ + ...input.initialEvents, + event(new Date().toISOString(), "scan", "completed", "Scanned the local company inventory."), + event(new Date().toISOString(), "preview", "completed", "Generated the transfer manifest."), + ], + }); + + const remoteRun = await remotePost(input.connection, `/api/companies/${encodeURIComponent(input.connection.targetCompanyId)}/upstream-imports/runs`, { + mode: "apply", + manifest: bundle.manifest, + entities: bundle.entities, + }); + const remoteRunId = remoteRunIdFromResponse(remoteRun); + await updateRun(input.runId, { + remoteRunId, + activeStep: "push", + progressPercent: 60, + events: [ + ...input.initialEvents, + event(new Date().toISOString(), "scan", "completed", "Scanned the local company inventory."), + event(new Date().toISOString(), "preview", "completed", "Generated the transfer manifest."), + event(new Date().toISOString(), "push", "updated", "Created or resumed the cloud import ledger run."), + ], + }); + + for (const chunk of bundle.chunks) { + await remotePost(input.connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/chunks`, chunk); + } + await updateRun(input.runId, { + activeStep: "verify", + progressPercent: 82, + events: [ + ...input.initialEvents, + event(new Date().toISOString(), "scan", "completed", "Scanned the local company inventory."), + event(new Date().toISOString(), "preview", "completed", "Generated the transfer manifest."), + event(new Date().toISOString(), "push", "completed", `Uploaded ${bundle.chunks.length} manifest chunk${bundle.chunks.length === 1 ? "" : "s"}.`), + ], + }); + + const applied = await remotePost(input.connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/apply`, {}); + const remoteEvents = await remoteGet(input.connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/events`).catch(() => null); + const completedAt = new Date(); + const finalEvents = [ + ...input.initialEvents, + event(completedAt.toISOString(), "push", "completed", "Pushed mapped objects without duplicate creation."), + event(completedAt.toISOString(), "verify", "completed", "Verified the cloud import ledger and generated a run report."), + event(completedAt.toISOString(), "activate", "completed", "Activation checklist is ready for manual unpause decisions."), + ...eventsFromRemote(remoteEvents), + ]; + const finalRun = await updateRun(input.runId, { + remoteRunId, + status: "succeeded", + activeStep: "activate", + progressPercent: 100, + warnings: mergeWarnings(preview.warnings, warningsFromRemote(applied)), + conflicts: conflictsFromRemote(applied), + events: finalEvents, + report: { + runId: input.runId, + remoteRunId, + target: targetFromConnectionRow(input.connection), + manifestHash: bundle.manifest.manifestHash, + idempotencyKey: bundle.manifest.idempotencyKey, + retryOfRunId: input.retryOfRunId, + result: applied, + events: remoteEvents, + }, + completedAt, + }); + await db + .update(cloudUpstreamConnections) + .set({ lastRunId: finalRun.id, updatedAt: new Date() }) + .where(eq(cloudUpstreamConnections.id, input.connection.id)); + } catch (error) { + const failedAt = new Date(); + const failure = cloudUpstreamRemoteFailureReport(error); + await updateRun(input.runId, { + status: "failed", + activeStep: "push", + progressPercent: 100, + ...(preview ? { + summary: preview.summary, + warnings: preview.warnings, + conflicts: preview.conflicts, + } : {}), + ...(bundle ? { + idempotencyKey: bundle.manifest.idempotencyKey, + manifestHash: bundle.manifest.manifestHash, + } : {}), + events: [ + ...input.initialEvents, + event(failedAt.toISOString(), "push", "failed", failure.errorMessage ?? failure.error), + ], + report: { + runId: input.runId, + target: targetFromConnectionRow(input.connection), + manifestHash: bundle?.manifest.manifestHash ?? null, + idempotencyKey: bundle?.manifest.idempotencyKey ?? null, + retryOfRunId: input.retryOfRunId, + ...failure, + }, + completedAt: failedAt, + }); + } + } + + async function recoverStaleRunningRuns(companyId: string): Promise { + const cutoff = new Date(Date.now() - RUN_STALE_AFTER_MS); + const staleRows = await db + .select() + .from(cloudUpstreamRuns) + .where(and( + eq(cloudUpstreamRuns.companyId, companyId), + eq(cloudUpstreamRuns.status, "running"), + lt(cloudUpstreamRuns.updatedAt, cutoff), + )); + await Promise.all(staleRows.map((row) => recoverStaleRun(row))); + } + + async function recoverStaleRun(row: RunRow): Promise { + if (row.status !== "running") return null; + if (Date.now() - row.updatedAt.getTime() <= RUN_STALE_AFTER_MS) return null; + return updateRun(row.id, { + status: "failed", + activeStep: row.activeStep, + progressPercent: 100, + completedAt: new Date(), + events: [ + ...row.events, + event(new Date().toISOString(), "push", "failed", "Cloud upstream push stopped before completion. Retry to resume with the same idempotency safeguards."), + ], + report: { + ...asRecord(row.report), + staleRecovery: { + recoveredAt: new Date().toISOString(), + reason: "server_restart_or_worker_timeout", + }, + }, + }); + } + async function localPreview(connection: ConnectionRow): Promise { return { connectionId: connection.id, @@ -873,7 +970,7 @@ function buildLocalChunks(entities: LocalUpstreamExportEntity[], maxEntitiesPerC async function remoteGet(connection: ConnectionRow, path: string): Promise { const response = await fetch(`${connection.targetOrigin}${path}`, { method: "GET", - headers: proofHeaders(connection, "GET", path), + headers: await proofHeaders(connection, "GET", path), }); return parseRemoteResponse(response); } @@ -883,17 +980,21 @@ async function remotePost(connection: ConnectionRow, path: string, body: unknown method: "POST", headers: { "Content-Type": "application/json", - ...proofHeaders(connection, "POST", path), + ...await proofHeaders(connection, "POST", path), }, body: JSON.stringify(body), }); return parseRemoteResponse(response); } -function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch: string): Record { - if (!connection.accessToken || !connection.tokenId) { +async function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch: string): Promise> { + if (!connection.accessTokenMaterial || !connection.tokenId) { throw badRequest("Cloud upstream connection is missing an import token"); } + const [accessToken, privateKeyPem] = await Promise.all([ + decryptCloudCredential(connection.accessTokenMaterial, "Cloud upstream import token"), + decryptCloudCredential(connection.privateKeyMaterial, "Cloud upstream signing key"), + ]); const timestamp = new Date().toISOString(); const nonce = crypto.randomUUID(); const payload = [ @@ -906,14 +1007,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"), }; } @@ -943,10 +1044,39 @@ export function cloudUpstreamRemoteFailureReport(error: unknown): { return { error: remote.error ?? error.message, ...(remote.message ? { errorMessage: remote.message } : {}), - ...(error.details !== undefined ? { details: error.details } : {}), + ...(error.details !== undefined ? { details: redactCloudCredentialDetails(error.details) } : {}), }; } +async function encryptCloudCredential(value: string): Promise> { + const prepared = await localEncryptedProvider.createSecret({ value }); + return prepared.material; +} + +async function decryptCloudCredential(material: Record, label: string): Promise { + try { + return await localEncryptedProvider.resolveVersion({ material, externalRef: null }); + } catch (error) { + throw badRequest(`${label} could not be decrypted`, { + error: error instanceof Error ? error.message : String(error), + }); + } +} + +function redactCloudCredentialDetails(value: unknown): unknown { + if (Array.isArray(value)) return value.map(redactCloudCredentialDetails); + if (!value || typeof value !== "object") return value; + const output: Record = {}; + for (const [key, entry] of Object.entries(value)) { + if (/access.?token|authorization|private.?key|bearer/i.test(key)) { + output[key] = "[redacted]"; + } else { + output[key] = redactCloudCredentialDetails(entry); + } + } + return output; +} + function remoteErrorBody(details: unknown): { error?: string; message?: string } { const record = optionalObject(details); if (!record) return {}; diff --git a/ui/src/pages/CloudUpstream.tsx b/ui/src/pages/CloudUpstream.tsx index 0e7dee0aa72..250f41c66d2 100644 --- a/ui/src/pages/CloudUpstream.tsx +++ b/ui/src/pages/CloudUpstream.tsx @@ -92,10 +92,14 @@ export function CloudUpstream() { queryKey: selectedCompanyId ? queryKeys.cloudUpstreams(selectedCompanyId) : ["cloud-upstreams", "__disabled__"], queryFn: () => cloudUpstreamsApi.list(selectedCompanyId!), enabled: !!selectedCompanyId && cloudSyncEnabled, + refetchInterval: (query) => + query.state.data?.runs.some((run) => run.status === "running") ? 2000 : false, }); const connection = upstreamQuery.data?.connections[0] ?? null; - const latestRun = activeRun ?? upstreamQuery.data?.runs[0] ?? null; + const latestRun = activeRun + ? upstreamQuery.data?.runs.find((run) => run.id === activeRun.id) ?? activeRun + : upstreamQuery.data?.runs[0] ?? null; const callbackParams = useMemo(() => new URLSearchParams(location.search), [location.search]); const code = callbackParams.get("code"); @@ -165,9 +169,11 @@ export function CloudUpstream() { cloudUpstreamsApi.createRun(input.connectionId, { retryOfRunId: input.retryOfRunId ?? null }), onSuccess: async (run) => { setActiveRun(run); - setNotice(run.status === "succeeded" - ? "Push run completed. Review activation before unpausing automations." - : "Push run failed. Review the run events and retry after correcting the issue."); + setNotice(run.status === "running" + ? "Push run started. Progress will update while the upload continues." + : run.status === "succeeded" + ? "Push run completed. Review activation before unpausing automations." + : "Push run failed. Review the run events and retry after correcting the issue."); setActionError(null); await invalidateUpstreams(); },