diff --git a/packages/cli/src/server/studioServer.ts b/packages/cli/src/server/studioServer.ts index e9d50f3cf3..2f57830594 100644 --- a/packages/cli/src/server/studioServer.ts +++ b/packages/cli/src/server/studioServer.ts @@ -394,7 +394,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, { @@ -403,7 +404,12 @@ 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, { + expectedContent: "media", + onTelemetry: writeUrlDownloadTelemetry, + }), + ), }); return injectDeterministicFontFaces(prepared.html); }, diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index b479920dc8..c96cceca28 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -262,7 +262,16 @@ export { type KeyframeAnalysis, } from "./utils/ffprobe.js"; -export { assertPublicHttpsUrl, downloadToTemp, isHttpUrl } from "./utils/urlDownloader.js"; +export { + assertPublicHttpsUrl, + downloadToTemp, + isHttpUrl, + safeDownloadUrlIdentity, + writeUrlDownloadTelemetry, + type SafeDownloadUrlIdentity, + type UrlDownloadOptions, + type UrlDownloadTelemetry, +} 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 bc49a149ed..36531822dc 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 1b063f3fd5..149bca51c3 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,12 @@ export async function processCompositionAudio( if (isHttpUrl(srcPath)) { try { - srcPath = await downloadToTemp(srcPath, workDir); + srcPath = await downloadToTemp(srcPath, workDir, undefined, signal, undefined, { + expectedContent: "media", + 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 6afa2f8fa6..6873c484a2 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 a3d7df8699..9e2de377d6 100644 --- a/packages/engine/src/services/videoFrameExtractor.ts +++ b/packages/engine/src/services/videoFrameExtractor.ts @@ -17,7 +17,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"; @@ -247,6 +252,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", @@ -1071,8 +1084,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), + { expectedContent: "media", onTelemetry: writeUrlDownloadTelemetry }, ); } diff --git a/packages/engine/src/utils/urlDownloader.test.ts b/packages/engine/src/utils/urlDownloader.test.ts index 8ef85317a8..d927108705 100644 --- a/packages/engine/src/utils/urlDownloader.test.ts +++ b/packages/engine/src/utils/urlDownloader.test.ts @@ -11,6 +11,7 @@ import { } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { createHash } from "node:crypto"; import { assertPublicHttpsUrl, downloadToTemp, UrlDownloadError } from "./urlDownloader.js"; const tempDirs: string[] = []; @@ -27,6 +28,17 @@ function temporaryDownloadEntries(dir: string): string[] { ); } +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(); for (const dir of tempDirs.splice(0)) { @@ -120,6 +132,18 @@ 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("downloadToTemp atomic publication and bounded retry", () => { @@ -141,7 +165,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 +239,365 @@ 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, + { expectedContent: "media" }, + ); + + 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("rejects Content-Range on a 200 response", 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("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, + { expectedContent: "media", expectedSha256 }, + ); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path)).toEqual(complete); + }); + + 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, + { expectedContent: "media" }, + ); + + 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, { + expectedContent: "media", + }), + ).rejects.toMatchObject({ + kind: "invalid_payload", + retryable: false, + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("rejects arbitrary non-media bytes 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, { + expectedContent: "media", + }), + ).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, + { expectedContent: "media" }, + ); + + 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, { + expectedContent: "media", + 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([ + [ + "mpeg-ts", + (() => { + const bytes = Buffer.alloc(188 * 3); + bytes[0] = 0x47; + bytes[188] = 0x47; + bytes[376] = 0x47; + return bytes; + })(), + ], + [ + "m2ts", + (() => { + const bytes = Buffer.alloc(4 + 192 * 3); + bytes[4] = 0x47; + bytes[196] = 0x47; + bytes[388] = 0x47; + return bytes; + })(), + ], + [ + "mxf", + Buffer.from([ + 0x06, 0x0e, 0x2b, 0x34, 0x02, 0x05, 0x01, 0x01, 0x0d, 0x01, 0x02, 0x01, 0x01, 0x02, 0x04, + 0x00, 0x83, 0x00, + ]), + ], + ])("accepts extensionless supported %s container signatures", 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, + undefined, + undefined, + { expectedContent: "media" }, + ); + + 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({ @@ -447,6 +833,46 @@ describe("downloadToTemp atomic publication and bounded retry", () => { expect(fetchMock).toHaveBeenCalledTimes(1); }); + 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, { + expectedContent: "media", + expectedSha256: firstSha, + }), + downloadToTemp(url, dir, 1_000, undefined, undefined, { + expectedContent: "media", + expectedSha256: secondSha, + }), + ]); + + expect(firstPath).not.toBe(secondPath); + expect(readFileSync(firstPath)).toEqual(firstBytes); + expect(readFileSync(secondPath)).toEqual(secondBytes); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + it("does not let one caller cancellation abort another caller", async () => { const firstController = new AbortController(); const secondController = new AbortController(); diff --git a/packages/engine/src/utils/urlDownloader.ts b/packages/engine/src/utils/urlDownloader.ts index 7e92345375..fcb97d7edc 100644 --- a/packages/engine/src/utils/urlDownloader.ts +++ b/packages/engine/src/utils/urlDownloader.ts @@ -1,20 +1,22 @@ import { closeSync, + createReadStream, createWriteStream, existsSync, fsyncSync, + linkSync, mkdtempSync, mkdirSync, lstatSync, openSync, - renameSync, rmSync, statSync, + unlinkSync, } from "fs"; import { createHash } 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 +42,92 @@ 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"; + etagFingerprint?: string; + etagWeak?: boolean; + localSize?: number; + localSha256?: string; + failureKind?: UrlDownloadFailureKind; +} + +export interface UrlDownloadOptions { + /** Require a recognized media container/file signature. Content-Type remains advisory. */ + expectedContent?: "media" | "opaque"; + /** Optional strong checksum supplied by a trusted caller. */ + expectedSha256?: string; + onTelemetry?: (event: UrlDownloadTelemetry) => void; +} + +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, ) { 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 +139,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,22 +231,19 @@ 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 === "opaque\0" ? 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}`; @@ -228,7 +304,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 +315,200 @@ 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); } } +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"; + } + return response.status === 200 && contentRange !== null ? "content_range_on_200" : "none"; +} + +function looksLikeHtmlErrorBody(prefix: Buffer): boolean { + const text = prefix + .toString("utf8") + .replace(/^\uFEFF?\s*/, "") + .toLowerCase(); + return ( + text.startsWith(" prefix.toString("ascii", start, end); + if ( + prefix.length >= 8 && + prefix.subarray(0, 8).equals(Buffer.from([137, 80, 78, 71, 13, 10, 26, 10])) + ) { + return true; + } + if (prefix[0] === 0xff && prefix[1] === 0xd8 && prefix[2] === 0xff) return true; + if (prefix.length >= 6 && (ascii(0, 6) === "GIF87a" || ascii(0, 6) === "GIF89a")) return true; + if (prefix.length >= 12 && ascii(0, 4) === "RIFF") { + const kind = ascii(8, 12); + if (kind === "WEBP" || kind === "WAVE" || kind === "AVI ") return true; + } + if (prefix.length >= 4) { + const firstFour = ascii(0, 4); + if (firstFour === "OggS" || firstFour === "fLaC") return true; + if (prefix.subarray(0, 4).equals(Buffer.from([0x1a, 0x45, 0xdf, 0xa3]))) return true; + if (prefix.subarray(0, 4).equals(Buffer.from([0x76, 0x2f, 0x31, 0x01]))) return true; + if (prefix.subarray(0, 4).equals(Buffer.from([0x00, 0x00, 0x01, 0xba]))) return true; + } + if (ascii(0, 3) === "ID3") return prefix.length >= 10 && localSize >= 10; + if (prefix[0] === 0xff && (prefix[1]! & 0xe0) === 0xe0) return true; + if (ascii(0, 2) === "BM") return prefix.length >= 14; + if ( + prefix.length >= 4 && + (prefix.subarray(0, 4).equals(Buffer.from([0x49, 0x49, 0x2a, 0x00])) || + prefix.subarray(0, 4).equals(Buffer.from([0x4d, 0x4d, 0x00, 0x2a])) || + prefix.subarray(0, 4).equals(Buffer.from([0x00, 0x00, 0x01, 0x00]))) + ) { + return true; + } + if ( + prefix.length >= 12 && + prefix + .subarray(0, 12) + .equals(Buffer.from([0, 0, 0, 12, 0x4a, 0x58, 0x4c, 0x20, 0x0d, 0x0a, 0x87, 0x0a])) + ) { + return true; + } + if (prefix[0] === 0xff && prefix[1] === 0x0a) return true; + const hasTransportStreamSync = (offset: number, stride: number): boolean => + prefix.length > offset + stride * 2 && + prefix[offset] === 0x47 && + prefix[offset + stride] === 0x47 && + prefix[offset + stride * 2] === 0x47; + if (hasTransportStreamSync(0, 188) || hasTransportStreamSync(4, 192)) return true; + const mxfHeaderPartitionPrefix = Buffer.from([ + 0x06, 0x0e, 0x2b, 0x34, 0x02, 0x05, 0x01, 0x01, 0x0d, 0x01, 0x02, 0x01, 0x01, 0x02, + ]); + if ( + prefix.length >= 16 && + prefix.subarray(0, mxfHeaderPartitionPrefix.length).equals(mxfHeaderPartitionPrefix) && + prefix[15] === 0x00 + ) { + return true; + } + if (prefix.length >= 16 && ascii(4, 8) === "ftyp") { + const boxSize = prefix.readUInt32BE(0); + return boxSize >= 16 && boxSize <= localSize && /^[\x20-\x7e]{4}$/.test(ascii(8, 12)); + } + const text = prefix.toString("utf8").replace(/^\uFEFF?\s*/, ""); + return /^)/i.test(text) || (/^<\?xml\b/i.test(text) && /)/i.test(text)); +} + +function expectedResponseSha256(response: Response, options: UrlDownloadOptions): string | null { + if (options.expectedSha256) return options.expectedSha256.trim().toLowerCase(); + const amazonChecksum = response.headers.get("x-amz-checksum-sha256")?.trim(); + const amazonChecksumType = response.headers.get("x-amz-checksum-type")?.trim().toUpperCase(); + if (amazonChecksum && amazonChecksumType !== "COMPOSITE") return `base64:${amazonChecksum}`; + const digest = response.headers.get("digest"); + const sha256Digest = digest?.match(/(?:^|,)\s*sha-256=:?([^,:\s]+):?/i)?.[1]; + return sha256Digest ? `base64:${sha256Digest}` : null; +} + +// 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") { + 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 +517,154 @@ 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 ( + options.expectedContent === "media" && + (looksLikeHtmlErrorBody(prefix) || !hasRecognizedMediaSignature(prefix, localSize)) + ) { + throw new UrlDownloadError( + "invalid_payload", + false, + "Download returned a payload without a recognized media signature", + response.status, + telemetry, + ); + } + + const expectedSha256 = expectedResponseSha256(response, options); + const checksumMatches = + expectedSha256 === null || + (expectedSha256.startsWith("base64:") + ? sha256Bytes.toString("base64") === expectedSha256.slice("base64:".length) + : /^[a-f0-9]{64}$/.test(expectedSha256) && localSha256 === expectedSha256); + const contentMd5 = response.headers.get("content-md5")?.trim(); + if (!checksumMatches || (contentMd5 && md5Base64 !== contentMd5)) { + throw new UrlDownloadError( + "hash_mismatch", + true, + "Download payload checksum did not match", + 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 +674,36 @@ 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" || !hasCompleteFile(localPath)) 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 +722,83 @@ 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") { + const inspection = await inspectExistingFile(localPath); + publishedIntegrity = { + ...integrity, + localSize: inspection.localSize, + localSha256: inspection.localSha256, + }; + if (!localInspectionMatchesOptions(inspection, options)) { + const invalidMedia = + options.expectedContent === "media" && + (looksLikeHtmlErrorBody(inspection.prefix) || + !hasRecognizedMediaSignature(inspection.prefix, inspection.localSize)); + throw new UrlDownloadError( + invalidMedia ? "invalid_payload" : "hash_mismatch", + !invalidMedia, + invalidMedia + ? "Concurrent download published a payload without a recognized media signature" + : "Concurrent download payload checksum did not match", + 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,25 +813,78 @@ 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); + 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 && + !( + options.expectedContent === "media" && + (looksLikeHtmlErrorBody(inspection.prefix) || + !hasRecognizedMediaSignature(inspection.prefix, inspection.localSize)) + ) && + (!expectedSha256 || inspection.localSha256 === expectedSha256) + ); +} + 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 @@ -377,7 +895,8 @@ export async function downloadToTemp( // 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 = `${options.expectedContent ?? "opaque"}\0${options.expectedSha256?.trim().toLowerCase() ?? ""}`; + const inFlightKey = `${cacheKey}\0${timeoutMs}\0${signalScopeKey(signal)}\0${validationScope}`; const inFlight = inFlightDownloads.get(inFlightKey); if (inFlight) { return inFlight; @@ -387,12 +906,37 @@ 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; + if (hasCompleteFile(localPath)) { + const identity = safeDownloadUrlIdentity(url); + const inspection = await inspectExistingFile(localPath); + const cacheValid = localInspectionMatchesOptions(inspection, options); + if (cacheValid) { + 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 localPath; + } + rmSync(localPath, { force: true }); + } - const downloadPromise = downloadWithRetry(url, localPath, timeoutMs, signal, onTransientRetry); + const downloadPromise = downloadWithRetry( + url, + localPath, + timeoutMs, + signal, + onTransientRetry, + options, + ); 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 8e24f92c9c..e116820d69 100644 --- a/packages/producer/src/services/htmlCompiler.test.ts +++ b/packages/producer/src/services/htmlCompiler.test.ts @@ -23,6 +23,22 @@ import { } from "./htmlCompiler.js"; import { validateNoSystemFonts } from "./render/planValidation.js"; +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 = @@ -1362,7 +1378,7 @@ describe("localizeRemoteMediaSources", () => { it("rewrites remote