diff --git a/apps/media-server/bunfig.toml b/apps/media-server/bunfig.toml new file mode 100644 index 00000000000..37ff1f94820 --- /dev/null +++ b/apps/media-server/bunfig.toml @@ -0,0 +1,2 @@ +[test] +preload = ["./src/__tests__/setup.ts"] diff --git a/apps/media-server/src/__tests__/lib/job-manager.test.ts b/apps/media-server/src/__tests__/lib/job-manager.test.ts index 5bada4e694c..bc1e06a4cc9 100644 --- a/apps/media-server/src/__tests__/lib/job-manager.test.ts +++ b/apps/media-server/src/__tests__/lib/job-manager.test.ts @@ -156,8 +156,10 @@ describe("durable recording workers", () => { const sending = sendWebhook(job); updateJob(job.jobId, { phase: "complete", progress: 100 }); expect(sendWebhook(job)).toBe(sending); + // The URL policy check runs before fetch, so wait for the request to start. + while (!release) await Bun.sleep(1); const started = performance.now(); - release?.(new Response(null, { status: 503 })); + release(new Response(null, { status: 503 })); await sending; expect(performance.now() - started).toBeLessThan(400); expect(maximumInFlight).toBe(1); diff --git a/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts b/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts index 44cd8698342..2c3871341e7 100644 --- a/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts @@ -1,5 +1,5 @@ import { describe, expect, spyOn, test } from "bun:test"; -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { @@ -14,8 +14,15 @@ const FIXTURES_DIR = join(import.meta.dir, "..", "fixtures"); const TEST_VIDEO_WITH_AUDIO = `file://${join(FIXTURES_DIR, "test-with-audio.mp4")}`; const TEST_VIDEO_NO_AUDIO = `file://${join(FIXTURES_DIR, "test-no-audio.mp4")}`; -async function createHlsFixture(): Promise { +async function createHlsFixture(encrypted = false): Promise { const dirPath = await mkdtemp(join(tmpdir(), "cap-audio-hls-")); + if (encrypted) { + await writeFile(join(dirPath, "key.bin"), Buffer.alloc(16, 7)); + await writeFile( + join(dirPath, "key.info"), + `key.bin\n${join(dirPath, "key.bin")}\n`, + ); + } const proc = Bun.spawn({ cmd: [ "ffmpeg", @@ -31,6 +38,7 @@ async function createHlsFixture(): Promise { "0.25", "-hls_list_size", "0", + ...(encrypted ? ["-hls_key_info_file", join(dirPath, "key.info")] : []), "-hls_segment_filename", join(dirPath, "segment-%03d.ts"), join(dirPath, "manifest.m3u8"), @@ -202,7 +210,7 @@ describe("mediaAudio integration tests", () => { try { await expect( checkHasAudioTrack(`${server.url}video.mp4`), - ).rejects.toThrow("FFprobe exited"); + ).rejects.toThrow("Failed to fetch media: 503"); await waitForAudioOperations(beforeCount); } finally { await server.stop(true); @@ -226,7 +234,9 @@ describe("mediaAudio integration tests", () => { describe("extractAudioStream", () => { test("preserves queued audio when the reader pauses before EOF", async () => { const expected = await extractAudio(TEST_VIDEO_WITH_AUDIO); - const { stream, cleanup } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream, cleanup } = await extractAudioStream( + TEST_VIDEO_WITH_AUDIO, + ); const reader = stream.getReader(); const chunks: Uint8Array[] = []; try { @@ -246,7 +256,9 @@ describe("mediaAudio integration tests", () => { test("drains all encoded bytes when the consumer waits for FFmpeg to exit", async () => { const expected = await extractAudio(TEST_VIDEO_WITH_AUDIO); const registration = spyOn(subprocess, "registerSubprocess"); - const { stream, cleanup } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream, cleanup } = await extractAudioStream( + TEST_VIDEO_WITH_AUDIO, + ); try { const proc = registration.mock.calls[0][0]; expect(await proc.exited).toBe(0); @@ -261,7 +273,7 @@ describe("mediaAudio integration tests", () => { test("errors the stream when FFmpeg cannot extract audio", async () => { const beforeCount = getActiveProcessCount(); - const { stream, cleanup } = extractAudioStream(TEST_VIDEO_NO_AUDIO); + const { stream, cleanup } = await extractAudioStream(TEST_VIDEO_NO_AUDIO); try { await expect(readStream(stream)).rejects.toThrow("FFmpeg exited"); await waitForAudioOperations(beforeCount); @@ -273,9 +285,12 @@ describe("mediaAudio integration tests", () => { test("errors the stream and terminates FFmpeg at the deadline", async () => { const beforeCount = getActiveProcessCount(); const registration = spyOn(subprocess, "registerSubprocess"); - const { stream, cleanup } = extractAudioStream(TEST_VIDEO_WITH_AUDIO, { - timeoutMs: 1, - }); + const { stream, cleanup } = await extractAudioStream( + TEST_VIDEO_WITH_AUDIO, + { + timeoutMs: 1, + }, + ); try { await expect(readStream(stream)).rejects.toThrow("timed out"); const proc = registration.mock.calls[0][0]; @@ -289,7 +304,7 @@ describe("mediaAudio integration tests", () => { test("streams real mp3 data and cleans up its operation", async () => { const beforeCount = getActiveProcessCount(); - const { stream } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream } = await extractAudioStream(TEST_VIDEO_WITH_AUDIO); const audioData = await readStream(stream); @@ -297,9 +312,26 @@ describe("mediaAudio integration tests", () => { await waitForAudioOperations(beforeCount); }); + test("releases the operation when a remote download fails", async () => { + const beforeCount = getActiveProcessCount(); + const server = Bun.serve({ + port: 0, + fetch: () => new Response("Missing", { status: 404 }), + }); + + try { + await expect( + extractAudioStream(`${server.url}clip.mp4`), + ).rejects.toThrow("Failed to fetch media: 404"); + await waitForAudioOperations(beforeCount); + } finally { + await server.stop(true); + } + }); + test("cleans up when a real stream is cancelled", async () => { const beforeCount = getActiveProcessCount(); - const { stream } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream } = await extractAudioStream(TEST_VIDEO_WITH_AUDIO); await stream.cancel(); @@ -307,3 +339,27 @@ describe("mediaAudio integration tests", () => { }); }); }); + +describe("encrypted remote HLS audio", () => { + test("extracts MP3 audio after staging AES-128 keys and segments locally", async () => { + const directory = await createHlsFixture(true); + const server = Bun.serve({ + port: 0, + fetch(request) { + return new Response( + Bun.file(join(directory, new URL(request.url).pathname)), + ); + }, + }); + try { + const url = new URL("manifest.m3u8", server.url).toString(); + expect(await checkHasAudioTrack(url)).toBe(true); + expectMp3(await extractAudio(url)); + const { stream } = await extractAudioStream(url); + expectMp3(await readStream(stream)); + } finally { + server.stop(true); + await rm(directory, { recursive: true, force: true }); + } + }); +}); diff --git a/apps/media-server/src/__tests__/lib/media-download-policy.integration.test.ts b/apps/media-server/src/__tests__/lib/media-download-policy.integration.test.ts new file mode 100644 index 00000000000..d0bf1ba79ac --- /dev/null +++ b/apps/media-server/src/__tests__/lib/media-download-policy.integration.test.ts @@ -0,0 +1,239 @@ +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; +import { mkdtemp, readdir, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { withMediaTransfers } from "../../lib/media-transfer"; +import { + materializeHlsPlaylist, + materializeMpdAsHlsPlaylist, + materializeMpdManifest, +} from "../../lib/media-video"; + +let directory: string; +let server: ReturnType | undefined; +const savedEnv = { ...process.env }; + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), "cap-download-policy-")); +}); + +afterEach(async () => { + server?.stop(true); + server = undefined; + await rm(directory, { recursive: true, force: true }); + for (const key of [ + "MEDIA_SERVER_MAX_PLAYLIST_RESOURCES", + "MEDIA_SERVER_MAX_PLAYLIST_BYTES", + ]) { + if (savedEnv[key] === undefined) delete process.env[key]; + else process.env[key] = savedEnv[key]; + } +}); + +function serve( + manifest: string, + resource: ( + path: string, + request: Request, + ) => Response | Promise = () => new Response("segment"), +) { + const requests: string[] = []; + server = Bun.serve({ + port: 0, + fetch(request) { + const path = new URL(request.url).pathname; + requests.push(path); + if (path.includes("manifest.")) return new Response(manifest); + return resource(path, request); + }, + }); + return { url: server.url, requests }; +} + +describe("localized media resources", () => { + test("inherits a Period template and timeline through representation overrides", async () => { + const target = serve(` + + + `); + const path = await materializeMpdAsHlsPlaylist( + new URL("manifest.mpd", target.url).toString(), + directory, + ); + expect(await readFile(path, "utf8")).not.toContain("http:"); + expect(target.requests).toContain("/seg-3.m4s"); + expect(target.requests).toContain("/seg-4.m4s"); + }); + + test("localizes single quotes and whitespace around XML attribute assignments", async () => { + const target = serve( + ``, + ); + const path = await materializeMpdManifest( + new URL("manifest.mpd", target.url).toString(), + directory, + ); + const content = await readFile(path, "utf8"); + expect(target.requests).toContain("/init.mp4"); + expect(target.requests).toContain("/segment.m4s"); + expect(content).not.toContain('media="segment.m4s"'); + expect(content).not.toContain("http:"); + }); + + test("uses a working sibling when the first SegmentBase URL is missing", async () => { + const target = serve( + `../elsewhere/updated.mpdprimary/video.mp4backup/video.mp4`, + (path) => + new Response("segment", { + status: path.includes("primary") ? 404 : 200, + }), + ); + const path = await materializeMpdManifest( + new URL("manifest.mpd", target.url).toString(), + directory, + ); + const content = await readFile(path, "utf8"); + expect(content.match(//g)).toHaveLength(1); + expect(content).not.toContain("Location"); + expect(target.requests).toContain("/backup/video.mp4"); + expect(target.requests).not.toContain("/primary/backup/video.mp4"); + }); + + test("fails when no SegmentBase alternative works", async () => { + const target = serve( + `primary.mp4backup.mp4`, + () => new Response("missing", { status: 404 }), + ); + await expect( + materializeMpdManifest( + new URL("manifest.mpd", target.url).toString(), + directory, + ), + ).rejects.toThrow("404"); + }); + + test("downloads one copy of a byte-range HLS resource and preserves the offsets", async () => { + const target = serve( + "#EXTM3U\n#EXT-X-BYTERANGE:3@0\nshared.mp4\n#EXT-X-BYTERANGE:3@3\nshared.mp4\n#EXT-X-ENDLIST", + ); + const path = await materializeHlsPlaylist( + new URL("manifest.m3u8", target.url).toString(), + directory, + ); + const content = await readFile(path, "utf8"); + const resources = content + .split("\n") + .filter((line) => line && !line.startsWith("#")); + expect(resources[0]).toBe(resources[1]); + expect(content).toContain("#EXT-X-BYTERANGE:3@3"); + expect( + target.requests.filter((path) => path === "/shared.mp4"), + ).toHaveLength(1); + }); + + test.each(["hls", "dash", "converted-dash"])( + "stops and joins sibling downloads before a %s failure returns", + async (kind) => { + const manifest = + kind === "hls" + ? "#EXTM3U\nfailed.m4s\nslow.m4s\nqueued-1.m4s\nqueued-2.m4s\nqueued-3.m4s" + : kind === "dash" + ? `${["failed", "slow", "queued-1", "queued-2", "queued-3"].map((name) => ``).join("")}` + : ``; + let releaseSlow = () => {}; + const slowStarted = new Promise((resolve) => { + releaseSlow = resolve; + }); + const target = serve(manifest, async (path) => { + if (path.includes("failed") || path === "/seg-1.m4s") { + await slowStarted; + return new Response("missing", { status: 404 }); + } + if (path.includes("slow") || path === "/seg-2.m4s") { + releaseSlow(); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1])); + }, + }), + ); + } + return new Response("segment"); + }); + const materialize = + kind === "hls" + ? materializeHlsPlaylist + : kind === "dash" + ? materializeMpdManifest + : materializeMpdAsHlsPlaylist; + const url = new URL( + kind === "hls" ? "manifest.m3u8" : "manifest.mpd", + target.url, + ).toString(); + const operation = () => + materialize === materializeHlsPlaylist + ? materializeHlsPlaylist(url, directory) + : materialize(url, directory); + const originalFetch = globalThis.fetch; + const slowSignals: AbortSignal[] = []; + let requestsStarted = 0; + const fetcher = spyOn(globalThis, "fetch").mockImplementation( + Object.assign( + (input: RequestInfo | URL, init?: RequestInit) => { + requestsStarted++; + const path = new URL( + input instanceof Request ? input.url : String(input), + ).pathname; + if ((path === "/slow.m4s" || path === "/seg-2.m4s") && init?.signal) + slowSignals.push(init.signal); + return originalFetch(input, init); + }, + { preconnect: originalFetch.preconnect }, + ), + ); + try { + await expect(operation()).rejects.toThrow("404"); + expect(slowSignals).toHaveLength(1); + expect(slowSignals[0].aborted).toBe(true); + const completedFiles = await readdir(directory); + const requests = requestsStarted; + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(await readdir(directory)).toEqual(completedFiles); + expect(requestsStarted).toBe(requests); + } finally { + fetcher.mockRestore(); + } + }, + 2000, + ); + + test("bounds resource counts before expanding a huge template", async () => { + process.env.MEDIA_SERVER_MAX_PLAYLIST_RESOURCES = "5"; + const target = serve( + ``, + ); + await expect( + materializeMpdAsHlsPlaylist( + new URL("manifest.mpd", target.url).toString(), + directory, + ), + ).rejects.toThrow("resource count limit"); + expect(target.requests).toHaveLength(1); + }); + + test("enforces a shared transfer byte budget while streaming resources", async () => { + const target = serve( + "#EXTM3U\nsegment.mp4", + () => new Response(new Uint8Array(2048)), + ); + await expect( + withMediaTransfers(1024, () => + materializeHlsPlaylist( + new URL("manifest.m3u8", target.url).toString(), + directory, + ), + ), + ).rejects.toThrow("transfer budget"); + }); +}); diff --git a/apps/media-server/src/__tests__/lib/media-request-budgets.integration.test.ts b/apps/media-server/src/__tests__/lib/media-request-budgets.integration.test.ts new file mode 100644 index 00000000000..48cc247aba2 --- /dev/null +++ b/apps/media-server/src/__tests__/lib/media-request-budgets.integration.test.ts @@ -0,0 +1,167 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import * as childProcess from "node:child_process"; +import { readdir, readFile } from "node:fs/promises"; +import { join } from "node:path"; +import app from "../../app"; +import { withMediaTransfers } from "../../lib/media-transfer"; +import { generateThumbnail } from "../../lib/media-video"; +import { verifyRecording } from "../../lib/recording-verification"; +import { ensureTempDir, getTempDir } from "../../lib/temp-files"; + +const originalEnv = { ...process.env }; +const servers: ReturnType[] = []; + +afterEach(() => { + for (const server of servers.splice(0)) server.stop(true); + for (const key of [ + "MEDIA_SERVER_WEBHOOK_SECRET", + "MEDIA_SERVER_ALLOW_PRIVATE_URLS", + "S3_INTERNAL_ENDPOINT", + ]) { + if (originalEnv[key] === undefined) delete process.env[key]; + else process.env[key] = originalEnv[key]; + } +}); + +function startServer( + fetch: (request: Request) => Response | Promise, +) { + const server = Bun.serve({ port: 0, fetch }); + servers.push(server); + return server; +} + +describe("media request budgets", () => { + test("applies the thumbnail deadline while downloading", async () => { + const timeout = AbortSignal.timeout.bind(AbortSignal); + const timer = spyOn(AbortSignal, "timeout").mockImplementation((ms) => + timeout(ms === 60_000 ? 30 : ms), + ); + let requested = false; + const server = startServer(() => { + requested = true; + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1])); + }, + }), + ); + }); + try { + await expect( + generateThumbnail(new URL("video.mp4", server.url).toString(), 5), + ).rejects.toThrow(); + expect(requested).toBe(true); + expect(timer).toHaveBeenCalledWith(60_000); + } finally { + timer.mockRestore(); + } + }, 1000); + + test("returns JSON URL_NOT_ALLOWED before an audio stream response is sent", async () => { + process.env.MEDIA_SERVER_WEBHOOK_SECRET = "test-secret"; + process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS = "false"; + let requested = false; + const server = startServer(() => { + requested = true; + return new Response(null, { + status: 307, + headers: { location: "http://169.254.169.254/private" }, + }); + }); + process.env.S3_INTERNAL_ENDPOINT = server.url.toString(); + const response = await app.fetch( + new Request("http://localhost/audio/convert", { + method: "POST", + headers: { + "Content-Type": "application/json", + "x-media-server-secret": "test-secret", + }, + body: JSON.stringify({ + audioUrl: new URL("audio.mp4", server.url).toString(), + }), + }), + ); + expect(requested).toBe(true); + expect(response.status).toBe(400); + expect(await response.json()).toMatchObject({ code: "URL_NOT_ALLOWED" }); + }); + + test("accounts for remote verification staging in the managed temp directory and removes it on timeout", async () => { + await ensureTempDir(); + const before = await readdir(getTempDir()); + let managedPath = ""; + const server = startServer(async () => { + managedPath = + (await readdir(getTempDir())).find( + (name) => name.startsWith("recording-") && !before.includes(name), + ) ?? ""; + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1])); + }, + }), + ); + }); + await expect( + verifyRecording(new URL("recording.mp4", server.url).toString(), { + allowObservedDuration: true, + requireAudio: false, + timeoutMs: 100, + }), + ).rejects.toThrow("timed out"); + expect(managedPath).not.toBe(""); + expect(await readdir(getTempDir())).not.toContain(managedPath); + }); + + test("shares the recording deadline between staging and decode", async () => { + const bytes = await readFile( + join(import.meta.dir, "../fixtures/test-with-audio.mp4"), + ); + const server = startServer(async () => { + await new Promise((resolve) => setTimeout(resolve, 300)); + return new Response(bytes); + }); + const originalSpawn = childProcess.spawn; + const decoder = spyOn(childProcess, "spawn").mockImplementation((( + ...args: Parameters + ) => + originalSpawn( + process.execPath, + ["-e", "setTimeout(() => {}, 2000)"], + args[2], + )) as typeof childProcess.spawn); + const startedAt = performance.now(); + try { + await expect( + verifyRecording(new URL("recording.mp4", server.url).toString(), { + allowObservedDuration: true, + requireAudio: false, + hasAudio: false, + timeoutMs: 500, + }), + ).rejects.toThrow("timed out"); + expect(decoder).toHaveBeenCalledTimes(1); + expect(performance.now() - startedAt).toBeLessThan(700); + } finally { + decoder.mockRestore(); + } + }, 1500); + + test("enforces the transfer budget on non-internal recording URLs", async () => { + const bytes = await readFile( + join(import.meta.dir, "../fixtures/test-with-audio.mp4"), + ); + const server = startServer(() => new Response(bytes)); + await expect( + withMediaTransfers(1024, () => + verifyRecording(new URL("recording.mp4", server.url).toString(), { + allowObservedDuration: true, + requireAudio: false, + }), + ), + ).rejects.toThrow("transfer budget"); + }); +}); diff --git a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts index f957ac607f1..dea435a1b9d 100644 --- a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts @@ -14,7 +14,6 @@ import { mkdtempSync, rmSync } from "node:fs"; import { readFile, writeFile } from "node:fs/promises"; import os, { tmpdir } from "node:os"; import { join } from "node:path"; -import { pathToFileURL } from "node:url"; import type appType from "../../app"; import * as audioLevels from "../../lib/audio-levels"; import * as containerCpu from "../../lib/container-cpu"; @@ -66,10 +65,6 @@ let transientFixtureFailures = 0; let permanentFixtureFailures = 0; let slowFixtureCancellations = 0; -function fileUrl(path: string) { - return pathToFileURL(path).toString(); -} - function fixtureUrl(name = "test-with-audio.mp4") { return `${baseUrl}/fixtures/${name}`; } @@ -1142,12 +1137,12 @@ describe("media routes real-world integration tests", () => { test("checks real audio tracks through the route stack", async () => { const withAudioResponse = await app.fetch( mediaPostRequest("/audio/check", { - videoUrl: fileUrl(TEST_VIDEO_WITH_AUDIO), + videoUrl: fixtureUrl("test-with-audio.mp4"), }), ); const noAudioResponse = await app.fetch( mediaPostRequest("/audio/check", { - videoUrl: fileUrl(TEST_VIDEO_NO_AUDIO), + videoUrl: fixtureUrl("test-no-audio.mp4"), }), ); @@ -1163,7 +1158,7 @@ describe("media routes real-world integration tests", () => { test("extracts real audio through the non-streaming route", async () => { const response = await app.fetch( mediaPostRequest("/audio/extract", { - videoUrl: fileUrl(TEST_VIDEO_WITH_AUDIO), + videoUrl: fixtureUrl("test-with-audio.mp4"), stream: false, }), ); @@ -1176,7 +1171,7 @@ describe("media routes real-world integration tests", () => { test("probes and thumbnails a real video through the route stack", async () => { const probeResponse = await app.fetch( mediaPostRequest("/video/probe", { - videoUrl: fileUrl(TEST_VIDEO_WITH_AUDIO), + videoUrl: fixtureUrl("test-with-audio.mp4"), }), ); expect(probeResponse.status).toBe(200); @@ -1188,7 +1183,7 @@ describe("media routes real-world integration tests", () => { const thumbnailResponse = await app.fetch( mediaPostRequest("/video/thumbnail", { - videoUrl: fileUrl(TEST_VIDEO_WITH_AUDIO), + videoUrl: fixtureUrl("test-with-audio.mp4"), timestamp: 0.2, width: 160, height: 120, diff --git a/apps/media-server/src/__tests__/lib/media-video.integration.test.ts b/apps/media-server/src/__tests__/lib/media-video.integration.test.ts index dd9844d100c..ae98c0d8dff 100644 --- a/apps/media-server/src/__tests__/lib/media-video.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-video.integration.test.ts @@ -3,7 +3,6 @@ import { execFileSync } from "node:child_process"; import { existsSync, mkdtempSync, - readdirSync, readFileSync, rmSync, statSync, @@ -329,7 +328,7 @@ describe("generateThumbnail integration tests", () => { } }); - test("joins an in-flight thumbnail decoder when its worker is cancelled", async () => { + test("cancels an in-flight remote thumbnail download", async () => { let ready: (() => void) | undefined; const started = new Promise((resolve) => { ready = resolve; @@ -358,35 +357,8 @@ describe("generateThumbnail integration tests", () => { ); try { await withTimeout(started, 5_000); - const commands = - process.platform === "linux" - ? readdirSync("/proc") - .filter((entry) => /^\d+$/.test(entry)) - .map((pid) => { - try { - return `${pid} ${readFileSync(`/proc/${pid}/cmdline`, "utf8")}`; - } catch (error) { - if ( - error instanceof Error && - "code" in error && - (error.code === "ENOENT" || error.code === "ESRCH") - ) - return ""; - throw error; - } - }) - : execFileSync("ps", ["-axo", "pid=,command="]) - .toString() - .split("\n"); - const pids = commands - .filter( - (command) => command.includes("ffmpeg") && command.includes(input), - ) - .map((command) => Number.parseInt(command.trim(), 10)); - expect(pids).toHaveLength(1); controller.abort(reason); expect(await withTimeout(outcome, 7_000)).toBe(reason); - for (const pid of pids) expect(() => process.kill(pid, 0)).toThrow(); } finally { controller.abort(reason); await server.stop(true); @@ -645,11 +617,16 @@ describe("processVideo integration tests", () => { const originalFetch = globalThis.fetch; const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-")); - globalThis.fetch = (async () => - new Response( - '', + const requestedUrls: string[] = []; + globalThis.fetch = (async (input: RequestInfo | URL) => { + requestedUrls.push( + input instanceof Request ? input.url : input.toString(), + ); + return new Response( + '', { status: 200, statusText: "OK" }, - )) as unknown as typeof fetch; + ); + }) as unknown as typeof fetch; try { const path = await materializeMpdManifest( @@ -658,13 +635,13 @@ describe("processVideo integration tests", () => { ); const content = readFileSync(path, "utf8"); - expect(content).toContain( - "init.mp4?Policy=a&Signature=b&Key-Pair-Id=c", + expect(requestedUrls).toContain( + "https://cdn.example/video/init.mp4?Policy=a&Signature=b&Key-Pair-Id=c", ); - expect(content).toContain( - "chunk-$Number$.m4s?Policy=a&Signature=b&Key-Pair-Id=c", + expect(requestedUrls).toContain( + "https://cdn.example/video/escaped.mp4?part=1&token=x", ); - expect(content).toContain("escaped.mp4?part=1&token=x"); + expect(content).not.toContain("https://cdn.example"); expect(content).not.toContain("&amp;"); } finally { globalThis.fetch = originalFetch; @@ -717,21 +694,10 @@ describe("processVideo integration tests", () => { const videoPlaylist = readFileSync(videoPlaylistPath ?? "", "utf8"); const audioPlaylist = readFileSync(mediaMatch?.[1] ?? "", "utf8"); - expect(videoPlaylist).toContain( - "https://cdn.example/video/video/init-v1.mp4?Policy=a&Signature=b", - ); - expect(videoPlaylist).toContain( - "https://cdn.example/video/video/chunk-00003.m4s?Policy=a&Signature=b", - ); - expect(videoPlaylist).toContain( - "https://cdn.example/video/video/chunk-00004.m4s?Policy=a&Signature=b", - ); - expect(audioPlaylist).toContain( - "https://cdn.example/video/audio/chunk-0.m4s?Policy=a&Signature=b", - ); - expect(audioPlaylist).toContain( - "https://cdn.example/video/audio/chunk-48000.m4s?Policy=a&Signature=b", - ); + expect(videoPlaylist).not.toContain("https://cdn.example"); + expect(audioPlaylist).not.toContain("https://cdn.example"); + expect(videoPlaylist).toContain(".m4s"); + expect(audioPlaylist).toContain(".m4s"); } finally { globalThis.fetch = originalFetch; rmSync(manifestDir, { recursive: true, force: true }); @@ -741,12 +707,17 @@ describe("processVideo integration tests", () => { test("falls back to generic DASH manifest materialization for unsupported MPD shapes", async () => { const originalFetch = globalThis.fetch; const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-fallback-")); - let requests = 0; + const requestedUrls: string[] = []; - globalThis.fetch = (async () => { - requests++; - return new Response( - ` + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : input.toString(); + requestedUrls.push(url); + if (new URL(url).pathname.endsWith("/")) { + return new Response("Missing directory", { status: 404 }); + } + if (url.includes("manifest.mpd")) { + return new Response( + ` @@ -759,8 +730,10 @@ describe("processVideo integration tests", () => { `, - { status: 200, statusText: "OK" }, - ); + { status: 200, statusText: "OK" }, + ); + } + return new Response("segment", { status: 200, statusText: "OK" }); }) as unknown as typeof fetch; try { @@ -771,11 +744,153 @@ describe("processVideo integration tests", () => { const content = readFileSync(path, "utf8"); expect(path.endsWith(".mpd")).toBe(true); - expect(requests).toBe(2); - expect(content).toContain( - "https://cdn.example/video/media/?Policy=a&Signature=b", + expect(requestedUrls).toContain( + "https://cdn.example/video/media/init.mp4?Policy=a&Signature=b", + ); + expect(requestedUrls).toContain( + "https://cdn.example/video/media/seg-1.m4s?Policy=a&Signature=b", ); - expect(content).toContain("seg-1.m4s?Policy=a&Signature=b"); + expect( + requestedUrls.some((url) => new URL(url).pathname.endsWith("/")), + ).toBe(false); + expect(content).not.toContain("https://cdn.example"); + expect(content).not.toContain(""); + } finally { + globalThis.fetch = originalFetch; + rmSync(manifestDir, { recursive: true, force: true }); + } + }); + + test("ignores MPD Location and resolves sibling BaseURLs against their parent", async () => { + const originalFetch = globalThis.fetch; + const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-alternatives-")); + const requestedUrls: string[] = []; + + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : input.toString(); + requestedUrls.push(url); + if (url.includes("manifest.mpd")) { + return new Response( + ` + https://cdn.example/elsewhere/live/updated.mpd + + + + primary/video.mp4 + backup/video.mp4 + + + + + `, + { status: 200, statusText: "OK" }, + ); + } + return new Response("segment", { status: 200, statusText: "OK" }); + }) as unknown as typeof fetch; + + try { + const path = await materializeStreamingInput( + "https://cdn.example/video/manifest.mpd?Policy=a&Signature=b", + manifestDir, + ); + const content = readFileSync(path, "utf8"); + + expect(requestedUrls).toContain( + "https://cdn.example/video/primary/video.mp4?Policy=a&Signature=b", + ); + expect(requestedUrls).toContain( + "https://cdn.example/video/backup/video.mp4?Policy=a&Signature=b", + ); + expect(requestedUrls.some((url) => url.includes("primary/backup"))).toBe( + false, + ); + expect(requestedUrls.some((url) => url.includes("elsewhere"))).toBe( + false, + ); + expect(content).not.toContain(""); + expect(content).not.toContain("https://cdn.example"); + expect(content.match(//g)?.length).toBe(2); + } finally { + globalThis.fetch = originalFetch; + rmSync(manifestDir, { recursive: true, force: true }); + } + }); + + test("resolves DASH segments against the parent when MPD Location is in another directory", async () => { + const originalFetch = globalThis.fetch; + const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-location-")); + const requestedUrls: string[] = []; + + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : input.toString(); + requestedUrls.push(url); + if (url.includes("manifest.mpd")) { + return new Response( + ` + ../other/updated.mpd + + + + + + + + + + + `, + { status: 200, statusText: "OK" }, + ); + } + return new Response("segment", { status: 200, statusText: "OK" }); + }) as unknown as typeof fetch; + + try { + await materializeStreamingInput( + "https://cdn.example/video/manifest.mpd", + manifestDir, + ); + + expect(requestedUrls).toContain("https://cdn.example/video/init.mp4"); + expect(requestedUrls).toContain("https://cdn.example/video/seg-1.m4s"); + expect(requestedUrls.some((url) => url.includes("/other/"))).toBe(false); + } finally { + globalThis.fetch = originalFetch; + rmSync(manifestDir, { recursive: true, force: true }); + } + }); + + test("drops a missing alternative BaseURL instead of failing", async () => { + const originalFetch = globalThis.fetch; + const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-alt-missing-")); + + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : input.toString(); + if (url.includes("manifest.mpd")) { + return new Response( + `primary/video.mp4backup/video.mp4`, + { status: 200, statusText: "OK" }, + ); + } + if (url.includes("backup/")) { + return new Response("missing", { + status: 404, + statusText: "Not Found", + }); + } + return new Response("segment", { status: 200, statusText: "OK" }); + }) as unknown as typeof fetch; + + try { + const path = await materializeStreamingInput( + "https://cdn.example/video/manifest.mpd", + manifestDir, + ); + const content = readFileSync(path, "utf8"); + + expect(content.match(//g)?.length).toBe(1); + expect(content).not.toContain("https://cdn.example"); } finally { globalThis.fetch = originalFetch; rmSync(manifestDir, { recursive: true, force: true }); @@ -877,10 +992,7 @@ describe("processVideo integration tests", () => { ); } - return new Response("not found", { - status: 404, - statusText: "Not Found", - }); + return new Response("segment", { status: 200, statusText: "OK" }); }) as unknown as typeof fetch; try { @@ -896,21 +1008,24 @@ describe("processVideo integration tests", () => { expect(requestedUrls).toContain( "https://cdn.example/video/variant.m3u8?Policy=a&Signature=b", ); - expect(variantPath).toBeTruthy(); - - const variant = readFileSync(variantPath ?? "", "utf8"); - expect(variant).toContain( - 'URI="https://cdn.example/video/init.mp4?Policy=a&Signature=b"', + expect(requestedUrls).toContain( + "https://cdn.example/video/init.mp4?Policy=a&Signature=b", ); - expect(variant).toContain( - 'URI="https://cdn.example/video/key.bin?Policy=a&Signature=b"', + expect(requestedUrls).toContain( + "https://cdn.example/video/key.bin?Policy=a&Signature=b", ); - expect(variant).toContain( + expect(requestedUrls).toContain( "https://cdn.example/video/segment-1.ts?Policy=a&Signature=b", ); - expect(variant).toContain( + expect(requestedUrls).toContain( "https://cdn.example/video/segment-2.ts?Policy=a&Signature=b", ); + expect(variantPath).toBeTruthy(); + + const variant = readFileSync(variantPath ?? "", "utf8"); + expect(variant).not.toContain("https://cdn.example"); + expect(variant).toContain(".ts"); + expect(variant).toContain(".bin"); } finally { globalThis.fetch = originalFetch; rmSync(manifestDir, { recursive: true, force: true }); diff --git a/apps/media-server/src/__tests__/lib/memory-leak.test.ts b/apps/media-server/src/__tests__/lib/memory-leak.test.ts index a51b40b5564..be266397cce 100644 --- a/apps/media-server/src/__tests__/lib/memory-leak.test.ts +++ b/apps/media-server/src/__tests__/lib/memory-leak.test.ts @@ -55,7 +55,7 @@ describe("memory and resource leak tests", () => { test("extractAudioStream cleans up after stream is fully consumed", async () => { const beforeCount = getActiveProcessCount(); - const { stream } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream } = await extractAudioStream(TEST_VIDEO_WITH_AUDIO); const reader = stream.getReader(); while (true) { @@ -71,7 +71,9 @@ describe("memory and resource leak tests", () => { test("extractAudioStream cleans up when stream is cancelled", async () => { const beforeCount = getActiveProcessCount(); - const { stream, cleanup } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream, cleanup } = await extractAudioStream( + TEST_VIDEO_WITH_AUDIO, + ); const reader = stream.getReader(); await reader.read(); @@ -86,7 +88,7 @@ describe("memory and resource leak tests", () => { test("extractAudioStream cleans up via cancel() method", async () => { const beforeCount = getActiveProcessCount(); - const { stream } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream } = await extractAudioStream(TEST_VIDEO_WITH_AUDIO); await stream.cancel(); @@ -133,7 +135,7 @@ describe("memory and resource leak tests", () => { checkHasAudioTrack(TEST_VIDEO_WITH_AUDIO), extractAudio(TEST_VIDEO_WITH_AUDIO), (async () => { - const { stream } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream } = await extractAudioStream(TEST_VIDEO_WITH_AUDIO); const reader = stream.getReader(); while (true) { const { done } = await reader.read(); @@ -166,7 +168,7 @@ describe("memory and resource leak tests", () => { await checkHasAudioTrack(TEST_VIDEO_WITH_AUDIO); await extractAudio(TEST_VIDEO_WITH_AUDIO); - const { stream } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream } = await extractAudioStream(TEST_VIDEO_WITH_AUDIO); const reader = stream.getReader(); while (true) { const { done } = await reader.read(); @@ -200,7 +202,7 @@ describe("memory and resource leak tests", () => { } else if (i % 3 === 1) { await extractAudio(TEST_VIDEO_WITH_AUDIO); } else { - const { stream, cleanup } = extractAudioStream( + const { stream, cleanup } = await extractAudioStream( TEST_VIDEO_WITH_AUDIO, ); const reader = stream.getReader(); @@ -226,7 +228,9 @@ describe("memory and resource leak tests", () => { await checkHasAudioTrack(TEST_VIDEO_WITH_AUDIO); await extractAudio(TEST_VIDEO_WITH_AUDIO); - const { stream, cleanup } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); + const { stream, cleanup } = await extractAudioStream( + TEST_VIDEO_WITH_AUDIO, + ); await stream.cancel(); cleanup(); @@ -241,7 +245,7 @@ describe("abort signal handling", () => { test("stream cleanup is called when response is aborted", async () => { let cleanupCalled = false; - const { stream, cleanup: originalCleanup } = extractAudioStream( + const { stream, cleanup: originalCleanup } = await extractAudioStream( TEST_VIDEO_WITH_AUDIO, ); diff --git a/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts b/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts index afc3199881a..f7d26c9363b 100644 --- a/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts @@ -1999,10 +1999,6 @@ async function decoderPids(input: string): Promise { .map((line) => Number.parseInt(line.trim(), 10)); } -function expectExited(pid: number) { - expect(() => process.kill(pid, 0)).toThrow(); -} - describe("recording timing metadata budget", () => { const mb = 1024 * 1024; test("keeps at least 64 MB for small recordings", () => { @@ -2434,15 +2430,13 @@ describe("recording verification lifetime", () => { (error: unknown) => error, ); await request; - const pids = await decoderPids(input); - expect(pids.length).toBe(1); if (cause === "cancel") controller.abort(); const error = await outcome; expect(error).toBeInstanceOf(Error); expect((error as Error).message).toContain( cause === "cancel" ? "cancelled" : "timed out", ); - for (const pid of pids) expectExited(pid); + expect(await decoderPids(input)).toEqual([]); } finally { controller.abort(); await server.stop(true); @@ -3531,9 +3525,7 @@ describe("remote recording object identity", () => { { expectedDuration: 5, requireAudio: true }, ).catch((error: unknown) => error); expect(error).toBeInstanceOf(Error); - expect((error as Error).message).toContain( - when === "before-decode" ? "full decode failed" : "object changed", - ); + expect((error as Error).message).toContain("object changed"); expect(isRetryableRecordingVerificationError(error)).toBe(false); expect(decoderRequests).toBeGreaterThan(0); if (when !== "before-decode") expect(probes).toBe(2); @@ -3583,8 +3575,6 @@ describe("remote recording object identity", () => { (error: unknown) => error, ); await decoding; - const pids = await decoderPids(input); - expect(pids.length).toBe(1); if (cause === "cancel") controller.abort(); const error = await outcome; expect(error).toBeInstanceOf(Error); @@ -3595,7 +3585,7 @@ describe("remote recording object identity", () => { expect(isRetryableRecordingVerificationError(error)).toBe( cause === "timeout", ); - for (const pid of pids) expectExited(pid); + expect(await decoderPids(input)).toEqual([]); } finally { controller.abort(); await server.stop(true); diff --git a/apps/media-server/src/__tests__/lib/redirect-follow.test.ts b/apps/media-server/src/__tests__/lib/redirect-follow.test.ts new file mode 100644 index 00000000000..16c4253f5c7 --- /dev/null +++ b/apps/media-server/src/__tests__/lib/redirect-follow.test.ts @@ -0,0 +1,134 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { extractAudio } from "../../lib/media-audio"; +import { probeVideo } from "../../lib/media-probe"; +import { downloadVideoToTemp, generateThumbnail } from "../../lib/media-video"; +import { verifyRecording } from "../../lib/recording-verification"; + +const video = readFileSync( + join(import.meta.dir, "..", "fixtures", "test-with-audio.mp4"), +); + +describe("redirects to a private address", () => { + const previous = process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS; + const previousEndpoint = process.env.S3_INTERNAL_ENDPOINT; + + afterEach(() => { + if (previousEndpoint === undefined) delete process.env.S3_INTERNAL_ENDPOINT; + else process.env.S3_INTERNAL_ENDPOINT = previousEndpoint; + if (previous === undefined) { + delete process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS; + } else { + process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS = previous; + } + }); + + function privateRedirectServer() { + let secretHits = 0; + let firstHits = 0; + const server: ReturnType = Bun.serve({ + port: 0, + fetch(request): Response { + const url = new URL(request.url); + if (url.pathname === "/secret") { + secretHits += 1; + return new Response(video, { + headers: { "content-type": "video/mp4" }, + }); + } + firstHits++; + const destination = new URL("/secret", server.url); + destination.hostname = "127.0.0.2"; + return new Response(null, { + status: 302, + headers: { + location: destination.toString(), + }, + }); + }, + }); + process.env.S3_INTERNAL_ENDPOINT = server.url.toString(); + return { + server, + firstHits: () => firstHits, + url: `${server.url}clip.mp4`, + secretHits: () => secretHits, + }; + } + + test("probe does not fetch the redirect target", async () => { + process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS = "false"; + const target = privateRedirectServer(); + try { + await expect(probeVideo(target.url)).rejects.toThrow("private address"); + expect(target.firstHits()).toBeGreaterThan(0); + expect(target.secretHits()).toBe(0); + } finally { + target.server.stop(true); + } + }); + + test("audio extraction does not fetch the redirect target", async () => { + process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS = "false"; + const target = privateRedirectServer(); + try { + await expect(extractAudio(target.url)).rejects.toThrow("private address"); + expect(target.firstHits()).toBeGreaterThan(0); + expect(target.secretHits()).toBe(0); + } finally { + target.server.stop(true); + } + }); + + test("thumbnail generation does not fetch the redirect target", async () => { + process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS = "false"; + const target = privateRedirectServer(); + try { + await expect(generateThumbnail(target.url, 1)).rejects.toThrow( + "private address", + ); + expect(target.firstHits()).toBeGreaterThan(0); + expect(target.secretHits()).toBe(0); + } finally { + target.server.stop(true); + } + }); + + test("recording verification does not fetch the redirect target", async () => { + process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS = "false"; + const target = privateRedirectServer(); + try { + await expect( + verifyRecording(target.url, { + allowObservedDuration: true, + requireAudio: false, + }), + ).rejects.toThrow("private address"); + expect(target.firstHits()).toBeGreaterThan(0); + expect(target.secretHits()).toBe(0); + } finally { + target.server.stop(true); + } + }); + test("parallel ranged downloads do not fetch the redirect target", async () => { + process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS = "false"; + const previousConcurrency = process.env.MEDIA_SERVER_DOWNLOAD_CONCURRENCY; + process.env.MEDIA_SERVER_DOWNLOAD_CONCURRENCY = "4"; + const target = privateRedirectServer(); + try { + await expect(downloadVideoToTemp(target.url, "mp4")).rejects.toThrow( + "private address", + ); + expect(target.firstHits()).toBeGreaterThan(0); + expect(target.secretHits()).toBe(0); + } finally { + if (previousConcurrency === undefined) { + delete process.env.MEDIA_SERVER_DOWNLOAD_CONCURRENCY; + } else { + process.env.MEDIA_SERVER_DOWNLOAD_CONCURRENCY = previousConcurrency; + } + target.server.stop(true); + } + }); +}); diff --git a/apps/media-server/src/__tests__/lib/url-policy.test.ts b/apps/media-server/src/__tests__/lib/url-policy.test.ts new file mode 100644 index 00000000000..f25198b0e3e --- /dev/null +++ b/apps/media-server/src/__tests__/lib/url-policy.test.ts @@ -0,0 +1,329 @@ +import { describe, expect, test } from "bun:test"; +import { + assertUrlAllowed, + collectUrls, + fetchAllowed, + getUrlPolicy, + isPrivateAddress, + resolveAllowedUrl, +} from "../../lib/url-policy"; + +const resolveTo = + (...addresses: string[]) => + async () => + addresses; + +describe("isPrivateAddress", () => { + test("flags loopback, private, link-local, CGNAT and ULA addresses", () => { + for (const address of [ + "127.0.0.1", + "10.1.2.3", + "172.20.0.5", + "192.168.1.10", + "169.254.169.254", + "100.100.1.1", + "::1", + "fd12:3456::1", + "fe80::1", + "::ffff:127.0.0.1", + ]) { + expect(isPrivateAddress(address)).toBe(true); + } + }); + + test("allows public addresses", () => { + expect(isPrivateAddress("93.184.216.34")).toBe(false); + expect(isPrivateAddress("2606:4700::6810:85e5")).toBe(false); + expect(isPrivateAddress("::ffff:8.8.8.8")).toBe(false); + }); + + test.each(["::a9fe:a9fe", "::7f00:1", "64:ff9b::a9fe:a9fe", "fec0::1"])( + "blocks IPv6 translation and site-local address %s", + async (address) => { + expect(isPrivateAddress(address)).toBe(true); + await expect( + assertUrlAllowed(`http://[${address}]/`, getUrlPolicy({})), + ).rejects.toThrow("private address"); + }, + ); +}); + +describe("assertUrlAllowed", () => { + const open = getUrlPolicy({}); + + test("blocks cloud metadata and private hosts by default", async () => { + await expect( + assertUrlAllowed("http://169.254.169.254/latest/meta-data/", open), + ).rejects.toThrow("private address"); + await expect( + assertUrlAllowed( + "http://internal.example/x", + open, + resolveTo("10.0.0.7"), + ), + ).rejects.toThrow("private address"); + }); + + test("blocks non-http schemes", async () => { + await expect(assertUrlAllowed("file:///etc/passwd", open)).rejects.toThrow( + "scheme", + ); + }); + + test("allows public hosts when no allow-list is configured", async () => { + await expect( + assertUrlAllowed( + "https://bucket.s3.amazonaws.com/v.mp4", + open, + resolveTo("52.216.1.1"), + ), + ).resolves.toBeUndefined(); + }); + + test("allows the configured S3 endpoint on a private network", async () => { + const policy = getUrlPolicy({ S3_INTERNAL_ENDPOINT: "http://minio:9000" }); + await expect( + assertUrlAllowed( + "http://minio:9000/cap/v.mp4", + policy, + resolveTo("172.18.0.3"), + ), + ).resolves.toBeUndefined(); + }); + + test("does not treat a trusted host as allowed on every port", async () => { + const policy = getUrlPolicy({ + S3_PUBLIC_ENDPOINT: "http://localhost:9000", + MEDIA_SERVER_WEBHOOK_URL: "http://cap-web:3000", + }); + await expect( + assertUrlAllowed( + "http://minio:445/cap/v.mp4", + policy, + resolveTo("10.0.0.2"), + ), + ).rejects.toThrow("private address"); + await expect( + assertUrlAllowed( + "http://localhost:6379/", + policy, + resolveTo("127.0.0.1"), + ), + ).rejects.toThrow("private address"); + await expect( + assertUrlAllowed("http://localhost:9000/bucket/v.mp4", policy), + ).resolves.toBeUndefined(); + }); + + test("blocks file URLs even when private URLs are allowed", async () => { + const policy = getUrlPolicy({ MEDIA_SERVER_ALLOW_PRIVATE_URLS: "true" }); + await expect( + assertUrlAllowed("file:///etc/passwd", policy), + ).rejects.toThrow("scheme"); + }); + + test("enforces MEDIA_SERVER_ALLOWED_HOSTS when set", async () => { + const policy = getUrlPolicy({ + MEDIA_SERVER_ALLOWED_HOSTS: "*.r2.cloudflarestorage.com", + MEDIA_SERVER_WEBHOOK_URL: "http://cap-web:3000", + }); + await expect( + assertUrlAllowed( + "https://acct.r2.cloudflarestorage.com/v.mp4", + policy, + resolveTo("104.18.1.1"), + ), + ).resolves.toBeUndefined(); + await expect( + assertUrlAllowed("http://cap-web:3000/api/webhooks", policy), + ).resolves.toBeUndefined(); + await expect( + assertUrlAllowed( + "https://evil.example/v.mp4", + policy, + resolveTo("1.1.1.1"), + ), + ).rejects.toThrow("MEDIA_SERVER_ALLOWED_HOSTS"); + }); +}); + +describe("redirect checks", () => { + test("does not follow a redirect to a private address", async () => { + const policy = getUrlPolicy({}); + const fetcher = async (input: string) => { + if (input === "https://93.184.216.34/v.mp4") { + return new Response(null, { + status: 302, + headers: { location: "http://169.254.169.254/latest/meta-data/" }, + }); + } + return new Response("ok"); + }; + await expect( + fetchAllowed( + "https://cdn.example/v.mp4", + {}, + policy, + resolveTo("93.184.216.34"), + fetcher, + ), + ).rejects.toThrow("private address"); + await expect( + resolveAllowedUrl( + "https://cdn.example/v.mp4", + policy, + resolveTo("93.184.216.34"), + fetcher, + ), + ).rejects.toThrow("private address"); + }); +}); + +describe("collectUrls", () => { + test("collects nested url fields only", () => { + expect( + collectUrls({ + videoUrl: "http://a/1", + name: "http://not-a-url-field", + segments: [{ url: "http://b/2" }], + videoSegmentUrls: ["http://c/3"], + }), + ).toEqual(["http://a/1", "http://b/2", "http://c/3"]); + }); +}); + +describe("connection address pinning", () => { + test.each(["fetch", "resolve"])( + "pins the checked address during %s without another hostname lookup", + async (mode) => { + let lookups = 0; + const resolve = async () => + ++lookups === 1 ? ["93.184.216.34"] : ["127.0.0.1"]; + let connections = 0; + const fetcher: NonNullable[4]> = async ( + input, + init, + ) => { + connections++; + expect(input).toBe("https://93.184.216.34:8443/video.mp4?token=signed"); + expect(new Headers(init?.headers).get("host")).toBe("cdn.example:8443"); + expect(init?.tls?.serverName).toBe("cdn.example"); + expect(init?.proxy).toBe(false); + return new Response("ok"); + }; + const url = "https://cdn.example:8443/video.mp4?token=signed"; + if (mode === "fetch") { + const response = await fetchAllowed( + url, + {}, + getUrlPolicy({}), + resolve, + fetcher, + ); + expect(response.url).toBe(url); + } else { + expect( + await resolveAllowedUrl(url, getUrlPolicy({}), resolve, fetcher), + ).toBe(url); + } + expect(lookups).toBe(1); + expect(connections).toBe(1); + }, + ); + + test("resolves relative redirects against the original hostname", async () => { + const seen: string[] = []; + const fetcher: NonNullable[4]> = async ( + input, + init, + ) => { + seen.push(input); + expect(new Headers(init?.headers).get("host")).toBe("cdn.example"); + return seen.length === 1 + ? new Response(null, { + status: 302, + headers: { location: "../video.mp4" }, + }) + : new Response("ok"); + }; + await fetchAllowed( + "https://cdn.example/start/redirect", + {}, + getUrlPolicy({}), + resolveTo("93.184.216.34"), + fetcher, + ); + expect(seen).toEqual([ + "https://93.184.216.34/start/redirect", + "https://93.184.216.34/video.mp4", + ]); + }); + + test("refuses callback redirects before sending a secret to the target", async () => { + let calls = 0; + const fetcher = async () => { + calls++; + return new Response(null, { + status: 307, + headers: { location: "https://other.example/secret" }, + }); + }; + await expect( + fetchAllowed( + "https://cdn.example/callback", + { + method: "POST", + redirect: "error", + headers: { "x-media-server-secret": "test-secret" }, + body: "recording-details", + }, + getUrlPolicy({}), + resolveTo("93.184.216.34"), + fetcher, + ), + ).rejects.toThrow("Redirects are not allowed"); + expect(calls).toBe(1); + }); + + test("returns manual redirects without following them", async () => { + let calls = 0; + const response = await fetchAllowed( + "https://cdn.example/callback", + { redirect: "manual" }, + getUrlPolicy({}), + resolveTo("93.184.216.34"), + async () => { + calls++; + return new Response(null, { + status: 302, + headers: { location: "http://127.0.0.1/" }, + }); + }, + ); + expect(response.status).toBe(302); + expect(calls).toBe(1); + }); +}); + +test("tries another checked address when a dual-stack media connection fails", async () => { + const seen: string[] = []; + const fetcher: NonNullable[4]> = async ( + input, + ) => { + seen.push(input); + if (seen.length === 1) throw new Error("IPv6 is unreachable"); + return new Response("ok"); + }; + const response = await fetchAllowed( + "https://cdn.example/video.mp4", + {}, + getUrlPolicy({}), + resolveTo("2606:4700::6810:85e5", "93.184.216.34"), + fetcher, + ); + expect(await response.text()).toBe("ok"); + expect(seen).toEqual([ + "https://[2606:4700::6810:85e5]/video.mp4", + "https://93.184.216.34/video.mp4", + ]); +}); diff --git a/apps/media-server/src/__tests__/routes/audio-memory.test.ts b/apps/media-server/src/__tests__/routes/audio-memory.test.ts index 82241f830ab..8f5cd532138 100644 --- a/apps/media-server/src/__tests__/routes/audio-memory.test.ts +++ b/apps/media-server/src/__tests__/routes/audio-memory.test.ts @@ -1,9 +1,27 @@ -import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { + afterAll, + afterEach, + beforeEach, + describe, + expect, + mock, + test, +} from "bun:test"; import { join } from "node:path"; import { getActiveProcessCount } from "../../lib/media-audio"; const FIXTURES_DIR = join(import.meta.dir, "..", "fixtures"); -const TEST_VIDEO_WITH_AUDIO = `file://${join(FIXTURES_DIR, "test-with-audio.mp4")}`; +// The routes only accept http(s) URLs, so serve the fixture locally. +const fixtureServer = Bun.serve({ + port: 0, + fetch: () => + new Response(Bun.file(join(FIXTURES_DIR, "test-with-audio.mp4"))), +}); +const TEST_VIDEO_WITH_AUDIO = `${fixtureServer.url}test-with-audio.mp4`; + +afterAll(() => { + fixtureServer.stop(true); +}); const MEDIA_SERVER_SECRET = "test-secret"; const AUTH_HEADERS = { "Content-Type": "application/json", diff --git a/apps/media-server/src/__tests__/setup.ts b/apps/media-server/src/__tests__/setup.ts new file mode 100644 index 00000000000..8f6bb606ef7 --- /dev/null +++ b/apps/media-server/src/__tests__/setup.ts @@ -0,0 +1 @@ +process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS ??= "true"; diff --git a/apps/media-server/src/app.ts b/apps/media-server/src/app.ts index 162a7652dd0..2a780327762 100644 --- a/apps/media-server/src/app.ts +++ b/apps/media-server/src/app.ts @@ -1,6 +1,7 @@ import { Hono } from "hono"; import { logger } from "hono/logger"; import { beginRequest } from "./lib/lifecycle"; +import { urlPolicyMiddleware } from "./lib/url-policy"; import audio from "./routes/audio"; import health from "./routes/health"; import video from "./routes/video"; @@ -17,6 +18,8 @@ app.use("*", async (c, next) => { end(); } }); +app.use("/audio/*", urlPolicyMiddleware); +app.use("/video/*", urlPolicyMiddleware); app.route("/health", health); app.route("/audio", audio); diff --git a/apps/media-server/src/lib/audio-levels.ts b/apps/media-server/src/lib/audio-levels.ts index 90b639adf46..9d286b63135 100644 --- a/apps/media-server/src/lib/audio-levels.ts +++ b/apps/media-server/src/lib/audio-levels.ts @@ -9,6 +9,7 @@ import { import { withMediaTransfers } from "./media-transfer"; import { uploadFileToStorage } from "./media-video"; import { verifyRemoteRecordingBytes } from "./recording-verification"; +import { fetchAllowed } from "./url-policy"; import { tryAcquireDirectVideoProcessSlot } from "./video-capacity"; const preparedSchema = z.object({ @@ -74,14 +75,14 @@ export async function enhanceLocalRecording(input: { AbortSignal.timeout(120_000), ]); const callback = async (body: Record) => { - const response = await fetch(webhookUrl, { + const response = await fetchAllowed(webhookUrl, { + redirect: "error", method: "POST", headers: { "Content-Type": "application/json", "x-media-server-secret": webhookSecret, }, body: JSON.stringify({ kind: "audio-levels", ...body }), - redirect: "error", signal, }); if (!response.ok) throw new Error("Audio publication unavailable"); diff --git a/apps/media-server/src/lib/auth.ts b/apps/media-server/src/lib/auth.ts index 9641d67c496..416acdddd8b 100644 --- a/apps/media-server/src/lib/auth.ts +++ b/apps/media-server/src/lib/auth.ts @@ -1,3 +1,5 @@ +import { createHash, timingSafeEqual } from "node:crypto"; + type MediaServerContext = { req: { header: (name: string) => string | undefined }; }; @@ -10,7 +12,11 @@ export function validateMediaServerSecret(c: MediaServerContext): boolean { ); return false; } - return c.req.header("x-media-server-secret") === secret; + const provided = c.req.header("x-media-server-secret"); + if (!provided) return false; + const expectedDigest = createHash("sha256").update(secret).digest(); + const providedDigest = createHash("sha256").update(provided).digest(); + return timingSafeEqual(expectedDigest, providedDigest); } export function hasMediaServerSecret(c: MediaServerContext): boolean { diff --git a/apps/media-server/src/lib/job-manager.ts b/apps/media-server/src/lib/job-manager.ts index 5c05238f20c..9cd45a6bd19 100644 --- a/apps/media-server/src/lib/job-manager.ts +++ b/apps/media-server/src/lib/job-manager.ts @@ -6,6 +6,7 @@ import { import { getContainerMemoryMetrics } from "./container-memory"; import type { MediaOperationHandle } from "./media-operations"; import type { TempFileHandle } from "./temp-files"; +import { fetchAllowed } from "./url-policy"; import { getActiveDirectVideoProcessCount } from "./video-capacity"; export type JobPhase = @@ -775,7 +776,8 @@ async function deliverWebhook( const startedAt = Date.now(); job.webhookLastAttemptAt = startedAt; try { - const resp = await fetch(job.webhookUrl, { + const resp = await fetchAllowed(job.webhookUrl, { + redirect: "error", method: "POST", headers, body, diff --git a/apps/media-server/src/lib/media-audio.ts b/apps/media-server/src/lib/media-audio.ts index eeb2096c011..700adcd0f6b 100644 --- a/apps/media-server/src/lib/media-audio.ts +++ b/apps/media-server/src/lib/media-audio.ts @@ -13,6 +13,7 @@ import { import { materializeStreamingInput } from "./media-video"; import { registerSubprocess, terminateProcess } from "./subprocess"; import { ensureTempDir, getTempDir } from "./temp-files"; +import { fetchAllowed } from "./url-policy"; export interface AudioExtractionOptions { format?: "mp3"; @@ -119,6 +120,11 @@ function getAudioExtractArgs( ): string[] { return [ "ffmpeg", + "-protocol_whitelist", + "file,pipe,crypto,data", + ...(videoUrl.toLowerCase().endsWith(".m3u8") + ? ["-allowed_extensions", "ALL"] + : []), "-i", videoUrl, "-vn", @@ -134,10 +140,13 @@ function getAudioExtractArgs( function getAudioProbeArgs(inputPath: string): string[] { const normalizedPath = (inputPath.split("?")[0] ?? "").toLowerCase(); - const args = ["ffprobe", "-v", "error"]; - if (normalizedPath.endsWith(".m3u8") || normalizedPath.endsWith(".mpd")) { - args.push("-protocol_whitelist", "file,http,https,tcp,tls,crypto,data"); - } + const args = [ + "ffprobe", + "-v", + "error", + "-protocol_whitelist", + "file,pipe,crypto,data", + ]; if (normalizedPath.endsWith(".m3u8")) { args.push( "-allowed_extensions", @@ -163,30 +172,41 @@ function isRemoteStreamingUrl(value: string): boolean { } } -async function prepareAudioProbeInput( +async function stageRemoteInput( videoUrl: string, - abortSignal: AbortSignal, + abortSignal?: AbortSignal, ): Promise<{ inputPath: string; cleanup: () => Promise }> { - if (!isRemoteStreamingUrl(videoUrl)) { + if (!/^https?:/i.test(videoUrl)) { return { inputPath: videoUrl, cleanup: async () => {} }; } await ensureTempDir(); - const dirPath = await mkdtemp(join(getTempDir(), "audio-probe-")); + const dirPath = await mkdtemp(join(getTempDir(), "audio-input-")); + const cleanup = async () => { + await rm(dirPath, { recursive: true, force: true }).catch(() => {}); + }; try { - const inputPath = await materializeStreamingInput( - videoUrl, - dirPath, - abortSignal, - ); - return { - inputPath, - cleanup: async () => { - await rm(dirPath, { recursive: true, force: true }); - }, - }; + if (isRemoteStreamingUrl(videoUrl)) { + return { + inputPath: await materializeStreamingInput( + videoUrl, + dirPath, + abortSignal, + ), + cleanup, + }; + } + const response = await fetchAllowed(videoUrl, { signal: abortSignal }); + if (!response.ok) { + throw new Error( + `Failed to fetch media: ${response.status} ${response.statusText}`, + ); + } + const inputPath = join(dirPath, "input"); + await Bun.write(inputPath, response); + return { inputPath, cleanup }; } catch (error) { - await rm(dirPath, { recursive: true, force: true }).catch(() => {}); + await cleanup(); throw error; } } @@ -270,7 +290,7 @@ export async function checkHasAudioTrack(videoUrl: string): Promise { setCancel(cancel); return await withTimeout( (async () => { - const preparedInput = await prepareAudioProbeInput( + const preparedInput = await stageRemoteInput( videoUrl, abortController.signal, ); @@ -305,18 +325,31 @@ export async function extractAudio( const opts = { ...DEFAULT_OPTIONS, ...options }; return await withMediaOperation("audio", async (setCancel) => { - const proc = registerSubprocess( - spawn({ - cmd: getAudioExtractArgs(videoUrl, opts), - stdout: "pipe", - stderr: "pipe", - }), - ); - setCancel(() => terminateProcess(proc)); + const abortController = new AbortController(); + let proc: Subprocess | undefined; + let cleanupStaged = async () => {}; + const cancel = async () => { + abortController.abort(); + if (proc) await terminateProcess(proc); + await cleanupStaged(); + }; + setCancel(cancel); try { return await withTimeout( (async () => { + const staged = await stageRemoteInput( + videoUrl, + abortController.signal, + ); + cleanupStaged = staged.cleanup; + proc = registerSubprocess( + spawn({ + cmd: getAudioExtractArgs(staged.inputPath, opts), + stdout: "pipe", + stderr: "pipe", + }), + ); const stderrPromise = readStreamWithLimit( proc.stderr as ReadableStream, MAX_STDERR_BYTES, @@ -368,23 +401,36 @@ export async function extractAudio( return output; })(), opts.timeoutMs, - () => terminateProcess(proc), + cancel, ); } finally { - await terminateProcess(proc); + if (proc) await terminateProcess(proc); + await cleanupStaged(); } }); } -export function extractAudioStream( +export async function extractAudioStream( videoUrl: string, options: AudioExtractionOptions = {}, -): StreamingExtractResult { + abortSignal?: AbortSignal, +): Promise { if (!canAcceptNewAudioOperation()) { throw new Error("Server is busy, please try again later"); } const opts = { ...DEFAULT_OPTIONS, ...options }; + if (/^https?:/i.test(videoUrl)) { + return openRemoteAudioExtractStream(videoUrl, opts, abortSignal); + } + return openAudioExtractStream(videoUrl, opts); +} + +function openAudioExtractStream( + videoUrl: string, + opts: Required, + cleanupInput: () => Promise = async () => {}, +): StreamingExtractResult { const proc: Subprocess = registerSubprocess( spawn({ cmd: getAudioExtractArgs(videoUrl, opts), @@ -414,6 +460,7 @@ export function extractAudioStream( reader = null; } unregisterMediaOperation(operation); + void cleanupInput(); void terminateProcess(proc); }; const cleanup = () => finish(new Error("Audio extraction cancelled")); @@ -474,3 +521,39 @@ export function extractAudioStream( return { stream, cleanup }; } + +async function openRemoteAudioExtractStream( + videoUrl: string, + opts: Required, + abortSignal?: AbortSignal, +): Promise { + const controller = new AbortController(); + const signal = AbortSignal.any([ + controller.signal, + AbortSignal.timeout(opts.timeoutMs), + ...(abortSignal ? [abortSignal] : []), + ]); + const operation = registerMediaOperation("audio", () => controller.abort()); + const startedAt = performance.now(); + let staged: Awaited> | undefined; + try { + staged = await stageRemoteInput(videoUrl, signal); + signal.throwIfAborted(); + return openAudioExtractStream( + staged.inputPath, + { + ...opts, + timeoutMs: Math.max( + 1, + opts.timeoutMs - (performance.now() - startedAt), + ), + }, + staged.cleanup, + ); + } catch (error) { + await staged?.cleanup(); + throw error; + } finally { + unregisterMediaOperation(operation); + } +} diff --git a/apps/media-server/src/lib/media-common.ts b/apps/media-server/src/lib/media-common.ts index 875a8faba37..739ae833a3b 100644 --- a/apps/media-server/src/lib/media-common.ts +++ b/apps/media-server/src/lib/media-common.ts @@ -24,13 +24,22 @@ import { UrlSource, } from "mediabunny"; import { registerMediaEngine } from "./media-engine"; +import { fetchAllowed } from "./url-policy"; export type ProgressCallback = (progress: number, message: string) => void; export const PROCESS_TIMEOUT_MS = 45 * 60 * 1000; export const DOWNLOAD_TIMEOUT_MS = 10 * 60 * 1000; export const UPLOAD_TIMEOUT_MS = 10 * 60 * 1000; -const mediaFetch: typeof fetch = globalThis.fetch.bind(globalThis); +const mediaFetch = ((input: string | URL | Request, init?: RequestInit) => { + const url = + typeof input === "string" + ? input + : input instanceof URL + ? input.toString() + : input.url; + return fetchAllowed(url, init); +}) as typeof fetch; export function decodeThreadsFromEnv( name: string, diff --git a/apps/media-server/src/lib/media-downloads.ts b/apps/media-server/src/lib/media-downloads.ts new file mode 100644 index 00000000000..546785a348b --- /dev/null +++ b/apps/media-server/src/lib/media-downloads.ts @@ -0,0 +1,171 @@ +import { randomUUID } from "node:crypto"; +import { join } from "node:path"; +import { DOWNLOAD_TIMEOUT_MS } from "./media-common"; +import { consumeMediaTransferBytes } from "./media-transfer"; +import { fetchAllowed } from "./url-policy"; + +function positiveLimit(name: string, fallback: number): number { + const value = Number(process.env[name]); + return Number.isSafeInteger(value) && value > 0 ? value : fallback; +} + +export class MediaDownloads { + private readonly controller = new AbortController(); + readonly signal: AbortSignal; + private readonly cache = new Map>(); + private readonly pending = new Set>(); + private readonly waiting: (() => void)[] = []; + private active = 0; + private resources = 0; + private bytes = 0; + readonly maxResources = positiveLimit( + "MEDIA_SERVER_MAX_PLAYLIST_RESOURCES", + 20_000, + ); + private readonly maxBytes = positiveLimit( + "MEDIA_SERVER_MAX_PLAYLIST_BYTES", + 32 * 1024 ** 3, + ); + + constructor(signal?: AbortSignal) { + this.signal = AbortSignal.any([ + this.controller.signal, + ...(signal ? [signal] : []), + ]); + } + + async run( + operation: (downloads: MediaDownloads) => Promise, + ): Promise { + try { + return await operation(this); + } catch (error) { + this.controller.abort(error); + throw error; + } finally { + await Promise.allSettled(this.pending); + } + } + + async all(operations: Promise[]): Promise { + try { + return await Promise.all(operations); + } catch (error) { + this.controller.abort(error); + await Promise.allSettled(operations); + throw error; + } + } + + private countResource() { + this.signal.throwIfAborted(); + if (++this.resources > this.maxResources) + throw new Error("Playlist resource count limit exceeded"); + } + + private countBytes(size: number) { + if (this.bytes + size > this.maxBytes) + throw new Error("Playlist download byte limit exceeded"); + consumeMediaTransferBytes(size); + this.bytes += size; + } + + async text( + url: string, + label = "media manifest", + ): Promise<{ content: string; url: string }> { + this.countResource(); + const response = await fetchAllowed(url, { + signal: AbortSignal.any([ + this.signal, + AbortSignal.timeout(DOWNLOAD_TIMEOUT_MS), + ]), + }); + if (!response.ok) { + await response.body?.cancel(); + throw new Error( + `Failed to fetch ${label}: ${response.status} ${response.statusText}`, + ); + } + const chunks: Uint8Array[] = []; + let bytes = 0; + await this.read(response, (value) => { + bytes += value.byteLength; + if (bytes > 8 * 1024 ** 2) + throw new Error("Media manifest size limit exceeded"); + chunks.push(value); + }); + return { + content: Buffer.concat(chunks).toString("utf8"), + url: response.url || url, + }; + } + + private async read(response: Response, write: (value: Uint8Array) => void) { + if (!response.body) throw new Error("Media response has no body"); + const reader = response.body.getReader(); + try { + while (true) { + this.signal.throwIfAborted(); + const { done, value } = await reader.read(); + if (done) break; + this.countBytes(value.byteLength); + write(value); + } + } finally { + await reader.cancel().catch(() => {}); + reader.releaseLock(); + } + } + + download(url: string, directory: string): Promise { + const cached = this.cache.get(url); + if (cached) return cached; + const operation = this.downloadResource(url, directory); + this.cache.set(url, operation); + this.pending.add(operation); + void operation + .finally(() => this.pending.delete(operation)) + .catch(() => {}); + return operation; + } + + private async downloadResource( + url: string, + directory: string, + ): Promise { + this.countResource(); + if (this.active >= 4) + await new Promise((resolve) => this.waiting.push(resolve)); + else this.active++; + try { + this.signal.throwIfAborted(); + const response = await fetchAllowed(url, { + signal: AbortSignal.any([ + this.signal, + AbortSignal.timeout(DOWNLOAD_TIMEOUT_MS), + ]), + }); + if (!response.ok) { + await response.body?.cancel(); + throw new Error(`Failed to fetch media resource: ${response.status}`); + } + const extension = + new URL(url).pathname.match(/(\.[a-z0-9]{1,8})$/i)?.[1] ?? ""; + const name = `${randomUUID()}${extension}`; + const writer = Bun.file(join(directory, name)).writer(); + try { + await this.read(response, (value) => { + writer.write(value); + }); + } finally { + await writer.end(); + } + return name; + } finally { + const next = this.waiting.shift(); + if (next) next(); + else this.active--; + } + } +} diff --git a/apps/media-server/src/lib/media-probe.ts b/apps/media-server/src/lib/media-probe.ts index ef6ee8ace8f..e1dc70235a8 100644 --- a/apps/media-server/src/lib/media-probe.ts +++ b/apps/media-server/src/lib/media-probe.ts @@ -12,9 +12,9 @@ import { withMediaOperation, } from "./media-operations"; import { materializeMedia } from "./media-transfer"; +import { BlockedUrlError, fetchAllowed } from "./url-policy"; const PROBE_TIMEOUT_MS = 30_000; -const probeFetch: typeof fetch = globalThis.fetch.bind(globalThis); export { canAcceptNewProbeOperation, getActiveProbeOperationCount }; @@ -46,21 +46,29 @@ function isHttpUrl(path: string): boolean { } } -async function hasHttpNetworkFailure(path: string): Promise { +async function hasHttpNetworkFailure( + path: string, + signal?: AbortSignal, +): Promise { try { - const response = await probeFetch(path, { + const timeout = AbortSignal.timeout(10_000); + const response = await fetchAllowed(path, { headers: { Range: "bytes=0-0" }, - signal: AbortSignal.timeout(10_000), + signal: signal ? AbortSignal.any([signal, timeout]) : timeout, }); await response.body?.cancel(); return !response.ok; - } catch { + } catch (error) { + if (error instanceof BlockedUrlError) throw error; return true; } } -async function probeMedia(path: string): Promise { - if (isHttpUrl(path) && (await hasHttpNetworkFailure(path))) { +async function probeMedia( + path: string, + signal?: AbortSignal, +): Promise { + if (isHttpUrl(path) && (await hasHttpNetworkFailure(path, signal))) { throw new Error("Media input is not accessible"); } @@ -126,9 +134,14 @@ export async function probeVideo(videoUrl: string): Promise { throw new Error("Server is busy, please try again later"); } - return await withMediaOperation("probe", () => - withTimeout(probeMedia(videoUrl), PROBE_TIMEOUT_MS), - ); + return await withMediaOperation("probe", () => { + const controller = new AbortController(); + return withTimeout( + probeMedia(videoUrl, controller.signal), + PROBE_TIMEOUT_MS, + () => controller.abort(), + ); + }); } export async function probeVideoFile(filePath: string): Promise { @@ -143,7 +156,7 @@ export async function probeVideoFile(filePath: string): Promise { export async function checkVideoAccessible(videoUrl: string): Promise { try { - const response = await probeFetch(videoUrl, { + const response = await fetchAllowed(videoUrl, { method: "HEAD", signal: AbortSignal.timeout(10_000), }); diff --git a/apps/media-server/src/lib/media-transfer.ts b/apps/media-server/src/lib/media-transfer.ts index a74cbc40932..6bd8d9fbea0 100644 --- a/apps/media-server/src/lib/media-transfer.ts +++ b/apps/media-server/src/lib/media-transfer.ts @@ -10,6 +10,7 @@ import { releaseInstanceDir, sweepOrphanedInstanceDirs, } from "./temp-files"; +import { fetchAllowed } from "./url-policy"; const targetSchema = z.object({ version: z.literal(1), @@ -177,6 +178,15 @@ export async function withMediaTransfers( }); } +export function consumeMediaTransferBytes(bytes: number): void { + const context = transfers.getStore(); + if (!context) return; + context.controller.signal.throwIfAborted(); + if (context.bytes + context.reserved + bytes > context.limit) + throw new MediaTransferBudgetError(); + context.bytes += bytes; +} + export async function getMediaDownloadTarget( input: string, signal?: AbortSignal, @@ -489,7 +499,7 @@ export async function fetchMedia( init.signal ?? undefined, headers.get("if-match") ?? undefined, ); - if (!local) return fetch(input, init); + if (!local) return fetchAllowed(input, init); const range = headers.get("range"); const match = range?.match(/^bytes=(\d+)-(\d*)$/); const start = match ? Number(match[1]) : 0; diff --git a/apps/media-server/src/lib/media-video.ts b/apps/media-server/src/lib/media-video.ts index e1030ae5f67..9be5ca2f7a6 100644 --- a/apps/media-server/src/lib/media-video.ts +++ b/apps/media-server/src/lib/media-video.ts @@ -21,6 +21,7 @@ import { UPLOAD_TIMEOUT_MS, withTimeout, } from "./media-common"; +import { MediaDownloads } from "./media-downloads"; import { probeVideoFile } from "./media-probe"; import { fetchMedia, materializeMedia } from "./media-transfer"; import { @@ -52,6 +53,7 @@ import { getTempDir, type TempFileHandle, } from "./temp-files"; +import { fetchAllowed } from "./url-policy"; const PROCESS_TIMEOUT_PER_SECOND_MS = 20_000; const MAX_PROCESS_TIMEOUT_MS = 2 * 60 * 60 * 1000; @@ -303,29 +305,26 @@ export async function materializeHlsPlaylist( dirPath: string, cache = new Map(), abortSignal?: AbortSignal, + downloads?: MediaDownloads, ): Promise { + if (!downloads) + return new MediaDownloads(abortSignal).run((group) => + materializeHlsPlaylist(playlistUrl, dirPath, cache, group.signal, group), + ); + abortSignal = downloads.signal; const cached = cache.get(playlistUrl); if (cached) return cached; - const response = await fetch(playlistUrl, { - signal: getFetchSignal(DOWNLOAD_TIMEOUT_MS, abortSignal), - }); - - if (!response.ok) { - throw new Error( - `Failed to fetch HLS playlist: ${response.status} ${response.statusText}`, - ); - } - - const content = await response.text(); - const parsedUrl = new URL(playlistUrl); + const manifest = await downloads.text(playlistUrl, "HLS playlist"); + const content = manifest.content; + const parsedUrl = new URL(manifest.url); const baseUrl = new URL(".", parsedUrl).toString(); const query = parsedUrl.search; const filePath = join(dirPath, `${randomUUID()}.m3u8`); cache.set(playlistUrl, filePath); - const lines = await Promise.all( + const lines = await downloads.all( content.split("\n").map(async (line) => { const trimmed = line.trim(); @@ -333,9 +332,16 @@ export async function materializeHlsPlaylist( if (!trimmed.startsWith("#")) { const resolved = resolveResourceUrl(trimmed, baseUrl, query); - return isHlsUrl(resolved) - ? await materializeHlsPlaylist(resolved, dirPath, cache, abortSignal) - : resolved; + if (isHlsUrl(resolved)) { + return await materializeHlsPlaylist( + resolved, + dirPath, + cache, + abortSignal, + downloads, + ); + } + return await downloads.download(resolved, dirPath); } if (!line.includes('URI="')) return line; @@ -349,8 +355,14 @@ export async function materializeHlsPlaylist( const resolved = resolveResourceUrl(original, baseUrl, query); const replacement = isHlsUrl(resolved) - ? await materializeHlsPlaylist(resolved, dirPath, cache, abortSignal) - : resolved; + ? await materializeHlsPlaylist( + resolved, + dirPath, + cache, + abortSignal, + downloads, + ) + : await downloads.download(resolved, dirPath); rewritten = rewritten.replace( `URI="${original}"`, @@ -366,51 +378,212 @@ export async function materializeHlsPlaylist( return filePath; } -export async function materializeMpdManifest( - manifestUrl: string, - dirPath: string, - abortSignal?: AbortSignal, -): Promise { - const response = await fetch(manifestUrl, { - signal: getFetchSignal(DOWNLOAD_TIMEOUT_MS, abortSignal), - }); +interface DashBaseFrame { + inherited: string; + effective: string; + hasBaseUrl: boolean; + alternatives?: DashAlternative[]; +} - if (!response.ok) { - throw new Error( - `Failed to fetch DASH manifest: ${response.status} ${response.statusText}`, - ); +function currentDashBase(frames: DashBaseFrame[], fallback: string): string { + return frames[frames.length - 1]?.effective ?? fallback; +} + +function rememberDashResource( + resource: string, + base: string, + query: string, + urls: string[], +): string { + const resolved = resolveResourceUrl(resource, base, query); + if (resolved.includes("$")) { + throw new Error("DASH template URL cannot be fetched locally"); } + if (!new URL(resolved).pathname.endsWith("/")) urls.push(resolved); + return resolved; +} - const content = await response.text(); - const parsedUrl = new URL(manifestUrl); - const baseUrl = new URL(".", parsedUrl).toString(); - const query = parsedUrl.search; - const filePath = join(dirPath, `${randomUUID()}.mpd`); - const rewrittenElements = content.replace( - /<(BaseURL|Location)(\b[^>]*)>([\s\S]*?)<\/\1>/gi, - (_, tag: string, attributes: string, resource: string) => { - const resolved = resolveResourceUrl( - decodeXmlAttribute(resource.trim()), - baseUrl, +function rewriteDashAttributes( + attributes: string, + base: string, + query: string, + urls: string[], +): string { + return attributes.replace( + /([\w:-]+)\s*=\s*(?:"([^"]*)"|'([^']*)')/g, + ( + original: string, + attribute: string, + doubleQuoted: string | undefined, + singleQuoted: string | undefined, + ) => { + if ( + ![ + "initialization", + "media", + "sourceurl", + "xlink:href", + "href", + ].includes(attribute.toLowerCase()) + ) + return original; + const resource = doubleQuoted ?? singleQuoted ?? ""; + if (!resource) return `${attribute}=""`; + const resolved = rememberDashResource( + decodeXmlAttribute(resource), + base, query, + urls, ); - return `<${tag}${attributes}>${escapeXmlAttribute(resolved)}`; + return `${attribute}="${escapeXmlAttribute(resolved)}"`; }, ); +} - const rewritten = rewrittenElements.replace( - /(initialization|media|sourceURL|xlink:href|href)="([^"]+)"/gi, - (_, attribute: string, resource: string) => { +interface DashAlternative { + url: string; + element: string; +} + +function rewriteDashManifest( + content: string, + manifestBase: string, + query: string, +): { urls: string[]; alternatives: DashAlternative[][]; xml: string } { + const urls: string[] = []; + const alternatives: DashAlternative[][] = []; + const frames: DashBaseFrame[] = [ + { inherited: manifestBase, effective: manifestBase, hasBaseUrl: false }, + ]; + const pattern = + /||<(BaseURL|Location)\b([^>]*)>([\s\S]*?)<\/\1>|<\/([A-Za-z0-9:_-]+)\s*>|<([A-Za-z0-9:_-]+)\b([^>]*?)(\/)?\s*>/gi; + let xml = ""; + let cursor = 0; + + for (const match of content.matchAll(pattern)) { + const index = match.index ?? 0; + xml += content.slice(cursor, index); + cursor = index + match[0].length; + if (match[0].startsWith("/g, ""); - const parsedUrl = new URL(manifestUrl); + const parsedUrl = new URL(manifest.url); const manifestBaseUrl = new URL(".", parsedUrl).toString(); const query = parsedUrl.search; const mpdAttrs = parseXmlAttributes( @@ -776,6 +951,10 @@ export async function materializeMpdAsHlsPlaylist( parseIsoDurationSeconds(periodAttrs.duration) ?? parseIsoDurationSeconds(mpdAttrs.mediaPresentationDuration); const playlists: DashRepresentationPlaylist[] = []; + const periodTemplate = getFirstXmlElementBlock( + periodContent.split(/ entry?.content ?? "") + .find((content) => + getFirstXmlElementBlock(content, "SegmentTimeline"), + ) ?? template.content; const templateAttrs = { + ...(periodTemplate?.attributes ?? {}), ...(adaptationTemplate?.attributes ?? {}), ...template.attributes, }; @@ -836,17 +1023,25 @@ export async function materializeMpdAsHlsPlaylist( const initUrl = resolveResourceUrl(initResource, resourceBaseUrl, query); const segments = buildDashSegments( templateAttrs, - template.content, + templateContent, representationId, bandwidth, resourceBaseUrl, query, presentationDuration, + downloads.maxResources, + ); + const localInit = await downloads.download(initUrl, dirPath); + const localSegments = await downloads.all( + segments.map(async (segment) => ({ + ...segment, + url: await downloads.download(segment.url, dirPath), + })), ); const path = await writeDashRepresentationPlaylist( dirPath, - initUrl, - segments, + localInit, + localSegments, ); playlists.push({ @@ -1143,7 +1338,7 @@ export function buildStreamingDownloadFfmpegArgs( "-threads", "2", "-protocol_whitelist", - "file,http,https,tcp,tls,crypto,data", + "file,crypto,data", "-allowed_extensions", "ALL", ...(capabilities.allowedSegmentExtensions @@ -1305,6 +1500,12 @@ export async function downloadVideoToTemp( if (parallel.concurrency > 1 && maxBytes === undefined) { await downloadUrlToPathParallel(videoUrl, tempFile.path, { ...parallel, + // Ranged parts go through the same URL policy as a single download. + fetchImpl: ((input, init) => + fetchAllowed( + input instanceof Request ? input.url : input.toString(), + init, + )) as typeof fetch, signal: combinedSignal, }); if (file(tempFile.path).size === 0) { @@ -2050,33 +2251,48 @@ export async function generateThumbnail( const opts = { ...DEFAULT_THUMBNAIL_OPTIONS, ...options }; const timestamp = getThumbnailTimestamp(duration, opts.timestamp); const deadline = performance.now() + THUMBNAIL_TIMEOUT_MS; + abortSignal = getFetchSignal(THUMBNAIL_TIMEOUT_MS, abortSignal); + let localInput: { path: string; cleanup: () => Promise }; try { - return await generateThumbnailFrame( - inputPath, - opts, - timestamp, - THUMBNAIL_TIMEOUT_MS, - abortSignal, - ); + localInput = /^https?:/i.test(inputPath) + ? await downloadVideoToTemp(inputPath, undefined, abortSignal) + : { path: inputPath, cleanup: async () => {} }; } catch (error) { abortSignal?.throwIfAborted(); - const remainingMs = deadline - performance.now(); - if ( - !(error instanceof EmptyThumbnailError) || - !isAbsolute(normalizeLocalPath(inputPath)) || - timestamp <= 0 || - remainingMs <= 0 - ) { - throw error; + throw error; + } + try { + try { + return await generateThumbnailFrame( + localInput.path, + opts, + timestamp, + Math.max(1, deadline - performance.now()), + abortSignal, + true, + ); + } catch (error) { + abortSignal?.throwIfAborted(); + const remainingMs = deadline - performance.now(); + if ( + !(error instanceof EmptyThumbnailError) || + !isAbsolute(normalizeLocalPath(localInput.path)) || + timestamp <= 0 || + remainingMs <= 0 + ) { + throw error; + } + return await generateThumbnailFrame( + localInput.path, + opts, + 0, + remainingMs, + abortSignal, + true, + ); } - return await generateThumbnailFrame( - inputPath, - opts, - 0, - remainingMs, - abortSignal, - true, - ); + } finally { + await localInput.cleanup(); } } @@ -2511,7 +2727,7 @@ async function uploadWithRetry( }; if (ifNoneMatch) headers["If-None-Match"] = ifNoneMatch; - response = await fetch(presignedUrl, { + response = await fetchAllowed(presignedUrl, { method: "PUT", headers, body: bodyFactory(), @@ -2618,8 +2834,9 @@ async function postMultipartJson>( headers["x-media-server-secret"] = webhookSecret; } - return await fetch(url, { + return await fetchAllowed(url, { method: "POST", + redirect: "error", headers, body: JSON.stringify(body), signal: uploadSignal(abortSignal), @@ -2675,7 +2892,7 @@ async function uploadMultipartPart( abortSignal?.throwIfAborted(); let response: Response; try { - response = await fetch(url, { + response = await fetchAllowed(url, { method: "PUT", headers: { "Content-Length": contentLength.toString(), diff --git a/apps/media-server/src/lib/recording-verification.ts b/apps/media-server/src/lib/recording-verification.ts index 6bbe6442cea..963b14744a1 100644 --- a/apps/media-server/src/lib/recording-verification.ts +++ b/apps/media-server/src/lib/recording-verification.ts @@ -1,13 +1,14 @@ import { spawn } from "node:child_process"; import { createHash, type Hash } from "node:crypto"; import { createReadStream } from "node:fs"; -import { lstat, open, rm } from "node:fs/promises"; +import { lstat, mkdtemp, open, rm } from "node:fs/promises"; import { availableParallelism } from "node:os"; import { dirname, isAbsolute, join } from "node:path"; import { Readable } from "node:stream"; import { setTimeout as sleep } from "node:timers/promises"; import { decodeThreadsFromEnv, PROCESS_TIMEOUT_MS } from "./media-common"; import { + consumeMediaTransferBytes, fetchMedia, MediaTransferBudgetError, materializeMedia, @@ -22,6 +23,7 @@ import { readRecordingVideoTiming, } from "./recording-timing"; import { registerSubprocess, unregisterSubprocess } from "./subprocess"; +import { ensureTempDir, getTempDir } from "./temp-files"; const MAX_LOCAL_VERIFICATION_MS = 2 * 60 * 60_000; const DECODER_STALL_MS = 5 * 60_000; @@ -1300,6 +1302,69 @@ export async function verifyRemuxedRecording( }; } +async function stageRemoteRecording( + input: string, + abortSignal?: AbortSignal, + objectIdentity?: string, +): Promise<{ path: string; cleanup: () => Promise }> { + const local = await materializeMedia(input, abortSignal, objectIdentity); + if (local) return { path: local.path, cleanup: async () => {} }; + await ensureTempDir(); + const directory = await mkdtemp(join(getTempDir(), "recording-")); + const path = join(directory, "recording.mp4"); + const cleanup = () => rm(directory, { force: true, recursive: true }); + try { + abortSignal?.throwIfAborted(); + const response = await fetchMedia(input, { + headers: { + Range: "bytes=0-", + "X-Cap-Recording-Verification": "1", + ...(objectIdentity ? { "If-Match": objectIdentity } : {}), + }, + signal: abortSignal, + }); + try { + if ( + response.status === 408 || + response.status === 429 || + response.status >= 500 + ) { + throw new RecordingVerificationError( + `Recording object read failed: ${response.status}`, + true, + ); + } + if (response.status === 412) + throw new Error("Recording object changed during verification"); + if (!response.ok || !response.body) { + throw new Error(`Failed to download recording: ${response.status}`); + } + const reader = response.body.getReader(); + const writer = Bun.file(path).writer(); + try { + while (true) { + abortSignal?.throwIfAborted(); + const { done, value } = await reader.read(); + if (done) break; + consumeMediaTransferBytes(value.byteLength); + writer.write(value); + } + } finally { + await writer.end(); + await reader.cancel().catch(() => {}); + reader.releaseLock(); + } + } finally { + if (!response.body?.locked) await response.body?.cancel().catch(() => {}); + } + return { path, cleanup }; + } catch (error) { + await cleanup(); + abortSignal?.throwIfAborted(); + throw error; + } +} + export async function verifyRecording( input: string, options: RecordingVerificationOptions, @@ -1332,13 +1397,7 @@ async function decodeRecording( if (options.abortSignal?.aborted) { throw new Error("Recording verification was cancelled"); } - let remote = false; - if (isAbsolute(input)) { - const metadata = await lstat(input); - if (!metadata.isFile()) { - throw new Error("Recording verification requires a regular MP4 file"); - } - } else { + if (!isAbsolute(input)) { const url = new URL(input); if ( (url.protocol !== "http:" && url.protocol !== "https:") || @@ -1347,7 +1406,46 @@ async function decodeRecording( ) { throw new Error("Recording verification requires an HTTP(S) MP4 URL"); } - remote = true; + const deadline = new AbortController(); + const timeout = setTimeout(() => deadline.abort(), timeoutMs); + const signal = options.abortSignal + ? AbortSignal.any([deadline.signal, options.abortSignal]) + : deadline.signal; + try { + const staged = await stageRemoteRecording(input, signal, objectIdentity); + try { + return await decodeRecording( + staged.path, + { + ...options, + abortSignal: signal, + timeoutMs: Math.max(1, timeoutMs - (performance.now() - startedAt)), + }, + undefined, + sourceAudioInput, + packetProof, + ); + } finally { + await staged.cleanup(); + } + } catch (error) { + if (options.abortSignal?.aborted) { + throw new Error("Recording verification was cancelled"); + } + if (deadline.signal.aborted) { + throw new RecordingVerificationError( + "Recording verification timed out", + true, + ); + } + throw error; + } finally { + clearTimeout(timeout); + } + } + const metadata = await lstat(input); + if (!metadata.isFile()) { + throw new Error("Recording verification requires a regular MP4 file"); } if (sourceAudioInput && !(await lstat(sourceAudioInput)).isFile()) { throw new Error("Recording verification requires a regular MP4 file"); @@ -1370,10 +1468,6 @@ async function decodeRecording( ? AbortSignal.any([deadline.signal, options.abortSignal]) : deadline.signal; try { - const remoteObject = remote - ? await readRemoteIdentity(input, signal, objectIdentity) - : undefined; - objectIdentity = remoteObject?.objectIdentity ?? objectIdentity; const timingBudgetMs = timeoutMs - (performance.now() - startedAt); if (timingBudgetMs <= 0) { throw new RecordingVerificationError( @@ -1382,7 +1476,6 @@ async function decodeRecording( ); } videoTiming = await readRecordingVideoTiming(input, { - remoteObject, abortSignal: signal, timeoutMs: Math.ceil(timingBudgetMs), }); @@ -1467,15 +1560,9 @@ async function decodeRecording( "-threads", decodeThreadsFromEnv("MEDIA_SERVER_VERIFY_DECODE_THREADS", 1), "-protocol_whitelist", - remote ? "http,https,tcp,tls" : "file", + "file", "-f", "mov", - ...(objectIdentity - ? [ - "-headers", - `If-Match: ${objectIdentity}\r\nX-Cap-Recording-Verification: 1\r\n`, - ] - : []), "-i", input, ...(sourceAudioInput @@ -2071,7 +2158,11 @@ export async function verifyRemoteRecording( const local = await materializeMedia(input, signal, before.objectIdentity); const evidence = await decodeRecording( local?.path ?? input, - { ...options, abortSignal: signal, timeoutMs: remainingMs }, + { + ...options, + abortSignal: signal, + timeoutMs: Math.max(1, timeoutMs - (performance.now() - startedAt)), + }, local ? undefined : before.objectIdentity, ); const remoteSha256 = diff --git a/apps/media-server/src/lib/url-policy.ts b/apps/media-server/src/lib/url-policy.ts new file mode 100644 index 00000000000..a8be8238119 --- /dev/null +++ b/apps/media-server/src/lib/url-policy.ts @@ -0,0 +1,428 @@ +import { lookup } from "node:dns/promises"; +import { BlockList, isIP } from "node:net"; +import type { MiddlewareHandler } from "hono"; +import { validateMediaServerSecret } from "./auth"; + +const blockedRanges = 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.168.0.0", 16], + ["198.18.0.0", 15], + ["224.0.0.0", 4], + ["240.0.0.0", 4], +] as const) { + blockedRanges.addSubnet(network, prefix, "ipv4"); +} +for (const [network, prefix] of [ + ["::", 96], + ["64:ff9b::", 96], + ["fc00::", 7], + ["fe80::", 10], + ["fec0::", 10], + ["ff00::", 8], +] as const) { + blockedRanges.addSubnet(network, prefix, "ipv6"); +} + +export function isPrivateAddress(address: string): boolean { + const family = isIP(address); + if (family === 4) return blockedRanges.check(address, "ipv4"); + if (family === 6) { + const mapped = address.toLowerCase().match(/^::ffff:(\d+\.\d+\.\d+\.\d+)$/); + if (mapped?.[1]) return blockedRanges.check(mapped[1], "ipv4"); + return blockedRanges.check(address, "ipv6"); + } + return true; +} + +function defaultPort(protocol: string): string { + if (protocol === "http:") return "80"; + if (protocol === "https:") return "443"; + return ""; +} + +function originOf(value: string | undefined): TrustedOrigin | null { + if (!value?.trim()) return null; + try { + const url = new URL(value.trim()); + if (url.protocol !== "http:" && url.protocol !== "https:") return null; + return { + host: normalizeHost(url.hostname), + port: url.port || defaultPort(url.protocol), + }; + } catch { + return null; + } +} + +function normalizeHost(host: string): string { + return host + .trim() + .toLowerCase() + .replace(/^\[|\]$/g, ""); +} + +export type TrustedOrigin = { host: string; port: string }; + +export type UrlPolicy = { + strictHosts: boolean; + allowedHosts: string[]; + trustedOrigins: TrustedOrigin[]; + allowPrivate: boolean; +}; + +export function getUrlPolicy( + env: Record = process.env, +): UrlPolicy { + const explicit = (env.MEDIA_SERVER_ALLOWED_HOSTS ?? "") + .split(",") + .map(normalizeHost) + .filter(Boolean); + const trustedOrigins = [ + env.S3_INTERNAL_ENDPOINT, + env.S3_PUBLIC_ENDPOINT, + env.CAP_AWS_ENDPOINT, + env.MEDIA_SERVER_WEBHOOK_URL, + env.WEB_URL, + ] + .map(originOf) + .filter((origin): origin is TrustedOrigin => Boolean(origin)); + const seen = new Set(); + return { + strictHosts: explicit.length > 0, + allowedHosts: [...new Set(explicit)], + trustedOrigins: trustedOrigins.filter((origin) => { + const key = `${origin.host}\0${origin.port}`; + if (seen.has(key)) return false; + seen.add(key); + return true; + }), + allowPrivate: env.MEDIA_SERVER_ALLOW_PRIVATE_URLS === "true", + }; +} + +function hostMatches(host: string, pattern: string): boolean { + if (pattern.startsWith("*.")) { + const suffix = pattern.slice(1); + return host.endsWith(suffix) && host.length > suffix.length; + } + return host === pattern; +} + +export class BlockedUrlError extends Error { + constructor( + readonly url: string, + readonly reason: string, + ) { + super(`URL not allowed (${reason}): ${redactUrl(url)}`); + this.name = "BlockedUrlError"; + } +} + +export function blockedUrlPayload( + error: unknown, +): { error: string; code: "URL_NOT_ALLOWED"; details: string } | null { + if (!(error instanceof BlockedUrlError)) return null; + return { + error: "URL not allowed", + code: "URL_NOT_ALLOWED", + details: error.message, + }; +} + +function redactUrl(value: string): string { + try { + const url = new URL(value); + return `${url.protocol}//${url.host}${url.pathname}`; + } catch { + return ""; + } +} + +type Resolver = (host: string) => Promise; + +const defaultResolver: Resolver = async (host) => + (await lookup(host, { all: true, verbatim: true })).map( + (entry) => entry.address, + ); + +async function checkedAddresses( + value: string, + policy: UrlPolicy = getUrlPolicy(), + resolve: Resolver = defaultResolver, +): Promise { + let url: URL; + try { + url = new URL(value); + } catch { + throw new BlockedUrlError(value, "invalid URL"); + } + if (url.protocol !== "http:" && url.protocol !== "https:") { + throw new BlockedUrlError(value, `scheme ${url.protocol} is not allowed`); + } + const host = normalizeHost(url.hostname); + const port = url.port || defaultPort(url.protocol); + const explicitlyAllowed = policy.allowedHosts.some((pattern) => + hostMatches(host, pattern), + ); + const trustedOrigin = policy.trustedOrigins.some( + (origin) => origin.host === host && origin.port === port, + ); + if (explicitlyAllowed || trustedOrigin || policy.allowPrivate) return []; + if (policy.strictHosts) { + throw new BlockedUrlError( + value, + `host ${host} is not in MEDIA_SERVER_ALLOWED_HOSTS`, + ); + } + + let addresses: string[]; + try { + addresses = isIP(host) ? [host] : await resolve(host); + } catch { + throw new BlockedUrlError(value, `host ${host} does not resolve`); + } + const privateAddress = addresses.find(isPrivateAddress); + if (addresses.length === 0 || privateAddress) { + throw new BlockedUrlError( + value, + `host ${host} resolves to private address ${privateAddress ?? "none"}; add it to MEDIA_SERVER_ALLOWED_HOSTS if this is your storage or Cap web host`, + ); + } + return addresses; +} + +export async function assertUrlAllowed( + value: string, + policy: UrlPolicy = getUrlPolicy(), + resolve: Resolver = defaultResolver, +): Promise { + await checkedAddresses(value, policy, resolve); +} + +const URL_KEY = /urls?$/i; + +export function collectUrls(input: unknown, key = ""): string[] { + if (typeof input === "string") { + return URL_KEY.test(key) && /^[a-z][a-z0-9+.-]*:/i.test(input) + ? [input] + : []; + } + if (Array.isArray(input)) { + return input.flatMap((item) => collectUrls(item, key)); + } + if (input && typeof input === "object") { + return Object.entries(input).flatMap(([childKey, value]) => + collectUrls(value, childKey), + ); + } + return []; +} + +const REDIRECT_STATUSES = new Set([301, 302, 303, 307, 308]); +const MAX_REDIRECTS = 3; + +type PinnedRequestInit = RequestInit & { + proxy?: false; + tls?: { serverName: string }; +}; +type Fetcher = (input: string, init?: PinnedRequestInit) => Promise; + +async function fetchChecked( + input: string, + init: RequestInit, + policy: UrlPolicy, + resolve: Resolver, + fetcher: Fetcher, +): Promise { + init.signal?.throwIfAborted(); + const addresses = await checkedAddresses(input, policy, resolve); + init.signal?.throwIfAborted(); + if (addresses.length === 0) return fetcher(input, withManualRedirect(init)); + const origin = new URL(input); + const headers = new Headers(init.headers); + headers.set("Host", origin.host); + // Dial only checked addresses. Keep the original authority for routing and + // TLS verification, and bypass proxies that could resolve the host again. + const candidates = ["GET", "HEAD"].includes( + (init.method ?? "GET").toUpperCase(), + ) + ? addresses + : addresses.slice(0, 1); + for (const [index, address] of candidates.entries()) { + const pinned = new URL(origin); + pinned.hostname = isIP(address) === 6 ? `[${address}]` : address; + try { + const response = await fetcher(pinned.toString(), { + ...withManualRedirect(init), + headers, + proxy: false, + ...(origin.protocol === "https:" + ? { tls: { serverName: normalizeHost(origin.hostname) } } + : {}), + }); + Object.defineProperty(response, "url", { + value: input, + configurable: true, + }); + return response; + } catch (error) { + init.signal?.throwIfAborted(); + if (index === candidates.length - 1) throw error; + } + } + throw new BlockedUrlError(input, "no connection address available"); +} + +async function redirectTarget( + response: Response, + current: string, +): Promise { + if (!REDIRECT_STATUSES.has(response.status)) return null; + const location = response.headers.get("location"); + if (!location) return null; + return new URL(location, current).toString(); +} + +function withManualRedirect(init: RequestInit): RequestInit { + const body = init.body; + return { + ...init, + body: body instanceof Blob ? body.slice() : body, + redirect: "manual", + }; +} + +export async function resolveAllowedUrl( + input: string, + policy: UrlPolicy = getUrlPolicy(), + resolve: Resolver = defaultResolver, + fetcher: Fetcher = fetch, + signal?: AbortSignal, +): Promise { + let current = input; + for (let hop = 0; hop <= MAX_REDIRECTS; hop++) { + let response = await fetchChecked( + current, + { + method: "HEAD", + redirect: "manual", + signal, + }, + policy, + resolve, + fetcher, + ); + if (response.status === 405 || response.status === 501) { + await response.body?.cancel(); + response = await fetchChecked( + current, + { + method: "GET", + redirect: "manual", + signal, + }, + policy, + resolve, + fetcher, + ); + } + const next = await redirectTarget(response, current); + await response.body?.cancel(); + if (!next) return current; + if (hop === MAX_REDIRECTS) { + throw new BlockedUrlError(current, "too many redirects"); + } + current = next; + } + throw new BlockedUrlError(current, "too many redirects"); +} + +export async function fetchAllowed( + input: string, + init: RequestInit = {}, + policy: UrlPolicy = getUrlPolicy(), + resolve: Resolver = defaultResolver, + fetcher: Fetcher = fetch, +): Promise { + let current = input; + for (let hop = 0; hop <= MAX_REDIRECTS; hop++) { + const response = await fetchChecked( + current, + init, + policy, + resolve, + fetcher, + ); + if ( + response.status >= 300 && + response.status < 400 && + init.redirect === "error" + ) { + await response.body?.cancel(); + throw new TypeError("Redirects are not allowed for this request"); + } + if (init.redirect === "manual") return response; + const next = await redirectTarget(response, current); + if (!next) return response; + await response.body?.cancel(); + if (hop === MAX_REDIRECTS) { + throw new BlockedUrlError(current, "too many redirects"); + } + if (new URL(next).origin !== new URL(current).origin) { + const headers = new Headers(init.headers); + for (const name of ["authorization", "cookie", "x-media-server-secret"]) + headers.delete(name); + init = { ...init, headers }; + } + if ( + response.status === 303 || + ((response.status === 301 || response.status === 302) && + init.method?.toUpperCase() === "POST") + ) { + const headers = new Headers(init.headers); + headers.delete("content-type"); + headers.delete("content-length"); + init = { ...init, method: "GET", body: undefined, headers }; + } + current = next; + } + throw new BlockedUrlError(current, "too many redirects"); +} + +export const urlPolicyMiddleware: MiddlewareHandler = async (c, next) => { + if (c.req.method !== "POST" || !validateMediaServerSecret(c)) { + return next(); + } + let body: unknown; + try { + body = await c.req.json(); + } catch { + return next(); + } + const policy = getUrlPolicy(); + const urls = [...new Set(collectUrls(body))]; + try { + await Promise.all(urls.map((url) => assertUrlAllowed(url, policy))); + } catch (error) { + if (error instanceof BlockedUrlError) { + console.warn(`[media-server] ${error.message}`); + return c.json( + { + error: "URL not allowed", + code: "URL_NOT_ALLOWED", + details: error.message, + }, + 400, + ); + } + throw error; + } + return next(); +}; diff --git a/apps/media-server/src/routes/audio.ts b/apps/media-server/src/routes/audio.ts index bd4c3a12c5d..54bab5870cd 100644 --- a/apps/media-server/src/routes/audio.ts +++ b/apps/media-server/src/routes/audio.ts @@ -16,6 +16,7 @@ import { extractAudioStream, getActiveAudioOperationCount as getActiveProcessCount, } from "../lib/media-audio"; +import { blockedUrlPayload } from "../lib/url-policy"; const audio = new Hono(); const MEDIA_ENGINE_ERROR_CODE = ["FF", "MPEG_ERROR"].join(""); @@ -92,6 +93,9 @@ audio.post("/check", async (c) => { } catch (err) { console.error("[audio/check] Error:", err); + const blocked = blockedUrlPayload(err); + if (blocked) return c.json(blocked, 400); + if (isBusyError(err)) { return c.json( { @@ -161,7 +165,11 @@ audio.post("/extract", async (c) => { } if (useStreaming) { - const { stream, cleanup } = extractAudioStream(videoUrl); + const { stream, cleanup } = await extractAudioStream( + videoUrl, + {}, + c.req.raw.signal, + ); c.req.raw.signal.addEventListener("abort", () => { cleanup(); @@ -190,6 +198,9 @@ audio.post("/extract", async (c) => { } catch (err) { console.error("[audio/extract] Error:", err); + const blocked = blockedUrlPayload(err); + if (blocked) return c.json(blocked, 400); + if (isBusyError(err)) { return c.json( { @@ -252,10 +263,14 @@ audio.post("/convert", async (c) => { let streaming = false; try { - const { stream, cleanup } = extractAudioStream(audioUrl, { - bitrate, - timeoutMs: 15 * 60 * 1000, - }); + const { stream, cleanup } = await extractAudioStream( + audioUrl, + { + bitrate, + timeoutMs: 15 * 60 * 1000, + }, + c.req.raw.signal, + ); c.req.raw.signal.addEventListener("abort", () => { cleanup(); @@ -274,6 +289,9 @@ audio.post("/convert", async (c) => { } catch (err) { console.error("[audio/convert] Error:", err); + const blocked = blockedUrlPayload(err); + if (blocked) return c.json(blocked, 400); + if (isBusyError(err)) { return c.json( { diff --git a/apps/media-server/src/routes/video.ts b/apps/media-server/src/routes/video.ts index 9bc9662f987..f4aaf085699 100644 --- a/apps/media-server/src/routes/video.ts +++ b/apps/media-server/src/routes/video.ts @@ -97,6 +97,7 @@ import { import { getRuntimeReport } from "../lib/runtime-report"; import type { TempFileHandle } from "../lib/temp-files"; import { cleanupStaleTempFiles } from "../lib/temp-files"; +import { blockedUrlPayload } from "../lib/url-policy"; import { getActiveDirectVideoProcessCount, getMaxConcurrentDirectVideoProcesses, @@ -528,6 +529,9 @@ video.post("/probe", async (c) => { } catch (err) { console.error("[video/probe] Error:", err); + const blocked = blockedUrlPayload(err); + if (blocked) return c.json(blocked, 400); + if (isBusyError(err)) { return c.json( { @@ -608,6 +612,9 @@ video.post("/thumbnail", async (c) => { } catch (err) { console.error("[video/thumbnail] Error:", err); + const blocked = blockedUrlPayload(err); + if (blocked) return c.json(blocked, 400); + if (isBusyError(err)) { return c.json( { @@ -752,6 +759,9 @@ video.post("/convert", async (c) => { }); console.error("[video/convert] Error:", err); + const blocked = blockedUrlPayload(err); + if (blocked) return c.json(blocked, 400); + if (isBusyError(err)) { c.header("Retry-After", VIDEO_BUSY_RETRY_AFTER_SECONDS.toString()); return c.json( diff --git a/apps/web/content/docs/self-hosting.mdx b/apps/web/content/docs/self-hosting.mdx index 32e16f0f975..4bb5c9d8da4 100644 --- a/apps/web/content/docs/self-hosting.mdx +++ b/apps/web/content/docs/self-hosting.mdx @@ -106,6 +106,21 @@ By default, login links are printed to the server logs. To send emails: RESEND_FROM_DOMAIN=yourdomain.com ``` +## Media Server URL Policy + +The media server only fetches the URLs cap-web sends it, and it checks each one first: + +- Only `http` and `https` URLs are accepted. +- Hosts that resolve to loopback, private, link-local (including cloud metadata at `169.254.169.254`), CGNAT or IPv6 ULA/site-local addresses are rejected, unless the host is trusted. IPv4-compatible IPv6 (`::/96`) and NAT64 (`64:ff9b::/96`) addresses are also blocked, matching the S3 endpoint policy. +- Trusted hosts are those in `S3_INTERNAL_ENDPOINT`, `S3_PUBLIC_ENDPOINT`, `CAP_AWS_ENDPOINT`, `MEDIA_SERVER_WEBHOOK_URL` and `WEB_URL` (set them on the media server too), plus anything in `MEDIA_SERVER_ALLOWED_HOSTS`. +- When `MEDIA_SERVER_ALLOWED_HOSTS` is set, it becomes a strict allow-list: every URL must use one of those hosts or the trusted hosts above. Entries are comma-separated and may use a `*.` prefix, e.g. `minio.railway.internal,*.r2.cloudflarestorage.com`. + +If your storage lives on a private network (MinIO in Docker or on Railway's private network), set `S3_INTERNAL_ENDPOINT` on the media server or add its host to `MEDIA_SERVER_ALLOWED_HOSTS`. Otherwise processing fails with `URL_NOT_ALLOWED`. `MEDIA_SERVER_ALLOW_PRIVATE_URLS=true` turns the private-address check off for `http` and `https` URLs. It is meant for local development and tests only. + +Remote media is connected to a checked IP address while preserving the original HTTP host and TLS server name. Untrusted downloads bypass HTTP proxy environment settings so a proxy cannot resolve the hostname again. Callback requests carrying the shared secret refuse redirects. + +HLS and DASH resources are cached by URL within each materialization. The default limits are 20,000 resources, 32 GiB of downloaded data, and 8 MiB per manifest. Set `MEDIA_SERVER_MAX_PLAYLIST_RESOURCES` and `MEDIA_SERVER_MAX_PLAYLIST_BYTES` on the media-server service to change the first two limits. Downloads also respect the enclosing job's transfer budget. A failed required resource cancels the remaining transfers before temporary files are removed. + ## Optional Features ### AI Features diff --git a/docker-compose.coolify.yml b/docker-compose.coolify.yml index 628af506939..80c26a6f180 100644 --- a/docker-compose.coolify.yml +++ b/docker-compose.coolify.yml @@ -50,6 +50,9 @@ services: restart: unless-stopped environment: PORT: 3456 + S3_INTERNAL_ENDPOINT: 'http://minio:9000' + S3_PUBLIC_ENDPOINT: '${SERVICE_URL_MINIO}' + MEDIA_SERVER_WEBHOOK_URL: 'http://cap-web:3000' MEDIA_SERVER_WEBHOOK_SECRET: '${SERVICE_HEX_32_MEDIASERVER}' healthcheck: test: diff --git a/docker-compose.template.yml b/docker-compose.template.yml index 1360b3d8715..9037f229f9c 100644 --- a/docker-compose.template.yml +++ b/docker-compose.template.yml @@ -56,6 +56,8 @@ services: dockerfile: Dockerfile.standalone restart: unless-stopped environment: + S3_INTERNAL_ENDPOINT: http://minio:3902 + MEDIA_SERVER_WEBHOOK_URL: http://cap-web:3000 PORT: 3456 MEDIA_SERVER_WEBHOOK_SECRET: fe337b52749070bb7b5d2c78cff9948439ea73cbc1869ba39d350e6c24db53b1 ports: diff --git a/docker-compose.yml b/docker-compose.yml index 98badf02671..8c427a915be 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -53,6 +53,9 @@ services: restart: unless-stopped environment: PORT: 3456 + S3_INTERNAL_ENDPOINT: http://minio:9000 + S3_PUBLIC_ENDPOINT: ${S3_PUBLIC_URL:-http://localhost:9000} + MEDIA_SERVER_WEBHOOK_URL: http://cap-web:3000 MEDIA_SERVER_WEBHOOK_SECRET: ${MEDIA_SERVER_WEBHOOK_SECRET:-fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210} healthcheck: test: ["CMD", "bun", "-e", "fetch('http://127.0.0.1:3456/health').then((response) => process.exit(response.ok ? 0 : 1)).catch(() => process.exit(1))"] diff --git a/packages/local-docker/docker-compose.yml b/packages/local-docker/docker-compose.yml index 5f250568836..3a472a7fc7f 100644 --- a/packages/local-docker/docker-compose.yml +++ b/packages/local-docker/docker-compose.yml @@ -30,6 +30,7 @@ services: environment: PORT: 3456 MEDIA_SERVER_WEBHOOK_SECRET: ${MEDIA_SERVER_WEBHOOK_SECRET:-local-media-server-secret} + MEDIA_SERVER_ALLOW_PRIVATE_URLS: "true" # Local S3 Strorage minio: