diff --git a/apps/media-server/Dockerfile b/apps/media-server/Dockerfile index 59ac24ad25..d5ab399ad0 100644 --- a/apps/media-server/Dockerfile +++ b/apps/media-server/Dockerfile @@ -10,7 +10,7 @@ COPY apps/media-server/package.json apps/media-server/bun.lock ./ RUN bun install --frozen-lockfile --production COPY apps/media-server/src ./src -COPY apps/media-server/capcodec.md ./capcodec.md +COPY apps/media-server/capcodec.md apps/media-server/bunfig.toml ./ RUN bun test src/__tests__/lib/audio-quality-policy.test.ts src/__tests__/lib/audio-quality.integration.test.ts \ && bun test src/__tests__/lib/audio-quality-formats.integration.test.ts src/__tests__/lib/audio-quality-benchmark.test.ts \ diff --git a/apps/media-server/Dockerfile.standalone b/apps/media-server/Dockerfile.standalone index 06d1e8837d..f304d8e73d 100644 --- a/apps/media-server/Dockerfile.standalone +++ b/apps/media-server/Dockerfile.standalone @@ -10,7 +10,7 @@ COPY package.json bun.lock ./ RUN bun install --frozen-lockfile --production COPY src ./src -COPY capcodec.md ./capcodec.md +COPY capcodec.md bunfig.toml ./ ENV PORT=3456 EXPOSE 3456 diff --git a/apps/media-server/bunfig.toml b/apps/media-server/bunfig.toml new file mode 100644 index 0000000000..37ff1f9482 --- /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 0a57e77084..d15a0a71e2 100644 --- a/apps/media-server/src/__tests__/lib/job-manager.test.ts +++ b/apps/media-server/src/__tests__/lib/job-manager.test.ts @@ -157,8 +157,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 88f0163106..8f53fd11b1 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 { afterAll, beforeAll, 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 { @@ -24,8 +24,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", @@ -41,6 +48,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"), @@ -212,7 +220,7 @@ describe("mediaAudio integration tests", () => { try { await expect( checkHasAudioTrack(`${server.url}video.mp4`), - ).rejects.toThrow("FFprobe exited"); + ).rejects.toThrow("Failed to fetch media resource: 503"); await waitForAudioOperations(beforeCount); } finally { await server.stop(true); @@ -236,7 +244,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 { @@ -256,7 +266,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); @@ -271,7 +283,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); @@ -283,9 +295,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]; @@ -299,7 +314,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); @@ -307,9 +322,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 resource: 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(); @@ -317,3 +349,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 0000000000..d0bf1ba79a --- /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 0000000000..3dbdf5ab08 --- /dev/null +++ b/apps/media-server/src/__tests__/lib/media-request-budgets.integration.test.ts @@ -0,0 +1,211 @@ +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 { extractAudioStream } from "../../lib/media-audio"; +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", + "MEDIA_SERVER_MAX_PLAYLIST_BYTES", + ]) { + 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.each(["input limit", "transfer budget"])( + "bounds ordinary audio staging with the %s and removes partial files", + async (limit) => { + await ensureTempDir(); + const before = await readdir(getTempDir()); + let stagedPath = ""; + let cancelled = false; + const server = startServer(async () => { + stagedPath = + (await readdir(getTempDir())).find( + (name) => name.startsWith("audio-input-") && !before.includes(name), + ) ?? ""; + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(2048)); + }, + cancel() { + cancelled = true; + }, + }), + ); + }); + const run = () => + extractAudioStream(new URL("audio.mp4", server.url).toString(), { + timeoutMs: 1000, + }); + if (limit === "input limit") { + process.env.MEDIA_SERVER_MAX_PLAYLIST_BYTES = "1024"; + await expect(run()).rejects.toThrow("download byte limit"); + } else { + await expect(withMediaTransfers(1024, run)).rejects.toThrow( + "transfer budget", + ); + } + expect(stagedPath).not.toBe(""); + expect(await readdir(getTempDir())).not.toContain(stagedPath); + await Bun.sleep(10); + expect(cancelled).toBe(true); + }, + ); + + 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 3ef76eb181..44b1a3d551 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"; @@ -67,10 +66,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}`; } @@ -690,6 +685,63 @@ describe("media routes real-world integration tests", () => { } }, 30_000); + test("blocks a private redirect while loading cached verification runs and still completes the recording", async () => { + const previousPrivate = process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS; + const previousHosts = process.env.MEDIA_SERVER_ALLOWED_HOSTS; + process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS = "false"; + process.env.MEDIA_SERVER_ALLOWED_HOSTS = "127.0.0.1"; + let targetHits = 0; + let firstHits = 0; + const target = Bun.serve({ + hostname: "127.0.0.2", + port: 0, + fetch() { + targetHits++; + return new Response("private verification run"); + }, + }); + const redirect = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch() { + firstHits++; + return Response.redirect(target.url.toString(), 302); + }, + }); + let jobId: string | undefined; + try { + const body = fencedMuxRequest("preverified-private-redirect"); + const response = await app.fetch( + mediaPostRequest("/video/mux-segments", { + ...body, + audioInitUrl: undefined, + audioSegmentUrls: undefined, + requiredAudio: false, + sourceObjects: body.sourceObjects.filter(({ url }) => + url.includes("/video-"), + ), + expectedDuration: 1, + preverifiedRunUrls: [redirect.url.toString()], + }), + ); + expect(response.status).toBe(200); + jobId = ((await response.json()) as { jobId: string }).jobId; + expect((await waitForTerminalJob(jobId)).phase).toBe("complete"); + expect(firstHits).toBe(1); + expect(targetHits).toBe(0); + } finally { + if (jobId) deleteJob(jobId); + redirect.stop(true); + target.stop(true); + if (previousPrivate === undefined) + delete process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS; + else process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS = previousPrivate; + if (previousHosts === undefined) + delete process.env.MEDIA_SERVER_ALLOWED_HOSTS; + else process.env.MEDIA_SERVER_ALLOWED_HOSTS = previousHosts; + } + }, 30_000); + test("uploads mux previews only after the recording output is verified", async () => { const body = { ...fencedMuxRequest("mux-previews"), @@ -1188,12 +1240,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"), }), ); @@ -1209,7 +1261,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, }), ); @@ -1222,7 +1274,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); @@ -1234,7 +1286,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 2342f9a6c9..aa50458fac 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 @@ -12,6 +12,7 @@ import { import { tmpdir } from "node:os"; import { join } from "node:path"; import { withTimeout } from "../../lib/media-common"; +import { MediaDownloads } from "../../lib/media-downloads"; import { probeVideo } from "../../lib/media-probe"; import { buildStreamingDownloadFfmpegArgs, @@ -358,7 +359,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; @@ -387,35 +388,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); @@ -746,11 +720,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( @@ -759,13 +738,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; @@ -818,21 +797,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 }); @@ -842,10 +810,15 @@ describe("processVideo integration tests", () => { test("skips the zero-length closing entry Loom adds to DASH timelines", async () => { const originalFetch = globalThis.fetch; const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-zero-")); + const requestedUrls: string[] = []; - globalThis.fetch = (async () => - new Response( - ` + globalThis.fetch = Object.assign( + async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : input.toString(); + requestedUrls.push(url); + if (!new URL(url).pathname.endsWith(".mpd")) return new Response(url); + return new Response( + ` @@ -860,8 +833,11 @@ describe("processVideo integration tests", () => { `, - { status: 200, statusText: "OK" }, - )) as unknown as typeof fetch; + { status: 200, statusText: "OK" }, + ); + }, + { preconnect: originalFetch.preconnect }, + ); try { const masterPath = await materializeMpdAsHlsPlaylist( @@ -878,9 +854,32 @@ describe("processVideo integration tests", () => { "utf8", ); - expect(videoPlaylist).toContain("loom-video-0.webm?Signature=s"); - expect(videoPlaylist).toContain("loom-video-1.webm?Signature=s"); - expect(videoPlaylist).not.toContain("loom-video-2.webm"); + const segmentPaths = videoPlaylist + .split("\n") + .filter((line) => line && !line.startsWith("#")); + expect(segmentPaths).toHaveLength(2); + expect( + segmentPaths.map((path) => + readFileSync(join(manifestDir, path), "utf8"), + ), + ).toEqual([ + "https://luna.example/id/x/resource/dash/loom-video-0.webm?Signature=s", + "https://luna.example/id/x/resource/dash/loom-video-1.webm?Signature=s", + ]); + const initializationPath = videoPlaylist.match( + /#EXT-X-MAP:URI="([^"]+)"/, + )?.[1]; + expect(initializationPath).toBeDefined(); + expect( + readFileSync(join(manifestDir, initializationPath ?? ""), "utf8"), + ).toBe( + "https://luna.example/id/x/resource/dash/loom-video-init.webm?Signature=s", + ); + expect(requestedUrls).toHaveLength(4); + expect( + requestedUrls.some((url) => url.includes("loom-video-2.webm")), + ).toBe(false); + expect(videoPlaylist).not.toContain("https:"); expect(videoPlaylist.match(/#EXTINF:/g)?.length).toBe(2); expect(videoPlaylist).toContain("#EXTINF:3.539000,"); expect(videoPlaylist).toContain("#EXTINF:3.479000,"); @@ -893,12 +892,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( + ` @@ -911,8 +915,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 { @@ -923,11 +929,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( + 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(content).toContain("seg-1.m4s?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 }); @@ -1029,10 +1177,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 { @@ -1048,21 +1193,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 }); @@ -1644,9 +1792,18 @@ describe("ffmpeg-backed media utilities integration tests", () => { }); describe("DASH segment expansion", () => { + class ManifestDownloads extends MediaDownloads { + override readonly maxResources = 200_001; + + override async download(url: string): Promise { + this.countResource(); + return url; + } + } + async function withManifest( manifest: string, - run: (dir: string) => Promise, + run: (dir: string, downloads: MediaDownloads) => Promise, ): Promise { const originalFetch = globalThis.fetch; const dir = mkdtempSync(join(tmpdir(), "cap-mpd-segments-")); @@ -1656,7 +1813,9 @@ describe("DASH segment expansion", () => { statusText: "OK", })) as unknown as typeof fetch; try { - return await run(dir); + return await new ManifestDownloads().run((downloads) => + run(dir, downloads), + ); } finally { globalThis.fetch = originalFetch; rmSync(dir, { recursive: true, force: true }); @@ -1664,9 +1823,9 @@ describe("DASH segment expansion", () => { } async function materializeNormalized(url: string, manifest: string) { - return withManifest(manifest, async (dir) => { + return withManifest(manifest, async (dir, downloads) => { let master = readFileSync( - await materializeMpdAsHlsPlaylist(url, dir), + await materializeMpdAsHlsPlaylist(url, dir, undefined, downloads), "utf8", ); const paths = [...master.matchAll(/(\/[^\s"]+\.m3u8)/g)].map( @@ -1859,28 +2018,30 @@ describe("DASH segment expansion", () => { } }); - test("falls back to the generic DASH path when segments exceed the limit", async () => { - await withManifest(longTemplate(300_000), async (dir) => { + test("rejects oversized templates when the generic DASH path cannot localize them", async () => { + await withManifest(longTemplate(300_000), async (dir, downloads) => { await expect( materializeMpdAsHlsPlaylist( "https://cdn.example/long/manifest.mpd", dir, + undefined, + downloads, ), ).rejects.toThrow(/^DASH /); - const path = await materializeStreamingInput( - "https://cdn.example/long/manifest.mpd", - dir, - ); - expect(path.endsWith(".mpd")).toBe(true); + await expect( + materializeStreamingInput("https://cdn.example/long/manifest.mpd", dir), + ).rejects.toThrow("DASH template URL cannot be fetched locally"); }); }); test("counts segments across every representation against the limit", async () => { - await withManifest(longTemplate(80_000, 3), async (dir) => { + await withManifest(longTemplate(80_000, 3), async (dir, downloads) => { await expect( materializeMpdAsHlsPlaylist( "https://cdn.example/long/manifest.mpd", dir, + undefined, + downloads, ), ).rejects.toThrow(/^DASH /); }); @@ -1890,18 +2051,21 @@ describe("DASH segment expansion", () => { const startedAt = performance.now(); await withManifest( '', - async (dir) => { + async (dir, downloads) => { await expect( materializeMpdAsHlsPlaylist( "https://cdn.example/day/manifest.mpd", dir, + undefined, + downloads, ), ).rejects.toThrow(/^DASH /); - const path = await materializeStreamingInput( - "https://cdn.example/day/manifest.mpd", - dir, - ); - expect(path.endsWith(".mpd")).toBe(true); + await expect( + materializeStreamingInput( + "https://cdn.example/day/manifest.mpd", + dir, + ), + ).rejects.toThrow("DASH template URL cannot be fetched locally"); }, ); expect(performance.now() - startedAt).toBeLessThan(5_000); @@ -1925,18 +2089,21 @@ describe("DASH segment expansion", () => { const startedAt = performance.now(); await withManifest( `${timeline}`, - async (dir) => { + async (dir, downloads) => { await expect( materializeMpdAsHlsPlaylist( "https://cdn.example/r/manifest.mpd", dir, + undefined, + downloads, ), ).rejects.toThrow(/^DASH /); - const path = await materializeStreamingInput( - "https://cdn.example/r/manifest.mpd", - dir, - ); - expect(path.endsWith(".mpd")).toBe(true); + await expect( + materializeStreamingInput( + "https://cdn.example/r/manifest.mpd", + dir, + ), + ).rejects.toThrow("DASH template URL cannot be fetched locally"); }, ); expect(performance.now() - startedAt).toBeLessThan(5_000); 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 a51b40b556..be266397cc 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 433d57a174..5d0466601b 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 @@ -2097,10 +2097,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", () => { @@ -2532,15 +2528,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); @@ -4040,9 +4034,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); @@ -4092,8 +4084,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); @@ -4104,7 +4094,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 0000000000..16c4253f5c --- /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 0000000000..320214d388 --- /dev/null +++ b/apps/media-server/src/__tests__/lib/url-policy.test.ts @@ -0,0 +1,443 @@ +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.each([ + "http://[::ffff:127.0.0.1]/", + "http://[::ffff:7f00:1]/", + "http://[::ffff:169.254.169.254]/", + "http://[::ffff:a9fe:a9fe]/", + ])("blocks canonicalized IPv4-mapped private URL %s", async (url) => { + await expect(assertUrlAllowed(url, open)).rejects.toThrow( + "private address", + ); + }); + + test("allows IPv4-mapped public literals after URL canonicalization", async () => { + await expect( + assertUrlAllowed("https://[::ffff:8.8.8.8]/", open), + ).resolves.toBeUndefined(); + }); + + 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.each([ + "http://169.254.169.254/latest/meta-data/", + "http://[::ffff:127.0.0.1]/", + "http://[::ffff:7f00:1]/", + "http://[::ffff:169.254.169.254]/", + "http://[::ffff:a9fe:a9fe]/", + ])("does not follow a redirect to private URL %s", async (target) => { + const policy = getUrlPolicy({}); + const requested: string[] = []; + const fetcher = async (input: string) => { + requested.push(input); + if (input === "https://93.184.216.34/v.mp4") { + return new Response(null, { + status: 302, + headers: { location: target }, + }); + } + 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"); + expect(requested).toEqual([ + "https://93.184.216.34/v.mp4", + "https://93.184.216.34/v.mp4", + ]); + }); +}); + +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", + ]); +}); + +describe("upload connection fallback", () => { + test.each(["PUT", "POST"])( + "tries the next checked address after a refused %s connection", + async (method) => { + const seen: string[] = []; + const body = + method === "PUT" ? new Blob(["upload-content"]) : "upload-content"; + const fetcher: NonNullable[4]> = async ( + input, + init, + ) => { + seen.push(input); + expect(new Headers(init?.headers).get("host")).toBe("storage.example"); + expect(init?.tls?.serverName).toBe("storage.example"); + if (seen.length === 1) + throw Object.assign(new TypeError("Unable to connect"), { + code: "ConnectionRefused", + }); + expect(await new Response(init?.body).text()).toBe("upload-content"); + return new Response(null, { status: 204 }); + }; + const response = await fetchAllowed( + "https://storage.example/upload", + { method, body }, + getUrlPolicy({}), + resolveTo("2606:4700::6810:85e5", "93.184.216.34"), + fetcher, + ); + expect(response.status).toBe(204); + expect(seen).toEqual([ + "https://[2606:4700::6810:85e5]/upload", + "https://93.184.216.34/upload", + ]); + }, + ); + + test.each(["ECONNRESET", "EPIPE", "ETIMEDOUT", "ConnectionClosed"])( + "does not replay a POST after %s", + async (code) => { + let calls = 0; + const error = Object.assign(new TypeError("Request failed"), { code }); + await expect( + fetchAllowed( + "https://storage.example/callback", + { method: "POST", body: "recording-details" }, + getUrlPolicy({}), + resolveTo("2606:4700::6810:85e5", "93.184.216.34"), + async () => { + calls++; + throw error; + }, + ), + ).rejects.toBe(error); + expect(calls).toBe(1); + }, + ); + + test("does not replay a stream body after a connection failure", async () => { + let calls = 0; + const error = Object.assign(new TypeError("Unable to connect"), { + code: "ConnectionRefused", + }); + await expect( + fetchAllowed( + "https://storage.example/upload", + { + method: "PUT", + body: new ReadableStream({ + start(controller) { + controller.close(); + }, + }), + }, + getUrlPolicy({}), + resolveTo("2606:4700::6810:85e5", "93.184.216.34"), + async () => { + calls++; + throw error; + }, + ), + ).rejects.toBe(error); + expect(calls).toBe(1); + }); +}); diff --git a/apps/media-server/src/__tests__/loom-importer/parsing.test.ts b/apps/media-server/src/__tests__/loom-importer/parsing.test.ts index 66627b1f1e..15695456fc 100644 --- a/apps/media-server/src/__tests__/loom-importer/parsing.test.ts +++ b/apps/media-server/src/__tests__/loom-importer/parsing.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test"; import { assertAllowedSourceUrl, + createAllowedFetch, LoomSourceRejectedError, parseAllowedHosts, Semaphore, @@ -18,6 +19,36 @@ import { UnsupportedStreamError, } from "../../loom-importer/streaming"; +test("the checked Loom fetch preserves a Request body and headers", async () => { + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + return Response.json({ + method: request.method, + header: request.headers.get("x-source-header"), + body: await request.text(), + }); + }, + }); + try { + const response = await createAllowedFetch(["127.0.0.1"])( + new Request(server.url, { + method: "POST", + headers: { "x-source-header": "source-header" }, + body: "source-body", + }), + ); + expect(await response.json()).toEqual({ + method: "POST", + header: "source-header", + body: "source-body", + }); + } finally { + server.stop(true); + } +}); + describe("selectHlsTracks", () => { const master = [ "#EXTM3U", 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 82241f830a..8f5cd53213 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__/routes/audio.test.ts b/apps/media-server/src/__tests__/routes/audio.test.ts index de070066fb..7e965c25fc 100644 --- a/apps/media-server/src/__tests__/routes/audio.test.ts +++ b/apps/media-server/src/__tests__/routes/audio.test.ts @@ -1,6 +1,7 @@ import { beforeEach, describe, expect, mock, test } from "bun:test"; import app from "../../app"; import * as mediaAudio from "../../lib/media-audio"; +import { BlockedUrlError } from "../../lib/url-policy"; const MEDIA_SERVER_SECRET = "test-secret"; const AUTH_HEADERS = { @@ -10,6 +11,35 @@ const AUTH_HEADERS = { process.env.MEDIA_SERVER_WEBHOOK_SECRET = MEDIA_SERVER_SECRET; +describe("audio staging failures", () => { + beforeEach(() => mock.restore()); + + test.each(["/audio/extract", "/audio/convert"])( + "returns URL_NOT_ALLOWED JSON before streaming from %s", + async (path) => { + mock.module("../../lib/media-audio", () => ({ + ...mediaAudio, + checkHasAudioTrack: async () => true, + extractAudioStream: async () => { + await Promise.resolve(); + throw new BlockedUrlError( + "http://127.0.0.1/private", + "private address", + ); + }, + })); + const response = await app.fetch( + audioPostRequest(path, { + videoUrl: "https://example.com/video.mp4", + audioUrl: "https://example.com/video.mp4", + }), + ); + expect(response.status).toBe(400); + expect(await response.json()).toMatchObject({ code: "URL_NOT_ALLOWED" }); + }, + ); +}); + function audioPostRequest(path: string, body: unknown): Request { return new Request(`http://localhost${path}`, { method: "POST", diff --git a/apps/media-server/src/__tests__/setup.ts b/apps/media-server/src/__tests__/setup.ts new file mode 100644 index 0000000000..8f6bb606ef --- /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 96e7ca32cb..0616401973 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 liveTranscribe from "./routes/live-transcribe"; @@ -18,6 +19,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 90b639adf4..9d286b6313 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 0b0c6943cf..2eb0eeed77 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 e98215389e..d7815a2989 100644 --- a/apps/media-server/src/lib/job-manager.ts +++ b/apps/media-server/src/lib/job-manager.ts @@ -11,6 +11,7 @@ import { } from "./media-input-policy"; 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 = @@ -800,7 +801,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 41383bf42a..f8ad58d428 100644 --- a/apps/media-server/src/lib/media-audio.ts +++ b/apps/media-server/src/lib/media-audio.ts @@ -3,6 +3,7 @@ import { join } from "node:path"; import type { Subprocess } from "bun"; import { z } from "zod"; import { withTimeout } from "./media-common"; +import { MediaDownloads } from "./media-downloads"; import { canAcceptNewAudioOperation, getActiveAudioOperationCount, @@ -120,6 +121,11 @@ function getAudioExtractArgs( ): string[] { return [ "ffmpeg", + "-protocol_whitelist", + "file,pipe,crypto,data", + ...(videoUrl.toLowerCase().endsWith(".m3u8") + ? ["-allowed_extensions", "ALL"] + : []), "-i", videoUrl, "-vn", @@ -135,10 +141,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", @@ -164,30 +173,36 @@ 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, + if (isRemoteStreamingUrl(videoUrl)) { + return { + inputPath: await materializeStreamingInput( + videoUrl, + dirPath, + abortSignal, + ), + cleanup, + }; + } + const filename = await new MediaDownloads(abortSignal).run((downloads) => + downloads.download(videoUrl, dirPath), ); - return { - inputPath, - cleanup: async () => { - await rm(dirPath, { recursive: true, force: true }); - }, - }; + return { inputPath: join(dirPath, filename), cleanup }; } catch (error) { - await rm(dirPath, { recursive: true, force: true }).catch(() => {}); + await cleanup(); throw error; } } @@ -271,7 +286,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, ); @@ -306,18 +321,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, @@ -369,23 +397,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), @@ -415,6 +456,7 @@ export function extractAudioStream( reader = null; } unregisterMediaOperation(operation); + void cleanupInput(); void terminateProcess(proc); }; const cleanup = () => finish(new Error("Audio extraction cancelled")); @@ -475,3 +517,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 875a8faba3..739ae833a3 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 0000000000..704eb857a8 --- /dev/null +++ b/apps/media-server/src/lib/media-downloads.ts @@ -0,0 +1,177 @@ +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, + private readonly fetcher: ( + url: string, + init?: RequestInit, + ) => Promise = fetchAllowed, + ) { + 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; + } + } + + protected 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 this.fetcher(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 this.fetcher(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 0aaec77d51..56715aa3a4 100644 --- a/apps/media-server/src/lib/media-probe.ts +++ b/apps/media-server/src/lib/media-probe.ts @@ -17,9 +17,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 }; @@ -51,21 +51,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"); } @@ -131,9 +139,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 { @@ -188,7 +201,7 @@ export async function probeTrackSpans(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 a74cbc4093..6bd8d9fbea 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 ee25ca9cf8..a11f86a5d1 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 { assertMediaFileContent, assertTrustedStreamingUrl, @@ -59,6 +60,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; @@ -310,29 +312,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(); @@ -340,9 +339,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; @@ -356,8 +362,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}"`, @@ -373,51 +385,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)}"`; }, ); +} + +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( @@ -798,6 +968,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, }; @@ -859,19 +1041,26 @@ export async function materializeMpdAsHlsPlaylist( const initUrl = resolveResourceUrl(initResource, resourceBaseUrl, query); const segments = buildDashSegments( templateAttrs, - template.content, + templateContent, representationId, bandwidth, resourceBaseUrl, query, presentationDuration, - MAX_DASH_SEGMENTS - segmentTotal, + Math.min(downloads.maxResources, MAX_DASH_SEGMENTS - segmentTotal), + ); + const localInit = await downloads.download(initUrl, dirPath); + const localSegments = await downloads.all( + segments.map(async (segment) => ({ + ...segment, + url: await downloads.download(segment.url, dirPath), + })), ); segmentTotal += segments.length; const path = await writeDashRepresentationPlaylist( dirPath, - initUrl, - segments, + localInit, + localSegments, ); playlists.push({ @@ -941,6 +1130,7 @@ export async function materializeStreamingInput( dirPath: string, abortSignal?: AbortSignal, trustedHosts?: readonly string[], + downloads?: MediaDownloads, ): Promise { let inputPath: string; @@ -953,6 +1143,7 @@ export async function materializeStreamingInput( dirPath, undefined, abortSignal, + downloads, ); } else if (isMpdUrl(videoUrl)) { try { @@ -960,10 +1151,16 @@ export async function materializeStreamingInput( videoUrl, dirPath, abortSignal, + downloads, ); } catch (err) { if (!shouldFallbackToGenericMpd(err)) throw err; - inputPath = await materializeMpdManifest(videoUrl, dirPath, abortSignal); + inputPath = await materializeMpdManifest( + videoUrl, + dirPath, + abortSignal, + downloads, + ); } } else { return videoUrl; @@ -1173,7 +1370,7 @@ export function buildStreamingDownloadFfmpegArgs( "-threads", "2", "-protocol_whitelist", - "file,http,https,tcp,tls,crypto,data", + "file,crypto,data", "-allowed_extensions", "ALL", ...(capabilities.allowedSegmentExtensions @@ -1335,6 +1532,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) { @@ -2109,33 +2312,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(); } } @@ -2570,7 +2788,7 @@ async function uploadWithRetry( }; if (ifNoneMatch) headers["If-None-Match"] = ifNoneMatch; - response = await fetch(presignedUrl, { + response = await fetchAllowed(presignedUrl, { method: "PUT", headers, body: bodyFactory(), @@ -2677,8 +2895,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), @@ -2734,7 +2953,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 4704a17a15..ad3c14f442 100644 --- a/apps/media-server/src/lib/recording-verification.ts +++ b/apps/media-server/src/lib/recording-verification.ts @@ -1,6 +1,6 @@ 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"; @@ -10,6 +10,7 @@ import { z } from "zod"; import { decodeThreadsFromEnv, PROCESS_TIMEOUT_MS } from "./media-common"; import { spawnNode as spawn } from "./media-spawn"; import { + consumeMediaTransferBytes, fetchMedia, MediaTransferBudgetError, materializeMedia, @@ -25,6 +26,7 @@ import { readRecordingVideoTiming, } from "./recording-timing"; import { registerSubprocess, unregisterSubprocess } from "./subprocess"; +import { ensureTempDir, getTempDir } from "./temp-files"; import { digestUploadFile } from "./upload-digest"; const MAX_LOCAL_VERIFICATION_MS = 2 * 60 * 60_000; @@ -1428,6 +1430,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, @@ -1460,13 +1525,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:") || @@ -1475,7 +1534,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"); @@ -1498,10 +1596,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( @@ -1510,7 +1604,6 @@ async function decodeRecording( ); } videoTiming = await readRecordingVideoTiming(input, { - remoteObject, abortSignal: signal, timeoutMs: Math.ceil(timingBudgetMs), }); @@ -1595,15 +1688,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 @@ -2199,7 +2286,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 0000000000..5954238352 --- /dev/null +++ b/apps/media-server/src/lib/url-policy.ts @@ -0,0 +1,449 @@ +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; + +function canRetryConnection( + error: unknown, + body: RequestInit["body"], +): boolean { + const repeatable = + body == null || + typeof body === "string" || + body instanceof Blob || + body instanceof ArrayBuffer || + ArrayBuffer.isView(body) || + body instanceof URLSearchParams || + body instanceof FormData; + if (!repeatable || !error || typeof error !== "object" || !("code" in error)) + return false; + // These connect failures precede request transmission. Resets and timeouts + // can occur after a write, so they must never replay a callback or upload. + return error.code === "ConnectionRefused" || error.code === "ECONNREFUSED"; +} + +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 readOnly = ["GET", "HEAD"].includes( + (init.method ?? "GET").toUpperCase(), + ); + for (const [index, address] of addresses.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 === addresses.length - 1 || + (!readOnly && !canRetryConnection(error, init.body)) + ) + 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/loom-importer/fetch.ts b/apps/media-server/src/loom-importer/fetch.ts index 8313e1126d..051ede4661 100644 --- a/apps/media-server/src/loom-importer/fetch.ts +++ b/apps/media-server/src/loom-importer/fetch.ts @@ -1,4 +1,5 @@ import { open, rename, rm } from "node:fs/promises"; +import { fetchAllowed } from "../lib/url-policy"; export class LoomSourceRejectedError extends Error {} @@ -77,7 +78,23 @@ export function createAllowedFetch(allowedHosts: string[]): typeof fetch { ? input.href : input; assertAllowedSourceUrl(url, allowedHosts); - return fetch(input, { ...init, redirect: "error" }); + if (!(input instanceof Request)) + return fetchAllowed(url, { ...init, redirect: "error" }); + const request = new Request(input, init); + return fetchAllowed(url, { + method: request.method, + headers: request.headers, + body: request.body, + signal: request.signal, + cache: request.cache, + credentials: request.credentials, + integrity: request.integrity, + keepalive: request.keepalive, + mode: request.mode, + referrer: request.referrer, + referrerPolicy: request.referrerPolicy, + redirect: "error", + }); }; return Object.assign(allowedFetch, { preconnect: fetch.preconnect }); } diff --git a/apps/media-server/src/loom-importer/job.ts b/apps/media-server/src/loom-importer/job.ts index efe07489fd..3bf67fc01a 100644 --- a/apps/media-server/src/loom-importer/job.ts +++ b/apps/media-server/src/loom-importer/job.ts @@ -7,6 +7,7 @@ import { updateJob, type VideoMetadata, } from "../lib/job-manager"; +import { MediaDownloads } from "../lib/media-downloads"; import { trustLocalPlaylist } from "../lib/media-input-policy"; import { abortStorageUploadTarget, @@ -180,6 +181,21 @@ interface PreparedSource { skippedSeconds: number; } +class LoomManifestDownloads extends MediaDownloads { + constructor( + signal: AbortSignal, + private readonly allowedHosts: string[], + ) { + super(signal, createAllowedFetch(allowedHosts)); + } + + // The prefetch pass applies Loom's missing-segment tolerance and byte budget. + override async download(url: string): Promise { + this.countResource(); + return assertAllowedSourceUrl(url, this.allowedHosts).toString(); + } +} + async function prepareSource( job: Job, request: ImportRequest, @@ -211,11 +227,17 @@ async function prepareSource( const segmentDir = join(workDir, "segments"); await mkdir(manifestDir); await mkdir(segmentDir); - const master = await materializeStreamingInput( - request.videoUrl, - manifestDir, + const master = await new LoomManifestDownloads( signal, settings.allowedHosts, + ).run((downloads) => + materializeStreamingInput( + request.videoUrl, + manifestDir, + downloads.signal, + settings.allowedHosts, + downloads, + ), ); if (!master.endsWith(".m3u8")) throw new UnsupportedStreamError("This Loom DASH layout is not supported"); diff --git a/apps/media-server/src/routes/audio.ts b/apps/media-server/src/routes/audio.ts index d79ae3f71b..44377bc471 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(""); @@ -95,6 +96,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( { @@ -164,9 +168,11 @@ audio.post("/extract", async (c) => { } if (useStreaming) { - const { stream, cleanup } = extractAudioStream(videoUrl, { - timeoutMs: STREAMED_AUDIO_TIMEOUT_MS, - }); + const { stream, cleanup } = await extractAudioStream( + videoUrl, + { timeoutMs: STREAMED_AUDIO_TIMEOUT_MS }, + c.req.raw.signal, + ); c.req.raw.signal.addEventListener("abort", () => { cleanup(); @@ -195,6 +201,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( { @@ -257,10 +266,14 @@ audio.post("/convert", async (c) => { let streaming = false; try { - const { stream, cleanup } = extractAudioStream(audioUrl, { - bitrate, - timeoutMs: STREAMED_AUDIO_TIMEOUT_MS, - }); + const { stream, cleanup } = await extractAudioStream( + audioUrl, + { + bitrate, + timeoutMs: STREAMED_AUDIO_TIMEOUT_MS, + }, + c.req.raw.signal, + ); c.req.raw.signal.addEventListener("abort", () => { cleanup(); @@ -279,6 +292,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 7e035e7b34..491e10316d 100644 --- a/apps/media-server/src/routes/video.ts +++ b/apps/media-server/src/routes/video.ts @@ -109,6 +109,7 @@ import { import { getRuntimeReport } from "../lib/runtime-report"; import type { TempFileHandle } from "../lib/temp-files"; import { cleanupStaleTempFiles } from "../lib/temp-files"; +import { blockedUrlPayload, fetchAllowed } from "../lib/url-policy"; import { getActiveDirectVideoProcessCount, getMaxConcurrentDirectVideoProcesses, @@ -553,6 +554,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( { @@ -633,6 +637,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( { @@ -777,6 +784,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( @@ -2902,7 +2912,7 @@ async function loadPreverifiedRuns( const url = urls[next++]; if (!url) return; try { - const response = await fetch(url, { signal: budget }); + const response = await fetchAllowed(url, { signal: budget }); const text = response.ok ? await readBoundedText(response, MAX_PREVERIFIED_RUN_BYTES) : null; diff --git a/apps/web/content/docs/self-hosting.mdx b/apps/web/content/docs/self-hosting.mdx index 6a87cb1134..789791aa28 100644 --- a/apps/web/content/docs/self-hosting.mdx +++ b/apps/web/content/docs/self-hosting.mdx @@ -139,6 +139,23 @@ Email is optional. Without it, login codes are printed to the cap-web logs, the ``` 3. Verify the domain in Resend. Login emails are sent from `auth@RESEND_FROM_DOMAIN`. +## 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. + +Streaming HLS and DASH inputs must also match `MEDIA_SERVER_STREAMING_HOSTS` (a comma-separated list, default `loom.com`); HTTPS subdomains of listed hosts are accepted. The Loom importer uses its own configured host list. Set this on the media server alongside the URL policy settings for other streaming storage hosts. + +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. Ordinary remote audio inputs share the byte limit, including inputs without a Content-Length header. Downloads also respect the enclosing job's transfer budget. A failed required resource cancels the remaining transfers before temporary files are removed. + ## Media Server Capacity The media server runs a limited number of video jobs at once. Unless you set a limit, it uses the smallest of 4 and half the CPUs, and, if the container has a memory limit, the memory limit × 0.85 ÷ 768 MB (at least 1). A media server with 2 CPUs therefore processes one video at a time. diff --git a/docker-compose.coolify.yml b/docker-compose.coolify.yml index 628af50693..80c26a6f18 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 1360b3d871..9037f229f9 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 98badf0267..8c427a915b 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 5f25056883..3a472a7fc7 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: