diff --git a/.env.self-hosting.example b/.env.self-hosting.example new file mode 100644 index 00000000000..cd24c6bece9 --- /dev/null +++ b/.env.self-hosting.example @@ -0,0 +1,20 @@ +# Copy to .env next to docker-compose.yml and fill in every value. +# Generate each secret with: openssl rand -hex 32 +# cap-web and the media server refuse to start when these are missing or left at a known placeholder. +NEXTAUTH_SECRET= +DATABASE_ENCRYPTION_KEY= +MEDIA_SERVER_WEBHOOK_SECRET= + +# Optional: database and storage passwords (internal-only defaults are used when unset) +MYSQL_PASSWORD= +MYSQL_ROOT_PASSWORD= +MINIO_ROOT_USER=cap-admin +MINIO_ROOT_PASSWORD= + +# Public URLs +CAP_URL=http://localhost:3000 +S3_PUBLIC_URL=http://localhost:9000 + +# Optional: email. Without both values, login codes are printed to the cap-web logs. +RESEND_API_KEY= +RESEND_FROM_DOMAIN= diff --git a/apps/media-server/bunfig.toml b/apps/media-server/bunfig.toml new file mode 100644 index 00000000000..37ff1f94820 --- /dev/null +++ b/apps/media-server/bunfig.toml @@ -0,0 +1,2 @@ +[test] +preload = ["./src/__tests__/setup.ts"] diff --git a/apps/media-server/src/__tests__/lib/audio-levels.integration.test.ts b/apps/media-server/src/__tests__/lib/audio-levels.integration.test.ts index bd88592b2ca..aa373732385 100644 --- a/apps/media-server/src/__tests__/lib/audio-levels.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/audio-levels.integration.test.ts @@ -20,6 +20,7 @@ import { withMediaTransfers } from "../../lib/media-transfer"; import { processVideo } from "../../lib/media-video"; import { getActiveDirectVideoProcessCount, + getMaxConcurrentDirectVideoProcesses, tryAcquireDirectVideoProcessSlot, } from "../../lib/video-capacity"; @@ -238,15 +239,19 @@ describe("local audio levels integration", () => { expect(objects.publications()).toBe(0); }); test("leaves an occupied conversion slot available to its current owner", async () => { - const slot = tryAcquireDirectVideoProcessSlot(() => true); - expect(slot).not.toBeNull(); + // The default slot count follows the CPUs, so occupy every slot. + const max = getMaxConcurrentDirectVideoProcesses(); + const slots = Array.from({ length: max }, () => + tryAcquireDirectVideoProcessSlot(() => true), + ); + expect(slots.every((slot) => slot !== null)).toBe(true); try { const objects = storage(); expect(await call()).toEqual({ status: "unchanged", reason: "capacity" }); expect(objects.preparations()).toBe(0); - expect(getActiveDirectVideoProcessCount()).toBe(1); + expect(getActiveDirectVideoProcessCount()).toBe(max); } finally { - slot?.release(); + for (const slot of slots) slot?.release(); } }); test("cancels optional work on shutdown and releases its processing slot", async () => { diff --git a/apps/media-server/src/__tests__/lib/job-manager.test.ts b/apps/media-server/src/__tests__/lib/job-manager.test.ts index 5bada4e694c..bc1e06a4cc9 100644 --- a/apps/media-server/src/__tests__/lib/job-manager.test.ts +++ b/apps/media-server/src/__tests__/lib/job-manager.test.ts @@ -156,8 +156,10 @@ describe("durable recording workers", () => { const sending = sendWebhook(job); updateJob(job.jobId, { phase: "complete", progress: 100 }); expect(sendWebhook(job)).toBe(sending); + // The URL policy check runs before fetch, so wait for the request to start. + while (!release) await Bun.sleep(1); const started = performance.now(); - release?.(new Response(null, { status: 503 })); + release(new Response(null, { status: 503 })); await sending; expect(performance.now() - started).toBeLessThan(400); expect(maximumInFlight).toBe(1); diff --git a/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts b/apps/media-server/src/__tests__/lib/media-audio.integration.test.ts index 826d49234f2..ae1191baf60 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 @@ -201,7 +201,7 @@ describe("mediaAudio integration tests", () => { try { await expect( checkHasAudioTrack(`${server.url}video.mp4`), - ).rejects.toThrow("FFprobe exited"); + ).rejects.toThrow("Failed to fetch media: 503"); await waitForAudioOperations(beforeCount); } finally { await server.stop(true); @@ -233,6 +233,24 @@ 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 { + const { stream } = extractAudioStream(`${server.url}clip.mp4`); + await expect(readStream(stream)).rejects.toThrow( + "Failed to fetch media: 404", + ); + await waitForAudioOperations(beforeCount); + } finally { + await server.stop(true); + } + }); + test("cleans up when a real stream is cancelled", async () => { const beforeCount = getActiveProcessCount(); const { stream } = extractAudioStream(TEST_VIDEO_WITH_AUDIO); 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 1c7a4cc54eb..a91a2d9a650 100644 --- a/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-routes-real-world.integration.test.ts @@ -14,7 +14,6 @@ import { mkdtempSync, rmSync } from "node:fs"; import { readFile, writeFile } from "node:fs/promises"; import os, { tmpdir } from "node:os"; import { join } from "node:path"; -import { pathToFileURL } from "node:url"; import type appType from "../../app"; import * as audioLevels from "../../lib/audio-levels"; import * as containerCpu from "../../lib/container-cpu"; @@ -66,10 +65,6 @@ let transientFixtureFailures = 0; let permanentFixtureFailures = 0; let slowFixtureCancellations = 0; -function fileUrl(path: string) { - return pathToFileURL(path).toString(); -} - function fixtureUrl(name = "test-with-audio.mp4") { return `${baseUrl}/fixtures/${name}`; } @@ -1000,12 +995,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"), }), ); @@ -1021,7 +1016,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, }), ); @@ -1034,7 +1029,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); @@ -1046,7 +1041,7 @@ describe("media routes real-world integration tests", () => { const thumbnailResponse = await app.fetch( mediaPostRequest("/video/thumbnail", { - videoUrl: fileUrl(TEST_VIDEO_WITH_AUDIO), + videoUrl: fixtureUrl("test-with-audio.mp4"), timestamp: 0.2, width: 160, height: 120, diff --git a/apps/media-server/src/__tests__/lib/media-video.integration.test.ts b/apps/media-server/src/__tests__/lib/media-video.integration.test.ts index dd9844d100c..ae98c0d8dff 100644 --- a/apps/media-server/src/__tests__/lib/media-video.integration.test.ts +++ b/apps/media-server/src/__tests__/lib/media-video.integration.test.ts @@ -3,7 +3,6 @@ import { execFileSync } from "node:child_process"; import { existsSync, mkdtempSync, - readdirSync, readFileSync, rmSync, statSync, @@ -329,7 +328,7 @@ describe("generateThumbnail integration tests", () => { } }); - test("joins an in-flight thumbnail decoder when its worker is cancelled", async () => { + test("cancels an in-flight remote thumbnail download", async () => { let ready: (() => void) | undefined; const started = new Promise((resolve) => { ready = resolve; @@ -358,35 +357,8 @@ describe("generateThumbnail integration tests", () => { ); try { await withTimeout(started, 5_000); - const commands = - process.platform === "linux" - ? readdirSync("/proc") - .filter((entry) => /^\d+$/.test(entry)) - .map((pid) => { - try { - return `${pid} ${readFileSync(`/proc/${pid}/cmdline`, "utf8")}`; - } catch (error) { - if ( - error instanceof Error && - "code" in error && - (error.code === "ENOENT" || error.code === "ESRCH") - ) - return ""; - throw error; - } - }) - : execFileSync("ps", ["-axo", "pid=,command="]) - .toString() - .split("\n"); - const pids = commands - .filter( - (command) => command.includes("ffmpeg") && command.includes(input), - ) - .map((command) => Number.parseInt(command.trim(), 10)); - expect(pids).toHaveLength(1); controller.abort(reason); expect(await withTimeout(outcome, 7_000)).toBe(reason); - for (const pid of pids) expect(() => process.kill(pid, 0)).toThrow(); } finally { controller.abort(reason); await server.stop(true); @@ -645,11 +617,16 @@ describe("processVideo integration tests", () => { const originalFetch = globalThis.fetch; const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-")); - globalThis.fetch = (async () => - new Response( - '', + const requestedUrls: string[] = []; + globalThis.fetch = (async (input: RequestInfo | URL) => { + requestedUrls.push( + input instanceof Request ? input.url : input.toString(), + ); + return new Response( + '', { status: 200, statusText: "OK" }, - )) as unknown as typeof fetch; + ); + }) as unknown as typeof fetch; try { const path = await materializeMpdManifest( @@ -658,13 +635,13 @@ describe("processVideo integration tests", () => { ); const content = readFileSync(path, "utf8"); - expect(content).toContain( - "init.mp4?Policy=a&Signature=b&Key-Pair-Id=c", + expect(requestedUrls).toContain( + "https://cdn.example/video/init.mp4?Policy=a&Signature=b&Key-Pair-Id=c", ); - expect(content).toContain( - "chunk-$Number$.m4s?Policy=a&Signature=b&Key-Pair-Id=c", + expect(requestedUrls).toContain( + "https://cdn.example/video/escaped.mp4?part=1&token=x", ); - expect(content).toContain("escaped.mp4?part=1&token=x"); + expect(content).not.toContain("https://cdn.example"); expect(content).not.toContain("&amp;"); } finally { globalThis.fetch = originalFetch; @@ -717,21 +694,10 @@ describe("processVideo integration tests", () => { const videoPlaylist = readFileSync(videoPlaylistPath ?? "", "utf8"); const audioPlaylist = readFileSync(mediaMatch?.[1] ?? "", "utf8"); - expect(videoPlaylist).toContain( - "https://cdn.example/video/video/init-v1.mp4?Policy=a&Signature=b", - ); - expect(videoPlaylist).toContain( - "https://cdn.example/video/video/chunk-00003.m4s?Policy=a&Signature=b", - ); - expect(videoPlaylist).toContain( - "https://cdn.example/video/video/chunk-00004.m4s?Policy=a&Signature=b", - ); - expect(audioPlaylist).toContain( - "https://cdn.example/video/audio/chunk-0.m4s?Policy=a&Signature=b", - ); - expect(audioPlaylist).toContain( - "https://cdn.example/video/audio/chunk-48000.m4s?Policy=a&Signature=b", - ); + expect(videoPlaylist).not.toContain("https://cdn.example"); + expect(audioPlaylist).not.toContain("https://cdn.example"); + expect(videoPlaylist).toContain(".m4s"); + expect(audioPlaylist).toContain(".m4s"); } finally { globalThis.fetch = originalFetch; rmSync(manifestDir, { recursive: true, force: true }); @@ -741,12 +707,17 @@ describe("processVideo integration tests", () => { test("falls back to generic DASH manifest materialization for unsupported MPD shapes", async () => { const originalFetch = globalThis.fetch; const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-fallback-")); - let requests = 0; + const requestedUrls: string[] = []; - globalThis.fetch = (async () => { - requests++; - return new Response( - ` + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : input.toString(); + requestedUrls.push(url); + if (new URL(url).pathname.endsWith("/")) { + return new Response("Missing directory", { status: 404 }); + } + if (url.includes("manifest.mpd")) { + return new Response( + ` @@ -759,8 +730,10 @@ describe("processVideo integration tests", () => { `, - { status: 200, statusText: "OK" }, - ); + { status: 200, statusText: "OK" }, + ); + } + return new Response("segment", { status: 200, statusText: "OK" }); }) as unknown as typeof fetch; try { @@ -771,11 +744,153 @@ describe("processVideo integration tests", () => { const content = readFileSync(path, "utf8"); expect(path.endsWith(".mpd")).toBe(true); - expect(requests).toBe(2); - expect(content).toContain( - "https://cdn.example/video/media/?Policy=a&Signature=b", + expect(requestedUrls).toContain( + "https://cdn.example/video/media/init.mp4?Policy=a&Signature=b", + ); + expect(requestedUrls).toContain( + "https://cdn.example/video/media/seg-1.m4s?Policy=a&Signature=b", ); - expect(content).toContain("seg-1.m4s?Policy=a&Signature=b"); + expect( + requestedUrls.some((url) => new URL(url).pathname.endsWith("/")), + ).toBe(false); + expect(content).not.toContain("https://cdn.example"); + expect(content).not.toContain(""); + } finally { + globalThis.fetch = originalFetch; + rmSync(manifestDir, { recursive: true, force: true }); + } + }); + + test("ignores MPD Location and resolves sibling BaseURLs against their parent", async () => { + const originalFetch = globalThis.fetch; + const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-alternatives-")); + const requestedUrls: string[] = []; + + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : input.toString(); + requestedUrls.push(url); + if (url.includes("manifest.mpd")) { + return new Response( + ` + https://cdn.example/elsewhere/live/updated.mpd + + + + primary/video.mp4 + backup/video.mp4 + + + + + `, + { status: 200, statusText: "OK" }, + ); + } + return new Response("segment", { status: 200, statusText: "OK" }); + }) as unknown as typeof fetch; + + try { + const path = await materializeStreamingInput( + "https://cdn.example/video/manifest.mpd?Policy=a&Signature=b", + manifestDir, + ); + const content = readFileSync(path, "utf8"); + + expect(requestedUrls).toContain( + "https://cdn.example/video/primary/video.mp4?Policy=a&Signature=b", + ); + expect(requestedUrls).toContain( + "https://cdn.example/video/backup/video.mp4?Policy=a&Signature=b", + ); + expect(requestedUrls.some((url) => url.includes("primary/backup"))).toBe( + false, + ); + expect(requestedUrls.some((url) => url.includes("elsewhere"))).toBe( + false, + ); + expect(content).not.toContain(""); + expect(content).not.toContain("https://cdn.example"); + expect(content.match(//g)?.length).toBe(2); + } finally { + globalThis.fetch = originalFetch; + rmSync(manifestDir, { recursive: true, force: true }); + } + }); + + test("resolves DASH segments against the parent when MPD Location is in another directory", async () => { + const originalFetch = globalThis.fetch; + const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-location-")); + const requestedUrls: string[] = []; + + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : input.toString(); + requestedUrls.push(url); + if (url.includes("manifest.mpd")) { + return new Response( + ` + ../other/updated.mpd + + + + + + + + + + + `, + { status: 200, statusText: "OK" }, + ); + } + return new Response("segment", { status: 200, statusText: "OK" }); + }) as unknown as typeof fetch; + + try { + await materializeStreamingInput( + "https://cdn.example/video/manifest.mpd", + manifestDir, + ); + + expect(requestedUrls).toContain("https://cdn.example/video/init.mp4"); + expect(requestedUrls).toContain("https://cdn.example/video/seg-1.m4s"); + expect(requestedUrls.some((url) => url.includes("/other/"))).toBe(false); + } finally { + globalThis.fetch = originalFetch; + rmSync(manifestDir, { recursive: true, force: true }); + } + }); + + test("drops a missing alternative BaseURL instead of failing", async () => { + const originalFetch = globalThis.fetch; + const manifestDir = mkdtempSync(join(tmpdir(), "cap-mpd-alt-missing-")); + + globalThis.fetch = (async (input: RequestInfo | URL) => { + const url = input instanceof Request ? input.url : input.toString(); + if (url.includes("manifest.mpd")) { + return new Response( + `primary/video.mp4backup/video.mp4`, + { status: 200, statusText: "OK" }, + ); + } + if (url.includes("backup/")) { + return new Response("missing", { + status: 404, + statusText: "Not Found", + }); + } + return new Response("segment", { status: 200, statusText: "OK" }); + }) as unknown as typeof fetch; + + try { + const path = await materializeStreamingInput( + "https://cdn.example/video/manifest.mpd", + manifestDir, + ); + const content = readFileSync(path, "utf8"); + + expect(content.match(//g)?.length).toBe(1); + expect(content).not.toContain("https://cdn.example"); } finally { globalThis.fetch = originalFetch; rmSync(manifestDir, { recursive: true, force: true }); @@ -877,10 +992,7 @@ describe("processVideo integration tests", () => { ); } - return new Response("not found", { - status: 404, - statusText: "Not Found", - }); + return new Response("segment", { status: 200, statusText: "OK" }); }) as unknown as typeof fetch; try { @@ -896,21 +1008,24 @@ describe("processVideo integration tests", () => { expect(requestedUrls).toContain( "https://cdn.example/video/variant.m3u8?Policy=a&Signature=b", ); - expect(variantPath).toBeTruthy(); - - const variant = readFileSync(variantPath ?? "", "utf8"); - expect(variant).toContain( - 'URI="https://cdn.example/video/init.mp4?Policy=a&Signature=b"', + expect(requestedUrls).toContain( + "https://cdn.example/video/init.mp4?Policy=a&Signature=b", ); - expect(variant).toContain( - 'URI="https://cdn.example/video/key.bin?Policy=a&Signature=b"', + expect(requestedUrls).toContain( + "https://cdn.example/video/key.bin?Policy=a&Signature=b", ); - expect(variant).toContain( + expect(requestedUrls).toContain( "https://cdn.example/video/segment-1.ts?Policy=a&Signature=b", ); - expect(variant).toContain( + expect(requestedUrls).toContain( "https://cdn.example/video/segment-2.ts?Policy=a&Signature=b", ); + expect(variantPath).toBeTruthy(); + + const variant = readFileSync(variantPath ?? "", "utf8"); + expect(variant).not.toContain("https://cdn.example"); + expect(variant).toContain(".ts"); + expect(variant).toContain(".bin"); } finally { globalThis.fetch = originalFetch; rmSync(manifestDir, { recursive: true, force: true }); diff --git a/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts b/apps/media-server/src/__tests__/lib/recording-verification.integration.test.ts index 129c807bdad..8f5d8d3361d 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 @@ -1960,10 +1960,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 lifetime", () => { test.each(["cancel", "timeout"])( "joins a stalled raw terminal-duration read after %s", @@ -2373,15 +2369,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); @@ -2943,8 +2937,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); @@ -2955,7 +2947,7 @@ describe("remote recording object identity", () => { expect(isRetryableRecordingVerificationError(error)).toBe( cause === "timeout", ); - for (const pid of pids) expectExited(pid); + expect(await decoderPids(input)).toEqual([]); } finally { controller.abort(); await server.stop(true); diff --git a/apps/media-server/src/__tests__/lib/redirect-follow.test.ts b/apps/media-server/src/__tests__/lib/redirect-follow.test.ts new file mode 100644 index 00000000000..83b6a72b2a9 --- /dev/null +++ b/apps/media-server/src/__tests__/lib/redirect-follow.test.ts @@ -0,0 +1,120 @@ +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; + + afterEach(() => { + 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; + 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" }, + }); + } + return new Response(null, { + status: 302, + headers: { + location: new URL("/secret", server.url).toString(), + }, + }); + }, + }); + return { + server, + 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.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.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.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.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.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/startup-checks.test.ts b/apps/media-server/src/__tests__/lib/startup-checks.test.ts new file mode 100644 index 00000000000..209f10adfc5 --- /dev/null +++ b/apps/media-server/src/__tests__/lib/startup-checks.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, test } from "bun:test"; +import { getWebhookSecretProblem } from "../../lib/startup-checks"; + +describe("getWebhookSecretProblem", () => { + test("rejects a missing secret", () => { + expect(getWebhookSecretProblem({})).toContain("not set"); + }); + + test("rejects the old docker-compose placeholder", () => { + expect( + getWebhookSecretProblem({ + MEDIA_SERVER_WEBHOOK_SECRET: + "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210", + }), + ).toContain("placeholder"); + }); + + test("allows a placeholder only with the explicit local override", () => { + expect( + getWebhookSecretProblem({ + MEDIA_SERVER_WEBHOOK_SECRET: "local-media-server-secret", + MEDIA_SERVER_ALLOW_INSECURE_SECRET: "true", + }), + ).toBeNull(); + }); + + test("accepts a unique secret", () => { + expect( + getWebhookSecretProblem({ + MEDIA_SERVER_WEBHOOK_SECRET: "4b1f0c9e7a2d43e8b6c5f1a09d8e7c6b", + }), + ).toBeNull(); + }); +}); diff --git a/apps/media-server/src/__tests__/lib/thumbnail-options.test.ts b/apps/media-server/src/__tests__/lib/thumbnail-options.test.ts new file mode 100644 index 00000000000..230a32314a6 --- /dev/null +++ b/apps/media-server/src/__tests__/lib/thumbnail-options.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, test } from "bun:test"; +import { resolveThumbnailOptions } from "../../lib/media-video"; + +describe("resolveThumbnailOptions", () => { + test("keeps defaults for fields a request leaves undefined", () => { + expect( + resolveThumbnailOptions({ + timestamp: undefined, + width: undefined, + height: undefined, + quality: undefined, + }), + ).toEqual({ timestamp: 1, width: 1280, height: 720, quality: 85 }); + }); + + test("applies provided values", () => { + expect(resolveThumbnailOptions({ width: 640, quality: 50 })).toMatchObject({ + width: 640, + height: 720, + quality: 50, + }); + }); +}); diff --git a/apps/media-server/src/__tests__/lib/url-policy.test.ts b/apps/media-server/src/__tests__/lib/url-policy.test.ts new file mode 100644 index 00000000000..9e3129fee9a --- /dev/null +++ b/apps/media-server/src/__tests__/lib/url-policy.test.ts @@ -0,0 +1,182 @@ +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); + }); +}); + +describe("assertUrlAllowed", () => { + const open = getUrlPolicy({}); + + test("blocks cloud metadata and private hosts by default", async () => { + await expect( + assertUrlAllowed("http://169.254.169.254/latest/meta-data/", open), + ).rejects.toThrow("private address"); + await expect( + assertUrlAllowed( + "http://internal.example/x", + open, + resolveTo("10.0.0.7"), + ), + ).rejects.toThrow("private address"); + }); + + test("blocks non-http schemes", async () => { + await expect(assertUrlAllowed("file:///etc/passwd", open)).rejects.toThrow( + "scheme", + ); + }); + + test("allows public hosts when no allow-list is configured", async () => { + await expect( + assertUrlAllowed( + "https://bucket.s3.amazonaws.com/v.mp4", + open, + resolveTo("52.216.1.1"), + ), + ).resolves.toBeUndefined(); + }); + + test("allows the configured S3 endpoint on a private network", async () => { + const policy = getUrlPolicy({ S3_INTERNAL_ENDPOINT: "http://minio:9000" }); + await expect( + assertUrlAllowed( + "http://minio:9000/cap/v.mp4", + policy, + resolveTo("172.18.0.3"), + ), + ).resolves.toBeUndefined(); + }); + + test("does not treat a trusted host as allowed on every port", async () => { + const policy = getUrlPolicy({ + S3_PUBLIC_ENDPOINT: "http://localhost:9000", + MEDIA_SERVER_WEBHOOK_URL: "http://cap-web:3000", + }); + await expect( + assertUrlAllowed( + "http://minio:445/cap/v.mp4", + policy, + resolveTo("10.0.0.2"), + ), + ).rejects.toThrow("private address"); + await expect( + assertUrlAllowed( + "http://localhost:6379/", + policy, + resolveTo("127.0.0.1"), + ), + ).rejects.toThrow("private address"); + await expect( + assertUrlAllowed("http://localhost:9000/bucket/v.mp4", policy), + ).resolves.toBeUndefined(); + }); + + test("blocks file URLs even when private URLs are allowed", async () => { + const policy = getUrlPolicy({ MEDIA_SERVER_ALLOW_PRIVATE_URLS: "true" }); + await expect( + assertUrlAllowed("file:///etc/passwd", policy), + ).rejects.toThrow("scheme"); + }); + + test("enforces MEDIA_SERVER_ALLOWED_HOSTS when set", async () => { + const policy = getUrlPolicy({ + MEDIA_SERVER_ALLOWED_HOSTS: "*.r2.cloudflarestorage.com", + MEDIA_SERVER_WEBHOOK_URL: "http://cap-web:3000", + }); + await expect( + assertUrlAllowed( + "https://acct.r2.cloudflarestorage.com/v.mp4", + policy, + resolveTo("104.18.1.1"), + ), + ).resolves.toBeUndefined(); + await expect( + assertUrlAllowed("http://cap-web:3000/api/webhooks", policy), + ).resolves.toBeUndefined(); + await expect( + assertUrlAllowed( + "https://evil.example/v.mp4", + policy, + resolveTo("1.1.1.1"), + ), + ).rejects.toThrow("MEDIA_SERVER_ALLOWED_HOSTS"); + }); +}); + +describe("redirect checks", () => { + test("does not follow a redirect to a private address", async () => { + const policy = getUrlPolicy({}); + const fetcher = async (input: string) => { + if (input === "https://cdn.example/v.mp4") { + return new Response(null, { + status: 302, + headers: { location: "http://169.254.169.254/latest/meta-data/" }, + }); + } + return new Response("ok"); + }; + await expect( + fetchAllowed( + "https://cdn.example/v.mp4", + {}, + policy, + resolveTo("93.184.216.34"), + fetcher, + ), + ).rejects.toThrow("private address"); + await expect( + resolveAllowedUrl( + "https://cdn.example/v.mp4", + policy, + resolveTo("93.184.216.34"), + fetcher, + ), + ).rejects.toThrow("private address"); + }); +}); + +describe("collectUrls", () => { + test("collects nested url fields only", () => { + expect( + collectUrls({ + videoUrl: "http://a/1", + name: "http://not-a-url-field", + segments: [{ url: "http://b/2" }], + videoSegmentUrls: ["http://c/3"], + }), + ).toEqual(["http://a/1", "http://b/2", "http://c/3"]); + }); +}); diff --git a/apps/media-server/src/__tests__/lib/video-capacity.test.ts b/apps/media-server/src/__tests__/lib/video-capacity.test.ts new file mode 100644 index 00000000000..d73be931803 --- /dev/null +++ b/apps/media-server/src/__tests__/lib/video-capacity.test.ts @@ -0,0 +1,16 @@ +import { describe, expect, test } from "bun:test"; +import { getDefaultMaxConcurrentDirectVideoProcesses } from "../../lib/video-capacity"; + +describe("getDefaultMaxConcurrentDirectVideoProcesses", () => { + test("keeps one conversion on small machines", () => { + expect(getDefaultMaxConcurrentDirectVideoProcesses(1)).toBe(1); + expect(getDefaultMaxConcurrentDirectVideoProcesses(2)).toBe(1); + expect(getDefaultMaxConcurrentDirectVideoProcesses(0.5)).toBe(1); + }); + + test("uses half the CPUs, capped at four", () => { + expect(getDefaultMaxConcurrentDirectVideoProcesses(4)).toBe(2); + expect(getDefaultMaxConcurrentDirectVideoProcesses(8)).toBe(4); + expect(getDefaultMaxConcurrentDirectVideoProcesses(32)).toBe(4); + }); +}); diff --git a/apps/media-server/src/__tests__/routes/audio-memory.test.ts b/apps/media-server/src/__tests__/routes/audio-memory.test.ts index 82241f830ab..8f5cd532138 100644 --- a/apps/media-server/src/__tests__/routes/audio-memory.test.ts +++ b/apps/media-server/src/__tests__/routes/audio-memory.test.ts @@ -1,9 +1,27 @@ -import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { + afterAll, + afterEach, + beforeEach, + describe, + expect, + mock, + test, +} from "bun:test"; import { join } from "node:path"; import { getActiveProcessCount } from "../../lib/media-audio"; const FIXTURES_DIR = join(import.meta.dir, "..", "fixtures"); -const TEST_VIDEO_WITH_AUDIO = `file://${join(FIXTURES_DIR, "test-with-audio.mp4")}`; +// The routes only accept http(s) URLs, so serve the fixture locally. +const fixtureServer = Bun.serve({ + port: 0, + fetch: () => + new Response(Bun.file(join(FIXTURES_DIR, "test-with-audio.mp4"))), +}); +const TEST_VIDEO_WITH_AUDIO = `${fixtureServer.url}test-with-audio.mp4`; + +afterAll(() => { + fixtureServer.stop(true); +}); const MEDIA_SERVER_SECRET = "test-secret"; const AUTH_HEADERS = { "Content-Type": "application/json", diff --git a/apps/media-server/src/__tests__/setup.ts b/apps/media-server/src/__tests__/setup.ts new file mode 100644 index 00000000000..8f6bb606ef7 --- /dev/null +++ b/apps/media-server/src/__tests__/setup.ts @@ -0,0 +1 @@ +process.env.MEDIA_SERVER_ALLOW_PRIVATE_URLS ??= "true"; diff --git a/apps/media-server/src/app.ts b/apps/media-server/src/app.ts index 162a7652dd0..2a780327762 100644 --- a/apps/media-server/src/app.ts +++ b/apps/media-server/src/app.ts @@ -1,6 +1,7 @@ import { Hono } from "hono"; import { logger } from "hono/logger"; import { beginRequest } from "./lib/lifecycle"; +import { urlPolicyMiddleware } from "./lib/url-policy"; import audio from "./routes/audio"; import health from "./routes/health"; import video from "./routes/video"; @@ -17,6 +18,8 @@ app.use("*", async (c, next) => { end(); } }); +app.use("/audio/*", urlPolicyMiddleware); +app.use("/video/*", urlPolicyMiddleware); app.route("/health", health); app.route("/audio", audio); diff --git a/apps/media-server/src/index.ts b/apps/media-server/src/index.ts index 05dbdac4fd3..bae43fd6124 100644 --- a/apps/media-server/src/index.ts +++ b/apps/media-server/src/index.ts @@ -6,8 +6,11 @@ import { cleanupMediaTransferCache, sweepMediaTransferCache, } from "./lib/media-transfer"; +import { assertSecureStartup } from "./lib/startup-checks"; import { removeOwnTempDir, startTempDirMaintenance } from "./lib/temp-files"; +assertSecureStartup(); + const port = Number(process.env.PORT) || 3456; console.log(`[media-server] Starting on port ${port}`); diff --git a/apps/media-server/src/lib/audio-levels.ts b/apps/media-server/src/lib/audio-levels.ts index 90b639adf46..273b15c393f 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,13 @@ export async function enhanceLocalRecording(input: { AbortSignal.timeout(120_000), ]); const callback = async (body: Record) => { - const response = await fetch(webhookUrl, { + const response = await fetchAllowed(webhookUrl, { method: "POST", headers: { "Content-Type": "application/json", "x-media-server-secret": webhookSecret, }, body: JSON.stringify({ kind: "audio-levels", ...body }), - redirect: "error", signal, }); if (!response.ok) throw new Error("Audio publication unavailable"); diff --git a/apps/media-server/src/lib/auth.ts b/apps/media-server/src/lib/auth.ts index 9641d67c496..416acdddd8b 100644 --- a/apps/media-server/src/lib/auth.ts +++ b/apps/media-server/src/lib/auth.ts @@ -1,3 +1,5 @@ +import { createHash, timingSafeEqual } from "node:crypto"; + type MediaServerContext = { req: { header: (name: string) => string | undefined }; }; @@ -10,7 +12,11 @@ export function validateMediaServerSecret(c: MediaServerContext): boolean { ); return false; } - return c.req.header("x-media-server-secret") === secret; + const provided = c.req.header("x-media-server-secret"); + if (!provided) return false; + const expectedDigest = createHash("sha256").update(secret).digest(); + const providedDigest = createHash("sha256").update(provided).digest(); + return timingSafeEqual(expectedDigest, providedDigest); } export function hasMediaServerSecret(c: MediaServerContext): boolean { diff --git a/apps/media-server/src/lib/job-manager.ts b/apps/media-server/src/lib/job-manager.ts index 5c05238f20c..5a8f86524a7 100644 --- a/apps/media-server/src/lib/job-manager.ts +++ b/apps/media-server/src/lib/job-manager.ts @@ -6,6 +6,7 @@ import { import { getContainerMemoryMetrics } from "./container-memory"; import type { MediaOperationHandle } from "./media-operations"; import type { TempFileHandle } from "./temp-files"; +import { fetchAllowed } from "./url-policy"; import { getActiveDirectVideoProcessCount } from "./video-capacity"; export type JobPhase = @@ -775,7 +776,7 @@ async function deliverWebhook( const startedAt = Date.now(); job.webhookLastAttemptAt = startedAt; try { - const resp = await fetch(job.webhookUrl, { + const resp = await fetchAllowed(job.webhookUrl, { 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 5e438acbd2b..a41d5140d39 100644 --- a/apps/media-server/src/lib/media-audio.ts +++ b/apps/media-server/src/lib/media-audio.ts @@ -13,6 +13,7 @@ import { import { materializeStreamingInput } from "./media-video"; import { registerSubprocess, terminateProcess } from "./subprocess"; import { ensureTempDir, getTempDir } from "./temp-files"; +import { fetchAllowed } from "./url-policy"; export interface AudioExtractionOptions { format?: "mp3"; @@ -119,6 +120,8 @@ function getAudioExtractArgs( ): string[] { return [ "ffmpeg", + "-protocol_whitelist", + "file,pipe", "-i", videoUrl, "-vn", @@ -134,10 +137,13 @@ function getAudioExtractArgs( function getAudioProbeArgs(inputPath: string): string[] { const normalizedPath = (inputPath.split("?")[0] ?? "").toLowerCase(); - const args = ["ffprobe", "-v", "error"]; - if (normalizedPath.endsWith(".m3u8") || normalizedPath.endsWith(".mpd")) { - args.push("-protocol_whitelist", "file,http,https,tcp,tls,crypto,data"); - } + const args = [ + "ffprobe", + "-v", + "error", + "-protocol_whitelist", + "file,pipe,crypto,data", + ]; if (normalizedPath.endsWith(".m3u8")) { args.push( "-allowed_extensions", @@ -163,30 +169,41 @@ function isRemoteStreamingUrl(value: string): boolean { } } -async function prepareAudioProbeInput( +async function stageRemoteInput( videoUrl: string, - abortSignal: AbortSignal, + abortSignal?: AbortSignal, ): Promise<{ inputPath: string; cleanup: () => Promise }> { - if (!isRemoteStreamingUrl(videoUrl)) { + if (!/^https?:/i.test(videoUrl)) { return { inputPath: videoUrl, cleanup: async () => {} }; } await ensureTempDir(); - const dirPath = await mkdtemp(join(getTempDir(), "audio-probe-")); + const dirPath = await mkdtemp(join(getTempDir(), "audio-input-")); + const cleanup = async () => { + await rm(dirPath, { recursive: true, force: true }).catch(() => {}); + }; try { - const inputPath = await materializeStreamingInput( - videoUrl, - dirPath, - abortSignal, - ); - return { - inputPath, - cleanup: async () => { - await rm(dirPath, { recursive: true, force: true }); - }, - }; + if (isRemoteStreamingUrl(videoUrl)) { + return { + inputPath: await materializeStreamingInput( + videoUrl, + dirPath, + abortSignal, + ), + cleanup, + }; + } + const response = await fetchAllowed(videoUrl, { signal: abortSignal }); + if (!response.ok) { + throw new Error( + `Failed to fetch media: ${response.status} ${response.statusText}`, + ); + } + const inputPath = join(dirPath, "input"); + await Bun.write(inputPath, response); + return { inputPath, cleanup }; } catch (error) { - await rm(dirPath, { recursive: true, force: true }).catch(() => {}); + await cleanup(); throw error; } } @@ -270,7 +287,7 @@ export async function checkHasAudioTrack(videoUrl: string): Promise { setCancel(cancel); return await withTimeout( (async () => { - const preparedInput = await prepareAudioProbeInput( + const preparedInput = await stageRemoteInput( videoUrl, abortController.signal, ); @@ -305,18 +322,31 @@ export async function extractAudio( const opts = { ...DEFAULT_OPTIONS, ...options }; return await withMediaOperation("audio", async (setCancel) => { - const proc = registerSubprocess( - spawn({ - cmd: getAudioExtractArgs(videoUrl, opts), - stdout: "pipe", - stderr: "pipe", - }), - ); - setCancel(() => terminateProcess(proc)); + const abortController = new AbortController(); + let proc: Subprocess | undefined; + let cleanupStaged = async () => {}; + const cancel = async () => { + abortController.abort(); + if (proc) await terminateProcess(proc); + await cleanupStaged(); + }; + setCancel(cancel); try { return await withTimeout( (async () => { + const staged = await stageRemoteInput( + videoUrl, + abortController.signal, + ); + cleanupStaged = staged.cleanup; + proc = registerSubprocess( + spawn({ + cmd: getAudioExtractArgs(staged.inputPath, opts), + stdout: "pipe", + stderr: "pipe", + }), + ); const stderrPromise = readStreamWithLimit( proc.stderr as ReadableStream, MAX_STDERR_BYTES, @@ -368,10 +398,11 @@ export async function extractAudio( return output; })(), opts.timeoutMs, - () => terminateProcess(proc), + cancel, ); } finally { - await terminateProcess(proc); + if (proc) await terminateProcess(proc); + await cleanupStaged(); } }); } @@ -385,6 +416,16 @@ export function extractAudioStream( } const opts = { ...DEFAULT_OPTIONS, ...options }; + if (/^https?:/i.test(videoUrl)) { + return openRemoteAudioExtractStream(videoUrl, opts); + } + return openAudioExtractStream(videoUrl, opts); +} + +function openAudioExtractStream( + videoUrl: string, + opts: Required, +): StreamingExtractResult { const proc: Subprocess = registerSubprocess( spawn({ cmd: getAudioExtractArgs(videoUrl, opts), @@ -460,3 +501,97 @@ export function extractAudioStream( return { stream, cleanup }; } + +function openRemoteAudioExtractStream( + videoUrl: string, + opts: Required, +): StreamingExtractResult { + const abortController = new AbortController(); + let cleaned = false; + let proc: Subprocess | undefined; + let cleanupStaged = async () => {}; + let reader: ReadableStreamDefaultReader | null = null; + let timeoutId: ReturnType | undefined; + let resolveReady: () => void = () => {}; + const ready = new Promise((resolve) => { + resolveReady = resolve; + }); + let operation: ReturnType | undefined; + + const cleanup = () => { + if (cleaned) return; + cleaned = true; + abortController.abort(); + if (timeoutId) clearTimeout(timeoutId); + if (reader) { + try { + reader.cancel().catch(() => {}); + reader.releaseLock(); + } catch {} + reader = null; + } + if (operation) unregisterMediaOperation(operation); + if (proc) void terminateProcess(proc); + void cleanupStaged(); + }; + operation = registerMediaOperation("audio", cleanup); + + timeoutId = setTimeout(cleanup, opts.timeoutMs); + + const stream = new ReadableStream( + { + async start() { + try { + const staged = await stageRemoteInput( + videoUrl, + abortController.signal, + ); + cleanupStaged = staged.cleanup; + if (cleaned) return; + proc = registerSubprocess( + spawn({ + cmd: getAudioExtractArgs(staged.inputPath, opts), + stdout: "pipe", + stderr: "pipe", + }), + ); + void drainStream(proc.stderr as ReadableStream); + proc.exited.then(() => { + cleanup(); + }); + reader = (proc.stdout as ReadableStream).getReader(); + } catch (error) { + cleanup(); + throw error; + } finally { + resolveReady(); + } + }, + async pull(controller) { + await ready; + if (!reader || cleaned) { + controller.close(); + return; + } + try { + const { done, value } = await reader.read(); + if (done) { + controller.close(); + cleanup(); + } else { + controller.enqueue(value); + } + } catch (err) { + controller.error(err); + cleanup(); + } + }, + cancel() { + cleanup(); + }, + }, + new CountQueuingStrategy({ highWaterMark: 4 }), + ); + + return { stream, cleanup }; +} diff --git a/apps/media-server/src/lib/media-common.ts b/apps/media-server/src/lib/media-common.ts index 875a8faba37..739ae833a3b 100644 --- a/apps/media-server/src/lib/media-common.ts +++ b/apps/media-server/src/lib/media-common.ts @@ -24,13 +24,22 @@ import { UrlSource, } from "mediabunny"; import { registerMediaEngine } from "./media-engine"; +import { fetchAllowed } from "./url-policy"; export type ProgressCallback = (progress: number, message: string) => void; export const PROCESS_TIMEOUT_MS = 45 * 60 * 1000; export const DOWNLOAD_TIMEOUT_MS = 10 * 60 * 1000; export const UPLOAD_TIMEOUT_MS = 10 * 60 * 1000; -const mediaFetch: typeof fetch = globalThis.fetch.bind(globalThis); +const mediaFetch = ((input: string | URL | Request, init?: RequestInit) => { + const url = + typeof input === "string" + ? input + : input instanceof URL + ? input.toString() + : input.url; + return fetchAllowed(url, init); +}) as typeof fetch; export function decodeThreadsFromEnv( name: string, diff --git a/apps/media-server/src/lib/media-probe.ts b/apps/media-server/src/lib/media-probe.ts index ef6ee8ace8f..185397d367b 100644 --- a/apps/media-server/src/lib/media-probe.ts +++ b/apps/media-server/src/lib/media-probe.ts @@ -12,9 +12,9 @@ import { withMediaOperation, } from "./media-operations"; import { materializeMedia } from "./media-transfer"; +import { fetchAllowed, getUrlPolicy, resolveAllowedUrl } from "./url-policy"; const PROBE_TIMEOUT_MS = 30_000; -const probeFetch: typeof fetch = globalThis.fetch.bind(globalThis); export { canAcceptNewProbeOperation, getActiveProbeOperationCount }; @@ -46,11 +46,15 @@ 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; @@ -59,8 +63,11 @@ async function hasHttpNetworkFailure(path: string): Promise { } } -async function probeMedia(path: string): Promise { - if (isHttpUrl(path) && (await hasHttpNetworkFailure(path))) { +async function probeMedia( + path: string, + signal?: AbortSignal, +): Promise { + if (isHttpUrl(path) && (await hasHttpNetworkFailure(path, signal))) { throw new Error("Media input is not accessible"); } @@ -126,9 +133,25 @@ 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( + (async () => { + const allowedUrl = /^https?:/i.test(videoUrl) + ? await resolveAllowedUrl( + videoUrl, + getUrlPolicy(), + undefined, + undefined, + controller.signal, + ) + : videoUrl; + return probeMedia(allowedUrl, controller.signal); + })(), + PROBE_TIMEOUT_MS, + () => controller.abort(), + ); + }); } export async function probeVideoFile(filePath: string): Promise { @@ -143,7 +166,7 @@ export async function probeVideoFile(filePath: string): Promise { export async function checkVideoAccessible(videoUrl: string): Promise { try { - const response = await probeFetch(videoUrl, { + const response = await fetchAllowed(videoUrl, { method: "HEAD", signal: AbortSignal.timeout(10_000), }); diff --git a/apps/media-server/src/lib/media-transfer.ts b/apps/media-server/src/lib/media-transfer.ts index a74cbc40932..8ae07d89c59 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), @@ -489,7 +490,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 bdff52896c7..86e2369f415 100644 --- a/apps/media-server/src/lib/media-video.ts +++ b/apps/media-server/src/lib/media-video.ts @@ -51,6 +51,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; @@ -255,6 +256,56 @@ function resolveResourceUrl( return withQuery(resolved.toString(), query); } +type Limit = (task: () => Promise) => Promise; + +function createLimiter(limit: number): Limit { + let active = 0; + const waiting: Array<() => void> = []; + return async function run(task: () => Promise): Promise { + active += 1; + if (active > limit) { + await new Promise((resolve) => waiting.push(resolve)); + } + try { + return await task(); + } finally { + active -= 1; + waiting.shift()?.(); + } + }; +} + +function localResourceName(url: string): string { + let extension = ""; + try { + const base = new URL(url).pathname.split("/").pop() ?? ""; + const match = base.match(/(\.[a-z0-9]{1,8})$/i); + if (match?.[1]) extension = match[1].toLowerCase(); + } catch {} + return `${randomUUID()}${extension}`; +} + +async function downloadAllowedResource( + url: string, + dirPath: string, + abortSignal: AbortSignal | undefined, + limit: Limit, +): Promise { + return limit(async () => { + const response = await fetchAllowed(url, { + signal: getFetchSignal(DOWNLOAD_TIMEOUT_MS, abortSignal), + }); + if (!response.ok) { + throw new Error( + `Failed to fetch media resource: ${response.status} ${response.statusText}`, + ); + } + const name = localResourceName(url); + await Bun.write(join(dirPath, name), response); + return name; + }); +} + function getFetchSignal(timeoutMs: number, abortSignal?: AbortSignal) { const timeoutSignal = AbortSignal.timeout(timeoutMs); return abortSignal @@ -300,11 +351,12 @@ export async function materializeHlsPlaylist( dirPath: string, cache = new Map(), abortSignal?: AbortSignal, + limit: Limit = createLimiter(4), ): Promise { const cached = cache.get(playlistUrl); if (cached) return cached; - const response = await fetch(playlistUrl, { + const response = await fetchAllowed(playlistUrl, { signal: getFetchSignal(DOWNLOAD_TIMEOUT_MS, abortSignal), }); @@ -330,9 +382,21 @@ 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, + limit, + ); + } + return await downloadAllowedResource( + resolved, + dirPath, + abortSignal, + limit, + ); } if (!line.includes('URI="')) return line; @@ -346,8 +410,19 @@ 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, + limit, + ) + : await downloadAllowedResource( + resolved, + dirPath, + abortSignal, + limit, + ); rewritten = rewritten.replace( `URI="${original}"`, @@ -363,12 +438,141 @@ export async function materializeHlsPlaylist( return filePath; } +interface DashBaseFrame { + inherited: string; + effective: string; + hasBaseUrl: boolean; +} + +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; +} + +function rewriteDashAttributes( + attributes: string, + base: string, + query: string, + urls: string[], +): string { + return attributes.replace( + /(initialization|media|sourceURL|xlink:href|href)="([^"]+)"/gi, + (_, attribute: string, resource: string) => { + const resolved = rememberDashResource( + decodeXmlAttribute(resource), + base, + query, + urls, + ); + 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("