diff --git a/src/application/Utf8JsonInput.ts b/src/application/Utf8JsonInput.ts index 7b0e5c0ef7..3844fd6454 100644 --- a/src/application/Utf8JsonInput.ts +++ b/src/application/Utf8JsonInput.ts @@ -81,6 +81,7 @@ export const parseUtf8Json = ( return { ok: false, error: "JSON input is not valid UTF-8", + userMessage: "JSON input is not valid UTF-8", cause, }; } diff --git a/src/cli/electronCommands.ts b/src/cli/electronCommands.ts index 4dd452e852..804d2da92e 100644 --- a/src/cli/electronCommands.ts +++ b/src/cli/electronCommands.ts @@ -121,6 +121,7 @@ const registerElectronActiveCommand = ( const input = await parseCliJsonInput( args.inputJson, "capture_electron_scenario", + signal, ); if (!input.ok) return input.error; const parsed = electronActiveObservationInputSchema.safeParse( diff --git a/src/cli/wasmCommands.ts b/src/cli/wasmCommands.ts index 862677ffa3..9748f03e63 100644 --- a/src/cli/wasmCommands.ts +++ b/src/cli/wasmCommands.ts @@ -38,6 +38,7 @@ export const registerWasmCommands = ( const input = await parseCliJsonInput( args.input, "compare_wasm_interfaces", + signal, ); if (!input.ok) return input.error; const result = await compareWasmInterfaces(input.value, { signal }); diff --git a/src/cliBrowserScenarioCommands.ts b/src/cliBrowserScenarioCommands.ts index 59aede588a..20e8923203 100644 --- a/src/cliBrowserScenarioCommands.ts +++ b/src/cliBrowserScenarioCommands.ts @@ -41,7 +41,11 @@ export const registerBrowserScenarioCommands = ( run: ({ args }) => withCommandCancellation((signal) => logCliCommand(logger, CLI_COMMANDS.captureBrowserScenario, async () => { - const input = await parseCliJsonInput(args.inputJson, OPERATION); + const input = await parseCliJsonInput( + args.inputJson, + OPERATION, + signal, + ); if (!input.ok) return input.error; const scenario = browserScenarioSchema.safeParse( resolveCliJsonPaths(input.value, [["browser", "executable_path"]]), diff --git a/src/cliJsonInput.ts b/src/cliJsonInput.ts index c7c0a36d72..d3df8608bb 100644 --- a/src/cliJsonInput.ts +++ b/src/cliJsonInput.ts @@ -1,3 +1,7 @@ +import { addAbortListener } from "node:events"; +import type { Readable } from "node:stream"; +import { getHeapStatistics } from "node:v8"; + import { absoluteCallerPath } from "./filesystem/callerPath.js"; import { sealOwnedJsonInput } from "./application/OwnedJsonInput.js"; @@ -5,23 +9,34 @@ import { JsonInputCancellationFailure, JsonInputPhysicalReadFailure, readJsonInputFile, + nativeJsonParseRequiredBytes, } from "./application/JsonInputFile.js"; import { NonRegularFileReadError } from "./filesystem/RegularFile.js"; import { AnalysisAccessDeniedError, AnalysisCancelledError, AnalysisInputError, + AnalysisResourceConstraintError, } from "./domain/analysisErrorCore.js"; import { projectAnalysisError } from "./domain/analysisErrorProjection.js"; +import { err, ok, type Result } from "./domain/result.js"; import type { JsonValue } from "./domain/jsonValue.js"; import { safeParseJson } from "./domain/safeJson.js"; +import { + JSON_INPUT_RESOURCE_REMEDIATION, + parseUtf8Json, +} from "./application/Utf8JsonInput.js"; -/** Parse inline JSON or one local JSON file, honoring optional caller cancellation. */ +/** Parse inline JSON, standard input ('-'), or one local JSON file, honoring optional caller cancellation. */ export const parseCliJsonInput = async ( value: string, operation: string, signal?: AbortSignal, - options?: { readonly snapshot?: boolean }, + options?: { + readonly snapshot?: boolean; + readonly stdin?: Readable; + readonly maxStdinBytes?: number; + }, ): Promise< | { readonly ok: true; readonly value: unknown } | { readonly ok: false; readonly error: JsonValue } @@ -53,13 +68,16 @@ export const parseCliJsonInput = async ( }, }; } - const parsed = await readJsonInputFile( - value, - operation, - signal, - undefined, - options?.snapshot ?? false, - ); + const parsed = + value === "-" + ? await readCliJsonStdin(operation, signal, options) + : await readJsonInputFile( + value, + operation, + signal, + undefined, + options?.snapshot ?? false, + ); if (parsed.ok && signal?.aborted) return cancelled(signal.reason); return parsed.ok ? { ok: true, value: parsed.value } @@ -106,6 +124,126 @@ export const parseCliJsonInput = async ( } }; +const DEFAULT_MAX_STDIN_BYTES = 8 * 1024 * 1024; + +const readCliJsonStdin = async ( + operation: string, + signal?: AbortSignal, + options?: { + readonly snapshot?: boolean; + readonly stdin?: Readable; + readonly maxStdinBytes?: number; + }, +): Promise< + Result +> => { + const stream = options?.stdin ?? process.stdin; + const maxBytes = options?.maxStdinBytes ?? DEFAULT_MAX_STDIN_BYTES; + const chunks: Buffer[] = []; + let totalBytes = 0; + const requireHeadroom = (): void => { + const heap = getHeapStatistics(); + const requiredBytes = nativeJsonParseRequiredBytes( + totalBytes, + options?.snapshot, + ); + if (requiredBytes > heap.total_available_size) + throw new AnalysisResourceConstraintError( + operation, + "memory", + "Insufficient heap headroom to read and parse the complete JSON stdin document", + { + input_path: "-", + input_bytes: totalBytes, + required_heap_bytes: requiredBytes, + available_heap_bytes: heap.total_available_size, + heap_size_limit_bytes: heap.heap_size_limit, + }, + { remediationAction: JSON_INPUT_RESOURCE_REMEDIATION }, + ); + }; + // Destroying the reader wakes an idle next(); polling inside the loop cannot. + const abort = + signal === undefined + ? undefined + : addAbortListener(signal, () => { + stream.destroy(new JsonInputCancellationFailure(signal.reason)); + }); + try { + try { + for await (const chunk of stream) { + signal?.throwIfAborted(); + totalBytes += + typeof chunk === "string" + ? Buffer.byteLength(chunk) + : chunk.byteLength; + if (totalBytes > maxBytes) + return err( + new AnalysisResourceConstraintError( + operation, + "memory", + "JSON input exceeds the maximum allowed stdin size", + { + input_path: "-", + input_bytes: totalBytes, + max_stdin_bytes: maxBytes, + }, + { remediationAction: JSON_INPUT_RESOURCE_REMEDIATION }, + ), + ); + requireHeadroom(); + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + } catch (cause: unknown) { + if ( + cause instanceof JsonInputCancellationFailure || + (signal?.aborted && isJsonInputCancellation(cause, signal)) + ) + throw cause; + if (cause instanceof AnalysisResourceConstraintError) return err(cause); + throw new JsonInputPhysicalReadFailure(cause); + } finally { + abort?.[Symbol.dispose](); + } + signal?.throwIfAborted(); + // Recheck live headroom before concatenation, UTF-8 decoding and JSON.parse. + requireHeadroom(); + const parsed = parseUtf8Json( + Buffer.concat(chunks, totalBytes), + operation, + "-", + ); + if (!parsed.ok) + return err( + new AnalysisInputError( + operation, + { + cause: parsed.cause, + ...(parsed.userMessage === undefined + ? {} + : { userMessage: parsed.userMessage }), + }, + [ + { + path: [], + reason: "invalid_format", + expected: "JSON", + ...(parsed.userMessage === undefined + ? {} + : { message: parsed.userMessage }), + }, + ], + ), + ); + return options?.snapshot + ? await sealOwnedJsonInput(parsed.value, operation, signal) + : ok(parsed.value); + } catch (cause: unknown) { + if (cause instanceof AnalysisResourceConstraintError) return err(cause); + throw cause; + } +}; + const isJsonInputCancellation = ( cause: unknown, signal: AbortSignal, diff --git a/tests/boundary/cli/cliJsonStdinCli.test.ts b/tests/boundary/cli/cliJsonStdinCli.test.ts new file mode 100644 index 0000000000..1b4150d1a8 --- /dev/null +++ b/tests/boundary/cli/cliJsonStdinCli.test.ts @@ -0,0 +1,168 @@ +import { resolve } from "node:path"; +import { pathToFileURL } from "node:url"; + +import { expect } from "vitest"; + +import { nativeJsonParseRequiredBytes } from "../../../src/application/JsonInputFile.js"; +import { JSON_BYTE_ORDER_MARK_MESSAGE } from "../../../src/application/Utf8JsonInput.js"; +import { cliTest } from "../../support/cli/cliFixture.js"; +import { waitForExit } from "../../support/process/processFixture.js"; + +const probe = pathToFileURL( + resolve("tests/fixtures/cliJsonStdinProbe.mjs"), +).href; + +cliTest.for([ + [Buffer.from("\uFEFF{}"), JSON_BYTE_ORDER_MARK_MESSAGE], + [Buffer.from([0xf0, 0x28, 0x8c, 0xbc]), "JSON input is not valid UTF-8"], + [Buffer.from('{"secret-value":'), undefined], +] as const)( + "projects piped JSON failures through the public CLI (%j)", + async ([input, message], { cli }) => { + const result = await cli.run({ + arguments: ["compare-wasm-interfaces", "-", "--json"], + input, + }); + expect(result.exitCode, result.stderr).toBe(1); + expect(result.json).toMatchObject({ + code: "invalid_request", + input_path: "-", + input_reason: "invalid-json", + ...(message === undefined ? {} : { details: { issues: [{ message }] } }), + }); + expect(result.stdout).not.toContain("secret-value"); + }, +); + +cliTest.for([false, true])( + "admits piped JSON against real heap headroom (snapshot: %s)", + async (snapshot, { processes }) => { + const child = processes.spawn( + process.execPath, + [ + resolve("scripts/rea.mjs"), + ...(snapshot + ? ["environment-serve", "-", "--evidence", "{}"] + : ["compare-wasm-interfaces", "-"]), + "--json", + ], + { + env: { + ...process.env, + NODE_OPTIONS: "--max-old-space-size=256", + REA_LOG_LEVEL: "silent", + }, + }, + ); + let stdout = "", + stderr = ""; + let writeFailure: unknown; + child.stdout.on("data", (chunk) => { + stdout += String(chunk); + }); + child.stderr.on("data", (chunk) => { + stderr += String(chunk); + }); + // Admission can close the reader before the producer finishes its write. + child.stdin.on("error", (cause) => { + writeFailure = cause; + }); + const closed = new Promise((resolve, reject) => { + child.once("error", reject); + child.once("close", () => resolve()); + }); + child.stdin.end(`[${"[],".repeat(1_400_000)}[]]`); + expect(await waitForExit(child, 10_000), stderr).toBe(true); + await closed; + if (writeFailure !== undefined) + expect(writeFailure).toMatchObject({ code: "EPIPE" }); + const result = { + exitCode: child.exitCode, + stderr, + json: JSON.parse(stdout) as unknown, + }; + expect(result.exitCode, result.stderr).toBe(1); + expect(result.json).toMatchObject({ + code: "resource_constraint", + input_path: "-", + input_reason: "too-large", + details: { + resource: "memory", + reported_limits: { + input_path: "-", + required_heap_bytes: expect.any(Number), + available_heap_bytes: expect.any(Number), + }, + }, + }); + const limits = ( + result.json as { + details: { + reported_limits: { + input_bytes: number; + required_heap_bytes: number; + available_heap_bytes: number; + }; + }; + } + ).details.reported_limits; + expect(limits.input_bytes).toBeLessThanOrEqual(8 * 1024 * 1024); + expect(limits.required_heap_bytes).toBe( + nativeJsonParseRequiredBytes(limits.input_bytes, snapshot), + ); + expect(limits.required_heap_bytes).toBeGreaterThan( + limits.available_heap_bytes, + ); + }, +); + +cliTest.skipIf(process.platform === "win32").for( + [ + "capture-browser-scenario", + "capture-electron-scenario", + "compare-wasm-interfaces", + ].flatMap((command) => + ( + [ + ["SIGINT", 130], + ["SIGTERM", 143], + ] as const + ).map(([signal, exitCode]) => ({ command, signal, exitCode })), + ), +)( + "$command cancels an idle stdin reader on $signal", + async ({ command, signal, exitCode }, { processes }) => { + const child = processes.spawn( + process.execPath, + [resolve("scripts/rea.mjs"), command, "-", "--json"], + { + env: { + ...process.env, + NODE_OPTIONS: `--import=${probe}`, + REA_LOG_LEVEL: "silent", + }, + }, + ); + let stdout = "", + stderr = ""; + child.stdout.on("data", (chunk) => { + stdout += String(chunk); + }); + child.stderr.on("data", (chunk) => { + stderr += String(chunk); + }); + const closed = new Promise((resolve, reject) => { + child.once("error", reject); + child.once("close", () => resolve()); + }); + await expect + .poll(() => stderr.includes("REA_TEST_STDIN_READ"), { timeout: 10_000 }) + .toBe(true); + // Keep the pipe open: EOF must not be what releases the pending read. + child.kill(signal); + expect(await waitForExit(child, 5_000), stderr).toBe(true); + await closed; + expect(child.exitCode, stderr).toBe(exitCode); + expect(JSON.parse(stdout)).toMatchObject({ code: "cancelled" }); + }, +); diff --git a/tests/boundary/filesystem/cliJsonInput.test.ts b/tests/boundary/filesystem/cliJsonInput.test.ts index 097811fcd5..443bcca113 100644 --- a/tests/boundary/filesystem/cliJsonInput.test.ts +++ b/tests/boundary/filesystem/cliJsonInput.test.ts @@ -1,6 +1,8 @@ +import { getEventListeners } from "node:events"; import { execFile } from "node:child_process"; import { chmod, symlink, writeFile } from "node:fs/promises"; import { join } from "node:path"; +import { Readable } from "node:stream"; import { promisify } from "node:util"; import { describe, expect, it, onTestFinished } from "vitest"; @@ -346,3 +348,211 @@ describe("CLI JSON file cancellation", () => { ).toEqual({ ok: true, value: ["index.js", "main.js", "package.json"] }); }); }); + +describe("CLI JSON standard input", () => { + it("reads valid JSON from stdin when input is '-'", async () => { + const stream = Readable.from(['{"seed":123,"label":"piped"}']); + const result = await parseCliJsonInput("-", "test-input", undefined, { + stdin: stream, + }); + expect(result).toEqual({ + ok: true, + value: { seed: 123, label: "piped" }, + }); + }); + + it("rejects empty stdin with invalid-json", async () => { + const stream = Readable.from([]); + const result = await parseCliJsonInput("-", "test-input", undefined, { + stdin: stream, + }); + expect(result).toMatchObject({ + ok: false, + error: { + code: "invalid_request", + input_path: "-", + input_reason: "invalid-json", + details: { + issues: [{ path: [], reason: "invalid_format" }], + }, + }, + }); + }); + + it("rejects malformed JSON from stdin without printing full payload", async () => { + const stream = Readable.from(["{ this is not valid json"]); + const result = await parseCliJsonInput("-", "test-input", undefined, { + stdin: stream, + }); + expect(result).toMatchObject({ + ok: false, + error: { + code: "invalid_request", + input_path: "-", + input_reason: "invalid-json", + details: { + issues: [{ path: [], reason: "invalid_format" }], + }, + }, + }); + expect(JSON.stringify(result)).not.toContain("this is not valid json"); + }); + + it("rejects oversized stdin with too-large", async () => { + const stream = Readable.from(['{"a":"' + "x".repeat(100) + '"}']); + const result = await parseCliJsonInput("-", "test-input", undefined, { + stdin: stream, + maxStdinBytes: 50, + }); + expect(result).toMatchObject({ + ok: false, + error: { + code: "resource_constraint", + input_path: "-", + input_reason: "too-large", + details: { + resource: "memory", + }, + }, + }); + }); + + it("preserves caller cancellation during stdin read", async () => { + const controller = new AbortController(); + controller.abort("aborted"); + const stream = Readable.from(['{"seed":1}']); + const result = await parseCliJsonInput( + "-", + "test-input", + controller.signal, + { + stdin: stream, + }, + ); + expect(result).toMatchObject({ + ok: false, + error: { + code: "cancelled", + }, + }); + }); +}); + +describe("CLI JSON stdin cancellation and encoding", () => { + it.each([false, true])( + "interrupts an idle stdin read (partial input: %s)", + async (partial) => { + const controller = new AbortController(); + let started!: () => void; + const reading = new Promise((resolve) => { + started = resolve; + }); + const stream = new Readable({ + read() { + started(); + }, + }); + onTestFinished(() => { + stream.destroy(); + }); + if (partial) stream.push('{"partial":'); + const pending = parseCliJsonInput("-", "test-input", controller.signal, { + stdin: stream, + }); + await reading; + // Let any initial chunk be consumed before aborting the next idle read. + await new Promise((resolve) => setImmediate(resolve)); + controller.abort("stop idle stdin"); + await expect(pending).resolves.toMatchObject({ + ok: false, + error: { code: "cancelled" }, + }); + expect(stream.destroyed).toBe(true); + expect(getEventListeners(controller.signal, "abort")).toHaveLength(0); + }, + 1_000, + ); + + it.each([ + [Buffer.from("\uFEFF{}"), JSON_BYTE_ORDER_MARK_MESSAGE], + [Buffer.from([0xf0, 0x28, 0x8c, 0xbc]), "JSON input is not valid UTF-8"], + [Buffer.from([0xe2, 0x82]), "JSON input is not valid UTF-8"], + ])( + "preserves safe stdin encoding diagnostics (%j)", + async (bytes, message) => { + expect( + await parseCliJsonInput("-", "test-input", undefined, { + stdin: Readable.from([bytes]), + }), + ).toMatchObject({ + ok: false, + error: { + input_path: "-", + input_reason: "invalid-json", + details: { issues: [{ message }] }, + }, + }); + }, + ); + + it("reads UTF-8 split between chunks and accepts the exact byte limit", async () => { + const bytes = Buffer.from('{"label":"中文🦀"}'); + expect( + await parseCliJsonInput("-", "test-input", undefined, { + stdin: Readable.from([...bytes].map((byte) => Buffer.from([byte]))), + maxStdinBytes: bytes.length, + }), + ).toEqual({ ok: true, value: { label: "中文🦀" } }); + }); + + it("projects cancellation during stdin snapshot sealing", async () => { + const controller = new AbortController(); + const stream = Readable.from([ + JSON.stringify(Array.from({ length: 20_000 }, (_, index) => ({ index }))), + ]); + stream.once("end", () => + setImmediate(() => controller.abort("stop sealing")), + ); + await expect( + parseCliJsonInput("-", "test-input", controller.signal, { + stdin: stream, + snapshot: true, + }), + ).resolves.toMatchObject({ ok: false, error: { code: "cancelled" } }); + expect(getEventListeners(controller.signal, "abort")).toHaveLength(0); + }); + + it("seals stdin snapshots and releases the abort listener on success", async () => { + const controller = new AbortController(); + const result = await parseCliJsonInput( + "-", + "test-input", + controller.signal, + { + stdin: Readable.from(['{"nested":{"value":1}}']), + snapshot: true, + }, + ); + expect(result).toEqual({ ok: true, value: { nested: { value: 1 } } }); + if (result.ok) expect(Object.isFrozen(result.value)).toBe(true); + expect(getEventListeners(controller.signal, "abort")).toHaveLength(0); + }); + + it("preserves stdin physical failure and releases the abort listener", async () => { + const controller = new AbortController(); + const stream = new Readable({ + read() { + this.destroy(Object.assign(new Error("read failed"), { code: "EIO" })); + }, + }); + expect( + await parseCliJsonInput("-", "test-input", controller.signal, { + stdin: stream, + }), + ).toMatchObject({ + ok: false, + error: { input_path: "-", input_reason: "read-failed" }, + }); + expect(getEventListeners(controller.signal, "abort")).toHaveLength(0); + }); +}); diff --git a/tests/fixtures/cliJsonStdinProbe.mjs b/tests/fixtures/cliJsonStdinProbe.mjs new file mode 100644 index 0000000000..cb9a56af56 --- /dev/null +++ b/tests/fixtures/cliJsonStdinProbe.mjs @@ -0,0 +1,6 @@ +// Observe the production stdin reader without supplying data or closing its pipe. +const iterate = process.stdin[Symbol.asyncIterator]; +process.stdin[Symbol.asyncIterator] = function (...args) { + process.stderr.write("REA_TEST_STDIN_READ\n"); + return iterate.apply(this, args); +};