diff --git a/src/artifacts/EspFlashImageArtifactReader.test.ts b/src/artifacts/EspFlashImageArtifactReader.test.ts index 8bb8f38208..896ec33cd0 100644 --- a/src/artifacts/EspFlashImageArtifactReader.test.ts +++ b/src/artifacts/EspFlashImageArtifactReader.test.ts @@ -376,6 +376,24 @@ describe("ESP flash layout reader identity", () => { } }); + it("opens a partition past the end as the empty range at the end of the file", async () => { + const cut = DUMP.subarray(0, 0x2_8000); + const reader = new EspFlashImageArtifactReader(await writeDump(cut)); + try { + const entry = (await list(reader)).find( + ({ path }) => path === "partitions/ota_1.bin", + ); + if (entry === undefined) throw new Error("Missing entry"); + expect(entry).toMatchObject({ byteOffset: cut.length, declaredSize: 0 }); + const chunks: Buffer[] = []; + for await (const chunk of await reader.open(entry)) + chunks.push(chunk as Buffer); + expect(Buffer.concat(chunks)).toHaveLength(0); + } finally { + await reader.close(); + } + }); + it("honors cancellation before admitting a file with the exact reason", async () => { const reader = new EspFlashImageArtifactReader(await writeDump(DUMP)); const controller = new AbortController(); diff --git a/src/artifacts/EspFlashImageArtifactReader.ts b/src/artifacts/EspFlashImageArtifactReader.ts index 2650a53fa7..7ab782f900 100644 --- a/src/artifacts/EspFlashImageArtifactReader.ts +++ b/src/artifacts/EspFlashImageArtifactReader.ts @@ -1,4 +1,3 @@ -import { Readable, Transform, pipeline } from "node:stream"; import type { FileHandle } from "node:fs/promises"; import type { EspFlashImageMetadata } from "../domain/artifactFormatMetadata.js"; @@ -21,18 +20,7 @@ import { import { offsetText } from "../domain/firmware/espImageStructures.js"; import { ESP_OTA_SELECT_ENTRY_BYTES } from "../domain/firmware/espOtaData.js"; import { isEspAppPartition } from "../domain/firmware/espPartitionTable.js"; -import { - ArtifactReaderFailure, - copyArtifactEntry, - sameArtifactEntry, - type ArtifactEntry, - type ArtifactReader, - type ArtifactReaderTraversal, - type ArtifactTraversalOptions, -} from "./ArtifactReader.js"; -import { ArtifactResourceScope } from "./ArtifactResourceScope.js"; -import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; -import { ArtifactFileSource } from "./ArtifactFileSource.js"; +import { ArtifactReaderFailure } from "./ArtifactReader.js"; import { decodeEspFlashImage, firstUnerasedInRange, @@ -40,7 +28,11 @@ import { writtenExtents, type EspFlashReadContext, } from "./EspFlashRegionReads.js"; -import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; +import { + PositionalArtifactReader, + type PositionalArtifactEntries, + type PositionalArtifactEntry, +} from "./PositionalArtifactReader.js"; import type { StableRegularFileDescriptor } from "../filesystem/RegularFile.js"; const LABEL = "ESP flash image"; @@ -62,18 +54,11 @@ type EspFlashImageFacts = NonNullable< * each entry is a positional read of the carrier, rechecked against its * identity. */ -export class EspFlashImageArtifactReader implements ArtifactReader { - readonly format = "esp-flash-image" as const; - readonly #scope = new ArtifactResourceScope(); - readonly #entries = new Map(); - readonly #streams = new Set(); - readonly #source: ArtifactFileSource; +export class EspFlashImageArtifactReader extends PositionalArtifactReader { + override readonly format = "esp-flash-image" as const; readonly #located: EspFlashLocation | undefined; readonly #selection: EspFlashSelection; - #prepared: Promise | undefined; #metadata: EspFlashImageMetadata | undefined; - #coverage: ArtifactReaderTraversal | undefined; - #closing = false; constructor( path: string, @@ -82,7 +67,7 @@ export class EspFlashImageArtifactReader implements ArtifactReader { ) { const admission = admitted !== undefined && "kind" in admitted ? admitted : undefined; - this.#source = new ArtifactFileSource( + super( path, LABEL, admission?.source ?? @@ -98,143 +83,10 @@ export class EspFlashImageArtifactReader implements ArtifactReader { : structuredClone(this.#metadata); } - async prepareContainer( - expectedSha256: string, + protected override async prepareEntries( signal?: AbortSignal, - ): Promise { - await this.#scope.run(async () => { - this.#assertOpen(signal); - await this.#source.bindSha256(expectedSha256, signal); - }); - } - - async *entries( - signal?: AbortSignal, - options?: ArtifactTraversalOptions, - ): AsyncIterable { - await this.#scope.run(async () => { - this.#assertOpen(signal); - this.#prepared ??= this.#prepare(signal); - await this.#prepared; - await this.#source.read(signal); - }); - this.#assertOpen(signal); - if (this.#coverage !== undefined) - options?.onCoverage?.({ - ...this.#coverage, - limitations: [...this.#coverage.limitations], - }); - for (const entry of this.#entries.values()) { - this.#assertOpen(signal); - yield copyArtifactEntry(entry); - } - this.#assertOpen(signal); - } - - async open(entry: ArtifactEntry, signal?: AbortSignal): Promise { - return this.#scope.run(async () => { - this.#assertOpen(signal); - const stored = this.#entries.get(entry.adapterKey); - if ( - stored === undefined || - !sameArtifactEntry(entry, stored) || - stored.byteOffset === null || - stored.declaredSize === null - ) - throw new ArtifactReaderFailure( - "integrity", - `${LABEL} entry does not match its owning inventory`, - ); - const { handle } = await this.#source.read(signal); - this.#assertOpen(signal); - const source = - stored.declaredSize === 0 - ? Readable.from([]) - : readFileHandleChunks(handle, { - start: stored.byteOffset, - end: stored.byteOffset + stored.declaredSize - 1, - signal, - cancellationFailure: (reason) => - new ArtifactReaderFailure( - "cancelled", - `${LABEL} read cancelled`, - { cause: reason }, - ), - }); - const output = this.#entryStream(stored, signal); - // Retirement can destroy the stream before its consumer listens. - output.on("error", () => undefined); - this.#streams.add(output); - let producerComplete = false; - const onAbort = (): void => { - // Preserve active read/validation errors; after production, the caller - // still owns buffered bytes and cancellation must retire that result. - if (!producerComplete || output.destroyed) return; - output.destroy( - new ArtifactReaderFailure("cancelled", `${LABEL} read cancelled`, { - cause: signal?.reason, - }), - ); - }; - const retired = new Promise((resolve) => { - output.once("close", () => { - this.#streams.delete(output); - signal?.removeEventListener("abort", onAbort); - resolve(); - }); - }); - signal?.addEventListener("abort", onAbort, { once: true }); - if (signal?.aborted) onAbort(); - void this.#scope - .run(async () => { - await Promise.all([ - new Promise((resolve) => { - pipeline(source, output, () => { - producerComplete = true; - if (signal?.aborted) onAbort(); - resolve(); - }); - }), - retired, - ]); - }) - .catch((cause: unknown) => - output.destroy( - cause instanceof Error - ? cause - : new Error(`${LABEL} stream ownership failed`, { cause }), - ), - ); - return output; - }); - } - - async close(): Promise { - this.#closing = true; - for (const stream of this.#streams) - stream.destroy( - new ArtifactReaderFailure("unavailable", `${LABEL} reader closed`), - ); - await this.#scope.close(); - await this.#source.close(); - this.#entries.clear(); - } - - provenance(): readonly [] { - return []; - } - - #assertOpen(signal?: AbortSignal): void { - assertArtifactReadActive(signal); - if (this.#closing) - throw new ArtifactReaderFailure( - "unavailable", - `${LABEL} reader is closed`, - ); - } - - async #prepare(signal?: AbortSignal): Promise { - const { handle, initial } = await this.#source.read(signal); + ): Promise { + const { handle, initial } = await this.verifyCarrier(signal); const fileSize = initial.size; const location = this.#located ?? (await this.#locate(handle, fileSize, signal)); @@ -251,11 +103,7 @@ export class EspFlashImageArtifactReader implements ArtifactReader { const images = await this.#decodeImages(context, plan); const written = await this.#readWritten(context, plan); // A changed carrier is an identity failure, not a different layout. - await this.#source.read(signal); - for (const region of plan.regions) { - const entry = regionEntry(region, written.get(region), fileSize); - this.#entries.set(entry.adapterKey, entry); - } + await this.verifyCarrier(signal); this.#metadata = layoutMetadata(location, plan, facts, images); const limitations = [ ...location.limitations, @@ -267,8 +115,15 @@ export class EspFlashImageArtifactReader implements ArtifactReader { ...facts.limitations, ...erasedSlotLimitations(images, written), ]; - if (facts.coverage === "unknown" || limitations.length > 0) - this.#coverage = { status: facts.coverage, limitations }; + return { + entries: plan.regions.map((region) => + regionEntry(region, written.get(region), fileSize), + ), + coverage: + facts.coverage === "unknown" || limitations.length > 0 + ? { status: facts.coverage, limitations } + : undefined, + }; } async #locate( @@ -308,12 +163,12 @@ export class EspFlashImageArtifactReader implements ArtifactReader { ); const gapFirstUnerased: (number | undefined)[] = []; for (const gap of plan.gaps) { - this.#assertOpen(context.signal); + this.assertOpen(context.signal); gapFirstUnerased.push( await firstUnerasedInRange( context, { start: gap.offset, length: gap.length }, - () => this.#assertOpen(context.signal), + () => this.assertOpen(context.signal), ), ); } @@ -332,14 +187,14 @@ export class EspFlashImageArtifactReader implements ArtifactReader { const held = plan.regions.filter( ({ partition, available }) => partition !== undefined && available > 0, ); - this.#assertOpen(context.signal); + this.assertOpen(context.signal); const extents = await writtenExtents( context, held.map(({ fileOffset, available }) => ({ start: fileOffset, length: available, })), - () => this.#assertOpen(context.signal), + () => this.assertOpen(context.signal), ); return new Map(held.map((region, index) => [region, extents[index]!])); } @@ -360,14 +215,14 @@ export class EspFlashImageArtifactReader implements ArtifactReader { region.role === "bootloader" || (region.partition !== undefined && isEspAppPartition(region.partition)); if (!image || region.available === 0) continue; - this.#assertOpen(context.signal); + this.assertOpen(context.signal); const range = `${String(region.fileOffset)}+${String(region.available)}`; let facts = decoded.get(range); if (facts === undefined) { facts = await decodeEspFlashImage( context, { start: region.fileOffset, length: region.available }, - () => this.#assertOpen(context.signal), + () => this.assertOpen(context.signal), ); decoded.set(range, facts); } @@ -376,35 +231,6 @@ export class EspFlashImageArtifactReader implements ArtifactReader { } return images; } - - #entryStream(entry: ArtifactEntry, signal?: AbortSignal): Transform { - let bytes = 0; - const check = async (): Promise => { - if (bytes !== entry.declaredSize) - throw new ArtifactReaderFailure( - "integrity", - `${LABEL} entry ended early: ${entry.path}`, - ); - await this.#source.read(signal); - }; - return new Transform({ - transform(chunk: Buffer, _encoding, callback) { - bytes += chunk.length; - callback(null, chunk); - }, - flush(callback) { - void check().then( - () => callback(), - (cause: unknown) => - callback( - cause instanceof Error - ? cause - : new Error(`${LABEL} read failed`, { cause }), - ), - ); - }, - }); - } } /** App slots the bootloader finds empty that still hold written bytes. */ @@ -428,7 +254,7 @@ const regionEntry = ( region: EspFlashRegion, written: EspFlashWrittenExtent | undefined, fileSize: number, -): ArtifactEntry => ({ +): PositionalArtifactEntry => ({ path: region.path, ...(region.partition?.label === null || region.partition === undefined ? {} @@ -440,7 +266,8 @@ const regionEntry = ( // The table's encrypted flag says how a device stores the partition, not // that these bytes are ciphertext. encrypted: false, - // A partition past the end holds no bytes; its entry sits at the end. + // A partition past the end holds no bytes: its entry is the empty range + // at the end of the file, which opens as an empty stream without a read. byteOffset: Math.min(region.fileOffset, fileSize), declaredIntegrity: null, unpacked: false, diff --git a/src/artifacts/EspImageArtifactReader.ts b/src/artifacts/EspImageArtifactReader.ts index 39cfec7147..7e61aeafbb 100644 --- a/src/artifacts/EspImageArtifactReader.ts +++ b/src/artifacts/EspImageArtifactReader.ts @@ -1,10 +1,4 @@ -import { Readable, Transform, pipeline } from "node:stream"; - -import type { - EspImageMetadata, - EspImageSegmentMetadata, - EspImageSignatureBlockMetadata, -} from "../domain/artifactFormatMetadata.js"; +import type { EspImageMetadata } from "../domain/artifactFormatMetadata.js"; import { espChipForId } from "../domain/firmware/espChips.js"; import { espFlashDumpOrigin, @@ -22,49 +16,29 @@ import { ESP_SIGNATURE_BLOCK_BYTES, offsetText, } from "../domain/firmware/espImageStructures.js"; +import { ArtifactReaderFailure } from "./ArtifactReader.js"; import { readEspFlashRange } from "./EspFlashRegionReads.js"; import { - ArtifactReaderFailure, - copyArtifactEntry, - sameArtifactEntry, - type ArtifactEntry, - type ArtifactReader, - type ArtifactReaderTraversal, - type ArtifactTraversalOptions, -} from "./ArtifactReader.js"; -import { ArtifactResourceScope } from "./ArtifactResourceScope.js"; -import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; -import { ArtifactFileSource } from "./ArtifactFileSource.js"; + PositionalArtifactReader, + type PositionalArtifactEntries, + type PositionalArtifactEntry, +} from "./PositionalArtifactReader.js"; import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; import type { StableRegularFileDescriptor } from "../filesystem/RegularFile.js"; const LABEL = "ESP image"; -type EspImageEntry = ArtifactEntry & { - readonly byteOffset: number; - readonly declaredSize: number; - readonly formatMetadata: - | EspImageSegmentMetadata - | EspImageSignatureBlockMetadata; -}; /** * Inventory one ESP32-family image as its header-declared segments and * Secure Boot v2 signature blocks. Decoding reads the file once; each entry * is a positional read of the carrier, rechecked against its identity. */ -export class EspImageArtifactReader implements ArtifactReader { - readonly format = "esp-image" as const; - readonly #scope = new ArtifactResourceScope(); - readonly #entries = new Map(); - readonly #streams = new Set(); - readonly #source: ArtifactFileSource; - #prepared: Promise | undefined; +export class EspImageArtifactReader extends PositionalArtifactReader { + override readonly format = "esp-image" as const; #metadata: EspImageMetadata | undefined; - #coverage: ArtifactReaderTraversal | undefined; - #closing = false; constructor(path: string, admitted?: StableRegularFileDescriptor) { - this.#source = new ArtifactFileSource(path, LABEL, admitted); + super(path, LABEL, admitted); } formatMetadata(): EspImageMetadata | undefined { @@ -73,138 +47,11 @@ export class EspImageArtifactReader implements ArtifactReader { : structuredClone(this.#metadata); } - async prepareContainer( - expectedSha256: string, + protected override async prepareEntries( signal?: AbortSignal, - ): Promise { - await this.#scope.run(async () => { - this.#assertOpen(signal); - await this.#source.bindSha256(expectedSha256, signal); - }); - } - - async *entries( - signal?: AbortSignal, - options?: ArtifactTraversalOptions, - ): AsyncIterable { - await this.#scope.run(async () => { - this.#assertOpen(signal); - this.#prepared ??= this.#prepare(signal); - await this.#prepared; - await this.#source.read(signal); - }); - this.#assertOpen(signal); - if (this.#coverage !== undefined) - options?.onCoverage?.({ - ...this.#coverage, - limitations: [...this.#coverage.limitations], - }); - for (const entry of this.#entries.values()) { - this.#assertOpen(signal); - yield copyArtifactEntry(entry); - } - this.#assertOpen(signal); - } - - async open(entry: ArtifactEntry, signal?: AbortSignal): Promise { - return this.#scope.run(async () => { - this.#assertOpen(signal); - const stored = this.#entries.get(entry.adapterKey); - if (stored === undefined || !sameArtifactEntry(entry, stored)) - throw new ArtifactReaderFailure( - "integrity", - `${LABEL} entry does not match its owning inventory`, - ); - const { handle } = await this.#source.read(signal); - this.#assertOpen(signal); - const source = - stored.declaredSize === 0 - ? Readable.from([]) - : readFileHandleChunks(handle, { - start: stored.byteOffset, - end: stored.byteOffset + stored.declaredSize - 1, - signal, - cancellationFailure: (reason) => - new ArtifactReaderFailure( - "cancelled", - `${LABEL} read cancelled`, - { cause: reason }, - ), - }); - const output = this.#entryStream(stored, signal); - // Retirement can destroy the stream before its consumer listens. - output.on("error", () => undefined); - this.#streams.add(output); - let producerComplete = false; - const onAbort = (): void => { - // Preserve active read/validation errors; after production, the caller - // still owns buffered bytes and cancellation must retire that result. - if (!producerComplete || output.destroyed) return; - output.destroy( - new ArtifactReaderFailure("cancelled", `${LABEL} read cancelled`, { - cause: signal?.reason, - }), - ); - }; - const retired = new Promise((resolve) => { - output.once("close", () => { - this.#streams.delete(output); - signal?.removeEventListener("abort", onAbort); - resolve(); - }); - }); - signal?.addEventListener("abort", onAbort, { once: true }); - if (signal?.aborted) onAbort(); - void this.#scope - .run(async () => { - await Promise.all([ - new Promise((resolve) => { - pipeline(source, output, () => { - producerComplete = true; - if (signal?.aborted) onAbort(); - resolve(); - }); - }), - retired, - ]); - }) - .catch((cause: unknown) => - output.destroy( - cause instanceof Error - ? cause - : new Error(`${LABEL} stream ownership failed`, { cause }), - ), - ); - return output; - }); - } - - async close(): Promise { - this.#closing = true; - for (const stream of this.#streams) - stream.destroy( - new ArtifactReaderFailure("unavailable", `${LABEL} reader closed`), - ); - await this.#scope.close(); - await this.#source.close(); - this.#entries.clear(); - } - - provenance(): readonly [] { - return []; - } - - #assertOpen(signal?: AbortSignal): void { - assertArtifactReadActive(signal); - if (this.#closing) - throw new ArtifactReaderFailure( - "unavailable", - `${LABEL} reader is closed`, - ); - } - - async #prepare(signal?: AbortSignal): Promise { + ): Promise { const image = await this.#decode(signal); + const entries: PositionalArtifactEntry[] = []; for (const segment of image.segments) { const { index, data_len } = segment.metadata; const entry = { @@ -229,8 +76,8 @@ export class EspImageArtifactReader implements ArtifactReader { : [], adapterKey: `esp-image:segment:${String(index)}`, formatMetadata: { ...segment.metadata }, - } satisfies EspImageEntry; - this.#entries.set(entry.adapterKey, entry); + } satisfies PositionalArtifactEntry; + entries.push(entry); } for (const block of image.signatureBlocks) { const entry = { @@ -246,8 +93,8 @@ export class EspImageArtifactReader implements ArtifactReader { limitations: [], adapterKey: `esp-image:signature-block:${String(block.metadata.index)}`, formatMetadata: { ...block.metadata }, - } satisfies EspImageEntry; - this.#entries.set(entry.adapterKey, entry); + } satisfies PositionalArtifactEntry; + entries.push(entry); } const dump = await this.#flashDump(image, signal); this.#metadata = { @@ -264,8 +111,13 @@ export class EspImageArtifactReader implements ArtifactReader { ...image.limitations, ...(dump === undefined ? [] : [flashDumpLimitation(image, dump)]), ]; - if (image.coverage === "unknown" || limitations.length > 0) - this.#coverage = { status: image.coverage, limitations }; + return { + entries, + coverage: + image.coverage === "unknown" || limitations.length > 0 + ? { status: image.coverage, limitations } + : undefined, + }; } /** @@ -283,7 +135,8 @@ export class EspImageArtifactReader implements ArtifactReader { image.metadata.descriptor?.kind === "app" ) return undefined; - const { handle, initial } = await this.#source.read(signal); + this.assertOpen(signal); + const { handle, initial } = await this.verifyCarrier(signal); const context = { handle, fileSize: initial.size, signal }; return espFlashDumpOrigin( (position, length) => readEspFlashRange(context, position, length), @@ -295,7 +148,7 @@ export class EspImageArtifactReader implements ArtifactReader { /** Stream the carrier once, stopping when later bytes cannot matter. */ async #decode(signal?: AbortSignal): Promise { - const { handle, initial } = await this.#source.read(signal); + const { handle, initial } = await this.verifyCarrier(signal); const decoder = createEspImageDecoder(initial.size); if (initial.size > 0) for await (const chunk of readFileHandleChunks(handle, { @@ -306,45 +159,16 @@ export class EspImageArtifactReader implements ArtifactReader { cause: reason, }), }) as AsyncIterable) { - this.#assertOpen(signal); + this.assertOpen(signal); decoder.write(chunk); if (decoder.satisfied) break; } // A changed carrier is an identity failure, not a malformed image. - await this.#source.read(signal); + await this.verifyCarrier(signal); const image = decoder.finish(); if (!image.ok) throw decodeFailure(image.error); return image.value; } - - #entryStream(entry: EspImageEntry, signal?: AbortSignal): Transform { - let bytes = 0; - const check = async (): Promise => { - if (bytes !== entry.declaredSize) - throw new ArtifactReaderFailure( - "integrity", - `${LABEL} entry ended early: ${entry.path}`, - ); - await this.#source.read(signal); - }; - return new Transform({ - transform(chunk: Buffer, _encoding, callback) { - bytes += chunk.length; - callback(null, chunk); - }, - flush(callback) { - void check().then( - () => callback(), - (cause: unknown) => - callback( - cause instanceof Error - ? cause - : new Error(`${LABEL} read failed`, { cause }), - ), - ); - }, - }); - } } const flashDumpLimitation = ( diff --git a/src/artifacts/GodotPckArtifactReader.boundaries.test.ts b/src/artifacts/GodotPckArtifactReader.boundaries.test.ts new file mode 100644 index 0000000000..331ed75cb5 --- /dev/null +++ b/src/artifacts/GodotPckArtifactReader.boundaries.test.ts @@ -0,0 +1,9 @@ +import { artifactReaderLifecycleCases } from "../../tests/fixtures/artifactReaderLifecycle.js"; +import { buildGodotPck } from "../../tests/fixtures/godotPck.js"; +import { GodotPckArtifactReader } from "./GodotPckArtifactReader.js"; + +artifactReaderLifecycleCases({ + image: () => + buildGodotPck(4, [{ path: "res://member.txt", data: "member bytes" }]), + createReader: (path, admitted) => new GodotPckArtifactReader(path, admitted), +}); diff --git a/src/artifacts/GodotPckArtifactReader.test.ts b/src/artifacts/GodotPckArtifactReader.test.ts index b7a9a81a15..40587a34bb 100644 --- a/src/artifacts/GodotPckArtifactReader.test.ts +++ b/src/artifacts/GodotPckArtifactReader.test.ts @@ -12,7 +12,7 @@ import { join } from "node:path"; import { text } from "node:stream/consumers"; import { once } from "node:events"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory.js"; import { renameOpenParent } from "../../tests/fixtures/renameOpenParent.js"; @@ -420,6 +420,44 @@ describe("GodotPckArtifactReader member streams", () => { ); }); +describe("GodotPckArtifactReader carrier truncation", () => { + it("reports a member that ends after the carrier recheck as an integrity failure", async () => { + const path = await writePck( + buildGodotPck(4, [{ path: "member.bin", data: "member bytes" }]), + ); + const descriptor = await openRegularFile(path, { symlinks: "reject" }); + const reader = new GodotPckArtifactReader(path, descriptor); + try { + const [entry] = await list(reader); + if (entry === undefined) throw new Error("expected one entry"); + // open() rechecks the carrier's identity; the file then ends mid-member. + const read = descriptor.handle.read.bind(descriptor.handle); + let calls = 0; + const truncated = vi + .spyOn(descriptor.handle, "read") + .mockImplementation((...args: unknown[]) => { + calls += 1; + if (calls > 1) + return Promise.resolve({ bytesRead: 0, buffer: args[0] }); + args[2] = 3; + return Reflect.apply(read, descriptor.handle, args); + }); + try { + await expect(text(await reader.open(entry))).rejects.toMatchObject({ + reason: "integrity", + message: "Godot PCK entry ended early: member.bin", + }); + expect(calls).toBe(2); + } finally { + truncated.mockRestore(); + } + } finally { + await reader.close(); + await descriptor.handle.close(); + } + }); +}); + describe("Godot PCK logical-byte admission", () => { it.each([2, 3, 4] as const)( "rejects total repeated plaintext work before yielding format %s members", diff --git a/src/artifacts/GodotPckArtifactReader.ts b/src/artifacts/GodotPckArtifactReader.ts index 50d796c7b8..920066a48d 100644 --- a/src/artifacts/GodotPckArtifactReader.ts +++ b/src/artifacts/GodotPckArtifactReader.ts @@ -3,16 +3,7 @@ import { readGodotPckHeaderFacts } from "../domain/godotPckFormat.js"; import { artifactExpansionBudgetBytes } from "../domain/artifactExpansionBudget.js"; import type { FileHandle } from "node:fs/promises"; import { setImmediate } from "node:timers/promises"; -import { Readable, Transform, pipeline } from "node:stream"; -import { - ArtifactReaderFailure, - copyArtifactEntry, - sameArtifactEntry, - type ArtifactEntry, - type ArtifactReader, - type ArtifactTraversalOptions, -} from "./ArtifactReader.js"; -import { ArtifactResourceScope } from "./ArtifactResourceScope.js"; +import { ArtifactReaderFailure } from "./ArtifactReader.js"; import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; import { locateGodotPck, @@ -21,9 +12,12 @@ import { } from "./GodotPckLocation.js"; import type { GodotPckMetadata } from "../domain/artifactFormatMetadata.js"; import { jsonValueByteLength } from "../domain/jsonValueByteLength.js"; -import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; import type { StableRegularFileDescriptor } from "../filesystem/RegularFile.js"; -import { ArtifactFileSource } from "./ArtifactFileSource.js"; +import { + PositionalArtifactReader, + type PositionalArtifactEntries, + type PositionalArtifactEntry, +} from "./PositionalArtifactReader.js"; const PACK_DIR_ENCRYPTED = 1; const PACK_REL_FILEBASE = 2; @@ -35,16 +29,10 @@ const MAX_DIRECTORY_METADATA_BYTES = 64 * 1024 * 1024; const UTF8 = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }); /** One descriptor, one validated directory, and bounded positional member reads. */ -export class GodotPckArtifactReader implements ArtifactReader { - readonly format = "godot-pck" as const; - readonly #scope = new ArtifactResourceScope(); - readonly #entries = new Map(); - readonly #streams = new Set(); - readonly #source: ArtifactFileSource; +export class GodotPckArtifactReader extends PositionalArtifactReader { + override readonly format = "godot-pck" as const; readonly #location: GodotPckLocation | undefined; - #prepared: Promise | undefined; #header: GodotPckMetadata | undefined; - #closing = false; constructor( path: string, @@ -52,7 +40,7 @@ export class GodotPckArtifactReader implements ArtifactReader { ) { const admission = admitted !== undefined && "kind" in admitted ? admitted : undefined; - this.#source = new ArtifactFileSource( + super( path, "Godot PCK", admitted !== undefined && "kind" in admitted ? admitted.source : admitted, @@ -64,131 +52,10 @@ export class GodotPckArtifactReader implements ArtifactReader { return this.#header === undefined ? undefined : { ...this.#header }; } - async prepareContainer( - expectedSha256: string, + protected override async prepareEntries( signal?: AbortSignal, - ): Promise { - await this.#scope.run(async () => { - this.#assertOpen(signal); - await this.#source.bindSha256(expectedSha256, signal); - }); - } - - async *entries( - signal?: AbortSignal, - options?: ArtifactTraversalOptions, - ): AsyncIterable { - await this.#scope.run(async () => { - this.#assertOpen(signal); - this.#prepared ??= this.#prepare(signal); - await this.#prepared; - await this.#source.read(signal); - }); - if (this.#header?.directory_status !== "available") { - options?.onCoverage?.({ - status: "unknown", - limitations: [ - this.#header?.directory_status === "encrypted" - ? "Godot PCK directory is encrypted; member identities and contents remain unknown." - : "Godot PCK sparse-bundle member bytes are external and were not traversed.", - ], - }); - return; - } - let index = 0; - for (const entry of this.#entries.values()) { - if (index++ % 256 === 0) await setImmediate(); - this.#assertOpen(signal); - yield copyArtifactEntry(entry); - } - } - - async open(entry: ArtifactEntry, signal?: AbortSignal): Promise { - return this.#scope.run(async () => { - this.#assertOpen(signal); - const stored = this.#entries.get(entry.adapterKey); - if (stored === undefined || !sameArtifactEntry(entry, stored)) - throw new ArtifactReaderFailure( - "integrity", - "Godot PCK entry does not match its owning inventory", - ); - if (stored.encrypted) - throw new ArtifactReaderFailure( - "unavailable", - `Encrypted Godot PCK member is unsupported: ${stored.path}`, - ); - const { handle } = await this.#source.read(signal); - this.#assertOpen(signal); - if (stored.byteOffset === null || stored.declaredSize === null) - throw formatFailure("member has no physical byte range"); - const source = - stored.declaredSize === 0 - ? Readable.from([]) - : readFileHandleChunks(handle, { - start: stored.byteOffset, - end: stored.byteOffset + stored.declaredSize - 1, - signal, - cancellationFailure: (reason) => - new ArtifactReaderFailure( - "cancelled", - "Godot PCK read cancelled", - { cause: reason }, - ), - }); - const output = this.#memberStream(stored, signal); - // The returned stream still reports failures to its consumer. Also handle - // retirement before a caller has attached its own error listener. - output.on("error", () => undefined); - this.#streams.add(output); - // Writable completion can leave an unread member buffered on the readable - // side. The returned stream stays owned until that side closes too. - output.once("close", () => this.#streams.delete(output)); - void this.#scope - .run( - () => - new Promise((resolve) => { - pipeline(source, output, () => { - resolve(); - }); - }), - ) - .catch((cause: unknown) => - output.destroy( - cause instanceof Error - ? cause - : new Error("Godot PCK stream ownership failed", { cause }), - ), - ); - return output; - }); - } - - async close(): Promise { - this.#closing = true; - for (const stream of this.#streams) - stream.destroy( - new ArtifactReaderFailure("unavailable", "Godot PCK reader closed"), - ); - await this.#scope.close(); - await this.#source.close(); - this.#entries.clear(); - } - - provenance(): readonly [] { - return []; - } - - #assertOpen(signal?: AbortSignal): void { - assertArtifactReadActive(signal); - if (this.#closing) - throw new ArtifactReaderFailure( - "unavailable", - "Godot PCK reader is closed", - ); - } - - async #prepare(signal?: AbortSignal): Promise { - const { handle, initial } = await this.#source.read(signal); + ): Promise { + const { handle, initial } = await this.verifyCarrier(signal); const header = await readGodotPckHeader( handle, initial.size, @@ -196,13 +63,25 @@ export class GodotPckArtifactReader implements ArtifactReader { this.#location, ); this.#header = header; - if (header.directory_status !== "available") return; + if (header.directory_status !== "available") + return { + entries: [], + coverage: { + status: "unknown", + limitations: [ + header.directory_status === "encrypted" + ? "Godot PCK directory is encrypted; member identities and contents remain unknown." + : "Godot PCK sparse-bundle member bytes are external and were not traversed.", + ], + }, + }; const cursor = new PckCursor( handle, header.directory_start + 4, header.pack_end, signal, ); + const entries: PositionalArtifactEntry[] = []; let retainedBytes = 0; let declaredPlaintextBytes = 0n; const expansionBudget = BigInt( @@ -281,41 +160,16 @@ export class GodotPckArtifactReader implements ArtifactReader { ? ("all-zero" as const) : ("declared" as const), }, - } satisfies ArtifactEntry; + } satisfies PositionalArtifactEntry; retainedBytes += jsonValueByteLength( metadata, MAX_DIRECTORY_METADATA_BYTES - retainedBytes, ); if (retainedBytes > MAX_DIRECTORY_METADATA_BYTES) throw metadataLimit(); - this.#entries.set(metadata.adapterKey, metadata); + entries.push(metadata); } - await this.#source.read(signal); - } - - #memberStream(entry: ArtifactEntry, signal?: AbortSignal): Transform { - let bytes = 0; - const check = async (): Promise => { - if (bytes !== entry.declaredSize) - throw formatFailure(`member ended early: ${entry.path}`); - await this.#source.read(signal); - }; - return new Transform({ - transform(chunk: Buffer, _encoding, callback) { - bytes += chunk.length; - callback(null, chunk); - }, - flush(callback) { - void check().then( - () => callback(), - (cause: unknown) => - callback( - cause instanceof Error - ? cause - : new Error("Godot PCK read failed", { cause }), - ), - ); - }, - }); + await this.verifyCarrier(signal); + return { entries, coverage: undefined }; } } diff --git a/src/artifacts/NodeSeaArtifactReader.boundaries.test.ts b/src/artifacts/NodeSeaArtifactReader.boundaries.test.ts new file mode 100644 index 0000000000..1a172fe30f --- /dev/null +++ b/src/artifacts/NodeSeaArtifactReader.boundaries.test.ts @@ -0,0 +1,8 @@ +import { artifactReaderLifecycleCases } from "../../tests/fixtures/artifactReaderLifecycle.js"; +import { nodeSeaElf } from "./nodeSea/NodeSea.fixture.js"; +import { NodeSeaArtifactReader } from "./NodeSeaArtifactReader.js"; + +artifactReaderLifecycleCases({ + image: () => nodeSeaElf({ fuse: "set" }), + createReader: (path, admitted) => new NodeSeaArtifactReader(path, admitted), +}); diff --git a/src/artifacts/NodeSeaArtifactReader.test.ts b/src/artifacts/NodeSeaArtifactReader.test.ts index 0e6f78d465..45b6950f4a 100644 --- a/src/artifacts/NodeSeaArtifactReader.test.ts +++ b/src/artifacts/NodeSeaArtifactReader.test.ts @@ -37,6 +37,7 @@ import { elfNote, elfNoteFixture } from "../native/elf/ElfNotes.fixture.js"; import { peResourceFixture } from "../native/pe/PeResources.fixture.js"; import { locateNodeSea } from "./nodeSea/NodeSeaLocation.js"; import { MAX_INSPECT_PE_FILE_BYTES } from "../domain/native/peResources.js"; +import { openRegularFile } from "../filesystem/RegularFile.js"; import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory.js"; import { scanCanonicalArtifactInventory } from "../../tests/fixtures/artifactInventory.js"; @@ -683,6 +684,42 @@ it("rejects forged entries and changed carrier identity, then closes idempotentl } }); +it("reports a member that ends after the carrier recheck as an integrity failure", async () => { + const path = await writeCarrier("ended", nodeSeaElf()); + const descriptor = await openRegularFile(path, { symlinks: "reject" }); + const reader = new NodeSeaArtifactReader(path, descriptor); + try { + const entries = []; + for await (const entry of reader.entries()) entries.push(entry); + const entry = entries[0]; + if (entry === undefined) throw new Error("Missing main script"); + // open() rechecks the carrier's identity; the file then ends mid-member. + const read = descriptor.handle.read.bind(descriptor.handle); + let calls = 0; + const truncated = vi + .spyOn(descriptor.handle, "read") + .mockImplementation((...args: unknown[]) => { + calls += 1; + if (calls > 1) + return Promise.resolve({ bytesRead: 0, buffer: args[0] }); + args[2] = 3; + return Reflect.apply(read, descriptor.handle, args); + }); + try { + await expect(buffer(await reader.open(entry))).rejects.toMatchObject({ + reason: "integrity", + message: "Node SEA entry ended early: sea.js", + }); + expect(calls).toBe(2); + } finally { + truncated.mockRestore(); + } + } finally { + await reader.close(); + await descriptor.handle.close(); + } +}); + it.each([8, 1024 * 1024])( "retires an unread %i-byte member through the shared reader factory", async (size) => { diff --git a/src/artifacts/NodeSeaArtifactReader.ts b/src/artifacts/NodeSeaArtifactReader.ts index d4b20ed235..bc86c50a1a 100644 --- a/src/artifacts/NodeSeaArtifactReader.ts +++ b/src/artifacts/NodeSeaArtifactReader.ts @@ -1,5 +1,4 @@ import type { NodeSeaAdmission } from "./nodeSea/NodeSeaAdmission.js"; -import { Readable, Transform, pipeline } from "node:stream"; import { setImmediate } from "node:timers/promises"; import { artifactExpansionBudgetBytes } from "../domain/artifactExpansionBudget.js"; @@ -8,16 +7,7 @@ import { ArtifactPathRegistry, normalizeArtifactPath, } from "./ArtifactPaths.js"; -import { - ArtifactReaderFailure, - copyArtifactEntry, - sameArtifactEntry, - type ArtifactEntry, - type ArtifactReader, - type ArtifactTraversalOptions, -} from "./ArtifactReader.js"; -import { ArtifactResourceScope } from "./ArtifactResourceScope.js"; -import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; +import { ArtifactReaderFailure } from "./ArtifactReader.js"; import { NODE_SEA_BLOB_TOO_LARGE, NODE_SEA_MAX_BLOB_BYTES, @@ -31,9 +21,12 @@ import { scanNodeSeaFuse, type NodeSeaLocation, } from "./nodeSea/NodeSeaLocation.js"; -import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; import type { StableRegularFileDescriptor } from "../filesystem/RegularFile.js"; -import { ArtifactFileSource } from "./ArtifactFileSource.js"; +import { + PositionalArtifactReader, + type PositionalArtifactEntries, + type PositionalArtifactEntry, +} from "./PositionalArtifactReader.js"; const SNAPSHOT_PATH = "__node_sea__/snapshot"; const CODE_CACHE_PATH = "__node_sea__/code-cache"; @@ -43,27 +36,23 @@ const ASSET_FALLBACK_PREFIX = "__node_sea__/assets/"; type NodeSeaRole = "main-script" | "snapshot" | "code-cache" | "asset"; /** Read a Node.js SEA blob out of a Mach-O, ELF, or PE carrier without running it. */ -export class NodeSeaArtifactReader implements ArtifactReader { - readonly format = "node-sea" as const; - readonly #scope = new ArtifactResourceScope(); - readonly #entries = new Map(); - readonly #streams = new Set(); - readonly #source: ArtifactFileSource; - #prepared: Promise | undefined; +export class NodeSeaArtifactReader extends PositionalArtifactReader { + override readonly format = "node-sea" as const; #header: NodeSeaMetadata | undefined; #admission: NodeSeaAdmission | undefined; - #closing = false; constructor( path: string, admitted?: StableRegularFileDescriptor | NodeSeaAdmission, ) { - if (admitted !== undefined && "kind" in admitted) { - this.#admission = admitted; - this.#source = new ArtifactFileSource(path, "Node SEA", admitted.source); - } else { - this.#source = new ArtifactFileSource(path, "Node SEA", admitted); - } + const admission = + admitted !== undefined && "kind" in admitted ? admitted : undefined; + super( + path, + "Node SEA", + admitted !== undefined && "kind" in admitted ? admitted.source : admitted, + ); + this.#admission = admission; } formatMetadata(): NodeSeaMetadata | undefined { @@ -71,123 +60,16 @@ export class NodeSeaArtifactReader implements ArtifactReader { return { ...this.#header, exec_argv: [...this.#header.exec_argv] }; } - async prepareContainer( - expectedSha256: string, - signal?: AbortSignal, - ): Promise { - await this.#scope.run(async () => { - this.#assertOpen(signal); - await this.#source.bindSha256(expectedSha256, signal); - }); - } - - async *entries( - signal?: AbortSignal, - options?: ArtifactTraversalOptions, - ): AsyncIterable { - await this.#scope.run(async () => { - this.#assertOpen(signal); - this.#prepared ??= this.#prepare(signal); - await this.#prepared; - await this.#source.read(signal); - }); - const contradiction = this.#header?.fuse_contradiction; - if (contradiction !== null && contradiction !== undefined) - options?.onCoverage?.({ - status: "complete", - limitations: [contradiction], - }); - let index = 0; - for (const entry of this.#entries.values()) { - if (index % 256 === 0) await setImmediate(); - index += 1; - this.#assertOpen(signal); - yield copyArtifactEntry(entry); - } - } - - async open(entry: ArtifactEntry, signal?: AbortSignal): Promise { - return this.#scope.run(async () => { - this.#assertOpen(signal); - const stored = this.#entries.get(entry.adapterKey); - if (stored === undefined || !sameArtifactEntry(entry, stored)) - throw new ArtifactReaderFailure( - "integrity", - "Node SEA entry does not match its owning inventory", - ); - const { handle } = await this.#source.read(signal); - this.#assertOpen(signal); - if (stored.byteOffset === null || stored.declaredSize === null) - throw formatFailure("member has no physical byte range"); - const source = - stored.declaredSize === 0 - ? Readable.from([]) - : readFileHandleChunks(handle, { - start: stored.byteOffset, - end: stored.byteOffset + stored.declaredSize - 1, - signal, - cancellationFailure: (reason) => - new ArtifactReaderFailure( - "cancelled", - "Node SEA read cancelled", - { - cause: reason, - }, - ), - }); - const output = this.#memberStream(stored, signal); - output.on("error", () => undefined); - this.#streams.add(output); - // A finished producer can leave source bytes buffered for its consumer. - // Keep the readable side owned until it closes as well. - output.once("close", () => this.#streams.delete(output)); - void this.#scope - .run( - () => - new Promise((resolve) => { - pipeline(source, output, () => { - resolve(); - }); - }), - ) - .catch((cause: unknown) => - output.destroy( - cause instanceof Error - ? cause - : new Error("Node SEA stream ownership failed", { cause }), - ), - ); - return output; - }); - } - - async close(): Promise { - this.#closing = true; - for (const stream of this.#streams) - stream.destroy( - new ArtifactReaderFailure("unavailable", "Node SEA reader closed"), - ); - await this.#scope.close(); - await this.#source.close(); - this.#entries.clear(); + /** Also drop the admitted decode, which only a completed preparation releases. */ + override async close(): Promise { + await super.close(); this.#admission = undefined; } - provenance(): readonly [] { - return []; - } - - #assertOpen(signal?: AbortSignal): void { - assertArtifactReadActive(signal); - if (this.#closing) - throw new ArtifactReaderFailure( - "unavailable", - "Node SEA reader is closed", - ); - } - - async #prepare(signal?: AbortSignal): Promise { - const { handle, initial } = await this.#source.read(signal); + protected override async prepareEntries( + signal?: AbortSignal, + ): Promise { + const { handle, initial } = await this.verifyCarrier(signal); // Freshness remains owned by ArtifactFileSource even when classification // already decoded this exact borrowed descriptor's blob. let location: NodeSeaLocation; @@ -208,6 +90,7 @@ export class NodeSeaArtifactReader implements ArtifactReader { ); } const fuse = await scanNodeSeaFuse(handle, initial.size, signal); + const entries: PositionalArtifactEntry[] = []; let declaredBytes = 0n; const expansionBudget = BigInt( artifactExpansionBudgetBytes(location.blobLength), @@ -215,7 +98,7 @@ export class NodeSeaArtifactReader implements ArtifactReader { let index = 0; for (const member of seaMembers(decoded)) { if (index % 256 === 0) await setImmediate(); - this.#assertOpen(signal); + this.assertOpen(signal); declaredBytes += BigInt(member.range.length); if (declaredBytes > expansionBudget) throw new ArtifactReaderFailure( @@ -245,8 +128,8 @@ export class NodeSeaArtifactReader implements ArtifactReader { index, }, sourcePath: member.sourcePath, - } satisfies ArtifactEntry; - this.#entries.set(entry.adapterKey, entry); + } satisfies PositionalArtifactEntry; + entries.push(entry); index += 1; } this.#header = { @@ -268,34 +151,15 @@ export class NodeSeaArtifactReader implements ArtifactReader { fuse_sentinel: fuse.fuseSentinel, fuse_contradiction: fuse.fuseContradiction, }; - await this.#source.read(signal); + await this.verifyCarrier(signal); this.#admission = undefined; - } - - #memberStream(entry: ArtifactEntry, signal?: AbortSignal): Transform { - let bytes = 0; - const check = async (): Promise => { - if (bytes !== entry.declaredSize) - throw formatFailure(`member ended early: ${entry.path}`); - await this.#source.read(signal); + return { + entries, + coverage: + fuse.fuseContradiction === null + ? undefined + : { status: "complete", limitations: [fuse.fuseContradiction] }, }; - return new Transform({ - transform(chunk: Buffer, _encoding, callback) { - bytes += chunk.length; - callback(null, chunk); - }, - flush(callback) { - void check().then( - () => callback(), - (cause: unknown) => - callback( - cause instanceof Error - ? cause - : new Error("Node SEA read failed", { cause }), - ), - ); - }, - }); } } diff --git a/src/artifacts/PositionalArtifactReader.ts b/src/artifacts/PositionalArtifactReader.ts new file mode 100644 index 0000000000..7b26e2420f --- /dev/null +++ b/src/artifacts/PositionalArtifactReader.ts @@ -0,0 +1,256 @@ +import { Readable, Transform, pipeline } from "node:stream"; +import { setImmediate } from "node:timers/promises"; + +import { + ArtifactReaderFailure, + copyArtifactEntry, + sameArtifactEntry, + type ArtifactEntry, + type ArtifactReader, + type ArtifactReaderTraversal, + type ArtifactTraversalOptions, +} from "./ArtifactReader.js"; +import { ArtifactResourceScope } from "./ArtifactResourceScope.js"; +import { assertArtifactReadActive } from "./ArtifactFileOperation.js"; +import { ArtifactFileSource } from "./ArtifactFileSource.js"; +import { readFileHandleChunks } from "../filesystem/readFileHandleChunks.js"; +import type { StableRegularFileDescriptor } from "../filesystem/RegularFile.js"; + +/** An entry whose bytes are one range of the carrier. */ +export type PositionalArtifactEntry = ArtifactEntry & { + readonly byteOffset: number; + readonly declaredSize: number; +}; + +/** The entries one preparation established and the coverage to report. */ +export interface PositionalArtifactEntries { + readonly entries: readonly PositionalArtifactEntry[]; + /** Undefined when coverage is complete without limitations. */ + readonly coverage: ArtifactReaderTraversal | undefined; +} + +/** + * Serve entries that are byte ranges of one carrier file. Preparation runs + * once; every traversal, open and completed entry stream rechecks the + * carrier's identity, and close retires open streams before the carrier. + * Encrypted entries are listed but never opened. + */ +export abstract class PositionalArtifactReader implements ArtifactReader { + abstract readonly format: ArtifactReader["format"]; + readonly #scope = new ArtifactResourceScope(); + readonly #entries = new Map(); + readonly #streams = new Set(); + readonly #label: string; + readonly #source: ArtifactFileSource; + #prepared: Promise | undefined; + #coverage: ArtifactReaderTraversal | undefined; + #closing = false; + + constructor( + path: string, + label: string, + admitted?: StableRegularFileDescriptor, + ) { + this.#label = label; + this.#source = new ArtifactFileSource(path, label, admitted); + } + + /** Read the carrier once and describe its entries. */ + protected abstract prepareEntries( + signal?: AbortSignal, + ): Promise; + + async prepareContainer( + expectedSha256: string, + signal?: AbortSignal, + ): Promise { + await this.#scope.run(async () => { + this.assertOpen(signal); + await this.#source.bindSha256(expectedSha256, signal); + }); + } + + async *entries( + signal?: AbortSignal, + options?: ArtifactTraversalOptions, + ): AsyncIterable { + await this.#scope.run(async () => { + this.assertOpen(signal); + this.#prepared ??= this.#prepare(signal); + await this.#prepared; + await this.#source.read(signal); + }); + this.assertOpen(signal); + if (this.#coverage !== undefined) + options?.onCoverage?.({ + ...this.#coverage, + limitations: [...this.#coverage.limitations], + }); + let index = 0; + for (const entry of this.#entries.values()) { + // Yield between batches so a large inventory does not hold the loop. + if (index++ % 256 === 0) await setImmediate(); + this.assertOpen(signal); + yield copyArtifactEntry(entry); + } + this.assertOpen(signal); + } + + async open(entry: ArtifactEntry, signal?: AbortSignal): Promise { + return this.#scope.run(async () => { + this.assertOpen(signal); + const stored = this.#entries.get(entry.adapterKey); + if (stored === undefined || !sameArtifactEntry(entry, stored)) + throw new ArtifactReaderFailure( + "integrity", + `${this.#label} entry does not match its owning inventory`, + ); + if (stored.encrypted) + throw new ArtifactReaderFailure( + "unavailable", + `Encrypted ${this.#label} member is unsupported: ${stored.path}`, + ); + const { handle } = await this.#source.read(signal); + this.assertOpen(signal); + const source = + stored.declaredSize === 0 + ? Readable.from([]) + : readFileHandleChunks(handle, { + start: stored.byteOffset, + end: stored.byteOffset + stored.declaredSize - 1, + signal, + cancellationFailure: (reason) => + new ArtifactReaderFailure( + "cancelled", + `${this.#label} read cancelled`, + { cause: reason }, + ), + }); + const output = this.#entryStream(stored, signal); + // Retirement can destroy the stream before its consumer listens. + output.on("error", () => undefined); + this.#streams.add(output); + let producerComplete = false; + const onAbort = (): void => { + // Preserve active read/validation errors; after production, the caller + // still owns buffered bytes and cancellation must retire that result. + if (!producerComplete || output.destroyed) return; + output.destroy( + new ArtifactReaderFailure( + "cancelled", + `${this.#label} read cancelled`, + { cause: signal?.reason }, + ), + ); + }; + // Writable completion can leave unread bytes buffered on the readable + // side, so the stream stays owned until that side closes too. + const retired = new Promise((resolve) => { + output.once("close", () => { + this.#streams.delete(output); + signal?.removeEventListener("abort", onAbort); + resolve(); + }); + }); + signal?.addEventListener("abort", onAbort, { once: true }); + if (signal?.aborted) onAbort(); + void this.#scope + .run(async () => { + await Promise.all([ + new Promise((resolve) => { + pipeline(source, output, () => { + producerComplete = true; + if (signal?.aborted) onAbort(); + resolve(); + }); + }), + retired, + ]); + }) + .catch((cause: unknown) => + output.destroy( + cause instanceof Error + ? cause + : new Error(`${this.#label} stream ownership failed`, { + cause, + }), + ), + ); + return output; + }); + } + + async close(): Promise { + this.#closing = true; + for (const stream of this.#streams) + stream.destroy( + new ArtifactReaderFailure( + "unavailable", + `${this.#label} reader closed`, + ), + ); + await this.#scope.close(); + await this.#source.close(); + this.#entries.clear(); + } + + provenance(): readonly [] { + return []; + } + + /** Recheck the carrier's identity and borrow its descriptor. */ + protected verifyCarrier( + signal?: AbortSignal, + ): Promise { + return this.#source.read(signal); + } + + /** Refuse work once the caller cancels or close begins. */ + protected assertOpen(signal?: AbortSignal): void { + assertArtifactReadActive(signal); + if (this.#closing) + throw new ArtifactReaderFailure( + "unavailable", + `${this.#label} reader is closed`, + ); + } + + async #prepare(signal?: AbortSignal): Promise { + const { entries, coverage } = await this.prepareEntries(signal); + for (const entry of entries) this.#entries.set(entry.adapterKey, entry); + this.#coverage = coverage; + } + + #entryStream( + entry: PositionalArtifactEntry, + signal?: AbortSignal, + ): Transform { + const label = this.#label; + let bytes = 0; + const check = async (): Promise => { + if (bytes !== entry.declaredSize) + throw new ArtifactReaderFailure( + "integrity", + `${label} entry ended early: ${entry.path}`, + ); + await this.#source.read(signal); + }; + return new Transform({ + transform(chunk: Buffer, _encoding, callback) { + bytes += chunk.length; + callback(null, chunk); + }, + flush(callback) { + void check().then( + () => callback(), + (cause: unknown) => + callback( + cause instanceof Error + ? cause + : new Error(`${label} read failed`, { cause }), + ), + ); + }, + }); + } +}