diff --git a/packages/engine/src/services/videoFrameExtractor.ts b/packages/engine/src/services/videoFrameExtractor.ts index ec0cedf2fe..953b3433f8 100644 --- a/packages/engine/src/services/videoFrameExtractor.ts +++ b/packages/engine/src/services/videoFrameExtractor.ts @@ -730,7 +730,7 @@ export async function extractAllVideoFrames( if (isHttpUrl(videoPath)) { const downloadDir = join(options.outputDir, "_downloads"); mkdirSync(downloadDir, { recursive: true }); - videoPath = await downloadToTemp(videoPath, downloadDir); + videoPath = await downloadToTemp(videoPath, downloadDir, undefined, signal); } if (!existsSync(videoPath)) { diff --git a/packages/engine/src/utils/urlDownloader.test.ts b/packages/engine/src/utils/urlDownloader.test.ts index 22fae4f93a..48a18e5334 100644 --- a/packages/engine/src/utils/urlDownloader.test.ts +++ b/packages/engine/src/utils/urlDownloader.test.ts @@ -1,5 +1,38 @@ -import { describe, expect, it } from "vitest"; -import { assertPublicHttpsUrl } from "./urlDownloader.js"; +// fallow-ignore-file code-duplication +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + existsSync, + mkdtempSync, + readFileSync, + readdirSync, + rmSync, + symlinkSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { assertPublicHttpsUrl, downloadToTemp, UrlDownloadError } from "./urlDownloader.js"; + +const tempDirs: string[] = []; + +function makeTempDir(): string { + const dir = mkdtempSync(join(tmpdir(), "hf-url-download-")); + tempDirs.push(dir); + return dir; +} + +function temporaryDownloadEntries(dir: string): string[] { + return readdirSync(dir).filter( + (name) => name.includes(".partial-") || name.startsWith(".hf-download-"), + ); +} + +afterEach(() => { + vi.unstubAllGlobals(); + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); describe("assertPublicHttpsUrl — SSRF guard", () => { it("accepts public HTTPS URLs", () => { @@ -58,8 +91,450 @@ describe("assertPublicHttpsUrl — SSRF guard", () => { expect(() => assertPublicHttpsUrl("https://[::1]/secret")).toThrow("private/reserved"); }); + it("rejects normalized reserved IPv6 and IPv4-mapped forms", () => { + for (const url of [ + "https://[::]/secret", + "https://[fe80::1]/secret", + "https://[fc00::1]/secret", + "https://[::ffff:127.0.0.1]/secret", + "https://[::ffff:169.254.169.254]/latest/meta-data/", + ]) { + expect(() => assertPublicHttpsUrl(url), url).toThrow("private/reserved"); + } + }); + + it("rejects CGNAT and other non-public IPv4 ranges", () => { + for (const url of [ + "https://100.64.0.1/secret", + "https://198.18.0.1/secret", + "https://192.0.2.1/secret", + "https://224.0.0.1/secret", + "https://240.0.0.1/secret", + "https://255.255.255.255/secret", + ]) { + expect(() => assertPublicHttpsUrl(url), url).toThrow("private/reserved"); + } + }); + it("rejects invalid URLs", () => { expect(() => assertPublicHttpsUrl("not-a-url")).toThrow("Invalid URL"); expect(() => assertPublicHttpsUrl("")).toThrow("Invalid URL"); }); }); + +describe("downloadToTemp atomic publication and bounded retry", () => { + it("follows a bounded redirect only after validating the next public HTTPS hop", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + new Response(null, { + status: 302, + headers: { location: "https://media.example/final.mp4" }, + }), + ) + .mockResolvedValueOnce(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/redirect.mp4", dir, 1_000); + + expect(fetchMock).toHaveBeenNthCalledWith( + 2, + "https://media.example/final.mp4", + expect.objectContaining({ redirect: "manual" }), + ); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("rejects a public redirect to a private host before issuing the second request", async () => { + const fetchMock = vi.fn().mockResolvedValueOnce( + new Response(null, { + status: 302, + headers: { location: "http://169.254.169.254/latest/meta-data/" }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/private-redirect.mp4", dir, 1_000), + ).rejects.toMatchObject({ + kind: "http_rejected", + retryable: false, + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("rejects an IPv4-mapped IMDS redirect before issuing the second request", async () => { + const fetchMock = vi.fn().mockResolvedValueOnce( + new Response(null, { + status: 302, + headers: { location: "https://[::ffff:169.254.169.254]/latest/meta-data/" }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/mapped-private-redirect.mp4", dir, 1_000), + ).rejects.toMatchObject({ + kind: "http_rejected", + retryable: false, + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("retries one HTTP 503 and publishes only the complete final file", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(null, { status: 503, statusText: "Service Unavailable" })) + .mockResolvedValueOnce(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/retry-503.mp4", dir, 1_000); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("cancels a streaming HTTP error body before retrying", async () => { + let errorBodyCancelled = false; + const errorBody = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("error details")); + }, + cancel() { + errorBodyCancelled = true; + }, + }); + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + new Response(errorBody, { status: 503, statusText: "Service Unavailable" }), + ) + .mockResolvedValueOnce(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/streaming-503.mp4", dir, 1_000); + + expect(errorBodyCancelled).toBe(true); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("does not retry a deterministic 404", async () => { + const fetchMock = vi + .fn() + .mockResolvedValue(new Response(null, { status: 404, statusText: "Not Found" })); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/missing-404.mp4", dir, 1_000), + ).rejects.toMatchObject({ + kind: "http_not_found", + retryable: false, + status: 404, + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("exhausts the transient retry budget after exactly one retry", async () => { + const fetchMock = vi + .fn() + .mockResolvedValue(new Response(null, { status: 503, statusText: "Service Unavailable" })); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + await expect( + downloadToTemp("https://cdn.example/always-503.mp4", dir, 1_000), + ).rejects.toMatchObject({ + kind: "http_transient", + retryable: true, + status: 503, + } satisfies Partial); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("removes a partial body after a network reset before retrying", async () => { + const resetBody = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("partial")); + const error = Object.assign(new Error("socket reset"), { code: "ECONNRESET" }); + controller.error(error); + }, + }); + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(resetBody)) + .mockResolvedValueOnce(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/reset-once.mp4", dir, 1_000); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("retries an Undici mid-body disconnect reported through a nested cause", async () => { + const disconnectedBody = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("partial")); + const cause = Object.assign(new Error("other side closed"), { + code: "UND_ERR_SOCKET", + }); + controller.error(new TypeError("terminated", { cause })); + }, + }); + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(disconnectedBody)) + .mockResolvedValueOnce(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/undici-reset-once.mp4", dir, 1_000); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("keeps the deadline active through a stalled response body", async () => { + const fetchMock = vi + .fn() + .mockImplementationOnce(async (_url: string, init: RequestInit) => { + const stalledBody = 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(stalledBody); + }) + .mockResolvedValueOnce(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/stalled-body.mp4", dir, 20); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("retries a zero-byte 200 response without publishing it", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response("")) + .mockResolvedValueOnce(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/empty-once.mp4", dir, 1_000); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("retries a 200 response with no body", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(null)) + .mockResolvedValueOnce(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const path = await downloadToTemp("https://cdn.example/null-body-once.mp4", dir, 1_000); + + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("removes a stale zero-byte final file before downloading", async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response("complete")); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + const stalePath = join(dir, "download_eda0de5dc5a3.mp4"); + writeFileSync(stalePath, ""); + + const path = await downloadToTemp("https://cdn.example/stale-empty.mp4", dir, 1_000); + + expect(path).toBe(stalePath); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(readFileSync(path, "utf8")).toBe("complete"); + 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); + const dir = makeTempDir(); + const target = join(dir, "attacker-controlled.mp4"); + const cachePath = join(dir, "download_eda0de5dc5a3.mp4"); + writeFileSync(target, "not-the-download"); + symlinkSync(target, cachePath); + + const path = await downloadToTemp("https://cdn.example/stale-empty.mp4", dir, 1_000); + + expect(path).toBe(cachePath); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(readFileSync(path, "utf8")).toBe("downloaded"); + expect(readFileSync(target, "utf8")).toBe("not-the-download"); + }); + + it("does not retry caller cancellation", async () => { + const fetchMock = vi.fn(); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + const controller = new AbortController(); + controller.abort(); + + await expect( + downloadToTemp("https://cdn.example/cancelled.mp4", dir, 1_000, controller.signal), + ).rejects.toMatchObject({ + kind: "cancelled", + retryable: false, + } satisfies Partial); + expect(fetchMock).not.toHaveBeenCalled(); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("deduplicates concurrent callers for the same URL and destination", async () => { + const fetchMock = vi.fn().mockImplementation( + () => + new Promise((resolve) => { + setTimeout(() => resolve(new Response("complete")), 10); + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const [first, second] = await Promise.all([ + downloadToTemp("https://cdn.example/concurrent.mp4", dir, 1_000), + downloadToTemp("https://cdn.example/concurrent.mp4", dir, 1_000), + ]); + + expect(first).toBe(second); + expect(existsSync(first)).toBe(true); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("does not let one caller cancellation abort another caller", async () => { + const firstController = new AbortController(); + const secondController = new AbortController(); + const fetchMock = vi + .fn() + .mockImplementationOnce(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); + }) + .mockImplementationOnce( + () => + new Promise((resolve) => { + setTimeout(() => resolve(new Response("complete")), 10); + }), + ); + vi.stubGlobal("fetch", fetchMock); + const dir = makeTempDir(); + + const first = downloadToTemp( + "https://cdn.example/cancellation-isolation-a.mp4", + dir, + 1_000, + firstController.signal, + ); + const second = downloadToTemp( + "https://cdn.example/cancellation-isolation-a.mp4", + dir, + 1_000, + secondController.signal, + ); + firstController.abort(); + + await expect(first).rejects.toMatchObject({ + kind: "cancelled", + retryable: false, + } satisfies Partial); + const path = await second; + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); + + it("does not let a later caller cancellation abort the first caller", async () => { + const firstController = new AbortController(); + const secondController = new AbortController(); + const fetchMock = vi + .fn() + .mockImplementationOnce( + () => + new Promise((resolve) => { + setTimeout(() => resolve(new Response("complete")), 10); + }), + ) + .mockImplementationOnce(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); + const dir = makeTempDir(); + + const first = downloadToTemp( + "https://cdn.example/cancellation-isolation-b.mp4", + dir, + 1_000, + firstController.signal, + ); + const second = downloadToTemp( + "https://cdn.example/cancellation-isolation-b.mp4", + dir, + 1_000, + secondController.signal, + ); + secondController.abort(); + + await expect(second).rejects.toMatchObject({ + kind: "cancelled", + retryable: false, + } satisfies Partial); + const path = await first; + expect(readFileSync(path, "utf8")).toBe("complete"); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(temporaryDownloadEntries(dir)).toEqual([]); + }); +}); diff --git a/packages/engine/src/utils/urlDownloader.ts b/packages/engine/src/utils/urlDownloader.ts index 3dbd65321d..8e25966c69 100644 --- a/packages/engine/src/utils/urlDownloader.ts +++ b/packages/engine/src/utils/urlDownloader.ts @@ -1,42 +1,145 @@ -import { createWriteStream, existsSync, mkdirSync } from "fs"; +import { + closeSync, + createWriteStream, + existsSync, + fsyncSync, + mkdtempSync, + mkdirSync, + lstatSync, + openSync, + renameSync, + rmSync, + statSync, +} from "fs"; import { createHash } from "crypto"; -import { join, extname } from "path"; +import { BlockList, isIP } from "node:net"; +import { dirname, extname, join } from "path"; import { Readable } from "stream"; -import { finished } from "stream/promises"; +import { pipeline } from "stream/promises"; -const downloadPathCache = new Map(); const inFlightDownloads = new Map>(); +const signalScopes = new WeakMap(); +let nextSignalScope = 1; -// SSRF guard: these prefixes identify non-public address space that -// compositions (customer-supplied) must never be able to reach via the -// download path. Blocks AWS IMDS (169.254.169.254), loopback, RFC1918, -// and unspecified addresses. All comparisons are on the raw hostname -// string; DNS resolution is NOT performed here, so DNS-rebinding bypasses -// are not closed by this check — that gap is acceptable for the risk level. -const BLOCKED_HOST_PREFIXES = [ - "169.254.", // link-local / AWS IMDS - "127.", // loopback IPv4 - "10.", // RFC1918 - "192.168.", // RFC1918 - "0.", // unspecified - "[::1]", // loopback IPv6 - "[fc", // RFC4193 unique-local IPv6 - "[fd", // RFC4193 unique-local IPv6 -]; -// 172.16.0.0 – 172.31.255.255 (RFC1918) -const BLOCKED_172_RANGE = { min: 16, max: 31 }; +function signalScopeKey(signal: AbortSignal | undefined): string { + if (!signal) return "none"; + let scope = signalScopes.get(signal); + if (scope === undefined) { + scope = nextSignalScope; + nextSignalScope += 1; + signalScopes.set(signal, scope); + } + return String(scope); +} + +export type UrlDownloadFailureKind = + | "cancelled" + | "timeout" + | "http_not_found" + | "http_rejected" + | "http_transient" + | "network" + | "empty_body" + | "filesystem"; + +export class UrlDownloadError extends Error { + constructor( + readonly kind: UrlDownloadFailureKind, + readonly retryable: boolean, + message: string, + readonly status?: number, + ) { + super(message); + this.name = "UrlDownloadError"; + } +} + +function classifyHttpFailure(status: number, statusText: string): UrlDownloadError { + const message = `HTTP ${status}: ${statusText}`; + if (status === 404 || status === 410) { + return new UrlDownloadError("http_not_found", false, message, status); + } + if (status === 408 || status === 429 || status >= 500) { + return new UrlDownloadError("http_transient", true, message, status); + } + return new UrlDownloadError("http_rejected", false, message, status); +} + +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}`); + } + current = + typeof current === "object" && current !== null && "cause" in current + ? current.cause + : undefined; + } + return new UrlDownloadError("filesystem", false, `Download failed: ${message}`); +} + +const RETRYABLE_NETWORK_CODES = new Set(["ECONNRESET", "ECONNREFUSED", "ETIMEDOUT", "EAI_AGAIN"]); + +function isRetryableNetworkCause(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error); + const code = + typeof error === "object" && error !== null && "code" in error && typeof error.code === "string" + ? error.code + : ""; + return ( + RETRYABLE_NETWORK_CODES.has(code) || + code.startsWith("UND_ERR_") || + /fetch failed|network|socket|connection reset|terminated/i.test(message) + ); +} + +const NON_PUBLIC_IPV4_ADDRESSES = new BlockList(); +for (const [network, prefix] of [ + ["0.0.0.0", 8], + ["10.0.0.0", 8], + ["100.64.0.0", 10], + ["127.0.0.0", 8], + ["169.254.0.0", 16], + ["172.16.0.0", 12], + ["192.0.0.0", 24], + ["192.0.2.0", 24], + ["192.88.99.0", 24], + ["192.168.0.0", 16], + ["198.18.0.0", 15], + ["198.51.100.0", 24], + ["203.0.113.0", 24], + ["224.0.0.0", 4], + ["240.0.0.0", 4], +] as const) { + NON_PUBLIC_IPV4_ADDRESSES.addSubnet(network, prefix, "ipv4"); +} +const NON_PUBLIC_IPV6_ADDRESSES = new BlockList(); +for (const [network, prefix] of [ + ["::", 128], + ["::1", 128], + ["::ffff:0:0", 96], + ["fc00::", 7], + ["fe80::", 10], + ["fec0::", 10], + ["ff00::", 8], + ["2001:db8::", 32], +] as const) { + NON_PUBLIC_IPV6_ADDRESSES.addSubnet(network, prefix, "ipv6"); +} function isBlockedHost(hostname: string): boolean { - const h = hostname.toLowerCase(); + const h = hostname.toLowerCase().replace(/^\[|\]$/g, ""); if (h === "localhost") return true; - if (BLOCKED_HOST_PREFIXES.some((p) => h.startsWith(p))) return true; - // 172.16–172.31 - const m = h.match(/^172\.(\d{1,3})\./); - if (m) { - const octet = parseInt(m[1] ?? "0", 10); - if (octet >= BLOCKED_172_RANGE.min && octet <= BLOCKED_172_RANGE.max) return true; - } - return false; + const addressType = isIP(h); + if (addressType === 0) return false; + return addressType === 4 + ? NON_PUBLIC_IPV4_ADDRESSES.check(h, "ipv4") + : NON_PUBLIC_IPV6_ADDRESSES.check(h, "ipv6"); } /** @@ -70,21 +173,209 @@ function getFilenameFromUrl(url: string): string { 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; +} + +const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]); +const MAX_REDIRECTS = 5; + +function assertAllowedDownloadUrl(url: string, redirect: boolean): void { + try { + assertPublicHttpsUrl(url); + } catch { + throw new UrlDownloadError( + "http_rejected", + false, + redirect ? "Download redirect target is not permitted" : "Download URL is not permitted", + ); + } +} + +async function cancelResponseBody(response: Response): Promise { + try { + await response.body?.cancel(); + } catch { + // Redirect validation remains authoritative if teardown also fails. + } +} + +function resolveRedirectUrl(response: Response, currentUrl: string, redirects: number): string { + if (redirects >= MAX_REDIRECTS) { + throw new UrlDownloadError("http_rejected", false, "Download exceeded redirect limit"); + } + const location = response.headers.get("location"); + if (!location) { + throw new UrlDownloadError( + "http_rejected", + false, + "Download redirect omitted a Location header", + ); + } + try { + return new URL(location, currentUrl).toString(); + } catch { + throw new UrlDownloadError("http_rejected", false, "Download redirect Location is invalid"); + } +} + +async function fetchWithValidatedRedirects( + initialUrl: string, + controller: AbortController, +): Promise { + let currentUrl = initialUrl; + for (let redirects = 0; ; redirects += 1) { + assertAllowedDownloadUrl(currentUrl, redirects > 0); + + // lgtm[js/file-access-to-http] — every redirect hop is fetched manually + // only after the HTTPS/private-host guard above; automatic redirect + // following is disabled so an allowed host cannot bounce into IMDS. + const response = await fetch(currentUrl, { + signal: controller.signal, + redirect: "manual", + }); + if (!REDIRECT_STATUSES.has(response.status)) return response; + + await cancelResponseBody(response); + currentUrl = resolveRedirectUrl(response, currentUrl, redirects); + } +} + +async function fetchToPartial( + url: string, + partialPath: string, + controller: AbortController, +): Promise { + const response = await fetchWithValidatedRedirects(url, controller); + if (!response.ok) { + // Do not leave a streaming error response holding an Undici connection + // while the bounded retry starts. + try { + await response.body?.cancel(); + } catch { + // The HTTP status remains the useful failure if teardown also fails. + } + throw classifyHttpFailure(response.status, response.statusText); + } + if (!response.body) { + throw new UrlDownloadError("empty_body", true, "Download response body is empty"); + } + + const fileStream = createWriteStream(partialPath, { flags: "wx" }); + // 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"); + } +} + +function syncAndPublishPartial(partialPath: string, localPath: string): void { + // 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+"); + try { + fsyncSync(fd); + } finally { + 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; + try { + renameSync(partialPath, localPath); + } catch (error) { + if (!hasCompleteFile(localPath)) throw error; + } +} + +async function runDownloadAttempt( + url: string, + localPath: string, + timeoutMs: number, + signal?: AbortSignal, +): Promise { + // A private, unguessable directory prevents symlink planting and keeps the + // partial on the destination filesystem so the final rename stays atomic. + const attemptDir = mkdtempSync(join(dirname(localPath), ".hf-download-")); + const partialPath = join(attemptDir, "payload"); + const controller = new AbortController(); + let timedOut = false; + let callerAborted = signal?.aborted ?? false; + const onCallerAbort = (): void => { + callerAborted = true; + controller.abort(); + }; + signal?.addEventListener("abort", onCallerAbort, { once: true }); + const timeoutId = setTimeout(() => { + timedOut = true; + controller.abort(); + }, timeoutMs); + + try { + if (callerAborted) { + throw new UrlDownloadError("cancelled", false, "Download cancelled"); + } + await fetchToPartial(url, partialPath, controller); + syncAndPublishPartial(partialPath, localPath); + return localPath; + } catch (error) { + if (callerAborted) { + throw new UrlDownloadError("cancelled", false, "Download cancelled"); + } + if (timedOut) { + throw new UrlDownloadError("timeout", true, `Download timeout after ${timeoutMs / 1000}s`); + } + throw classifyDownloadFailure(error); + } finally { + clearTimeout(timeoutId); + signal?.removeEventListener("abort", onCallerAbort); + controller.abort(); + rmSync(attemptDir, { recursive: true, force: true }); + } +} + +async function downloadWithRetry( + url: string, + localPath: string, + timeoutMs: number, + signal?: AbortSignal, +): Promise { + const maxTransientRetries = 1; + for (let attempt = 0; ; attempt += 1) { + try { + return await runDownloadAttempt(url, localPath, timeoutMs, signal); + } catch (error) { + const classified = classifyDownloadFailure(error); + if (!classified.retryable || attempt >= maxTransientRetries) throw classified; + } + } +} + export async function downloadToTemp( url: string, destDir: string, timeoutMs: number = 300000, + signal?: AbortSignal, ): 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 cachedPath = downloadPathCache.get(url); - if (cachedPath && existsSync(cachedPath)) { - return cachedPath; - } - const inFlight = inFlightDownloads.get(url); + 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 inFlight = inFlightDownloads.get(inFlightKey); if (inFlight) { return inFlight; } @@ -96,46 +387,14 @@ export async function downloadToTemp( const filename = getFilenameFromUrl(url); const localPath = join(destDir, filename); - if (existsSync(localPath)) { - downloadPathCache.set(url, localPath); - return localPath; - } + if (hasCompleteFile(localPath)) return localPath; - const downloadPromise = (async () => { - try { - const controller = new AbortController(); - const timeoutId = setTimeout(() => controller.abort(), timeoutMs); - - const response = await fetch(url, { signal: controller.signal }); - clearTimeout(timeoutId); - - if (!response.ok) { - throw new Error(`HTTP ${response.status}: ${response.statusText}`); - } - - if (!response.body) { - throw new Error("Response body is empty"); - } - - const fileStream = createWriteStream(localPath); - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const readableStream = Readable.fromWeb(response.body as any); - await finished(readableStream.pipe(fileStream)); - - downloadPathCache.set(url, localPath); - return localPath; - } catch (err) { - const message = err instanceof Error ? err.message : String(err); - if (message.includes("aborted")) { - throw new Error(`[URLDownloader] Download timeout after ${timeoutMs / 1000}s: ${url}`); - } - throw new Error(`[URLDownloader] Download failed: ${message}`); - } finally { - inFlightDownloads.delete(url); - } - })(); - inFlightDownloads.set(url, downloadPromise); - return downloadPromise; + const downloadPromise = downloadWithRetry(url, localPath, timeoutMs, signal); + const trackedDownload = downloadPromise.finally(() => { + inFlightDownloads.delete(inFlightKey); + }); + inFlightDownloads.set(inFlightKey, trackedDownload); + return trackedDownload; } export function isHttpUrl(path: string): boolean {