refactor(artifacts): unify stream lifecycle and nested archive traversal
This commit is contained in:
parent
be14aa0de9
commit
c5a16b7e21
|
|
@ -0,0 +1,52 @@
|
|||
import { Readable, Transform } from "node:stream";
|
||||
import { buffer } from "node:stream/consumers";
|
||||
import { expect, it } from "vitest";
|
||||
import { ArtifactReaderFailure } from "./ArtifactReader.js";
|
||||
import { artifactStreamPipeline } from "./ArtifactStreamPipeline.js";
|
||||
|
||||
it("preserves tagged failures from intermediate verification stages", async () => {
|
||||
const failure = new ArtifactReaderFailure(
|
||||
"integrity",
|
||||
"stored checksum disagrees",
|
||||
);
|
||||
const verifier = new Transform({
|
||||
transform(_chunk, _encoding, done) {
|
||||
done(failure);
|
||||
},
|
||||
});
|
||||
const output = artifactStreamPipeline(
|
||||
Readable.from([Buffer.from("encoded")]),
|
||||
[verifier],
|
||||
{ path: "member" },
|
||||
);
|
||||
await expect(buffer(output)).rejects.toBe(failure);
|
||||
});
|
||||
|
||||
it("closes the source when the consumer stops before exhausting member data", async () => {
|
||||
const source = Readable.from(
|
||||
(function* () {
|
||||
for (let index = 0; index < 1000; index++) yield Buffer.alloc(64 * 1024);
|
||||
})(),
|
||||
);
|
||||
const closed = new Promise<void>((resolve) =>
|
||||
source.once("close", () => resolve()),
|
||||
);
|
||||
const output = artifactStreamPipeline(source, [], { path: "member" });
|
||||
for await (const chunk of output) {
|
||||
expect(Buffer.isBuffer(chunk)).toBe(true);
|
||||
break;
|
||||
}
|
||||
await closed;
|
||||
expect(source.destroyed).toBe(true);
|
||||
});
|
||||
|
||||
it("observes abort before a consumer starts reading", async () => {
|
||||
const controller = new AbortController();
|
||||
const source = Readable.from([Buffer.from("data")]);
|
||||
const output = artifactStreamPipeline(source, [], {
|
||||
path: "member",
|
||||
signal: controller.signal,
|
||||
});
|
||||
controller.abort();
|
||||
await expect(buffer(output)).rejects.toMatchObject({ reason: "cancelled" });
|
||||
});
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
import { PassThrough, Readable, pipeline, type Transform } from "node:stream";
|
||||
import { ArtifactReaderFailure } from "./ArtifactReader.js";
|
||||
|
||||
/**
|
||||
* Connect a decoder chain as one lifecycle: failures propagate in both
|
||||
* directions, cancellation destroys every stage, and consumer early-return
|
||||
* cannot leave the archive source reading after its handle is closed.
|
||||
*/
|
||||
export const artifactStreamPipeline = (
|
||||
source: Readable,
|
||||
stages: readonly Transform[],
|
||||
context: {
|
||||
readonly path: string;
|
||||
readonly signal?: AbortSignal | undefined;
|
||||
},
|
||||
): Readable => {
|
||||
const output = new PassThrough();
|
||||
const aborted = () =>
|
||||
output.destroy(
|
||||
new ArtifactReaderFailure(
|
||||
"cancelled",
|
||||
`Archive read cancelled: ${context.path}`,
|
||||
),
|
||||
);
|
||||
pipeline([source, ...stages, output], () => {
|
||||
// The readable destination retains any failure for its consumer. Keep
|
||||
// cancellation active until buffered output has also been consumed.
|
||||
});
|
||||
context.signal?.addEventListener("abort", aborted, { once: true });
|
||||
if (context.signal?.aborted) aborted();
|
||||
const dispose = () => {
|
||||
context.signal?.removeEventListener("abort", aborted);
|
||||
output.destroy();
|
||||
};
|
||||
const consumer = Readable.from(
|
||||
(async function* () {
|
||||
try {
|
||||
for await (const chunk of output) yield chunk;
|
||||
} catch (cause: unknown) {
|
||||
if (cause instanceof ArtifactReaderFailure) throw cause;
|
||||
if (context.signal?.aborted)
|
||||
throw new ArtifactReaderFailure(
|
||||
"cancelled",
|
||||
`Archive read cancelled: ${context.path}`,
|
||||
{ cause },
|
||||
);
|
||||
throw new ArtifactReaderFailure(
|
||||
"format",
|
||||
`Archive member could not be decoded: ${context.path}`,
|
||||
{ cause },
|
||||
);
|
||||
} finally {
|
||||
dispose();
|
||||
}
|
||||
})(),
|
||||
);
|
||||
// Also covers destroy() before the async generator is first advanced.
|
||||
consumer.once("close", dispose);
|
||||
return consumer;
|
||||
};
|
||||
|
|
@ -0,0 +1,91 @@
|
|||
import { isAbsolute } from "node:path";
|
||||
import { normalizeArtifactPath } from "./ArtifactPaths.js";
|
||||
import type { ArtifactEntry, ArtifactReader } from "./ArtifactReader.js";
|
||||
import { AsarArtifactReader } from "./AsarArtifactReader.js";
|
||||
import { CpioArtifactReader } from "./CpioArtifactReader.js";
|
||||
|
||||
/** One reader entry plus its stable logical path and expansion eligibility. */
|
||||
export interface ArtifactTreeEntry {
|
||||
readonly reader: ArtifactReader;
|
||||
readonly entry: ArtifactEntry;
|
||||
readonly path: string;
|
||||
readonly container: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared inventory/extraction traversal. The visitor decides whether verified
|
||||
* content may expand; this boundary owns nested paths, reader creation and
|
||||
* cleanup so the two workflows cannot disagree about container topology.
|
||||
*/
|
||||
export const visitArtifactTree = async (
|
||||
root: ArtifactReader,
|
||||
visit: (item: ArtifactTreeEntry) => Promise<boolean>,
|
||||
options: {
|
||||
readonly signal?: AbortSignal | undefined;
|
||||
readonly integrity?: "fail" | "record-and-continue";
|
||||
readonly expandAsar?: boolean;
|
||||
} = {},
|
||||
): Promise<void> => {
|
||||
const stack = [
|
||||
{
|
||||
reader: root,
|
||||
prefix: "",
|
||||
iterator: root.entries(options.signal)[Symbol.asyncIterator](),
|
||||
owned: false,
|
||||
},
|
||||
];
|
||||
try {
|
||||
while (stack.length > 0) {
|
||||
const frame = stack.at(-1);
|
||||
if (frame === undefined) break;
|
||||
const next = await frame.iterator.next();
|
||||
if (next.done === true) {
|
||||
stack.pop();
|
||||
if (frame.owned) await frame.reader.close();
|
||||
continue;
|
||||
}
|
||||
const entry = next.value;
|
||||
const path = normalizeArtifactPath(
|
||||
frame.prefix === "" ? entry.path : `${frame.prefix}/${entry.path}`,
|
||||
);
|
||||
const cpio = entry.nestedArchive === "gzip-cpio";
|
||||
const asar =
|
||||
options.expandAsar !== false &&
|
||||
entry.kind === "file" &&
|
||||
path.toLowerCase().endsWith(".asar") &&
|
||||
isAbsolute(entry.adapterKey);
|
||||
const container =
|
||||
!entry.encrypted && entry.contentUnavailable !== true && (cpio || asar);
|
||||
const expand = await visit({
|
||||
reader: frame.reader,
|
||||
entry,
|
||||
path,
|
||||
container,
|
||||
});
|
||||
if (!container || !expand) continue;
|
||||
const parent = frame.reader;
|
||||
const nested = cpio
|
||||
? new CpioArtifactReader(
|
||||
(signal) => parent.open(entry, signal),
|
||||
options.integrity ?? "fail",
|
||||
parent.decodedBudget,
|
||||
)
|
||||
: new AsarArtifactReader(entry.adapterKey);
|
||||
stack.push({
|
||||
reader: nested,
|
||||
prefix: path,
|
||||
iterator: nested.entries(options.signal)[Symbol.asyncIterator](),
|
||||
owned: true,
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
await Promise.allSettled(
|
||||
stack
|
||||
.filter(({ owned }) => owned)
|
||||
.map(async ({ reader, iterator }) => {
|
||||
await iterator.return?.();
|
||||
await reader.close();
|
||||
}),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
|
@ -4,6 +4,7 @@ import {
|
|||
ArtifactDecodedBudget,
|
||||
} from "./ArtifactDecodedBudget.js";
|
||||
import { createGunzip } from "node:zlib";
|
||||
import { artifactStreamPipeline } from "./ArtifactStreamPipeline.js";
|
||||
|
||||
import type { ArtifactCommand } from "../domain/artifactGraph.js";
|
||||
import {
|
||||
|
|
@ -88,6 +89,18 @@ class ByteSource {
|
|||
async close(): Promise<void> {
|
||||
await this.#iterator.return?.();
|
||||
}
|
||||
|
||||
/** Consume legal zero padding through decoder EOF, including gzip trailer verification. */
|
||||
async finish(): Promise<void> {
|
||||
do {
|
||||
if (this.#buffer.some((byte) => byte !== 0))
|
||||
throw new ArtifactReaderFailure(
|
||||
"format",
|
||||
"cpio has non-padding bytes after its trailer",
|
||||
);
|
||||
this.#buffer = Buffer.alloc(0);
|
||||
} while (await this.#fill());
|
||||
}
|
||||
}
|
||||
|
||||
interface CpioHeader {
|
||||
|
|
@ -262,41 +275,16 @@ export class CpioArtifactReader implements ArtifactReader {
|
|||
async *entries(signal?: AbortSignal): AsyncIterable<ArtifactEntry> {
|
||||
this.#signal = signal;
|
||||
const compressed = await this.openCompressed(signal);
|
||||
const gunzip = createGunzip();
|
||||
// `pipe()` does not forward source errors (cancellation, truncation);
|
||||
// propagate them so the MCP process fails as a tagged Artifact failure.
|
||||
compressed.on("error", (cause: unknown) => {
|
||||
gunzip.destroy(
|
||||
cause instanceof ArtifactReaderFailure
|
||||
? cause
|
||||
: new ArtifactReaderFailure("format", "cpio stream failed", {
|
||||
cause,
|
||||
}),
|
||||
);
|
||||
});
|
||||
gunzip.on("error", () => {
|
||||
// Prevent unhandled source errors after the decoder fails first.
|
||||
compressed.destroy();
|
||||
});
|
||||
const budget = new ArtifactBudgetTransform(
|
||||
this.decodedBudget,
|
||||
"gzip-cpio",
|
||||
signal,
|
||||
const source = new ByteSource(
|
||||
artifactStreamPipeline(
|
||||
compressed,
|
||||
[
|
||||
createGunzip(),
|
||||
new ArtifactBudgetTransform(this.decodedBudget, "gzip-cpio", signal),
|
||||
],
|
||||
{ path: "gzip-cpio", signal },
|
||||
),
|
||||
);
|
||||
gunzip.on("error", (cause: unknown) => {
|
||||
budget.destroy(
|
||||
cause instanceof ArtifactReaderFailure
|
||||
? cause
|
||||
: new ArtifactReaderFailure("format", "cpio stream failed", {
|
||||
cause,
|
||||
}),
|
||||
);
|
||||
});
|
||||
budget.on("error", () => {
|
||||
compressed.destroy();
|
||||
gunzip.destroy();
|
||||
});
|
||||
const source = new ByteSource(compressed.pipe(gunzip).pipe(budget));
|
||||
this.#source = source;
|
||||
for (let index = 0; ; index++) {
|
||||
if (signal?.aborted === true)
|
||||
|
|
@ -342,6 +330,12 @@ export class CpioArtifactReader implements ArtifactReader {
|
|||
);
|
||||
}
|
||||
if (raw === TRAILER) {
|
||||
if (header.fileSize !== 0)
|
||||
throw new ArtifactReaderFailure(
|
||||
"format",
|
||||
"cpio trailer declares member data",
|
||||
);
|
||||
await source.finish();
|
||||
yield* this.#unresolvedLinks();
|
||||
return;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -155,3 +155,20 @@ it("observes cancellation during internal symlink drains", async () => {
|
|||
await reader.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("verifies gzip integrity even when the cpio trailer has already been decoded", async () => {
|
||||
const bytes = Buffer.from(
|
||||
gzipCpio([{ name: "a", mode: MODE.file, data: "ok" }]),
|
||||
);
|
||||
bytes[bytes.length - 8] = (bytes[bytes.length - 8] ?? 0) ^ 1;
|
||||
const reader = new CpioArtifactReader(async () => Readable.from([bytes]));
|
||||
try {
|
||||
await expect(
|
||||
(async () => {
|
||||
for await (const entry of reader.entries()) void entry;
|
||||
})(),
|
||||
).rejects.toMatchObject({ reason: "format" });
|
||||
} finally {
|
||||
await reader.close();
|
||||
}
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,17 +1,13 @@
|
|||
import { createHash, type Hash } from "node:crypto";
|
||||
import { open, type FileHandle } from "node:fs/promises";
|
||||
import {
|
||||
PassThrough,
|
||||
Readable,
|
||||
Transform,
|
||||
type TransformCallback,
|
||||
} from "node:stream";
|
||||
import { Readable, Transform, type TransformCallback } from "node:stream";
|
||||
import { createInflate, inflateSync } from "node:zlib";
|
||||
|
||||
import { DOMParser, type Element, type Node } from "@xmldom/xmldom";
|
||||
|
||||
import type { ArtifactCommand } from "../domain/artifactGraph.js";
|
||||
import { ArtifactDecodedBudget } from "./ArtifactDecodedBudget.js";
|
||||
import { artifactStreamPipeline } from "./ArtifactStreamPipeline.js";
|
||||
import {
|
||||
xarInteger as integer,
|
||||
xarHeapPosition,
|
||||
|
|
@ -29,18 +25,18 @@ const XAR_MAGIC = 0x78617221;
|
|||
const MAX_TOC_BYTES = 64 * 1024 * 1024;
|
||||
const READ_CHUNK_BYTES = 1024 * 1024;
|
||||
|
||||
const CHECKSUM_ALGORITHMS: Readonly<Record<string, string>> = {
|
||||
sha1: "sha1",
|
||||
md5: "md5",
|
||||
sha256: "sha256",
|
||||
sha512: "sha512",
|
||||
};
|
||||
const DIGEST_BYTES: Readonly<Record<string, number>> = {
|
||||
sha1: 20,
|
||||
md5: 16,
|
||||
sha256: 32,
|
||||
sha512: 64,
|
||||
};
|
||||
const CHECKSUM_ALGORITHMS: ReadonlyMap<string, string> = new Map([
|
||||
["sha1", "sha1"],
|
||||
["md5", "md5"],
|
||||
["sha256", "sha256"],
|
||||
["sha512", "sha512"],
|
||||
]);
|
||||
const DIGEST_BYTES: ReadonlyMap<string, number> = new Map([
|
||||
["sha1", 20],
|
||||
["md5", 16],
|
||||
["sha256", 32],
|
||||
["sha512", 64],
|
||||
]);
|
||||
|
||||
const cancelled = (signal?: AbortSignal): void => {
|
||||
if (signal?.aborted === true)
|
||||
|
|
@ -102,7 +98,7 @@ const parseChecksum = (
|
|||
if (element === undefined) return undefined;
|
||||
const style = element.getAttribute("style")?.toLowerCase() ?? "";
|
||||
const value = (element.textContent ?? "").trim().toLowerCase();
|
||||
if (CHECKSUM_ALGORITHMS[style] === undefined)
|
||||
if (CHECKSUM_ALGORITHMS.get(style) === undefined)
|
||||
return { unsupported: style || "(missing style)", value };
|
||||
return { algorithm: style, value };
|
||||
};
|
||||
|
|
@ -529,7 +525,7 @@ export class XarArtifactReader implements ArtifactReader {
|
|||
"format",
|
||||
`xar TOC checksum style ${style || "(missing)"} does not match header algorithm ${expectedStyle ?? "none"}`,
|
||||
);
|
||||
const algorithm = CHECKSUM_ALGORITHMS[style];
|
||||
const algorithm = CHECKSUM_ALGORITHMS.get(style);
|
||||
if (algorithm === undefined)
|
||||
throw new ArtifactReaderFailure(
|
||||
"format",
|
||||
|
|
@ -537,7 +533,7 @@ export class XarArtifactReader implements ArtifactReader {
|
|||
);
|
||||
const size = integer(textOf(checksum, "size"), "checksum size");
|
||||
// The stored checksum is exactly one digest; never allocate a declared size.
|
||||
if (size !== DIGEST_BYTES[style])
|
||||
if (size !== DIGEST_BYTES.get(style))
|
||||
throw new ArtifactReaderFailure(
|
||||
"format",
|
||||
`xar TOC declares a ${size}-byte ${style} checksum`,
|
||||
|
|
@ -748,58 +744,21 @@ export class XarArtifactReader implements ArtifactReader {
|
|||
this.#integrity.set(entry.adapterKey, [observation]);
|
||||
},
|
||||
);
|
||||
const raw = Readable.from(this.#chunks(data.offset, data.length, signal));
|
||||
raw.on("error", (cause: unknown) =>
|
||||
archivedVerifier?.destroy(memberFailure(entry.path, cause)),
|
||||
);
|
||||
const archivedChecked =
|
||||
archivedVerifier === undefined ? raw : raw.pipe(archivedVerifier);
|
||||
let decoded: Readable;
|
||||
if (data.encoding === "application/octet-stream") decoded = archivedChecked;
|
||||
else {
|
||||
// xar's "x-gzip" members are zlib streams; bound decoded output to the
|
||||
// declared extracted size so a gzip bomb cannot exhaust the process.
|
||||
const inflate = createInflate();
|
||||
raw.on("error", (cause: unknown) =>
|
||||
inflate.destroy(memberFailure(entry.path, cause)),
|
||||
const stages: Transform[] =
|
||||
archivedVerifier === undefined ? [] : [archivedVerifier];
|
||||
if (data.encoding === "application/x-gzip")
|
||||
stages.push(
|
||||
createInflate(),
|
||||
new SizeBound(entry.path, data.size, this.decodedBudget, signal),
|
||||
);
|
||||
// Checksum failures emit on archivedChecked, which pipe() does not
|
||||
// forward; without this the integrity error becomes unhandled instead
|
||||
// of rolling back extraction with the tagged failure.
|
||||
archivedChecked.on("error", (cause: unknown) =>
|
||||
inflate.destroy(memberFailure(entry.path, cause)),
|
||||
);
|
||||
const inflated = archivedChecked.pipe(inflate);
|
||||
const bounded = new SizeBound(
|
||||
entry.path,
|
||||
data.size,
|
||||
this.decodedBudget,
|
||||
signal,
|
||||
);
|
||||
inflated.on("error", (cause: unknown) =>
|
||||
bounded.destroy(memberFailure(entry.path, cause)),
|
||||
);
|
||||
decoded = inflated.pipe(bounded);
|
||||
}
|
||||
const checksum = data.extractedChecksum;
|
||||
const verified =
|
||||
checksum === undefined || !this.#verifyChecksums
|
||||
? new PassThrough()
|
||||
: new ChecksumVerifier(entry.path, checksum);
|
||||
// Decoder errors become tagged failures; cancellation stays tagged.
|
||||
decoded.on("error", (cause: unknown) =>
|
||||
verified.destroy(memberFailure(entry.path, cause)),
|
||||
if (checksum !== undefined && this.#verifyChecksums)
|
||||
stages.push(new ChecksumVerifier(entry.path, checksum));
|
||||
return artifactStreamPipeline(
|
||||
Readable.from(this.#chunks(data.offset, data.length, signal)),
|
||||
stages,
|
||||
{ path: entry.path, signal },
|
||||
);
|
||||
const out = decoded.pipe(verified);
|
||||
// Tear down the source chain when the consumer is done: otherwise a
|
||||
// consumer that stops early (or a checksum failure at flush) leaves the
|
||||
// generator suspended, and its next read after close() surfaces as an
|
||||
// unhandled "archive is closed" failure.
|
||||
out.on("close", () => {
|
||||
raw.destroy();
|
||||
archivedVerifier?.destroy();
|
||||
});
|
||||
return out;
|
||||
}
|
||||
|
||||
async *#chunks(
|
||||
|
|
@ -832,13 +791,3 @@ export class XarArtifactReader implements ArtifactReader {
|
|||
await handle?.close();
|
||||
}
|
||||
}
|
||||
|
||||
/** Keep tagged failures; any other decoder error means malformed member bytes. */
|
||||
const memberFailure = (path: string, cause: unknown): ArtifactReaderFailure =>
|
||||
cause instanceof ArtifactReaderFailure
|
||||
? cause
|
||||
: new ArtifactReaderFailure(
|
||||
"format",
|
||||
`xar member ${path} could not be decoded`,
|
||||
{ cause },
|
||||
);
|
||||
|
|
|
|||
|
|
@ -2,12 +2,9 @@ import { lstat, realpath } from "node:fs/promises";
|
|||
|
||||
import { canonicalDigest } from "../../domain/comparisonSemantics.js";
|
||||
import { AsarArtifactReader } from "../AsarArtifactReader.js";
|
||||
import { CpioArtifactReader } from "../CpioArtifactReader.js";
|
||||
import { XarArtifactReader } from "../XarArtifactReader.js";
|
||||
import {
|
||||
ArtifactPathRegistry,
|
||||
normalizeArtifactPath,
|
||||
} from "../ArtifactPaths.js";
|
||||
import { ArtifactPathRegistry } from "../ArtifactPaths.js";
|
||||
import { visitArtifactTree } from "../ArtifactTraversal.js";
|
||||
import {
|
||||
ArtifactReaderFailure,
|
||||
type ArtifactEntry,
|
||||
|
|
@ -131,91 +128,55 @@ const materializeSelection = async ({
|
|||
const output = await SafeOutputTree.create(input.outputRoot);
|
||||
let readerClosed = false;
|
||||
const extracted: ExtractedOccurrence[] = [];
|
||||
const stack: Array<{
|
||||
readonly reader: ArtifactReader;
|
||||
readonly prefix: string;
|
||||
readonly iterator: AsyncIterator<ArtifactEntry>;
|
||||
readonly owned: boolean;
|
||||
}> = [
|
||||
{
|
||||
reader,
|
||||
prefix: "",
|
||||
iterator: reader.entries(signal)[Symbol.asyncIterator](),
|
||||
owned: false,
|
||||
},
|
||||
];
|
||||
try {
|
||||
const materialized: SelectedOccurrence[] = [];
|
||||
const registry = new ArtifactPathRegistry();
|
||||
while (stack.length > 0) {
|
||||
const frame = stack[stack.length - 1];
|
||||
if (frame === undefined) break;
|
||||
const next = await frame.iterator.next();
|
||||
if (next.done === true) {
|
||||
stack.pop();
|
||||
if (frame.owned) await frame.reader.close();
|
||||
continue;
|
||||
}
|
||||
const entry = next.value;
|
||||
const path = normalizeArtifactPath(
|
||||
frame.prefix === "" ? entry.path : `${frame.prefix}/${entry.path}`,
|
||||
);
|
||||
// Traverse expanded PKG Payload/Scripts like inventory does, so nested
|
||||
// members are materialized instead of silently skipped. An expanded
|
||||
// member acts as a container: register it as a directory and leave its
|
||||
// archive bytes to its children, since a file and a directory cannot
|
||||
// share one output path.
|
||||
const expandable =
|
||||
entry.nestedArchive === "gzip-cpio" &&
|
||||
!entry.encrypted &&
|
||||
entry.contentUnavailable !== true;
|
||||
registry.add(path, expandable ? "directory" : entry.kind);
|
||||
if (expandable) {
|
||||
const parent = frame.reader;
|
||||
const nested = new CpioArtifactReader(
|
||||
(nestedSignal) => parent.open(entry, nestedSignal),
|
||||
"fail",
|
||||
parent.decodedBudget,
|
||||
await visitArtifactTree(
|
||||
reader,
|
||||
async ({ reader: currentReader, entry, path, container: expandable }) => {
|
||||
// Traverse expanded PKG Payload/Scripts like inventory does, so nested
|
||||
// members are materialized instead of silently skipped. An expanded
|
||||
// member acts as a container: register it as a directory and leave its
|
||||
// archive bytes to its children, since a file and a directory cannot
|
||||
// share one output path.
|
||||
registry.add(path, expandable ? "directory" : entry.kind);
|
||||
if (expandable) {
|
||||
return true;
|
||||
}
|
||||
const selectedItem = byPath.get(path);
|
||||
if (selectedItem === undefined) {
|
||||
if (
|
||||
entry.contentUnavailable === true &&
|
||||
inventory.unavailablePaths.includes(path)
|
||||
)
|
||||
return false;
|
||||
if (entry.kind === "file" || entry.kind === "slice")
|
||||
throw new ArtifactReaderFailure(
|
||||
"integrity",
|
||||
`Regular artifact entry is missing from inventory: ${path}`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
preflight(entry);
|
||||
const stream = await currentReader.open(entry, signal);
|
||||
const written = await output.write(
|
||||
path,
|
||||
stream,
|
||||
selectedItem.node.sha256,
|
||||
signal,
|
||||
);
|
||||
stack.push({
|
||||
reader: nested,
|
||||
prefix: path,
|
||||
iterator: nested.entries(signal)[Symbol.asyncIterator](),
|
||||
owned: true,
|
||||
extracted.push({
|
||||
artifact_id: selectedItem.node.artifact_id,
|
||||
relative_path: written.relativePath,
|
||||
sha256: written.sha256,
|
||||
bytes_written: written.bytesWritten,
|
||||
created: true,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const selectedItem = byPath.get(path);
|
||||
if (selectedItem === undefined) {
|
||||
if (
|
||||
entry.contentUnavailable === true &&
|
||||
inventory.unavailablePaths.includes(path)
|
||||
)
|
||||
continue;
|
||||
if (entry.kind === "file" || entry.kind === "slice")
|
||||
throw new ArtifactReaderFailure(
|
||||
"integrity",
|
||||
`Regular artifact entry is missing from inventory: ${path}`,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
preflight(entry);
|
||||
const stream = await frame.reader.open(entry, signal);
|
||||
const written = await output.write(
|
||||
path,
|
||||
stream,
|
||||
selectedItem.node.sha256,
|
||||
signal,
|
||||
);
|
||||
extracted.push({
|
||||
artifact_id: selectedItem.node.artifact_id,
|
||||
relative_path: written.relativePath,
|
||||
sha256: written.sha256,
|
||||
bytes_written: written.bytesWritten,
|
||||
created: true,
|
||||
});
|
||||
materialized.push(selectedItem);
|
||||
}
|
||||
materialized.push(selectedItem);
|
||||
return false;
|
||||
},
|
||||
{ signal, expandAsar: false },
|
||||
);
|
||||
await reader.close();
|
||||
readerClosed = true;
|
||||
extracted.sort((left, right) =>
|
||||
|
|
@ -230,14 +191,6 @@ const materializeSelection = async ({
|
|||
await output.commit();
|
||||
return result;
|
||||
} catch (cause: unknown) {
|
||||
await Promise.allSettled(
|
||||
stack
|
||||
.filter(({ owned }) => owned)
|
||||
.map(async ({ reader, iterator }) => {
|
||||
await iterator.return?.();
|
||||
await reader.close();
|
||||
}),
|
||||
);
|
||||
if (!readerClosed)
|
||||
await reader.close().catch((cause: unknown) => {
|
||||
// best-effort cleanup: reader close must not mask the extraction failure.
|
||||
|
|
|
|||
|
|
@ -1,16 +1,10 @@
|
|||
import { isAbsolute } from "node:path";
|
||||
|
||||
import {
|
||||
ArtifactPathRegistry,
|
||||
normalizeArtifactPath,
|
||||
} from "../ArtifactPaths.js";
|
||||
import { ArtifactPathRegistry } from "../ArtifactPaths.js";
|
||||
import { visitArtifactTree } from "../ArtifactTraversal.js";
|
||||
import {
|
||||
ArtifactReaderFailure,
|
||||
type ArtifactEntry,
|
||||
type ArtifactReader,
|
||||
} from "../ArtifactReader.js";
|
||||
import { AsarArtifactReader } from "../AsarArtifactReader.js";
|
||||
import { CpioArtifactReader } from "../CpioArtifactReader.js";
|
||||
import type { ArtifactNode } from "../../domain/artifactGraph.js";
|
||||
import {
|
||||
classifyArtifactContent,
|
||||
|
|
@ -77,7 +71,7 @@ export const scanReader = async (
|
|||
registry: new ArtifactPathRegistry(),
|
||||
expandedContainerIds: new Set<string>(),
|
||||
};
|
||||
await visitArtifactEntries(context, reader, "");
|
||||
await visitArtifactEntries(context, reader);
|
||||
// Archive directories may appear after their children. Resolve containment
|
||||
// against the complete index before directory identities are materialized.
|
||||
for (const occurrence of occurrences)
|
||||
|
|
@ -93,47 +87,17 @@ export const scanReader = async (
|
|||
const visitArtifactEntries = async (
|
||||
context: ScanContext,
|
||||
currentReader: ArtifactReader,
|
||||
prefix: string,
|
||||
): Promise<void> => {
|
||||
const stack: Array<{
|
||||
readonly reader: ArtifactReader;
|
||||
readonly prefix: string;
|
||||
readonly iterator: AsyncIterator<ArtifactEntry>;
|
||||
readonly owned: boolean;
|
||||
}> = [
|
||||
{
|
||||
reader: currentReader,
|
||||
prefix,
|
||||
iterator: currentReader.entries(context.signal)[Symbol.asyncIterator](),
|
||||
owned: false,
|
||||
},
|
||||
];
|
||||
try {
|
||||
while (stack.length > 0) {
|
||||
const frame = stack.at(-1);
|
||||
if (frame === undefined) break;
|
||||
const next = await frame.iterator.next();
|
||||
if (next.done) {
|
||||
stack.pop();
|
||||
if (frame.owned) await frame.reader.close();
|
||||
continue;
|
||||
}
|
||||
const entry = next.value;
|
||||
const logicalPath = normalizeArtifactPath(
|
||||
frame.prefix.length === 0
|
||||
? entry.path
|
||||
: `${frame.prefix}/${entry.path}`,
|
||||
);
|
||||
const expandable =
|
||||
isExpandableAsar(entry, logicalPath) ||
|
||||
entry.nestedArchive === "gzip-cpio";
|
||||
await visitArtifactTree(
|
||||
currentReader,
|
||||
async ({ reader, entry, path: logicalPath, container: expandable }) => {
|
||||
context.registry.add(logicalPath, expandable ? "directory" : entry.kind);
|
||||
const occurrence = createOccurrence(entry, logicalPath, null);
|
||||
let digested: DigestedEntry | undefined;
|
||||
try {
|
||||
digested = await digestArtifactEntry(
|
||||
context,
|
||||
frame.reader,
|
||||
reader,
|
||||
entry,
|
||||
logicalPath,
|
||||
);
|
||||
|
|
@ -156,7 +120,7 @@ const visitArtifactEntries = async (
|
|||
);
|
||||
context.occurrences.push(occurrence);
|
||||
context.occurrenceByPath.set(logicalPath, occurrence);
|
||||
continue;
|
||||
return false;
|
||||
}
|
||||
occurrence.hash_status = "unavailable";
|
||||
occurrence.limitations.push(UNAVAILABLE_UNPACKED_LIMITATION);
|
||||
|
|
@ -176,35 +140,14 @@ const visitArtifactEntries = async (
|
|||
context.occurrences.push(occurrence);
|
||||
context.occurrenceByPath.set(logicalPath, occurrence);
|
||||
if (expandable && digested?.mismatched !== true) {
|
||||
const parent = frame.reader;
|
||||
const nested: ArtifactReader =
|
||||
entry.nestedArchive === "gzip-cpio"
|
||||
? new CpioArtifactReader(
|
||||
(signal) => parent.open(entry, signal),
|
||||
context.integrity.mode,
|
||||
parent.decodedBudget,
|
||||
)
|
||||
: new AsarArtifactReader(entry.adapterKey);
|
||||
// Only traversed containers can own members, not opaque archive-named files.
|
||||
context.expandedContainerIds.add(occurrence.occurrence_id);
|
||||
stack.push({
|
||||
reader: nested,
|
||||
prefix: logicalPath,
|
||||
iterator: nested.entries(context.signal)[Symbol.asyncIterator](),
|
||||
owned: true,
|
||||
});
|
||||
return true;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await Promise.allSettled(
|
||||
stack
|
||||
.filter(({ owned }) => owned)
|
||||
.map(async ({ reader, iterator }) => {
|
||||
await iterator.return?.();
|
||||
await reader.close();
|
||||
}),
|
||||
);
|
||||
}
|
||||
return false;
|
||||
},
|
||||
{ signal: context.signal, integrity: context.integrity.mode },
|
||||
);
|
||||
};
|
||||
|
||||
interface DigestedEntry {
|
||||
|
|
@ -296,11 +239,6 @@ const digestArtifactEntry = async (
|
|||
};
|
||||
};
|
||||
|
||||
const isExpandableAsar = (entry: ArtifactEntry, logicalPath: string): boolean =>
|
||||
entry.kind === "file" &&
|
||||
logicalPath.toLowerCase().endsWith(".asar") &&
|
||||
isAbsolute(entry.adapterKey);
|
||||
|
||||
const isUnavailableUnpackedEntry = (
|
||||
cause: unknown,
|
||||
entry: ArtifactEntry,
|
||||
|
|
|
|||
Loading…
Reference in New Issue