diff --git a/src/application/ProcessCaptureAuthority.ts b/src/application/ProcessCaptureAuthority.ts new file mode 100644 index 0000000000..c33ce29548 --- /dev/null +++ b/src/application/ProcessCaptureAuthority.ts @@ -0,0 +1,48 @@ +import { realpath } from "node:fs/promises"; + +import type { + ProcessExecutionPolicy, + ProcessScenario, +} from "../domain/processCapture.js"; +import { ProcessCaptureError } from "./ProcessCaptureError.js"; + +const isWithin = (candidate: string, root: string): boolean => + candidate === root || + candidate.startsWith(`${root.endsWith("/") ? root.slice(0, -1) : root}/`); + +export const assertRealPathAuthority = async ( + scenario: ProcessScenario, + policy: ProcessExecutionPolicy, +): Promise => { + const executable = await realpath(scenario.executable); + const executableRoots = await Promise.all( + policy.executableRoots.map((root) => realpath(root)), + ); + if (!executableRoots.some((root) => isWithin(executable, root))) + throw new ProcessCaptureError( + "resolved executable is outside approved roots", + ); + const workingDirectory = await realpath(scenario.working_directory); + const workingRoots = await Promise.all( + policy.workingRoots.map((root) => realpath(root)), + ); + if (!workingRoots.some((root) => isWithin(workingDirectory, root))) + throw new ProcessCaptureError( + "resolved working directory is outside approved roots", + ); + for (const root of scenario.filesystem_roots) { + const resolvedRoot = await realpath(root); + if (!workingRoots.some((approved) => isWithin(resolvedRoot, approved))) + throw new ProcessCaptureError( + "resolved filesystem root is outside approved roots", + ); + } +}; + +/** Expected refusal or runtime failure from the process capture adapter. */ +export const assertNotCancelled = (signal: AbortSignal | undefined): void => { + if (signal?.aborted === true) + throw new ProcessCaptureError("process capture was cancelled"); +}; + +/** Runtime availability of the native PTY adapter on this host. */ diff --git a/src/application/ProcessCaptureCapability.ts b/src/application/ProcessCaptureCapability.ts new file mode 100644 index 0000000000..542fe9845d --- /dev/null +++ b/src/application/ProcessCaptureCapability.ts @@ -0,0 +1,38 @@ +import { tmpdir } from "node:os"; +import { spawn } from "node-pty"; + +export type ProcessCaptureCapability = + | { readonly available: true; readonly backend: "node-pty" } + | { + readonly available: false; + readonly backend: "node-pty"; + readonly reason: string; + }; + +/** Probe the actual native PTY seam instead of inferring support from the OS name. */ +export const probeProcessCaptureCapability = + async (): Promise => { + try { + const terminal = spawn( + process.platform === "win32" ? "cmd.exe" : "/bin/sh", + process.platform === "win32" ? ["/c", "exit", "0"] : ["-c", "exit 0"], + { + cwd: tmpdir(), + env: { HOME: tmpdir(), TERM: "xterm-256color" }, + cols: 80, + rows: 24, + name: "xterm-256color", + }, + ); + await new Promise((resolveExit) => + terminal.onExit(() => resolveExit()), + ); + return { available: true, backend: "node-pty" }; + } catch { + return { + available: false, + backend: "node-pty", + reason: "the native PTY backend could not start a probe process", + }; + } + }; diff --git a/src/application/ProcessCaptureError.ts b/src/application/ProcessCaptureError.ts new file mode 100644 index 0000000000..17593e7d10 --- /dev/null +++ b/src/application/ProcessCaptureError.ts @@ -0,0 +1,6 @@ +import { AnalysisError } from "../domain/errors.js"; + +/** Typed application failure produced by controlled process capture. */ +export class ProcessCaptureError extends AnalysisError { + readonly _tag = "ProcessCaptureError"; +} diff --git a/src/application/ProcessCaptureLifecycle.ts b/src/application/ProcessCaptureLifecycle.ts new file mode 100644 index 0000000000..7470079fd6 --- /dev/null +++ b/src/application/ProcessCaptureLifecycle.ts @@ -0,0 +1,417 @@ +import { createHash, randomUUID } from "node:crypto"; +import { createReadStream } from "node:fs"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { IPty } from "node-pty"; + +import type { + FilesystemCheckpoint, + InteractionEvent, + ProcessCapture, + ProcessExecutionPolicy, + ProcessSample, + ProcessScenario, + TerminalFrame, +} from "../domain/processCapture.js"; +import { + authorizeProcessScenario, + digestProcessCommitment, + processComparisonContract, + processScenarioCommitment, +} from "../domain/processCapture.js"; +import { PRODUCT_IDENTITY } from "../identity.js"; +import type { SnapshotResult } from "./FilesystemSnapshot.js"; +import { snapshotRoots } from "./FilesystemSnapshot.js"; +import type { LoopbackReplay } from "./LoopbackReplay.js"; +import { + cleanupOwnedProcessGroup, + observeOwnedProcessGroup, +} from "./ProcessOwnership.js"; +import { + classifyFilesystemEffects, + ProcessCheckpoints, +} from "./ProcessCheckpoints.js"; +import { ProcessCaptureError } from "./ProcessCaptureError.js"; +import { + assertNotCancelled, + assertRealPathAuthority, +} from "./ProcessCaptureAuthority.js"; +import { + normalizeProcessSamples, + normalizeProcessText, + redactProtocolEvents, +} from "./ProcessNormalization.js"; +import type { CommandShimReplay } from "./CommandShimReplay.js"; +import { PROCESS_PROVIDER } from "./ProcessEvidence.js"; +import { TerminalRenderer } from "./TerminalRenderer.js"; + +interface TerminalExitOptions { + readonly terminal: IPty; + readonly scenario: ProcessScenario; + readonly started: number; + readonly lastOutput: () => number; + readonly signal: AbortSignal | undefined; + readonly timers: Set; + readonly interactions: InteractionEvent[]; + readonly dispatchedEventIndexes: ReadonlySet; +} + +interface CaptureResultOptions { + readonly frames: readonly TerminalFrame[]; + readonly exit: { + readonly exitCode: number; + readonly signal?: number; + readonly reason: "exited" | "timeout" | "idle_timeout"; + }; + readonly samples: readonly ProcessSample[]; + readonly replay: LoopbackReplay; + readonly before: SnapshotResult; + readonly after: SnapshotResult; + readonly truncated: boolean; + readonly scenario: ProcessScenario; + readonly rootPid: number; + readonly samplingPartial: boolean; + readonly renderedFrames: ProcessCapture["rendered_frames"]; + readonly interactions: readonly InteractionEvent[]; + readonly checkpoints: readonly FilesystemCheckpoint[]; + readonly shimEvents: ProcessCapture["shim_events"]; + readonly settlement: ProcessCapture["settlement"]; + readonly manifest: ProcessCapture["manifest"]; +} + +export const buildCaptureResult = ( + options: CaptureResultOptions, +): ProcessCapture => ({ + schema_version: 4, + manifest: options.manifest, + normalization: options.scenario.normalization, + frames: options.frames, + rendered_frames: [...options.renderedFrames] + .sort( + (left, right) => + left.at_ms - right.at_ms || left.sequence - right.sequence, + ) + .map((frame, sequence) => ({ ...frame, sequence })), + interaction_events: options.interactions, + exit: { + code: + options.exit.reason === "exited" && options.exit.exitCode >= 0 + ? options.exit.exitCode + : null, + signal: options.exit.signal ?? null, + reason: options.exit.reason, + }, + settlement: options.settlement, + process_samples: normalizeProcessSamples( + options.samples, + options.scenario, + options.rootPid, + ), + filesystem_checkpoints: options.checkpoints, + shim_events: options.shimEvents, + protocol_events: redactProtocolEvents( + options.replay.events, + options.scenario, + ), + files_before: options.before.files, + files_after: options.after.files, + filesystem_effects: classifyFilesystemEffects( + options.before.files, + options.after.files, + ), + truncated: options.truncated, + limitations: [ + "Process trees are sampled and may omit short-lived descendants.", + ...(options.samplingPartial + ? ["Process-tree sampling ended with an incomplete observation."] + : []), + "Filesystem observations are before/after snapshots, not syscall traces.", + "The harness does not enforce external network isolation.", + ], + residual_unknowns: [ + { + scope: "process", + reason: "Process trees are sampled and may omit short-lived descendants.", + }, + { + scope: "network", + reason: "External network isolation is not enforced by this adapter.", + }, + ], + cleanup: { + owned_process_group: "verified", + temporary_root: "removed", + }, +}); + +const hashFile = async (path: string): Promise => { + const hash = createHash("sha256"); + for await (const chunk of createReadStream(path)) hash.update(chunk); + return hash.digest("hex"); +}; + +export const createRunManifest = async ( + scenario: ProcessScenario, + startedAt: Date, + completedAt: Date, +): Promise => { + const executableSha256 = await hashFile(scenario.executable); + const scenarioCommitment = processScenarioCommitment( + scenario, + executableSha256, + ); + const comparisonContract = processComparisonContract(scenario); + return { + rea_version: PRODUCT_IDENTITY.packageVersion, + provider_version: PROCESS_PROVIDER.version, + platform: process.platform, + architecture: process.arch, + pty_backend: "node-pty", + started_at: startedAt.toISOString(), + completed_at: completedAt.toISOString(), + scenario: scenarioCommitment, + comparison_contract: comparisonContract, + shim_plan: scenario.command_shims, + replay_plan: scenario.replay, + full_scenario_sha256: digestProcessCommitment(scenarioCommitment), + comparison_contract_sha256: digestProcessCommitment(comparisonContract), + executable_sha256: executableSha256, + normalization_sha256: digestProcessCommitment(scenario.normalization), + shim_plan_sha256: digestProcessCommitment(scenario.command_shims), + replay_plan_sha256: digestProcessCommitment(scenario.replay), + }; +}; + +export const observeSettlement = async ( + runId: string, + processGroupIds: readonly number[], + settleMs: number, +): Promise> => { + if (process.platform === "win32") + return { state: "unverifiable", elapsed_ms: 0 }; + const started = Date.now(); + let consecutiveEmpty = 0; + let deadlineReached = false; + while (!deadlineReached) { + const observations = await Promise.all( + [...new Set(processGroupIds)].map((processGroupId) => + observeOwnedProcessGroup({ + runId, + leaderPid: processGroupId, + processGroupId, + }), + ), + ); + if (observations.some(({ state }) => state === "unverifiable")) + return { state: "unverifiable", elapsed_ms: Date.now() - started }; + if (observations.every(({ state }) => state === "empty")) { + consecutiveEmpty += 1; + if (consecutiveEmpty >= 2) + return { state: "quiesced", elapsed_ms: Date.now() - started }; + } else consecutiveEmpty = 0; + deadlineReached = Date.now() - started >= settleMs; + if (deadlineReached) break; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + return { state: "alive_at_deadline", elapsed_ms: Date.now() - started }; +}; + +export const awaitTerminalExit = async ({ + terminal, + scenario, + started, + lastOutput, + signal, + timers, + interactions, + dispatchedEventIndexes, +}: TerminalExitOptions): Promise<{ + exitCode: number; + signal?: number; + reason: "exited" | "timeout" | "idle_timeout" | "cancelled"; +}> => + new Promise((resolveExit) => { + // The kill caused by a deadline is observed later as an ordinary PTY exit. + // Keep the initiating lifecycle reason so comparisons distinguish a target + // exit from harness-owned timeout, idle-timeout, and cancellation cleanup. + let reason: "exited" | "timeout" | "idle_timeout" | "cancelled" = "exited"; + terminal.onExit((exit) => { + for (const [eventIndex, event] of scenario.events.entries()) { + if (dispatchedEventIndexes.has(eventIndex)) continue; + interactions.push({ + sequence: interactions.length, + scheduled_at_ms: event.at_ms, + dispatched_at_ms: Math.max(0, Date.now() - started), + type: event.type, + data: + event.type === "input" + ? "" + : event.type === "resize" + ? `${String(event.columns)}x${String(event.rows)}` + : event.signal, + outcome: "target_exited", + }); + } + for (const timer of timers) { + clearTimeout(timer); + clearInterval(timer); + } + timers.clear(); + resolveExit({ ...exit, reason }); + }); + const timeout = setInterval(() => { + if (signal?.aborted === true) { + reason = "cancelled"; + terminal.kill("SIGKILL"); + } else if (Date.now() - started >= scenario.timeout_ms) { + reason = "timeout"; + terminal.kill("SIGKILL"); + } else if (Date.now() - lastOutput() >= scenario.idle_timeout_ms) { + reason = "idle_timeout"; + terminal.kill("SIGKILL"); + } + }, 20); + timers.add(timeout); + }); + +export const releaseProcessResources = async (options: { + readonly timers: ReadonlySet; + readonly replay: LoopbackReplay | undefined; + readonly terminal: IPty | undefined; + readonly renderer: TerminalRenderer | undefined; + readonly shimReplay: CommandShimReplay | undefined; + readonly checkpoints: ProcessCheckpoints | undefined; + readonly runId: string; + readonly temporaryRoot: string; + readonly sampledProcessGroupIds: readonly number[]; +}): Promise => { + for (const timer of options.timers) clearTimeout(timer); + let failure: string | undefined; + try { + await options.replay?.close(); + } catch { + failure = "loopback replay cleanup failed"; + } + try { + await options.shimReplay?.close(); + } catch { + failure ??= "command shim replay cleanup failed"; + } + try { + await options.renderer?.dispose(); + } catch { + failure ??= "terminal renderer cleanup failed"; + } + try { + await options.checkpoints?.dispose(); + } catch { + failure ??= "filesystem checkpoint cleanup failed"; + } + if (options.terminal !== undefined && process.platform !== "win32") { + const processGroupIds = new Set([ + options.terminal.pid, + ...options.sampledProcessGroupIds, + ]); + for (const processGroupId of processGroupIds) { + const cleaned = await cleanupOwnedProcessGroup({ + runId: options.runId, + leaderPid: processGroupId, + processGroupId, + }); + if (!cleaned.cleaned) failure ??= cleaned.reason; + } + } + try { + await rm(options.temporaryRoot, { recursive: true, force: true }); + } catch { + failure ??= "temporary process root cleanup failed"; + } + return failure; +}; + +export const captureTerminalFrames = (options: { + readonly terminal: IPty; + readonly scenario: ProcessScenario; + readonly frames: TerminalFrame[]; + readonly started: number; + readonly temporaryRoot: string; + readonly onOutput: () => void; + readonly onFirstOutput: () => void; + readonly renderer: TerminalRenderer; + readonly checkpoints: ProcessCheckpoints; +}): (() => boolean) => { + let outputBytes = 0; + let truncated = false; + options.terminal.onData((data) => { + options.onFirstOutput(); + options.onOutput(); + const bytes = Buffer.byteLength(data); + if ( + options.frames.length >= options.scenario.limits.frames || + outputBytes + bytes > options.scenario.limits.output_bytes + ) { + truncated = true; + return; + } + outputBytes += bytes; + const atMs = + Math.floor( + (Date.now() - options.started) / + options.scenario.normalization.time_bucket_ms, + ) * options.scenario.normalization.time_bucket_ms; + const normalized = normalizeProcessText( + data, + options.scenario, + options.temporaryRoot, + options.terminal.pid, + ); + options.frames.push({ + sequence: options.frames.length, + at_ms: atMs, + data: normalized, + }); + options.renderer.write(data, atMs); + options.checkpoints.observeTerminal(normalized); + }); + return () => truncated; +}; + +export const resolveProcessResult = ( + capture: ProcessCapture | undefined, + executionFailure: unknown, + cleanupFailure: string | undefined, +): ProcessCapture => { + if (executionFailure instanceof ProcessCaptureError) throw executionFailure; + if (executionFailure !== undefined) + throw new ProcessCaptureError("process capture failed", { + cause: executionFailure, + }); + if (cleanupFailure !== undefined) + throw new ProcessCaptureError(cleanupFailure); + if (capture === undefined) + throw new ProcessCaptureError("process capture produced no result"); + return capture; +}; + +export const prepareProcessCapture = async ( + scenario: ProcessScenario, + policy: ProcessExecutionPolicy, + signal: AbortSignal | undefined, +): Promise<{ + readonly temporaryRoot: string; + readonly runId: string; + readonly home: string; + readonly before: SnapshotResult; +}> => { + const decision = authorizeProcessScenario(scenario, policy); + if (!decision.allowed) throw new ProcessCaptureError(decision.reason); + await assertRealPathAuthority(scenario, policy); + assertNotCancelled(signal); + const before = await snapshotRoots(scenario, signal); + const temporaryRoot = await mkdtemp(join(tmpdir(), "rea-process-")); + const runId = randomUUID(); + const home = join(temporaryRoot, "home"); + await import("node:fs/promises").then(({ mkdir }) => mkdir(home)); + return { temporaryRoot, runId, home, before }; +}; diff --git a/src/application/ProcessHarness.ts b/src/application/ProcessHarness.ts index 21e37e0250..986867a537 100644 --- a/src/application/ProcessHarness.ts +++ b/src/application/ProcessHarness.ts @@ -1,11 +1,5 @@ -import { createHash, randomUUID } from "node:crypto"; -import { createReadStream } from "node:fs"; -import { mkdtemp, realpath, rm } from "node:fs/promises"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; import { spawn, type IPty } from "node-pty"; import type { - FilesystemCheckpoint, InteractionEvent, ProcessCapture, ProcessExecutionPolicy, @@ -13,118 +7,35 @@ import type { ProcessScenario, TerminalFrame, } from "../domain/processCapture.js"; -import { - authorizeProcessScenario, - digestProcessCommitment, - processComparisonContract, - processScenarioCommitment, - validateProcessCapture, -} from "../domain/processCapture.js"; +import { validateProcessCapture } from "../domain/processCapture.js"; import { err, ok, type Result } from "../domain/result.js"; -import { AnalysisError } from "../domain/errors.js"; -import { PRODUCT_IDENTITY } from "../identity.js"; +import { ProcessCaptureError } from "./ProcessCaptureError.js"; + +export { ProcessCaptureError } from "./ProcessCaptureError.js"; import { startLoopbackReplay, type LoopbackReplay } from "./LoopbackReplay.js"; -import { - cleanupOwnedProcessGroup, - observeOwnedProcessGroup, -} from "./ProcessOwnership.js"; import { startProcessSampler } from "./ProcessSampling.js"; -import { snapshotRoots, type SnapshotResult } from "./FilesystemSnapshot.js"; -import { - classifyFilesystemEffects, - ProcessCheckpoints, -} from "./ProcessCheckpoints.js"; +import { snapshotRoots } from "./FilesystemSnapshot.js"; +import { ProcessCheckpoints } from "./ProcessCheckpoints.js"; import { TerminalRenderer } from "./TerminalRenderer.js"; import { startCommandShimReplay, type CommandShimReplay, } from "./CommandShimReplay.js"; +import { normalizeProcessText } from "./ProcessNormalization.js"; import { - normalizeProcessSamples, - normalizeProcessText, - redactProtocolEvents, -} from "./ProcessNormalization.js"; -import { PROCESS_PROVIDER } from "./ProcessEvidence.js"; + awaitTerminalExit, + buildCaptureResult, + captureTerminalFrames, + createRunManifest, + observeSettlement, + prepareProcessCapture, + releaseProcessResources, + resolveProcessResult, +} from "./ProcessCaptureLifecycle.js"; +import { assertNotCancelled } from "./ProcessCaptureAuthority.js"; -const isWithin = (candidate: string, root: string): boolean => - candidate === root || - candidate.startsWith(`${root.endsWith("/") ? root.slice(0, -1) : root}/`); - -const assertRealPathAuthority = async ( - scenario: ProcessScenario, - policy: ProcessExecutionPolicy, -): Promise => { - const executable = await realpath(scenario.executable); - const executableRoots = await Promise.all( - policy.executableRoots.map((root) => realpath(root)), - ); - if (!executableRoots.some((root) => isWithin(executable, root))) - throw new ProcessCaptureError( - "resolved executable is outside approved roots", - ); - const workingDirectory = await realpath(scenario.working_directory); - const workingRoots = await Promise.all( - policy.workingRoots.map((root) => realpath(root)), - ); - if (!workingRoots.some((root) => isWithin(workingDirectory, root))) - throw new ProcessCaptureError( - "resolved working directory is outside approved roots", - ); - for (const root of scenario.filesystem_roots) { - const resolvedRoot = await realpath(root); - if (!workingRoots.some((approved) => isWithin(resolvedRoot, approved))) - throw new ProcessCaptureError( - "resolved filesystem root is outside approved roots", - ); - } -}; - -/** Expected refusal or runtime failure from the process capture adapter. */ -export class ProcessCaptureError extends AnalysisError { - readonly _tag = "ProcessCaptureError"; -} - -const assertNotCancelled = (signal: AbortSignal | undefined): void => { - if (signal?.aborted === true) - throw new ProcessCaptureError("process capture was cancelled"); -}; - -/** Runtime availability of the native PTY adapter on this host. */ -export type ProcessCaptureCapability = - | { readonly available: true; readonly backend: "node-pty" } - | { - readonly available: false; - readonly backend: "node-pty"; - readonly reason: string; - }; - -/** Probe the actual native PTY seam instead of inferring support from the OS name. */ -export const probeProcessCaptureCapability = - async (): Promise => { - try { - const terminal = spawn( - process.platform === "win32" ? "cmd.exe" : "/bin/sh", - process.platform === "win32" ? ["/c", "exit", "0"] : ["-c", "exit 0"], - { - cwd: tmpdir(), - env: { HOME: tmpdir(), TERM: "xterm-256color" }, - cols: 80, - rows: 24, - name: "xterm-256color", - }, - ); - await new Promise((resolveExit) => - terminal.onExit(() => resolveExit()), - ); - return { available: true, backend: "node-pty" }; - } catch { - return { - available: false, - backend: "node-pty", - reason: "the native PTY backend could not start a probe process", - }; - } - }; +export { probeProcessCaptureCapability } from "./ProcessCaptureCapability.js"; +export type { ProcessCaptureCapability } from "./ProcessCaptureCapability.js"; interface EnvironmentOptions { readonly scenario: ProcessScenario; @@ -231,235 +142,22 @@ const eventsRequireReadiness = (scenario: ProcessScenario): boolean => (event) => event.type === "input" || event.type === "resize", ); -interface TerminalExitOptions { - readonly terminal: IPty; - readonly scenario: ProcessScenario; - readonly started: number; - readonly lastOutput: () => number; - readonly signal: AbortSignal | undefined; - readonly timers: Set; - readonly interactions: InteractionEvent[]; - readonly dispatchedEventIndexes: ReadonlySet; -} - -interface CaptureResultOptions { - readonly frames: readonly TerminalFrame[]; - readonly exit: { - readonly exitCode: number; - readonly signal?: number; - readonly reason: "exited" | "timeout" | "idle_timeout"; - }; - readonly samples: readonly ProcessSample[]; +interface StartedCaptureRuntime { readonly replay: LoopbackReplay; - readonly before: SnapshotResult; - readonly after: SnapshotResult; - readonly truncated: boolean; - readonly scenario: ProcessScenario; - readonly rootPid: number; - readonly samplingPartial: boolean; - readonly renderedFrames: ProcessCapture["rendered_frames"]; - readonly interactions: readonly InteractionEvent[]; - readonly checkpoints: readonly FilesystemCheckpoint[]; - readonly shimEvents: ProcessCapture["shim_events"]; - readonly settlement: ProcessCapture["settlement"]; - readonly manifest: ProcessCapture["manifest"]; + readonly renderer: TerminalRenderer; + readonly checkpoints: ProcessCheckpoints; + readonly shimReplay: CommandShimReplay; + readonly terminal: IPty; + readonly started: number; + readonly startedAt: Date; + readonly lastOutput: () => number; + readonly framesTruncated: () => boolean; + readonly stopSampler: () => Promise<{ readonly partial: boolean }>; } -const buildCaptureResult = (options: CaptureResultOptions): ProcessCapture => ({ - schema_version: 4, - manifest: options.manifest, - normalization: options.scenario.normalization, - frames: options.frames, - rendered_frames: [...options.renderedFrames] - .sort( - (left, right) => - left.at_ms - right.at_ms || left.sequence - right.sequence, - ) - .map((frame, sequence) => ({ ...frame, sequence })), - interaction_events: options.interactions, - exit: { - code: - options.exit.reason === "exited" && options.exit.exitCode >= 0 - ? options.exit.exitCode - : null, - signal: options.exit.signal ?? null, - reason: options.exit.reason, - }, - settlement: options.settlement, - process_samples: normalizeProcessSamples( - options.samples, - options.scenario, - options.rootPid, - ), - filesystem_checkpoints: options.checkpoints, - shim_events: options.shimEvents, - protocol_events: redactProtocolEvents( - options.replay.events, - options.scenario, - ), - files_before: options.before.files, - files_after: options.after.files, - filesystem_effects: classifyFilesystemEffects( - options.before.files, - options.after.files, - ), - truncated: options.truncated, - limitations: [ - "Process trees are sampled and may omit short-lived descendants.", - ...(options.samplingPartial - ? ["Process-tree sampling ended with an incomplete observation."] - : []), - "Filesystem observations are before/after snapshots, not syscall traces.", - "The harness does not enforce external network isolation.", - ], - residual_unknowns: [ - { - scope: "process", - reason: "Process trees are sampled and may omit short-lived descendants.", - }, - { - scope: "network", - reason: "External network isolation is not enforced by this adapter.", - }, - ], - cleanup: { - owned_process_group: "verified", - temporary_root: "removed", - }, -}); - -const hashFile = async (path: string): Promise => { - const hash = createHash("sha256"); - for await (const chunk of createReadStream(path)) hash.update(chunk); - return hash.digest("hex"); -}; - -const createRunManifest = async ( - scenario: ProcessScenario, - startedAt: Date, - completedAt: Date, -): Promise => { - const executableSha256 = await hashFile(scenario.executable); - const scenarioCommitment = processScenarioCommitment( - scenario, - executableSha256, - ); - const comparisonContract = processComparisonContract(scenario); - return { - rea_version: PRODUCT_IDENTITY.packageVersion, - provider_version: PROCESS_PROVIDER.version, - platform: process.platform, - architecture: process.arch, - pty_backend: "node-pty", - started_at: startedAt.toISOString(), - completed_at: completedAt.toISOString(), - scenario: scenarioCommitment, - comparison_contract: comparisonContract, - shim_plan: scenario.command_shims, - replay_plan: scenario.replay, - full_scenario_sha256: digestProcessCommitment(scenarioCommitment), - comparison_contract_sha256: digestProcessCommitment(comparisonContract), - executable_sha256: executableSha256, - normalization_sha256: digestProcessCommitment(scenario.normalization), - shim_plan_sha256: digestProcessCommitment(scenario.command_shims), - replay_plan_sha256: digestProcessCommitment(scenario.replay), - }; -}; - -const observeSettlement = async ( - runId: string, - processGroupIds: readonly number[], - settleMs: number, -): Promise> => { - if (process.platform === "win32") - return { state: "unverifiable", elapsed_ms: 0 }; - const started = Date.now(); - let consecutiveEmpty = 0; - let deadlineReached = false; - while (!deadlineReached) { - const observations = await Promise.all( - [...new Set(processGroupIds)].map((processGroupId) => - observeOwnedProcessGroup({ - runId, - leaderPid: processGroupId, - processGroupId, - }), - ), - ); - if (observations.some(({ state }) => state === "unverifiable")) - return { state: "unverifiable", elapsed_ms: Date.now() - started }; - if (observations.every(({ state }) => state === "empty")) { - consecutiveEmpty += 1; - if (consecutiveEmpty >= 2) - return { state: "quiesced", elapsed_ms: Date.now() - started }; - } else consecutiveEmpty = 0; - deadlineReached = Date.now() - started >= settleMs; - if (deadlineReached) break; - await new Promise((resolve) => setTimeout(resolve, 50)); - } - return { state: "alive_at_deadline", elapsed_ms: Date.now() - started }; -}; - -const awaitTerminalExit = async ({ - terminal, - scenario, - started, - lastOutput, - signal, - timers, - interactions, - dispatchedEventIndexes, -}: TerminalExitOptions): Promise<{ - exitCode: number; - signal?: number; - reason: "exited" | "timeout" | "idle_timeout" | "cancelled"; -}> => - new Promise((resolveExit) => { - // The kill caused by a deadline is observed later as an ordinary PTY exit. - // Keep the initiating lifecycle reason so comparisons distinguish a target - // exit from harness-owned timeout, idle-timeout, and cancellation cleanup. - let reason: "exited" | "timeout" | "idle_timeout" | "cancelled" = "exited"; - terminal.onExit((exit) => { - for (const [eventIndex, event] of scenario.events.entries()) { - if (dispatchedEventIndexes.has(eventIndex)) continue; - interactions.push({ - sequence: interactions.length, - scheduled_at_ms: event.at_ms, - dispatched_at_ms: Math.max(0, Date.now() - started), - type: event.type, - data: - event.type === "input" - ? "" - : event.type === "resize" - ? `${String(event.columns)}x${String(event.rows)}` - : event.signal, - outcome: "target_exited", - }); - } - for (const timer of timers) { - clearTimeout(timer); - clearInterval(timer); - } - timers.clear(); - resolveExit({ ...exit, reason }); - }); - const timeout = setInterval(() => { - if (signal?.aborted === true) { - reason = "cancelled"; - terminal.kill("SIGKILL"); - } else if (Date.now() - started >= scenario.timeout_ms) { - reason = "timeout"; - terminal.kill("SIGKILL"); - } else if (Date.now() - lastOutput() >= scenario.idle_timeout_ms) { - reason = "idle_timeout"; - terminal.kill("SIGKILL"); - } - }, 20); - timers.add(timeout); - }); - -const releaseProcessResources = async (options: { - readonly timers: ReadonlySet; +const cleanupFailedStartup = async (options: { + readonly cause: unknown; + readonly timers: Set; readonly replay: LoopbackReplay | undefined; readonly terminal: IPty | undefined; readonly renderer: TerminalRenderer | undefined; @@ -467,136 +165,264 @@ const releaseProcessResources = async (options: { readonly checkpoints: ProcessCheckpoints | undefined; readonly runId: string; readonly temporaryRoot: string; - readonly sampledProcessGroupIds: readonly number[]; -}): Promise => { - for (const timer of options.timers) clearTimeout(timer); - let failure: string | undefined; - try { - await options.replay?.close(); - } catch { - failure = "loopback replay cleanup failed"; - } - try { - await options.shimReplay?.close(); - } catch { - failure ??= "command shim replay cleanup failed"; - } - try { - await options.renderer?.dispose(); - } catch { - failure ??= "terminal renderer cleanup failed"; - } - try { - await options.checkpoints?.dispose(); - } catch { - failure ??= "filesystem checkpoint cleanup failed"; - } - if (options.terminal !== undefined && process.platform !== "win32") { - const processGroupIds = new Set([ - options.terminal.pid, - ...options.sampledProcessGroupIds, - ]); - for (const processGroupId of processGroupIds) { - const cleaned = await cleanupOwnedProcessGroup({ - runId: options.runId, - leaderPid: processGroupId, - processGroupId, - }); - if (!cleaned.cleaned) failure ??= cleaned.reason; - } - } - try { - await rm(options.temporaryRoot, { recursive: true, force: true }); - } catch { - failure ??= "temporary process root cleanup failed"; - } - return failure; -}; - -const captureTerminalFrames = (options: { - readonly terminal: IPty; - readonly scenario: ProcessScenario; - readonly frames: TerminalFrame[]; - readonly started: number; - readonly temporaryRoot: string; - readonly onOutput: () => void; - readonly onFirstOutput: () => void; - readonly renderer: TerminalRenderer; - readonly checkpoints: ProcessCheckpoints; -}): (() => boolean) => { - let outputBytes = 0; - let truncated = false; - options.terminal.onData((data) => { - options.onFirstOutput(); - options.onOutput(); - const bytes = Buffer.byteLength(data); - if ( - options.frames.length >= options.scenario.limits.frames || - outputBytes + bytes > options.scenario.limits.output_bytes - ) { - truncated = true; - return; - } - outputBytes += bytes; - const atMs = - Math.floor( - (Date.now() - options.started) / - options.scenario.normalization.time_bucket_ms, - ) * options.scenario.normalization.time_bucket_ms; - const normalized = normalizeProcessText( - data, - options.scenario, - options.temporaryRoot, - options.terminal.pid, - ); - options.frames.push({ - sequence: options.frames.length, - at_ms: atMs, - data: normalized, - }); - options.renderer.write(data, atMs); - options.checkpoints.observeTerminal(normalized); +}): Promise => { + options.terminal?.kill("SIGKILL"); + const cleanupFailure = await releaseProcessResources({ + ...options, + sampledProcessGroupIds: [], }); - return () => truncated; -}; - -const resolveProcessResult = ( - capture: ProcessCapture | undefined, - executionFailure: unknown, - cleanupFailure: string | undefined, -): ProcessCapture => { - if (executionFailure instanceof ProcessCaptureError) throw executionFailure; - if (executionFailure !== undefined) - throw new ProcessCaptureError("process capture failed", { - cause: executionFailure, - }); if (cleanupFailure !== undefined) - throw new ProcessCaptureError(cleanupFailure); - if (capture === undefined) - throw new ProcessCaptureError("process capture produced no result"); - return capture; + throw new ProcessCaptureError(cleanupFailure, { cause: options.cause }); + throw options.cause; }; -const prepareProcessCapture = async ( +const createTerminalRenderer = ( scenario: ProcessScenario, - policy: ProcessExecutionPolicy, - signal: AbortSignal | undefined, -): Promise<{ + temporaryRoot: string, + terminalPid: () => number, +): TerminalRenderer => + new TerminalRenderer({ + columns: scenario.terminal.columns, + rows: scenario.terminal.rows, + scrollback: scenario.terminal.scrollback, + maxFrames: scenario.limits.frames, + maxBytes: scenario.limits.output_bytes, + normalize: (value) => + normalizeProcessText(value, scenario, temporaryRoot, terminalPid()), + }); + +const startCaptureRuntime = async (options: { + readonly scenario: ProcessScenario; + readonly home: string; readonly temporaryRoot: string; readonly runId: string; - readonly home: string; - readonly before: SnapshotResult; -}> => { - const decision = authorizeProcessScenario(scenario, policy); - if (!decision.allowed) throw new ProcessCaptureError(decision.reason); - await assertRealPathAuthority(scenario, policy); - assertNotCancelled(signal); - const before = await snapshotRoots(scenario, signal); - const temporaryRoot = await mkdtemp(join(tmpdir(), "rea-process-")); - const runId = randomUUID(); - const home = join(temporaryRoot, "home"); - await import("node:fs/promises").then(({ mkdir }) => mkdir(home)); - return { temporaryRoot, runId, home, before }; + readonly before: Awaited>; + readonly frames: TerminalFrame[]; + readonly samples: ProcessSample[]; + readonly interactions: InteractionEvent[]; + readonly timers: Set; + readonly dispatchedEventIndexes: Set; + readonly signal?: AbortSignal; +}): Promise => { + const { scenario } = options; + let replay: LoopbackReplay | undefined; + let renderer: TerminalRenderer | undefined; + let checkpoints: ProcessCheckpoints | undefined; + let shimReplay: CommandShimReplay | undefined; + let terminal: IPty | undefined; + try { + replay = await startLoopbackReplay(scenario); + const startedAt = new Date(); + const started = Date.now(); + let lastOutput = started; + renderer = createTerminalRenderer( + scenario, + options.temporaryRoot, + () => terminal?.pid ?? -1, + ); + checkpoints = new ProcessCheckpoints( + scenario, + started, + options.before, + options.signal, + ); + shimReplay = await startCommandShimReplay( + scenario, + options.temporaryRoot, + started, + ); + terminal = spawn(scenario.executable, [...scenario.arguments], { + cwd: scenario.working_directory, + env: makeEnvironment({ ...options, replay, shimReplay }), + cols: scenario.terminal.columns, + rows: scenario.terminal.rows, + name: "xterm-256color", + }); + const startEvents = startScenarioEvents({ + ...options, + getTerminal: () => terminal, + renderer, + started, + }); + if (!eventsRequireReadiness(scenario)) startEvents(); + const framesTruncated = captureTerminalFrames({ + ...options, + terminal, + started, + onOutput: () => (lastOutput = Date.now()), + onFirstOutput: startEvents, + renderer, + checkpoints, + }); + const stopSampler = startProcessSampler( + terminal.pid, + started, + scenario.limits.processes, + options.samples, + ); + return { + replay, + renderer, + checkpoints, + shimReplay, + terminal, + started, + startedAt, + lastOutput: () => lastOutput, + framesTruncated, + stopSampler, + }; + } catch (cause: unknown) { + return cleanupFailedStartup({ + cause, + timers: options.timers, + replay, + terminal, + renderer, + shimReplay, + checkpoints, + runId: options.runId, + temporaryRoot: options.temporaryRoot, + }); + } +}; + +const normalizedShimEvents = ( + runtime: StartedCaptureRuntime, + scenario: ProcessScenario, + temporaryRoot: string, +): ProcessCapture["shim_events"] => + runtime.shimReplay.events.map((event) => ({ + ...event, + arguments: event.arguments.map((argument) => + normalizeProcessText( + argument, + scenario, + temporaryRoot, + runtime.terminal.pid, + ), + ), + working_directory: normalizeProcessText( + event.working_directory, + scenario, + temporaryRoot, + runtime.terminal.pid, + ), + })); + +const finishProcessRun = async (options: { + readonly runtime: StartedCaptureRuntime | undefined; + readonly timers: Set; + readonly runId: string; + readonly temporaryRoot: string; + readonly samples: readonly ProcessSample[]; + readonly stopSampler: () => Promise<{ readonly partial: boolean }>; + readonly capture: ProcessCapture | undefined; + readonly executionFailure: unknown; +}): Promise => { + await options.stopSampler(); + const cleanupFailure = await releaseProcessResources({ + timers: options.timers, + replay: options.runtime?.replay, + terminal: options.runtime?.terminal, + renderer: options.runtime?.renderer, + shimReplay: options.runtime?.shimReplay, + checkpoints: options.runtime?.checkpoints, + runId: options.runId, + temporaryRoot: options.temporaryRoot, + sampledProcessGroupIds: options.samples.flatMap(({ process_group_id }) => + process_group_id === null ? [] : [process_group_id], + ), + }); + let { capture, executionFailure } = options; + if (capture !== undefined) { + capture = { + ...capture, + settlement: { + ...capture.settlement, + cleanup_outcome: + capture.settlement.state === "quiesced" + ? "not_required" + : cleanupFailure === undefined + ? "cleaned" + : "failed", + }, + }; + const issue = validateProcessCapture(capture)[0]; + if (issue !== undefined) + executionFailure = new ProcessCaptureError( + `process capture validation failed: ${issue.path}: ${issue.message}`, + ); + } + return resolveProcessResult(capture, executionFailure, cleanupFailure); +}; + +const completeCapture = async (options: { + readonly scenario: ProcessScenario; + readonly runtime: StartedCaptureRuntime; + readonly runId: string; + readonly temporaryRoot: string; + readonly before: Awaited>; + readonly frames: readonly TerminalFrame[]; + readonly samples: readonly ProcessSample[]; + readonly interactions: readonly InteractionEvent[]; + readonly exit: Awaited>; + readonly signal?: AbortSignal; + readonly initiallyTruncated: boolean; +}): Promise => { + const { runtime, scenario } = options; + runtime.checkpoints.trigger("root_exit"); + const { reason } = options.exit; + if (reason === "cancelled") + throw new ProcessCaptureError("process capture was cancelled"); + const settlement = await observeSettlement( + options.runId, + [ + runtime.terminal.pid, + ...options.samples.flatMap(({ process_group_id }) => + process_group_id === null ? [] : [process_group_id], + ), + ], + scenario.settle_ms, + ); + runtime.checkpoints.trigger("settled"); + const samplingPartial = (await runtime.stopSampler()).partial; + assertNotCancelled(options.signal); + const after = await snapshotRoots(scenario, options.signal); + const renderedFrames = await runtime.renderer.frames(); + const checkpoints = await runtime.checkpoints.finish(after); + const manifest = await createRunManifest( + scenario, + runtime.startedAt, + new Date(), + ); + const truncated = + options.initiallyTruncated || + after.truncated || + runtime.replay.truncated || + runtime.shimReplay.truncated || + runtime.framesTruncated() || + checkpoints.some(({ truncated: partial }) => partial) || + samplingPartial || + runtime.renderer.truncated(); + return buildCaptureResult({ + frames: options.frames, + exit: { ...options.exit, reason }, + samples: options.samples, + replay: runtime.replay, + before: options.before, + after, + truncated, + scenario, + rootPid: runtime.terminal.pid, + samplingPartial, + renderedFrames, + interactions: options.interactions, + checkpoints, + shimEvents: normalizedShimEvents(runtime, scenario, options.temporaryRoot), + settlement: { ...settlement, cleanup_outcome: "not_required" }, + manifest, + }); }; /** Execute one authorized scenario and return bounded observations. */ @@ -612,194 +438,66 @@ const runProcessScenario = async ( ); const frames: TerminalFrame[] = []; const samples: ProcessSample[] = []; - let truncated = before.truncated; - let terminal: IPty | undefined; - let replay: LoopbackReplay | undefined; - let renderer: TerminalRenderer | undefined; - let checkpoints: ProcessCheckpoints | undefined; - let shimReplay: CommandShimReplay | undefined; - let started = 0; - let startedAt = new Date(0); - let lastOutput = 0; + let runtime: StartedCaptureRuntime | undefined; const timers = new Set(); let capture: ProcessCapture | undefined; let executionFailure: unknown; - let framesTruncated = (): boolean => false; let stopSampler = async () => ({ partial: false }); - let samplingPartial = false; - let settlement: Omit = { - state: "unverifiable", - elapsed_ms: 0, - }; const interactions: InteractionEvent[] = []; const dispatchedEventIndexes = new Set(); try { - replay = await startLoopbackReplay(scenario); - startedAt = new Date(); - started = Date.now(); - lastOutput = started; - renderer = new TerminalRenderer({ - columns: scenario.terminal.columns, - rows: scenario.terminal.rows, - scrollback: scenario.terminal.scrollback, - maxFrames: scenario.limits.frames, - maxBytes: scenario.limits.output_bytes, - normalize: (value) => - normalizeProcessText( - value, - scenario, - temporaryRoot, - terminal?.pid ?? -1, - ), - }); - checkpoints = new ProcessCheckpoints(scenario, started, before, signal); - shimReplay = await startCommandShimReplay(scenario, temporaryRoot, started); - terminal = spawn(scenario.executable, [...scenario.arguments], { - cwd: scenario.working_directory, - env: makeEnvironment({ scenario, home, replay, shimReplay, runId }), - cols: scenario.terminal.columns, - rows: scenario.terminal.rows, - name: "xterm-256color", - }); - const startEvents = startScenarioEvents({ + runtime = await startCaptureRuntime({ scenario, - getTerminal: () => terminal, - timers, - interactions, - renderer, - started, - dispatchedEventIndexes, - }); - if (!eventsRequireReadiness(scenario)) startEvents(); - framesTruncated = captureTerminalFrames({ - terminal, - scenario, - frames, - started, + home, temporaryRoot, - onOutput: () => (lastOutput = Date.now()), - onFirstOutput: startEvents, - renderer, - checkpoints, - }); - stopSampler = startProcessSampler( - terminal.pid, - started, - scenario.limits.processes, + runId, + before, + frames, samples, - ); + interactions, + timers, + dispatchedEventIndexes, + ...(signal === undefined ? {} : { signal }), + }); + stopSampler = runtime.stopSampler; const exit = await awaitTerminalExit({ - terminal, + terminal: runtime.terminal, scenario, - started, - lastOutput: () => lastOutput, + started: runtime.started, + lastOutput: runtime.lastOutput, signal, timers, interactions, dispatchedEventIndexes, }); - const exitReason = exit.reason; - checkpoints.trigger("root_exit"); - if (exitReason === "cancelled") - throw new ProcessCaptureError("process capture was cancelled"); - settlement = await observeSettlement( - runId, - [ - terminal.pid, - ...samples.flatMap(({ process_group_id }) => - process_group_id === null ? [] : [process_group_id], - ), - ], - scenario.settle_ms, - ); - checkpoints.trigger("settled"); - samplingPartial = (await stopSampler()).partial; - assertNotCancelled(signal); - const after = await snapshotRoots(scenario, signal); - const renderedFrames = await renderer.frames(); - const filesystemCheckpoints = await checkpoints.finish(after); - const manifest = await createRunManifest(scenario, startedAt, new Date()); - truncated ||= - after.truncated || - replay.truncated || - shimReplay.truncated || - framesTruncated() || - filesystemCheckpoints.some(({ truncated: partial }) => partial) || - samplingPartial; - truncated ||= renderer.truncated(); - capture = buildCaptureResult({ - frames, - exit: { ...exit, reason: exitReason }, - samples, - replay, - before, - after, - truncated, + capture = await completeCapture({ scenario, - rootPid: terminal.pid, - samplingPartial, - renderedFrames, + runtime, + runId, + temporaryRoot, + before, + frames, + samples, interactions, - checkpoints: filesystemCheckpoints, - shimEvents: shimReplay.events.map((event) => ({ - ...event, - arguments: event.arguments.map((argument) => - normalizeProcessText( - argument, - scenario, - temporaryRoot, - terminal?.pid ?? -1, - ), - ), - working_directory: normalizeProcessText( - event.working_directory, - scenario, - temporaryRoot, - terminal?.pid ?? -1, - ), - })), - settlement: { ...settlement, cleanup_outcome: "not_required" }, - manifest, + exit, + initiallyTruncated: before.truncated, + ...(signal === undefined ? {} : { signal }), }); } catch (cause: unknown) { - terminal?.kill("SIGKILL"); + runtime?.terminal.kill("SIGKILL"); executionFailure = cause; } - samplingPartial ||= (await stopSampler()).partial; - const cleanupFailure = await releaseProcessResources({ + return finishProcessRun({ + runtime, timers, - replay, - terminal, - renderer, - shimReplay, - checkpoints, runId, temporaryRoot, - sampledProcessGroupIds: samples.flatMap(({ process_group_id }) => - process_group_id === null ? [] : [process_group_id], - ), + samples, + stopSampler, + capture, + executionFailure, }); - if (capture !== undefined) { - capture = { - ...capture, - settlement: { - ...capture.settlement, - cleanup_outcome: - capture.settlement.state === "quiesced" - ? "not_required" - : cleanupFailure === undefined - ? "cleaned" - : "failed", - }, - }; - const issues = validateProcessCapture(capture); - if (issues.length > 0) - executionFailure = new ProcessCaptureError( - `process capture validation failed: ${issues[0]!.path}: ${issues[0]!.message}`, - ); - } - return resolveProcessResult(capture, executionFailure, cleanupFailure); }; /** Execute one scenario through a typed expected-failure channel. */ diff --git a/src/domain/processCapture.ts b/src/domain/processCapture.ts index eb0becbc97..03d451408f 100644 --- a/src/domain/processCapture.ts +++ b/src/domain/processCapture.ts @@ -1,448 +1,11 @@ -import { createHash } from "node:crypto"; -import { resolve } from "node:path"; -import canonicalize from "canonicalize"; import { z } from "zod"; -const positiveBudget = z.number().int().positive(); -const timedEventBase = { at_ms: z.number().int().nonnegative() }; -const environmentName = z.string().regex(/^[A-Za-z_][A-Za-z0-9_]*$/u); -const reservedEnvironment = new Set([ - "HOME", - "TERM", - "REA_PROCESS_RUN_ID", - "REA_REPLAY_HTTP_URL", - "REA_REPLAY_WEBSOCKET_URL", - "REA_SHIM_LEDGER_URL", -]); -const normalizationSchema = z.object({ - paths: z.boolean(), - pids: z.boolean(), - ports: z.boolean(), - time_bucket_ms: positiveBudget.max(60_000), - patterns: z.array( - z.object({ - pattern: z.string().max(500), - replacement: z.string().max(100), - }), - ), -}); -const checkpointNameSchema = z - .string() - .regex(/^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$/u); -const commandNameSchema = z - .string() - .regex(/^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/u); -const outputChunkSchema = z.object({ - at_ms: z.number().int().nonnegative().max(300_000), - stream: z.enum(["stdout", "stderr"]), - data: z.string().max(1_000_000), -}); +import { + LEGACY_PROCESS_CAPTURE_MESSAGE, + normalizationSchema, +} from "./processScenario.js"; -/** Actionable refusal shared by CLI, MCP, and Evidence import boundaries. */ -export const LEGACY_PROCESS_CAPTURE_MESSAGE = - "Process Capture v3 is unsupported. Re-run the scenario with this REA version using capture_process_scenario to produce Process Capture v4; captures cannot be upgraded because v4 requires new manifest and settlement evidence."; - -const canonicalSha256 = (value: unknown): string => { - const serialized = canonicalize(value); - if (serialized === undefined) - throw new TypeError("Process capture commitment is not canonical JSON"); - return createHash("sha256").update(serialized).digest("hex"); -}; - -/** Build the secret-safe scenario projection committed by Process Capture v4. */ -export const processScenarioCommitment = ( - scenario: ProcessScenario, - executableSha256?: string, -): Readonly> => ({ - ...scenario, - environment: Object.fromEntries( - Object.entries(scenario.environment).map(([name, value]) => [ - name, - scenario.secret_aliases.includes(name) ? "" : value, - ]), - ), - unknown_registry_approved: scenario.unknown_registry_approved === true, - executable_sha256: executableSha256 ?? null, -}); - -/** Project observation settings shared by authority and reconstruction. */ -export const processComparisonContract = ( - scenario: ProcessScenario, -): Readonly> => ({ - working_directory: scenario.working_directory, - inherit_environment: scenario.inherit_environment, - secret_aliases: scenario.secret_aliases, - network_access: scenario.network_access, - filesystem_roots: scenario.filesystem_roots, - terminal: scenario.terminal, - checkpoints: scenario.checkpoints, - events: scenario.events, - timeout_ms: scenario.timeout_ms, - idle_timeout_ms: scenario.idle_timeout_ms, - settle_ms: scenario.settle_ms, - limits: scenario.limits, - normalization: scenario.normalization, - command_shims: scenario.command_shims, - replay: scenario.replay, -}); - -/** Compute a canonical SHA-256 commitment independent of object key order. */ -export const digestProcessCommitment = canonicalSha256; - -/** - * Boundary schema for one explicitly approved, bounded process experiment. - * Defaults are part of the evidence contract and must remain deterministic. - */ -export const processScenarioSchema = z - .object({ - approved: z - .literal(true) - .describe( - "Explicit per-call acknowledgement that this operation launches the target", - ), - unknown_registry_approved: z - .literal(true) - .optional() - .describe("Explicit approval to record capture residuals durably"), - executable: z.string().startsWith("/"), - arguments: z.array(z.string()).max(256).default([]), - working_directory: z.string().startsWith("/"), - environment: z.record(environmentName, z.string()).default({}), - inherit_environment: z.array(environmentName).max(64).default([]), - secret_aliases: z.array(environmentName).max(64).default([]), - network_access: z.literal("host").default("host"), - filesystem_roots: z.array(z.string().startsWith("/")).max(16).default([]), - terminal: z - .object({ - columns: z.number().int().min(1).max(1_000).default(80), - rows: z.number().int().min(1).max(1_000).default(24), - scrollback: z.number().int().min(0).max(10_000).default(1_000), - }) - .default({ columns: 80, rows: 24, scrollback: 1_000 }), - checkpoints: z - .array( - z.object({ - name: checkpointNameSchema, - trigger: z.discriminatedUnion("type", [ - z.object({ - type: z.literal("time"), - at_ms: z.number().int().nonnegative().max(300_000), - }), - z.object({ - type: z.literal("terminal_literal"), - value: z.string().min(1).max(10_000), - occurrence: z.number().int().positive().max(1_000).default(1), - }), - z.object({ type: z.literal("root_exit") }), - z.object({ type: z.literal("settled") }), - ]), - }), - ) - .max(64) - .default([]), - command_shims: z - .array( - z.object({ - name: commandNameSchema, - routes: z - .array( - z.object({ - arguments: z.array(z.string()).max(256), - outputs: z.array(outputChunkSchema).max(1_000).default([]), - termination: z.discriminatedUnion("type", [ - z.object({ - type: z.literal("exit"), - code: z.number().int().min(0).max(255), - }), - z.object({ - type: z.literal("signal"), - signal: z.enum(["SIGINT", "SIGTERM", "SIGKILL"]), - }), - ]), - max_calls: z.number().int().positive().max(100).default(1), - }), - ) - .min(1) - .max(100), - }), - ) - .max(32) - .default([]), - events: z - .array( - z.discriminatedUnion("type", [ - z.object({ - ...timedEventBase, - type: z.literal("input"), - data: z.string(), - sensitive: z.boolean().default(false), - }), - z.object({ - ...timedEventBase, - type: z.literal("resize"), - columns: z.number().int().min(1).max(1_000), - rows: z.number().int().min(1).max(1_000), - }), - z.object({ - ...timedEventBase, - type: z.literal("signal"), - signal: z.enum(["SIGINT", "SIGTERM", "SIGKILL"]), - }), - ]), - ) - .max(1_000) - .default([]), - timeout_ms: positiveBudget.max(300_000).default(30_000), - idle_timeout_ms: positiveBudget.max(300_000).default(30_000), - settle_ms: z.number().int().nonnegative().max(10_000).default(100), - limits: z - .object({ - output_bytes: positiveBudget.max(10_000_000).default(1_000_000), - frames: positiveBudget.max(100_000).default(10_000), - files: positiveBudget.max(100_000).default(10_000), - file_bytes: positiveBudget.max(100_000_000).default(10_000_000), - processes: positiveBudget.max(10_000).default(1_000), - protocol_events: positiveBudget.max(100_000).default(10_000), - protocol_body_bytes: positiveBudget.max(10_000_000).default(1_000_000), - connections: positiveBudget.max(1_000).default(100), - filesystem_depth: z.number().int().min(1).max(64).default(16), - }) - .default({ - output_bytes: 1_000_000, - frames: 10_000, - files: 10_000, - file_bytes: 10_000_000, - processes: 1_000, - protocol_events: 10_000, - protocol_body_bytes: 1_000_000, - connections: 100, - filesystem_depth: 16, - }), - normalization: z - .object({ - paths: z.boolean().default(true), - pids: z.boolean().default(true), - ports: z.boolean().default(true), - time_bucket_ms: positiveBudget.max(60_000).default(10), - patterns: z - .array( - z.object({ - pattern: z.string().max(500), - replacement: z.string().max(100), - }), - ) - .max(32) - .default([]), - }) - .default({ - paths: true, - pids: true, - ports: true, - time_bucket_ms: 10, - patterns: [], - }), - replay: z - .object({ - http: z - .array( - z.object({ - method: z.string().max(16), - path: z.string().startsWith("/"), - status: z.number().int().min(100).max(599), - body: z.string().max(1_000_000), - request_headers: z.record(z.string(), z.string()).default({}), - request_body: z.string().max(1_000_000).optional(), - response_headers: z.record(z.string(), z.string()).default({}), - delay_ms: z.number().int().nonnegative().max(30_000).default(0), - disconnect: z.boolean().default(false), - max_calls: z.number().int().min(1).max(100).default(1), - }), - ) - .max(100) - .default([]), - websocket_messages: z - .array(z.string().max(1_000_000)) - .max(100) - .default([]), - websocket_connections: z - .array( - z.object({ - messages: z - .array( - z.object({ - data: z.string().max(1_000_000), - delay_ms: z - .number() - .int() - .nonnegative() - .max(30_000) - .default(0), - }), - ) - .max(100), - disconnect_after: z.boolean().default(false), - }), - ) - .max(100) - .default([]), - }) - .default({ - http: [], - websocket_messages: [], - websocket_connections: [], - }), - }) - .strict() - .superRefine((scenario, context) => { - for (let index = 1; index < scenario.events.length; index += 1) { - if (scenario.events[index]!.at_ms < scenario.events[index - 1]!.at_ms) { - context.addIssue({ - code: "custom", - message: "events must be ordered by at_ms", - path: ["events", index], - }); - } - } - const secrets = new Set(scenario.secret_aliases); - for (const alias of secrets) { - if (!(alias in scenario.environment)) { - context.addIssue({ - code: "custom", - message: "secret alias has no environment value", - path: ["secret_aliases"], - }); - } - } - const explicit = new Set(Object.keys(scenario.environment)); - for (const name of [...explicit, ...scenario.inherit_environment]) { - if (reservedEnvironment.has(name)) { - context.addIssue({ - code: "custom", - message: `${name} is reserved by the process adapter`, - path: ["environment", name], - }); - } - } - for (const name of scenario.inherit_environment) { - if (explicit.has(name)) { - context.addIssue({ - code: "custom", - message: "an environment name cannot be explicit and inherited", - path: ["inherit_environment"], - }); - } - } - for (const event of scenario.events) { - if (event.at_ms > scenario.timeout_ms) { - context.addIssue({ - code: "custom", - message: "event occurs after the scenario timeout", - path: ["events"], - }); - } - } - const checkpointNames = scenario.checkpoints.map(({ name }) => name); - for (const [index, name] of checkpointNames.entries()) { - if (name === "before" || name === "after_settlement") - context.addIssue({ - code: "custom", - message: "checkpoint name is reserved by the capture lifecycle", - path: ["checkpoints", index, "name"], - }); - } - if (new Set(checkpointNames).size !== checkpointNames.length) - context.addIssue({ - code: "custom", - message: "checkpoint names must be unique", - path: ["checkpoints"], - }); - const shimNames = scenario.command_shims.map(({ name }) => name); - if (new Set(shimNames).size !== shimNames.length) - context.addIssue({ - code: "custom", - message: "command shim names must be unique", - path: ["command_shims"], - }); - }); - -/** Parsed instructions and resource bounds for one process experiment. */ -export type ProcessScenario = z.infer; - -/** Parse untrusted process scenario input and apply safe default budgets. */ -export const parseProcessScenario = (input: unknown): ProcessScenario => - processScenarioSchema.parse(input); - -/** Operator-owned ceiling applied in addition to per-scenario approval. */ -export interface ProcessExecutionPolicy { - readonly enabled: boolean; - readonly executableRoots: readonly string[]; - readonly workingRoots: readonly string[]; - readonly allowedEnvironment: readonly string[]; - readonly allowExternalNetwork: boolean; -} - -/** Explicit authorization result; denial reasons are safe to show callers. */ -export type ProcessPolicyDecision = - | { readonly allowed: true } - | { readonly allowed: false; readonly reason: string }; - -const isWithin = (candidate: string, root: string): boolean => - resolve(candidate) === resolve(root) || - resolve(candidate).startsWith( - `${resolve(root)}${resolve(root).endsWith("/") ? "" : "/"}`, - ); - -/** Evaluate scenario authority before any process or filesystem side effect occurs. */ -export const authorizeProcessScenario = ( - scenario: ProcessScenario, - policy: ProcessExecutionPolicy, -): ProcessPolicyDecision => { - if (!policy.enabled) - return { allowed: false, reason: "process capture is disabled" }; - if (scenario.network_access === "host" && !policy.allowExternalNetwork) - return { - allowed: false, - reason: "host network access is not approved by operator policy", - }; - if ( - !policy.executableRoots.some((root) => isWithin(scenario.executable, root)) - ) { - return { allowed: false, reason: "executable is outside approved roots" }; - } - if ( - !policy.workingRoots.some((root) => - isWithin(scenario.working_directory, root), - ) - ) { - return { - allowed: false, - reason: "working directory is outside approved roots", - }; - } - const requestedNames = [ - ...Object.keys(scenario.environment), - ...scenario.inherit_environment, - ]; - if ( - requestedNames.some((name) => !policy.allowedEnvironment.includes(name)) - ) { - return { - allowed: false, - reason: "scenario requests an environment variable not allowed by policy", - }; - } - if ( - scenario.filesystem_roots.some( - (path) => !policy.workingRoots.some((root) => isWithin(path, root)), - ) - ) { - return { - allowed: false, - reason: "filesystem root is outside approved roots", - }; - } - return { allowed: true }; -}; +export * from "./processScenario.js"; /** Normalized raw PTY chunk, preserving transport-level output differences. */ export interface TerminalFrame { @@ -761,154 +324,9 @@ export const processCaptureSchema: z.ZodType = z.object({ }), }); -/** One pure semantic validation failure in an otherwise shaped v4 capture. */ -export interface ProcessCaptureValidationIssue { - readonly path: string; - readonly message: string; -} - -const orderedBy = ( - values: readonly Value[], - project: (value: Value) => number, -): boolean => - values.every( - (value, index) => - index === 0 || project(value) >= project(values[index - 1]!), - ); - -const orderedTimestamps = ( - values: readonly { readonly at_ms: number }[], -): boolean => orderedBy(values, ({ at_ms }) => at_ms); - -/** Recompute v4 commitments and cross-field invariants without side effects. */ -export const validateProcessCapture = ( - capture: ProcessCapture, -): readonly ProcessCaptureValidationIssue[] => { - const issues: ProcessCaptureValidationIssue[] = []; - const require = (condition: boolean, path: string, message: string): void => { - if (!condition) issues.push({ path, message }); - }; - const { manifest } = capture; - require(manifest.scenario.executable_sha256 === - manifest.executable_sha256, "manifest.executable_sha256", "executable commitment does not match the scenario projection"); - const commitments = [ - ["full_scenario_sha256", manifest.scenario], - ["comparison_contract_sha256", manifest.comparison_contract], - ["normalization_sha256", capture.normalization], - ["shim_plan_sha256", manifest.shim_plan], - ["replay_plan_sha256", manifest.replay_plan], - ] as const; - for (const [field, value] of commitments) - require(manifest[field] === - canonicalSha256( - value, - ), `manifest.${field}`, "commitment does not match its canonical value"); - require(Date.parse(manifest.started_at) <= - Date.parse( - manifest.completed_at, - ), "manifest.completed_at", "completion precedes start"); - const sequences = [ - ["frames", capture.frames], - ["rendered_frames", capture.rendered_frames], - ["interaction_events", capture.interaction_events], - ["shim_events", capture.shim_events], - ["protocol_events", capture.protocol_events], - ] as const; - for (const [name, values] of sequences) { - require(values.every( - ({ sequence }, index) => sequence === index, - ), name, "sequence values must be contiguous from zero"); - } - for (const [name, values] of [ - ["frames", capture.frames], - ["rendered_frames", capture.rendered_frames], - ["process_samples", capture.process_samples], - ["filesystem_checkpoints", capture.filesystem_checkpoints], - ["shim_events", capture.shim_events], - ["protocol_events", capture.protocol_events], - ] as const) - require(orderedTimestamps(values), name, "timestamps must be ordered"); - const checkpointNames = capture.filesystem_checkpoints.map( - ({ name }) => name, - ); - require(new Set(checkpointNames).size === - checkpointNames.length, "filesystem_checkpoints", "checkpoint names must be unique"); - require(checkpointNames[0] === - "before", "filesystem_checkpoints", "first checkpoint must be before"); - require(checkpointNames.at(-1) === - "after_settlement", "filesystem_checkpoints", "last checkpoint must be after_settlement"); - require(!capture.filesystem_checkpoints.some(({ truncated }) => truncated) || - capture.truncated, "truncated", "checkpoint truncation must propagate to the capture"); - require(capture.exit.reason === "exited" || - capture.exit.code === - null, "exit", "deadline termination cannot declare a normal exit code"); - require(capture.settlement.state !== "quiesced" || - capture.settlement.cleanup_outcome === - "not_required", "settlement.cleanup_outcome", "quiesced settlement cannot require cleanup"); - require(capture.settlement.state === "quiesced" || - capture.settlement.cleanup_outcome !== - "not_required", "settlement.cleanup_outcome", "non-quiesced settlement must report cleanup"); - const shimPlans = manifest.shim_plan.flatMap((shim) => - typeof shim === "object" && - shim !== null && - "name" in shim && - "routes" in shim && - typeof shim.name === "string" && - Array.isArray(shim.routes) - ? shim.routes.map((route, routeIndex) => ({ - command: shim.name as string, - route, - routeIndex, - })) - : [], - ); - for (const event of capture.shim_events) { - const declaredRoute = shimPlans.find( - ({ command, routeIndex }) => - command === event.command && routeIndex === event.route_index, - ); - require(event.outcome === "unmatched" - ? event.route_index === null - : declaredRoute !== - undefined, "shim_events", "shim event has no declared route"); - } - for (const plan of shimPlans) { - const maximum = - typeof plan.route === "object" && - plan.route !== null && - "max_calls" in plan.route && - typeof plan.route.max_calls === "number" - ? plan.route.max_calls - : 0; - require(capture.shim_events.filter( - (event) => - event.command === plan.command && - event.route_index === plan.routeIndex && - event.outcome === "matched", - ).length <= - maximum, "shim_events", "matched shim events exceed the declared route max_calls"); - } - const replayHttp = - "http" in manifest.replay_plan && Array.isArray(manifest.replay_plan.http) - ? manifest.replay_plan.http - : []; - for (const event of capture.protocol_events) - if ( - event.protocol === "http" && - event.direction === "request" && - event.outcome === "matched" - ) - require(replayHttp.some( - (route) => - typeof route === "object" && - route !== null && - "method" in route && - route.method === event.method && - "path" in route && - route.path === event.path, - ), "protocol_events", "matched HTTP event has no declared replay route"); - return issues; -}; +export { validateProcessCapture } from "./processCaptureValidation.js"; +export type { ProcessCaptureValidationIssue } from "./processCaptureValidation.js"; +import { validateProcessCapture } from "./processCaptureValidation.js"; /** Parse unknown input as v4 and reject invalid commitments or semantics. */ export const parseProcessCapture = (input: unknown): ProcessCapture => { diff --git a/src/domain/processCaptureValidation.ts b/src/domain/processCaptureValidation.ts new file mode 100644 index 0000000000..5716177462 --- /dev/null +++ b/src/domain/processCaptureValidation.ts @@ -0,0 +1,190 @@ +import type { ProcessCapture } from "./processCapture.js"; +import { canonicalProcessSha256 } from "./processScenario.js"; + +/** One pure semantic validation failure in an otherwise shaped v4 capture. */ +export interface ProcessCaptureValidationIssue { + readonly path: string; + readonly message: string; +} + +type RequireInvariant = ( + condition: boolean, + path: string, + message: string, +) => void; + +const orderedTimestamps = ( + values: readonly { readonly at_ms: number }[], +): boolean => + values.every( + (value, index) => index === 0 || value.at_ms >= values[index - 1]!.at_ms, + ); + +const validateCommitments = ( + capture: ProcessCapture, + require: RequireInvariant, +): void => { + const { manifest } = capture; + require(manifest.scenario.executable_sha256 === + manifest.executable_sha256, "manifest.executable_sha256", "executable commitment does not match the scenario projection"); + for (const [field, value] of [ + ["full_scenario_sha256", manifest.scenario], + ["comparison_contract_sha256", manifest.comparison_contract], + ["normalization_sha256", capture.normalization], + ["shim_plan_sha256", manifest.shim_plan], + ["replay_plan_sha256", manifest.replay_plan], + ] as const) + require(manifest[field] === + canonicalProcessSha256( + value, + ), `manifest.${field}`, "commitment does not match its canonical value"); + require(Date.parse(manifest.started_at) <= + Date.parse( + manifest.completed_at, + ), "manifest.completed_at", "completion precedes start"); +}; + +const validateOrdering = ( + capture: ProcessCapture, + require: RequireInvariant, +): void => { + for (const [name, values] of [ + ["frames", capture.frames], + ["rendered_frames", capture.rendered_frames], + ["interaction_events", capture.interaction_events], + ["shim_events", capture.shim_events], + ["protocol_events", capture.protocol_events], + ] as const) + require(values.every( + ({ sequence }, index) => sequence === index, + ), name, "sequence values must be contiguous from zero"); + for (const [name, values] of [ + ["frames", capture.frames], + ["rendered_frames", capture.rendered_frames], + ["process_samples", capture.process_samples], + ["filesystem_checkpoints", capture.filesystem_checkpoints], + ["shim_events", capture.shim_events], + ["protocol_events", capture.protocol_events], + ] as const) + require(orderedTimestamps(values), name, "timestamps must be ordered"); +}; + +const validateLifecycle = ( + capture: ProcessCapture, + require: RequireInvariant, +): void => { + const checkpointNames = capture.filesystem_checkpoints.map( + ({ name }) => name, + ); + require(new Set(checkpointNames).size === + checkpointNames.length, "filesystem_checkpoints", "checkpoint names must be unique"); + require(checkpointNames[0] === + "before", "filesystem_checkpoints", "first checkpoint must be before"); + require(checkpointNames.at(-1) === + "after_settlement", "filesystem_checkpoints", "last checkpoint must be after_settlement"); + require(!capture.filesystem_checkpoints.some(({ truncated }) => truncated) || + capture.truncated, "truncated", "checkpoint truncation must propagate to the capture"); + require(capture.exit.reason === "exited" || + capture.exit.code === + null, "exit", "deadline termination cannot declare a normal exit code"); + require(capture.settlement.state !== "quiesced" || + capture.settlement.cleanup_outcome === + "not_required", "settlement.cleanup_outcome", "quiesced settlement cannot require cleanup"); + require(capture.settlement.state === "quiesced" || + capture.settlement.cleanup_outcome !== + "not_required", "settlement.cleanup_outcome", "non-quiesced settlement must report cleanup"); +}; + +const shimPlans = (capture: ProcessCapture) => + capture.manifest.shim_plan.flatMap((shim) => + typeof shim === "object" && + shim !== null && + "name" in shim && + "routes" in shim && + typeof shim.name === "string" && + Array.isArray(shim.routes) + ? shim.routes.map((route, routeIndex) => ({ + command: shim.name as string, + route, + routeIndex, + })) + : [], + ); + +const validateShimEvents = ( + capture: ProcessCapture, + require: RequireInvariant, +): void => { + const plans = shimPlans(capture); + for (const event of capture.shim_events) { + const declared = plans.some( + ({ command, routeIndex }) => + command === event.command && routeIndex === event.route_index, + ); + require(event.outcome === "unmatched" + ? event.route_index === null + : declared, "shim_events", "shim event has no declared route"); + } + for (const plan of plans) { + const maximum = + typeof plan.route === "object" && + plan.route !== null && + "max_calls" in plan.route && + typeof plan.route.max_calls === "number" + ? plan.route.max_calls + : 0; + const matches = capture.shim_events.filter( + (event) => + event.command === plan.command && + event.route_index === plan.routeIndex && + event.outcome === "matched", + ).length; + require(matches <= + maximum, "shim_events", "matched shim events exceed the declared route max_calls"); + } +}; + +const validateReplayEvents = ( + capture: ProcessCapture, + require: RequireInvariant, +): void => { + const { replay_plan: replayPlan } = capture.manifest; + const routes = + "http" in replayPlan && Array.isArray(replayPlan.http) + ? replayPlan.http + : []; + for (const event of capture.protocol_events) { + if ( + event.protocol !== "http" || + event.direction !== "request" || + event.outcome !== "matched" + ) + continue; + const declared = routes.some( + (route) => + typeof route === "object" && + route !== null && + "method" in route && + route.method === event.method && + "path" in route && + route.path === event.path, + ); + require(declared, "protocol_events", "matched HTTP event has no declared replay route"); + } +}; + +/** Recompute v4 commitments and cross-field invariants without side effects. */ +export const validateProcessCapture = ( + capture: ProcessCapture, +): readonly ProcessCaptureValidationIssue[] => { + const issues: ProcessCaptureValidationIssue[] = []; + const require: RequireInvariant = (condition, path, message) => { + if (!condition) issues.push({ path, message }); + }; + validateCommitments(capture, require); + validateOrdering(capture, require); + validateLifecycle(capture, require); + validateShimEvents(capture, require); + validateReplayEvents(capture, require); + return issues; +}; diff --git a/src/domain/processComparison.ts b/src/domain/processComparison.ts index 9f25e11a9f..0c9b3e105e 100644 --- a/src/domain/processComparison.ts +++ b/src/domain/processComparison.ts @@ -197,6 +197,108 @@ const chooseFirstDivergence = ( ); }; +const assertComparable = ( + left: ProcessCapture, + right: ProcessCapture, + options: { readonly maxCaptureAgeMs?: number; readonly now?: () => number }, +): void => { + const invalid = [ + ...validateProcessCapture(left), + ...validateProcessCapture(right), + ]; + if (invalid.length > 0) + throw new TypeError(`Invalid Process Capture v4: ${invalid[0]!.path}`); + if ( + left.schema_version !== right.schema_version || + left.manifest.comparison_contract_sha256 !== + right.manifest.comparison_contract_sha256 + ) + throw new TypeError( + "Process captures have incompatible comparison contracts", + ); + if (options.maxCaptureAgeMs === undefined) return; + const maxCaptureAgeMs = options.maxCaptureAgeMs; + const now = (options.now ?? Date.now)(); + if ( + [left, right].some( + ({ manifest }) => + now - Date.parse(manifest.completed_at) > maxCaptureAgeMs, + ) + ) + throw new TypeError("Process capture exceeds max_capture_age_ms"); +}; + +const truncatedComparison = (): ProcessCaptureComparison => ({ + status: "truncated", + terminal: "truncated", + interaction: "truncated", + exit: "truncated", + filesystem: "truncated", + protocol: "truncated", + process: "truncated", + shim: "truncated", + first_divergence: { + status: "unknown", + reason: "At least one capture is truncated.", + }, + limitations: ["At least one capture is truncated."], +}); + +type ComparisonDimensions = Pick< + ProcessCaptureComparison, + | "terminal" + | "interaction" + | "exit" + | "filesystem" + | "protocol" + | "process" + | "shim" +>; + +const compareDimensions = ( + left: ProcessCapture, + right: ProcessCapture, + sameNormalization: boolean, +): ComparisonDimensions => { + const classify = ( + scope: ProcessCapture["residual_unknowns"][number]["scope"], + leftValues: readonly unknown[], + rightValues: readonly unknown[], + ): ComparisonStatus => + !sameNormalization || hasUnknown(left, scope) || hasUnknown(right, scope) + ? "unknown" + : classifyCollection(leftValues, rightValues); + return { + terminal: classify( + "terminal", + terminalObservations(left), + terminalObservations(right), + ), + interaction: classify( + "terminal", + left.interaction_events, + right.interaction_events, + ), + exit: + hasUnknown(left, "exit") || hasUnknown(right, "exit") + ? "unknown" + : JSON.stringify({ exit: left.exit, settlement: left.settlement }) === + JSON.stringify({ exit: right.exit, settlement: right.settlement }) + ? "unchanged" + : "changed", + filesystem: classify( + "filesystem", + filesystemObservations(left), + filesystemObservations(right), + ), + protocol: classify("protocol", left.protocol_events, right.protocol_events), + process: classify("process", left.process_samples, right.process_samples), + shim: sameNormalization + ? classifyCollection(left.shim_events, right.shim_events) + : "unknown", + }; +}; + /** Compare bounded captures without equating missing or incompatible evidence. */ /** * Compare two Process Capture v4 observations without treating absence as proof. @@ -212,93 +314,11 @@ export const compareProcessCaptures = ( readonly now?: () => number; } = {}, ): ProcessCaptureComparison => { - const invalid = [ - ...validateProcessCapture(left), - ...validateProcessCapture(right), - ]; - if (invalid.length > 0) - throw new TypeError(`Invalid Process Capture v4: ${invalid[0]!.path}`); - if ( - left.schema_version !== right.schema_version || - left.manifest.comparison_contract_sha256 !== - right.manifest.comparison_contract_sha256 - ) - throw new TypeError( - "Process captures have incompatible comparison contracts", - ); - if (options.maxCaptureAgeMs !== undefined) { - const maxCaptureAgeMs = options.maxCaptureAgeMs; - const now = (options.now ?? Date.now)(); - if ( - [left, right].some( - ({ manifest }) => - now - Date.parse(manifest.completed_at) > maxCaptureAgeMs, - ) - ) - throw new TypeError("Process capture exceeds max_capture_age_ms"); - } - if (left.truncated || right.truncated) { - return { - status: "truncated", - terminal: "truncated", - interaction: "truncated", - exit: "truncated", - filesystem: "truncated", - protocol: "truncated", - process: "truncated", - shim: "truncated", - first_divergence: { - status: "unknown", - reason: "At least one capture is truncated.", - }, - limitations: ["At least one capture is truncated."], - }; - } + assertComparable(left, right, options); + if (left.truncated || right.truncated) return truncatedComparison(); const sameNormalization = JSON.stringify(left.normalization) === JSON.stringify(right.normalization); - const classify = ( - scope: ProcessCapture["residual_unknowns"][number]["scope"], - leftValues: readonly unknown[], - rightValues: readonly unknown[], - ): ComparisonStatus => - !sameNormalization || hasUnknown(left, scope) || hasUnknown(right, scope) - ? "unknown" - : classifyCollection(leftValues, rightValues); - const terminal = classify( - "terminal", - terminalObservations(left), - terminalObservations(right), - ); - const interaction = classify( - "terminal", - left.interaction_events, - right.interaction_events, - ); - const exit = - hasUnknown(left, "exit") || hasUnknown(right, "exit") - ? "unknown" - : JSON.stringify({ exit: left.exit, settlement: left.settlement }) === - JSON.stringify({ exit: right.exit, settlement: right.settlement }) - ? "unchanged" - : "changed"; - const filesystem = classify( - "filesystem", - filesystemObservations(left), - filesystemObservations(right), - ); - const protocol = classify( - "protocol", - left.protocol_events, - right.protocol_events, - ); - const process = classify( - "process", - left.process_samples, - right.process_samples, - ); - const shim = sameNormalization - ? classifyCollection(left.shim_events, right.shim_events) - : "unknown"; + const dimensions = compareDimensions(left, right, sameNormalization); const observedFirstDivergence = chooseFirstDivergence([ firstCollectionDivergence( "terminal", @@ -346,21 +366,15 @@ export const compareProcessCaptures = ( : observedFirstDivergence; return { status: comparisonStatus([ - terminal, - interaction, - exit, - filesystem, - protocol, - process, - shim, + dimensions.terminal, + dimensions.interaction, + dimensions.exit, + dimensions.filesystem, + dimensions.protocol, + dimensions.process, + dimensions.shim, ]), - terminal, - interaction, - exit, - filesystem, - protocol, - process, - shim, + ...dimensions, first_divergence: firstDivergence, limitations: [ ...left.limitations, diff --git a/src/domain/processScenario.ts b/src/domain/processScenario.ts new file mode 100644 index 0000000000..5d7d5cea16 --- /dev/null +++ b/src/domain/processScenario.ts @@ -0,0 +1,445 @@ +import { createHash } from "node:crypto"; +import { resolve } from "node:path"; +import canonicalize from "canonicalize"; +import { z } from "zod"; + +const positiveBudget = z.number().int().positive(); +const timedEventBase = { at_ms: z.number().int().nonnegative() }; +const environmentName = z.string().regex(/^[A-Za-z_][A-Za-z0-9_]*$/u); +const reservedEnvironment = new Set([ + "HOME", + "TERM", + "REA_PROCESS_RUN_ID", + "REA_REPLAY_HTTP_URL", + "REA_REPLAY_WEBSOCKET_URL", + "REA_SHIM_LEDGER_URL", +]); +export const normalizationSchema = z.object({ + paths: z.boolean(), + pids: z.boolean(), + ports: z.boolean(), + time_bucket_ms: positiveBudget.max(60_000), + patterns: z.array( + z.object({ + pattern: z.string().max(500), + replacement: z.string().max(100), + }), + ), +}); +const checkpointNameSchema = z + .string() + .regex(/^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$/u); +const commandNameSchema = z + .string() + .regex(/^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/u); +const outputChunkSchema = z.object({ + at_ms: z.number().int().nonnegative().max(300_000), + stream: z.enum(["stdout", "stderr"]), + data: z.string().max(1_000_000), +}); + +/** Actionable refusal shared by CLI, MCP, and Evidence import boundaries. */ +export const LEGACY_PROCESS_CAPTURE_MESSAGE = + "Process Capture v3 is unsupported. Re-run the scenario with this REA version using capture_process_scenario to produce Process Capture v4; captures cannot be upgraded because v4 requires new manifest and settlement evidence."; + +export const canonicalProcessSha256 = (value: unknown): string => { + const serialized = canonicalize(value); + if (serialized === undefined) + throw new TypeError("Process capture commitment is not canonical JSON"); + return createHash("sha256").update(serialized).digest("hex"); +}; + +/** Build the secret-safe scenario projection committed by Process Capture v4. */ +export const processScenarioCommitment = ( + scenario: ProcessScenario, + executableSha256?: string, +): Readonly> => ({ + ...scenario, + environment: Object.fromEntries( + Object.entries(scenario.environment).map(([name, value]) => [ + name, + scenario.secret_aliases.includes(name) ? "" : value, + ]), + ), + unknown_registry_approved: scenario.unknown_registry_approved === true, + executable_sha256: executableSha256 ?? null, +}); + +/** Project observation settings shared by authority and reconstruction. */ +export const processComparisonContract = ( + scenario: ProcessScenario, +): Readonly> => ({ + working_directory: scenario.working_directory, + inherit_environment: scenario.inherit_environment, + secret_aliases: scenario.secret_aliases, + network_access: scenario.network_access, + filesystem_roots: scenario.filesystem_roots, + terminal: scenario.terminal, + checkpoints: scenario.checkpoints, + events: scenario.events, + timeout_ms: scenario.timeout_ms, + idle_timeout_ms: scenario.idle_timeout_ms, + settle_ms: scenario.settle_ms, + limits: scenario.limits, + normalization: scenario.normalization, + command_shims: scenario.command_shims, + replay: scenario.replay, +}); + +/** Compute a canonical SHA-256 commitment independent of object key order. */ +export const digestProcessCommitment = canonicalProcessSha256; + +/** + * Boundary schema for one explicitly approved, bounded process experiment. + * Defaults are part of the evidence contract and must remain deterministic. + */ +export const processScenarioSchema = z + .object({ + approved: z + .literal(true) + .describe( + "Explicit per-call acknowledgement that this operation launches the target", + ), + unknown_registry_approved: z + .literal(true) + .optional() + .describe("Explicit approval to record capture residuals durably"), + executable: z.string().startsWith("/"), + arguments: z.array(z.string()).max(256).default([]), + working_directory: z.string().startsWith("/"), + environment: z.record(environmentName, z.string()).default({}), + inherit_environment: z.array(environmentName).max(64).default([]), + secret_aliases: z.array(environmentName).max(64).default([]), + network_access: z.literal("host").default("host"), + filesystem_roots: z.array(z.string().startsWith("/")).max(16).default([]), + terminal: z + .object({ + columns: z.number().int().min(1).max(1_000).default(80), + rows: z.number().int().min(1).max(1_000).default(24), + scrollback: z.number().int().min(0).max(10_000).default(1_000), + }) + .default({ columns: 80, rows: 24, scrollback: 1_000 }), + checkpoints: z + .array( + z.object({ + name: checkpointNameSchema, + trigger: z.discriminatedUnion("type", [ + z.object({ + type: z.literal("time"), + at_ms: z.number().int().nonnegative().max(300_000), + }), + z.object({ + type: z.literal("terminal_literal"), + value: z.string().min(1).max(10_000), + occurrence: z.number().int().positive().max(1_000).default(1), + }), + z.object({ type: z.literal("root_exit") }), + z.object({ type: z.literal("settled") }), + ]), + }), + ) + .max(64) + .default([]), + command_shims: z + .array( + z.object({ + name: commandNameSchema, + routes: z + .array( + z.object({ + arguments: z.array(z.string()).max(256), + outputs: z.array(outputChunkSchema).max(1_000).default([]), + termination: z.discriminatedUnion("type", [ + z.object({ + type: z.literal("exit"), + code: z.number().int().min(0).max(255), + }), + z.object({ + type: z.literal("signal"), + signal: z.enum(["SIGINT", "SIGTERM", "SIGKILL"]), + }), + ]), + max_calls: z.number().int().positive().max(100).default(1), + }), + ) + .min(1) + .max(100), + }), + ) + .max(32) + .default([]), + events: z + .array( + z.discriminatedUnion("type", [ + z.object({ + ...timedEventBase, + type: z.literal("input"), + data: z.string(), + sensitive: z.boolean().default(false), + }), + z.object({ + ...timedEventBase, + type: z.literal("resize"), + columns: z.number().int().min(1).max(1_000), + rows: z.number().int().min(1).max(1_000), + }), + z.object({ + ...timedEventBase, + type: z.literal("signal"), + signal: z.enum(["SIGINT", "SIGTERM", "SIGKILL"]), + }), + ]), + ) + .max(1_000) + .default([]), + timeout_ms: positiveBudget.max(300_000).default(30_000), + idle_timeout_ms: positiveBudget.max(300_000).default(30_000), + settle_ms: z.number().int().nonnegative().max(10_000).default(100), + limits: z + .object({ + output_bytes: positiveBudget.max(10_000_000).default(1_000_000), + frames: positiveBudget.max(100_000).default(10_000), + files: positiveBudget.max(100_000).default(10_000), + file_bytes: positiveBudget.max(100_000_000).default(10_000_000), + processes: positiveBudget.max(10_000).default(1_000), + protocol_events: positiveBudget.max(100_000).default(10_000), + protocol_body_bytes: positiveBudget.max(10_000_000).default(1_000_000), + connections: positiveBudget.max(1_000).default(100), + filesystem_depth: z.number().int().min(1).max(64).default(16), + }) + .default({ + output_bytes: 1_000_000, + frames: 10_000, + files: 10_000, + file_bytes: 10_000_000, + processes: 1_000, + protocol_events: 10_000, + protocol_body_bytes: 1_000_000, + connections: 100, + filesystem_depth: 16, + }), + normalization: z + .object({ + paths: z.boolean().default(true), + pids: z.boolean().default(true), + ports: z.boolean().default(true), + time_bucket_ms: positiveBudget.max(60_000).default(10), + patterns: z + .array( + z.object({ + pattern: z.string().max(500), + replacement: z.string().max(100), + }), + ) + .max(32) + .default([]), + }) + .default({ + paths: true, + pids: true, + ports: true, + time_bucket_ms: 10, + patterns: [], + }), + replay: z + .object({ + http: z + .array( + z.object({ + method: z.string().max(16), + path: z.string().startsWith("/"), + status: z.number().int().min(100).max(599), + body: z.string().max(1_000_000), + request_headers: z.record(z.string(), z.string()).default({}), + request_body: z.string().max(1_000_000).optional(), + response_headers: z.record(z.string(), z.string()).default({}), + delay_ms: z.number().int().nonnegative().max(30_000).default(0), + disconnect: z.boolean().default(false), + max_calls: z.number().int().min(1).max(100).default(1), + }), + ) + .max(100) + .default([]), + websocket_messages: z + .array(z.string().max(1_000_000)) + .max(100) + .default([]), + websocket_connections: z + .array( + z.object({ + messages: z + .array( + z.object({ + data: z.string().max(1_000_000), + delay_ms: z + .number() + .int() + .nonnegative() + .max(30_000) + .default(0), + }), + ) + .max(100), + disconnect_after: z.boolean().default(false), + }), + ) + .max(100) + .default([]), + }) + .default({ + http: [], + websocket_messages: [], + websocket_connections: [], + }), + }) + .strict() + .superRefine((scenario, context) => { + for (let index = 1; index < scenario.events.length; index += 1) { + if (scenario.events[index]!.at_ms < scenario.events[index - 1]!.at_ms) { + context.addIssue({ + code: "custom", + message: "events must be ordered by at_ms", + path: ["events", index], + }); + } + } + const secrets = new Set(scenario.secret_aliases); + for (const alias of secrets) { + if (!(alias in scenario.environment)) { + context.addIssue({ + code: "custom", + message: "secret alias has no environment value", + path: ["secret_aliases"], + }); + } + } + const explicit = new Set(Object.keys(scenario.environment)); + for (const name of [...explicit, ...scenario.inherit_environment]) { + if (reservedEnvironment.has(name)) { + context.addIssue({ + code: "custom", + message: `${name} is reserved by the process adapter`, + path: ["environment", name], + }); + } + } + for (const name of scenario.inherit_environment) { + if (explicit.has(name)) { + context.addIssue({ + code: "custom", + message: "an environment name cannot be explicit and inherited", + path: ["inherit_environment"], + }); + } + } + for (const event of scenario.events) { + if (event.at_ms > scenario.timeout_ms) { + context.addIssue({ + code: "custom", + message: "event occurs after the scenario timeout", + path: ["events"], + }); + } + } + const checkpointNames = scenario.checkpoints.map(({ name }) => name); + for (const [index, name] of checkpointNames.entries()) { + if (name === "before" || name === "after_settlement") + context.addIssue({ + code: "custom", + message: "checkpoint name is reserved by the capture lifecycle", + path: ["checkpoints", index, "name"], + }); + } + if (new Set(checkpointNames).size !== checkpointNames.length) + context.addIssue({ + code: "custom", + message: "checkpoint names must be unique", + path: ["checkpoints"], + }); + const shimNames = scenario.command_shims.map(({ name }) => name); + if (new Set(shimNames).size !== shimNames.length) + context.addIssue({ + code: "custom", + message: "command shim names must be unique", + path: ["command_shims"], + }); + }); + +/** Parsed instructions and resource bounds for one process experiment. */ +export type ProcessScenario = z.infer; + +/** Parse untrusted process scenario input and apply safe default budgets. */ +export const parseProcessScenario = (input: unknown): ProcessScenario => + processScenarioSchema.parse(input); + +/** Operator-owned ceiling applied in addition to per-scenario approval. */ +export interface ProcessExecutionPolicy { + readonly enabled: boolean; + readonly executableRoots: readonly string[]; + readonly workingRoots: readonly string[]; + readonly allowedEnvironment: readonly string[]; + readonly allowExternalNetwork: boolean; +} + +/** Explicit authorization result; denial reasons are safe to show callers. */ +export type ProcessPolicyDecision = + | { readonly allowed: true } + | { readonly allowed: false; readonly reason: string }; + +const isWithin = (candidate: string, root: string): boolean => + resolve(candidate) === resolve(root) || + resolve(candidate).startsWith( + `${resolve(root)}${resolve(root).endsWith("/") ? "" : "/"}`, + ); + +/** Evaluate scenario authority before any process or filesystem side effect occurs. */ +export const authorizeProcessScenario = ( + scenario: ProcessScenario, + policy: ProcessExecutionPolicy, +): ProcessPolicyDecision => { + if (!policy.enabled) + return { allowed: false, reason: "process capture is disabled" }; + if (scenario.network_access === "host" && !policy.allowExternalNetwork) + return { + allowed: false, + reason: "host network access is not approved by operator policy", + }; + if ( + !policy.executableRoots.some((root) => isWithin(scenario.executable, root)) + ) { + return { allowed: false, reason: "executable is outside approved roots" }; + } + if ( + !policy.workingRoots.some((root) => + isWithin(scenario.working_directory, root), + ) + ) { + return { + allowed: false, + reason: "working directory is outside approved roots", + }; + } + const requestedNames = [ + ...Object.keys(scenario.environment), + ...scenario.inherit_environment, + ]; + if ( + requestedNames.some((name) => !policy.allowedEnvironment.includes(name)) + ) { + return { + allowed: false, + reason: "scenario requests an environment variable not allowed by policy", + }; + } + if ( + scenario.filesystem_roots.some( + (path) => !policy.workingRoots.some((root) => isWithin(path, root)), + ) + ) { + return { + allowed: false, + reason: "filesystem root is outside approved roots", + }; + } + return { allowed: true }; +}; diff --git a/src/server/registerProcessComparisonTool.ts b/src/server/registerProcessComparisonTool.ts index 49c8c2cc99..708859e512 100644 --- a/src/server/registerProcessComparisonTool.ts +++ b/src/server/registerProcessComparisonTool.ts @@ -11,7 +11,7 @@ import type { RecordUnknownInput } from "../domain/residualUnknown.js"; import { compareProcessCaptures, processCaptureSchema, - validateProcessCapture, + parseProcessCapture, } from "../domain/processCapture.js"; import { recordDerivedEvidence } from "./recordDerivedEvidence.js"; import { err } from "../domain/result.js"; @@ -103,7 +103,12 @@ const validateCaptureSources = ( [parsed.right_evidence_id, parsed.right], ] as const) { const evidence = session.evidenceById(evidenceId); - const result = processCaptureSchema.safeParse(evidence?.normalized_result); + let evidenceCapture: z.infer | undefined; + try { + evidenceCapture = parseProcessCapture(evidence?.normalized_result); + } catch { + evidenceCapture = undefined; + } if ( evidence === undefined || evidence.operation !== "capture_process_scenario" || @@ -113,10 +118,8 @@ const validateCaptureSources = ( evidence.provider.version !== PROCESS_PROVIDER.version || evidence.confidence !== "observed" || evidence.authority !== "controlled-replay" || - !result.success || - (result.success && validateProcessCapture(result.data).length > 0) || - validateProcessCapture(capture).length > 0 || - canonicalJson(result.data) !== canonicalJson(capture) + evidenceCapture === undefined || + canonicalJson(evidenceCapture) !== canonicalJson(capture) ) return err( new EvidenceIntegrityError(