mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 06:16:06 -10:00
fix(native): recover exact historical provider pack code (#19779)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work.
> - Native sessions retain the provider pack that admitted their saved
state.
> - An image update can replace that pack's executable, proxy, or
contract.
> - Older sessions can predate the durable three-file launch artifact
receipt.
> - The current image's archive cannot satisfy a different pinned
manifest.
> - This pull request obtains the exact public code from its recorded
release image.
> - Full source, archive, and installed-file checks still control
provider admission.
## Linked Issues or Issue Description
Refs #19558, #19622, #19755.
A saved OpenCode session can fail with
`runner_remote_provider_archive_invalid` after an image update. Its
original provider state and checkpoint remain valid, but its pinned pack
differs from the controller's pack. The old manifest can name a release
that still has the exact code in GHCR. Matching the OpenCode version
does not prove matching proxy or contract bytes.
## What Changed
- Resolve a pinned historical pack from the recorded source-SHA image in
the public Paperclip GHCR repository. Verify the platform manifest, raw
config, source identity, dedicated code layer, selected pack manifest,
archive contents, paths, and hashes.
- Keep a private temporary cache of verified public code. Rebuild it on
an empty host. Bound each source to 4 GiB and three minutes, with two
active imports or cache verifications, 32 distinct queued sources, and
disk-headroom checks. Share warm-cache verification by manifest while
each caller keeps its own cancellation.
- Select this verified archive through the existing uploader. Preserve
installed-pack and preinstalled-pack checks. Refuse unavailable
historical transport before clearing the target. Keep the current-source
identity guard from #19755. Refuse an oversized layer argument before
any target effect.
- Attach the source wait to the actual per-turn startup. Stop, lease
loss, and startup completion cancel it. Keep existing Runner admission
and turn limits.
- Document the registry retention requirement and the distinction
between public code recovery and encrypted provider-state backup.
## Verification
- Genuine baseline RED: the real uploader rejects the selected old pack
before any remote effect. Treatment installs the exact old code through
that uploader.
- Current head: all 19 source/integrity/cache/deadline controls, all 45
archive controls, and 29 actual-caller and manifest/startup controls
pass (93 total). Four review regressions fail against the prior head and
pass with the fixes. The server and colocated test TypeScript graph
passes. Module-boundary and whitespace checks pass.
- Actual public GHCR cold reconstruction: 1,054,587,008 compressed code
bytes, 115.4 s, about 248 MiB maximum owner RSS. All 114 file layers and
the overlay passed source validation.
- Linux Node 24.22.2 fixture: the real historical pack passes the normal
uploader, full installed-file verifier, and pinned OpenCode version
check. It used one transfer, no application startup, no network, and no
provider calls. Revision12b0d2ba passed in 220.2 s with about 282 MiB
owner RSS. The earlier prototype passed in 108.5 s; both measurements
are retained. The three-minute source limit and the fixture's
four-minute limit were unchanged. The later review fixes add shared
cache verification, hash cancellation, and an early size guard. They do
not change extraction or the accepted Linux fixture input.
- Earlier fixture failures are retained: unsafe hardlink classification
was corrected with confined-link controls; a cached image lacked the
required zstd CLI; Docker Desktop bind storage caused GNU tar
directory-identity errors. The passing fixture uses the official Debian
zstd binary and an owned Linux tmpfs target. Production extraction code
is unchanged.
- Fresh Greptile review is 5/5 on
7b1a249478 with no open findings. All
required CI gates passed on the reviewed head. [CI
run](https://github.com/paperclipai/paperclip/actions/runs/42353011485),
[fresh
review](https://github.com/paperclipai/paperclip/pull/19779#issuecomment-6704065994).
No hosted session was retried by this change.
## Risks
- Cold historical recovery reads about 1 GiB of public code for the
measured release. Registry latency or missing images can block it. The
source deadline remains three minutes.
- GHCR administrators can delete release images. Source-SHA tags and
this temporary cache are not a permanent code-retention guarantee. A
provider-state S3 backup does not replace legacy executable bytes.
- Targets must support the existing qualified archive transport and
selected platform. Failed checks preserve the saved state and prevent
provider dispatch.
- The Linux fixture proves code installation and verification. Hosted
conversation continuation still needs acceptance after normal delivery.
## Model Used
OpenAI Codex, GPT-6 family, with reasoning, repository tools, and local
code execution. The host does not expose the exact deployment ID or
context-window size.
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes:` / `Closes:`
/ `Refs:` OR (b) described the issue in-PR following the relevant issue
template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub references)
- [x] My branch name describes the change and contains no internal
Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
---------
Co-authored-by: Paperclip <[email protected]>
This commit is contained in:
1 parent
ae2d2956b3
commit
909f4d1a33
10 files changed
+1051
-24
No files matched your search
@@ -60,12 +60,68 @@ with the prior session state, receipt, and backups. The cache is not a new
|
||||
execution authority and must not remain active for the fresh provider session.
|
||||
|
||||
Live Runner reattachment does not download or replace the launch files.
|
||||
Historical sessions that recorded image-rooted paths keep those paths. This
|
||||
change cannot reconstruct files that were never saved. A governed fresh-session
|
||||
Historical sessions that recorded image-rooted paths keep those paths. A governed fresh-session
|
||||
transition can use a new pack after existing effects and receipts are reconciled.
|
||||
It must not replay a previous turn or treat a missing result as proof that no
|
||||
effect occurred.
|
||||
|
||||
## Historical provider-pack code sources
|
||||
|
||||
An older session can retain its exact pack manifest while the current server
|
||||
image contains a different pack. When neither the staged nor the preinstalled
|
||||
pack verifies, the normal uploader resolves the selected historical code from
|
||||
`ghcr.io/paperclipai/paperclip`. It uses the manifest's recorded source revision
|
||||
and Linux architecture. The full source-SHA image tag is a lookup hint; a tag
|
||||
name alone does not prove the code identity.
|
||||
|
||||
The resolver verifies the OCI index, selected platform manifest, image config,
|
||||
source revision, and dedicated provider-pack transport layer. It then verifies
|
||||
the selected canonical pack manifest, compressed layer hashes, tar paths and
|
||||
links, and each file layer's plaintext bytes. The final installed pack still
|
||||
passes the normal verification against the original selected manifest. The
|
||||
OpenCode version alone cannot admit a different proxy or contract. This source
|
||||
selection also serves other providers that use the full provider-pack uploader;
|
||||
it does not change their compatibility or qualification rules.
|
||||
|
||||
The code source is downloaded and verified before the uploader can clear or
|
||||
transfer anything to the target. A historical source requires the existing
|
||||
qualified archive transport. An unavailable transport leaves the target intact.
|
||||
Missing images, changed bytes, invalid source identity, and unsupported packs
|
||||
fail closed. The resolver never resets provider state, substitutes the current
|
||||
pack, changes the stored launch paths, or starts an old application image.
|
||||
|
||||
The temporary cache contains public code and source provenance only. It uses
|
||||
private, process-account-owned directories and exact manifest keys. An empty
|
||||
host can reconstruct the cache from the registry. A cached source is checked
|
||||
again before upload. Imports stream data with a four-GiB code limit and a
|
||||
three-minute deadline, allow two active imports or cache verifications, and bound queued distinct
|
||||
sources to 32. Concurrent callers share verification of the same cached pack.
|
||||
Import waits and downloads respond to cancellation; one cancelled
|
||||
caller does not cancel a shared import still needed by another caller. Inactive
|
||||
instances perform no registry reads or cache scans. Each import requires disk
|
||||
headroom before writing; completed cached code remains until temporary storage
|
||||
is removed.
|
||||
|
||||
The normal per-turn startup owns the source wait, including OpenCode starts
|
||||
without a warm execution context. Stop, execution-lease loss, and startup
|
||||
completion cancel that wait. An existing execution context can cancel it sooner.
|
||||
The source deadline applies before Runner admission. It does not increase the
|
||||
separate cold-admission or provider-turn time limits.
|
||||
|
||||
The resolver uses only an anonymous token scoped to reading this public code
|
||||
repository. It does not use database, sandbox, model, or object-store credentials.
|
||||
Its cache and registry dependency are separate from the encrypted S3 backup of
|
||||
provider conversation state and workspace files.
|
||||
|
||||
The Docker publication workflow publishes full source-SHA tags. There is no
|
||||
image-deletion workflow in this repository, but this is not a permanent registry
|
||||
retention guarantee. GitHub package administrators can
|
||||
[delete package versions](https://docs.github.com/en/packages/learn-github-packages/deleting-and-restoring-a-package).
|
||||
Historical continuation therefore depends on retaining the corresponding code
|
||||
image in GHCR, or retaining a verified source cache. Deleting both keeps the
|
||||
session blocked with its original state intact; an S3 provider-state backup alone
|
||||
does not reconstruct legacy image-owned executable code.
|
||||
|
||||
Artifact cleanup uses the existing bounded checkpoint and backup expiration
|
||||
owners. It requires closed work, settled native ownership, and exact stopped
|
||||
remote execution receipts. Backup retirement releases only that backup reference.
|
||||
|
||||
@@ -112,6 +112,7 @@ import { forwardRemoteMcpRequest } from "./remote-mcp-relay.js";
|
||||
import { createCapabilityRunnerdProviderEnvironment } from "@paperclipai/paperclip-runner/live";
|
||||
import * as nativeSessionResume from "./native-session-resume.js";
|
||||
import * as nativeLeaseRenewal from "./native-execution-lease.js";
|
||||
import * as retainedPackSource from "./retained-provider-pack-source.js";
|
||||
import { buildNativeHeartbeatPreparationSpans, createNativeRunTrace } from "./native-run-trace.js";
|
||||
import { NativeRunnerOwnershipUnverifiedError } from "./native-runner-ownership.js";
|
||||
import type { AdapterRuntimeEvent } from "../../adapters/index.js";
|
||||
@@ -7488,6 +7489,65 @@ function cancellationDb(options?: {
|
||||
}
|
||||
|
||||
describe("native startup cancellation fence", () => {
|
||||
it.each(["stop", "lease-loss"])("cancels the actual per-turn OpenCode code-source wait on %s before provider dispatch", async mode => {
|
||||
const root = await mkdtemp(join(tmpdir(), "native-pack-source-caller-"));
|
||||
const daemonPath = join(root, "paperclip-runnerd");
|
||||
const daemon = Buffer.alloc(64); daemon.set([0x7f, 0x45, 0x4c, 0x46, 2, 1]); daemon.writeUInt16LE(62, 18);
|
||||
await writeFile(daemonPath, daemon);
|
||||
const sha = "sha256:" + "a".repeat(64);
|
||||
const artifacts = Object.fromEntries(Object.entries({ grokLauncher: "dist/providers/grok/launcher.cjs", nodeCommand: "node_modules/node/bin/node",
|
||||
productionLock: "pnpm-lock.yaml", opencodeCommand: "node_modules/.bin/opencode", opencodeExecutable: "node_modules/opencode-ai/bin/opencode.exe",
|
||||
opencodeProxy: "dist/cli/opencode-app-server-proxy.cjs", acpxSidecar: "dist/cli/acpx-runtime-sidecar.cjs" }).map(([name, path]) => [name, { path, sha256: sha }]));
|
||||
const payload = { compatibility: { opencode: 1, acpx: 1 }, contractDigest: sha, target: { platform: "linux", architecture: "x64" }, runnerSourceRevision: "a".repeat(40),
|
||||
pins: { nodeMinimum: "24.11.0", codex: QUALIFIED_ACPX_PROFILES.codex.agentRuntimeVersion, opencode: "1.18.34", acpx: "0.13.1",
|
||||
grok: QUALIFIED_ACPX_PROFILES.grok.agentRuntimeVersion, claudeAcp: QUALIFIED_ACPX_PROFILES.claude.agentServerVersion, codexAcp: QUALIFIED_ACPX_PROFILES.codex.agentServerVersion },
|
||||
acpxProfileDigests: Object.fromEntries(["grok", "claude", "codex"].map(agent => [agent, QUALIFIED_ACPX_PROFILES[agent as "grok" | "claude" | "codex"].commandDigest])),
|
||||
artifacts, distDigest: sha, bridgeDigest: "sha256:" + createHash("sha256").update(sha + "\n" + sha + "\n" + sha).digest("hex") };
|
||||
await writeFile(join(root, "provider-pack.json"), JSON.stringify({ schema: "paperclip-runner/remote-provider-pack/v1", payload,
|
||||
digest: "sha256:" + createHash("sha256").update(canonicalProviderPackJson(payload)).digest("hex") }));
|
||||
const current = { ...execution, binding: { ...execution.binding, runId: randomUUID(), executionWorkspaceId: randomUUID() },
|
||||
workspace: { ...execution.workspace, cwd: root }, provider: { kind: "opencode", model: "openai/test" },
|
||||
session: { ...execution.session, normalizedSessionId: randomUUID(), driverKind: "opencode_server", lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } } as NativeExecutionInputV1;
|
||||
const execute = vi.fn(async (command: { args?: string[] }) => ({ exitCode: command.args?.[0] === "-e" ? 1 : 0, timedOut: false, signal: null, stderr: "", stdout: command.args?.[0] === "--build-metadata" ? JSON.stringify({
|
||||
schema: "paperclip-runner/runnerd-build-metadata/v1", binaryName: "paperclip-runnerd", packageName: "@paperclipai/paperclip-runner", binaryContractVersion: 2,
|
||||
durableSessionCapabilities: ["unlimited_runtime", "connection_lease_renewal"], prpTransportModes: ["listen_ws"], providerPacks: [],
|
||||
}) : "" }));
|
||||
const syncIn = vi.fn(async (_operations: Array<{ files: Array<{ targetPath?: string }> }>) => ({ operations: [] }));
|
||||
let entered!: () => void, release!: () => void, sourceSignal: AbortSignal | undefined, lost: (() => Promise<unknown>) | undefined;
|
||||
const reached = new Promise<void>(done => { entered = done; });
|
||||
const source = vi.spyOn(retainedPackSource, "resolveProviderPackUploadSource").mockImplementation(input => {
|
||||
sourceSignal = input.signal; entered();
|
||||
return new Promise((_resolve, reject) => {
|
||||
release = () => reject(new Error("owned fixture cleanup"));
|
||||
input.signal?.addEventListener("abort", () => reject(input.signal!.reason), { once: true });
|
||||
});
|
||||
});
|
||||
const originalRenewal = nativeLeaseRenewal.startNativeSessionExecutionLeaseRenewal;
|
||||
const renewal = vi.spyOn(nativeLeaseRenewal, "startNativeSessionExecutionLeaseRenewal").mockImplementation(input => {
|
||||
lost = () => Promise.resolve(input.onLost());
|
||||
return originalRenewal(input);
|
||||
});
|
||||
const running = executePaperclipNativeSession({ db: leaseDb(current), execution: current, useRunnerd: true,
|
||||
runnerInstanceId: "source-caller-runner", runnerRemoteBinaryPath: daemonPath, runnerRemoteProviderPackPath: root,
|
||||
runnerEnvironment: { ...configuredEnvironmentProjection({}), PAPERCLIP_NATIVE_MCP_NAME: "source-caller", PAPERCLIP_NATIVE_MCP_URL: "http://127.0.0.1/mcp/gateways/source-caller", PAPERCLIP_NATIVE_MCP_TOKEN: "owned-fixture" },
|
||||
runnerIngressAuthorized: true, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace", environmentId: "environment", leaseId: "lease", providerKey: "daytona",
|
||||
effectiveCapabilities: { runnerWebSocketIngress: true }, runner: { execute, syncIn } } as never });
|
||||
const failure = running.then(() => { throw Error("unexpected provider completion"); }, error => error);
|
||||
try {
|
||||
await Promise.race([reached, failure.then(error => { throw error; })]);
|
||||
expect(sourceSignal).toBeInstanceOf(AbortSignal);
|
||||
const stopping = mode === "stop" ? cancelNativeSession(current.binding.runId, "owned source wait Stop") : lost!();
|
||||
expect(await failure).toMatchObject({ message: mode === "stop" ? "native_cancellation_pending_recovery" : "native_controller_detached_for_restart" });
|
||||
await stopping;
|
||||
expect(sourceSignal!.reason).toMatchObject({ message: mode === "stop" ? "native_cancellation_pending_recovery" : "native_session_lease_lost" });
|
||||
expect(sourceSignal!.aborted).toBe(true);
|
||||
expect(state.createBackend).not.toHaveBeenCalled();
|
||||
expect(syncIn.mock.calls.flatMap(call => (call[0] as { files?: Array<{ targetPath?: string }> } | undefined)?.files ?? []).some(file => file.targetPath?.endsWith("/provider-pack"))).toBe(false);
|
||||
} finally {
|
||||
release?.(); await failure; source.mockRestore(); renewal.mockRestore(); await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["deadline", "stop"])("retires a claimed warm owner when %s interrupts its reuse diagnostic", async mode => {
|
||||
const first: NativeExecutionInputV1 = { ...execution,
|
||||
binding: { ...execution.binding, runId: `reuse-log-first-${mode}`, executionWorkspaceId: `reuse-log-workspace-${mode}` },
|
||||
|
||||
@@ -378,6 +378,7 @@ type NativeSessionStartup = {
|
||||
priorRestartDetach?: symbol;
|
||||
leaseOwner?: string;
|
||||
executionLeaseRenewal?: NativeExecutionLeaseRenewal;
|
||||
providerPackSourceAbort: AbortController;
|
||||
/** A fresh claim whose preparation failed before any provider turn. */
|
||||
unstartedClaim?: { leaseOwner: string; attempt: number };
|
||||
/** A provider session handle was published; a turn may have followed it. */
|
||||
@@ -8780,6 +8781,7 @@ async function performNativeCancellation(
|
||||
&& startup.sessionPublished !== true && startup.providerWorkMayPrecede !== true));
|
||||
if (startup) {
|
||||
startup.stopRequested = true;
|
||||
startup.providerPackSourceAbort.abort(new NativeCancellationPendingRecoveryError());
|
||||
startup.executionContext?.abort(new Error("native_session_cancelled"));
|
||||
startup.cancellationSettled = new Promise<void>(resolve => { settleStartupCancellation = resolve; });
|
||||
}
|
||||
@@ -9161,6 +9163,7 @@ export async function executePaperclipNativeSession(input: {
|
||||
let resolveStartup!: (session: ActiveNativeSession | null) => void;
|
||||
let settle!: () => void;
|
||||
const startup: NativeSessionStartup = {
|
||||
providerPackSourceAbort: new AbortController(),
|
||||
priorRestartDetach: nativeRunsDetachingForRestart.get(runId),
|
||||
providerWorkMayPrecede: Boolean(input.restartRecovery),
|
||||
settled: new Promise<void>(resolve => { settle = resolve; }),
|
||||
@@ -9317,6 +9320,7 @@ export async function executePaperclipNativeSession(input: {
|
||||
}
|
||||
throw error;
|
||||
} finally {
|
||||
startup.providerPackSourceAbort.abort(new Error("native_execution_startup_completed"));
|
||||
for (const lease of diagnosticLeases) lease.retire();
|
||||
let startupUnsettled = false;
|
||||
if (startup.executionContext) {
|
||||
@@ -9876,6 +9880,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
remote: input.runnerExecutionTarget?.kind === "remote",
|
||||
confirmedDeadline: confirmedLeaseDeadline,
|
||||
onLost: () => {
|
||||
startup.providerPackSourceAbort.abort(new Error("native_session_lease_lost"));
|
||||
startup.executionContext?.abort(new Error("native_session_lease_lost"));
|
||||
return input.runnerExecutionTarget?.kind === "remote"
|
||||
? detachNativeSessionsForRestart([input.execution.binding.runId], leaseOwner)
|
||||
@@ -12801,6 +12806,7 @@ export async function prepareRemoteProviderPackOnTarget(input: {
|
||||
runner: CommandManagedRuntimeRunner;
|
||||
configuredProviderPackRoot: string | null;
|
||||
manifest: RemoteProviderPackManifest;
|
||||
signal?: AbortSignal;
|
||||
trace?: NativeRunTrace;
|
||||
onLog?: (stream: "stdout" | "stderr", chunk: string) => void | Promise<void>;
|
||||
}): Promise<{ root: string; manifest: RemoteProviderPackManifest; source: "staged" | "preinstalled" | "uploaded" }> {
|
||||
@@ -12891,16 +12897,18 @@ export async function prepareRemoteProviderPackOnTarget(input: {
|
||||
root: configuredProviderPackRoot,
|
||||
sourceDigest: readRemoteProviderPackIdentity(configuredProviderPackRoot, false).digest,
|
||||
expectedDigest: expectedProviderPackManifest.digest,
|
||||
manifest: expectedProviderPackManifest,
|
||||
signal: input.signal,
|
||||
runner: remoteCommandRunner,
|
||||
cwd: remoteTarget.remoteCwd,
|
||||
targetPath: stagedRemoteProviderPackRoot,
|
||||
reuseRoots: preinstalledProviderPackRoots,
|
||||
trace: input.trace,
|
||||
verifyDirectory: () => { readRemoteProviderPackManifest(configuredProviderPackRoot); },
|
||||
stageDirectory: () => stageRemoteRunnerDirectory({
|
||||
verifyDirectory: (sourceRoot) => { readRemoteProviderPackManifest(sourceRoot); },
|
||||
stageDirectory: (sourceRoot) => stageRemoteRunnerDirectory({
|
||||
target: remoteTarget,
|
||||
runner: remoteCommandRunner,
|
||||
sourcePath: configuredProviderPackRoot,
|
||||
sourcePath: sourceRoot,
|
||||
targetPath: stagedRemoteProviderPackRoot,
|
||||
mode: 0o700,
|
||||
}),
|
||||
@@ -14895,9 +14903,14 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
expectedProviderPackManifest &&
|
||||
stagedRemoteProviderPackRoot
|
||||
) {
|
||||
const sourceRunId = sessionToolAuthorityEpochs.get(sessionScopeId)?.runId ?? input.execution.binding.runId;
|
||||
const startupSignal = nativeSessionStartups.get(sourceRunId)?.providerPackSourceAbort.signal;
|
||||
const contextSignal = input.executionContext?.active ? input.executionContext.signal : undefined;
|
||||
const sourceSignals = [startupSignal, contextSignal].filter((signal): signal is AbortSignal => !!signal);
|
||||
const prepared = await prepareRemoteProviderPackOnTarget({
|
||||
target: remoteTarget, runner: remoteCommandRunner, configuredProviderPackRoot,
|
||||
manifest: expectedProviderPackManifest, trace: input.trace, onLog: input.onLog,
|
||||
signal: sourceSignals.length ? AbortSignal.any(sourceSignals) : undefined,
|
||||
});
|
||||
activeRemoteProviderPackRoot = prepared.root;
|
||||
providerPackVerified = true;
|
||||
|
||||
@@ -66,6 +66,25 @@ it("keeps historical/source packs without an archive on the existing path", asyn
|
||||
expect(await readProviderPackArchive("/does-not-exist/provider-pack", digest)).toBeNull();
|
||||
});
|
||||
|
||||
it.each(["v1", "v2"])("cancels a real %s archive hash after its first read chunk", async version => {
|
||||
const { pack } = version === "v2" ? await layeredFixture() : await fixture();
|
||||
const archivePath = version === "v2" ? `${pack}.archives/overlay.tar.zst` : `${pack}.tar.zst`;
|
||||
const controller = new AbortController();
|
||||
const handle = await fs.open(archivePath, "r");
|
||||
const prototype = Object.getPrototypeOf(handle) as typeof handle;
|
||||
await handle.close();
|
||||
const originalStream = prototype.createReadStream;
|
||||
const streaming = vi.spyOn(prototype, "createReadStream").mockImplementation(function(this: typeof handle, options) {
|
||||
const stream = originalStream.call(this, options);
|
||||
stream.once("data", () => controller.abort(new Error("owned archive read stopped")));
|
||||
return stream;
|
||||
});
|
||||
try {
|
||||
await expect(readProviderPackArchive(pack, digest, controller.signal)).rejects.toThrow(/abort|owned archive read stopped/i);
|
||||
expect(controller.signal.aborted).toBe(true);
|
||||
} finally { streaming.mockRestore(); }
|
||||
});
|
||||
|
||||
async function layeredFixture() {
|
||||
const { root, pack } = await fixture();
|
||||
await fs.writeFile(path.join(pack, "bin", "large"), Buffer.alloc(1024 * 1024, 42), { mode: 0o755 });
|
||||
|
||||
@@ -7,16 +7,18 @@ import { providerPackExtractionProbe } from "./provider-pack-extraction.js";
|
||||
|
||||
const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
|
||||
|
||||
export async function readProviderPackArchive(root: string, expectedDigest: string) {
|
||||
export async function readProviderPackArchive(root: string, expectedDigest: string, signal?: AbortSignal) {
|
||||
signal?.throwIfAborted();
|
||||
let receipt: { schema?: string; encoding?: string; providerPackDigest?: string; sha256?: string; bytes?: number };
|
||||
try {
|
||||
receipt = JSON.parse(await readFile(`${root}.archive.json`, "utf8"));
|
||||
receipt = JSON.parse(await readFile(`${root}.archive.json`, { encoding: "utf8", signal }));
|
||||
} catch (error) {
|
||||
signal?.throwIfAborted();
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
|
||||
throw invalidArchive("receipt_unreadable", error);
|
||||
}
|
||||
if (receipt.schema === "paperclip-runner/provider-pack-archive/v2") {
|
||||
return readProviderPackLayers(root, expectedDigest, receipt);
|
||||
return readProviderPackLayers(root, expectedDigest, receipt, signal);
|
||||
}
|
||||
if (receipt.schema !== "paperclip-runner/provider-pack-archive/v1"
|
||||
|| receipt.encoding !== "tar+zstd"
|
||||
@@ -26,7 +28,7 @@ export async function readProviderPackArchive(root: string, expectedDigest: stri
|
||||
throw invalidArchive(receipt.providerPackDigest !== expectedDigest ? "receipt_identity_mismatch" : "receipt_invalid");
|
||||
}
|
||||
const sourcePath = `${root}.tar.zst`;
|
||||
await verifyArchive(sourcePath, { sha256: receipt.sha256!, bytes: receipt.bytes! });
|
||||
await verifyArchive(sourcePath, { sha256: receipt.sha256!, bytes: receipt.bytes! }, signal);
|
||||
return { sourcePath, sha256: receipt.sha256!, bytes: receipt.bytes! };
|
||||
}
|
||||
|
||||
|
||||
@@ -28,7 +28,8 @@ const validArchive = (value: unknown): value is Archive => {
|
||||
export const invalidArchive = (reason: string, cause?: unknown) =>
|
||||
new Error(`runner_remote_provider_archive_invalid: ${reason}`, cause === undefined ? undefined : { cause });
|
||||
|
||||
export async function verifyArchive(path: string, expected: Archive) {
|
||||
export async function verifyArchive(path: string, expected: Archive, signal?: AbortSignal) {
|
||||
signal?.throwIfAborted();
|
||||
let file;
|
||||
try { file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW); }
|
||||
catch (error) { throw invalidArchive("archive_unreadable", error); }
|
||||
@@ -37,12 +38,13 @@ export async function verifyArchive(path: string, expected: Archive) {
|
||||
if (!metadata.isFile()) throw invalidArchive("archive_not_file");
|
||||
if (metadata.size !== expected.bytes) throw invalidArchive("size_mismatch");
|
||||
const hash = createHash("sha256");
|
||||
for await (const chunk of file.createReadStream({ autoClose: false })) hash.update(chunk);
|
||||
for await (const chunk of file.createReadStream({ autoClose: false, signal })) hash.update(chunk);
|
||||
signal?.throwIfAborted();
|
||||
if (hash.digest("hex") !== expected.sha256) throw invalidArchive("sha256_mismatch");
|
||||
} finally { await file.close(); }
|
||||
}
|
||||
|
||||
export async function readProviderPackLayers(root: string, expectedDigest: string, value: unknown): Promise<ProviderPackLayers> {
|
||||
export async function readProviderPackLayers(root: string, expectedDigest: string, value: unknown, signal?: AbortSignal): Promise<ProviderPackLayers> {
|
||||
const receipt = value as { encoding: string; providerPackDigest: string; overlay: Archive; files: FileLayer[] };
|
||||
if (receipt.providerPackDigest !== expectedDigest) throw invalidArchive("receipt_identity_mismatch");
|
||||
if (receipt.encoding !== "tar+zstd"
|
||||
@@ -70,7 +72,7 @@ export async function readProviderPackLayers(root: string, expectedDigest: strin
|
||||
const sourceRoot = `${root}.archives`;
|
||||
const layers = await lstat(sourceRoot).catch(error => { throw invalidArchive("layers_unreadable", error); });
|
||||
if (!layers.isDirectory()) throw invalidArchive("layers_unreadable");
|
||||
await verifyArchive(`${sourceRoot}/overlay.tar.zst`, receipt.overlay);
|
||||
await verifyArchive(`${sourceRoot}/overlay.tar.zst`, receipt.overlay, signal);
|
||||
return { kind: "layers", sourceRoot, overlay: receipt.overlay, files: receipt.files };
|
||||
}
|
||||
|
||||
@@ -110,6 +112,11 @@ const reuseScript = `
|
||||
})().catch(()=>process.exitCode=1);
|
||||
`;
|
||||
|
||||
export function providerPackLayersArgument(archive: ProviderPackLayers, targetPath: string, reuseRoots: readonly string[]) {
|
||||
const spec = Buffer.from(JSON.stringify({ target: targetPath, roots: reuseRoots, files: archive.files })).toString("base64");
|
||||
return Buffer.byteLength(spec) > 120_000 ? null : spec;
|
||||
}
|
||||
|
||||
export async function stageProviderPackLayers(input: {
|
||||
archive: ProviderPackLayers;
|
||||
runner: CommandManagedRuntimeRunner;
|
||||
@@ -120,9 +127,9 @@ export async function stageProviderPackLayers(input: {
|
||||
const { archive, runner } = input;
|
||||
if (!runner.syncIn) throw Error("runner_remote_provider_archive_transport_unavailable");
|
||||
const measure = <T>(name: string, fn: () => Promise<T>) => input.trace ? input.trace.measure(name, fn) : fn();
|
||||
const spec = Buffer.from(JSON.stringify({ target: input.targetPath, roots: input.reuseRoots, files: archive.files })).toString("base64");
|
||||
const spec = providerPackLayersArgument(archive, input.targetPath, input.reuseRoots);
|
||||
// Respect the per-argument Linux limit; unusual packs retain directory staging.
|
||||
if (Buffer.byteLength(spec) > 120_000) return false;
|
||||
if (spec === null) return false;
|
||||
const capability = await measure("provider_pack.archive.probe", () => runner.execute({
|
||||
command: "sh", args: ["-c", `${providerPackExtractionProbe} && node -e 'process.exit(Number(process.versions.node.split(".")[0])>=18?0:1)'`],
|
||||
bypassSession: true, timeoutMs: 10_000,
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import type { NativeRunTrace } from "./native-run-trace.js";
|
||||
import { readProviderPackArchive, stageProviderPackArchive } from "./provider-pack-archive.js";
|
||||
import type { ProviderPackIdentity } from "./remote-provider-pack-verification.js";
|
||||
import { resolveProviderPackUploadSource } from "./retained-provider-pack-source.js";
|
||||
import { providerPackExtractionProbe } from "./provider-pack-extraction.js";
|
||||
import { providerPackLayersArgument } from "./provider-pack-layers.js";
|
||||
|
||||
const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
|
||||
|
||||
@@ -11,30 +15,46 @@ export async function uploadVerifiedProviderPack(input: {
|
||||
sourceDigest: string;
|
||||
/** Identity selected for this execution; a retained session can pin an older release. */
|
||||
expectedDigest: string;
|
||||
manifest?: ProviderPackIdentity;
|
||||
signal?: AbortSignal;
|
||||
runner: CommandManagedRuntimeRunner;
|
||||
cwd: string;
|
||||
targetPath: string;
|
||||
reuseRoots: readonly string[];
|
||||
trace?: NativeRunTrace;
|
||||
verifyDirectory: () => void;
|
||||
stageDirectory: () => Promise<void>;
|
||||
verifyDirectory: (root: string) => void | Promise<void>;
|
||||
stageDirectory: (root: string) => Promise<void>;
|
||||
verifyUploaded: () => Promise<void>;
|
||||
}): Promise<void> {
|
||||
// The controller can only stage the pack it ships. A retained continuation
|
||||
// pinned to an earlier release's pack cannot be satisfied by these bytes, by
|
||||
// either transport. Fail before clearing the remote pack or uploading.
|
||||
if (input.sourceDigest !== input.expectedDigest) {
|
||||
if (input.sourceDigest !== input.expectedDigest && !input.manifest) {
|
||||
const short = (digest: string) => digest.replace(/^sha256:/, "").slice(0, 12);
|
||||
throw new Error(`runner_remote_provider_artifact_incompatible: selected provider pack ${short(input.expectedDigest)} is not the controller's pack ${short(input.sourceDigest)}; install the matching Paperclip package and Daytona image`);
|
||||
}
|
||||
const measure = <T>(name: string, fn: () => Promise<T>) => input.trace ? input.trace.measure(name, fn) : fn();
|
||||
const verifyDirectory = () => measure("provider_pack.directory.verify", async () => input.verifyDirectory());
|
||||
input.signal?.throwIfAborted();
|
||||
if (input.manifest && input.manifest.digest !== input.expectedDigest) throw Error("runner_remote_provider_source_invalid");
|
||||
const source = input.manifest ? await measure("provider_pack.source", () => resolveProviderPackUploadSource({
|
||||
root: input.root, sourceDigest: input.sourceDigest, manifest: input.manifest!, signal: input.signal,
|
||||
})) : { root: input.root, archiveOnly: false };
|
||||
const verifyDirectory = () => measure("provider_pack.directory.verify", async () => input.verifyDirectory(source.root));
|
||||
// Archive installation reads the sealed archives, not the expanded tree.
|
||||
// Hash the directory only when it is actually the upload source. The
|
||||
// installed pack still passes full verification against the selected identity.
|
||||
const archive = await measure("provider_pack.archive.verify", () =>
|
||||
readProviderPackArchive(input.root, input.expectedDigest));
|
||||
readProviderPackArchive(source.root, input.expectedDigest, input.signal));
|
||||
if (source.archiveOnly) {
|
||||
if (!archive) throw Error("runner_remote_provider_source_invalid");
|
||||
if ("kind" in archive && providerPackLayersArgument(archive, input.targetPath, input.reuseRoots) === null) {
|
||||
throw Error("runner_remote_provider_archive_transport_unavailable");
|
||||
}
|
||||
const capability = await measure("provider_pack.archive.probe", () => input.runner.execute({
|
||||
command: "sh", args: ["-c", `${providerPackExtractionProbe}${"kind" in archive ? " && node -e 'process.exit(Number(process.versions.node.split(\".\")[0])>=18?0:1)'" : ""}`],
|
||||
bypassSession: true, timeoutMs: 10_000,
|
||||
}));
|
||||
if (capability.exitCode !== 0 || capability.timedOut) throw Error("runner_remote_provider_archive_transport_unavailable");
|
||||
}
|
||||
if (!archive) await verifyDirectory();
|
||||
input.signal?.throwIfAborted();
|
||||
const cleared = await measure("provider_pack.clear_stale", () => input.runner.execute({
|
||||
command: "sh", args: ["-c", `rm -rf ${quote(input.targetPath)}`],
|
||||
cwd: input.cwd, bypassSession: true, timeoutMs: 10_000,
|
||||
@@ -47,8 +67,9 @@ export async function uploadVerifiedProviderPack(input: {
|
||||
archive, runner: input.runner, targetPath: input.targetPath,
|
||||
reuseRoots: input.reuseRoots, trace: input.trace,
|
||||
})) return;
|
||||
if (source.archiveOnly) throw Error("runner_remote_provider_archive_transport_unavailable");
|
||||
if (archive) await verifyDirectory();
|
||||
await input.stageDirectory();
|
||||
await input.stageDirectory(source.root);
|
||||
});
|
||||
await measure("provider_pack.verify_uploaded", input.verifyUploaded);
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { createReadStream, createWriteStream } from "node:fs";
|
||||
import { mkdir } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { Readable, Transform } from "node:stream";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import { createGunzip } from "node:zlib";
|
||||
import tar from "tar-stream";
|
||||
import type { ProviderPackIdentity } from "./remote-provider-pack-verification.js";
|
||||
|
||||
const repository = "paperclipai/paperclip";
|
||||
const registry = "https://ghcr.io";
|
||||
export const MAX_RETAINED_PACK_BYTES = 4 * 1024 ** 3;
|
||||
const maxMetadataBytes = 1024 * 1024;
|
||||
const digest = (bytes: Buffer) => `sha256:${createHash("sha256").update(bytes).digest("hex")}`;
|
||||
const validDigest = (value: unknown): value is string => typeof value === "string" && /^sha256:[a-f0-9]{64}$/.test(value);
|
||||
function fail(): never { throw Error("runner_remote_provider_source_invalid"); }
|
||||
type Blob = { digest: string; size: number; mediaType?: string };
|
||||
type Manifest = { schemaVersion: number; config: Blob; layers: Blob[] };
|
||||
export type RetainedPackProvenance = {
|
||||
schema: "paperclip.retained-provider-pack-source.v1";
|
||||
sourceRevision: string;
|
||||
providerPackDigest: string;
|
||||
imageIndexDigest: string;
|
||||
imageManifestDigest: string;
|
||||
imageConfigDigest: string;
|
||||
transportLayerDigest: string;
|
||||
transportLayerBytes: number;
|
||||
};
|
||||
|
||||
function blob(value: Blob, maximum: number): Blob {
|
||||
if (!value || !validDigest(value.digest) || !Number.isSafeInteger(value.size) || value.size < 1 || value.size > maximum) fail();
|
||||
return value;
|
||||
}
|
||||
|
||||
async function metadata(response: Response, signal: AbortSignal, expected?: Blob): Promise<Buffer> {
|
||||
if (!response.body) fail();
|
||||
const chunks: Buffer[] = [];
|
||||
let bytes = 0;
|
||||
try {
|
||||
for await (const chunk of Readable.fromWeb(response.body as Parameters<typeof Readable.fromWeb>[0])) {
|
||||
signal.throwIfAborted();
|
||||
bytes += chunk.length;
|
||||
if (bytes > maxMetadataBytes) fail();
|
||||
chunks.push(Buffer.from(chunk));
|
||||
}
|
||||
} catch (error) {
|
||||
await response.body.cancel().catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
const result = Buffer.concat(chunks);
|
||||
if (expected && (result.length !== expected.size || digest(result) !== expected.digest)) fail();
|
||||
return result;
|
||||
}
|
||||
|
||||
function bounded(limit: number) {
|
||||
let bytes = 0;
|
||||
return new Transform({ transform(chunk, _encoding, done) {
|
||||
bytes += chunk.length;
|
||||
done(bytes > limit ? Error("runner_remote_provider_source_size_limit") : null, chunk);
|
||||
} });
|
||||
}
|
||||
|
||||
async function unpackTransport(layer: string, target: string, signal: AbortSignal) {
|
||||
const archive = tar.extract();
|
||||
const names = new Set<string>();
|
||||
let bytes = 0;
|
||||
archive.on("entry", (header, body, next) => {
|
||||
body.on("error", () => undefined);
|
||||
void (async () => {
|
||||
signal.throwIfAborted();
|
||||
const normalized = header.name.replace(/^\.\//u, "");
|
||||
const name = header.type === "directory" ? normalized.replace(/\/$/u, "") : normalized;
|
||||
if (names.has(name) || names.size >= 10_010) fail();
|
||||
names.add(name);
|
||||
if (header.type === "directory") {
|
||||
if (!["opt", "opt/paperclip-runner", "opt/paperclip-runner/provider-pack.archives"].includes(name)) fail();
|
||||
body.resume();
|
||||
return;
|
||||
}
|
||||
if (header.type !== "file" || !/^opt\/paperclip-runner\/provider-pack(?:\.archive\.json|\.archives\/(?:overlay|[0-9]{1,4})\.tar\.zst)$/u.test(name)
|
||||
|| !Number.isSafeInteger(header.size) || header.size! < 1 || (bytes += header.size!) > MAX_RETAINED_PACK_BYTES
|
||||
|| name.endsWith(".json") && header.size! > maxMetadataBytes) fail();
|
||||
const relative = name.slice("opt/paperclip-runner/".length);
|
||||
if (relative.includes("/")) await mkdir(join(target, "provider-pack.archives"), { recursive: true, mode: 0o700 });
|
||||
await pipeline(body, createWriteStream(join(target, relative), { flags: "wx", mode: 0o600 }), { signal });
|
||||
})().then(() => next(), error => archive.destroy(error));
|
||||
});
|
||||
await pipeline(createReadStream(layer), createGunzip(), bounded(MAX_RETAINED_PACK_BYTES + 16 * 1024 ** 2), archive, { signal });
|
||||
}
|
||||
|
||||
export async function receiveRetainedProviderPack(input: {
|
||||
manifest: ProviderPackIdentity;
|
||||
directory: string;
|
||||
signal: AbortSignal;
|
||||
}): Promise<RetainedPackProvenance> {
|
||||
const { manifest: selected, directory, signal } = input;
|
||||
const revision = selected.payload.runnerSourceRevision;
|
||||
const architecture = selected.payload.target.architecture === "x64" ? "amd64" : selected.payload.target.architecture === "arm64" ? "arm64" : null;
|
||||
if (!/^[a-f0-9]{40}$/.test(revision) || selected.payload.target.platform !== "linux" || !architecture) fail();
|
||||
const request = async (path: string, token?: string, allowRedirect = false) => {
|
||||
signal.throwIfAborted();
|
||||
let response = await fetch(`${registry}${path}`, { redirect: "manual", signal, headers: {
|
||||
Accept: "application/vnd.oci.image.index.v1+json, application/vnd.oci.image.manifest.v1+json, application/vnd.docker.distribution.manifest.list.v2+json, application/vnd.docker.distribution.manifest.v2+json",
|
||||
...(token ? { Authorization: `Bearer ${token}` } : {}),
|
||||
} });
|
||||
if ([302, 307].includes(response.status) && allowRedirect) {
|
||||
const location = response.headers.get("location");
|
||||
await response.body?.cancel();
|
||||
if (!location || location.length > 8192) fail();
|
||||
const url = new URL(location, registry);
|
||||
if (url.protocol !== "https:" || url.username || url.password || !["ghcr.io", "pkg-containers.githubusercontent.com"].includes(url.hostname)) fail();
|
||||
response = await fetch(url, { redirect: "error", signal });
|
||||
}
|
||||
if (!response.ok) {
|
||||
await response.body?.cancel();
|
||||
throw Error(`runner_remote_provider_source_unavailable: HTTP ${response.status}`);
|
||||
}
|
||||
return response;
|
||||
};
|
||||
const authentication = JSON.parse((await metadata(await request(`/token?service=ghcr.io&scope=repository:${repository}:pull`), signal)).toString()) as { token?: unknown };
|
||||
if (typeof authentication.token !== "string" || authentication.token.length < 1 || authentication.token.length > 32 * 1024) fail();
|
||||
const token = authentication.token;
|
||||
const indexBytes = await metadata(await request(`/v2/${repository}/manifests/sha-${revision}`, token), signal);
|
||||
const index = JSON.parse(indexBytes.toString()) as { schemaVersion: number; manifests: Array<Blob & { platform?: { os?: string; architecture?: string } }> };
|
||||
if (index.schemaVersion !== 2 || !Array.isArray(index.manifests) || index.manifests.length > 32) fail();
|
||||
const candidates = index.manifests.filter(row => row.platform?.os === "linux" && row.platform.architecture === architecture);
|
||||
if (candidates.length !== 1) fail();
|
||||
const child = blob(candidates[0]!, maxMetadataBytes);
|
||||
const manifestBytes = await metadata(await request(`/v2/${repository}/manifests/${child.digest}`, token), signal, child);
|
||||
const image = JSON.parse(manifestBytes.toString()) as Manifest;
|
||||
if (image.schemaVersion !== 2 || !Array.isArray(image.layers) || image.layers.length > 128) fail();
|
||||
const configBlob = blob(image.config, maxMetadataBytes);
|
||||
const configBytes = await metadata(await request(`/v2/${repository}/blobs/${configBlob.digest}`, token, true), signal, configBlob);
|
||||
const config = JSON.parse(configBytes.toString()) as { architecture: string; os: string; config?: { Labels?: Record<string, string> }; history: Array<{ empty_layer?: boolean; created_by?: string }> };
|
||||
if (config.architecture !== architecture || config.os !== "linux" || config.config?.Labels?.["org.opencontainers.image.revision"] !== revision
|
||||
|| config.config.Labels["org.opencontainers.image.source"] !== "https://github.com/paperclipai/paperclip"
|
||||
|| !Array.isArray(config.history) || config.history.length > 1024) fail();
|
||||
const history = config.history.filter(row => !row.empty_layer);
|
||||
if (history.length !== image.layers.length) fail();
|
||||
const indices = history.flatMap((row, index) => /^COPY \/provider-pack-transport\/? \/opt\/paperclip-runner\/? # buildkit$/u.test(row.created_by ?? "") ? [index] : []);
|
||||
if (indices.length !== 1) fail();
|
||||
const transport = blob(image.layers[indices[0]!]!, MAX_RETAINED_PACK_BYTES);
|
||||
if (!["application/vnd.oci.image.layer.v1.tar+gzip", "application/vnd.docker.image.rootfs.diff.tar.gzip"].includes(transport.mediaType ?? "")) fail();
|
||||
const response = await request(`/v2/${repository}/blobs/${transport.digest}`, token, true);
|
||||
if (!response.body) fail();
|
||||
let bytes = 0;
|
||||
const hash = createHash("sha256");
|
||||
const meter = new Transform({ transform(chunk, _encoding, done) {
|
||||
bytes += chunk.length;
|
||||
hash.update(chunk);
|
||||
done(bytes > transport.size ? Error("runner_remote_provider_source_size_limit") : null, chunk);
|
||||
} });
|
||||
const layer = join(directory, "transport.tar.gz");
|
||||
await pipeline(Readable.fromWeb(response.body as Parameters<typeof Readable.fromWeb>[0]), meter, createWriteStream(layer, { flags: "wx", mode: 0o600 }), { signal });
|
||||
if (bytes !== transport.size || `sha256:${hash.digest("hex")}` !== transport.digest) fail();
|
||||
await unpackTransport(layer, directory, signal);
|
||||
return { schema: "paperclip.retained-provider-pack-source.v1", sourceRevision: revision, providerPackDigest: selected.digest,
|
||||
imageIndexDigest: digest(indexBytes), imageManifestDigest: child.digest, imageConfigDigest: configBlob.digest,
|
||||
transportLayerDigest: transport.digest, transportLayerBytes: transport.size };
|
||||
}
|
||||
@@ -0,0 +1,380 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import { createHash } from "node:crypto";
|
||||
import { createServer } from "node:http";
|
||||
import { promises as fs } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { gzipSync, zstdCompressSync } from "node:zlib";
|
||||
import tar from "tar-stream";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
|
||||
import { uploadVerifiedProviderPack } from "./provider-pack-upload.js";
|
||||
import * as archiveOwner from "./provider-pack-archive.js";
|
||||
import { resolveProviderPackUploadSource } from "./retained-provider-pack-source.js";
|
||||
const archiveScript = pathToFileURL(path.resolve(import.meta.dirname, "../../../../scripts/create-provider-pack-archive.mjs")).href;
|
||||
const { createProviderPackLayers } = await import(archiveScript);
|
||||
|
||||
const exec = promisify(execFile);
|
||||
const hash = (value: string | Buffer) => createHash("sha256").update(value).digest("hex");
|
||||
const roots: string[] = [];
|
||||
const servers: ReturnType<typeof createServer>[] = [];
|
||||
const actualFetch = globalThis.fetch;
|
||||
|
||||
afterEach(async () => {
|
||||
vi.unstubAllGlobals();
|
||||
vi.unstubAllEnvs();
|
||||
vi.restoreAllMocks();
|
||||
await Promise.all(servers.splice(0).map(server => new Promise<void>((done, reject) => {
|
||||
server.close(error => error ? reject(error) : done());
|
||||
server.closeAllConnections();
|
||||
})));
|
||||
await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
async function fixture(options: { wrongCode?: boolean; unsafeOverlay?: boolean; changedFileLayer?: boolean; hardlink?: "safe" | "escape" | "unknown" } = {}) {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "retained-pack-test-"));
|
||||
roots.push(root);
|
||||
const temporary = path.join(root, "tmp");
|
||||
await fs.mkdir(temporary);
|
||||
vi.stubEnv("TMPDIR", temporary);
|
||||
const makePack = async (name: string, tool: string, source: string) => {
|
||||
const pack = path.join(root, name);
|
||||
await fs.mkdir(path.join(pack, "bin"), { recursive: true });
|
||||
const artifact = { path: "bin/tool", sha256: `sha256:${hash(tool)}` };
|
||||
const payload = { pins: { opencode: "1.18.34" }, target: { platform: "linux", architecture: "x64" }, runnerSourceRevision: source,
|
||||
contractDigest: `sha256:${hash(tool)}`, artifacts: { nodeCommand: artifact, opencodeCommand: artifact, opencodeProxy: artifact } };
|
||||
const manifest = { schema: "paperclip-runner/remote-provider-pack/v1", payload, digest: `sha256:${hash(canonicalProviderPackJson(payload))}` };
|
||||
await fs.writeFile(path.join(pack, "provider-pack.json"), JSON.stringify(manifest));
|
||||
await fs.writeFile(path.join(pack, "bin/tool"), tool, { mode: 0o755 });
|
||||
await fs.writeFile(path.join(pack, "bin/large"), Buffer.alloc(1024 * 1024, 42), { mode: 0o755 });
|
||||
await createProviderPackLayers(pack);
|
||||
return { pack, manifest };
|
||||
};
|
||||
const selected = await makePack("old", "retained provider code", "a".repeat(40));
|
||||
const current = await makePack("current", "current incompatible code", "b".repeat(40));
|
||||
if (options.changedFileLayer) {
|
||||
const unsafe = tar.pack(), chunks: Buffer[] = [];
|
||||
const collected = (async () => { for await (const chunk of unsafe) chunks.push(Buffer.from(chunk)); })();
|
||||
const body = Buffer.alloc(1024 * 1024, 43);
|
||||
await new Promise<void>((done, reject) => unsafe.entry({ name: "bin/large", type: "file", size: body.length, mode: 0o755 }, body, error => error ? reject(error) : done()));
|
||||
unsafe.finalize(); await collected;
|
||||
const bytes = zstdCompressSync(Buffer.concat(chunks));
|
||||
await fs.writeFile(selected.pack + ".archives/0.tar.zst", bytes);
|
||||
const receipt = JSON.parse(await fs.readFile(selected.pack + ".archive.json", "utf8"));
|
||||
receipt.files[0].archive = { bytes: bytes.length, sha256: hash(bytes) };
|
||||
await fs.writeFile(selected.pack + ".archive.json", JSON.stringify(receipt));
|
||||
}
|
||||
if (options.unsafeOverlay || options.hardlink) {
|
||||
const unsafe = tar.pack(), chunks: Buffer[] = [];
|
||||
const collected = (async () => { for await (const chunk of unsafe) chunks.push(Buffer.from(chunk)); })();
|
||||
for (const [name, body] of [["provider-pack.json", JSON.stringify(selected.manifest)], ["bin/tool", "retained provider code"], ...(options.unsafeOverlay ? [["../outside", "must not be written"]] : [])]) {
|
||||
await new Promise<void>((done, reject) => unsafe.entry({ name: name!, type: "file", size: Buffer.byteLength(body!) }, body!, error => error ? reject(error) : done()));
|
||||
}
|
||||
if (options.hardlink) await new Promise<void>((done, reject) => unsafe.entry({ name: "bin/tool-copy", type: "link", linkname: options.hardlink === "safe" ? "./bin/tool" : options.hardlink === "escape" ? "../outside" : "bin/missing", mode: 0o644 }, error => error ? reject(error) : done()));
|
||||
unsafe.finalize(); await collected;
|
||||
const bytes = zstdCompressSync(Buffer.concat(chunks));
|
||||
await fs.writeFile(selected.pack + ".archives/overlay.tar.zst", bytes);
|
||||
const receipt = JSON.parse(await fs.readFile(selected.pack + ".archive.json", "utf8"));
|
||||
receipt.overlay = { bytes: bytes.length, sha256: hash(bytes) };
|
||||
await fs.writeFile(selected.pack + ".archive.json", JSON.stringify(receipt));
|
||||
}
|
||||
const transportPack = options.wrongCode ? current.pack : selected.pack;
|
||||
const archive = tar.pack();
|
||||
const chunks: Buffer[] = [];
|
||||
const collected = (async () => { for await (const chunk of archive) chunks.push(Buffer.from(chunk)); })();
|
||||
for (const name of ["provider-pack.archive.json", ...await fs.readdir(transportPack + ".archives").then(names => names.map(name => `provider-pack.archives/${name}`))]) {
|
||||
const bytes = await fs.readFile(path.join(path.dirname(transportPack), name.replace("provider-pack", path.basename(transportPack))));
|
||||
await new Promise<void>((done, reject) => archive.entry({ name: `opt/paperclip-runner/${name}`, type: "file", size: bytes.length }, bytes, error => error ? reject(error) : done()));
|
||||
}
|
||||
archive.finalize();
|
||||
await collected;
|
||||
const layer = gzipSync(Buffer.concat(chunks));
|
||||
const config = Buffer.from(JSON.stringify({ architecture: "amd64", os: "linux", config: { Labels: {
|
||||
"org.opencontainers.image.revision": selected.manifest.payload.runnerSourceRevision,
|
||||
"org.opencontainers.image.source": "https://github.com/paperclipai/paperclip",
|
||||
} }, history: [{ created_by: "COPY /provider-pack-transport/ /opt/paperclip-runner/ # buildkit" }] }));
|
||||
const manifest = Buffer.from(JSON.stringify({ schemaVersion: 2, mediaType: "application/vnd.oci.image.manifest.v1+json", config: { digest: `sha256:${hash(config)}`, size: config.length }, layers: [{ digest: `sha256:${hash(layer)}`, size: layer.length, mediaType: "application/vnd.oci.image.layer.v1.tar+gzip" }] }));
|
||||
const index = Buffer.from(JSON.stringify({ schemaVersion: 2, mediaType: "application/vnd.oci.image.index.v1+json", manifests: [{ digest: `sha256:${hash(manifest)}`, size: manifest.length, platform: { os: "linux", architecture: "amd64" } }] }));
|
||||
const routes = new Map<string, Buffer>([
|
||||
["/token", Buffer.from(JSON.stringify({ token: "public-code-fixture-token" }))],
|
||||
[`/v2/paperclipai/paperclip/manifests/sha-${selected.manifest.payload.runnerSourceRevision}`, index],
|
||||
[`/v2/paperclipai/paperclip/manifests/sha256:${hash(manifest)}`, manifest],
|
||||
[`/v2/paperclipai/paperclip/blobs/sha256:${hash(config)}`, config],
|
||||
[`/v2/paperclipai/paperclip/blobs/sha256:${hash(layer)}`, layer],
|
||||
]);
|
||||
const requests: string[] = [];
|
||||
const stalled = new Set<string>();
|
||||
const observers = new Map<string, () => void>();
|
||||
const server = createServer((request, response) => {
|
||||
const route = new URL(request.url!, "http://fixture").pathname;
|
||||
requests.push(route);
|
||||
const bytes = routes.get(route);
|
||||
response.writeHead(bytes ? 200 : 404, { "content-length": bytes?.length ?? 0 });
|
||||
response.flushHeaders();
|
||||
observers.get(route)?.();
|
||||
if (stalled.has(route)) return;
|
||||
response.end(bytes);
|
||||
});
|
||||
servers.push(server);
|
||||
await new Promise<void>(done => server.listen(0, "127.0.0.1", done));
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") throw Error("fixture server unavailable");
|
||||
vi.stubGlobal("fetch", (input: string | URL | Request, init?: RequestInit) => {
|
||||
const url = new URL(typeof input === "string" || input instanceof URL ? input : input.url);
|
||||
expect(url.hostname).toBe("ghcr.io");
|
||||
return actualFetch(`http://127.0.0.1:${address.port}${url.pathname}${url.search}`, init);
|
||||
});
|
||||
const bin = path.join(root, "commands");
|
||||
await fs.mkdir(bin);
|
||||
const tarCommand = (await exec("sh", ["-c", "command -v gtar || command -v tar"])).stdout.trim();
|
||||
await fs.symlink(tarCommand, path.join(bin, "tar"));
|
||||
const execute = vi.fn(async (input: { command: string; args?: string[] }) => {
|
||||
try {
|
||||
const result = await exec(input.command, input.args ?? [], { env: { ...process.env, PATH: `${bin}:${process.env.PATH}` } });
|
||||
return { exitCode: 0, timedOut: false, stdout: result.stdout, stderr: result.stderr };
|
||||
} catch (error) { return { exitCode: 1, timedOut: false, stdout: "", stderr: String(error) }; }
|
||||
});
|
||||
const syncIn = vi.fn(async (operations: Array<{ files: Array<{ sourcePath: string; targetPath: string }> }>) => {
|
||||
for (const operation of operations) for (const file of operation.files) {
|
||||
await fs.mkdir(path.dirname(file.targetPath), { recursive: true });
|
||||
await fs.copyFile(file.sourcePath, file.targetPath);
|
||||
}
|
||||
return { operations: [] };
|
||||
});
|
||||
return { root, selected, current, requests, routes, runner: { execute, syncIn }, targetPath: path.join(root, "target"),
|
||||
temporary, layerRoute: `/v2/paperclipai/paperclip/blobs/sha256:${hash(layer)}`,
|
||||
stall: (route: string) => { stalled.add(route); },
|
||||
observed: (route: string) => new Promise<void>(done => { if (requests.includes(route)) done(); else observers.set(route, done); }),
|
||||
};
|
||||
}
|
||||
|
||||
function upload(input: Awaited<ReturnType<typeof fixture>>, extra: Partial<Parameters<typeof uploadVerifiedProviderPack>[0]> = {}) {
|
||||
return uploadVerifiedProviderPack({ root: input.current.pack, sourceDigest: input.current.manifest.digest, expectedDigest: input.selected.manifest.digest,
|
||||
manifest: input.selected.manifest, runner: input.runner, cwd: input.root, targetPath: input.targetPath,
|
||||
reuseRoots: [], verifyDirectory: () => { throw Error("current_directory_fallback_forbidden"); },
|
||||
stageDirectory: async () => { throw Error("current_directory_fallback_forbidden"); }, verifyUploaded: async () => {}, ...extra } as Parameters<typeof uploadVerifiedProviderPack>[0]);
|
||||
}
|
||||
|
||||
it("installs the exact retained pack through the real uploader on an empty host instead of the current image archive", async () => {
|
||||
const input = await fixture();
|
||||
const verifyUploaded = vi.fn(async () => {
|
||||
expect(await fs.readFile(path.join(input.targetPath, "bin/tool"), "utf8")).toBe("retained provider code");
|
||||
expect(JSON.parse(await fs.readFile(path.join(input.targetPath, "provider-pack.json"), "utf8"))).toEqual(input.selected.manifest);
|
||||
});
|
||||
await upload(input, { verifyUploaded });
|
||||
expect(verifyUploaded).toHaveBeenCalledOnce();
|
||||
expect(input.requests.some(route => route.includes("/blobs/"))).toBe(true);
|
||||
});
|
||||
|
||||
it("keeps a matching current pack on its existing path without registry or cache reads", async () => {
|
||||
const input = await fixture();
|
||||
await upload(input, { root: input.selected.pack, sourceDigest: input.selected.manifest.digest });
|
||||
expect(input.requests).toEqual([]);
|
||||
await expect(fs.stat(path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(await fs.readFile(path.join(input.targetPath, "bin/tool"), "utf8")).toBe("retained provider code");
|
||||
});
|
||||
|
||||
it("refuses a different historical code pack even when OCI source labels and all transport hashes are valid", async () => {
|
||||
const input = await fixture({ wrongCode: true });
|
||||
await expect(upload(input)).rejects.toThrow("runner_remote_provider_archive_invalid");
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
expect(await fs.readFile(path.join(input.current.pack, "bin/tool"), "utf8")).toBe("current incompatible code");
|
||||
});
|
||||
|
||||
it("does not mask a corrupt matching current archive by looking up a different code source", async () => {
|
||||
const input = await fixture();
|
||||
await fs.writeFile(input.selected.pack + ".archive.json", "{");
|
||||
await expect(upload(input, { root: input.selected.pack, sourceDigest: input.selected.manifest.digest })).rejects.toThrow("runner_remote_provider_archive_invalid");
|
||||
expect(input.requests).toEqual([]);
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects a changed controller code identity before looking up a historical source", async () => {
|
||||
const input = await fixture();
|
||||
await expect(upload(input, { sourceDigest: input.selected.manifest.digest })).rejects.toThrow("runner_remote_provider_source_invalid");
|
||||
expect(input.requests).toEqual([]);
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("refuses an unknown historical source without clearing or falling back to the current image", async () => {
|
||||
const input = await fixture();
|
||||
input.routes.delete(`/v2/paperclipai/paperclip/manifests/sha-${input.selected.manifest.payload.runnerSourceRevision}`);
|
||||
await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_unavailable: HTTP 404");
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects escaping inner tar paths despite valid OCI and layer digests before any remote effects", async () => {
|
||||
const input = await fixture({ unsafeOverlay: true });
|
||||
await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_invalid");
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
await expect(fs.stat(path.join(input.temporary, "outside"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("checks the file layer's actual plaintext against its declared code digest before any remote effects", async () => {
|
||||
const input = await fixture({ changedFileLayer: true });
|
||||
await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_invalid");
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("preserves historical code hardlinks only to earlier confined regular files", async () => {
|
||||
const input = await fixture({ hardlink: "safe" });
|
||||
await upload(input);
|
||||
const original = await fs.stat(path.join(input.targetPath, "bin/tool")), copy = await fs.stat(path.join(input.targetPath, "bin/tool-copy"));
|
||||
expect(copy.ino).toBe(original.ino);
|
||||
expect(await fs.readFile(path.join(input.targetPath, "bin/tool-copy"), "utf8")).toBe("retained provider code");
|
||||
});
|
||||
|
||||
it.each(["escape", "unknown"] as const)("refuses a historical code hardlink with an %s target before remote effects", async hardlink => {
|
||||
const input = await fixture({ hardlink });
|
||||
await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_invalid");
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("cancels a stalled source body and removes only the unfinished private import", async () => {
|
||||
const input = await fixture();
|
||||
input.stall(input.layerRoute);
|
||||
const controller = new AbortController();
|
||||
const operation = upload(input, { signal: controller.signal });
|
||||
const rejected = expect(operation).rejects.toThrow("owned caller stopped");
|
||||
await input.observed(input.layerRoute);
|
||||
controller.abort(new Error("owned caller stopped"));
|
||||
await rejected;
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
const cache = path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`);
|
||||
await vi.waitFor(async () => expect(await fs.readdir(cache)).toEqual([]));
|
||||
});
|
||||
|
||||
it("enforces the source deadline while the registry is stalled", async () => {
|
||||
const input = await fixture();
|
||||
input.stall(input.layerRoute);
|
||||
const nativeTimeout = AbortSignal.timeout;
|
||||
vi.spyOn(AbortSignal, "timeout").mockImplementation(milliseconds => {
|
||||
expect(milliseconds).toBe(180_000);
|
||||
return nativeTimeout(80);
|
||||
});
|
||||
await expect(upload(input)).rejects.toMatchObject({ name: "TimeoutError" });
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("reuses a verified derived cache without registry reads and reconstructs it after an empty-host reset", async () => {
|
||||
const input = await fixture();
|
||||
await upload(input);
|
||||
const first = input.requests.length;
|
||||
await upload(input, { targetPath: input.targetPath + "-cached" });
|
||||
expect(input.requests).toHaveLength(first);
|
||||
await fs.rm(path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`), { recursive: true });
|
||||
await upload(input, { targetPath: input.targetPath + "-empty" });
|
||||
expect(input.requests.length).toBeGreaterThan(first);
|
||||
expect(await fs.readFile(path.join(input.targetPath + "-empty", "bin/tool"), "utf8")).toBe("retained provider code");
|
||||
});
|
||||
|
||||
it("shares warm-cache verification while a stopped caller leaves another waiter active", async () => {
|
||||
const input = await fixture();
|
||||
await upload(input);
|
||||
const requests = input.requests.length;
|
||||
const firstController = new AbortController(), secondController = new AbortController();
|
||||
let enter!: () => void, release!: () => void, joined!: () => void, duplicate!: () => void;
|
||||
const entered = new Promise<void>(done => { enter = done; });
|
||||
const gate = new Promise<void>(done => { release = done; });
|
||||
const secondJoined = new Promise<void>(done => { joined = done; });
|
||||
const repeated = new Promise<void>(done => { duplicate = done; });
|
||||
const originalRead = archiveOwner.readProviderPackArchive;
|
||||
const verification = vi.spyOn(archiveOwner, "readProviderPackArchive").mockImplementation(async (...args) => {
|
||||
enter();
|
||||
if (verification.mock.calls.length > 1) duplicate();
|
||||
await gate;
|
||||
return originalRead(...args);
|
||||
});
|
||||
const originalAny = AbortSignal.any;
|
||||
vi.spyOn(AbortSignal, "any").mockImplementation(signals => {
|
||||
const combined = originalAny(signals);
|
||||
if (signals.includes(secondController.signal)) {
|
||||
const add = combined.addEventListener.bind(combined);
|
||||
vi.spyOn(combined, "addEventListener").mockImplementation((type, listener, options) => {
|
||||
const result = add(type, listener, options);
|
||||
if (type === "abort") joined();
|
||||
return result;
|
||||
});
|
||||
}
|
||||
return combined;
|
||||
});
|
||||
const source = { root: input.current.pack, sourceDigest: input.current.manifest.digest, manifest: input.selected.manifest };
|
||||
const first = resolveProviderPackUploadSource({ ...source, signal: firstController.signal });
|
||||
const firstStopped = expect(first).rejects.toThrow("owned shared-cache waiter stopped");
|
||||
await entered;
|
||||
const second = resolveProviderPackUploadSource({ ...source, signal: secondController.signal });
|
||||
try {
|
||||
await Promise.race([secondJoined, repeated.then(() => { throw Error("warm-cache verification repeated"); })]);
|
||||
expect(verification).toHaveBeenCalledOnce();
|
||||
firstController.abort(new Error("owned shared-cache waiter stopped"));
|
||||
await firstStopped;
|
||||
expect(verification.mock.calls[0]?.[2]?.aborted).toBe(false);
|
||||
release();
|
||||
expect(await second).toMatchObject({ archiveOnly: true });
|
||||
expect(input.requests).toHaveLength(requests);
|
||||
} finally {
|
||||
firstController.abort(new Error("owned shared-cache waiter stopped"));
|
||||
secondController.abort(new Error("owned shared-cache fixture cleanup"));
|
||||
release();
|
||||
await Promise.allSettled([first, second, firstStopped]);
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects a changed cached archive before any remote command or transfer", async () => {
|
||||
const input = await fixture();
|
||||
await upload(input);
|
||||
const cache = path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`);
|
||||
const publication = JSON.parse(await fs.readFile(path.join(cache, input.selected.manifest.digest.slice(7) + ".json"), "utf8"));
|
||||
await fs.appendFile(path.join(cache, publication.entry, "provider-pack.archives/overlay.tar.zst"), "changed");
|
||||
input.runner.execute.mockClear();
|
||||
input.runner.syncIn.mockClear();
|
||||
await expect(upload(input, { targetPath: input.targetPath + "-new" })).rejects.toThrow("runner_remote_provider_archive_invalid");
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
expect(await fs.readFile(path.join(input.targetPath, "bin/tool"), "utf8")).toBe("retained provider code");
|
||||
});
|
||||
|
||||
it("refuses an unsafe cache directory instead of following it", async () => {
|
||||
const input = await fixture();
|
||||
const outside = path.join(input.root, "outside");
|
||||
await fs.mkdir(outside);
|
||||
await fs.symlink(outside, path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`));
|
||||
await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_invalid");
|
||||
expect(await fs.readdir(outside)).toEqual([]);
|
||||
expect(input.requests).toEqual([]);
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("leaves a target intact when its archive transport cannot install the verified source", async () => {
|
||||
const input = await fixture();
|
||||
await fs.mkdir(input.targetPath);
|
||||
await fs.writeFile(path.join(input.targetPath, "marker"), "untouched");
|
||||
input.runner.execute.mockResolvedValue({ exitCode: 127, timedOut: false, stdout: "", stderr: "unavailable" });
|
||||
await expect(upload(input)).rejects.toThrow("runner_remote_provider_archive_transport_unavailable");
|
||||
expect(input.runner.execute).toHaveBeenCalledOnce();
|
||||
expect(input.runner.execute.mock.calls[0]?.[0].args?.join(" ")).not.toContain("rm -rf");
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
expect(await fs.readFile(path.join(input.targetPath, "marker"), "utf8")).toBe("untouched");
|
||||
});
|
||||
|
||||
it("refuses an oversized historical layer specification before any target effects", async () => {
|
||||
const input = await fixture();
|
||||
await fs.mkdir(input.targetPath);
|
||||
await fs.writeFile(path.join(input.targetPath, "marker"), "untouched");
|
||||
const reuseRoots = Array.from({ length: 2000 }, (_, index) => `/qualified-image-${index}/` + "a".repeat(64));
|
||||
await expect(upload(input, { reuseRoots })).rejects.toThrow("runner_remote_provider_archive_transport_unavailable");
|
||||
expect(input.runner.execute).not.toHaveBeenCalled();
|
||||
expect(input.runner.syncIn).not.toHaveBeenCalled();
|
||||
expect(await fs.readFile(path.join(input.targetPath, "marker"), "utf8")).toBe("untouched");
|
||||
});
|
||||
@@ -0,0 +1,308 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { constants, createReadStream } from "node:fs";
|
||||
import { link, lstat, mkdir, mkdtemp, open, readdir, realpath, rm, statfs, unlink, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { basename, join } from "node:path";
|
||||
import { Transform } from "node:stream";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import { createZstdDecompress } from "node:zlib";
|
||||
import tar from "tar-stream";
|
||||
import { canonicalProviderPackJson, type ProviderPackIdentity } from "./remote-provider-pack-verification.js";
|
||||
import { readProviderPackArchive } from "./provider-pack-archive.js";
|
||||
import { MAX_RETAINED_PACK_BYTES, receiveRetainedProviderPack, type RetainedPackProvenance } from "./retained-provider-pack-oci.js";
|
||||
|
||||
type Source = { root: string; archiveOnly: true };
|
||||
type ImportJob = { operation: Promise<Source>; controller: AbortController; waiters: number };
|
||||
const pending = new Map<string, ImportJob>();
|
||||
const waiting = new Set<() => void>();
|
||||
let activeImports = 0;
|
||||
const digest = (value: Buffer | string) => createHash("sha256").update(value).digest("hex");
|
||||
function fail(): never { throw Error("runner_remote_provider_source_invalid"); }
|
||||
const maximumMetadata = 1024 * 1024;
|
||||
type Published = RetainedPackProvenance & { entry: string; receiptSha256: string; verifiedTarPaths: true };
|
||||
const safeName = (value: string) => value.length > 0 && Buffer.byteLength(value) <= 4096 && !/[\\\u0000-\u001f\u007f]/u.test(value)
|
||||
&& !value.startsWith("/") && value.split("/").every(part => part !== "" && part !== "." && part !== "..");
|
||||
|
||||
async function waitForSource<T>(operation: Promise<T>, signal: AbortSignal): Promise<T> {
|
||||
signal.throwIfAborted();
|
||||
let stopped!: () => void;
|
||||
const aborted = new Promise<never>((_resolve, reject) => {
|
||||
stopped = () => reject(signal.reason);
|
||||
signal.addEventListener("abort", stopped, { once: true });
|
||||
if (signal.aborted) stopped();
|
||||
});
|
||||
try { return await Promise.race([operation, aborted]); }
|
||||
finally { signal.removeEventListener("abort", stopped); }
|
||||
}
|
||||
|
||||
async function importSlot(signal: AbortSignal) {
|
||||
while (activeImports >= 2) {
|
||||
let resume!: () => void;
|
||||
const available = new Promise<void>(done => { resume = done; waiting.add(done); });
|
||||
try { await waitForSource(available, signal); }
|
||||
finally { waiting.delete(resume); }
|
||||
}
|
||||
signal.throwIfAborted();
|
||||
activeImports++;
|
||||
return () => { activeImports--; for (const resume of waiting) resume(); };
|
||||
}
|
||||
|
||||
async function readStable(path: string, maximum = maximumMetadata) {
|
||||
const file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
|
||||
try {
|
||||
const before = await file.stat();
|
||||
if (!before.isFile() || before.size > maximum) fail();
|
||||
const bytes = await file.readFile();
|
||||
const after = await file.stat();
|
||||
if (before.ino !== after.ino || before.dev !== after.dev || before.size !== after.size || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs) fail();
|
||||
return bytes;
|
||||
} finally { await file.close(); }
|
||||
}
|
||||
|
||||
async function regularDigest(path: string, expected: { sha256: string; bytes: number }, signal: AbortSignal) {
|
||||
const file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
|
||||
try {
|
||||
const before = await file.stat();
|
||||
if (!before.isFile() || before.size !== expected.bytes) fail();
|
||||
const hash = createHash("sha256");
|
||||
let bytes = 0;
|
||||
for await (const chunk of file.createReadStream({ autoClose: false, signal })) { bytes += chunk.length; hash.update(chunk); }
|
||||
const after = await file.stat();
|
||||
if (bytes !== before.size || before.ino !== after.ino || before.dev !== after.dev || before.size !== after.size || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs || hash.digest("hex") !== expected.sha256) fail();
|
||||
} finally { await file.close(); }
|
||||
}
|
||||
|
||||
async function selectedArchive(root: string, manifest: ProviderPackIdentity, signal: AbortSignal) {
|
||||
signal.throwIfAborted();
|
||||
const value = JSON.parse((await readStable(join(root, "provider-pack.json"))).toString()) as ProviderPackIdentity;
|
||||
if (canonicalProviderPackJson(value) !== canonicalProviderPackJson(manifest)) fail();
|
||||
const archive = await readProviderPackArchive(root, manifest.digest, signal);
|
||||
if (!archive || !("kind" in archive)) fail();
|
||||
const names = ["overlay.tar.zst", ...archive.files.map((_file, index) => `${index}.tar.zst`)];
|
||||
if (canonicalProviderPackJson((await readdir(archive.sourceRoot)).sort()) !== canonicalProviderPackJson(names.sort())) fail();
|
||||
let bytes = archive.overlay.bytes;
|
||||
for (const [index, file] of archive.files.entries()) {
|
||||
signal.throwIfAborted();
|
||||
bytes += file.archive.bytes;
|
||||
if (bytes > MAX_RETAINED_PACK_BYTES) fail();
|
||||
await regularDigest(join(archive.sourceRoot, `${index}.tar.zst`), file.archive, signal);
|
||||
}
|
||||
signal.throwIfAborted();
|
||||
return archive;
|
||||
}
|
||||
|
||||
async function extractManifest(root: string, manifest: ProviderPackIdentity, signal: AbortSignal) {
|
||||
const source = await readProviderPackArchive(root, manifest.digest, signal);
|
||||
if (!source || !("kind" in source)) fail();
|
||||
const regular = new Set(source.files.map(file => file.path));
|
||||
const archive = tar.extract();
|
||||
const names = new Set<string>(), files = new Set<string>(), links = new Map<string, string>();
|
||||
let bytes = 0, count = 0;
|
||||
let found: Buffer | undefined;
|
||||
archive.on("entry", (header, body, next) => {
|
||||
body.on("error", () => undefined);
|
||||
void (async () => {
|
||||
signal.throwIfAborted();
|
||||
if (++count > 100_000) fail();
|
||||
const normalized = header.name.replace(/^\.\//u, "");
|
||||
const name = header.type === "directory" ? normalized.replace(/\/$/u, "") : normalized;
|
||||
if (!safeName(name) || names.has(name) || !["file", "directory", "symlink", "link"].includes(header.type ?? "")
|
||||
|| !Number.isSafeInteger(header.size ?? 0) || (header.size ?? 0) < 0 || ((header.mode ?? 0) & ~0o777) !== 0) fail();
|
||||
names.add(name);
|
||||
if (header.type !== "directory") files.add(name);
|
||||
if (header.type === "link") {
|
||||
const target = header.linkname?.replace(/^\.\//u, "");
|
||||
if (!target || !safeName(target) || !regular.has(target)) fail();
|
||||
regular.add(name);
|
||||
} else if (header.type === "file") regular.add(name);
|
||||
if (header.type === "symlink") {
|
||||
if (!header.linkname || header.linkname.startsWith("/") || /[\\\u0000-\u001f\u007f]/u.test(header.linkname) || Buffer.byteLength(header.linkname) > 4096) fail();
|
||||
links.set(name, header.linkname);
|
||||
}
|
||||
if (name !== "provider-pack.json") { body.resume(); return; }
|
||||
if (found || header.type !== "file" || !Number.isSafeInteger(header.size) || header.size! > maximumMetadata) fail();
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of body) {
|
||||
bytes += chunk.length;
|
||||
if (bytes > maximumMetadata) fail();
|
||||
chunks.push(Buffer.from(chunk));
|
||||
}
|
||||
found = Buffer.concat(chunks);
|
||||
})().then(() => next(), error => archive.destroy(error));
|
||||
});
|
||||
let expanded = 0;
|
||||
const bound = new Transform({ transform(chunk, _encoding, done) {
|
||||
expanded += chunk.length;
|
||||
done(expanded > MAX_RETAINED_PACK_BYTES ? Error("runner_remote_provider_source_size_limit") : null, chunk);
|
||||
} });
|
||||
await pipeline(createReadStream(root + ".archives/overlay.tar.zst"), createZstdDecompress(), bound, archive, { signal });
|
||||
for (const name of names) {
|
||||
let parent = name;
|
||||
while (parent.includes("/")) {
|
||||
parent = parent.slice(0, parent.lastIndexOf("/"));
|
||||
if (files.has(parent)) fail();
|
||||
}
|
||||
}
|
||||
for (const [name, target] of links) {
|
||||
let parts = [...name.split("/").slice(0, -1), ...target.split("/")];
|
||||
for (let followed = 0; ; followed++) {
|
||||
if (followed > 32) fail();
|
||||
const resolved: string[] = [];
|
||||
let changed = false;
|
||||
for (let index = 0; index < parts.length; index++) {
|
||||
const part = parts[index]!;
|
||||
if (part === "" || part === ".") continue;
|
||||
if (part === "..") { if (!resolved.length) fail(); resolved.pop(); continue; }
|
||||
resolved.push(part);
|
||||
const link = links.get(resolved.join("/"));
|
||||
if (link !== undefined) { parts = [...resolved.slice(0, -1), ...link.split("/"), ...parts.slice(index + 1)]; changed = true; break; }
|
||||
}
|
||||
if (!changed) break;
|
||||
}
|
||||
}
|
||||
if (!found || canonicalProviderPackJson(JSON.parse(found.toString())) !== canonicalProviderPackJson(manifest)) fail();
|
||||
await mkdir(root, { mode: 0o700 });
|
||||
await writeFile(join(root, "provider-pack.json"), found, { flag: "wx", mode: 0o600 });
|
||||
return { expanded, files };
|
||||
}
|
||||
|
||||
async function verifyFileContents(archivePath: string, file: { path: string; bytes: number; sha256: string; mode: number }, signal: AbortSignal) {
|
||||
const archive = tar.extract();
|
||||
let found = false, count = 0;
|
||||
archive.on("entry", (header, body, next) => {
|
||||
body.on("error", () => undefined);
|
||||
void (async () => {
|
||||
signal.throwIfAborted();
|
||||
const normalized = header.name.replace(/^\.\//u, "");
|
||||
const name = header.type === "directory" ? normalized.replace(/\/$/u, "") : normalized;
|
||||
if (++count > 128 || !safeName(name)) fail();
|
||||
if (header.type === "directory" && file.path.startsWith(name + "/")) { body.resume(); return; }
|
||||
if (found || header.type !== "file" || name !== file.path || header.size !== file.bytes || header.mode !== file.mode) fail();
|
||||
found = true;
|
||||
const hash = createHash("sha256");
|
||||
let bytes = 0;
|
||||
for await (const chunk of body) {
|
||||
signal.throwIfAborted();
|
||||
bytes += chunk.length;
|
||||
if (bytes > file.bytes) fail();
|
||||
hash.update(chunk);
|
||||
}
|
||||
if (bytes !== file.bytes || hash.digest("hex") !== file.sha256) fail();
|
||||
})().then(() => next(), error => archive.destroy(error));
|
||||
});
|
||||
let expanded = 0;
|
||||
const bound = new Transform({ transform(chunk, _encoding, done) {
|
||||
expanded += chunk.length;
|
||||
done(expanded > file.bytes + 2 * 1024 * 1024 ? Error("runner_remote_provider_source_size_limit") : null, chunk);
|
||||
} });
|
||||
await pipeline(createReadStream(archivePath), createZstdDecompress(), bound, archive, { signal });
|
||||
if (!found) fail();
|
||||
}
|
||||
|
||||
async function privateDirectory(root: string) {
|
||||
const metadata = await lstat(root);
|
||||
if (!metadata.isDirectory() || metadata.isSymbolicLink() || metadata.uid !== process.getuid?.() || (metadata.mode & 0o077) !== 0) fail();
|
||||
}
|
||||
|
||||
async function cached(cache: string, pointer: string, manifest: ProviderPackIdentity, signal: AbortSignal) {
|
||||
let bytes: Buffer;
|
||||
try { bytes = await readStable(pointer); } catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
|
||||
throw error;
|
||||
}
|
||||
const publication = JSON.parse(bytes.toString()) as Published;
|
||||
if (publication.schema !== "paperclip.retained-provider-pack-source.v1" || publication.verifiedTarPaths !== true || publication.providerPackDigest !== manifest.digest
|
||||
|| publication.sourceRevision !== manifest.payload.runnerSourceRevision || !new RegExp(`^${manifest.digest.slice(7)}-[A-Za-z0-9]{6}$`).test(publication.entry)
|
||||
|| !/^[a-f0-9]{64}$/.test(publication.receiptSha256)
|
||||
|| [publication.imageIndexDigest, publication.imageManifestDigest, publication.imageConfigDigest, publication.transportLayerDigest].some(value => !/^sha256:[a-f0-9]{64}$/.test(value))) fail();
|
||||
const directory = join(cache, publication.entry);
|
||||
await privateDirectory(directory);
|
||||
const root = join(directory, "provider-pack");
|
||||
await privateDirectory(root);
|
||||
if (digest(await readStable(root + ".archive.json")) !== publication.receiptSha256) fail();
|
||||
await selectedArchive(root, manifest, signal);
|
||||
return { root, archiveOnly: true as const };
|
||||
}
|
||||
|
||||
async function importSource(cache: string, pointer: string, manifest: ProviderPackIdentity, signal: AbortSignal) {
|
||||
const space = await statfs(cache, { bigint: true });
|
||||
if (space.bavail * space.bsize < BigInt(2 * MAX_RETAINED_PACK_BYTES + 1024 ** 3)) throw Error("runner_remote_provider_source_disk_headroom_required");
|
||||
const directory = await mkdtemp(join(cache, manifest.digest.slice(7) + "-"));
|
||||
const temporaryPointer = join(cache, `.${randomUUID()}.pending`);
|
||||
let published = false;
|
||||
try {
|
||||
const provenance = await receiveRetainedProviderPack({ manifest, directory, signal });
|
||||
const root = join(directory, "provider-pack");
|
||||
const overlay = await extractManifest(root, manifest, signal);
|
||||
const archive = await selectedArchive(root, manifest, signal);
|
||||
let expanded = overlay.expanded;
|
||||
for (const [index, file] of archive.files.entries()) {
|
||||
if (!Number.isSafeInteger(file.bytes) || file.bytes < 1 || (expanded += file.bytes) > MAX_RETAINED_PACK_BYTES || overlay.files.has(file.path)) fail();
|
||||
let parent = file.path;
|
||||
while (parent.includes("/")) { parent = parent.slice(0, parent.lastIndexOf("/")); if (overlay.files.has(parent)) fail(); }
|
||||
await verifyFileContents(join(archive.sourceRoot, `${index}.tar.zst`), file, signal);
|
||||
}
|
||||
await unlink(join(directory, "transport.tar.gz"));
|
||||
signal.throwIfAborted();
|
||||
const publication: Published = { ...provenance, entry: basename(directory), receiptSha256: digest(await readStable(root + ".archive.json")), verifiedTarPaths: true };
|
||||
await writeFile(temporaryPointer, JSON.stringify(publication), { flag: "wx", mode: 0o600 });
|
||||
signal.throwIfAborted();
|
||||
try {
|
||||
await link(temporaryPointer, pointer);
|
||||
published = true;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error;
|
||||
const winner = await cached(cache, pointer, manifest, signal);
|
||||
if (!winner) fail();
|
||||
return winner;
|
||||
}
|
||||
return { root, archiveOnly: true as const };
|
||||
} finally {
|
||||
await rm(temporaryPointer, { force: true });
|
||||
if (!published) await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
export async function resolveProviderPackUploadSource(input: {
|
||||
root: string;
|
||||
sourceDigest?: string;
|
||||
manifest: ProviderPackIdentity;
|
||||
signal?: AbortSignal;
|
||||
}): Promise<{ root: string; archiveOnly: boolean }> {
|
||||
const manifest = input.manifest;
|
||||
if (manifest?.schema !== "paperclip-runner/remote-provider-pack/v1" || manifest.digest !== `sha256:${digest(canonicalProviderPackJson(manifest.payload))}`) fail();
|
||||
input.signal?.throwIfAborted();
|
||||
const configured = JSON.parse((await readStable(join(input.root, "provider-pack.json"))).toString()) as ProviderPackIdentity;
|
||||
if (configured?.schema !== manifest.schema || configured.digest !== `sha256:${digest(canonicalProviderPackJson(configured.payload))}`
|
||||
|| input.sourceDigest !== undefined && input.sourceDigest !== configured.digest) fail();
|
||||
if (configured.digest === manifest.digest) return { root: input.root, archiveOnly: false };
|
||||
const cache = join(await realpath(tmpdir()), `paperclip-provider-pack-sources-${process.getuid?.()}`);
|
||||
await mkdir(cache, { recursive: true, mode: 0o700 });
|
||||
await privateDirectory(cache);
|
||||
const signal = input.signal ? AbortSignal.any([input.signal, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000);
|
||||
const pointer = join(cache, manifest.digest.slice(7) + ".json");
|
||||
const key = `${cache}:${manifest.digest}`;
|
||||
let job = pending.get(key);
|
||||
if (!job || job.controller.signal.aborted) {
|
||||
if (pending.size >= 32) throw Error("runner_remote_provider_source_import_busy");
|
||||
const controller = new AbortController();
|
||||
const sharedSignal = AbortSignal.any([controller.signal, AbortSignal.timeout(180_000)]);
|
||||
let created!: ImportJob;
|
||||
const operation = (async () => {
|
||||
const release = await importSlot(sharedSignal);
|
||||
try { return await cached(cache, pointer, manifest, sharedSignal) ?? await importSource(cache, pointer, manifest, sharedSignal); }
|
||||
finally { release(); }
|
||||
})().finally(() => { if (pending.get(key) === created) pending.delete(key); });
|
||||
job = created = { operation, controller, waiters: 0 };
|
||||
pending.set(key, job);
|
||||
}
|
||||
job.waiters++;
|
||||
try {
|
||||
const source = await waitForSource(job.operation, signal);
|
||||
signal.throwIfAborted();
|
||||
return source;
|
||||
} finally {
|
||||
job.waiters--;
|
||||
if (job.waiters === 0 && pending.get(key) === job) job.controller.abort(new DOMException("provider source no longer requested", "AbortError"));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user