diff --git a/packages/cli/src/server/studioServer.ts b/packages/cli/src/server/studioServer.ts index a7316ea31..8d6ebf1e7 100644 --- a/packages/cli/src/server/studioServer.ts +++ b/packages/cli/src/server/studioServer.ts @@ -399,7 +399,8 @@ export function createStudioServer(options: StudioServerOptions): StudioServer { await import("../../../producer/src/services/deterministicFonts.js"); const { prepareAnimatedGifInputs } = await import("../../../producer/src/services/animatedGifPrep.js"); - const { downloadToTemp } = await import("../../../producer/src/utils/urlDownloader.js"); + const { downloadToTemp, writeUrlDownloadTelemetry } = + await import("../../../producer/src/utils/urlDownloader.js"); const gifOutputDir = join(project.dir, ".hyperframes", "prepared-assets", "gif"); const gifDownloadDir = join(project.dir, ".hyperframes", "prepared-assets", "downloads"); const prepared = await prepareAnimatedGifInputs(html, { @@ -408,7 +409,11 @@ export function createStudioServer(options: StudioServerOptions): StudioServer { outputDir: gifOutputDir, outputSrcPrefix: ".hyperframes/prepared-assets/gif", cacheDir: gifOutputDir, - sourceAssets: await downloadRemoteGifImageSources(html, gifDownloadDir, downloadToTemp), + sourceAssets: await downloadRemoteGifImageSources(html, gifDownloadDir, (url, destDir) => + downloadToTemp(url, destDir, undefined, undefined, undefined, { + onTelemetry: writeUrlDownloadTelemetry, + }), + ), }); return injectDeterministicFontFaces(prepared.html); }, diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 7a1630659..0e9a7209c 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -273,7 +273,18 @@ export { type KeyframeAnalysis, } from "./utils/ffprobe.js"; -export { assertPublicHttpsUrl, downloadToTemp, isHttpUrl } from "./utils/urlDownloader.js"; +export { + assertPublicHttpsUrl, + downloadToTemp, + fetchPublicHttpsText, + isHttpUrl, + safeDownloadUrlIdentity, + writeUrlDownloadTelemetry, + type SafeDownloadUrlIdentity, + type UrlDownloadOptions, + type UrlDownloadTelemetry, + type PublicHttpsTextOptions, +} from "./utils/urlDownloader.js"; export { runFfmpeg, formatFfmpegError, diff --git a/packages/engine/src/services/audioMixer.test.ts b/packages/engine/src/services/audioMixer.test.ts index bc49a149e..36531822d 100644 --- a/packages/engine/src/services/audioMixer.test.ts +++ b/packages/engine/src/services/audioMixer.test.ts @@ -52,6 +52,7 @@ describe("processCompositionAudio", () => { const tempDirs: string[] = []; afterEach(() => { + vi.unstubAllGlobals(); runFfmpegMock.mockClear(); extractAudioMetadataMock.mockReset(); extractAudioMetadataMock.mockResolvedValue({ @@ -66,6 +67,44 @@ describe("processCompositionAudio", () => { } }); + it("classifies an HTML-as-200 audio source as deterministic user input", async () => { + const baseDir = mkdtempSync(join(tmpdir(), "hf-audio-base-")); + const workDir = mkdtempSync(join(tmpdir(), "hf-audio-work-")); + tempDirs.push(baseDir, workDir); + const fetchMock = vi + .fn() + .mockResolvedValue(new Response("denied")); + vi.stubGlobal("fetch", fetchMock); + + const result = await processCompositionAudio( + [ + { + id: "remote-voice", + src: "https://cdn.example/voice", + start: 0, + end: 2, + mediaStart: 0, + layer: 0, + volume: 1, + type: "audio", + }, + ], + baseDir, + workDir, + join(baseDir, "out.m4a"), + 2, + ); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(result.failures).toEqual([ + expect.objectContaining({ + stage: "download", + owner: "user", + retryable: false, + }), + ]); + }); + it.each([ { message: "AbortError: ffprobe operation aborted", diff --git a/packages/engine/src/services/audioMixer.ts b/packages/engine/src/services/audioMixer.ts index 1b063f3fd..a3b3ad060 100644 --- a/packages/engine/src/services/audioMixer.ts +++ b/packages/engine/src/services/audioMixer.ts @@ -9,7 +9,12 @@ import { closeSync, existsSync, mkdirSync, mkdtempSync, openSync, rmSync, writeF import { join, dirname } from "path"; import { parseHTML } from "linkedom"; import { extractAudioMetadata } from "../utils/ffprobe.js"; -import { downloadToTemp, isHttpUrl } from "../utils/urlDownloader.js"; +import { + downloadToTemp, + isHttpUrl, + UrlDownloadError, + writeUrlDownloadTelemetry, +} from "../utils/urlDownloader.js"; import { DEFAULT_CONFIG, type EngineConfig } from "../config.js"; import { formatFfmpegError, runFfmpeg, type RunFfmpegResult } from "../utils/runFfmpeg.js"; import { unwrapTemplate } from "../utils/htmlTemplate.js"; @@ -241,16 +246,23 @@ function probeFailure(message: string, elementId: string): AudioProcessingFailur }; } -function downloadFailure(message: string, elementId: string): AudioProcessingFailure { +function downloadFailure(error: unknown, elementId: string): AudioProcessingFailure { + const message = error instanceof Error ? error.message : String(error); const invalidSource = - /(?:invalid URL|only HTTPS|private\/reserved|HTTP (?:400|401|403|404|405|410|422)\b)/i.test( - message, - ); + error instanceof UrlDownloadError + ? error.kind === "http_not_found" || + error.kind === "http_rejected" || + error.kind === "invalid_payload" || + error.kind === "cancelled" + : /(?:invalid URL|only HTTPS|private\/reserved|HTTP (?:400|401|403|404|405|410|422)\b)/i.test( + message, + ); + const retryable = error instanceof UrlDownloadError ? error.retryable : !invalidSource; return { stage: "download", reason: "download_failed", owner: invalidSource ? "user" : "system", - retryable: !invalidSource, + retryable, elementId, detail: boundedDetail(`Download failed for audio element ${elementId}: ${message}`), }; @@ -712,11 +724,11 @@ export async function processCompositionAudio( if (isHttpUrl(srcPath)) { try { - srcPath = await downloadToTemp(srcPath, workDir); + srcPath = await downloadToTemp(srcPath, workDir, undefined, signal, undefined, { + onTelemetry: writeUrlDownloadTelemetry, + }); } catch (err: unknown) { - failures.push( - downloadFailure(err instanceof Error ? err.message : String(err), element.id), - ); + failures.push(downloadFailure(err, element.id)); return; } } diff --git a/packages/engine/src/services/videoFrameExtractor.errorClassification.test.ts b/packages/engine/src/services/videoFrameExtractor.errorClassification.test.ts index 6afa2f8fa..6873c484a 100644 --- a/packages/engine/src/services/videoFrameExtractor.errorClassification.test.ts +++ b/packages/engine/src/services/videoFrameExtractor.errorClassification.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; -import { classifyFfmpegSpawnError } from "./videoFrameExtractor.js"; +import { UrlDownloadError } from "../utils/urlDownloader.js"; +import { classifyFfmpegSpawnError, classifyVideoExtractionError } from "./videoFrameExtractor.js"; describe("classifyFfmpegSpawnError", () => { it.each(["ENOENT", "EACCES", "ENOEXEC", "UNKNOWN"])( @@ -18,3 +19,21 @@ describe("classifyFfmpegSpawnError", () => { }); }); }); + +describe("classifyVideoExtractionError download integrity", () => { + it("keeps deterministic HTML payloads non-retryable and user-owned as invalid media", () => { + const classified = classifyVideoExtractionError( + new UrlDownloadError("invalid_payload", false, "HTML payload"), + ); + expect(classified).toMatchObject({ kind: "invalid_media", retryable: false }); + }); + + it.each(["range_protocol", "length_mismatch", "hash_mismatch"] as const)( + "keeps %s retryable after the downloader's one clean refetch is exhausted", + (kind) => { + expect( + classifyVideoExtractionError(new UrlDownloadError(kind, true, "integrity failure")), + ).toMatchObject({ kind: "download_transient", retryable: true }); + }, + ); +}); diff --git a/packages/engine/src/services/videoFrameExtractor.ts b/packages/engine/src/services/videoFrameExtractor.ts index 501485aa1..1ecc62d4e 100644 --- a/packages/engine/src/services/videoFrameExtractor.ts +++ b/packages/engine/src/services/videoFrameExtractor.ts @@ -28,7 +28,12 @@ import { isHdrColorSpace as isHdrColorSpaceUtil, type HdrTransfer, } from "../utils/hdr.js"; -import { downloadToTemp, isHttpUrl, UrlDownloadError } from "../utils/urlDownloader.js"; +import { + downloadToTemp, + isHttpUrl, + UrlDownloadError, + writeUrlDownloadTelemetry, +} from "../utils/urlDownloader.js"; import { runFfmpeg } from "../utils/runFfmpeg.js"; import { DEFAULT_CONFIG, type EngineConfig } from "../config.js"; import { unwrapTemplate } from "../utils/htmlTemplate.js"; @@ -342,6 +347,14 @@ export function classifyVideoExtractionError(error: unknown): VideoSourceExtract diagnostic, ); } + if (error.kind === "invalid_payload") { + return new VideoSourceExtractionError( + "invalid_media", + false, + "Video source download returned a non-media payload", + diagnostic, + ); + } if (error.retryable) { return new VideoSourceExtractionError( "download_transient", @@ -1424,8 +1437,13 @@ export async function extractAllVideoFrames( if (isHttpUrl(videoPath)) { const downloadDir = join(options.outputDir, "_downloads"); mkdirSync(downloadDir, { recursive: true }); - videoPath = await downloadToTemp(videoPath, downloadDir, undefined, signal, () => - recordTransientRetries(1), + videoPath = await downloadToTemp( + videoPath, + downloadDir, + undefined, + signal, + () => recordTransientRetries(1), + { onTelemetry: writeUrlDownloadTelemetry }, ); } diff --git a/packages/engine/src/utils/urlDownloader.test.ts b/packages/engine/src/utils/urlDownloader.test.ts index 8ef85317a..908765c18 100644 --- a/packages/engine/src/utils/urlDownloader.test.ts +++ b/packages/engine/src/utils/urlDownloader.test.ts @@ -2,16 +2,89 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { existsSync, + mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, symlinkSync, + utimesSync, writeFileSync, } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { assertPublicHttpsUrl, downloadToTemp, UrlDownloadError } from "./urlDownloader.js"; +import { createHash } from "node:crypto"; +import { + assertPublicHttpsUrl, + downloadToTemp, + fetchPublicHttpsText, + UrlDownloadError, +} from "./urlDownloader.js"; + +const fsRaceControls = vi.hoisted(() => ({ + deleteBeforeLstatPath: undefined as string | undefined, + deleteInjectedWinnerBeforeLstatPath: undefined as string | undefined, + injectRaceAtLinkPath: undefined as string | undefined, + deleteBeforeReadPath: undefined as string | undefined, + replaceStaleLockAfterObservationPath: undefined as string | undefined, + replaceLockOnReleasePath: undefined as string | undefined, +})); + +vi.mock("fs", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + lstatSync: ((...args: unknown[]) => { + const path = String(args[0]); + if (path === fsRaceControls.deleteBeforeLstatPath) { + fsRaceControls.deleteBeforeLstatPath = undefined; + actual.rmSync(path, { force: true }); + } + if (path === fsRaceControls.deleteInjectedWinnerBeforeLstatPath && actual.existsSync(path)) { + fsRaceControls.deleteInjectedWinnerBeforeLstatPath = undefined; + actual.rmSync(path, { force: true }); + } + return Reflect.apply(actual.lstatSync, actual, args); + }) as typeof actual.lstatSync, + readdirSync: ((...args: unknown[]) => { + const path = String(args[0]); + const observed = Reflect.apply(actual.readdirSync, actual, args); + if (path === fsRaceControls.replaceStaleLockAfterObservationPath) { + fsRaceControls.replaceStaleLockAfterObservationPath = undefined; + actual.rmSync(path, { recursive: true, force: true }); + actual.mkdirSync(path); + actual.mkdirSync(join(path, ".hf-owner-successor")); + } + return observed; + }) as typeof actual.readdirSync, + rmdirSync: ((...args: unknown[]) => { + const path = String(args[0]); + if (path === fsRaceControls.replaceLockOnReleasePath) { + fsRaceControls.replaceLockOnReleasePath = undefined; + actual.rmSync(path, { recursive: true, force: true }); + actual.mkdirSync(path); + actual.mkdirSync(join(path, ".hf-owner-successor")); + } + return Reflect.apply(actual.rmdirSync, actual, args); + }) as typeof actual.rmdirSync, + linkSync: ((...args: unknown[]) => { + const destination = String(args[1]); + if (destination === fsRaceControls.injectRaceAtLinkPath) { + fsRaceControls.injectRaceAtLinkPath = undefined; + actual.writeFileSync(destination, "concurrent-winner"); + } + return Reflect.apply(actual.linkSync, actual, args); + }) as typeof actual.linkSync, + createReadStream: ((...args: unknown[]) => { + const path = String(args[0]); + if (path === fsRaceControls.deleteBeforeReadPath) { + fsRaceControls.deleteBeforeReadPath = undefined; + actual.rmSync(path, { force: true }); + } + return Reflect.apply(actual.createReadStream, actual, args); + }) as typeof actual.createReadStream, + }; +}); const tempDirs: string[] = []; @@ -23,12 +96,30 @@ function makeTempDir(): string { function temporaryDownloadEntries(dir: string): string[] { return readdirSync(dir).filter( - (name) => name.includes(".partial-") || name.startsWith(".hf-download-"), + (name) => + name.includes(".partial-") || name.startsWith(".hf-download-") || name.endsWith(".hf-lock"), ); } +function isoBmffMediaBytes(marker: string): Buffer { + const ftyp = Buffer.alloc(24); + ftyp.writeUInt32BE(24, 0); + ftyp.write("ftyp", 4, 4, "ascii"); + ftyp.write("isom", 8, 4, "ascii"); + ftyp.writeUInt32BE(0, 12); + ftyp.write("isom", 16, 4, "ascii"); + ftyp.write("mp42", 20, 4, "ascii"); + return Buffer.concat([ftyp, Buffer.from(marker)]); +} + afterEach(() => { vi.unstubAllGlobals(); + fsRaceControls.deleteBeforeLstatPath = undefined; + fsRaceControls.deleteInjectedWinnerBeforeLstatPath = undefined; + fsRaceControls.injectRaceAtLinkPath = undefined; + fsRaceControls.deleteBeforeReadPath = undefined; + fsRaceControls.replaceStaleLockAfterObservationPath = undefined; + fsRaceControls.replaceLockOnReleasePath = undefined; for (const dir of tempDirs.splice(0)) { rmSync(dir, { recursive: true, force: true }); } @@ -120,6 +211,78 @@ describe("assertPublicHttpsUrl — SSRF guard", () => { expect(() => assertPublicHttpsUrl("not-a-url")).toThrow("Invalid URL"); expect(() => assertPublicHttpsUrl("")).toThrow("Invalid URL"); }); + + it("never echoes a rejected signed URL in diagnostics", () => { + const signed = "http://127.0.0.1/private/customer.mp4?X-Amz-Signature=super-secret"; + let message = ""; + try { + assertPublicHttpsUrl(signed); + } catch (error) { + message = error instanceof Error ? error.message : String(error); + } + expect(message).not.toContain("customer.mp4"); + expect(message).not.toContain("super-secret"); + }); +}); + +describe("fetchPublicHttpsText", () => { + it("validates every redirect before issuing the next request", async () => { + const fetchMock = vi.fn().mockResolvedValue( + new Response(null, { + status: 302, + headers: { location: "https://169.254.169.254/latest/meta-data/" }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + + await expect( + fetchPublicHttpsText("https://styles.example/fonts.css", { maxBytes: 1024 }), + ).rejects.toMatchObject({ kind: "http_rejected", retryable: false }); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("enforces the byte cap while consuming a chunked response", async () => { + let cancelled = false; + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("123456")); + controller.enqueue(new TextEncoder().encode("789012")); + }, + cancel() { + cancelled = true; + }, + }); + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(new Response(body))); + + await expect( + fetchPublicHttpsText("https://styles.example/large.css", { maxBytes: 10 }), + ).rejects.toMatchObject({ kind: "length_mismatch", retryable: false }); + expect(cancelled).toBe(true); + }); + + it("keeps the timeout active while the response body is stalled", async () => { + const fetchMock = vi.fn().mockImplementation(async (_url: string, init: RequestInit) => { + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("partial")); + init.signal?.addEventListener( + "abort", + () => controller.error(new DOMException("aborted", "AbortError")), + { once: true }, + ); + }, + }); + return new Response(body); + }); + vi.stubGlobal("fetch", fetchMock); + + await expect( + fetchPublicHttpsText("https://styles.example/stalled.css", { + maxBytes: 1024, + timeoutMs: 20, + }), + ).rejects.toMatchObject({ kind: "timeout", retryable: true }); + }); }); describe("downloadToTemp atomic publication and bounded retry", () => { @@ -141,7 +304,10 @@ describe("downloadToTemp atomic publication and bounded retry", () => { expect(fetchMock).toHaveBeenNthCalledWith( 2, "https://media.example/final.mp4", - expect.objectContaining({ redirect: "manual" }), + expect.objectContaining({ + redirect: "manual", + headers: { "accept-encoding": "identity" }, + }), ); expect(readFileSync(path, "utf8")).toBe("complete"); expect(temporaryDownloadEntries(dir)).toEqual([]); @@ -212,6 +378,598 @@ describe("downloadToTemp atomic publication and bounded retry", () => { expect(temporaryDownloadEntries(dir)).toEqual([]); }); + it("does not expose a signed URL supplied through hostile HTTP status text", async () => { + const signedUrl = + "https://cdn.example/private/customer-video?X-Amz-Signature=super-secret-signature"; + const fetchMock = vi + .fn() + .mockResolvedValue(new Response(null, { status: 503, statusText: signedUrl })); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + let message = ""; + try { + await downloadToTemp(signedUrl, dir, 1_000); + } catch (error) { + message = error instanceof Error ? error.message : String(error); + } + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(message).toBe("HTTP 503"); + expect(message).not.toContain("customer-video"); + expect(message).not.toContain("super-secret-signature"); + }); + + it("rejects a truncated Content-Length response and cleanly refetches once", async () => { + const complete = isoBmffMediaBytes("complete"); + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + new Response(complete.subarray(0, complete.length - 1), { + headers: { "content-length": String(complete.length) }, + }), + ) + .mockResolvedValueOnce( + new Response(complete, { headers: { "content-length": String(complete.length) } }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + const onTransientRetry = vi.fn(); + + const path = await downloadToTemp( + "https://cdn.example/truncated.mp4", + dir, + 1_000, + undefined, + onTransientRetry, + {}, + ); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(onTransientRetry).toHaveBeenCalledWith( + expect.objectContaining({ kind: "length_mismatch", retryable: true }), + ); + expect(readFileSync(path)).toEqual(complete); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("does not publish after two repeated length mismatches", async () => { + const fetchMock = vi + .fn() + .mockImplementation(() => + Promise.resolve(new Response("short", { headers: { "content-length": "12" } })), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/always-short.mp4", dir, 1_000), + ).rejects.toMatchObject({ + kind: "length_mismatch", + retryable: true, + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readdirSync(dir).filter((name) => name.startsWith("download_"))).toEqual([]); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("rejects an unsolicited well-formed 206 and succeeds after one clean refetch", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + new Response("part", { + status: 206, + headers: { "content-range": "bytes 0-3/8", "content-length": "4" }, + }), + ) + .mockResolvedValueOnce(new Response("complete", { headers: { "content-length": "8" } })); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/unsolicited-range.mp4", dir, 1_000); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("rejects malformed 206 responses after exactly one refetch", async () => { + const fetchMock = vi.fn().mockResolvedValue( + new Response("part", { + status: 206, + headers: { "content-range": "bytes nonsense" }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/malformed-range.mp4", dir, 1_000), + ).rejects.toMatchObject({ + kind: "range_protocol", + retryable: true, + telemetry: expect.objectContaining({ rangeDisposition: "malformed_206" }), + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("accepts a complete identity-encoded object with a noncompliant Content-Range on 200", async () => { + const onTelemetry = vi.fn(); + const fetchMock = vi.fn().mockResolvedValue( + new Response("complete", { + status: 200, + headers: { "content-range": "bytes 0-7/8", "content-length": "8" }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp( + "https://cdn.example/full-range-on-200.mp4", + dir, + 1_000, + undefined, + undefined, + { onTelemetry }, + ); + + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(onTelemetry).toHaveBeenCalledWith( + expect.objectContaining({ outcome: "published", rangeDisposition: "full_object_200" }), + ); + }); + + it("rejects Content-Range on a 200 response without a matching declared length", async () => { + const fetchMock = vi.fn().mockResolvedValue( + new Response("complete", { + status: 200, + headers: { "content-range": "bytes 0-7/8" }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/range-on-200.mp4", dir, 1_000), + ).rejects.toMatchObject({ + kind: "range_protocol", + retryable: true, + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it("rejects a full-object Content-Range when the streamed body is shorter than declared", async () => { + const fetchMock = vi.fn().mockImplementation(() => + Promise.resolve( + new Response("short", { + status: 200, + headers: { "content-range": "bytes 0-7/8", "content-length": "8" }, + }), + ), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/short-full-range-on-200.mp4", dir, 1_000), + ).rejects.toMatchObject({ + kind: "length_mismatch", + retryable: true, + telemetry: expect.objectContaining({ rangeDisposition: "full_object_200" }), + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it.each([ + ["partial object", "bytes 0-3/8", "4", undefined], + ["nonzero start", "bytes 1-7/8", "8", undefined], + ["wildcard total", "bytes 0-7/*", "8", undefined], + ["mismatched length", "bytes 0-7/8", "7", undefined], + ["unsafe integer", "bytes 0-9007199254740991/9007199254740992", "8", undefined], + ["encoded body", "bytes 0-7/8", "8", "gzip"], + ])("rejects a %s Content-Range on 200", async (_case, range, length, encoding) => { + const fetchMock = vi.fn().mockResolvedValue( + new Response("complete", { + status: 200, + headers: { + "content-range": range, + "content-length": length, + ...(encoding ? { "content-encoding": encoding } : {}), + }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/invalid-range-on-200.mp4", dir, 1_000), + ).rejects.toMatchObject({ + kind: "range_protocol", + retryable: true, + telemetry: expect.objectContaining({ rangeDisposition: "content_range_on_200" }), + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it("retries a checksum mismatch once and publishes only matching bytes", async () => { + const corrupt = isoBmffMediaBytes("corrupt"); + const complete = isoBmffMediaBytes("complete"); + const expectedSha256 = createHash("sha256").update(complete).digest("hex"); + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(corrupt)) + .mockResolvedValueOnce(new Response(complete)); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp( + "https://cdn.example/checksum.mp4", + dir, + 1_000, + undefined, + undefined, + { expectedSha256 }, + ); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path)).toEqual(complete); + }); + + it("rejects a malformed caller checksum before fetching", async () => { + const fetchMock = vi.fn(); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/checksum.mp4", dir, 1_000, undefined, undefined, { + expectedSha256: "not-a-sha256", + }), + ).rejects.toMatchObject({ kind: "hash_mismatch", retryable: false }); + expect(fetchMock).not.toHaveBeenCalled(); + expect(readdirSync(dir)).toEqual([]); + }); + + it("locally refetches a caller checksum mismatch but keeps the final error non-retryable", async () => { + const bytes = isoBmffMediaBytes("always-wrong"); + const expectedSha256 = createHash("sha256").update("different").digest("hex"); + const serverSha256 = createHash("sha256").update(bytes).digest("base64"); + const contentMd5 = createHash("md5").update(bytes).digest("base64"); + const fetchMock = vi.fn().mockImplementation(() => + Promise.resolve( + new Response(bytes, { + headers: { "x-amz-checksum-sha256": serverSha256, "content-md5": contentMd5 }, + }), + ), + ); + const onTransientRetry = vi.fn(); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp( + "https://cdn.example/caller-checksum.mp4", + dir, + 1_000, + undefined, + onTransientRetry, + { expectedSha256 }, + ), + ).rejects.toMatchObject({ + kind: "hash_mismatch", + retryable: false, + locallyRetryable: true, + }); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(onTransientRetry).not.toHaveBeenCalled(); + }); + + it("prefers a matching caller SHA-256 over contradictory server checksums", async () => { + const bytes = isoBmffMediaBytes("caller-authoritative"); + const callerSha256 = createHash("sha256").update(bytes).digest("hex"); + const fetchMock = vi.fn().mockResolvedValue( + new Response(bytes, { + headers: { + "x-amz-checksum-sha256": Buffer.from("wrong-sha256").toString("base64"), + "content-md5": Buffer.from("wrong-md5").toString("base64"), + }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp( + "https://cdn.example/caller-authoritative.mp4", + dir, + 1_000, + undefined, + undefined, + { expectedSha256: callerSha256 }, + ); + + expect(readFileSync(path)).toEqual(bytes); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("prefers a matching server SHA-256 over a contradictory Content-MD5", async () => { + const bytes = isoBmffMediaBytes("server-sha-authoritative"); + const serverSha256 = createHash("sha256").update(bytes).digest("base64"); + const fetchMock = vi.fn().mockResolvedValue( + new Response(bytes, { + headers: { + "x-amz-checksum-sha256": serverSha256, + "content-md5": Buffer.from("wrong-md5").toString("base64"), + }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/server-sha.mp4", dir, 1_000); + + expect(readFileSync(path)).toEqual(bytes); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("accepts a matching Content-MD5 when no SHA-256 is available", async () => { + const bytes = isoBmffMediaBytes("legacy-md5"); + const contentMd5 = createHash("md5").update(bytes).digest("base64"); + const fetchMock = vi + .fn() + .mockResolvedValue(new Response(bytes, { headers: { "content-md5": contentMd5 } })); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/legacy-md5.mp4", dir, 1_000); + + expect(readFileSync(path)).toEqual(bytes); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("keeps repeated Content-MD5 mismatches retryable upstream", async () => { + const bytes = isoBmffMediaBytes("legacy-md5-mismatch"); + const fetchMock = vi.fn().mockImplementation(() => + Promise.resolve( + new Response(bytes, { + headers: { "content-md5": Buffer.from("wrong-md5").toString("base64") }, + }), + ), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/legacy-md5-mismatch.mp4", dir, 1_000), + ).rejects.toMatchObject({ kind: "hash_mismatch", retryable: true }); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it("keeps repeated server checksum mismatches retryable upstream", async () => { + const bytes = isoBmffMediaBytes("server-corrupt"); + const fetchMock = vi.fn().mockImplementation(() => + Promise.resolve( + new Response(bytes, { + headers: { digest: `sha-256=${Buffer.from("wrong-checksum").toString("base64")}` }, + }), + ), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/server-checksum.mp4", dir, 1_000), + ).rejects.toMatchObject({ kind: "hash_mismatch", retryable: true }); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it("ignores S3 composite checksums that are not full-object SHA-256", async () => { + const mediaBytes = isoBmffMediaBytes("multipart-object"); + const fetchMock = vi.fn().mockResolvedValue( + new Response(mediaBytes, { + headers: { + "x-amz-checksum-sha256": Buffer.from("not-a-full-object-checksum").toString("base64"), + "x-amz-checksum-type": "COMPOSITE", + }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp( + "https://cdn.example/multipart.mp4", + dir, + 1_000, + undefined, + undefined, + {}, + ); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(readFileSync(path)).toEqual(mediaBytes); + }); + + it("rejects an HTML-as-200 media payload without retrying", async () => { + const fetchMock = vi + .fn() + .mockResolvedValue(new Response(" denied")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/html-error.mp4", dir, 1_000, undefined, undefined, {}), + ).rejects.toMatchObject({ + kind: "invalid_payload", + retryable: false, + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("rejects a BOM-prefixed remote error document without retrying", async () => { + const payload = Buffer.from("\uFEFF denied"); + const fetchMock = vi.fn().mockResolvedValue(new Response(payload)); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/bom-error.mp4", dir, 1_000), + ).rejects.toMatchObject({ kind: "invalid_payload", retryable: false }); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(readdirSync(dir).filter((name) => name.startsWith("download_"))).toEqual([]); + }); + + it("invalidates a cached HTML error document before localizing a font", async () => { + const url = "https://cdn.example/font.woff2"; + const dir = makeTempDir(); + const cacheName = `download_${createHash("md5").update(url).digest("hex").slice(0, 12)}.woff2`; + const cachePath = join(dir, cacheName); + writeFileSync(cachePath, "expired"); + const fontBytes = Buffer.from("wOF2valid-font-payload"); + const fetchMock = vi.fn().mockResolvedValue(new Response(fontBytes)); + vi.stubGlobal("fetch", fetchMock); + + const path = await downloadToTemp(url, dir, 1_000); + + expect(path).toBe(cachePath); + expect(readFileSync(path)).toEqual(fontBytes); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("rejects a JSON error document served as video without retrying", async () => { + const fetchMock = vi.fn().mockResolvedValue( + new Response('{"message":"access denied"}', { + headers: { "content-type": "video/mp4" }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp( + "https://cdn.example/wrong-signature.mp4", + dir, + 1_000, + undefined, + undefined, + {}, + ), + ).rejects.toMatchObject({ kind: "invalid_payload", retryable: false }); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(readdirSync(dir).filter((name) => name.startsWith("download_"))).toEqual([]); + }); + + it("accepts valid media bytes through a wrong-MIME redirect", async () => { + const mediaBytes = isoBmffMediaBytes("wrong-mime"); + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + new Response(null, { + status: 302, + headers: { location: "https://media.example/asset" }, + }), + ) + .mockResolvedValueOnce( + new Response(mediaBytes, { headers: { "content-type": "text/html" } }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp( + "https://cdn.example/wrong-mime", + dir, + 1_000, + undefined, + undefined, + {}, + ); + + expect(readFileSync(path)).toEqual(mediaBytes); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it("accepts extensionless media and emits safe complete integrity telemetry", async () => { + const mediaBytes = isoBmffMediaBytes("extensionless"); + const etag = '"customer-secret-etag"'; + const fetchMock = vi.fn().mockResolvedValue( + new Response(mediaBytes, { + headers: { "content-length": String(mediaBytes.length), etag }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + const events: unknown[] = []; + const signedUrl = + "https://cdn.example/private/customer-video?X-Amz-Signature=super-secret-signature"; + + const path = await downloadToTemp(signedUrl, dir, 1_000, undefined, undefined, { + onTelemetry: (event) => events.push(event), + }); + + expect(readFileSync(path)).toEqual(mediaBytes); + expect(events).toEqual([ + expect.objectContaining({ + initialHost: "cdn.example", + finalHost: "cdn.example", + attempt: 1, + outcome: "published", + status: 200, + expectedBytes: mediaBytes.length, + receivedBytes: mediaBytes.length, + localSize: mediaBytes.length, + localSha256: createHash("sha256").update(mediaBytes).digest("hex"), + etagFingerprint: createHash("sha256").update(etag).digest("hex"), + }), + ]); + const serialized = JSON.stringify(events); + expect(serialized).not.toContain("customer-video"); + expect(serialized).not.toContain("super-secret-signature"); + expect(serialized).not.toContain("customer-secret-etag"); + }); + + it.each([ + ["aiff", Buffer.from("FORM\0\0\0\0AIFF")], + ["caf", Buffer.from("caff\0\x01\0\0")], + ["amr", Buffer.from("#!AMR\n")], + ["flv", Buffer.from("FLV\x01\x05")], + ])( + "does not reject valid %s inputs via a duplicate format allowlist", + async (kind, mediaBytes) => { + const fetchMock = vi.fn().mockResolvedValue(new Response(mediaBytes)); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp(`https://cdn.example/extensionless-${kind}`, dir, 1_000); + + expect(readFileSync(path)).toEqual(mediaBytes); + expect(fetchMock).toHaveBeenCalledTimes(1); + }, + ); + + it("redacts signed URLs embedded in underlying fetch failures", async () => { + const signedUrl = + "https://cdn.example/private/customer-video?X-Amz-Signature=super-secret-signature"; + const fetchMock = vi + .fn() + .mockRejectedValue(new TypeError(`fetch failed while requesting ${signedUrl}`)); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + let message = ""; + try { + await downloadToTemp(signedUrl, dir, 1_000); + } catch (error) { + message = error instanceof Error ? error.message : String(error); + } + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(message).toContain("transient network error"); + expect(message).not.toContain("customer-video"); + expect(message).not.toContain("super-secret-signature"); + }); + it("cancels a streaming HTTP error body before retrying", async () => { let errorBodyCancelled = false; const errorBody = new ReadableStream({ @@ -393,6 +1151,24 @@ describe("downloadToTemp atomic publication and bounded retry", () => { expect(temporaryDownloadEntries(dir)).toEqual([]); }); + it("refetches when the cache path disappears before its first inspection", async () => { + const url = "https://cdn.example/first-lstat-race.mp4"; + const dir = makeTempDir(); + const cacheName = `download_${createHash("md5").update(url).digest("hex").slice(0, 12)}.mp4`; + const cachePath = join(dir, cacheName); + writeFileSync(cachePath, "vanishing-cache-entry"); + fsRaceControls.deleteBeforeLstatPath = cachePath; + const fetchMock = vi.fn().mockResolvedValue(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + + const path = await downloadToTemp(url, dir, 1_000); + + expect(path).toBe(cachePath); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + it("does not trust a nonempty symlink at the final cache path", async () => { const fetchMock = vi.fn().mockResolvedValue(new Response("downloaded")); vi.stubGlobal("fetch", fetchMock); @@ -447,6 +1223,206 @@ describe("downloadToTemp atomic publication and bounded retry", () => { expect(fetchMock).toHaveBeenCalledTimes(1); }); + it("waits for a cross-process cache-path lock before inspecting the final path", async () => { + const url = "https://cdn.example/locked.mp4"; + const dir = makeTempDir(); + const cacheName = `download_${createHash("md5").update(url).digest("hex").slice(0, 12)}.mp4`; + const lockPath = join(dir, `${cacheName}.hf-lock`); + mkdirSync(lockPath); + const fetchMock = vi.fn().mockResolvedValue(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + + const pending = downloadToTemp(url, dir, 1_000); + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(fetchMock).not.toHaveBeenCalled(); + rmSync(lockPath, { recursive: true, force: true }); + + const path = await pending; + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("recovers a stale cross-process cache-path lock", async () => { + const url = "https://cdn.example/stale-lock.mp4"; + const dir = makeTempDir(); + const cacheName = `download_${createHash("md5").update(url).digest("hex").slice(0, 12)}.mp4`; + const lockPath = join(dir, `${cacheName}.hf-lock`); + mkdirSync(lockPath); + mkdirSync(join(lockPath, ".hf-owner-stale")); + const staleTime = new Date(Date.now() - 6 * 60_000); + utimesSync(lockPath, staleTime, staleTime); + const fetchMock = vi.fn().mockResolvedValue(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + + const path = await downloadToTemp(url, dir, 1_000); + + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(existsSync(lockPath)).toBe(false); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("does not mix a stale lstat with a successor owner read", async () => { + const url = "https://cdn.example/stale-lock-successor.mp4"; + const dir = makeTempDir(); + const cacheName = `download_${createHash("md5").update(url).digest("hex").slice(0, 12)}.mp4`; + const lockPath = join(dir, `${cacheName}.hf-lock`); + mkdirSync(lockPath); + mkdirSync(join(lockPath, ".hf-owner-stale")); + const staleTime = new Date(Date.now() - 6 * 60_000); + utimesSync(lockPath, staleTime, staleTime); + fsRaceControls.replaceStaleLockAfterObservationPath = lockPath; + const fetchMock = vi.fn().mockResolvedValue(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + + const pending = downloadToTemp(url, dir, 1_000); + await new Promise((resolve) => setTimeout(resolve, 25)); + + expect(existsSync(lockPath)).toBe(true); + expect(fetchMock).not.toHaveBeenCalled(); + rmSync(lockPath, { recursive: true, force: true }); + + const path = await pending; + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("does not release a successor cache lock created at the same path", async () => { + const url = "https://cdn.example/successor-lock.mp4"; + const dir = makeTempDir(); + const cacheName = `download_${createHash("md5").update(url).digest("hex").slice(0, 12)}.mp4`; + const cachePath = join(dir, cacheName); + const lockPath = `${cachePath}.hf-lock`; + writeFileSync(cachePath, "cached"); + fsRaceControls.replaceLockOnReleasePath = lockPath; + const fetchMock = vi.fn(); + vi.stubGlobal("fetch", fetchMock); + + const path = await downloadToTemp(url, dir, 1_000); + + expect(path).toBe(cachePath); + expect(fetchMock).not.toHaveBeenCalled(); + expect(existsSync(lockPath)).toBe(true); + rmSync(lockPath, { recursive: true, force: true }); + }); + + it("keeps the replacement path intact across cross-signal stale-cache callers", async () => { + const url = "https://cdn.example/stale-concurrent.woff2"; + const dir = makeTempDir(); + const cacheName = `download_${createHash("md5").update(url).digest("hex").slice(0, 12)}.woff2`; + writeFileSync(join(dir, cacheName), "expired"); + const firstController = new AbortController(); + const secondController = new AbortController(); + const body = Buffer.from("wOF2replacement"); + const fetchMock = vi.fn().mockImplementation( + () => + new Promise((resolve) => { + setTimeout(() => resolve(new Response(body)), 10); + }), + ); + vi.stubGlobal("fetch", fetchMock); + + const [first, second] = await Promise.all([ + downloadToTemp(url, dir, 1_000, firstController.signal), + downloadToTemp(url, dir, 1_000, secondController.signal), + ]); + + expect(first).toBe(second); + expect(existsSync(first)).toBe(true); + expect(readFileSync(first)).toEqual(body); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("never overwrites an artifact returned by a concurrent checksum scope", async () => { + const firstBytes = isoBmffMediaBytes("first-version"); + const secondBytes = isoBmffMediaBytes("second-version"); + const firstSha = createHash("sha256").update(firstBytes).digest("hex"); + const secondSha = createHash("sha256").update(secondBytes).digest("hex"); + const fetchMock = vi + .fn() + .mockImplementationOnce( + () => + new Promise((resolve) => + setTimeout(() => resolve(new Response(firstBytes)), 20), + ), + ) + .mockImplementationOnce( + () => + new Promise((resolve) => + setTimeout(() => resolve(new Response(secondBytes)), 5), + ), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + const url = "https://cdn.example/versioned.mp4"; + + const [firstPath, secondPath] = await Promise.all([ + downloadToTemp(url, dir, 1_000, undefined, undefined, { + expectedSha256: firstSha, + }), + downloadToTemp(url, dir, 1_000, undefined, undefined, { + expectedSha256: secondSha, + }), + ]); + + expect(firstPath).not.toBe(secondPath); + expect(readFileSync(firstPath)).toEqual(firstBytes); + expect(readFileSync(secondPath)).toEqual(secondBytes); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it("locally refetches when a concurrent race winner disappears before validation", async () => { + const url = "https://cdn.example/vanishing-race-winner.mp4"; + const dir = makeTempDir(); + const cacheName = `download_${createHash("md5").update(url).digest("hex").slice(0, 12)}.mp4`; + const cachePath = join(dir, cacheName); + fsRaceControls.injectRaceAtLinkPath = cachePath; + fsRaceControls.deleteBeforeReadPath = cachePath; + const onTelemetry = vi.fn(); + const fetchMock = vi.fn().mockImplementation(() => Promise.resolve(new Response("complete"))); + vi.stubGlobal("fetch", fetchMock); + + const path = await downloadToTemp(url, dir, 1_000, undefined, undefined, { onTelemetry }); + + expect(path).toBe(cachePath); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(onTelemetry).toHaveBeenCalledWith( + expect.objectContaining({ outcome: "attempt_failed", failureKind: "filesystem" }), + ); + expect(onTelemetry).toHaveBeenCalledWith( + expect.objectContaining({ outcome: "retrying", failureKind: "filesystem" }), + ); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("locally refetches when an EEXIST race winner disappears before its first lstat", async () => { + const url = "https://cdn.example/vanishing-race-winner-before-lstat.mp4"; + const dir = makeTempDir(); + const cacheName = `download_${createHash("md5").update(url).digest("hex").slice(0, 12)}.mp4`; + const cachePath = join(dir, cacheName); + fsRaceControls.injectRaceAtLinkPath = cachePath; + fsRaceControls.deleteInjectedWinnerBeforeLstatPath = cachePath; + const onTelemetry = vi.fn(); + const fetchMock = vi.fn().mockImplementation(() => Promise.resolve(new Response("complete"))); + vi.stubGlobal("fetch", fetchMock); + + const path = await downloadToTemp(url, dir, 1_000, undefined, undefined, { onTelemetry }); + + expect(path).toBe(cachePath); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(onTelemetry).toHaveBeenCalledWith( + expect.objectContaining({ outcome: "attempt_failed", failureKind: "filesystem" }), + ); + expect(onTelemetry).toHaveBeenCalledWith( + expect.objectContaining({ outcome: "retrying", failureKind: "filesystem" }), + ); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + it("does not let one caller cancellation abort another caller", async () => { const firstController = new AbortController(); const secondController = new AbortController(); @@ -545,7 +1521,7 @@ describe("downloadToTemp atomic publication and bounded retry", () => { } satisfies Partial); const path = await first; expect(readFileSync(path, "utf8")).toBe("complete"); - expect(fetchMock).toHaveBeenCalledTimes(2); + expect(fetchMock).toHaveBeenCalledTimes(1); expect(temporaryDownloadEntries(dir)).toEqual([]); }); }); diff --git a/packages/engine/src/utils/urlDownloader.ts b/packages/engine/src/utils/urlDownloader.ts index 7e9234537..3ba5fe650 100644 --- a/packages/engine/src/utils/urlDownloader.ts +++ b/packages/engine/src/utils/urlDownloader.ts @@ -1,20 +1,25 @@ import { closeSync, + createReadStream, createWriteStream, existsSync, fsyncSync, + linkSync, mkdtempSync, mkdirSync, lstatSync, openSync, - renameSync, + readdirSync, + rmdirSync, rmSync, statSync, + type Stats, + unlinkSync, } from "fs"; -import { createHash } from "crypto"; +import { createHash, randomUUID } from "crypto"; import { BlockList, isIP } from "node:net"; import { dirname, extname, join } from "path"; -import { Readable } from "stream"; +import { Readable, Transform } from "stream"; import { pipeline } from "stream/promises"; const inFlightDownloads = new Map>(); @@ -40,22 +45,104 @@ export type UrlDownloadFailureKind = | "http_transient" | "network" | "empty_body" + | "range_protocol" + | "length_mismatch" + | "hash_mismatch" + | "invalid_payload" | "filesystem"; +export type UrlDownloadTelemetryOutcome = + | "attempt_failed" + | "cache_hit" + | "published" + | "race_reused" + | "retrying"; + +export interface UrlDownloadTelemetry { + urlFingerprint: string; + initialHost?: string; + finalHost?: string; + attempt: number; + outcome: UrlDownloadTelemetryOutcome; + status?: number; + expectedBytes?: number; + receivedBytes?: number; + rangeDisposition?: + | "none" + | "malformed_206" + | "unsolicited_206" + | "content_range_on_200" + | "full_object_200"; + etagFingerprint?: string; + etagWeak?: boolean; + localSize?: number; + localSha256?: string; + failureKind?: UrlDownloadFailureKind; +} + +export interface UrlDownloadOptions { + /** Optional strong checksum supplied by a trusted caller. */ + expectedSha256?: string; + onTelemetry?: (event: UrlDownloadTelemetry) => void; +} + +export interface PublicHttpsTextOptions { + /** Maximum decoded response bytes retained in memory. */ + maxBytes: number; + timeoutMs?: number; + signal?: AbortSignal; +} + +export interface SafeDownloadUrlIdentity { + urlFingerprint: string; + host?: string; +} + +/** Query-free, non-reversible URL identity suitable for logs and metrics. */ +export function safeDownloadUrlIdentity(url: string): SafeDownloadUrlIdentity { + let canonical = url; + let host: string | undefined; + try { + const parsed = new URL(url); + canonical = `${parsed.origin}${parsed.pathname}`; + host = parsed.hostname.toLowerCase(); + } catch { + // Invalid input is still fingerprinted; never echo it in diagnostics. + } + return { + urlFingerprint: createHash("sha256").update(canonical).digest("hex"), + host, + }; +} + +/** Default safe structured sink for engine media call sites without a logger. */ +export function writeUrlDownloadTelemetry(event: UrlDownloadTelemetry): void { + try { + process.stderr.write(`[hyperframes:download] ${JSON.stringify(event)}\n`); + } catch { + // Observability must never change download correctness. + } +} + export class UrlDownloadError extends Error { constructor( readonly kind: UrlDownloadFailureKind, readonly retryable: boolean, message: string, readonly status?: number, + readonly telemetry?: Partial, + /** A bounded in-call refetch can be safe even when upstream retry is not. */ + readonly locallyRetryable: boolean = retryable, ) { super(message); this.name = "UrlDownloadError"; } } -function classifyHttpFailure(status: number, statusText: string): UrlDownloadError { - const message = `HTTP ${status}: ${statusText}`; +function classifyHttpFailure(status: number): UrlDownloadError { + // Response.statusText is remote-controlled and some CDNs/proxies echo the + // signed request URL into it. The numeric status is sufficient and bounded. + const message = `HTTP ${status}`; if (status === 404 || status === 410) { return new UrlDownloadError("http_not_found", false, message, status); } @@ -67,20 +154,27 @@ function classifyHttpFailure(status: number, statusText: string): UrlDownloadErr function classifyDownloadFailure(error: unknown): UrlDownloadError { if (error instanceof UrlDownloadError) return error; - const message = error instanceof Error ? error.message : String(error); let current: unknown = error; // Undici often wraps a mid-body socket failure as `TypeError: terminated` // with the actionable `UND_ERR_*` code on `cause`. for (let depth = 0; current && depth < 4; depth += 1) { if (isRetryableNetworkCause(current)) { - return new UrlDownloadError("network", true, `Download failed: ${message}`); + return new UrlDownloadError( + "network", + true, + "Download failed due to a transient network error", + ); } current = typeof current === "object" && current !== null && "cause" in current ? current.cause : undefined; } - return new UrlDownloadError("filesystem", false, `Download failed: ${message}`); + return new UrlDownloadError( + "filesystem", + false, + "Download failed while writing the local artifact", + ); } const RETRYABLE_NETWORK_CODES = new Set(["ECONNRESET", "ECONNREFUSED", "ETIMEDOUT", "EAI_AGAIN"]); @@ -152,35 +246,202 @@ export function assertPublicHttpsUrl(url: string): void { try { parsed = new URL(url); } catch { - throw new Error(`[URLDownloader] Invalid URL: ${url}`); + throw new Error("[URLDownloader] Invalid URL"); } if (parsed.protocol !== "https:") { - throw new Error( - `[URLDownloader] Only HTTPS URLs are permitted in compositions (got ${parsed.protocol}): ${url}`, - ); + throw new Error(`[URLDownloader] Only HTTPS URLs are permitted in compositions`); } if (isBlockedHost(parsed.hostname)) { - throw new Error( - `[URLDownloader] URL targets a private/reserved address and is not permitted: ${url}`, - ); + throw new Error("[URLDownloader] URL targets a private/reserved address and is not permitted"); } } -function getFilenameFromUrl(url: string): string { - const hash = createHash("md5").update(url).digest("hex").slice(0, 12); +function getFilenameFromUrl(url: string, validationScope: string): string { + const physicalIdentity = validationScope === "" ? url : `${url}\0${validationScope}`; + const hash = createHash("md5").update(physicalIdentity).digest("hex").slice(0, 12); const urlObj = new URL(url); const ext = extname(urlObj.pathname) || ".mp4"; return `download_${hash}${ext}`; } -function hasCompleteFile(path: string): boolean { - if (!existsSync(path)) return false; - const entry = lstatSync(path); - if (entry.isFile() && entry.size > 0) return true; - // Old versions could leave an empty file behind. Never trust that stale - // cache entry—or a symlink/special entry planted at the final path. - rmSync(path, { recursive: entry.isDirectory(), force: true }); - return false; +function sameFileIdentity(left: Stats, right: Stats): boolean { + return left.dev === right.dev && left.ino === right.ino; +} + +const CACHE_LOCK_POLL_MS = 10; +const CACHE_LOCK_STALE_MS = 5 * 60_000; +const CACHE_LOCK_RECLAIM_NAME = ".hf-reclaim"; +const CACHE_LOCK_OWNER_PREFIX = ".hf-owner-"; + +interface CacheLockObservation { + stats: Stats; + owner?: string; +} + +function sameCacheLockStatGeneration(left: Stats, right: Stats): boolean { + return ( + sameFileIdentity(left, right) && + left.mtimeMs === right.mtimeMs && + left.ctimeMs === right.ctimeMs && + left.birthtimeMs === right.birthtimeMs + ); +} + +function observeCachePathLock(lockPath: string): CacheLockObservation { + for (let pass = 0; pass < 3; pass += 1) { + const before = lstatSync(lockPath); + const owner = readdirSync(lockPath).find((name) => name.startsWith(CACHE_LOCK_OWNER_PREFIX)); + const after = lstatSync(lockPath); + if (sameCacheLockStatGeneration(before, after)) return { stats: after, owner }; + } + throw new UrlDownloadError("filesystem", true, "Cache lock changed repeatedly during inspection"); +} + +function sameCachePathLock(left: CacheLockObservation, right: CacheLockObservation): boolean { + if (left.owner !== undefined || right.owner !== undefined) { + return left.owner !== undefined && left.owner === right.owner; + } + // Backward-compatible fallback for lock directories created by an older + // process before ownership markers were introduced. + return sameFileIdentity(left.stats, right.stats); +} + +function removeCacheLockDirectoryIfEmpty(lockPath: string): void { + try { + rmdirSync(lockPath); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code !== "ENOENT" && code !== "ENOTEMPTY" && code !== "EEXIST") throw error; + } +} + +function releaseOwnedCachePathLock(lockPath: string, owner: string): void { + try { + // Consuming the unique owner marker elects exactly one releaser. The + // non-recursive rmdir below cannot delete a successor with its own marker. + rmdirSync(join(lockPath, owner)); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return; + throw error; + } + removeCacheLockDirectoryIfEmpty(lockPath); +} + +async function waitForCacheLock(signal?: AbortSignal): Promise { + if (signal?.aborted) { + throw new UrlDownloadError("cancelled", false, "Download cancelled"); + } + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + signal?.removeEventListener("abort", onAbort); + resolve(); + }, CACHE_LOCK_POLL_MS); + const onAbort = (): void => { + clearTimeout(timeout); + signal?.removeEventListener("abort", onAbort); + reject(new UrlDownloadError("cancelled", false, "Download cancelled")); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + +// The lock loop keeps filesystem races, stale-lock recovery, cancellation, and timeout together. +// fallow-ignore-next-line complexity +async function acquireCachePathLock( + localPath: string, + timeoutMs: number, + signal?: AbortSignal, +): Promise<() => void> { + const lockPath = `${localPath}.hf-lock`; + const startedAt = Date.now(); + for (;;) { + if (signal?.aborted) { + throw new UrlDownloadError("cancelled", false, "Download cancelled"); + } + let createdLock = false; + try { + mkdirSync(lockPath); + createdLock = true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; + // Another process owns the path; inspect it below. + } + if (createdLock) { + const owner = `${CACHE_LOCK_OWNER_PREFIX}${randomUUID()}`; + try { + mkdirSync(join(lockPath, owner)); + const entries = readdirSync(lockPath); + if (entries.length === 1 && entries[0] === owner) { + return () => releaseOwnedCachePathLock(lockPath, owner); + } + // The empty directory was replaced or another creator reached it + // before our marker. Consume only our marker and retry ownership. + rmdirSync(join(lockPath, owner)); + removeCacheLockDirectoryIfEmpty(lockPath); + continue; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; + throw error; + } + } + + let observedLock: CacheLockObservation; + try { + observedLock = observeCachePathLock(lockPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; + throw error; + } + if (Date.now() - observedLock.stats.mtimeMs > CACHE_LOCK_STALE_MS) { + if (observedLock.owner) { + // Removing the exact unique marker is an atomic ownership claim. + // A competing releaser/reclaimer gets ENOENT and must re-observe. + try { + rmdirSync(join(lockPath, observedLock.owner)); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; + throw error; + } + removeCacheLockDirectoryIfEmpty(lockPath); + continue; + } + + // Compatibility path for stale marker-less locks from an older process. + const reclaimPath = join(lockPath, CACHE_LOCK_RECLAIM_NAME); + try { + // A marker inside the observed lock serializes competing stale + // reclaimers without introducing another independently stale lock. + mkdirSync(reclaimPath); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === "EEXIST" || code === "ENOENT") continue; + throw error; + } + try { + const currentLock = observeCachePathLock(lockPath); + if (sameCachePathLock(currentLock, observedLock)) { + rmdirSync(reclaimPath); + removeCacheLockDirectoryIfEmpty(lockPath); + } else { + // The stale lock was replaced after our observation. Remove only + // our marker from the successor and leave its ownership intact. + rmdirSync(reclaimPath); + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + continue; + } + + if (Date.now() - startedAt >= timeoutMs) { + throw new UrlDownloadError( + "timeout", + true, + `Download cache lock timeout after ${timeoutMs / 1000}s`, + ); + } + await waitForCacheLock(signal); + } } const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]); @@ -228,7 +489,7 @@ function resolveRedirectUrl(response: Response, currentUrl: string, redirects: n async function fetchWithValidatedRedirects( initialUrl: string, controller: AbortController, -): Promise { +): Promise<{ response: Response; finalUrl: string }> { let currentUrl = initialUrl; for (let redirects = 0; ; redirects += 1) { assertAllowedDownloadUrl(currentUrl, redirects > 0); @@ -239,20 +500,292 @@ async function fetchWithValidatedRedirects( const response = await fetch(currentUrl, { signal: controller.signal, redirect: "manual", + headers: { "accept-encoding": "identity" }, }); - if (!REDIRECT_STATUSES.has(response.status)) return response; + if (!REDIRECT_STATUSES.has(response.status)) return { response, finalUrl: currentUrl }; await cancelResponseBody(response); currentUrl = resolveRedirectUrl(response, currentUrl, redirects); } } +/** Fetch bounded UTF-8 text while applying the downloader's redirect and SSRF policy to every hop. */ +// fallow-ignore-next-line complexity +export async function fetchPublicHttpsText( + url: string, + options: PublicHttpsTextOptions, +): Promise { + const timeoutMs = options.timeoutMs ?? 15_000; + if (!Number.isSafeInteger(options.maxBytes) || options.maxBytes <= 0) { + throw new RangeError("maxBytes must be a positive safe integer"); + } + assertPublicHttpsUrl(url); + + const controller = new AbortController(); + let timedOut = false; + let callerAborted = options.signal?.aborted ?? false; + const onCallerAbort = (): void => { + callerAborted = true; + controller.abort(); + }; + options.signal?.addEventListener("abort", onCallerAbort, { once: true }); + const timeoutId = setTimeout(() => { + timedOut = true; + controller.abort(); + }, timeoutMs); + + try { + if (callerAborted) { + throw new UrlDownloadError("cancelled", false, "Text fetch cancelled"); + } + const { response } = await fetchWithValidatedRedirects(url, controller); + if (!response.ok) { + await cancelResponseBody(response); + throw classifyHttpFailure(response.status); + } + if (!response.body) return ""; + + let declaredLength: number | undefined; + try { + declaredLength = parseDeclaredLength(response); + } catch (error) { + await cancelResponseBody(response); + throw error; + } + const contentEncoding = response.headers.get("content-encoding")?.trim().toLowerCase(); + const expectedBytes = + !contentEncoding || contentEncoding === "identity" ? declaredLength : undefined; + if (declaredLength !== undefined && declaredLength > options.maxBytes) { + await cancelResponseBody(response); + throw new UrlDownloadError( + "length_mismatch", + false, + "Text response exceeded the configured byte limit", + response.status, + ); + } + + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let receivedBytes = 0; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + receivedBytes += value.byteLength; + if (receivedBytes > options.maxBytes) { + await reader.cancel(); + throw new UrlDownloadError( + "length_mismatch", + false, + "Text response exceeded the configured byte limit", + response.status, + { receivedBytes }, + ); + } + chunks.push(value); + } + if (expectedBytes !== undefined && receivedBytes !== expectedBytes) { + throw new UrlDownloadError( + "length_mismatch", + true, + "Text response byte count did not match its declared length", + response.status, + { expectedBytes, receivedBytes }, + ); + } + return new TextDecoder().decode(Buffer.concat(chunks.map((chunk) => Buffer.from(chunk)))); + } catch (error) { + if (callerAborted) { + throw new UrlDownloadError("cancelled", false, "Text fetch cancelled"); + } + if (timedOut) { + throw new UrlDownloadError("timeout", true, `Text fetch timeout after ${timeoutMs / 1000}s`); + } + throw classifyDownloadFailure(error); + } finally { + clearTimeout(timeoutId); + options.signal?.removeEventListener("abort", onCallerAbort); + controller.abort(); + } +} + +interface PartialIntegrity { + finalHost?: string; + status: number; + expectedBytes?: number; + receivedBytes: number; + rangeDisposition: UrlDownloadTelemetry["rangeDisposition"]; + etagFingerprint?: string; + etagWeak?: boolean; + localSize: number; + localSha256: string; +} + +function parseDeclaredLength(response: Response): number | undefined { + const raw = response.headers.get("content-length"); + if (raw === null) return undefined; + if (!/^\d+$/.test(raw.trim())) { + throw new UrlDownloadError( + "length_mismatch", + true, + "Download response Content-Length is malformed", + response.status, + { status: response.status }, + ); + } + const value = Number(raw); + if (!Number.isSafeInteger(value) || value < 0) { + throw new UrlDownloadError( + "length_mismatch", + true, + "Download response Content-Length is out of range", + response.status, + { status: response.status }, + ); + } + return value; +} + +// Keep every Content-Range invariant in one parser so malformed and unsolicited +// partial responses cannot drift into different retry classifications. +// fallow-ignore-next-line complexity +function classifyRangeDisposition(response: Response): UrlDownloadTelemetry["rangeDisposition"] { + const contentRange = response.headers.get("content-range"); + if (response.status === 206) { + const match = contentRange?.match(/^bytes (\d+)-(\d+)\/(\d+|\*)$/i); + if (!match) return "malformed_206"; + const start = Number(match[1]); + const end = Number(match[2]); + const total = match[3] === "*" ? undefined : Number(match[3]); + if ( + !Number.isSafeInteger(start) || + !Number.isSafeInteger(end) || + start < 0 || + end < start || + (total !== undefined && (!Number.isSafeInteger(total) || total <= end)) + ) { + return "malformed_206"; + } + return "unsolicited_206"; + } + if (response.status !== 200 || contentRange === null) return "none"; + + const match = contentRange.match(/^bytes (\d+)-(\d+)\/(\d+)$/i); + const contentLength = response.headers.get("content-length")?.trim(); + const contentEncoding = response.headers.get("content-encoding")?.trim().toLowerCase(); + if (!match || !contentLength || !/^\d+$/.test(contentLength)) { + return "content_range_on_200"; + } + const start = Number(match[1]); + const end = Number(match[2]); + const total = Number(match[3]); + const declaredLength = Number(contentLength); + return Number.isSafeInteger(start) && + Number.isSafeInteger(end) && + Number.isSafeInteger(total) && + Number.isSafeInteger(declaredLength) && + start === 0 && + total > 0 && + end === total - 1 && + declaredLength === total && + (!contentEncoding || contentEncoding === "identity") + ? "full_object_200" + : "content_range_on_200"; +} + +function looksLikeRemoteErrorDocument(prefix: Buffer): boolean { + const text = prefix + .toString("utf8") + .replace(/^\uFEFF?\s*/, "") + .toLowerCase(); + return ( + text.startsWith(", +): UrlDownloadError { + return new UrlDownloadError( + "hash_mismatch", + source === "server", + "Download payload checksum did not match", + status, + telemetry, + true, + ); +} + +// Response protocol, streamed byte accounting, hashes, and payload validation +// share one lifecycle so no validation can happen after publication. +// fallow-ignore-next-line complexity async function fetchToPartial( url: string, partialPath: string, controller: AbortController, -): Promise { - const response = await fetchWithValidatedRedirects(url, controller); + options: UrlDownloadOptions, +): Promise { + const { response, finalUrl } = await fetchWithValidatedRedirects(url, controller); + const finalIdentity = safeDownloadUrlIdentity(finalUrl); + const rangeDisposition = classifyRangeDisposition(response); + if (rangeDisposition !== "none" && rangeDisposition !== "full_object_200") { + await cancelResponseBody(response); + throw new UrlDownloadError( + "range_protocol", + true, + rangeDisposition === "malformed_206" + ? "Download received a malformed unsolicited partial response" + : "Download received an unsolicited partial response", + response.status, + { + finalHost: finalIdentity.host, + status: response.status, + rangeDisposition, + }, + ); + } if (!response.ok) { // Do not leave a streaming error response holding an Undici connection // while the bounded retry starts. @@ -261,22 +794,148 @@ async function fetchToPartial( } catch { // The HTTP status remains the useful failure if teardown also fails. } - throw classifyHttpFailure(response.status, response.statusText); + const classified = classifyHttpFailure(response.status); + throw new UrlDownloadError( + classified.kind, + classified.retryable, + classified.message, + classified.status, + { finalHost: finalIdentity.host, status: response.status, rangeDisposition }, + ); } if (!response.body) { - throw new UrlDownloadError("empty_body", true, "Download response body is empty"); + throw new UrlDownloadError( + "empty_body", + true, + "Download response body is empty", + response.status, + { + finalHost: finalIdentity.host, + status: response.status, + }, + ); } - const fileStream = createWriteStream(partialPath, { flags: "wx" }); + let declaredLength: number | undefined; + try { + declaredLength = parseDeclaredLength(response); + } catch (error) { + await cancelResponseBody(response); + if (error instanceof UrlDownloadError) { + throw new UrlDownloadError(error.kind, error.retryable, error.message, error.status, { + ...error.telemetry, + finalHost: finalIdentity.host, + rangeDisposition, + }); + } + throw error; + } + const contentEncoding = response.headers.get("content-encoding")?.trim().toLowerCase(); + const expectedBytes = + !contentEncoding || contentEncoding === "identity" ? declaredLength : undefined; + + let receivedBytes = 0; + const sha256 = createHash("sha256"); + const md5 = createHash("md5"); + const prefixChunks: Buffer[] = []; + let prefixBytes = 0; + const inspector = new Transform({ + transform(chunk: Buffer, _encoding, callback) { + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + receivedBytes += bytes.length; + sha256.update(bytes); + md5.update(bytes); + if (prefixBytes < 1024) { + const remaining = 1024 - prefixBytes; + const sample = bytes.subarray(0, remaining); + prefixChunks.push(sample); + prefixBytes += sample.length; + } + callback(null, bytes); + }, + }); + // eslint-disable-next-line @typescript-eslint/no-explicit-any const readableStream = Readable.fromWeb(response.body as any); - await pipeline(readableStream, fileStream); - if (statSync(partialPath).size === 0) { - throw new UrlDownloadError("empty_body", true, "Download response body contained zero bytes"); + const fileStream = createWriteStream(partialPath, { flags: "wx" }); + await pipeline(readableStream, inspector, fileStream); + const localSize = statSync(partialPath).size; + const sha256Bytes = sha256.digest(); + const localSha256 = sha256Bytes.toString("hex"); + const md5Base64 = md5.digest("base64"); + const telemetry = { + finalHost: finalIdentity.host, + status: response.status, + expectedBytes, + receivedBytes, + rangeDisposition, + localSize, + localSha256, + } satisfies Partial; + if (receivedBytes === 0 || localSize === 0) { + throw new UrlDownloadError( + "empty_body", + true, + "Download response body contained zero bytes", + response.status, + telemetry, + ); } + if ( + localSize !== receivedBytes || + (expectedBytes !== undefined && receivedBytes !== expectedBytes) + ) { + throw new UrlDownloadError( + "length_mismatch", + true, + "Download response byte count did not match its declared length", + response.status, + telemetry, + ); + } + const prefix = Buffer.concat(prefixChunks); + if (looksLikeRemoteErrorDocument(prefix)) { + throw new UrlDownloadError( + "invalid_payload", + false, + "Download returned an HTML or JSON error document", + response.status, + telemetry, + ); + } + + const expectedSha256 = expectedResponseSha256(response, options.expectedSha256); + const checksumMatches = + expectedSha256 === null || + (expectedSha256.encoding === "base64" + ? sha256Bytes.toString("base64") === expectedSha256.value + : localSha256 === expectedSha256.value); + const contentMd5 = response.headers.get("content-md5")?.trim(); + if (!checksumMatches) { + throw checksumMismatchError(expectedSha256?.source ?? "server", response.status, telemetry); + } + if (expectedSha256 === null && contentMd5 && md5Base64 !== contentMd5) { + throw checksumMismatchError("server", response.status, telemetry); + } + + const etag = response.headers.get("etag")?.trim(); + return { + finalHost: finalIdentity.host, + status: response.status, + expectedBytes, + receivedBytes, + rangeDisposition, + etagFingerprint: etag ? createHash("sha256").update(etag).digest("hex") : undefined, + etagWeak: etag ? /^W\//i.test(etag) : undefined, + localSize, + localSha256, + }; } -function syncAndPublishPartial(partialPath: string, localPath: string): void { +function syncAndPublishPartial( + partialPath: string, + localPath: string, +): Extract { // Windows rejects fsync on a read-only handle (EPERM); the partial is ours // and writable, so r+ preserves the same flush semantics cross-platform. const fd = openSync(partialPath, "r+"); @@ -286,20 +945,48 @@ function syncAndPublishPartial(partialPath: string, localPath: string): void { closeSync(fd); } - // Different cancellation scopes intentionally do not share a physical - // request. If another complete attempt won the final-path race, reuse it. - if (hasCompleteFile(localPath)) return; + // A hard-link publish is atomic and no-clobber: unlike rename(), it cannot + // replace a path that another successful caller has already returned. try { - renameSync(partialPath, localPath); + linkSync(partialPath, localPath); + unlinkSync(partialPath); + return "published"; } catch (error) { - if (!hasCompleteFile(localPath)) throw error; + const code = (error as NodeJS.ErrnoException).code; + if (code !== "EEXIST") throw error; + let winner: Stats; + try { + winner = lstatSync(localPath); + } catch (inspectionError) { + if ((inspectionError as NodeJS.ErrnoException).code !== "ENOENT") throw inspectionError; + throw new UrlDownloadError( + "filesystem", + true, + "Concurrent cache artifact disappeared before validation", + ); + } + if (!winner.isFile() || winner.size === 0) throw error; + return "race_reused"; } } +function emitDownloadTelemetry(options: UrlDownloadOptions, event: UrlDownloadTelemetry): void { + try { + options.onTelemetry?.(event); + } catch { + // Metrics/logging callbacks cannot affect publication or retry policy. + } +} + +// Attempt-scoped cancellation, cleanup, publication, and telemetry deliberately +// remain under one try/finally so every exit removes the unique partial directory. +// fallow-ignore-next-line complexity async function runDownloadAttempt( url: string, localPath: string, timeoutMs: number, + attempt: number, + options: UrlDownloadOptions, signal?: AbortSignal, ): Promise { // A private, unguessable directory prevents symlink planting and keeps the @@ -318,22 +1005,92 @@ async function runDownloadAttempt( timedOut = true; controller.abort(); }, timeoutMs); + const identity = safeDownloadUrlIdentity(url); try { if (callerAborted) { throw new UrlDownloadError("cancelled", false, "Download cancelled"); } - await fetchToPartial(url, partialPath, controller); - syncAndPublishPartial(partialPath, localPath); + const integrity = await fetchToPartial(url, partialPath, controller, options); + let outcome = syncAndPublishPartial(partialPath, localPath); + let publishedIntegrity = integrity; + if (outcome === "race_reused") { + let inspection: Awaited>; + try { + inspection = await inspectExistingFile(localPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + throw new UrlDownloadError( + "filesystem", + true, + "Concurrent cache artifact disappeared before validation", + integrity.status, + integrity, + ); + } + publishedIntegrity = { + ...integrity, + localSize: inspection.localSize, + localSha256: inspection.localSha256, + }; + if (!localInspectionMatchesOptions(inspection, options)) { + if (looksLikeRemoteErrorDocument(inspection.prefix)) { + throw new UrlDownloadError( + "invalid_payload", + false, + "Concurrent download published an HTML or JSON error document", + integrity.status, + publishedIntegrity, + ); + } + throw checksumMismatchError("caller", integrity.status, publishedIntegrity); + } + } + emitDownloadTelemetry(options, { + urlFingerprint: identity.urlFingerprint, + initialHost: identity.host, + attempt, + outcome, + ...publishedIntegrity, + }); return localPath; } catch (error) { if (callerAborted) { - throw new UrlDownloadError("cancelled", false, "Download cancelled"); + const classified = new UrlDownloadError("cancelled", false, "Download cancelled"); + emitDownloadTelemetry(options, { + urlFingerprint: identity.urlFingerprint, + initialHost: identity.host, + attempt, + outcome: "attempt_failed", + failureKind: classified.kind, + }); + throw classified; } if (timedOut) { - throw new UrlDownloadError("timeout", true, `Download timeout after ${timeoutMs / 1000}s`); + const classified = new UrlDownloadError( + "timeout", + true, + `Download timeout after ${timeoutMs / 1000}s`, + ); + emitDownloadTelemetry(options, { + urlFingerprint: identity.urlFingerprint, + initialHost: identity.host, + attempt, + outcome: "attempt_failed", + failureKind: classified.kind, + }); + throw classified; } - throw classifyDownloadFailure(error); + const classified = classifyDownloadFailure(error); + emitDownloadTelemetry(options, { + urlFingerprint: identity.urlFingerprint, + initialHost: identity.host, + attempt, + outcome: "attempt_failed", + failureKind: classified.kind, + ...classified.telemetry, + }); + throw classified; } finally { clearTimeout(timeoutId); signal?.removeEventListener("abort", onCallerAbort); @@ -348,36 +1105,161 @@ async function downloadWithRetry( timeoutMs: number, signal?: AbortSignal, onTransientRetry?: (error: UrlDownloadError) => void, + options: UrlDownloadOptions = {}, ): Promise { const maxTransientRetries = 1; for (let attempt = 0; ; attempt += 1) { try { - return await runDownloadAttempt(url, localPath, timeoutMs, signal); + return await runDownloadAttempt(url, localPath, timeoutMs, attempt + 1, options, signal); } catch (error) { const classified = classifyDownloadFailure(error); - if (!classified.retryable || attempt >= maxTransientRetries) throw classified; - onTransientRetry?.(classified); + if (!classified.locallyRetryable || attempt >= maxTransientRetries) throw classified; + if (classified.retryable) onTransientRetry?.(classified); + const identity = safeDownloadUrlIdentity(url); + emitDownloadTelemetry(options, { + urlFingerprint: identity.urlFingerprint, + initialHost: identity.host, + attempt: attempt + 1, + outcome: "retrying", + failureKind: classified.kind, + ...classified.telemetry, + }); } } } +async function inspectExistingFile(path: string): Promise<{ + localSize: number; + localSha256: string; + prefix: Buffer; +}> { + const sha256 = createHash("sha256"); + const prefixChunks: Buffer[] = []; + let prefixBytes = 0; + let localSize = 0; + for await (const chunk of createReadStream(path)) { + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + localSize += bytes.length; + sha256.update(bytes); + if (prefixBytes < 1024) { + const sample = bytes.subarray(0, 1024 - prefixBytes); + prefixChunks.push(sample); + prefixBytes += sample.length; + } + } + return { + localSize, + localSha256: sha256.digest("hex"), + prefix: Buffer.concat(prefixChunks), + }; +} + +function localInspectionMatchesOptions( + inspection: { localSize: number; localSha256: string; prefix: Buffer }, + options: UrlDownloadOptions, +): boolean { + const expectedSha256 = options.expectedSha256?.trim().toLowerCase(); + return ( + inspection.localSize > 0 && + !looksLikeRemoteErrorDocument(inspection.prefix) && + (!expectedSha256 || inspection.localSha256 === expectedSha256) + ); +} + +function sameCacheEntry(before: Stats, after: Stats): boolean { + return ( + sameFileIdentity(before, after) && + before.size === after.size && + before.mtimeMs === after.mtimeMs + ); +} + +// Cache identity checks must remain adjacent to invalidation to avoid widening the TOCTOU window. +// fallow-ignore-next-line complexity +async function reuseOrInvalidateCachedFile( + url: string, + localPath: string, + timeoutMs: number, + signal: AbortSignal | undefined, + options: UrlDownloadOptions, +): Promise { + const releaseLock = await acquireCachePathLock(localPath, timeoutMs, signal); + try { + // Re-inspect when a mixed-version process changes the path while it is open. + for (let pass = 0; pass < 3; pass += 1) { + let before: Stats; + try { + before = lstatSync(localPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; + throw error; + } + if (!before.isFile() || before.size === 0) { + rmSync(localPath, { recursive: before.isDirectory(), force: true }); + return false; + } + + let inspection: Awaited>; + try { + inspection = await inspectExistingFile(localPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; + throw error; + } + let after: Stats; + try { + after = lstatSync(localPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; + throw error; + } + if (!sameCacheEntry(before, after)) continue; + + if (localInspectionMatchesOptions(inspection, options)) { + const identity = safeDownloadUrlIdentity(url); + emitDownloadTelemetry(options, { + urlFingerprint: identity.urlFingerprint, + initialHost: identity.host, + attempt: 0, + outcome: "cache_hit", + receivedBytes: inspection.localSize, + localSize: inspection.localSize, + localSha256: inspection.localSha256, + rangeDisposition: "none", + }); + return true; + } + + rmSync(localPath, { force: true }); + return false; + } + return false; + } finally { + releaseLock(); + } +} + export async function downloadToTemp( url: string, destDir: string, timeoutMs: number = 300000, signal?: AbortSignal, onTransientRetry?: (error: UrlDownloadError) => void, + options: UrlDownloadOptions = {}, ): Promise { // Reject non-HTTPS URLs and private/reserved address ranges before // touching the cache or filesystem — customer-supplied compositions must // not be able to trigger outbound fetches to internal infrastructure. assertPublicHttpsUrl(url); + const expectedSha256 = normalizeCallerSha256(options.expectedSha256); + const normalizedOptions = { ...options, expectedSha256 }; const cacheKey = `${url}\0${destDir}`; // The physical request may be shared only by callers with the same // cancellation scope and deadline. Otherwise the first caller's abort or // timeout would incorrectly own every waiter. - const inFlightKey = `${cacheKey}\0${timeoutMs}\0${signalScopeKey(signal)}`; + const validationScope = expectedSha256 ?? ""; + const inFlightKey = `${cacheKey}\0${timeoutMs}\0${signalScopeKey(signal)}\0${validationScope}`; const inFlight = inFlightDownloads.get(inFlightKey); if (inFlight) { return inFlight; @@ -387,12 +1269,38 @@ export async function downloadToTemp( mkdirSync(destDir, { recursive: true }); } - const filename = getFilenameFromUrl(url); + const filename = getFilenameFromUrl(url, validationScope); const localPath = join(destDir, filename); - if (hasCompleteFile(localPath)) return localPath; - - const downloadPromise = downloadWithRetry(url, localPath, timeoutMs, signal, onTransientRetry); + // Register before the first asynchronous cache inspection so same-scope + // callers cannot both race through stale-entry invalidation. + const downloadPromise = (async () => { + const cacheStartedAt = Date.now(); + const reused = await reuseOrInvalidateCachedFile( + url, + localPath, + timeoutMs, + signal, + normalizedOptions, + ); + const remainingTimeoutMs = timeoutMs - (Date.now() - cacheStartedAt); + if (remainingTimeoutMs <= 0) { + throw new UrlDownloadError( + "timeout", + true, + `Download cache inspection timeout after ${timeoutMs / 1000}s`, + ); + } + if (reused) return localPath; + return downloadWithRetry( + url, + localPath, + remainingTimeoutMs, + signal, + onTransientRetry, + normalizedOptions, + ); + })(); const trackedDownload = downloadPromise.finally(() => { inFlightDownloads.delete(inFlightKey); }); diff --git a/packages/producer/src/services/htmlCompiler.test.ts b/packages/producer/src/services/htmlCompiler.test.ts index a69be7e02..6fda36d7f 100644 --- a/packages/producer/src/services/htmlCompiler.test.ts +++ b/packages/producer/src/services/htmlCompiler.test.ts @@ -85,6 +85,22 @@ describe("discoverMediaFromBrowser", () => { }); }); +function validTestMediaResponse(): Response { + const bytes = new Uint8Array([ + 0, 0, 0, 24, 0x66, 0x74, 0x79, 0x70, 0x69, 0x73, 0x6f, 0x6d, 0, 0, 0, 0, 0x69, 0x73, 0x6f, 0x6d, + 0x6d, 0x70, 0x34, 0x32, + ]); + return new Response(bytes, { status: 200 }); +} + +function validTestImageResponse(): Response { + const png = Buffer.from( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=", + "base64", + ); + return new Response(png, { status: 200 }); +} + describe("injectSdkPositionEditsRenderScript", () => { it("injects before when SDK position-edit markers are present", () => { const html = @@ -1424,7 +1440,7 @@ describe("localizeRemoteMediaSources", () => { it("rewrites remote