From 599a8ee11de6b1a7fb270e669b88d78b6be982d2 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Wed, 14 Oct 2026 08:19:07 +0800 Subject: [PATCH] fix(process): preserve cleanup ownership and normalize captures once Retain unresolved process identity and launch evidence through retryable cleanup. Share capture path normalization and preserve failed filesystem observations. Bound child output and drain Windows pipes without one-chunk throttling. Reviewed fixes include #1628, #2632, #2644, #2784, #2928, #2931, #2965, #2992, and #3011. Validation: 92 focused source tests passed; four platform-specific skips. Real Windows and Hopper lifecycle coverage was not rerun. --- docs/process-capture.md | 3 + native/windows/src/process.cc | 32 +- scripts/lib/clean-linux-hopper-runtime.mjs | 92 +++- src/domain/process/processCaptureCoverage.ts | 18 +- src/domain/process/processCaptureExample.ts | 2 + .../process/processCaptureValidation.ts | 12 + src/domain/process/processComparison.ts | 36 +- .../process/processTraceComparison.test.ts | 111 +++++ src/domain/process/processTraceEvaluation.ts | 98 +++-- src/hopper/BridgeLauncher.ts | 12 +- src/hopper/HopperCleanup.test.ts | 63 ++- src/hopper/HopperCleanup.ts | 39 +- src/hopper/HopperClient.ts | 27 ++ src/hopper/HopperSessionValues.ts | 10 +- src/hopper/HopperTargetLease.test.ts | 79 ++++ src/hopper/HopperTargetLease.ts | 69 ++- src/process/ArtifactAnalysisResources.test.ts | 85 ++++ src/process/ArtifactAnalysisResources.ts | 63 ++- src/process/ExecFileBoundedOutput.test.ts | 50 +++ src/process/ExecFileBoundedOutput.ts | 73 ++++ src/process/ProcessOwnership.test.ts | 179 ++++++++ src/process/ProcessOwnership.ts | 88 +++- .../ProcessOwnership.validation.test.ts | 34 ++ src/process/ProcessOwnershipObservation.ts | 16 + src/process/ProcessStartIdentity.test.ts | 21 + src/process/ProviderProcess.ts | 12 +- src/process/ProviderProcessOutput.test.ts | 24 + src/process/WindowsAuthority.ts | 139 +++--- .../capture/FilesystemCoverage.test.ts | 39 ++ .../capture/FilesystemSnapshot.test.ts | 63 +++ src/process/capture/FilesystemSnapshot.ts | 44 +- .../capture/ProcessCaptureLifecycle.ts | 10 +- src/process/capture/ProcessHarness.ts | 15 +- .../capture/ProcessNormalization.test.ts | 37 ++ src/process/capture/ProcessNormalization.ts | 97 +++-- src/windows/WindowsOwnedProcess.test.ts | 223 +++++++++- src/windows/WindowsOwnedProcess.ts | 87 +++- .../process/cleanLinuxHopperRuntime.test.ts | 411 ++++++++++++++++++ .../boundary/process/providerProcess.test.ts | 27 +- .../hopper/bridgeBootstrapCleanup.test.ts | 58 +++ .../cleanLinuxHopperRuntime.test.ts | 6 +- ...systemSnapshotFallbackCancellation.test.ts | 118 +++++ .../hopperLeaseDescriptorPermissions.test.ts | 79 ++++ 43 files changed, 2561 insertions(+), 240 deletions(-) create mode 100644 src/process/ArtifactAnalysisResources.test.ts create mode 100644 src/process/ExecFileBoundedOutput.test.ts create mode 100644 src/process/ExecFileBoundedOutput.ts create mode 100644 tests/boundary/process/cleanLinuxHopperRuntime.test.ts create mode 100644 tests/boundary/providers/hopper/bridgeBootstrapCleanup.test.ts create mode 100644 tests/process-global/filesystemSnapshotFallbackCancellation.test.ts create mode 100644 tests/process-global/hopperLeaseDescriptorPermissions.test.ts diff --git a/docs/process-capture.md b/docs/process-capture.md index cc163f715..5b2bbbeeb 100644 --- a/docs/process-capture.md +++ b/docs/process-capture.md @@ -380,6 +380,9 @@ Every capture requires `truncation_details`, with separate accounting for: When a directory exceeds that capacity, the retained subset follows filesystem enumeration order and is then sorted; it is explicitly incomplete and cannot establish that an omitted path is absent. + An unreadable directory is recorded in `enumeration_omissions` with its + root-aliased path and operating-system error code. Entries observed before + that directory remain available, while descendants beneath it stay unknown. Each retained regular file without a digest has an aliased path, size, remaining budget and reason: `file_bytes_budget`, `file_changed_or_short_read`, or `file_unavailable`. Its `system_code` is diff --git a/native/windows/src/process.cc b/native/windows/src/process.cc index 42ee9582b..55e415345 100644 --- a/native/windows/src/process.cc +++ b/native/windows/src/process.cc @@ -214,18 +214,32 @@ std::unique_ptr spawnProcess(const std::wstring& command, const std::ws } static napi_value readOutput(napi_env env, Handle& read, bool& ended) { - std::array bytes; - DWORD available = 0, count = 0; + constexpr size_t kMaximumReadPerPoll = 1024 * 1024; + std::array chunk; + std::vector bytes; if (!read.valid()) ended = true; - else if (!PeekNamedPipe(read.get(), nullptr, 0, nullptr, &available, nullptr)) { - require(GetLastError() == ERROR_BROKEN_PIPE, "Observe owned process output failed"); - read.reset(); ended = true; - } else if (available > 0) { - require(ReadFile(read.get(), bytes.data(), std::min(available, bytes.size()), &count, nullptr), - "Read owned process output failed"); + else { + while (bytes.size() < kMaximumReadPerPoll) { + DWORD available = 0; + if (!PeekNamedPipe(read.get(), nullptr, 0, nullptr, &available, nullptr)) { + require(GetLastError() == ERROR_BROKEN_PIPE, + "Observe owned process output failed"); + read.reset(); + ended = true; + break; + } + if (available == 0) break; + const auto requested = static_cast(std::min( + {available, chunk.size(), kMaximumReadPerPoll - bytes.size()})); + DWORD count = 0; + require(ReadFile(read.get(), chunk.data(), requested, &count, nullptr) && + count > 0, + "Read owned process output failed"); + bytes.insert(bytes.end(), chunk.begin(), chunk.begin() + count); + } } napi_value result; - check(napi_create_buffer_copy(env, count, bytes.data(), nullptr, &result)); + check(napi_create_buffer_copy(env, bytes.size(), bytes.data(), nullptr, &result)); return result; } diff --git a/scripts/lib/clean-linux-hopper-runtime.mjs b/scripts/lib/clean-linux-hopper-runtime.mjs index 4d6bc8163..41ccfed6c 100644 --- a/scripts/lib/clean-linux-hopper-runtime.mjs +++ b/scripts/lib/clean-linux-hopper-runtime.mjs @@ -1,8 +1,36 @@ -import { execFile } from "node:child_process"; -import { promisify } from "node:util"; -import { readdir, stat } from "node:fs/promises"; +import { readFile, readdir, stat } from "node:fs/promises"; +import { basename } from "node:path"; -const execFileAsync = promisify(execFile); +// CPython's interface option ends interpreter parsing; following values are +// script arguments. -W/-X consume either an attached value or the next operand. +// https://docs.python.org/3.12/using/cmdline.html +const pythonScriptOperand = (arguments_) => { + for (let index = 1; index < arguments_.length; index++) { + const option = arguments_[index]; + if (option === "--") return arguments_[index + 1]; + if (option === "-" || option === "") return undefined; + if (!option.startsWith("-")) return option; + if (option === "--check-hash-based-pycs") { + index++; + continue; + } + if (/^--(?:help(?:-env|-xoptions|-all)?|version)$/u.test(option)) + return undefined; + if (option.startsWith("--")) + throw new Error(`Unknown Python helper invocation option: ${option}`); + for (let character = 1; character < option.length; character++) { + const flag = option[character]; + if ("cmh?V".includes(flag)) return undefined; + if (flag === "W" || flag === "X") { + if (character === option.length - 1) index++; + break; + } + if (!"bBdEiIOPqRsSuvx".includes(flag)) + throw new Error(`Unknown Python helper invocation option: ${option}`); + } + } + return undefined; +}; /** * Observe leftover demo processes on a dedicated clean Linux runner, including @@ -18,9 +46,6 @@ export const assertCleanLinuxHopperRuntime = async ( ], procRoot = "/proc", ) => { - const { stdout } = await execFileAsync("ps", ["-ax", "-o", "command="]); - if (/\/opt\/hopper\/bin\/Hopper|\bXvfb\b|hopper-demo-x11\.py/u.test(stdout)) - throw new Error("Clean Linux runner retained a Hopper demo process"); const launchers = []; for (const path of launcherPaths) { try { @@ -29,20 +54,55 @@ export const assertCleanLinuxHopperRuntime = async ( if (error?.code !== "ENOENT") throw error; } } - if (launchers.length === 0) return; // /proc exposes the executed file even when argv[0] names a removed symlink. // Compare file identity, not a display path or a process name. let inspectionError; for (const entry of await readdir(procRoot, { withFileTypes: true })) { if (!entry.isDirectory() || !/^\d+$/u.test(entry.name)) continue; - let executable; + let retainedRuntime; try { - executable = await stat(`${procRoot}/${entry.name}/exe`, { - bigint: true, - }); + // procfs preserves argument boundaries. A caller merely mentioning a + // runtime in its script text or other arguments is not that runtime. + const commandLine = await readFile( + `${procRoot}/${entry.name}/cmdline`, + "utf8", + ); + const arguments_ = commandLine.split("\0"); + // The kernel command field identifies renamed runtime processes such + // as Xvfb. Keep it separate from arguments so ordinary script text does + // not count as a retained runtime. + const kernelCommand = ( + await readFile(`${procRoot}/${entry.name}/comm`, "utf8") + ).replace(/\n$/u, ""); + if ( + commandLine.length === 0 && + /^python3(?:\.\d+)?$/u.test(kernelCommand) + ) + throw new Error( + "Cannot identify a command-less Python process on the clean Linux runner", + ); + retainedRuntime = + kernelCommand === "Hopper" || + kernelCommand === "Xvfb" || + arguments_[0] === "/opt/hopper/bin/Hopper" || + basename(arguments_[0] ?? "") === "Xvfb" || + (/^python3(?:\.\d+)?$/u.test(basename(arguments_[0] ?? "")) && + arguments_.some( + (value) => basename(value) === "hopper-demo-x11.py", + ) && + basename(pythonScriptOperand(arguments_) ?? "") === + "hopper-demo-x11.py"); + if (!retainedRuntime && launchers.length > 0) { + const executable = await stat(`${procRoot}/${entry.name}/exe`, { + bigint: true, + }); + retainedRuntime = launchers.some( + ({ dev, ino }) => dev === executable.dev && ino === executable.ino, + ); + } } catch (error) { if (error?.code === "ENOENT" || error?.code === "ESRCH") continue; - // Other users' executables may be unreadable. The setup lane launches + // Other users' process facts may be unreadable. The setup lane launches // providers as the caller; unreadable caller-owned processes fail closed. if (error?.code === "EACCES") { const owner = await stat(`${procRoot}/${entry.name}`).catch((cause) => { @@ -56,11 +116,7 @@ export const assertCleanLinuxHopperRuntime = async ( inspectionError ??= error; continue; } - if ( - launchers.some( - ({ dev, ino }) => dev === executable.dev && ino === executable.ino, - ) - ) + if (retainedRuntime) throw new Error("Clean Linux runner retained a Hopper demo process"); } if (inspectionError !== undefined) throw inspectionError; diff --git a/src/domain/process/processCaptureCoverage.ts b/src/domain/process/processCaptureCoverage.ts index 72bbb9fb1..1a0e73ee5 100644 --- a/src/domain/process/processCaptureCoverage.ts +++ b/src/domain/process/processCaptureCoverage.ts @@ -16,7 +16,19 @@ export const filesystemCoverageSchema = z.strictObject({ depth_limit: z.number().int().nonnegative(), enumeration_truncated: z.boolean(), enumeration_reasons: z.array( - z.enum(["files_limit", "depth_limit", "identity_changed"]), + z.enum([ + "files_limit", + "depth_limit", + "identity_changed", + "directory_unavailable", + ]), + ), + enumeration_omissions: z.array( + z.strictObject({ + path: z.string(), + reason: z.literal("directory_unavailable"), + system_code: z.string().nullable(), + }), ), hash_budget_bytes: z.number().int().nonnegative(), hashed_bytes: z.number().int().nonnegative(), @@ -66,7 +78,9 @@ export type ProcessSamplingCoverage = const omitted = (retention: TerminalRetention): boolean => retention.observed_frames > retention.retained_frames; const filesystemTruncated = (coverage: FilesystemCoverage): boolean => - coverage.enumeration_truncated || coverage.hash_omissions.length > 0; + coverage.enumeration_truncated || + coverage.enumeration_omissions.length > 0 || + coverage.hash_omissions.length > 0; /** Derive the aggregate truncation flag from independent producer observations. */ export const hasCaptureTruncation = ( diff --git a/src/domain/process/processCaptureExample.ts b/src/domain/process/processCaptureExample.ts index c8dbff897..8259f3850 100644 --- a/src/domain/process/processCaptureExample.ts +++ b/src/domain/process/processCaptureExample.ts @@ -79,6 +79,7 @@ export const EMPTY_PROCESS_CAPTURE_EXAMPLE = { depth_limit: 16, enumeration_truncated: false, enumeration_reasons: [], + enumeration_omissions: [], hash_budget_bytes: 10_000_000, hashed_bytes: 0, hash_omissions: [], @@ -88,6 +89,7 @@ export const EMPTY_PROCESS_CAPTURE_EXAMPLE = { depth_limit: 16, enumeration_truncated: false, enumeration_reasons: [], + enumeration_omissions: [], hash_budget_bytes: 10_000_000, hashed_bytes: 0, hash_omissions: [], diff --git a/src/domain/process/processCaptureValidation.ts b/src/domain/process/processCaptureValidation.ts index 537f6f92f..53a783798 100644 --- a/src/domain/process/processCaptureValidation.ts +++ b/src/domain/process/processCaptureValidation.ts @@ -334,6 +334,9 @@ const validateCoverage = ( .filter((file) => file.type === "file" && file.sha256 === null) .map((file) => [file.path, file]), ); + const directories = new Set( + files.filter((file) => file.type === "directory").map(({ path }) => path), + ); require(coverage.hash_omissions.length === unhashedFiles.size && new Set(coverage.hash_omissions.map(({ path }) => path)).size === unhashedFiles.size && @@ -345,6 +348,14 @@ const validateCoverage = ( ? omission.system_code !== null : omission.system_code === null), ), `truncation_details.${name}.hash_omissions`, "every retained regular file without a digest must have one matching omission reason"); + require(new Set(coverage.enumeration_omissions.map(({ path }) => path)) + .size === coverage.enumeration_omissions.length && + (coverage.enumeration_omissions.length === 0 || + coverage.enumeration_reasons.includes("directory_unavailable")) && + coverage.enumeration_omissions.every( + (omission) => + directories.has(omission.path) && omission.system_code !== null, + ), `truncation_details.${name}.enumeration_omissions`, "every directory enumeration omission must identify one retained directory and its system error"); require(coverage.hashed_bytes === files .filter((file) => file.type === "file" && file.sha256 !== null) @@ -356,6 +367,7 @@ const validateCoverage = ( coverage.enumeration_reasons.length > 0 && checkpoint?.truncated === (coverage.enumeration_truncated || + coverage.enumeration_omissions.length > 0 || coverage.hash_omissions.length > 0), `truncation_details.${name}`, "filesystem checkpoint truncation must match enumeration and digest coverage"); } diff --git a/src/domain/process/processComparison.ts b/src/domain/process/processComparison.ts index eafcd4fd8..06e134061 100644 --- a/src/domain/process/processComparison.ts +++ b/src/domain/process/processComparison.ts @@ -1,7 +1,10 @@ import canonicalize from "canonicalize"; import { z } from "zod"; import { processSourceTruncated } from "./processCaptureCoverage.js"; -import type { ProcessObservationSource } from "./processObservation.js"; +import type { + ProcessObservationLocation, + ProcessObservationSource, +} from "./processObservation.js"; import type { ProcessCapture } from "./processCaptureParsing.js"; import { comparableTerminalFrame } from "./processObservation.js"; @@ -299,6 +302,18 @@ const differingContractFields = ( ) .sort(); +const LOCATION_SOURCES = { + frames: "terminal_raw", + rendered_frames: "terminal_rendered", + interaction_events: "interaction", + lifecycle: "lifecycle", + process_samples: "process", + filesystem_checkpoints: "filesystem", +} as const satisfies Record< + ProcessObservationLocation["collection"], + ProcessObservationSource +>; + const applyTraceVerdict = ( dimensions: ComparisonDimensions, specification: ProcessTraceSpecification, @@ -314,6 +329,20 @@ const applyTraceVerdict = ( : "changed"; const sources = new Set(specification.events.map(({ source }) => source)); const covered = dimensionsForTraceSources(sources); + // A different verdict names the failing events and observations; attribute + // it only to their dimensions so equal observations keep their exact status. + const diagnosticSources = new Set([ + ...specification.events + .filter(({ id }) => trace.diagnostic?.event_ids.includes(id) === true) + .map(({ source }) => source), + ...(trace.diagnostic?.locations ?? []).map( + ({ collection }) => LOCATION_SOURCES[collection], + ), + ]); + const attributed = + diagnosticSources.size > 0 + ? dimensionsForTraceSources(diagnosticSources) + : covered; const statusFor = ( dimension: (typeof PROCESS_COMPARISON_DIMENSIONS)[number], ): ComparisonStatus => { @@ -321,7 +350,10 @@ const applyTraceVerdict = ( if (!covered.has(dimension)) return exact; if (trace.verdict === "nonconforming") return exact; if (trace.verdict === "different") - return traceShowsDifference ? "changed" : exact; + return traceShowsDifference && + (exact !== "unchanged" || attributed.has(dimension)) + ? "changed" + : exact; if ( trace.verdict === "unknown" && ["changed", "added", "removed"].includes(exact) diff --git a/src/domain/process/processTraceComparison.test.ts b/src/domain/process/processTraceComparison.test.ts index 2207352e3..07113bfad 100644 --- a/src/domain/process/processTraceComparison.test.ts +++ b/src/domain/process/processTraceComparison.test.ts @@ -11,6 +11,7 @@ import { values, } from "./processTraceComparison.fixture.js"; +import { compareProcessCaptures } from "./processComparison.js"; import { compareProcessTraces } from "./processTraceComparison.js"; describe("generic capture trace comparison", () => { it("requires explicit partial ordering and never infers causality from timestamps", () => { @@ -285,3 +286,113 @@ describe("unknown process trace evidence", () => { ).toThrow("event_journal"); }); }); + +describe("process trace dimension attribution", () => { + const traced = ( + left: ReturnType, + right: ReturnType, + ) => + compareProcessCaptures(left, right, { + traceSpecification: partialSpecification(), + }); + const make = (data: string, command: string) => + capture({ + ...values(["terminal", "process"]), + frames: [{ ...terminal, data }], + process_samples: [{ ...processStarted, command }], + }); + + it("keeps equal dimensions unchanged when one side fails a predicate", () => { + expect( + traced(make("Ready", "worker"), make("Different", "worker")), + ).toMatchObject({ + trace: { verdict: "different" }, + terminal: "changed", + process: "unchanged", + }); + expect( + traced(make("Ready", "worker"), make("Ready", "other-worker")), + ).toMatchObject({ + trace: { verdict: "different" }, + terminal: "unchanged", + process: "changed", + }); + }); + + it("still attributes an ordering difference to both ordered dimensions", () => { + expect( + traced( + capture(values(["terminal", "process"])), + capture(values(["process", "terminal"])), + ), + ).toMatchObject({ + trace: { verdict: "different" }, + terminal: "changed", + process: "changed", + }); + }); + + it("attributes every independent failed ordering edge", () => { + const terminals = [ + terminal, + { ...terminal, sequence: 1, data: "Ready again" }, + ]; + const workers = [processStarted, { ...processStarted, at_ms: 2, pid: 2 }]; + const events = ["ready1", "worker1", "ready2", "worker2"] as const; + const make = (order: readonly (typeof events)[number][]) => + capture({ + frames: terminals, + process_samples: workers, + event_journal: order.map((event) => ({ + collection: event.startsWith("ready") + ? ("frames" as const) + : ("process_samples" as const), + index: Number(event.slice(-1)) - 1, + })), + }); + const left = make(events); + const right = make(["worker1", "ready1", "worker2", "ready2"]); + const specification: ProcessTraceSpecification = { + events: [ + ...terminals.map((frame, index) => ({ + id: `ready${String(index + 1)}`, + source: "terminal_raw" as const, + exact: frame, + cardinality: { kind: "required" as const }, + })), + ...workers.map((sample, index) => ({ + id: `worker${String(index + 1)}`, + source: "process" as const, + exact: sample, + cardinality: { kind: "required" as const }, + })), + ], + language: { + kind: "partial_order", + happens_before: [ + { before: "ready1", after: "worker1" }, + { before: "ready2", after: "worker2" }, + ], + not_before: [], + unordered_groups: [ + { events: ["ready1", "ready2"] }, + { events: ["worker1", "worker2"] }, + { events: ["ready1", "worker2"] }, + { events: ["worker1", "ready2"] }, + ], + prefix: [], + suffix: [], + }, + }; + + expect( + compareProcessCaptures(left, right, { + traceSpecification: specification, + }), + ).toMatchObject({ + trace: { verdict: "different" }, + terminal: "changed", + process: "changed", + }); + }); +}); diff --git a/src/domain/process/processTraceEvaluation.ts b/src/domain/process/processTraceEvaluation.ts index 7c232a480..4cf42d8b7 100644 --- a/src/domain/process/processTraceEvaluation.ts +++ b/src/domain/process/processTraceEvaluation.ts @@ -248,38 +248,33 @@ const evaluateOrderingConstraints = ( language: PartialOrderLanguage, ): { readonly failure: EvaluatedSide | null; readonly satisfied: string[] } => { const satisfied: string[] = []; + const failures: FailureInput[] = []; for (const edge of language.happens_before) { const before = trace.locationsById.get(edge.before) ?? []; const after = trace.locationsById.get(edge.after) ?? []; const lastBefore = before.at(-1); const firstAfter = after[0]; if (lastBefore === undefined && firstAfter !== undefined) - return { - failure: failure({ - kind: "edge", - message: `${edge.before} is required when ${edge.after} occurs.`, - eventIds: [edge.before, edge.after], - locations: [firstAfter], - rawTrace: trace.rawTrace, - }), - satisfied, - }; - if ( + failures.push({ + kind: "edge", + message: `${edge.before} is required when ${edge.after} occurs.`, + eventIds: [edge.before, edge.after], + locations: [firstAfter], + rawTrace: trace.rawTrace, + }); + else if ( lastBefore !== undefined && firstAfter !== undefined && lastBefore.capture_order >= firstAfter.capture_order ) - return { - failure: failure({ - kind: "edge", - message: `${edge.before} must happen before ${edge.after}.`, - eventIds: [edge.before, edge.after], - locations: [lastBefore, firstAfter], - rawTrace: trace.rawTrace, - }), - satisfied, - }; - satisfied.push(`happens_before:${edge.before}:${edge.after}`); + failures.push({ + kind: "edge", + message: `${edge.before} must happen before ${edge.after}.`, + eventIds: [edge.before, edge.after], + locations: [lastBefore, firstAfter], + rawTrace: trace.rawTrace, + }); + else satisfied.push(`happens_before:${edge.before}:${edge.after}`); } for (const constraint of language.not_before) { const events = trace.locationsById.get(constraint.event) ?? []; @@ -287,34 +282,47 @@ const evaluateOrderingConstraints = ( const firstEvent = events[0]; const firstAnchor = anchors[0]; if (firstEvent !== undefined && firstAnchor === undefined) - return { - failure: failure({ - kind: "edge", - message: `${constraint.anchor} is required when ${constraint.event} occurs.`, - eventIds: [constraint.event, constraint.anchor], - locations: [firstEvent], - rawTrace: trace.rawTrace, - }), - satisfied, - }; - if ( + failures.push({ + kind: "edge", + message: `${constraint.anchor} is required when ${constraint.event} occurs.`, + eventIds: [constraint.event, constraint.anchor], + locations: [firstEvent], + rawTrace: trace.rawTrace, + }); + else if ( firstEvent !== undefined && firstAnchor !== undefined && firstEvent.capture_order < firstAnchor.capture_order ) - return { - failure: failure({ - kind: "edge", - message: `${constraint.event} must not occur before ${constraint.anchor}.`, - eventIds: [constraint.event, constraint.anchor], - locations: [firstEvent, firstAnchor], - rawTrace: trace.rawTrace, - }), - satisfied, - }; - satisfied.push(`not_before:${constraint.event}:${constraint.anchor}`); + failures.push({ + kind: "edge", + message: `${constraint.event} must not occur before ${constraint.anchor}.`, + eventIds: [constraint.event, constraint.anchor], + locations: [firstEvent, firstAnchor], + rawTrace: trace.rawTrace, + }); + else satisfied.push(`not_before:${constraint.event}:${constraint.anchor}`); } - return { failure: null, satisfied }; + if (failures.length === 0) return { failure: null, satisfied }; + const unique = (values: readonly T[], key: (value: T) => string): T[] => [ + ...new Map(values.map((value) => [key(value), value])).values(), + ]; + return { + failure: failure({ + kind: "edge", + message: failures.map(({ message }) => message).join(" "), + eventIds: unique( + failures.flatMap(({ eventIds }) => eventIds), + String, + ), + locations: unique( + failures.flatMap(({ locations }) => locations), + ({ collection, index }) => `${collection}:${String(index)}`, + ), + rawTrace: trace.rawTrace, + }), + satisfied, + }; }; const boundaryFailure = ( diff --git a/src/hopper/BridgeLauncher.ts b/src/hopper/BridgeLauncher.ts index 3f83867bf..40cba4288 100644 --- a/src/hopper/BridgeLauncher.ts +++ b/src/hopper/BridgeLauncher.ts @@ -1,5 +1,5 @@ import { execFile } from "node:child_process"; -import { chmod, writeFile } from "node:fs/promises"; +import { chmod, rm, writeFile } from "node:fs/promises"; import { basename, dirname, extname, join } from "node:path"; import { promisify } from "node:util"; @@ -213,6 +213,7 @@ export class HopperApplicationLauncher implements BridgeLauncher { action, input.loadedPath, ]; + let launched = false; try { if (signal?.aborted === true) return err(new HopperCancelledError()); const display = await this.#privateDisplay(signal); @@ -246,6 +247,7 @@ export class HopperApplicationLauncher implements BridgeLauncher { signal, }); if (started === null) return err(new HopperCancelledError()); + launched = true; const cleanup = () => cleanupOwnedProcessGroup(started.ownership); const ownedLauncher = { ...(this.options.preparedImage === undefined @@ -283,6 +285,12 @@ export class HopperApplicationLauncher implements BridgeLauncher { }); } catch (cause: unknown) { return err(hopperLaunchFailure(cause, signal)); + } finally { + // Remove the token-bearing bootstrap file on every non-success path. + // Only the launched process needs the file; all error returns, early + // cancellations, and thrown exceptions must not leave it on disk. + if (!launched) + await rm(bootstrapPath, { force: true }).catch(() => undefined); } } @@ -347,6 +355,8 @@ export class HopperApplicationLauncher implements BridgeLauncher { await chmod(bootstrapPath, 0o600); return ok(bootstrapPath); } catch (cause: unknown) { + // Clean up any partially written bootstrap file that may contain the token. + await rm(bootstrapPath, { force: true }).catch(() => undefined); return err(new HopperStartError({ cause })); } } diff --git a/src/hopper/HopperCleanup.test.ts b/src/hopper/HopperCleanup.test.ts index 8e078b13d..b61009ca3 100644 --- a/src/hopper/HopperCleanup.test.ts +++ b/src/hopper/HopperCleanup.test.ts @@ -7,11 +7,15 @@ import { EventEmitter } from "node:events"; import { Socket } from "node:net"; import pino from "pino"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { ok } from "../domain/result.js"; import type { BridgeLaunch } from "./BridgeLauncher.js"; -import { cleanupHopperSession } from "./HopperCleanup.js"; +import { + cleanupHopperSession, + confirmExternalApplicationExit, + type HopperOwnedResources, +} from "./HopperCleanup.js"; const loggerHarness = () => { const logs: unknown[] = []; @@ -47,6 +51,58 @@ const processCleanupLaunch = (): BridgeLaunch => ({ const failure = (message: string) => Object.assign(new TypeError(message), { code: "ECONNRESET" }); +const externalApplicationResources = (): HopperOwnedResources => ({ + launch: { + process: Object.assign(new EventEmitter(), { + stdout: null, + stderr: null, + exitCode: null, + signalCode: null, + kill: () => true, + }), + ownsProcessLifetime: false, + providerLifetime: "external-application", + shutdownMode: "bridge-request", + }, + processSupervisor: undefined, + runtimeRoot: undefined, + shutdownConfirmed: false, + externalApplication: { + pid: 123, + identity: { state: "readable", identity: "captured-start" }, + }, +}); + +describe("external Hopper process exit confirmation", () => { + it("confirms only the captured GUI process exit and releases its transport", async () => { + const resources = externalApplicationResources(); + const released: unknown[] = []; + const observeExit = vi.fn().mockResolvedValue("exited"); + await confirmExternalApplicationExit( + resources, + undefined, + (socket) => released.push(socket), + observeExit, + ); + expect(observeExit).toHaveBeenCalledWith(123, "captured-start"); + expect(resources.shutdownConfirmed).toBe(true); + expect(released).toEqual([undefined]); + }); + + it("retains cleanup when process identity cannot be verified", async () => { + const resources = externalApplicationResources(); + const releaseTransport = vi.fn(); + await confirmExternalApplicationExit( + resources, + undefined, + releaseTransport, + async () => "unverified", + ); + expect(resources.shutdownConfirmed).toBe(false); + expect(releaseTransport).not.toHaveBeenCalled(); + }); +}); + describe("Hopper shutdown rejection diagnostics", () => { it("retains the primary shutdown operation and rejection cause", async () => { const { logger, logs } = loggerHarness(); @@ -57,6 +113,7 @@ describe("Hopper shutdown rejection diagnostics", () => { processSupervisor: undefined, runtimeRoot: undefined, shutdownConfirmed: false, + externalApplication: undefined, }, activeRequest: null, retainDocument: false, @@ -96,6 +153,7 @@ describe("Hopper shutdown rejection diagnostics", () => { processSupervisor: undefined, runtimeRoot: undefined, shutdownConfirmed: false, + externalApplication: undefined, }, activeRequest: null, retainDocument: false, @@ -152,6 +210,7 @@ it("keeps a prepared backing image when native document closure cannot be confir runtimeRoot, processSupervisor: undefined, shutdownConfirmed: false, + externalApplication: undefined, }, activeRequest: null, retainDocument: true, diff --git a/src/hopper/HopperCleanup.ts b/src/hopper/HopperCleanup.ts index eb155a9fb..ac4cce659 100644 --- a/src/hopper/HopperCleanup.ts +++ b/src/hopper/HopperCleanup.ts @@ -10,8 +10,12 @@ import { ProviderCleanupError } from "../domain/providerCleanupError.js"; import { err, ok, type Result } from "../domain/result.js"; import type { Logger } from "pino"; import type { PrivateRuntimeRoot } from "../process/PrivateRuntimeRoot.js"; -import type { ProcessCleanupResult } from "../process/ProcessOwnership.js"; +import type { + ProcessCleanupResult, + ProcessIdentityObservation, +} from "../process/ProcessOwnership.js"; import type { ProviderProcessSupervisor } from "../process/ProviderProcess.js"; +import { observeProcessStartIdentityExit } from "../process/ProcessOwnershipObservation.js"; import type { BridgeLaunch } from "./BridgeLauncher.js"; import { createOwnedHopperShutdownDiagnostic, @@ -42,6 +46,12 @@ export interface HopperOwnedResources { processSupervisor: ProviderProcessSupervisor | undefined; runtimeRoot: PrivateRuntimeRoot | undefined; shutdownConfirmed: boolean; + externalApplication: + | { + readonly pid: number; + readonly identity: ProcessIdentityObservation; + } + | undefined; } export interface HopperCleanupInput { @@ -68,6 +78,11 @@ export const cleanupHopperSession = async ( cleanupResult: undefined, processStopped: input.resources.processSupervisor === undefined, }; + await confirmExternalApplicationExit( + input.resources, + input.socket, + input.releaseTransport, + ); await report( input.progress, 0, @@ -119,6 +134,28 @@ export const cleanupHopperSession = async ( return cleanupOutcome(input, state); }; +export const confirmExternalApplicationExit = async ( + resources: HopperOwnedResources, + socket: Socket | undefined, + releaseTransport: (socket: Socket | undefined) => void, + observeExit: typeof observeProcessStartIdentityExit = observeProcessStartIdentityExit, +): Promise => { + const application = resources.externalApplication; + if ( + resources.shutdownConfirmed || + resources.launch?.providerLifetime !== "external-application" || + application?.identity.state !== "readable" + ) + return; + if ( + (await observeExit(application.pid, application.identity.identity)) === + "exited" + ) { + resources.shutdownConfirmed = true; + releaseTransport(socket); + } +}; + const requestShutdown = async ( input: HopperCleanupInput, state: CleanupState, diff --git a/src/hopper/HopperClient.ts b/src/hopper/HopperClient.ts index 2943839f4..c81a2f2fa 100644 --- a/src/hopper/HopperClient.ts +++ b/src/hopper/HopperClient.ts @@ -44,11 +44,13 @@ import { type ProviderProcessDiagnostic, ProviderProcessSupervisor, } from "../process/ProviderProcess.js"; +import { observeProcessStartIdentity } from "../process/ProcessOwnershipObservation.js"; import type { BridgeLaunch, BridgeLauncher } from "./BridgeLauncher.js"; import { hopperOperationStage } from "./HopperOperationStage.js"; import type { HopperDiagnostic } from "./HopperDiagnostics.js"; import { cleanupHopperSession, + confirmExternalApplicationExit, type HopperOwnedResources, } from "./HopperCleanup.js"; import { @@ -113,6 +115,7 @@ export class HopperClient { processSupervisor: undefined, runtimeRoot: undefined, shutdownConfirmed: false, + externalApplication: undefined, }; #token: string | undefined; readonly #lineage = new ProviderRunLineage(); @@ -434,6 +437,27 @@ export class HopperClient { } const parsed = parseHopperServerInfo(health.value, runId); if (!parsed.ok) return this.#startupFailure(parsed.error); + if (this.#resources.launch?.providerLifetime === "external-application") { + let identity; + try { + identity = await observeProcessStartIdentity(parsed.value.processId); + } catch (cause: unknown) { + identity = { + state: "unavailable" as const, + reason: + cause instanceof Error + ? cause.message + : "process identity observation failed", + }; + } + this.#resources.externalApplication = { + pid: parsed.value.processId, + identity: identity ?? { + state: "unavailable", + reason: "process identity was not returned while Hopper was running", + }, + }; + } await this.#lineage.observe(this.#resources.launch); return parsed; } @@ -535,6 +559,9 @@ export class HopperClient { // best-effort cleanup: a rejected startup must not mask close/cleanup; // #cleanup reports the authoritative close result. await starting?.catch(() => undefined); + await confirmExternalApplicationExit(this.#resources, this.#socket, () => + this.#releaseTransport(), + ); await this.#reconnectForCleanup(); return await this.#cleanup(options); } finally { diff --git a/src/hopper/HopperSessionValues.ts b/src/hopper/HopperSessionValues.ts index 86d15b135..abee06e04 100644 --- a/src/hopper/HopperSessionValues.ts +++ b/src/hopper/HopperSessionValues.ts @@ -9,6 +9,7 @@ export const HOPPER_SHUTDOWN_TIMEOUT_MS = 30_000; export interface HopperServerInfo { readonly name: string; readonly version: string; + readonly processId: number; } /** Validate the bridge identity and exact per-launch run identifier. */ @@ -22,9 +23,16 @@ export const parseHopperServerInfo = ( !Array.isArray(value) && value.name === "REA Hopper bridge" && typeof value.version === "string" && + typeof value.process_id === "number" && + Number.isSafeInteger(value.process_id) && + value.process_id > 0 && value.run_id === expectedRunId ) - return ok({ name: value.name, version: value.version }); + return ok({ + name: value.name, + version: value.version, + processId: value.process_id, + }); return err(new HopperProtocolError("Hopper bridge health result is invalid")); }; diff --git a/src/hopper/HopperTargetLease.test.ts b/src/hopper/HopperTargetLease.test.ts index 2c72909fa..a0c8e0296 100644 --- a/src/hopper/HopperTargetLease.test.ts +++ b/src/hopper/HopperTargetLease.test.ts @@ -1,6 +1,7 @@ import { execFile } from "node:child_process"; import { once } from "node:events"; import { + chmod, lstat, mkdir, mkdtemp, @@ -31,6 +32,84 @@ afterEach(async () => { ); }); +describe.skipIf(process.platform === "win32")( + "Hopper lease directory admission", + () => { + it("secures an existing owned directory without replacing it", async () => { + const directory = await temporaryDirectory(); + await chmod(directory, 0o755); + const before = await lstat(directory); + const acquired = await acquireLinuxHopperApplicationLease({ + directory, + runId: "permission-hardening", + }); + if (!acquired.acquired) + throw new Error("Expected the owned directory lease"); + try { + const after = await lstat(directory); + expect(after.dev).toBe(before.dev); + expect(after.ino).toBe(before.ino); + expect(after.mode & 0o777).toBe(0o700); + } finally { + await acquired.lease.release(); + } + }); + + it("rejects a directory symlink without changing its target", async () => { + const directory = await temporaryDirectory(); + const realDirectory = join(directory, "actual"); + const alias = join(directory, "selected"); + await mkdir(realDirectory); + await chmod(realDirectory, 0o755); + await symlink(realDirectory, alias); + await expect( + acquireLinuxHopperApplicationLease({ + directory: alias, + runId: "alias", + }), + ).rejects.toMatchObject({ _tag: "HopperStartError" }); + expect(await readdir(realDirectory)).toEqual([]); + expect((await lstat(realDirectory)).mode & 0o777).toBe(0o755); + expect((await lstat(alias)).isSymbolicLink()).toBe(true); + }); + + it.skipIf(process.getuid?.() === 0)( + "explains unreadable directory recovery without changing unverified permissions", + async () => { + const directory = await temporaryDirectory(); + await chmod(directory, 0o333); + try { + await expect( + acquireLinuxHopperApplicationLease({ + directory, + runId: "unreadable", + }), + ).rejects.toMatchObject({ + _tag: "HopperStartError", + cause: expect.objectContaining({ code: "EACCES" }), + userMessage: expect.stringContaining( + "restore owner-only read, write, and search permissions (0700)", + ), + }); + expect((await lstat(directory)).mode & 0o777).toBe(0o333); + // This fixture owns the directory. Apply the documented recovery, + // then exercise the same production acquisition again. + await chmod(directory, 0o700); + const recovered = await acquireLinuxHopperApplicationLease({ + directory, + runId: "recovered-access", + }); + if (!recovered.acquired) throw new Error("Expected recovered access"); + await recovered.lease.release(); + expect(await readdir(directory)).toEqual([]); + } finally { + await chmod(directory, 0o700); + } + }, + ); + }, +); + describe("Hopper target leases", () => { it("reports the owning REA session for a duplicate target and profile", async () => { const directory = await temporaryDirectory(); diff --git a/src/hopper/HopperTargetLease.ts b/src/hopper/HopperTargetLease.ts index 150700267..53d09aba1 100644 --- a/src/hopper/HopperTargetLease.ts +++ b/src/hopper/HopperTargetLease.ts @@ -1,5 +1,6 @@ import { createHash } from "node:crypto"; -import { chmod, lstat, mkdir, realpath, rm, rmdir } from "node:fs/promises"; +import { chmod, mkdir, open, realpath, rm, rmdir } from "node:fs/promises"; +import { constants as fsConstants } from "node:fs"; import { createConnection, createServer, @@ -164,27 +165,69 @@ const tryAcquireLease = async ( }; const ensureLeaseDirectory = async (directory: string): Promise => { - let directoryStat: Awaited>; try { await mkdir(directory, { recursive: true, mode: 0o700 }); - directoryStat = await lstat(directory); } catch (cause: unknown) { throw new HopperStartError({ cause, - userMessage: - "REA could not create or inspect its private Hopper lease directory.", + userMessage: "REA could not create its private Hopper lease directory.", }); } - if ( - directoryStat.isSymbolicLink() || - directoryStat.isDirectory() === false || - (process.getuid !== undefined && directoryStat.uid !== process.getuid()) - ) + // Use O_DIRECTORY|O_NOFOLLOW to open the directory without following symlinks, + // then inspect and secure that descriptor. Later socket operations still use + // the path and are not made atomic by this directory validation. + let handle; + try { + handle = await open( + directory, + fsConstants.O_RDONLY | + (fsConstants.O_DIRECTORY ?? 0) | + (fsConstants.O_NOFOLLOW ?? 0), + ); + const directoryStat = await handle.stat(); + if ( + directoryStat.isSymbolicLink() || + directoryStat.isDirectory() === false || + (process.getuid !== undefined && directoryStat.uid !== process.getuid()) + ) + throw new HopperStartError({ + userMessage: + "REA's Hopper lease directory is not a private directory owned by this user.", + }); + // POSIX permission hardening: chmod on the already-opened descriptor avoids + // a second TOCTOU window where the directory could be swapped between check + // and permission fix. Windows DACL-based security is handled separately. + const posixPermissionsApply = process.getuid !== undefined; + if (posixPermissionsApply && (directoryStat.mode & 0o077) !== 0) { + await handle.chmod(0o700); + // Re-verify permissions on the same opened directory. + const afterChmod = await handle.stat(); + if ( + afterChmod.isSymbolicLink() || + afterChmod.isDirectory() === false || + (afterChmod.mode & 0o077) !== 0 + ) + throw new HopperStartError({ + userMessage: + "REA's Hopper lease directory permissions could not be secured.", + }); + } + } catch (cause: unknown) { + if (cause instanceof HopperStartError) throw cause; + const accessDenied = + process.getuid !== undefined && + cause instanceof Error && + "code" in cause && + cause.code === "EACCES"; throw new HopperStartError({ - userMessage: - "REA's Hopper lease directory is not a private directory owned by this user.", + cause, + userMessage: accessDenied + ? `REA could not access its Hopper lease directory at ${JSON.stringify(directory)}. Check its ownership and parent-directory access. For a directory you own, restore owner-only read, write, and search permissions (0700), then retry.` + : "REA could not inspect its private Hopper lease directory.", }); - if ((directoryStat.mode & 0o077) !== 0) await chmod(directory, 0o700); + } finally { + await handle?.close(); + } }; const bind = (server: Server, socketPath: string): Promise => diff --git a/src/process/ArtifactAnalysisResources.test.ts b/src/process/ArtifactAnalysisResources.test.ts new file mode 100644 index 000000000..84ce0e6d7 --- /dev/null +++ b/src/process/ArtifactAnalysisResources.test.ts @@ -0,0 +1,85 @@ +import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; +import { expect, it } from "vitest"; +import { readStableArtifact } from "../artifacts/readStableArtifact.js"; +import { projectAnalysisError } from "../domain/analysisErrorProjection.js"; +import { ProviderAdapterError } from "../domain/providerAdapterError.js"; +import { err } from "../domain/result.js"; +import { ArtifactAnalysisResources } from "./ArtifactAnalysisResources.js"; +import { readLinuxResourceLimits } from "./LinuxResourceLimits.js"; +import { ProviderProcessSupervisor } from "./ProviderProcess.js"; + +it("projects the specific reason and member evidence for retained cleanup", async () => { + const producer = Object.assign(new EventEmitter(), { + stdout: new PassThrough(), + stderr: new PassThrough(), + exitCode: null, + signalCode: null, + kill: () => true, + }); + const owner = new ProviderProcessSupervisor({ + process: producer, + ownsProcessLifetime: true, + ownership: { + runId: "rea-tshark-run-1", + leaderPid: 4242, + processGroupId: 4242, + }, + cleanup: async () => ({ + cleaned: false as const, + reason: "process ownership token could not be read for 1 live process", + failures: [ + { + pid: 101, + reason: "environment-unreadable" as const, + diagnostic: "sysctl_errno_22", + }, + ], + unverified: [ + { pid: 900, diagnostic: "platform_binary_environment_withheld" }, + ], + }), + }); + const resources = new ArtifactAnalysisResources( + "fixture-provider", + "fixture-operation", + readStableArtifact, + readLinuxResourceLimits, + { message: "busy", details: {} }, + ); + resources.retain({ path: "runtime-root", close: async () => {} }, owner); + + const result = await resources.finish( + err(new ProviderAdapterError("fixture-provider", "fixture-operation")), + () => { + throw new Error("unexpected success observation"); + }, + ); + owner.dispose(); + + if (result.ok) throw new Error("expected cleanup failure"); + expect(projectAnalysisError(result.error)).toMatchObject({ + code: "cleanup_incomplete", + details: { + resources: [ + "rea-tshark-run-1", + "runtime-root", + "pid:4242", + "process-group:4242", + ], + diagnostics: { + reason: "process ownership token could not be read for 1 live process", + failures: [ + { + pid: 101, + reason: "environment-unreadable", + diagnostic: "sysctl_errno_22", + }, + ], + unverified: [ + { pid: 900, diagnostic: "platform_binary_environment_withheld" }, + ], + }, + }, + }); +}); diff --git a/src/process/ArtifactAnalysisResources.ts b/src/process/ArtifactAnalysisResources.ts index ae3d9f803..ff7b63048 100644 --- a/src/process/ArtifactAnalysisResources.ts +++ b/src/process/ArtifactAnalysisResources.ts @@ -24,7 +24,7 @@ import { } from "./LinuxResourceLimits.js"; import type { PrivateRuntimeRoot } from "./PrivateRuntimeRoot.js"; import type { ProviderProcessSupervisor } from "./ProviderProcess.js"; -import { OwnedCommandFailure } from "./OwnedCommand.js"; +import { OwnedCommandFailure, ownedCommandResources } from "./OwnedCommand.js"; export type ArtifactAnalysisRuntime = Pick< PrivateRuntimeRoot, @@ -129,22 +129,67 @@ export class ArtifactAnalysisResources { return [...this.#pending.keys()].map((root) => root.path); } + #resourcesFor(root: ArtifactAnalysisRuntime): readonly string[] { + const ownership = this.#pending.get(root)?.owner?.launch.ownership; + return ownership === undefined + ? [root.path] + : ownedCommandResources(ownership, root.path); + } + + get #pendingResources(): readonly string[] { + return [...this.#pending.keys()].flatMap((root) => + this.#resourcesFor(root), + ); + } + async release(): Promise { for (const [root, retained] of this.#pending) { try { const { owner } = retained; if (owner !== undefined) { const stopped = await owner.stop(); - if (stopped.status === "incomplete") throw new Error(stopped.reason); + if (stopped.status === "incomplete") { + const cleanup = stopped.cleanupResult; + throw new ProviderCleanupError( + this.providerId, + this.#resourcesFor(root), + { + reason: stopped.reason, + ...(cleanup?.failures === undefined + ? {} + : { + failures: cleanup.failures.map((failure) => ({ + ...failure, + })), + }), + ...(cleanup?.unverified === undefined + ? {} + : { + unverified: cleanup.unverified.map((process) => ({ + ...process, + })), + }), + }, + { operation: retained.operation }, + ); + } this.#pending.set(root, { ...retained, owner: undefined }); } await root.close(); this.#pending.delete(root); } catch (cause: unknown) { + const diagnostics = + cause instanceof ProviderCleanupError ? cause.diagnostics : undefined; throw new ProviderCleanupError( this.providerId, - [root.path], - { reason: String(cause) }, + this.#resourcesFor(root), + { + ...diagnostics, + reason: + cause instanceof ProviderCleanupError + ? String(diagnostics?.reason ?? cause.message) + : String(cause), + }, { operation: retained.operation, cause }, ); } @@ -236,12 +281,18 @@ export class ArtifactAnalysisResources { const partialObservation = result.ok ? observation(result.value) : result.error.partialObservation; + const cleanupDiagnostics = + cause instanceof ProviderCleanupError ? cause.diagnostics : undefined; return err( new ProviderCleanupError( this.providerId, - this.pendingPaths, + this.#pendingResources, { - reason: String(cause), + ...cleanupDiagnostics, + reason: + cause instanceof ProviderCleanupError + ? String(cleanupDiagnostics?.reason ?? cause.message) + : String(cause), previous_error: result.ok ? null : projectAnalysisError(result.error), diff --git a/src/process/ExecFileBoundedOutput.test.ts b/src/process/ExecFileBoundedOutput.test.ts new file mode 100644 index 000000000..c055bc002 --- /dev/null +++ b/src/process/ExecFileBoundedOutput.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from "vitest"; + +import { execFileBoundedOutput } from "./ExecFileBoundedOutput.js"; + +describe("execFileBoundedOutput", () => { + it("lets a command finish after its output exceeds the capture limit", async () => { + const result = await execFileBoundedOutput( + process.execPath, + [ + "-e", + "process.stdout.write('x'.repeat(70000)); process.stderr.write('failure detail');", + ], + { env: process.env, limit: 64 * 1024 }, + ); + expect(result.exitCode).toBe(0); + expect(Buffer.byteLength(result.stdout)).toBeLessThan(64 * 1024 + 50); + expect(result.stdout).toContain("[output omitted]"); + expect(result.stderr).toBe("failure detail"); + expect(result.outputTruncated).toBe(true); + }); + + it("retains a failing command's status and stderr after verbose stdout", async () => { + const result = await execFileBoundedOutput( + process.execPath, + [ + "-e", + "process.stdout.write('x'.repeat(70000)); process.stderr.write('package conflict'); process.exitCode = 1;", + ], + { env: process.env, limit: 64 * 1024 }, + ); + expect(result.exitCode).toBe(1); + expect(result.stderr).toBe("package conflict"); + expect(result.outputTruncated).toBe(true); + }); + + it("retains a final error after a long prefix in the same stream", async () => { + const result = await execFileBoundedOutput( + process.execPath, + [ + "-e", + "process.stderr.write('warning '.repeat(10000)); process.stderr.write('error: final conflict'); process.exitCode = 1;", + ], + { env: process.env, limit: 64 * 1024 }, + ); + expect(result.exitCode).toBe(1); + expect(result.stderr).toContain("[output omitted]"); + expect(result.stderr).toContain("error: final conflict"); + expect(result.outputTruncated).toBe(true); + }); +}); diff --git a/src/process/ExecFileBoundedOutput.ts b/src/process/ExecFileBoundedOutput.ts new file mode 100644 index 000000000..fad51e2bc --- /dev/null +++ b/src/process/ExecFileBoundedOutput.ts @@ -0,0 +1,73 @@ +import { spawn } from "node:child_process"; + +export interface BoundedCommandOutput { + readonly exitCode: string | number | null; + readonly stdout: string; + readonly stderr: string; + readonly outputTruncated: boolean; +} + +/** Drain both streams while retaining bounded head and tail evidence. */ +export const execFileBoundedOutput = ( + executable: string, + args: readonly string[], + options: { + readonly env: Readonly; + readonly limit: number; + }, +): Promise => + new Promise((resolve) => { + const child = spawn(executable, [...args], { + env: options.env, + stdio: ["ignore", "pipe", "pipe"], + }); + const headLimit = Math.ceil(options.limit / 2); + const tailLimit = options.limit - headLimit; + const stdout = { head: Buffer.alloc(0), tail: Buffer.alloc(0), total: 0 }; + const stderr = { head: Buffer.alloc(0), tail: Buffer.alloc(0), total: 0 }; + let outputTruncated = false; + const capture = + (stream: typeof stdout) => + (chunk: Buffer): void => { + stream.total += chunk.length; + const headCount = Math.min( + chunk.length, + Math.max(0, headLimit - stream.head.length), + ); + if (headCount > 0) + stream.head = Buffer.concat([ + stream.head, + chunk.subarray(0, headCount), + ]); + const rest = chunk.subarray(headCount); + if (rest.length > 0 && tailLimit > 0) + stream.tail = Buffer.concat([stream.tail, rest]).subarray(-tailLimit); + if (stream.total > options.limit) outputTruncated = true; + }; + child.stdout.on("data", capture(stdout)); + child.stderr.on("data", capture(stderr)); + let spawnError: NodeJS.ErrnoException | undefined; + child.on("error", (error: NodeJS.ErrnoException) => { + spawnError = error; + }); + child.on("close", (code, signal) => { + resolve({ + exitCode: spawnError?.code ?? code ?? signal, + stdout: retainedOutput(stdout, options.limit), + stderr: retainedOutput(stderr, options.limit), + outputTruncated, + }); + }); + }); + +const retainedOutput = ( + stream: { + readonly head: Buffer; + readonly tail: Buffer; + readonly total: number; + }, + limit: number, +): string => + stream.total > limit + ? `${stream.head.toString("utf8")} … [output omitted] … ${stream.tail.toString("utf8")}` + : Buffer.concat([stream.head, stream.tail]).toString("utf8"); diff --git a/src/process/ProcessOwnership.test.ts b/src/process/ProcessOwnership.test.ts index de233698c..fafa39950 100644 --- a/src/process/ProcessOwnership.test.ts +++ b/src/process/ProcessOwnership.test.ts @@ -209,6 +209,185 @@ describe("opaque process neighbors during cleanup", () => { }); describe("related sweep candidates during cleanup", () => { + it("counts each unreadable live member once", async () => { + const processes = [ + { + pid: 100, + parentPid: 1, + processGroupId: 100, + state: "S", + command: "capture", + }, + { + pid: 101, + parentPid: 100, + processGroupId: 100, + state: "S", + command: "child-a", + }, + { + pid: 102, + parentPid: 100, + processGroupId: 100, + state: "S", + command: "child-b", + }, + ]; + const signalGroup = vi.fn(); + const adapter: ProcessOwnershipHost = { + listProcesses: async () => processes, + environment: async () => ({}), + runTokens: async (members) => + new Map( + members.map(({ pid }) => [ + pid, + pid === 100 + ? { state: "readable" as const, runId: ownership.runId } + : { + state: "unavailable" as const, + reason: "environment_unavailable", + }, + ]), + ), + signalGroup, + }; + const result = await cleanupOwnedProcessGroup( + { ...ownership, sweepTokenOwnedProcesses: true, captureBaseline: [] }, + adapter, + ); + if (result.cleaned) throw new Error("expected cleanup failure"); + expect(signalGroup).not.toHaveBeenCalled(); + expect(result.failures?.map(({ pid }) => pid).sort()).toEqual([101, 102]); + expect(result.reason).toContain("2 live process(es)"); + }); + + it("upgrades unreadable evidence when later group validation finds a foreign token", async () => { + const processes = [ + { + pid: 100, + parentPid: 1, + processGroupId: 100, + state: "S", + command: "capture", + }, + { + pid: 200, + parentPid: 100, + processGroupId: 200, + state: "S", + command: "detached-child", + }, + ]; + const signalGroup = vi.fn(); + let childReads = 0; + const adapter: ProcessOwnershipHost = { + listProcesses: async () => processes, + processIdentities: async (members) => + new Map( + members.map(({ pid }) => [ + pid, + { state: "readable" as const, identity: `start-${String(pid)}` }, + ]), + ), + environment: async () => ({}), + runTokens: async (members) => + new Map( + members.map(({ pid }) => [ + pid, + pid === 100 + ? { state: "readable" as const, runId: ownership.runId } + : ++childReads <= 4 + ? { + state: "unavailable" as const, + reason: "environment_unavailable", + } + : { state: "readable" as const, runId: "foreign-token" }, + ]), + ), + signalGroup, + }; + const result = await cleanupOwnedProcessGroup( + { ...ownership, sweepTokenOwnedProcesses: true, captureBaseline: [] }, + adapter, + ); + expect(result).toMatchObject({ + cleaned: false, + reason: "process ownership token mismatch for PID(s): 200", + failures: [{ pid: 200, reason: "run-token-mismatch" }], + }); + expect(signalGroup).not.toHaveBeenCalledWith(200, "SIGKILL"); + }); +}); + +describe("withheld process evidence during cleanup", () => { + it.each(["revalidation", "signal-error"] as const)( + "retains withheld-process evidence after %s failure", + async (mode) => { + const processes = [ + { + pid: 100, + parentPid: 1, + processGroupId: 100, + state: "S", + command: "fixture", + }, + { + pid: 900, + parentPid: 1, + processGroupId: 900, + state: "S", + command: "opaque", + }, + ]; + let rootReads = 0; + const signalGroup = vi.fn(() => { + if (mode === "signal-error") + throw Object.assign(new Error("signal denied"), { code: "EACCES" }); + }); + const adapter: ProcessOwnershipHost = { + listProcesses: async () => processes, + environment: async () => ({}), + runTokens: async (members) => + new Map( + members.map(({ pid }) => [ + pid, + pid === 900 + ? { + state: "unavailable" as const, + reason: "platform_binary_environment_withheld", + } + : { + state: "readable" as const, + runId: + mode === "revalidation" && ++rootReads > 2 + ? "other-run" + : "run-token", + }, + ]), + ), + signalGroup, + }; + + const result = await cleanupOwnedProcessGroup( + { ...ownership, sweepTokenOwnedProcesses: true, captureBaseline: [] }, + adapter, + ); + + expect(result.unverified).toEqual([ + { pid: 900, diagnostic: "platform_binary_environment_withheld" }, + ]); + if (result.cleaned) throw new Error("expected cleanup failure"); + if (mode === "revalidation") + expect(result.reason).toBe( + "process ownership token mismatch for PID(s): 100", + ); + else + expect(result.reason).toBe( + "owned process group signal failed: signal denied", + ); + }, + ); + it.each([ "environment_unavailable", "platform_binary_environment_withheld", diff --git a/src/process/ProcessOwnership.ts b/src/process/ProcessOwnership.ts index 399b9cd06..91d681f79 100644 --- a/src/process/ProcessOwnership.ts +++ b/src/process/ProcessOwnership.ts @@ -227,6 +227,15 @@ export const readProcessRunId = async ( ): Promise => (await host.environment(pid, signal)).REA_PROCESS_RUN_ID; +/** Keep unrelated-process uncertainty attached to every outcome of this cleanup attempt. */ +const withUnverifiedObservations = ( + result: ProcessCleanupFailure, + unverified: readonly OwnershipSweepUnverifiedProcess[], +): ProcessCleanupFailure => + unverified.length === 0 || (result.unverified?.length ?? 0) > 0 + ? result + : { ...result, unverified }; + /** Token-validate every rooted POSIX process group before signaling any. */ export const cleanupOwnedProcessGroup = async ( ownership: OwnedProcessGroup, @@ -252,7 +261,16 @@ export const cleanupOwnedProcessGroup = async ( host, plan.tokenOwnedIdentities, ); - if ("cleaned" in revalidation) return revalidation; + if ("cleaned" in revalidation) { + if (!revalidation.cleaned) + return withUnverifiedObservations(revalidation, plan.unverified); + return { + ...revalidation, + ...(revalidation.unverified?.length || plan.unverified.length === 0 + ? {} + : { unverified: plan.unverified }), + }; + } if (revalidation.empty) continue; try { host.signalGroup(processGroupId, "SIGKILL"); @@ -261,10 +279,13 @@ export const cleanupOwnedProcessGroup = async ( const code = cause instanceof Error && "code" in cause ? cause.code : undefined; if (code !== "ESRCH") - return { - cleaned: false, - reason: `owned process group signal failed: ${errorMessage(cause)}`, - }; + return withUnverifiedObservations( + { + cleaned: false, + reason: `owned process group signal failed: ${errorMessage(cause)}`, + }, + plan.unverified, + ); } } if (plan.unresolved.length > 0) @@ -507,7 +528,8 @@ const createOwnedCleanupPlan = async ( const scanFailures = tokenOwned.failures.filter(({ pid }) => members.some((member) => member.pid === pid), ); - if (failures.length > 0) unresolved.push(...failures); + for (const failure of failures) + mergeProcessOwnershipFailure(unresolved, failure); if ( failures.length === 0 && scanFailures.length === 0 && @@ -533,6 +555,36 @@ const createOwnedCleanupPlan = async ( }; }; +/** Keep one current, most useful ownership diagnosis for each live PID. */ +const mergeProcessOwnershipFailure = ( + failures: ProcessOwnershipValidationFailure[], + next: ProcessOwnershipValidationFailure, +): void => { + const index = failures.findIndex(({ pid }) => pid === next.pid); + if (index < 0) { + failures.push(next); + return; + } + const previous = failures[index]; + if ( + previous !== undefined && + ownershipFailurePriority(next.reason) >= + ownershipFailurePriority(previous.reason) + ) + failures[index] = next; +}; + +const ownershipFailurePriority = ( + reason: ProcessOwnershipValidationFailure["reason"], +): number => + reason === "run-token-mismatch" + ? 3 + : reason === "process-identity-unavailable" + ? 2 + : reason === "environment-unreadable" + ? 1 + : 0; + const revalidateOwnedProcessGroup = async ( processGroupId: number, ownership: OwnedProcessGroup, @@ -768,9 +820,27 @@ const recheckOwnershipFailures = async ( if (!failures.some(({ reason }) => reason !== "run-token-mismatch")) return failures; try { - const live = new Set( - liveProcesses(await host.listProcesses()).map(({ pid }) => pid), - ); + let processes = await host.listProcesses(); + const isDarwin = (host.platform ?? process.platform) === "darwin"; + for ( + let attempt = 0; + isDarwin && + attempt < TOKEN_READ_RETRY_ATTEMPTS && + processes.some( + ({ pid, state }) => + failures.some( + (failure) => + failure.pid === pid && failure.reason !== "run-token-mismatch", + ) && state.slice(1).includes("E"), + ); + attempt += 1 + ) { + await new Promise((resolve) => + setTimeout(resolve, TOKEN_READ_RETRY_DELAY_MS), + ); + processes = await host.listProcesses(); + } + const live = new Set(liveProcesses(processes).map(({ pid }) => pid)); return failures.filter( ({ pid, reason }) => reason === "run-token-mismatch" || live.has(pid), ); diff --git a/src/process/ProcessOwnership.validation.test.ts b/src/process/ProcessOwnership.validation.test.ts index c718f4b93..ed6179a77 100644 --- a/src/process/ProcessOwnership.validation.test.ts +++ b/src/process/ProcessOwnership.validation.test.ts @@ -21,6 +21,40 @@ const launcher: ProcessTableEntry = { }; describe("owned process-group cleanup validation: ownership and lineage", () => { + it("settles unreadable observations for Darwin processes exiting in the kernel", async () => { + let snapshots = 0; + const child = { + pid: 101, + parentPid: 100, + processGroupId: 100, + state: "?E", + command: "fixture-child", + }; + const adapter: ProcessOwnershipHost = { + platform: "darwin", + listProcesses: () => { + snapshots += 1; + return Promise.resolve( + snapshots === 1 + ? [launcher, child] + : snapshots === 2 + ? [launcher, child] + : [launcher, { ...child, state: "Z" }], + ); + }, + environment: (pid) => + pid === 100 + ? Promise.resolve({ REA_PROCESS_RUN_ID: "run-token" }) + : Promise.reject(new Error("sysctl_errno_22")), + signalGroup: vi.fn(), + }; + + await expect(cleanupOwnedProcessGroup(ownership, adapter)).resolves.toEqual( + { cleaned: true, signaled: true }, + ); + expect(snapshots).toBeGreaterThanOrEqual(3); + }); + it("fails closed when a descendant in another process group lacks the token", async () => { const adapter: ProcessOwnershipHost = { listProcesses: () => diff --git a/src/process/ProcessOwnershipObservation.ts b/src/process/ProcessOwnershipObservation.ts index 5747cded8..3b4311c94 100644 --- a/src/process/ProcessOwnershipObservation.ts +++ b/src/process/ProcessOwnershipObservation.ts @@ -510,6 +510,22 @@ export const observeProcessStartIdentity = async ( }; }; +/** Confirm whether the process that held an authenticated bridge has exited. */ +export const observeProcessStartIdentityExit = async ( + pid: number, + expectedIdentity: string, + host: ProcessOwnershipHost = systemProcessOwnershipHost, +): Promise<"exited" | "present" | "unverified"> => { + try { + const observed = await observeProcessStartIdentity(pid, host); + if (observed === undefined) return "exited"; + if (observed.state !== "readable") return "unverified"; + return observed.identity === expectedIdentity ? "present" : "exited"; + } catch { + return "unverified"; + } +}; + /** Revalidate a launch-time start identity immediately before signaling a PID. */ export const signalProcessWithStartIdentity = async ( pid: number, diff --git a/src/process/ProcessStartIdentity.test.ts b/src/process/ProcessStartIdentity.test.ts index 20b2862ee..8d2f53261 100644 --- a/src/process/ProcessStartIdentity.test.ts +++ b/src/process/ProcessStartIdentity.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from "vitest"; import type { ProcessOwnershipHost } from "./ProcessOwnership.js"; import { observeProcessStartIdentity, + observeProcessStartIdentityExit, signalProcessWithStartIdentity, } from "./ProcessOwnershipObservation.js"; @@ -37,6 +38,26 @@ describe("process start identity lease", () => { }); }); + it("confirms exit or PID replacement without treating unreadable state as exit", async () => { + await expect( + observeProcessStartIdentityExit(42, "start-1", host(undefined)), + ).resolves.toBe("exited"); + await expect( + observeProcessStartIdentityExit(42, "start-1", host("replacement-start")), + ).resolves.toBe("exited"); + await expect( + observeProcessStartIdentityExit(42, "start-1", host("start-1")), + ).resolves.toBe("present"); + const unavailable: ProcessOwnershipHost = { + ...host("start-1"), + processIdentities: async () => + new Map([[42, { state: "unavailable", reason: "reader unavailable" }]]), + }; + await expect( + observeProcessStartIdentityExit(42, "start-1", unavailable), + ).resolves.toBe("unverified"); + }); + it("does not adopt a replacement PID's first identity for an owned launch", async () => { const replacement: ProcessOwnershipHost = { ...host("replacement-start"), diff --git a/src/process/ProviderProcess.ts b/src/process/ProviderProcess.ts index c5e547ac7..17eda48d7 100644 --- a/src/process/ProviderProcess.ts +++ b/src/process/ProviderProcess.ts @@ -66,6 +66,8 @@ export interface ProviderProcessHandle extends Pick< readonly signalCode: NodeJS.Signals | null; /** True once an enforced process memory limit reported an allocation failure. */ readonly memoryLimitReached?: boolean; + /** True if a Windows backpressured stream was destroyed before output drained. */ + readonly outputIncomplete?: boolean; kill(signal?: NodeJS.Signals | number): boolean; unref?(): void; ref?(): void; @@ -97,7 +99,7 @@ export interface ProviderProcessExitObservation { /** Detached diagnostics captured for one supervised provider process. */ export interface ProviderProcessSnapshot { - /** True only when a configured diagnostic budget excluded output bytes. */ + /** True when retained diagnostics omit bytes due to the budget or provider output loss. */ readonly diagnosticTruncated?: boolean; readonly stdout: { readonly text: string; @@ -120,6 +122,8 @@ export interface ProviderProcessSnapshot { readonly signal: NodeJS.Signals | null | undefined; /** Present only when the process tree reached its enforced memory limit. */ readonly memoryLimitReached?: true; + /** Present when a provider stream was destroyed before all bytes were observed. */ + readonly outputIncomplete?: true; } /** Resource-safe events emitted without imposing provider protocol semantics. */ @@ -338,8 +342,11 @@ export class ProviderProcessSupervisor { /** Latest retained output and exit observation, with explicit configured truncation. */ snapshot(): ProviderProcessSnapshot { + const outputIncomplete = this.launch.process.outputIncomplete === true; return { - ...(this.#diagnosticTruncated ? { diagnosticTruncated: true } : {}), + ...(this.#diagnosticTruncated || outputIncomplete + ? { diagnosticTruncated: true } + : {}), stdout: this.#stdout.snapshot(this.#stdoutObservedBytes), stderr: this.#stderr.snapshot(this.#stderrObservedBytes), exitCode: this.#exitObservation?.code, @@ -347,6 +354,7 @@ export class ProviderProcessSupervisor { ...(this.launch.process.memoryLimitReached === true ? { memoryLimitReached: true } : {}), + ...(outputIncomplete ? { outputIncomplete: true } : {}), }; } diff --git a/src/process/ProviderProcessOutput.test.ts b/src/process/ProviderProcessOutput.test.ts index 3ca2621f0..993386e79 100644 --- a/src/process/ProviderProcessOutput.test.ts +++ b/src/process/ProviderProcessOutput.test.ts @@ -122,6 +122,30 @@ it("reports a reached process memory limit only when the handle observed one", ( } }); +it("projects provider output loss as truncated diagnostics without inventing a byte count", () => { + const producer = Object.assign(new EventEmitter(), { + stdout: new PassThrough(), + stderr: new PassThrough(), + exitCode: null, + signalCode: null, + kill: () => false, + }); + Object.defineProperty(producer, "outputIncomplete", { value: true }); + const supervisor = new ProviderProcessSupervisor({ + process: producer, + ownsProcessLifetime: false, + }); + try { + expect(supervisor.snapshot()).toMatchObject({ + diagnosticTruncated: true, + outputIncomplete: true, + }); + expect(supervisor.snapshot()).not.toHaveProperty("discardedOutputBytes"); + } finally { + supervisor.dispose(); + } +}); + it("unreferences process and stream handles when stop reports incomplete cleanup", async () => { const unrefProcess = vi.fn(); const unrefStdin = vi.fn(); diff --git a/src/process/WindowsAuthority.ts b/src/process/WindowsAuthority.ts index b25a22863..e05080739 100644 --- a/src/process/WindowsAuthority.ts +++ b/src/process/WindowsAuthority.ts @@ -1,4 +1,7 @@ -import { systemWindowsNativeAuthority } from "../windows/WindowsNativeLoader.js"; +import { + systemWindowsNativeAuthority, + type WindowsNativeLoadResult, +} from "../windows/WindowsNativeLoader.js"; /** Windows controls are proven by the packaged native boundary, not POSIX flags. */ export type WindowsNativeCapability = @@ -12,6 +15,15 @@ export type WindowsNativeCapability = readonly reason: string; readonly proof: "not-proven" | "not-applicable"; readonly retryable?: true; + readonly code: + | "unsupported_host" + | "runtime_missing" + | "runtime_unavailable"; + readonly failureStage: Exclude< + WindowsNativeLoadResult, + { available: true } + >["failureStage"]; + readonly remediation: string; }; /** Security controls required before REA may claim Windows isolation. */ @@ -25,64 +37,87 @@ export interface WindowsNativeCapabilitySet { export const WINDOWS_NATIVE_AUTHORITY_UNAVAILABLE_REASON = "Windows native controls are unavailable in this runtime; use the Windows x64 REA package with its matching bundled native artifact."; +/** One readiness observation, shared by diagnostics and security controls. */ +export type WindowsNativeReadiness = + | { readonly available: true } + | { + readonly available: false; + readonly reason: string; + readonly failureStage: Exclude< + WindowsNativeLoadResult, + { available: true } + >["failureStage"]; + readonly code: + | "unsupported_host" + | "runtime_missing" + | "runtime_unavailable"; + readonly retryable?: true; + readonly remediation: string; + readonly proof: "not-proven" | "not-applicable"; + }; + +/** Interpret the loader's tagged failure without guessing from OS error text. */ +export const inspectWindowsNativeReadiness = ( + platform: NodeJS.Platform, + observation?: WindowsNativeLoadResult, +): WindowsNativeReadiness => { + if (platform !== "win32") + return { + available: false, + failureStage: "host", + reason: "Windows native authority is not applicable on this host", + proof: "not-applicable", + code: "unsupported_host", + remediation: "Use a Windows x64 host for Windows native controls.", + }; + const loaded = observation ?? systemWindowsNativeAuthority(); + if (loaded.available) return { available: true }; + const failureStage = loaded.failureStage; + return { + available: false, + failureStage, + reason: loaded.reason, + ...(loaded.retryable === true ? { retryable: true as const } : {}), + proof: "not-proven", + code: + failureStage === "host" + ? "unsupported_host" + : failureStage === "artifact" + ? "runtime_missing" + : "runtime_unavailable", + remediation: + failureStage === "host" + ? "Use a Windows x64 host and a supported Node.js runtime with Node-API 8 or newer." + : failureStage === "artifact" + ? "Install the matching Windows x64 REA package with its bundled native addon; check the reported artifact load or integrity failure." + : failureStage === "inspection" + ? "Install the matching Windows x64 REA package; its native inspection must prove every required security control." + : "Inspect the preserved native self-test error. Check that the process's temporary directory is writable local NTFS and that its filesystem, DACL, Job Object, and cleanup operations are permitted. Resolve the reported constraint, then rerun rea doctor; reinstalling Ghidra or Java cannot fix this self-test.", + }; +}; + /** * Describe native Windows controls without inferring them from taskkill, * chmod, symlink creation, or a numeric POSIX open flag. */ export const windowsNativeCapabilities = ( platform: NodeJS.Platform, + readiness: WindowsNativeReadiness = inspectWindowsNativeReadiness(platform), ): WindowsNativeCapabilitySet => { - if (platform !== "win32") { - const notApplicable = { - available: false, - reason: "Windows native authority is not applicable on this host", - proof: "not-applicable", - } as const; - return { - job_object_process_ownership: notApplicable, - private_runtime_dacl: notApplicable, - reparse_safe_path_admission: notApplicable, - }; - } - - if (process.platform === "win32") { - const loaded = systemWindowsNativeAuthority(); - const capability = loaded.available - ? ({ available: true, reason: null, proof: "native-authority" } as const) - : ({ - available: false, - reason: loaded.reason, - proof: "not-proven", - ...(loaded.retryable === true ? { retryable: true as const } : {}), - } as const); - return { - job_object_process_ownership: capability, - private_runtime_dacl: capability, - reparse_safe_path_admission: capability, - }; - } - - const notProven = { - available: false, - reason: WINDOWS_NATIVE_AUTHORITY_UNAVAILABLE_REASON, - proof: "not-proven", - } as const; + const capability = readiness.available + ? ({ available: true, reason: null, proof: "native-authority" } as const) + : ({ + available: false, + reason: readiness.reason, + proof: readiness.proof, + code: readiness.code, + failureStage: readiness.failureStage, + remediation: readiness.remediation, + ...(readiness.retryable === true ? { retryable: true as const } : {}), + } as const); return { - job_object_process_ownership: notProven, - private_runtime_dacl: notProven, - reparse_safe_path_admission: notProven, + job_object_process_ownership: capability, + private_runtime_dacl: capability, + reparse_safe_path_admission: capability, }; }; - -/** Whether every native control needed for Windows isolation is proven. */ -export const hasWindowsNativeAuthority = (platform: NodeJS.Platform): boolean => - Object.values(windowsNativeCapabilities(platform)).every( - ({ available }) => available, - ); - -/** Preserve the actual loader or operating-system constraint in provider errors. */ -export const windowsNativeAuthorityUnavailableReason = ( - platform: NodeJS.Platform, -): string => - windowsNativeCapabilities(platform).job_object_process_ownership.reason ?? - WINDOWS_NATIVE_AUTHORITY_UNAVAILABLE_REASON; diff --git a/src/process/capture/FilesystemCoverage.test.ts b/src/process/capture/FilesystemCoverage.test.ts index baa5f5c94..4d77ebfaa 100644 --- a/src/process/capture/FilesystemCoverage.test.ts +++ b/src/process/capture/FilesystemCoverage.test.ts @@ -100,6 +100,45 @@ it("distinguishes unavailable content hashes from completed path enumeration", a ).toMatchObject({ status: "deleted", before: { sha256: null }, after: null }); }); +it.skipIf(process.platform === "win32" || process.getuid?.() === 0)( + "retains readable entries and marks an unreadable directory subtree unknown", + async () => { + const root = await createTestTempDirectory("rea-fs-unreadable-directory-"); + const visible = join(root, "a-visible.txt"); + const blocked = join(root, "z-blocked"); + await writeFile(visible, "visible"); + await mkdir(blocked); + await chmod(blocked, 0); + const scenario = parseProcessScenario({ + executable: process.execPath, + working_directory: root, + filesystem_observation_paths: [root], + }); + + try { + const snapshot = await snapshotRoots(scenario); + expect(snapshot.files.map(({ path }) => path)).toContain( + "root_0:a-visible.txt", + ); + expect(snapshot.files.map(({ path }) => path)).toContain( + "root_0:z-blocked", + ); + expect(snapshot.completeRoots).toEqual([]); + expect(snapshot.coverage.enumeration_omissions).toEqual([ + { + path: "root_0:z-blocked", + reason: "directory_unavailable", + system_code: "EACCES", + }, + ]); + expect(snapshot.truncated).toBe(true); + } finally { + await chmod(blocked, 0o700); + await rm(root, { recursive: true, force: true }); + } + }, +); + it.skipIf(process.platform === "win32" || process.getuid?.() === 0)( "retains unreadable-file and sibling observations in a complete snapshot", async () => { diff --git a/src/process/capture/FilesystemSnapshot.test.ts b/src/process/capture/FilesystemSnapshot.test.ts index 16799b973..3aeab0391 100644 --- a/src/process/capture/FilesystemSnapshot.test.ts +++ b/src/process/capture/FilesystemSnapshot.test.ts @@ -1,6 +1,8 @@ import { fstatSync } from "node:fs"; import { mkdtemp, + mkdir, + symlink, lstat, open, readdir, @@ -35,6 +37,27 @@ const countOpenDescriptorsFor = async (path: string): Promise => { return matches.filter(Boolean).length; }; +it.skipIf(process.platform === "win32")( + "omits a captured regular file replaced by a symlink", + async () => { + const directory = await mkdtemp(join(tmpdir(), "rea-snapshot-symlink-")); + const selected = join(directory, "selected"); + const replacement = join(directory, "replacement"); + try { + await writeFile(selected, "captured bytes"); + await writeFile(replacement, "unselected bytes"); + const expected = await lstat(selected); + await rm(selected); + await symlink(replacement, selected); + const outcome = await hashFile(selected, expected, 1_000); + expect(outcome.state).toBe("omitted"); + expect(outcome).not.toHaveProperty("sha256"); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }, +); + it("does not hash a replacement file using the earlier path metadata", async () => { const directory = await mkdtemp(join(tmpdir(), "rea-snapshot-identity-")); const path = join(directory, "observed-file"); @@ -154,3 +177,43 @@ it.skipIf(process.platform === "win32")( } }, ); + +for (const variant of [ + "unchanged", + "case-alias", + "ancestor-junction", + "replaced-junction", +] as const) { + it.runIf(process.platform === "win32")( + `hashes only the captured Windows file through ${variant}`, + async () => { + const directory = await mkdtemp(join(tmpdir(), "rea-snapshot-windows-")); + try { + const original = join(directory, "original"); + const replacement = join(directory, "replacement"); + await mkdir(original); + await mkdir(replacement); + await writeFile(join(original, "TestFile.txt"), "captured bytes"); + await writeFile(join(replacement, "TestFile.txt"), "replaced bytes"); + const alias = join(directory, "alias"); + if (variant === "ancestor-junction" || variant === "replaced-junction") + await symlink(original, alias, "junction"); + const path = join( + variant.includes("junction") ? alias : original, + variant === "case-alias" ? "testfile.txt" : "TestFile.txt", + ); + const expected = await lstat(path); + if (variant === "replaced-junction") { + await rm(alias); + await symlink(replacement, alias, "junction"); + } + const result = await hashFile(path, expected, 1_000); + expect(result.state).toBe( + variant === "replaced-junction" ? "omitted" : "hashed", + ); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }, + ); +} diff --git a/src/process/capture/FilesystemSnapshot.ts b/src/process/capture/FilesystemSnapshot.ts index 5740f6525..3ec2e0716 100644 --- a/src/process/capture/FilesystemSnapshot.ts +++ b/src/process/capture/FilesystemSnapshot.ts @@ -74,12 +74,14 @@ export const hashFile = async ( signal?: AbortSignal, ): Promise => { signal?.throwIfAborted(); + const noFollowAvailable = + typeof fsConstants.O_NOFOLLOW === "number" && fsConstants.O_NOFOLLOW !== 0; let handle: FileHandle; try { handle = await open( path, fsConstants.O_RDONLY | - (fsConstants.O_NOFOLLOW ?? 0) | + (noFollowAvailable ? fsConstants.O_NOFOLLOW : 0) | (fsConstants.O_NONBLOCK ?? 0), ); } catch (cause: unknown) { @@ -94,6 +96,18 @@ export const hashFile = async ( } try { signal?.throwIfAborted(); + // On hosts without O_NOFOLLOW, compare the leaf's identity with the + // captured regular file. Ancestor aliases do not change file identity. + if (!noFollowAvailable) { + const named = await lstat(path); + signal?.throwIfAborted(); + if (!hasSameFileState(expected, named)) + return { + state: "omitted", + reason: "file_changed_or_short_read", + system_code: null, + }; + } const stats = await handle.stat(); signal?.throwIfAborted(); if (!hasSameFileState(expected, stats) || stats.size > maxBytes) @@ -124,7 +138,13 @@ export const hashFile = async ( } const after = await handle.stat(); signal?.throwIfAborted(); - if (!hasSameFileState(stats, after)) + let unchanged = hasSameFileState(stats, after); + if (unchanged && !noFollowAvailable) { + const namedAfter = await lstat(path); + signal?.throwIfAborted(); + unchanged = hasSameFileState(after, namedAfter); + } + if (!unchanged) return { state: "omitted", reason: "file_changed_or_short_read", @@ -158,6 +178,7 @@ export const snapshotRoots = async ( const enumerationReasons = new Set< FilesystemCoverage["enumeration_reasons"][number] >(); + const enumerationOmissions: FilesystemCoverage["enumeration_omissions"] = []; const hashOmissions: FilesystemCoverage["hash_omissions"] = []; const visit = async ( root: string, @@ -252,7 +273,23 @@ export const snapshotRoots = async ( let complete = true; const remainingEntries = scenario.limits.files - entries.length - pendingNames; - const directory = await opendir(path); + let directory; + try { + directory = await opendir(path); + } catch (cause: unknown) { + signal?.throwIfAborted(); + const systemCode = fileUnavailableSystemCode(cause); + if (systemCode === undefined) throw cause; + truncated = true; + complete = false; + enumerationReasons.add("directory_unavailable"); + enumerationOmissions.push({ + path: `${rootAlias}:${relativePath}`, + reason: "directory_unavailable", + system_code: systemCode, + }); + return false; + } for await (const child of directory) { signal?.throwIfAborted(); if ( @@ -305,6 +342,7 @@ export const snapshotRoots = async ( enumeration_truncated: completeRoots.length !== scenario.filesystem_observation_paths.length, enumeration_reasons: [...enumerationReasons], + enumeration_omissions: enumerationOmissions, hash_budget_bytes: scenario.limits.file_bytes, hashed_bytes: scenario.limits.file_bytes - remainingBytes, hash_omissions: hashOmissions, diff --git a/src/process/capture/ProcessCaptureLifecycle.ts b/src/process/capture/ProcessCaptureLifecycle.ts index ab9271dda..0e1ca73ee 100644 --- a/src/process/capture/ProcessCaptureLifecycle.ts +++ b/src/process/capture/ProcessCaptureLifecycle.ts @@ -38,7 +38,6 @@ import { assertNotCancelled } from "./ProcessScenarioRuntimeValidation.js"; import { normalizeProcessElapsedTime, normalizeProcessSamples, - normalizeProcessText, } from "./ProcessNormalization.js"; import { PROCESS_PROVIDER } from "../../domain/process/processEvidenceProvider.js"; import { TerminalRenderer } from "./TerminalRenderer.js"; @@ -1302,9 +1301,9 @@ export const captureTerminalFrames = (options: { readonly scenario: ProcessScenario; readonly frames: TerminalFrame[]; readonly started: number; - readonly temporaryRoot: string; readonly onOutput: () => void; readonly renderer: TerminalRenderer; + readonly normalizeText: (value: string) => string; readonly recordEvent: RecordProcessCaptureEvent; }): (() => TerminalRetention) => { let outputBytes = 0; @@ -1323,12 +1322,7 @@ export const captureTerminalFrames = (options: { monotonicProcessCaptureClock() - options.started, options.scenario.normalization.time_bucket_ms, ); - const normalized = normalizeProcessText( - data, - options.scenario, - options.temporaryRoot, - options.terminal.pid, - ); + const normalized = options.normalizeText(data); const sequence = options.frames.length; options.frames.push({ sequence, diff --git a/src/process/capture/ProcessHarness.ts b/src/process/capture/ProcessHarness.ts index 847e512f8..d12e66837 100644 --- a/src/process/capture/ProcessHarness.ts +++ b/src/process/capture/ProcessHarness.ts @@ -24,7 +24,7 @@ import { startProcessSampler } from "./ProcessSampling.js"; import { snapshotRoots } from "./FilesystemSnapshot.js"; import { TerminalRenderer } from "./TerminalRenderer.js"; import type { ProcessTimer } from "./ProcessTimer.js"; -import { normalizeProcessText } from "./ProcessNormalization.js"; +import { createProcessTextNormalizer } from "./ProcessNormalization.js"; import { selectCapturedProcessGroupIds } from "../ProcessOwnershipProcessTree.js"; import { awaitTerminalExit, @@ -222,8 +222,7 @@ const cleanupFailedStartup = async (options: { const createTerminalRenderer = ( scenario: ProcessScenario, - temporaryRoot: string, - terminalPid: () => number, + normalizeText: (value: string) => string, recordEvent: RecordProcessCaptureEvent, ): TerminalRenderer => new TerminalRenderer({ @@ -231,8 +230,7 @@ const createTerminalRenderer = ( rows: scenario.terminal.rows, scrollback: scenario.terminal.scrollback, maxBytes: scenario.limits.output_bytes, - normalize: (value) => - normalizeProcessText(value, scenario, temporaryRoot, terminalPid()), + normalize: normalizeText, recordEvent, }); @@ -271,10 +269,14 @@ const startCaptureRuntime = async ( const startedAt = new Date(); const started = monotonicProcessCaptureClock(); let lastOutput = started; - renderer = createTerminalRenderer( + const normalizeText = createProcessTextNormalizer( scenario, options.temporaryRoot, () => terminal?.pid ?? -1, + ); + renderer = createTerminalRenderer( + scenario, + normalizeText, options.recordEvent, ); terminal = spawn(scenario.executable, [...scenario.arguments], { @@ -299,6 +301,7 @@ const startCaptureRuntime = async ( started, onOutput: () => (lastOutput = monotonicProcessCaptureClock()), renderer, + normalizeText, }); scheduleScenarioInteractions({ ...options, diff --git a/src/process/capture/ProcessNormalization.test.ts b/src/process/capture/ProcessNormalization.test.ts index e5ef61062..54b042dd4 100644 --- a/src/process/capture/ProcessNormalization.test.ts +++ b/src/process/capture/ProcessNormalization.test.ts @@ -1,6 +1,7 @@ import { expect, it } from "vitest"; import { + createProcessTextNormalizer, normalizeProcessSamples, normalizeProcessText, } from "./ProcessNormalization.js"; @@ -94,6 +95,42 @@ it("replaces caller-declared patterns literally, without expanding replacement p ).toBe("listening on $&$&; $$; $1=alpha; tail $1"); }); +it("reuses matchers while observing the terminal PID after launch", () => { + let pid = -1; + const normalize = createProcessTextNormalizer( + scenario({ ports: false, patterns: [] }), + "/temporary", + () => pid, + ); + expect(normalize("root 41010 /workspace/item")).toBe( + "root 41010 /item", + ); + pid = 41_010; + expect(normalize("root 41010 /workspace/item")).toBe( + "root /item", + ); + pid = 42_020; + expect(normalize("root 41010 peer 42020")).toBe("root 41010 peer "); +}); + +it("keeps equal-path priority and replacement aliases local to each capture", () => { + const declared = parseProcessScenario({ + executable: "/selected/tool", + working_directory: "/selected", + filesystem_observation_paths: ["/selected", ""], + normalization: { paths: true, pids: false, ports: false, patterns: [] }, + }); + const normalize = createProcessTextNormalizer(declared, "/selected", -1); + const other = createProcessTextNormalizer(declared, "/other", -1); + const text = "/selected/item /source"; + expect(normalize(text)).toBe( + "/item /source", + ); + expect(other(text)).toBe( + "/item /source", + ); +}); + it("preserves empty-pattern boundaries, including empty text", () => { const declared = scenario({ patterns: [{ pattern: "", replacement: "$$" }], diff --git a/src/process/capture/ProcessNormalization.ts b/src/process/capture/ProcessNormalization.ts index 74db31b1c..85dc40929 100644 --- a/src/process/capture/ProcessNormalization.ts +++ b/src/process/capture/ProcessNormalization.ts @@ -9,36 +9,51 @@ export const normalizeProcessElapsedTime = ( timeBucketMs: number, ): number => Math.floor(elapsedMs / timeBucketMs) * timeBucketMs; -/** Normalize one terminal/protocol payload under explicitly selected scenario rules. */ +/** Prepare capture-local matchers while retaining a live terminal PID lookup. */ +export const createProcessTextNormalizer = ( + scenario: ProcessScenario, + temporaryRoot: string, + pid: number | (() => number), +): ((value: string) => string) => { + const normalizePaths = scenario.normalization.paths + ? createPathNormalizer(scenario, temporaryRoot) + : (value: string): string => value; + let previousPid: number | undefined; + let normalizePids = (value: string): string => value; + return (value) => { + let normalized = normalizePaths(value); + if (scenario.normalization.ports) + normalized = normalizePortTokens(normalized); + if (scenario.normalization.pids) { + const currentPid = typeof pid === "number" ? pid : pid(); + if (currentPid !== previousPid) { + previousPid = currentPid; + normalizePids = createPidNormalizer([currentPid]); + } + normalized = normalizePids(normalized); + } + for (const pattern of scenario.normalization.patterns) + normalized = normalized.replaceAll( + pattern.pattern, + () => pattern.replacement, + ); + return normalized; + }; +}; + +/** Normalize one payload when its caller does not retain a capture-local matcher. */ export const normalizeProcessText = ( value: string, scenario: ProcessScenario, temporaryRoot: string, pid: number, -): string => { - let normalized = value; - if (scenario.normalization.paths) - normalized = normalizePathTokens(normalized, scenario, temporaryRoot); - if (scenario.normalization.ports) - normalized = normalizePortTokens(normalized); - if (scenario.normalization.pids) - normalized = normalizePidTokens(normalized, [pid]); - // A callback keeps replacement text literal while preserving replaceAll's - // empty-pattern boundaries, which the process scenario contract permits. - for (const pattern of scenario.normalization.patterns) - normalized = normalized.replaceAll( - pattern.pattern, - () => pattern.replacement, - ); - return normalized; -}; +): string => createProcessTextNormalizer(scenario, temporaryRoot, pid)(value); /** Match original paths once, preferring the most specific selected identity. */ -const normalizePathTokens = ( - value: string, +const createPathNormalizer = ( scenario: ProcessScenario, temporaryRoot: string, -): string => { +): ((value: string) => string) => { const aliases = new Map(); const selected: readonly (readonly [string, string])[] = [ [temporaryRoot, ""], @@ -56,14 +71,12 @@ const normalizePathTokens = ( const paths = [...aliases.keys()].sort( (left, right) => right.length - left.length, ); - if (paths.length === 0) return value; + if (paths.length === 0) return (value) => value; const pattern = paths .map((path) => path.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&")) .join("|"); - return value.replace( - new RegExp(pattern, "gu"), - (path) => aliases.get(path) ?? path, - ); + const matcher = new RegExp(pattern, "gu"); + return (value) => value.replace(matcher, (path) => aliases.get(path) ?? path); }; // A port may end a sentence, but a period followed by a letter, digit, or @@ -109,16 +122,17 @@ export const normalizeProcessSamples = ( for (const identifier of identifiers) if (identifier > 0 && !mapping.has(identifier)) mapping.set(identifier, mapping.size + 1); + const normalizeText = createProcessTextNormalizer( + scenario, + "", + rootPid, + ); + const normalizeSamplePids = scenario.normalization.pids + ? createPidNormalizer(identifiers) + : undefined; const normalizeCommand = (command: string): string => { - const normalized = normalizeProcessText( - command, - scenario, - "", - rootPid, - ); - const withNormalizedPids = scenario.normalization.pids - ? normalizePidTokens(normalized, identifiers) - : normalized; + const normalized = normalizeText(command); + const withNormalizedPids = normalizeSamplePids?.(normalized) ?? normalized; return withNormalizedPids.trim(); }; return samples.map((sample) => ({ @@ -144,17 +158,14 @@ export const normalizeProcessSamples = ( })); }; -const normalizePidTokens = ( - value: string, +const createPidNormalizer = ( identifiers: readonly number[], -): string => { +): ((value: string) => string) => { const tokens = [...new Set(identifiers)] .filter((identifier) => Number.isSafeInteger(identifier) && identifier > 0) .sort((left, right) => right - left) .map(String); - if (tokens.length === 0) return value; - return value.replace( - new RegExp(`(?", - ); + if (tokens.length === 0) return (value) => value; + const matcher = new RegExp(`(? value.replace(matcher, ""); }; diff --git a/src/windows/WindowsOwnedProcess.test.ts b/src/windows/WindowsOwnedProcess.test.ts index 5beec9520..2ec1452bb 100644 --- a/src/windows/WindowsOwnedProcess.test.ts +++ b/src/windows/WindowsOwnedProcess.test.ts @@ -1,5 +1,5 @@ import { once } from "node:events"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import type { WindowsNativeAuthority } from "./WindowsNativeLoader.js"; import { @@ -82,6 +82,7 @@ const authority = ( : {}), }; if (operation === "process_poll") return polls.shift(); + if (operation === "process_terminate") return true; if (operation === "process_close") return null; throw new Error(`Unmodeled native operation ${operation}`); }, @@ -205,3 +206,223 @@ describe("Windows Job Object memory limits", () => { expect(() => launch(native, LIMIT)).toThrow(); }); }); + +describe("Windows output capture drain", () => { + it("repolls without the 25ms idle delay while the child keeps producing output", async () => { + vi.useFakeTimers(); + try { + const chunk = Buffer.alloc(8192, 0x61); + const polls: Array<{ + stdout: Buffer; + stderr: Buffer; + stdoutEnded: boolean; + stderrEnded: boolean; + exitCode: number | null; + activeProcesses: number; + }> = Array.from({ length: 10 }, () => ({ + stdout: chunk, + stderr: Buffer.alloc(0), + stdoutEnded: false, + stderrEnded: false, + exitCode: null, + activeProcesses: 1, + })); + polls.push({ ...settledPoll }); + const { native, calls } = authority({ polls }); + const child = launch(native); + child.stdout.resume(); + const closed = once(child, "close"); + // 100ms of simulated time: the fixed 25ms poll cadence can complete + // at most ~4 polls, so all 10 data polls plus the settle poll must + // still be pending. Draining repolls immediately while data flows. + await vi.advanceTimersByTimeAsync(100); + expect( + calls.filter((call) => call.operation === "process_poll"), + ).toHaveLength(11); + await closed; + } finally { + vi.useRealTimers(); + } + }); + + it("keeps the 25ms idle delay after a poll with no output", async () => { + vi.useFakeTimers(); + try { + const idle = { + stdout: Buffer.alloc(0), + stderr: Buffer.alloc(0), + stdoutEnded: false, + stderrEnded: false, + exitCode: null, + activeProcesses: 1, + }; + const { native, calls } = authority({ + polls: [idle, { ...settledPoll }], + }); + const child = launch(native); + const closed = once(child, "close"); + const pollCount = () => + calls.filter((call) => call.operation === "process_poll").length; + // The constructor schedules the first poll 25ms out. + await vi.advanceTimersByTimeAsync(25); + expect(pollCount()).toBe(1); + // An idle child must not be repolled early: 10ms of simulated time + // must not reach the next poll, the remaining 15ms must. + await vi.advanceTimersByTimeAsync(10); + expect(pollCount()).toBe(1); + await vi.advanceTimersByTimeAsync(15); + expect(pollCount()).toBe(2); + await closed; + } finally { + vi.useRealTimers(); + } + }); + + it("repolls without the 25ms idle delay while only stderr carries output", async () => { + vi.useFakeTimers(); + try { + const chunk = Buffer.alloc(4096, 0x62); + const { native, calls } = authority({ + polls: [ + { + stdout: Buffer.alloc(0), + stderr: chunk, + stdoutEnded: false, + stderrEnded: false, + exitCode: null, + activeProcesses: 1, + }, + { ...settledPoll }, + ], + }); + const child = launch(native); + const closed = once(child, "close"); + const pollCount = () => + calls.filter((call) => call.operation === "process_poll").length; + await vi.advanceTimersByTimeAsync(25); + expect(pollCount()).toBe(1); + // stderr-only output still means the child is producing: the next + // poll must be immediate, not 25ms out. + await vi.advanceTimersByTimeAsync(1); + expect(pollCount()).toBe(2); + await closed; + } finally { + vi.useRealTimers(); + } + }); +}); + +describe("Windows output backpressure and cleanup", () => { + it("stops native reads until both paused output consumers drain", async () => { + vi.useFakeTimers(); + try { + const chunk = Buffer.alloc(65536, 0x63); + const { native, calls } = authority({ + polls: [ + { + stdout: chunk, + stderr: chunk, + stdoutEnded: false, + stderrEnded: false, + exitCode: null, + activeProcesses: 1, + }, + { ...settledPoll, stderr: Buffer.alloc(0) }, + ], + }); + const child = launch(native); + const closed = once(child, "close"); + child.stdout.pause(); + child.stderr.pause(); + const pollCount = () => + calls.filter((call) => call.operation === "process_poll").length; + await vi.advanceTimersByTimeAsync(25); + expect(child.stdout.writableNeedDrain).toBe(true); + expect(child.stderr.writableNeedDrain).toBe(true); + await vi.advanceTimersByTimeAsync(5000); + expect(pollCount()).toBe(1); + const stdoutDrained = once(child.stdout, "drain"); + child.stdout.resume(); + await stdoutDrained; + await vi.advanceTimersByTimeAsync(25); + expect(pollCount()).toBe(1); + const stderrDrained = once(child.stderr, "drain"); + child.stderr.resume(); + await stderrDrained; + await vi.advanceTimersByTimeAsync(1); + await closed; + expect(pollCount()).toBe(2); + } finally { + vi.useRealTimers(); + } + }); + + it("discards blocked output during cleanup so the owned job can settle", async () => { + vi.useFakeTimers(); + try { + const chunk = Buffer.alloc(65536, 0x64); + const { native, calls } = authority({ + polls: [ + { + stdout: chunk, + stderr: Buffer.alloc(0), + stdoutEnded: false, + stderrEnded: false, + exitCode: null, + activeProcesses: 1, + }, + { ...settledPoll, stderr: Buffer.alloc(0) }, + ], + }); + const child = launch(native); + child.stdout.pause(); + await vi.advanceTimersByTimeAsync(25); + expect(child.stdout.writableNeedDrain).toBe(true); + const closed = once(child, "close"); + const cleanup = child.cleanup(); + await vi.advanceTimersByTimeAsync(25); + expect(await cleanup).toEqual({ cleaned: true, signaled: true }); + await closed; + expect( + calls.filter((call) => call.operation === "process_poll"), + ).toHaveLength(2); + expect(child.stdout.destroyed).toBe(true); + expect(child.outputIncomplete).toBe(true); + } finally { + vi.useRealTimers(); + } + }); + + it("resumes polling when a blocked output stream is destroyed", async () => { + vi.useFakeTimers(); + try { + const chunk = Buffer.alloc(65536, 0x65); + const { native, calls } = authority({ + polls: [ + { + stdout: chunk, + stderr: Buffer.alloc(0), + stdoutEnded: false, + stderrEnded: false, + exitCode: null, + activeProcesses: 1, + }, + { ...settledPoll, stderr: Buffer.alloc(0) }, + ], + }); + const child = launch(native); + const closed = once(child, "close"); + await vi.advanceTimersByTimeAsync(25); + expect(child.stdout.writableNeedDrain).toBe(true); + child.stdout.destroy(); + await vi.advanceTimersByTimeAsync(0); + await closed; + expect( + calls.filter((call) => call.operation === "process_poll"), + ).toHaveLength(2); + expect(child.outputIncomplete).toBe(true); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/src/windows/WindowsOwnedProcess.ts b/src/windows/WindowsOwnedProcess.ts index b6a530461..062f3b78e 100644 --- a/src/windows/WindowsOwnedProcess.ts +++ b/src/windows/WindowsOwnedProcess.ts @@ -67,11 +67,14 @@ export class WindowsOwnedProcess extends EventEmitter { #closed = false; #failure: string | undefined; #timer: NodeJS.Timeout | undefined; + #blockedWake: NodeJS.Timeout | undefined; + readonly #blockedOutput = new Set(); #unreferenced = false; #cleanup: Promise | undefined; readonly #memoryLimitBytes: number | undefined; #memoryLimitReached = false; #peakJobMemoryBytes: number | undefined; + #outputIncomplete = false; constructor( command: string, @@ -179,6 +182,11 @@ export class WindowsOwnedProcess extends EventEmitter { return this.#peakJobMemoryBytes; } + /** True when cleanup destroyed a backpressured stream before all output was observed. */ + get outputIncomplete(): boolean { + return this.#outputIncomplete; + } + async #writeInput(chunk: unknown): Promise { const bytes = z.instanceof(Buffer).parse(chunk); const written = z @@ -227,12 +235,14 @@ export class WindowsOwnedProcess extends EventEmitter { unref(): void { this.#unreferenced = true; this.#timer?.unref(); + this.#blockedWake?.unref(); } /** Restore event-loop references when retrying process lifecycle observation. */ ref(): void { this.#unreferenced = false; this.#timer?.ref(); + this.#blockedWake?.ref(); } async #stop(): Promise { @@ -240,6 +250,7 @@ export class WindowsOwnedProcess extends EventEmitter { let timer: NodeJS.Timeout | undefined; try { const signaled = this.kill(); + this.#discardBlockedOutput(); this.#schedule(); const settled = await Promise.race([ this.#settled.then(() => true), @@ -265,12 +276,50 @@ export class WindowsOwnedProcess extends EventEmitter { } } - #schedule(): void { - if (this.#closed || this.#timer !== undefined) return; - this.#timer = setTimeout(() => this.#poll(), 25); + #schedule(delay = 25): void { + if ( + this.#closed || + this.#timer !== undefined || + this.#blockedOutput.size > 0 + ) + return; + this.#timer = setTimeout(() => this.#poll(), delay); if (this.#unreferenced) this.#timer.unref(); } + #scheduleBlockedWake(): void { + if ( + this.#closed || + this.#blockedOutput.size === 0 || + this.#blockedWake !== undefined + ) + return; + this.#blockedWake = setTimeout(() => { + this.#blockedWake = undefined; + this.#scheduleBlockedWake(); + }, 1_000); + if (this.#unreferenced) this.#blockedWake.unref(); + } + + #clearBlockedWake(): void { + if (this.#blockedWake !== undefined) clearTimeout(this.#blockedWake); + this.#blockedWake = undefined; + } + + #unblockOutput(stream: PassThrough): void { + if (stream.destroyed) this.#outputIncomplete = true; + this.#blockedOutput.delete(stream); + if (this.#blockedOutput.size === 0) this.#clearBlockedWake(); + this.#schedule(0); + } + + #discardBlockedOutput(): void { + if (this.#blockedOutput.size > 0) this.#outputIncomplete = true; + for (const stream of this.#blockedOutput) stream.destroy(); + this.#blockedOutput.clear(); + this.#clearBlockedWake(); + } + #poll(): void { this.#timer = undefined; if (this.#closed) return; @@ -289,8 +338,11 @@ export class WindowsOwnedProcess extends EventEmitter { state = limited; } this.#failure = undefined; - if (state.stdout.length > 0) this.stdout.write(state.stdout); - if (state.stderr.length > 0) this.stderr.write(state.stderr); + // Stop native reads while either consumer is backpressured. A slower + // cadence still accumulates unbounded unread output in the parent heap. + const idle = state.stdout.length === 0 && state.stderr.length === 0; + this.#writeOutput(this.stdout, state.stdout); + this.#writeOutput(this.stderr, state.stderr); if (state.exitCode !== null && this.exitCode === null) { this.exitCode = state.exitCode; this.emit("exit", state.exitCode, null); @@ -302,7 +354,7 @@ export class WindowsOwnedProcess extends EventEmitter { state.activeProcesses === 0 ) { this.#release(); - } else this.#schedule(); + } else this.#schedule(idle ? 25 : 0); } catch (cause: unknown) { const failure = cause instanceof Error ? cause : new Error(String(cause)); this.#failure = failure.message; @@ -313,14 +365,33 @@ export class WindowsOwnedProcess extends EventEmitter { } } + #writeOutput(stream: PassThrough, bytes: Buffer): void { + if (bytes.length === 0 || stream.writableEnded) return; + if (stream.destroyed) { + this.#outputIncomplete = true; + return; + } + if (stream.write(bytes)) return; + this.#blockedOutput.add(stream); + const unblock = () => { + stream.off("drain", unblock); + stream.off("close", unblock); + this.#unblockOutput(stream); + }; + stream.once("drain", unblock); + stream.once("close", unblock); + this.#scheduleBlockedWake(); + } + #release(): void { if (this.#closed) return; if (this.#timer !== undefined) clearTimeout(this.#timer); + this.#clearBlockedWake(); this.#authority.call("process_close", [this.#handle]); this.#closed = true; this.stdin?.destroy(); - this.stdout.end(); - this.stderr.end(); + if (!this.stdout.destroyed) this.stdout.end(); + if (!this.stderr.destroyed) this.stderr.end(); this.#settle?.(); this.emit("close", this.exitCode, null); } diff --git a/tests/boundary/process/cleanLinuxHopperRuntime.test.ts b/tests/boundary/process/cleanLinuxHopperRuntime.test.ts new file mode 100644 index 000000000..410d713e4 --- /dev/null +++ b/tests/boundary/process/cleanLinuxHopperRuntime.test.ts @@ -0,0 +1,411 @@ +import { execFile } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { + chmod, + copyFile, + mkdir, + readFile, + symlink, + unlink, + writeFile, +} from "node:fs/promises"; +import { join } from "node:path"; +import { promisify } from "node:util"; + +import { expect, it } from "vitest"; + +import { assertCleanLinuxHopperRuntime } from "../../../scripts/lib/clean-linux-hopper-runtime.mjs"; +import { cleanupOwnedProcessGroup } from "../../../src/process/ProcessOwnership.js"; +import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; + +const execFileAsync = promisify(execFile); + +it.runIf(process.platform === "linux").each([ + ["Xvfb", "Clean Linux runner retained a Hopper demo process"], + ["Hopper", "Clean Linux runner retained a Hopper demo process"], + ["python3", "Cannot identify a command-less Python process"], + ["unrelated", undefined], +] as const)( + "observes an unreaped %s without signaling or reaping it", + async (name, refusal) => { + const directory = await createTestTempDirectory("rea-runtime-zombie-"); + const procRoot = join(directory, "proc"); + await mkdir(procRoot); + const executable = join(directory, name); + await copyFile("/usr/bin/true", executable); + await chmod(executable, 0o700); + // WNOWAIT observes exit without reaping. The parent keeps the zombie until + // the test closes its stdin, then waitpid performs the actual cleanup. + const parent = execFileAsync( + "/usr/bin/python3", + [ + "-c", + ` +import os, sys +pid = os.fork() +if pid == 0: + os.execl(sys.argv[1], sys.argv[1]) +try: + os.waitid(os.P_PID, pid, os.WEXITED | os.WNOWAIT) + print(pid, flush=True) + sys.stdin.buffer.read(1) +finally: + os.waitpid(pid, 0) +`, + executable, + ], + { timeout: 10_000 }, + ); + const ready = new Promise((resolve, reject) => { + parent.child.stdout?.once("data", (data: Buffer) => + resolve(Number(data.toString().trim())), + ); + parent.catch(reject); + }); + let zombiePid: number | undefined; + try { + zombiePid = await ready; + expect(Number.isSafeInteger(zombiePid)).toBe(true); + const prefix = `/proc/${String(zombiePid)}`; + const procEntry = join(procRoot, String(zombiePid)); + await mkdir(procEntry); + for (const name of ["cmdline", "comm"]) + await symlink(`${prefix}/${name}`, join(procEntry, name)); + expect(await readFile(`${prefix}/cmdline`, "utf8")).toBe(""); + expect((await readFile(`${prefix}/comm`, "utf8")).trim()).toBe(name); + const observation = assertCleanLinuxHopperRuntime([], procRoot); + if (refusal === undefined) + await expect(observation).resolves.toBeUndefined(); + else await expect(observation).rejects.toThrow(refusal); + expect(await readFile(`${prefix}/stat`, "utf8")).toContain( + `(${name}) Z `, + ); + } finally { + parent.child.stdin?.end("reap"); + await parent; + if (zombiePid !== undefined) + await expect( + readFile(`/proc/${String(zombiePid)}/stat`, "utf8"), + ).rejects.toMatchObject({ code: "ENOENT" }); + } + }, +); + +it.runIf(process.platform === "linux")( + "does not treat runtime names in another process's arguments as a runtime", + async () => { + const directory = await createTestTempDirectory("rea-runtime-arguments-"); + const procRoot = join(directory, "proc"); + await mkdir(procRoot); + const observer = new URL( + "../../../scripts/lib/clean-linux-hopper-runtime.mjs", + import.meta.url, + ).href; + const code = ` + import { mkdir, symlink } from "node:fs/promises"; + import { join } from "node:path"; + const { assertCleanLinuxHopperRuntime } = await import(${JSON.stringify(observer)}); + const entry = join(${JSON.stringify(procRoot)}, String(process.pid)); + await mkdir(entry); + await symlink("/proc/" + process.pid + "/cmdline", join(entry, "cmdline")); + await symlink("/proc/" + process.pid + "/comm", join(entry, "comm")); + await assertCleanLinuxHopperRuntime([], ${JSON.stringify(procRoot)}); + console.log('clean'); + `; + const { stdout } = await execFileAsync( + process.execPath, + [ + "--input-type=module", + "-e", + code, + procRoot, + "Xvfb", + "/opt/hopper/bin/Hopper", + "hopper-demo-x11.py", + ], + { timeout: 10_000 }, + ); + expect(stdout.trim()).toBe("clean"); + }, +); + +it.runIf(process.platform === "linux")( + "distinguishes an ordinary Node script with the helper's filename", + async () => { + const directory = await createTestTempDirectory("rea-runtime-arguments-"); + const procRoot = join(directory, "proc"); + await mkdir(procRoot); + const script = join(directory, "hopper-demo-x11.py"); + const observer = new URL( + "../../../scripts/lib/clean-linux-hopper-runtime.mjs", + import.meta.url, + ).href; + const code = ` + import { mkdir, symlink } from "node:fs/promises"; + import { join } from "node:path"; + const entry = join(${JSON.stringify(procRoot)}, String(process.pid)); + await mkdir(entry); + await symlink("/proc/" + process.pid + "/cmdline", join(entry, "cmdline")); + await symlink("/proc/" + process.pid + "/comm", join(entry, "comm")); + const { assertCleanLinuxHopperRuntime } = await import(${JSON.stringify(observer)}); + await assertCleanLinuxHopperRuntime([], ${JSON.stringify(procRoot)}); + console.log('clean'); + `; + const { stdout } = await execFileAsync( + process.execPath, + ["--input-type=module", "--eval", code, script], + { timeout: 10_000 }, + ); + expect(stdout.trim()).toBe("clean"); + }, +); + +const launchReapedOrphan = (runId: string, parentCode: string) => { + // The short-lived Node launcher still exits before observation. A retained + // subreaper adopts its child and waits after owned cleanup, even when the + // container's PID 1 does not reap. It is a sibling of the verifier below. + const reaper = execFileAsync( + "/usr/bin/python3", + [ + "-c", + ` +import ctypes, json, os, subprocess, sys +libc = ctypes.CDLL(None, use_errno=True) +if libc.prctl(36, 1, 0, 0, 0) != 0: # PR_SET_CHILD_SUBREAPER + raise OSError(ctypes.get_errno(), "PR_SET_CHILD_SUBREAPER") +launcher = subprocess.run(sys.argv[1:], capture_output=True, text=True, check=True) +child = json.loads(launcher.stdout) +try: + print(json.dumps(child), flush=True) + sys.stdin.buffer.read(1) +finally: + os.waitpid(child["pid"], 0) +`, + process.execPath, + "-e", + parentCode, + ], + { + env: { ...process.env, REA_PROCESS_RUN_ID: runId }, + timeout: 40_000, + }, + ); + const ready = new Promise<{ pid: number; parentPid: number }>( + (resolve, reject) => { + reaper.child.stdout?.once("data", (data: Buffer) => { + try { + resolve(JSON.parse(data.toString())); + } catch (error) { + reject(error); + } + }); + reaper.then( + () => reject(new Error("Fixture reaper exited before readiness")), + reject, + ); + }, + ); + return { reaper, ready }; +}; + +const observeFromSeparateVerifier = async ( + runId: string, + launcherPaths: string[], + procRoot: string, +) => { + const observer = new URL( + "../../../scripts/lib/clean-linux-hopper-runtime.mjs", + import.meta.url, + ).href; + const verifierRun = new URL( + "../../../scripts/lib/verifier-run.mjs", + import.meta.url, + ).href; + const { stdout: observed } = await execFileAsync( + process.execPath, + [ + "--input-type=module", + "-e", + ` +const { assertCleanLinuxHopperRuntime } = await import(${JSON.stringify(observer)}); +const { completeVerifierRun } = await import(${JSON.stringify(verifierRun)}); +const { process_lineage } = await completeVerifierRun({ + run_id: ${JSON.stringify(runId)}, verifier_pid: process.pid, parent_pid: process.ppid, +}); +let refusal; +try { await assertCleanLinuxHopperRuntime(${JSON.stringify(launcherPaths)}, ${JSON.stringify(procRoot)}); } +catch (error) { refusal = error.message; } +console.log(JSON.stringify({ process_lineage, refusal })); +`, + ], + { timeout: 10_000 }, + ); + return JSON.parse(observed) as { + process_lineage: { + status: string; + descendants: { pid: number }[]; + }; + refusal?: string; + }; +}; + +it.runIf(process.platform === "linux").each([ + ["Xvfb-named Node", null, true], + ["Python helper", [], true], + ["Python helper with -u", ["-u"], true], + [ + "Python helper with combined/value options", + ["-Bu", "-W", "ignore", "-X", "dev", "--check-hash-based-pycs", "always"], + true, + ], + ["Python helper after --", ["--"], true], + [ + "Python -c script argument", + ["-c", "import time; print('ready', flush=True); time.sleep(30)"], + false, + ], + ["Python -m script argument", ["-m", "unrelated"], false], + ["alternate launcher", null, true], +] as const)( + "classifies a reparented %s without signaling it", + async (mode, pythonOptions, retained) => { + const runId = randomUUID(); + const childCode = + "process.title = 'Xvfb'; setInterval(() => {}, 1000); process.send('ready');"; + const directory = await createTestTempDirectory("rea-clean-runtime-test-"); + const procRoot = join(directory, "proc"); + await mkdir(procRoot); + const launcher = join(directory, "Hopper"); + const alternate = join(directory, "Hopper-alternate"); + const helper = join(directory, "hopper-demo-x11.py"); + const alternateMode = mode === "alternate launcher"; + const pythonMode = pythonOptions !== null; + if (alternateMode) { + await copyFile("/usr/bin/sleep", launcher); + await chmod(launcher, 0o700); + await symlink(launcher, alternate); + } + if (pythonMode) { + await writeFile( + helper, + "import time\nprint('ready', flush=True)\ntime.sleep(30)\n", + ); + await copyFile(helper, join(directory, "unrelated.py")); + } + // The fixture must be the cause of a refusal; a retained host runtime + // would otherwise mask a broken classification or executable-identity check. + await assertCleanLinuxHopperRuntime( + alternateMode ? [launcher] : [], + procRoot, + ); + const command = alternateMode + ? alternate + : pythonMode + ? "/usr/bin/python3" + : process.execPath; + const arguments_ = alternateMode + ? ["30"] + : pythonMode + ? [...pythonOptions, helper] + : ["-e", childCode]; + const stdio = alternateMode + ? "ignore" + : pythonMode + ? ["ignore", "pipe", "ignore"] + : ["ignore", "ignore", "ignore", "ipc"]; + const parentCode = ` + const { spawn } = require('node:child_process'); + const child = spawn(${JSON.stringify(command)}, ${JSON.stringify(arguments_)}, { + detached: true, stdio: ${JSON.stringify(stdio)}, cwd: ${JSON.stringify(directory)} + }); + child.once('error', () => process.exit(1)); + const ready = () => { + process.stdout.write(JSON.stringify({ pid: child.pid, parentPid: process.pid })); + if (child.connected) child.disconnect(); + child.stdout?.destroy(); + child.unref(); + }; + ${pythonMode ? "child.stdout.once('data', ready);" : `child.once(${JSON.stringify(alternateMode ? "spawn" : "message")}, ready);`} + `; + const { reaper, ready } = launchReapedOrphan(runId, parentCode); + let orphanPid: number | undefined; + try { + const { pid, parentPid } = await ready; + orphanPid = pid; + const procEntry = join(procRoot, String(pid)); + await mkdir(procEntry); + for (const name of ["cmdline", "comm", "exe"]) + await symlink(`/proc/${String(pid)}/${name}`, join(procEntry, name)); + const { stdout: coordinates } = await execFileAsync("ps", [ + "-p", + String(pid), + "-o", + "ppid=,pgid=", + ]); + const [currentParent, group] = coordinates + .trim() + .split(/\s+/u) + .map(Number); + expect(currentParent).not.toBe(parentPid); + expect(currentParent).toBe(reaper.child.pid); + expect(group).toBe(pid); + + if (alternateMode) { + const commandLine = await readFile( + `/proc/${String(pid)}/cmdline`, + "utf8", + ); + expect(commandLine.split("\0")[0]).toBe(alternate); + await unlink(alternate); + } + if (pythonMode) { + const commandLine = await readFile( + `/proc/${String(pid)}/cmdline`, + "utf8", + ); + expect( + commandLine.split("\0").slice(0, pythonOptions.length + 2), + ).toEqual([command, ...pythonOptions, helper]); + } + const { process_lineage: lineage, refusal } = + await observeFromSeparateVerifier( + runId, + alternateMode ? [launcher] : [], + procRoot, + ); + expect(lineage.status).toBe("verified"); + expect(lineage.descendants.some((p) => p.pid === pid)).toBe(false); + if (retained) + expect(refusal).toBe( + "Clean Linux runner retained a Hopper demo process", + ); + else expect(refusal).toBeUndefined(); + // Observation must not terminate the fixture, even though it resembles Xvfb. + const { stdout: alive } = await execFileAsync("ps", [ + "-p", + String(pid), + "-o", + "stat=", + ]); + expect(alive.trim()).not.toMatch(/^Z/u); + } finally { + try { + if (orphanPid !== undefined) { + const cleanup = await cleanupOwnedProcessGroup({ + runId, + leaderPid: orphanPid, + processGroupId: orphanPid, + }); + expect(cleanup.cleaned, JSON.stringify(cleanup)).toBe(true); + } + } finally { + reaper.child.stdin?.end("reap"); + await reaper; + if (orphanPid !== undefined) + await expect( + readFile(`/proc/${String(orphanPid)}/stat`, "utf8"), + ).rejects.toMatchObject({ code: "ENOENT" }); + } + } + }, +); diff --git a/tests/boundary/process/providerProcess.test.ts b/tests/boundary/process/providerProcess.test.ts index b283df2a1..6e3b70995 100644 --- a/tests/boundary/process/providerProcess.test.ts +++ b/tests/boundary/process/providerProcess.test.ts @@ -18,7 +18,7 @@ import { spawnOwnedProviderProcess, type ProviderProcessDiagnostic, } from "../../../src/process/ProviderProcess.js"; -import { WINDOWS_NATIVE_AUTHORITY_UNAVAILABLE_REASON } from "../../../src/process/WindowsAuthority.js"; +import * as nativeLoader from "../../../src/windows/WindowsNativeLoader.js"; import { spawnProviderProcessFixture, stopProviderProcessFixture, @@ -71,12 +71,25 @@ describe("provider process runtime and wait primitives", () => { }); it("fails closed for a Windows runtime root without native DACL authority", async () => { - await expect( - PrivateRuntimeRoot.create({ platform: "win32" }), - ).rejects.toMatchObject({ - code: "private-runtime-root-authority-unavailable", - message: WINDOWS_NATIVE_AUTHORITY_UNAVAILABLE_REASON, - }); + const authority = vi + .spyOn(nativeLoader, "systemWindowsNativeAuthority") + .mockReturnValue({ + available: false, + failureStage: "self-test", + reason: "Native DACL self-test failed with Win32 error 145", + retryable: true, + }); + try { + await expect( + PrivateRuntimeRoot.create({ platform: "win32" }), + ).rejects.toMatchObject({ + code: "private-runtime-root-authority-unavailable", + message: "Native DACL self-test failed with Win32 error 145", + retryable: true, + }); + } finally { + authority.mockRestore(); + } }); it("classifies external cancellation during startup", async () => { diff --git a/tests/boundary/providers/hopper/bridgeBootstrapCleanup.test.ts b/tests/boundary/providers/hopper/bridgeBootstrapCleanup.test.ts new file mode 100644 index 000000000..81a8ffc5e --- /dev/null +++ b/tests/boundary/providers/hopper/bridgeBootstrapCleanup.test.ts @@ -0,0 +1,58 @@ +import { lstat } from "node:fs/promises"; +import { join } from "node:path"; + +import { expect, it } from "vitest"; + +import { HopperApplicationLauncher } from "../../../../src/hopper/BridgeLauncher.js"; +import { createTestTempDirectory } from "../../../fixtures/temporaryDirectory.js"; + +it.each(["failed", "cancelled"] as const)( + "removes the token-bearing bootstrap and releases the lease after a %s launch", + async (outcome) => { + const directory = await createTestTempDirectory("rea-bootstrap-cleanup-"); + const controller = new AbortController(); + if (outcome === "cancelled") controller.abort(); + let releases = 0; + const launcher = new HopperApplicationLauncher( + { + environment: {}, + launcherPath: join(directory, "nonexistent-hopper"), + targetPath: join(directory, "test-binary"), + targetKind: "executable", + loaderArgs: [], + bridgeScriptPath: join(directory, "bridge.py"), + launchMode: "native", + }, + { + platform: "darwin", + acquireTargetLease: async () => ({ + acquired: true, + lease: { + release: async () => { + releases++; + }, + }, + }), + }, + ); + const result = await launcher.launch( + { + directory, + socketPath: join(directory, "bridge.sock"), + token: "fixture-token", + runId: "bootstrap-cleanup", + }, + { signal: controller.signal }, + ); + expect(result.ok).toBe(false); + if (outcome === "cancelled") + expect(result).toMatchObject({ + ok: false, + error: { _tag: "HopperCancelledError" }, + }); + await expect(lstat(join(directory, "bootstrap.py"))).rejects.toMatchObject({ + code: "ENOENT", + }); + expect(releases).toBe(1); + }, +); diff --git a/tests/process-global/cleanLinuxHopperRuntime.test.ts b/tests/process-global/cleanLinuxHopperRuntime.test.ts index fdd19e8ee..55fdf0ec9 100644 --- a/tests/process-global/cleanLinuxHopperRuntime.test.ts +++ b/tests/process-global/cleanLinuxHopperRuntime.test.ts @@ -77,7 +77,11 @@ it if (alternateMode) { const processEntry = join(procRoot, String(pid)); await mkdir(processEntry); - await symlink(`/proc/${String(pid)}/exe`, join(processEntry, "exe")); + for (const name of ["cmdline", "comm", "exe"]) + await symlink( + `/proc/${String(pid)}/${name}`, + join(processEntry, name), + ); } const { stdout: coordinates } = await execFileAsync("ps", [ "-p", diff --git a/tests/process-global/filesystemSnapshotFallbackCancellation.test.ts b/tests/process-global/filesystemSnapshotFallbackCancellation.test.ts new file mode 100644 index 000000000..cc984dceb --- /dev/null +++ b/tests/process-global/filesystemSnapshotFallbackCancellation.test.ts @@ -0,0 +1,118 @@ +import { createHash } from "node:crypto"; +import { lstat, writeFile } from "node:fs/promises"; +import type { FileHandle } from "node:fs/promises"; +import { join } from "node:path"; + +import { afterEach, expect, it, vi } from "vitest"; + +const observation = vi.hoisted(() => ({ + path: "", + namedReads: 0, + cancelAt: 0, + controller: undefined as AbortController | undefined, + reason: undefined as Error | undefined, + handles: [] as FileHandle[], +})); + +// Force the capability fallback while keeping actual filesystem IO. These +// cases establish cancellation semantics, not native Windows compatibility. +vi.mock("node:fs", async (importOriginal) => { + const original = await importOriginal(); + return { + ...original, + constants: { ...original.constants, O_NOFOLLOW: 0 }, + }; +}); + +vi.mock("node:fs/promises", async (importOriginal) => { + const original = await importOriginal(); + return { + ...original, + open: async (...args: Parameters) => { + const handle = await original.open(...args); + if (args[0] === observation.path) observation.handles.push(handle); + return handle; + }, + lstat: async (...args: Parameters) => { + const stats = await original.lstat(...args); + if ( + args[0] === observation.path && + ++observation.namedReads === observation.cancelAt + ) + observation.controller?.abort(observation.reason); + return stats; + }, + }; +}); + +import { parseProcessScenario } from "../../src/domain/process/processScenario.js"; +import { + hashFile, + snapshotRoots, +} from "../../src/process/capture/FilesystemSnapshot.js"; +import { createTestTempDirectory } from "../fixtures/temporaryDirectory.js"; + +afterEach(() => { + observation.path = ""; + observation.namedReads = 0; + observation.cancelAt = 0; + observation.controller = undefined; + observation.reason = undefined; + observation.handles.length = 0; +}); + +it.each(["hash", "snapshot"] as const)( + "rejects cancellation during the final %s fallback inspection and closes the descriptor", + async (operation) => { + const directory = await createTestTempDirectory("rea-fs-fallback-cancel-"); + const path = join(directory, "observed-file"); + await writeFile(path, "captured bytes"); + const expected = await lstat(path); + const controller = new AbortController(); + const reason = Object.assign( + new Error("cancelled during final named-file inspection"), + { code: "EACCES" }, + ); + observation.path = path; + observation.controller = controller; + observation.reason = reason; + // A root file snapshot first inspects presence and entry metadata, then + // reaches the same two named-file inspections as direct hashing. + observation.cancelAt = operation === "hash" ? 2 : 4; + const result = + operation === "hash" + ? hashFile(path, expected, 1_000, controller.signal) + : snapshotRoots( + parseProcessScenario({ + executable: process.execPath, + working_directory: directory, + filesystem_observation_paths: [path], + }), + controller.signal, + ); + await expect(result).rejects.toBe(reason); + expect(controller.signal.aborted).toBe(true); + expect(observation.handles).toHaveLength(1); + await expect(observation.handles[0]?.stat()).rejects.toMatchObject({ + code: "EBADF", + }); + }, +); + +it("keeps an unchanged fallback digest and closes its descriptor", async () => { + const directory = await createTestTempDirectory("rea-fs-fallback-digest-"); + const path = join(directory, "observed-file"); + const bytes = Buffer.from("captured bytes"); + await writeFile(path, bytes); + const expected = await lstat(path); + observation.path = path; + await expect(hashFile(path, expected, 1_000)).resolves.toEqual({ + state: "hashed", + sha256: createHash("sha256").update(bytes).digest("hex"), + }); + expect(observation.namedReads).toBe(2); + expect(observation.handles).toHaveLength(1); + await expect(observation.handles[0]?.stat()).rejects.toMatchObject({ + code: "EBADF", + }); +}); diff --git a/tests/process-global/hopperLeaseDescriptorPermissions.test.ts b/tests/process-global/hopperLeaseDescriptorPermissions.test.ts new file mode 100644 index 000000000..174097abe --- /dev/null +++ b/tests/process-global/hopperLeaseDescriptorPermissions.test.ts @@ -0,0 +1,79 @@ +import { chmod, lstat, mkdir, mkdtemp, rm } from "node:fs/promises"; +import type { FileHandle } from "node:fs/promises"; +import { join } from "node:path"; + +import { afterEach, expect, it, vi } from "vitest"; + +const replacement = vi.hoisted(() => ({ + selected: "", + retained: "", + handle: undefined as FileHandle | undefined, +})); + +vi.mock("node:fs/promises", async (importOriginal) => { + const original = await importOriginal(); + return { + ...original, + open: async (...args: Parameters) => { + const handle = await original.open(...args); + if (args[0] === replacement.selected) { + replacement.handle = handle; + try { + await original.rename(replacement.selected, replacement.retained); + await original.mkdir(replacement.selected); + await original.chmod(replacement.selected, 0o755); + } catch (cause) { + await handle.close(); + throw cause; + } + } + return handle; + }, + }; +}); + +import { acquireLinuxHopperApplicationLease } from "../../src/hopper/HopperTargetLease.js"; + +afterEach(() => { + replacement.selected = ""; + replacement.retained = ""; + replacement.handle = undefined; +}); + +it.skipIf(process.platform === "win32")( + "secures only the admitted directory descriptor when its path is replaced", + async () => { + // macOS limits Unix-domain socket paths to 104 bytes. Keep the selected + // directory path short enough for the production lease socket beneath it. + const root = await mkdtemp("/tmp/r-"); + const selected = join(root, "selected"); + const retained = join(root, "retained"); + await mkdir(selected); + await chmod(selected, 0o755); + const before = await lstat(selected); + replacement.selected = selected; + replacement.retained = retained; + const acquired = await acquireLinuxHopperApplicationLease({ + directory: selected, + runId: "descriptor-permissions", + }); + if (!acquired.acquired) { + await rm(root, { recursive: true, force: true }); + throw new Error("Expected the fixture lease"); + } + try { + expect((await lstat(retained)).ino).toBe(before.ino); + expect((await lstat(retained)).mode & 0o777).toBe(0o700); + expect((await lstat(selected)).ino).not.toBe(before.ino); + expect((await lstat(selected)).mode & 0o777).toBe(0o755); + await expect(replacement.handle?.stat()).rejects.toMatchObject({ + code: "EBADF", + }); + } finally { + await acquired.lease.release(); + await rm(root, { recursive: true, force: true }); + } + // This asserts descriptor permission ownership only. Later socket + // operations still resolve the path, as the production comment states. + }, +);