diff --git a/server/src/services/remote-workspace-migration.test.ts b/server/src/services/remote-workspace-migration.test.ts index 52a3764ac5c..d50b58dcf3e 100644 --- a/server/src/services/remote-workspace-migration.test.ts +++ b/server/src/services/remote-workspace-migration.test.ts @@ -857,12 +857,18 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); } if (state === "missing-base-uploaded-retry") { - const get = storage.getObject.bind(storage); let reads = 0; - storage.getObject = async (...args) => { - if (args[1].includes("/workspace-snapshots/") && ++reads === 2) throw new Error("restore reply lost"); - return get(...args); - }; - await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("restore reply lost"); + const transaction = db.transaction.bind(db); let interrupted = false; + const fault = vi.spyOn(db, "transaction").mockImplementation(async (action, config) => { + const result = await transaction(action, config); + if (!interrupted && result && typeof result === "object" && "status" in result && result.status === "uploaded") { + interrupted = true; throw new Error("uploaded checkpoint reply lost"); + } + return result; + }); + try { + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("uploaded checkpoint reply lost"); + expect(interrupted).toBe(true); + } finally { fault.mockRestore(); } expect((await db.select().from(executionWorkspaceSnapshots))[0]?.status).toBe("uploaded"); } const interrupted = state.startsWith("missing-base-") || state.startsWith("nested-") @@ -1099,6 +1105,15 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(1); expect(await db.select().from(activityLog)).toHaveLength(1); }); + it("keeps the existing download path for Windows imports", async () => { + const f = await seed(), get = vi.spyOn(storage, "getObject"), platform = vi.spyOn(process, "platform", "get").mockReturnValue("win32"); + try { + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true }); + expect(get).toHaveBeenCalledTimes(2); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy(); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); + } finally { platform.mockRestore(); } + }); it.each([false, true])("removes plaintext verification files with read-only folders after import (source changed: %s)", async sourceChanged => { const f = await seed(), protectedDir = path.join(f.cwd, "protected"), scratchRoot = path.join(root, "scratch"); await fs.mkdir(protectedDir); await fs.mkdir(scratchRoot); @@ -1107,13 +1122,14 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { vi.stubEnv("TMPDIR", scratchRoot); const get = storage.getObject.bind(storage); let reads = 0; if (sourceChanged) storage.getObject = async (...args) => { - if (++reads === 2) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); + if (++reads === 1) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); return get(...args); }; try { const migration = migrateRemoteWorkspace({ db, storage, ...f, apply: true }); if (sourceChanged) { await expect(migration).rejects.toThrow("remote_workspace_migration_source_changed"); + expect(reads).toBe(1); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); } else { await expect(migration).resolves.toMatchObject({ applied: true, alreadyMigrated: true }); @@ -1452,29 +1468,32 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { it("rechecks a service that becomes unresolved during snapshot verification", async () => { const f = await seed(), get = storage.getObject.bind(storage); let reads = 0; storage.getObject = async (...args) => { - if (++reads === 2) await db.insert(workspaceRuntimeServices).values({ id: randomUUID(), companyId: f.companyId, + if (++reads === 1) await db.insert(workspaceRuntimeServices).values({ id: randomUUID(), companyId: f.companyId, executionWorkspaceId: f.workspaceId, scopeType: "execution_workspace", scopeId: f.workspaceId, serviceName: "Late synthetic preview", status: "running", lifecycle: "shared", provider: "local_process", cwd: f.cwd }); return get(...args); }; await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("workspace_runtime_service_unsettled"); + expect(reads).toBe(1); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); }); it("rechecks a writer that appears while the restored copy is being verified", async () => { const f = await seed(), get = storage.getObject.bind(storage); let reads = 0; storage.getObject = async (...args) => { - if (++reads === 2) await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId, + if (++reads === 1) await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId, phase: "workspace_command", status: "running" }); return get(...args); }; await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("workspace_operation_unresolved"); + expect(reads).toBe(1); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); }); it("preserves a local edit made during verification and refuses activation", async () => { const f = await seed(), get = storage.getObject.bind(storage); let reads = 0; - storage.getObject = async (...args) => { if (++reads === 2) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); return get(...args); }; + storage.getObject = async (...args) => { if (++reads === 1) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); return get(...args); }; await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("source_changed"); + expect(reads).toBe(1); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("new local work\n"); }); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index c20bf712ec8..8b9b2552baa 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -402,7 +402,7 @@ export async function migrateRemoteWorkspace(input: { } const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.companyId, workspaceId: input.workspaceId, operationId: operation.id, generation: 1, epoch: 1 }); - const scratch = await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-")); + const scratch = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-"))); try { input.signal?.throwIfAborted(); const retained = reserved.retainedDirectory?.restorePoint; @@ -435,15 +435,18 @@ export async function migrateRemoteWorkspace(input: { } const exclude = nested.map(child => child.relativePath); const captured = recovered ?? await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, requireNoGit: onlineSource?.kind === "ordinary", gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal }); + const restoredArchive = path.join(scratch, "verified.pcws"), restored = path.join(scratch, "restored"); + const verifiedArchivePath = process.platform === "win32" ? undefined : restoredArchive; const ref = await uploadWorkspaceSnapshot({ db: input.db, storage: input.storage, snapshotId: snapshot.id, companyId: input.companyId, workspaceId: input.workspaceId, archivePath, manifestDigest: captured.manifestDigest, signal: input.signal, + verifiedArchivePath, // Private Git reconstruction can change administrative bytes between // attempts. Its journaled logical recipe is checked at reservation and // again under source locks before authority publication below. ...(reserved.archivedGitSource ? { unpublishedRecapture: { sourceOperationId: operation.id } } : {}), }); - const restoredArchive = path.join(scratch, "verified.pcws"), restored = path.join(scratch, "restored"); - await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, targetPath: restoredArchive, signal: input.signal }); + if (!verifiedArchivePath) await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, + targetPath: restoredArchive, signal: input.signal }); await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir: restored, expectedManifestDigest: ref.manifestDigest, signal: input.signal }); if (!recovered) { const currentArchive = path.join(scratch, "current.pcws"); diff --git a/server/src/services/remote-workspace-online-source.test.ts b/server/src/services/remote-workspace-online-source.test.ts index 76c8cb99646..576af4605f9 100644 --- a/server/src/services/remote-workspace-online-source.test.ts +++ b/server/src/services/remote-workspace-online-source.test.ts @@ -267,7 +267,9 @@ describe.skipIf(!support.supported)("online ordinary physical source fence", () const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Unrelated active agent" }).returning(); const run = await remoteWriter(f.companyId, agent.id); expect(await inventoryRemoteWorkspaceMigration(db, f)).toMatchObject({ eligible: true }); + const get = vi.spyOn(storage, "getObject"); await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true }); + expect(get).toHaveBeenCalledTimes(1); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toEqual(run); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("saved legacy edits\n"); expect(await stored(f.aliases[1].workspace.id)).toEqual(f.aliases[1].workspace); diff --git a/server/src/services/remote-workspace-snapshot-storage.test.ts b/server/src/services/remote-workspace-snapshot-storage.test.ts index a2d938f009b..b34c6888a3c 100644 --- a/server/src/services/remote-workspace-snapshot-storage.test.ts +++ b/server/src/services/remote-workspace-snapshot-storage.test.ts @@ -3,6 +3,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { Readable } from "node:stream"; +import { appendFileSync, readdirSync } from "node:fs"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; import { applyPendingMigrations, closeRegisteredClients, createDb, executionWorkspaceSnapshots } from "@paperclipai/db"; @@ -10,6 +11,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; import { createStorageService } from "../storage/service.js"; import type { StorageService } from "../storage/types.js"; +import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; import { captureRemoteWorkspaceDirectory, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; import { beginWorkspaceSnapshot, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, recoverWorkspaceSnapshotArchive, uploadWorkspaceSnapshot } from "./remote-workspace-snapshot-storage.js"; @@ -30,7 +32,7 @@ suite("remote workspace snapshot storage", () => { afterAll(async () => { if (externalDatabaseUrl) await closeRegisteredClients(externalDatabaseUrl); else await database?.cleanup(); }); beforeEach(async () => { vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "41".repeat(32)); - root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-store-test-")); + root = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-store-test-"))); storage = createStorageService(createLocalDiskStorageProvider(path.join(root, "objects"))); }); afterEach(async () => { vi.unstubAllEnvs(); await db.delete(executionWorkspaceSnapshots); await fs.rm(root, { recursive: true, force: true }); }); @@ -57,6 +59,67 @@ suite("remote workspace snapshot storage", () => { const [row] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, ref.id)); expect(row?.status).toBe("uploaded"); // The owning service still controls atomic publication. }); + it.each(["fresh", "uploaded", "lost-reply"])("reuses one authenticated %s readback for a complete cold restore", async state => { + const f = await fixture(), put = storage.putFile.bind(storage); + if (state === "uploaded") await uploadWorkspaceSnapshot(f.upload); + if (state === "lost-reply") { + storage.putFile = async input => { await put(input); throw new Error("lost upload reply"); }; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("lost upload reply"); + storage.putFile = put; + } + const get = vi.spyOn(storage, "getObject"), verifiedArchivePath = path.join(root, "verified.pcws"); + const request = { ...f.upload, verifiedArchivePath }, ref = await uploadWorkspaceSnapshot(request); + if (!await fs.lstat(verifiedArchivePath).catch(() => null)) + await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: verifiedArchivePath }); + expect(get).toHaveBeenCalledTimes(1); + expect(await fs.readFile(verifiedArchivePath)).toEqual(await fs.readFile(f.archivePath)); + expect((await fs.stat(verifiedArchivePath)).mode & 0o777).toBe(0o600); + expect(await getWorkspaceSnapshot(db, { ...f, snapshotId: ref.id })).toEqual(ref); + await fs.rm(f.cwd, { recursive: true }); await fs.rm(f.archivePath); + await restoreRemoteWorkspaceArchive({ archivePath: verifiedArchivePath, targetDir: path.join(root, "cold"), expectedManifestDigest: ref.manifestDigest }); + expect(await fs.readFile(path.join(root, "cold", "saved.txt"), "utf8")).toBe("acknowledged workspace content\n"); + expect((await fs.readdir(root)).some(name => name.startsWith(".workspace-upload-readback-"))).toBe(false); + }); + it.each(["existing", "public-parent", "linked-parent"])("refuses a caller-owned readback path with %s ownership", async mode => { + const f = await fixture(), parent = path.join(root, "destination"); await fs.mkdir(parent, { mode: 0o700 }); + let verifiedArchivePath = path.join(parent, "archive.pcws"); + if (mode === "existing") await fs.writeFile(verifiedArchivePath, "keep caller bytes"); + if (mode === "public-parent") await fs.chmod(parent, 0o755); + if (mode === "linked-parent") { + const alias = path.join(root, "alias"); await fs.symlink(parent, alias); verifiedArchivePath = path.join(alias, "archive.pcws"); + } + const get = vi.spyOn(storage, "getObject"), put = vi.spyOn(storage, "putFile"); + await expect(uploadWorkspaceSnapshot({ ...f.upload, verifiedArchivePath })).rejects.toThrow("readback_"); + expect(get).not.toHaveBeenCalled(); expect(put).not.toHaveBeenCalled(); + if (mode === "existing") expect(await fs.readFile(verifiedArchivePath, "utf8")).toBe("keep caller bytes"); + expect((await db.select().from(executionWorkspaceSnapshots))[0].status).toBe("pending"); + }); + it("rejects a tampered verified plaintext before publishing the caller path", async () => { + const f = await fixture(), verifiedArchivePath = path.join(root, "tampered.pcws"), debug = db.$client.options.debug; + let changed = false; + db.$client.options.debug = (_connection, query) => { + if (changed || !query.startsWith('update "execution_workspace_snapshots"') || !query.includes('"status" =')) return; + const directory = readdirSync(root).find(name => name.startsWith(".workspace-upload-readback-")); + if (!directory) return; + const file = readdirSync(path.join(root, directory)).find(name => name.endsWith(".pcws")); + if (file) { appendFileSync(path.join(root, directory, file), "changed plaintext"); changed = true; } + }; + try { + await expect(uploadWorkspaceSnapshot({ ...f.upload, verifiedArchivePath })).rejects.toThrow("readback_file_changed"); + expect(changed).toBe(true); + expect(await fs.lstat(verifiedArchivePath).catch(() => null)).toBeNull(); + expect((await fs.readdir(root)).some(name => name.startsWith(".workspace-upload-readback-"))).toBe(false); + } finally { db.$client.options.debug = debug; } + }); + it("keeps GCM authentication mandatory for an uploaded readback and removes all failed plaintext", async () => { + const f = await fixture(), ref = await uploadWorkspaceSnapshot(f.upload), verifiedArchivePath = path.join(root, "bad-key.pcws"); + const wrapped = JSON.parse(await localEncryptedProvider.resolveVersion({ material: ref.key, externalRef: null })); + const { material } = await localEncryptedProvider.createVersion({ value: JSON.stringify({ ...wrapped, dataKey: Buffer.alloc(32, 8).toString("base64") }) }); + await db.update(executionWorkspaceSnapshots).set({ key: material }).where(eq(executionWorkspaceSnapshots.id, ref.id)); + await expect(uploadWorkspaceSnapshot({ ...f.upload, verifiedArchivePath })).rejects.toThrow(); + expect(await fs.lstat(verifiedArchivePath).catch(() => null)).toBeNull(); + expect((await fs.readdir(root)).some(name => name.startsWith(".workspace-upload-readback-"))).toBe(false); + }); it("reconciles an upload whose reply was lost after object persistence, without the old scratch file", async () => { const f = await fixture(); const put = storage.putFile.bind(storage); let loseReply = true; storage.putFile = async input => { const result = await put(input); if (loseReply) { loseReply = false; throw new Error("lost upload reply"); } return result; }; @@ -156,7 +219,8 @@ suite("remote workspace snapshot storage", () => { expect(saved).toMatchObject({ status: "pending", objectKey: null }); expect(saved!.metadata!.uploads).toHaveLength(1); }); - it.each(["original", "recapture", "original-with-lost-reply"])("keeps a %s winner when another physical capture finishes late", async winner => { + it.each(["original", "recapture", "original-with-lost-reply"].flatMap(winner => [false, true].map(readback => ({ winner, readback }))))( + "keeps a $winner winner when another physical capture finishes late (readback: $readback)", async ({ winner, readback }) => { const f = await fixture(), put = storage.putFile.bind(storage); const deferred = () => { let resolve!: () => void; @@ -174,14 +238,16 @@ suite("remote workspace snapshot storage", () => { return saved; }; const unpublishedRecapture = { sourceOperationId: f.operationId }; - const original = uploadWorkspaceSnapshot({ ...f.upload, unpublishedRecapture }); + const originalPath = path.join(root, "original-verified.pcws"), recapturedPath = path.join(root, "recaptured-verified.pcws"); + const original = uploadWorkspaceSnapshot({ ...f.upload, unpublishedRecapture, ...(readback ? { verifiedArchivePath: originalPath } : {}) }); let recapture: ReturnType | undefined; try { await originalReady.promise; const archivePath = path.join(root, "recaptured.pcws"); await fs.writeFile(path.join(f.cwd, "saved.txt"), "different physical capture\n"); const captured = await captureRemoteWorkspaceDirectory({ cwd: f.cwd, archivePath }); - recapture = uploadWorkspaceSnapshot({ ...f.upload, archivePath, manifestDigest: captured.manifestDigest, unpublishedRecapture }); + recapture = uploadWorkspaceSnapshot({ ...f.upload, archivePath, manifestDigest: captured.manifestDigest, unpublishedRecapture, + ...(readback ? { verifiedArchivePath: recapturedPath } : {}) }); await recaptureReady.promise; (winner === "recapture" ? recaptureRelease : originalRelease).resolve(); if (winner === "original-with-lost-reply") { @@ -193,6 +259,11 @@ suite("remote workspace snapshot storage", () => { originalRelease.resolve(); recaptureRelease.resolve(); if (winner !== "original-with-lost-reply") expect(await original).toEqual(saved); expect(await recapture).toEqual(saved); + if (readback) { + const winningBytes = await fs.readFile(winner === "recapture" ? archivePath : f.archivePath); + expect(await fs.readFile(recapturedPath)).toEqual(winningBytes); + if (winner !== "original-with-lost-reply") expect(await fs.readFile(originalPath)).toEqual(winningBytes); + } expect(await getWorkspaceSnapshot(db, { ...f, snapshotId: f.row.id })).toEqual(saved); expect((await db.select().from(executionWorkspaceSnapshots))[0]!.metadata!.uploads).toHaveLength(2); } finally { originalRelease.resolve(); recaptureRelease.resolve(); await Promise.allSettled([original, recapture]); } diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 12c38b45bae..a9bae63c069 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -1,6 +1,6 @@ import { createCipheriv, createDecipheriv, createHash, randomBytes, randomUUID } from "node:crypto"; -import { createReadStream, createWriteStream } from "node:fs"; -import { link, mkdtemp, open, rm } from "node:fs/promises"; +import { constants, createReadStream, createWriteStream } from "node:fs"; +import { link, lstat, mkdtemp, open, realpath, rm } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { pipeline } from "node:stream/promises"; @@ -121,22 +121,74 @@ export async function recoverWorkspaceSnapshotArchive(input: { return null; } -/** Upload attempts are journaled before remote I/O. The caller owns publication. */ -export async function uploadWorkspaceSnapshot(input: { +type WorkspaceSnapshotUpload = { db: Db; storage: StorageService; snapshotId: string; companyId: string; workspaceId: string; - archivePath: string; manifestDigest: string; signal?: AbortSignal; + archivePath: string; manifestDigest: string; signal?: AbortSignal; verifiedArchivePath?: string; // Only an owner with an immutable logical source may replace an unavailable, // unpublished capture. A recoverable or concurrently published attempt wins, // and its returned manifest may differ from the new physical capture. unpublishedRecapture?: { sourceOperationId: string }; -}): Promise { +}; + +/** Upload attempts are journaled before remote I/O. The caller owns publication. */ +export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): Promise { + if (input.verifiedArchivePath === undefined) + return uploadVerifiedWorkspaceSnapshot(input, ref => verifyObject(input.storage, ref, input.signal)); + input.signal?.throwIfAborted(); + const target = input.verifiedArchivePath, parent = path.dirname(target); + if (!path.isAbsolute(target) || path.normalize(target) !== target) failure("readback_parent_invalid"); + const owner = await lstat(parent); + const assertParent = async () => { + const current = await lstat(parent); + if (!current.isDirectory() || current.isSymbolicLink() || (current.mode & 0o777) !== 0o700 + || typeof process.getuid === "function" && current.uid !== process.getuid() + || current.dev !== owner.dev || current.ino !== owner.ino || await realpath(parent) !== parent) failure("readback_parent_invalid"); + }; + await assertParent(); + await lstat(target).then(() => failure("readback_target_exists"), (error: NodeJS.ErrnoException) => { + if (error.code !== "ENOENT") throw error; + }); + const scratch = await mkdtemp(path.join(parent, ".workspace-upload-readback-")); + const verified: Array<{ ref: WorkspaceSnapshotRef; file: string }> = []; + const verify = async (ref: WorkspaceSnapshotRef) => { + if (verified.some(saved => isDeepStrictEqual(saved.ref, ref))) return; + const file = path.join(scratch, `${randomUUID()}.pcws`); + await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, targetPath: file, + signal: input.signal, temporaryParent: scratch }); + verified.push({ ref: structuredClone(ref), file }); + }; + try { + const ref = await uploadVerifiedWorkspaceSnapshot(input, verify); + await verify(ref); + const file = verified.find(saved => isDeepStrictEqual(saved.ref, ref))!.file; + const wrapped = await snapshotKey(ref), handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK); + try { + const before = await handle.stat(), digest = createHash("sha256"); let bytes = 0; + if (!before.isFile() || (before.mode & 0o777) !== 0o600 || before.nlink !== 1 + || typeof process.getuid === "function" && before.uid !== process.getuid()) failure("readback_file_changed"); + for await (const chunk of createReadStream(file, { fd: handle.fd, autoClose: false, signal: input.signal })) { + digest.update(chunk); bytes += chunk.length; + } + const after = await handle.stat(), named = await lstat(file); + if (!named.isFile() || named.isSymbolicLink() || named.dev !== before.dev || named.ino !== before.ino + || named.nlink !== 1 || named.mode !== before.mode || after.size !== before.size + || after.mtimeMs !== before.mtimeMs || after.ctimeMs !== before.ctimeMs + || bytes !== wrapped.plaintextBytes || digest.digest("hex") !== wrapped.plaintextSha256) failure("readback_file_changed"); + await assertParent(); input.signal?.throwIfAborted(); + await link(file, target); + } finally { await handle.close(); } + return ref; + } finally { await rm(scratch, { recursive: true, force: true }); } +} + +async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, verify: (ref: WorkspaceSnapshotRef) => Promise): Promise { const [row] = await input.db.select().from(executionWorkspaceSnapshots).where(scope(input)); if (!row) failure("missing"); if (input.unpublishedRecapture && row!.sourceOperationId !== input.unpublishedRecapture.sourceOperationId) failure("operation_binding_changed"); if (["uploaded", "committed", "superseded"].includes(row!.status)) { const saved = reference(row!); if (saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed"); - await verifyObject(input.storage, saved, input.signal); return saved; + await verify(saved); return saved; } if (row!.status !== "pending" || !/^[a-f0-9]{64}$/.test(input.manifestDigest)) failure("not_writable"); const recapture = Boolean(input.unpublishedRecapture); @@ -157,7 +209,7 @@ export async function uploadWorkspaceSnapshot(input: { if (!winning || !["uploaded", "committed", "superseded"].includes(winning.status)) failure("publication_changed"); const saved = reference(winning!); if (!recapture && saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed"); - if (saved.objectKey !== attempt.objectKey) await verifyObject(input.storage, saved, input.signal); + if (saved.objectKey !== attempt.objectKey) await verify(saved); return saved; }; // Recover a completed upload even when the previous process lost its reply @@ -169,7 +221,7 @@ export async function uploadWorkspaceSnapshot(input: { const ref = { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef; validateReference(ref); if ((await input.storage.headObject(ref.companyId, ref.objectKey, { signal: input.signal })).exists) { - await verifyObject(input.storage, ref, input.signal); return attempt; + await verify(ref); return attempt; } } return null; @@ -217,14 +269,14 @@ export async function uploadWorkspaceSnapshot(input: { if (!recorded) { const saved = await getWorkspaceSnapshot(input.db, input); if (!saved || (!recapture && saved.manifestDigest !== input.manifestDigest)) failure("publication_changed"); - await verifyObject(input.storage, saved, input.signal); return saved; + await verify(saved); return saved; } // The object key is unique per attempt, so a late losing uploader cannot // overwrite an object already used by a committed generation. await input.storage.putFile({ companyId: identity.companyId, namespace: "workspace-snapshots", objectKey: attempt.objectKey, originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted), byteSize: attempt.byteSize, sha256: attempt.sha256, signal: input.signal }); - await verifyObject(input.storage, { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef, input.signal); + await verify({ schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef); if (recapture) { const recovered = await availableAttempt(attempts); if (recovered) return publish(recovered); @@ -233,16 +285,21 @@ export async function uploadWorkspaceSnapshot(input: { } finally { await rm(scratch, { recursive: true, force: true }); } } +async function snapshotKey(ref: WorkspaceSnapshotRef) { + validateReference(ref); + const wrapped = JSON.parse(await localEncryptedProvider.resolveVersion({ material: ref.key, externalRef: null })); + if (wrapped.schema !== KEY_SCHEMA || wrapped.companyId !== ref.companyId || wrapped.workspaceId !== ref.workspaceId + || wrapped.id !== ref.id || wrapped.generation !== ref.generation || !/^[a-f0-9]{64}$/.test(wrapped.plaintextSha256 ?? "") + || !Number.isSafeInteger(wrapped.plaintextBytes) || wrapped.plaintextBytes < 0) failure("key_binding_invalid"); + return wrapped; +} + /** Authentication finishes in private scratch before any plaintext is published. */ export async function loadWorkspaceSnapshotArchive(input: { storage: StorageService; reference: WorkspaceSnapshotRef; targetPath: string; signal?: AbortSignal; temporaryParent?: string; }): Promise { input.signal?.throwIfAborted(); - const ref = input.reference; validateReference(ref); - const wrapped = JSON.parse(await localEncryptedProvider.resolveVersion({ material: ref.key, externalRef: null })); - if (wrapped.schema !== KEY_SCHEMA || wrapped.companyId !== ref.companyId || wrapped.workspaceId !== ref.workspaceId - || wrapped.id !== ref.id || wrapped.generation !== ref.generation || !/^[a-f0-9]{64}$/.test(wrapped.plaintextSha256 ?? "") - || !Number.isSafeInteger(wrapped.plaintextBytes) || wrapped.plaintextBytes < 0) failure("key_binding_invalid"); + const ref = input.reference, wrapped = await snapshotKey(ref); const scratch = await mkdtemp(path.join(input.temporaryParent ?? os.tmpdir(), "pc-workspace-decrypt-")); try { input.signal?.throwIfAborted();