diff --git a/src/application/JsonFiles.ts b/src/application/JsonFiles.ts index 916d4fd61b..1d3875570c 100644 --- a/src/application/JsonFiles.ts +++ b/src/application/JsonFiles.ts @@ -16,8 +16,7 @@ import { AnalysisResourceConstraintError, } from "../domain/analysisErrorCore.js"; import { err, ok, type Result } from "../domain/result.js"; -import { parseUtf8Json } from "./Utf8JsonInput.js"; -import { readRegularFile } from "./RegularFileRead.js"; +import { readJsonInputFile } from "./JsonInputFile.js"; import { NonRegularFileReadError } from "../filesystem/RegularFile.js"; /** Request control and its owning operation for an interruptible atomic write. */ @@ -37,12 +36,20 @@ export const readJsonFile = async ( > => { const requestedPath = resolve(path); try { - const encoded = await readRegularFile(requestedPath); - const decoded = parseUtf8Json(encoded, "read_evidence_file", requestedPath); + const decoded = await readJsonInputFile( + requestedPath, + "read_evidence_file", + ); if (!decoded.ok) { + if (decoded.error instanceof AnalysisResourceConstraintError) + return err(decoded.error); return err( new EvidenceFileError("read", "invalid-json", { - cause: decoded.cause ?? new SyntaxError(decoded.error), + cause: + decoded.error.cause ?? + new SyntaxError( + decoded.error.issues[0]?.message ?? decoded.error.message, + ), path: requestedPath, }), ); diff --git a/src/cliJsonFile.ts b/src/application/JsonInputFile.ts similarity index 58% rename from src/cliJsonFile.ts rename to src/application/JsonInputFile.ts index c0d474f698..96d26e61e0 100644 --- a/src/cliJsonFile.ts +++ b/src/application/JsonInputFile.ts @@ -6,61 +6,76 @@ import { getHeapStatistics } from "node:v8"; import parser from "stream-json/parser.js"; import streamValues from "stream-json/streamers/stream-values.js"; -import { withRegularFile } from "./application/RegularFileRead.js"; +import { RegularFileChangedError } from "../filesystem/RegularFile.js"; +import { withRegularFile } from "./RegularFileRead.js"; import { JSON_BYTE_ORDER_MARK_MESSAGE, JSON_INPUT_RESOURCE_REMEDIATION, parseUtf8Json, -} from "./application/Utf8JsonInput.js"; +} from "./Utf8JsonInput.js"; import { AnalysisInputError, AnalysisResourceConstraintError, -} from "./domain/analysisErrorCore.js"; -import { err, ok, type Result } from "./domain/result.js"; +} from "../domain/analysisErrorCore.js"; +import { err, ok, type Result } from "../domain/result.js"; -type JsonFileFailure = AnalysisInputError | AnalysisResourceConstraintError; +type JsonInputFileFailure = + | AnalysisInputError + | AnalysisResourceConstraintError; const READ_CHUNK_BYTES = 64 * 1024; -const PACKED_STRING_PART_CODE_UNITS = 64 * 1024; +const PACKED_TOKEN_PART_CODE_UNITS = 64 * 1024; // Retain native-parser speed for modest inputs while bounding its extra copies. // This selects an implementation; larger files remain accepted through streaming. const NATIVE_PARSE_READ_BUDGET = 8 * 1024 * 1024; +/** Keep native parsing only when input copies and value expansion fit live headroom. */ +const nativeParseFits = (inputBytes: number): boolean => + inputBytes <= NATIVE_PARSE_READ_BUDGET && + // The assembler estimate reserves up to 128 bytes per two-byte container. + // Allow that 64x expansion plus byte/text copies before native JSON.parse, + // whose individual container allocations cannot be intercepted. + inputBytes * 72 + READ_CHUNK_BYTES <= + getHeapStatistics().total_available_size; + /** Parse strict UTF-8 JSON without requiring a whole-document string. */ -export const readCliJsonFile = ( +export const readJsonInputFile = ( path: string, operation: string, signal?: AbortSignal, -): Promise> => +): Promise> => withRegularFile( path, async (handle, stats) => { let found = false; let value: unknown; try { - const prefix = - stats.size <= NATIVE_PARSE_READ_BUDGET - ? await readPrefix(handle, stats.size + 1, signal) - : undefined; - if (prefix?.complete === true) { + const prefix = nativeParseFits(stats.size) + ? await readPrefix(handle, stats.size + 1, signal) + : undefined; + if (prefix !== undefined) { + if (!prefix.complete || prefix.bytes.length !== stats.size) + throw new RegularFileChangedError(path); const parsed = parseUtf8Json( prefix.bytes, operation, path, - "cli-json-input", + "json-file-input", ); return parsed.ok ? ok(parsed.value) : err(invalidJson(operation, parsed.error, parsed.cause)); } await pipeline( - decodedChunks(handle, signal, prefix?.bytes), + decodedChunks(handle, stats.size, path, signal), parser.asStream({ jsonStreaming: false, streamValues: false, packKeys: false, packStrings: false, + packNumbers: false, }), - packJsonStrings(operation, stats.size), + packJsonScalars(operation, stats.size, path), + admitJsonAssembly(operation, stats.size, path), streamValues.asStream(), async (rows: AsyncIterable) => { for await (const row of rows) { @@ -107,8 +122,9 @@ export const readCliJsonFile = ( new AnalysisResourceConstraintError( operation, "memory", - "An individual JSON string exceeds the runtime string limit", + "An individual JSON string or number token exceeds the runtime string limit", { + input_path: path, input_file_bytes: stats.size, max_string_code_units: constants.MAX_STRING_LENGTH, }, @@ -121,12 +137,16 @@ export const readCliJsonFile = ( throw cause; } }, - signal, + { signal }, ); -const packJsonStrings = (operation: string, inputFileBytes: number) => +const packJsonScalars = ( + operation: string, + inputFileBytes: number, + path: string, +) => async function* (tokens: AsyncIterable): AsyncGenerator { - let kind: "keyValue" | "stringValue" | undefined; + let kind: "keyValue" | "stringValue" | "numberValue" | undefined; let length = 0; let value = ""; let parts: string[] = []; @@ -139,23 +159,32 @@ const packJsonStrings = (operation: string, inputFileBytes: number) => case "startString": kind = token.name === "startKey" ? "keyValue" : "stringValue"; break; + case "startNumber": + kind = "numberValue"; + break; + case "numberChunk": case "stringChunk": if ( kind === undefined || !("value" in token) || typeof token.value !== "string" ) - throw new Error("Unexpected JSON string fragment"); + throw new Error("Unexpected JSON scalar fragment"); length += token.value.length; if (length > constants.MAX_STRING_LENGTH) throw new RangeError("Invalid string length"); parts.push(token.value); partLength += token.value.length; - if (partLength >= PACKED_STRING_PART_CODE_UNITS) { + if (partLength >= PACKED_TOKEN_PART_CODE_UNITS) { // The tokenizer emits short fragments. Coalesce them before retaining - // a string so millions of substring/rope nodes cannot exhaust the heap + // a scalar so millions of substring/rope nodes cannot exhaust the heap // before the native string-length constraint can be reported. - requireStringAssemblyHeadroom(operation, inputFileBytes, length); + requireStringAssemblyHeadroom( + operation, + inputFileBytes, + path, + length, + ); value += parts.join(""); parts = []; partLength = 0; @@ -163,8 +192,14 @@ const packJsonStrings = (operation: string, inputFileBytes: number) => break; case "endKey": case "endString": - if (kind === undefined) throw new Error("Unexpected JSON string end"); - requireStringAssemblyHeadroom(operation, inputFileBytes, length); + case "endNumber": + if (kind === undefined) throw new Error("Unexpected JSON scalar end"); + requireStringAssemblyHeadroom( + operation, + inputFileBytes, + path, + length, + ); yield { name: kind, value: value + parts.join("") }; kind = undefined; length = 0; @@ -178,22 +213,78 @@ const packJsonStrings = (operation: string, inputFileBytes: number) => } }; +/** Admit the retained value graph before the upstream assembler grows it. */ +const admitJsonAssembly = ( + operation: string, + inputFileBytes: number, + path: string, +) => + async function* (tokens: AsyncIterable): AsyncGenerator { + let projectedBytes = 0; + let nextCheck = 0; + for await (const token of tokens) { + if (typeof token !== "object" || token === null || !("name" in token)) + throw new Error("Unexpected JSON assembly token"); + if (token.name === "startObject" || token.name === "startArray") { + projectedBytes += 128; + } else if (token.name === "keyValue" || token.name === "stringValue") { + if (!("value" in token) || typeof token.value !== "string") + throw new Error("Unexpected packed JSON string"); + projectedBytes += 64 + token.value.length * 2; + } else if ( + token.name === "numberValue" || + token.name === "nullValue" || + token.name === "trueValue" || + token.name === "falseValue" + ) { + projectedBytes += 64; + } + if (projectedBytes >= nextCheck) { + const heap = getHeapStatistics(); + // Conservative capacity for containers, property/array slots and UTF-16 + // values. Keep another whole projection available for container growth, + // plus the next chunk of estimates between samples. This is admission + // against live runtime headroom, not a fixed document-size policy. + const requiredBytes = projectedBytes + READ_CHUNK_BYTES; + if (requiredBytes > heap.total_available_size) + throw new AnalysisResourceConstraintError( + operation, + "memory", + "Insufficient heap headroom to materialize the complete JSON value", + { + input_path: path, + input_file_bytes: inputFileBytes, + projected_value_bytes: projectedBytes, + assembly_headroom_bytes: requiredBytes, + available_heap_bytes: heap.total_available_size, + heap_size_limit_bytes: heap.heap_size_limit, + }, + { remediationAction: JSON_INPUT_RESOURCE_REMEDIATION }, + ); + nextCheck = projectedBytes + READ_CHUNK_BYTES; + } + yield token; + } + }; + const requireStringAssemblyHeadroom = ( operation: string, inputFileBytes: number, + path: string, stringCodeUnits: number, ): void => { - if (stringCodeUnits < PACKED_STRING_PART_CODE_UNITS) return; + if (stringCodeUnits < PACKED_TOKEN_PART_CODE_UNITS) return; const heap = getHeapStatistics(); // Concatenation or key interning may flatten the retained rope. Reserve its // UTF-16 representation and the next coalesced part before that allocation. - const requiredBytes = 2 * (stringCodeUnits + PACKED_STRING_PART_CODE_UNITS); + const requiredBytes = 2 * (stringCodeUnits + PACKED_TOKEN_PART_CODE_UNITS); if (requiredBytes > heap.total_available_size) throw new AnalysisResourceConstraintError( operation, "memory", "Insufficient heap headroom to assemble a complete JSON string", { + input_path: path, input_file_bytes: inputFileBytes, string_code_units: stringCodeUnits, string_assembly_headroom_bytes: requiredBytes, @@ -208,8 +299,9 @@ const requireStringAssemblyHeadroom = ( async function* decodedChunks( handle: FileHandle, + admittedBytes: number, + path: string, signal: AbortSignal | undefined, - prefix?: Uint8Array, ): AsyncGenerator { const decoder = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }); let leading = true; @@ -220,25 +312,25 @@ async function* decodedChunks( } return text; }; - if (prefix !== undefined) - for (let offset = 0; offset < prefix.length; offset += READ_CHUNK_BYTES) { - signal?.throwIfAborted(); - yield checked( - decoder.decode(prefix.subarray(offset, offset + READ_CHUNK_BYTES), { - stream: true, - }), - ); - } const bytes = Buffer.allocUnsafe(READ_CHUNK_BYTES); + let observedBytes = 0; while (true) { signal?.throwIfAborted(); - const read = await handle.read(bytes, 0, bytes.length, null); + const read = await handle.read( + bytes, + 0, + Math.min(bytes.length, admittedBytes - observedBytes + 1), + null, + ); signal?.throwIfAborted(); if (read.bytesRead === 0) break; + observedBytes += read.bytesRead; + if (observedBytes > admittedBytes) throw new RegularFileChangedError(path); yield checked( decoder.decode(bytes.subarray(0, read.bytesRead), { stream: true }), ); } + if (observedBytes !== admittedBytes) throw new RegularFileChangedError(path); yield checked(decoder.decode()); } @@ -264,8 +356,7 @@ const readPrefix = async ( return { bytes: bytes.subarray(0, offset), complete: true }; offset += read.bytesRead; } - // The file grew after admission. Continue from this same handle, preserving - // the prefix already read rather than reopening or dropping its bytes. + // One byte beyond the admitted extent proves the selected file changed. return { bytes, complete: false }; }; diff --git a/src/application/RegularFileRead.ts b/src/application/RegularFileRead.ts index df87b7c5a2..f584f1f6ea 100644 --- a/src/application/RegularFileRead.ts +++ b/src/application/RegularFileRead.ts @@ -1,20 +1,44 @@ import type { Stats } from "node:fs"; -import type { FileHandle } from "node:fs/promises"; +import { lstat, stat, type FileHandle } from "node:fs/promises"; -import { openRegularFile } from "../filesystem/RegularFile.js"; +import { + openRegularFile, + sameRegularFileState, + RegularFileChangedError, +} from "../filesystem/RegularFile.js"; +import { readBoundedFileBytes } from "../process/BoundedFileBytes.js"; + +/** Filesystem admission policy for one selected regular input. */ +export interface RegularFileReadOptions { + readonly signal?: AbortSignal | undefined; + readonly symlinks?: "follow" | "reject"; +} /** Admit one regular-file handle, retain it through the consumer, and always close it. */ export const withRegularFile = async ( path: string, read: (handle: FileHandle, stats: Stats) => Promise, - signal?: AbortSignal, + options: RegularFileReadOptions = {}, ): Promise => { - const handle = await openRegularFile(path, { symlinks: "follow", signal }); + const { signal } = options; + const handle = await openRegularFile(path, { + symlinks: options.symlinks ?? "follow", + signal, + }); try { const stats = await handle.stat(); signal?.throwIfAborted(); const value = await read(handle, stats); signal?.throwIfAborted(); + const [opened, currentPath] = await Promise.all([ + handle.stat(), + options.symlinks === "reject" ? lstat(path) : stat(path), + ]); + if ( + !sameRegularFileState(stats, opened) || + !sameRegularFileState(stats, currentPath) + ) + throw new RegularFileChangedError(path); return value; } finally { await handle.close(); @@ -24,6 +48,25 @@ export const withRegularFile = async ( /** Admit a regular file without waiting for a pipe, then read its verified handle. */ export const readRegularFile = ( path: string, - signal?: AbortSignal, + options: RegularFileReadOptions = {}, ): Promise => - withRegularFile(path, (handle) => handle.readFile({ signal }), signal); + withRegularFile( + path, + async (handle, stats) => { + const bytes = await readBoundedFileBytes( + handle, + stats.size, + options.signal, + ); + if (bytes === undefined || bytes.length !== stats.size) + throw new RegularFileChangedError(path); + return bytes; + }, + options, + ); + +/** Read one admitted regular file as UTF-8 without waiting on a pipe. */ +export const readRegularFileText = async ( + path: string, + options: RegularFileReadOptions = {}, +): Promise => (await readRegularFile(path, options)).toString("utf8"); diff --git a/src/application/WebScriptExportService.ts b/src/application/WebScriptExportService.ts index 1b3a47b7e7..3117cb69a8 100644 --- a/src/application/WebScriptExportService.ts +++ b/src/application/WebScriptExportService.ts @@ -27,7 +27,10 @@ import { WebScriptExportError } from "../domain/webScriptExportError.js"; import type { ExecutionOptions } from "./AnalysisProvider.js"; import { WEB_SCRIPT_EXPORT_PROVIDER as PROVIDER } from "./InvestigationProviders.js"; import { readRegularFile } from "./RegularFileRead.js"; -import { NonRegularFileReadError } from "../filesystem/RegularFile.js"; +import { + NonRegularFileReadError, + RegularFileChangedError, +} from "../filesystem/RegularFile.js"; const OPERATION = "export_web_scripts"; @@ -153,7 +156,7 @@ const readCapture = async ( signal: AbortSignal | undefined, ): Promise> => { try { - return ok(await readRegularFile(path, signal)); + return ok(await readRegularFile(path, { signal })); } catch (cause: unknown) { const code = cause instanceof Error && "code" in cause ? String(cause.code) : ""; @@ -163,6 +166,7 @@ const readCapture = async ( ); if ( cause instanceof NonRegularFileReadError || + cause instanceof RegularFileChangedError || code === "ENOENT" || code === "ENOTDIR" || code === "EISDIR" || diff --git a/src/cliJsonInput.ts b/src/cliJsonInput.ts index 47b1431e4b..4dd38a90e2 100644 --- a/src/cliJsonInput.ts +++ b/src/cliJsonInput.ts @@ -1,6 +1,6 @@ import { resolve } from "node:path"; -import { readCliJsonFile } from "./cliJsonFile.js"; +import { readJsonInputFile } from "./application/JsonInputFile.js"; import { NonRegularFileReadError } from "./filesystem/RegularFile.js"; import { AnalysisAccessDeniedError, @@ -21,7 +21,7 @@ export const parseCliJsonInput = async ( const inline = safeParseJson(value); if (inline.ok) return { ok: true, value: inline.value }; try { - const parsed = await readCliJsonFile(value, operation); + const parsed = await readJsonInputFile(value, operation); return parsed.ok ? { ok: true, value: parsed.value } : { diff --git a/src/filesystem/RegularFile.ts b/src/filesystem/RegularFile.ts index b9a0914c36..69edcad312 100644 --- a/src/filesystem/RegularFile.ts +++ b/src/filesystem/RegularFile.ts @@ -16,6 +16,15 @@ export class NonRegularFileReadError extends Error { } } +/** A regular input no longer matches the extent or identity admitted for reading. */ +export class RegularFileChangedError extends Error { + readonly code = "ESTALE"; + constructor(readonly path: string) { + super(`Selected input changed while being read: ${path}`); + this.name = "RegularFileChangedError"; + } +} + /** One admitted file descriptor and the metadata that identified its contents. */ export interface StableRegularFileDescriptor { readonly handle: FileHandle; diff --git a/src/filesystem/StableFileHash.ts b/src/filesystem/StableFileHash.ts new file mode 100644 index 0000000000..11c9193613 --- /dev/null +++ b/src/filesystem/StableFileHash.ts @@ -0,0 +1,55 @@ +import { createHash } from "node:crypto"; +import { stat } from "node:fs/promises"; + +import { + openRegularFile, + RegularFileChangedError, + sameRegularFileState, +} from "./RegularFile.js"; + +/** Hash one regular file from its admitted descriptor and verify its path state. */ +export const hashStableFile = async ( + path: string, + signal?: AbortSignal, +): Promise<{ readonly sha256: string; readonly bytes: number }> => { + const handle = await openRegularFile(path, { + symlinks: "follow", + signal, + }); + try { + const initial = await handle.stat(); + if (!Number.isSafeInteger(initial.size) || initial.size < 0) + throw new RangeError( + `Regular file size cannot be represented safely: ${path}`, + ); + const hash = createHash("sha256"); + const chunk = Buffer.allocUnsafe(64 * 1024); + let position = 0; + while (position <= initial.size) { + signal?.throwIfAborted(); + const length = Math.min(chunk.length, initial.size - position + 1); + if (length <= 0) break; + const read = await handle.read(chunk, 0, length, position); + signal?.throwIfAborted(); + if (read.bytesRead === 0) break; + hash.update(chunk.subarray(0, read.bytesRead)); + position += read.bytesRead; + if (position > initial.size) throw new RegularFileChangedError(path); + } + signal?.throwIfAborted(); + const [opened, currentPath] = await Promise.all([ + handle.stat(), + stat(path), + ]); + signal?.throwIfAborted(); + if ( + position !== initial.size || + !sameRegularFileState(initial, opened) || + !sameRegularFileState(initial, currentPath) + ) + throw new RegularFileChangedError(path); + return { sha256: hash.digest("hex"), bytes: position }; + } finally { + await handle.close(); + } +}; diff --git a/tests/boundary/filesystem/cliJsonInput.test.ts b/tests/boundary/filesystem/cliJsonInput.test.ts index dd4809ea81..d2bdcd5403 100644 --- a/tests/boundary/filesystem/cliJsonInput.test.ts +++ b/tests/boundary/filesystem/cliJsonInput.test.ts @@ -7,7 +7,7 @@ import { describe, expect, it, onTestFinished } from "vitest"; import { createCli } from "../../../src/cli.js"; import { parseCliJsonInput } from "../../../src/cliJsonInput.js"; -import { readCliJsonFile } from "../../../src/cliJsonFile.js"; +import { readJsonInputFile } from "../../../src/application/JsonInputFile.js"; import { analysisCliErrorEnvelopeSchema } from "../../../src/contracts/errorSchemas.js"; import { JSON_BYTE_ORDER_MARK_MESSAGE } from "../../../src/application/Utf8JsonInput.js"; import { readWithoutFifoWriter } from "../../fixtures/fifoInput.js"; @@ -203,11 +203,11 @@ describe("CLI JSON streamed input", () => { const path = join(root, "input.json"); await writeFile(path, `${" ".repeat(9 * 1024 * 1024)}{"after":true}`); const controller = new AbortController(); - const cancelled = readCliJsonFile(path, "test-input", controller.signal); + const cancelled = readJsonInputFile(path, "test-input", controller.signal); setImmediate(() => controller.abort(new Error("cancel JSON file parse"))); await expect(cancelled).rejects.toThrow("cancel JSON file parse"); await writeFile(path, '{"after":true}'); - expect(await readCliJsonFile(path, "test-input")).toEqual({ + expect(await readJsonInputFile(path, "test-input")).toEqual({ ok: true, value: { after: true }, });