fix(infra): refuse a URL transcript source that is not media from its first bytes

apps/web/__tests__/integration/developer-transcripts-database.test.ts

@@ -64,7 +64,8 @@ vi.mock("@/workflows/process-developer-video", () => ({
 vi.mock("@/lib/media-client", () => ({
 	isMediaServerConfigured: () => false,
 }));
-vi.mock("@/lib/media-container", () => ({
+vi.mock("@/lib/media-container", async (importOriginal) => ({
+	...(await importOriginal<typeof import("@/lib/media-container")>()),
 	urlIsSelfContainedMedia: async () => true,
 }));
 vi.mock("@/lib/developer-video-storage", () => ({}));

apps/web/__tests__/unit/developer-transcribe-download.test.ts

@@ -108,7 +108,8 @@ vi.mock("@/lib/public-media-fetch", async (importOriginal) => {
 	};
 });
 
-vi.mock("@/lib/media-container", () => ({
+vi.mock("@/lib/media-container", async (importOriginal) => ({
+	...(await importOriginal<typeof import("@/lib/media-container")>()),
 	urlIsSelfContainedMedia: async () => true,
 }));
 
@@ -199,6 +200,7 @@ import {
 import { developerTranscribeWorkflow } from "@/workflows/developer-transcribe";
 
 const SOURCE_KEY = "developer/_uploads/app-1/t-tr-1";
+const MEDIA_HEAD = new TextEncoder().encode("ID3");
 const SIGNED_URL = "https://media.example.com/clip.mp3?sig=secret&expires=1";
 
 type Origin = {
@@ -216,6 +218,7 @@ const serve = async (options: {
 	everyMs: number;
 	contentLength: boolean;
 	stopAfterChunks?: number;
+	head?: Uint8Array;
 }) => {
 	const state: Origin = {
 		server: createServer((_request, response) => {
@@ -238,12 +241,14 @@ const serve = async (options: {
 				sent += next;
 				chunk += 1;
 				state.served = sent;
+				const body = Buffer.alloc(next, 7);
+				if (chunk === 1) Buffer.from(options.head ?? MEDIA_HEAD).copy(body);
 				if (sent >= options.bytes) {
 					state.finishedAt = Date.now();
-					response.end(Buffer.alloc(next, 7));
+					response.end(body);
 					return;
 				}
-				response.write(Buffer.alloc(next, 7));
+				response.write(body);
 				setTimeout(tick, options.everyMs);
 			};
 			response.on("close", () => {
@@ -415,4 +420,103 @@ describe("downloading a URL source from a slow server", () => {
 		});
 		expect(balance()).toBe(100_000);
 	});
+
+	describe("a URL that is not an audio or video file", () => {
+		const notMedia = (head: Uint8Array) =>
+			serve({
+				bytes: 50 * 1024 * 1024,
+				chunks: 50 * 16,
+				everyMs: 1,
+				contentLength: true,
+				head,
+			});
+
+		it.each([
+			["an HTML page", new TextEncoder().encode("<!doctype html><html>")],
+			["a playlist", new TextEncoder().encode("#EXTM3U\n#EXT-X-VERSION:3\n")],
+			["random bytes", new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8])],
+		])(
+			"is refused from its first bytes when it is %s, with nothing stored",
+			async (_label, head) => {
+				const server = await notMedia(head);
+
+				const result = await run();
+
+				expect(result).toEqual({
+					success: false,
+					message: "The file is not a supported audio or video format",
+				});
+				expect(transcript()).toMatchObject({
+					status: "error",
+					errorCode: "unsupported_format",
+					errorMessage: "The file is not a supported audio or video format",
+					billedMicroCredits: 0,
+					reservedMicroCredits: 0,
+				});
+				expect(ledger()).toHaveLength(0);
+				expect(context.streamedPuts).toBe(0);
+				expect(context.filePuts).toEqual([]);
+				expect(context.objects.has(SOURCE_KEY)).toBe(false);
+				expect(context.extract).not.toHaveBeenCalled();
+				await new Promise((resolve) => setTimeout(resolve, 50));
+				expect(server.closedEarly).toBe(true);
+				expect(server.served).toBeLessThan(2 * 1024 * 1024);
+			},
+		);
+
+		const storedAtExtract = () => {
+			const seen: { bytes: Uint8Array | null } = { bytes: null };
+			const answer = context.extract.getMockImplementation();
+			context.extract.mockImplementationOnce(async (...args: unknown[]) => {
+				seen.bytes = context.objects.get(SOURCE_KEY) ?? null;
+				return answer?.(...args);
+			});
+			return seen;
+		};
+
+		it("still stores every byte of a media file, its first bytes included", async () => {
+			const server = await serve({
+				bytes: 40_000,
+				chunks: 7,
+				everyMs: 5,
+				contentLength: false,
+			});
+			const seen = storedAtExtract();
+
+			await run();
+
+			const stored = seen.bytes;
+			expect(stored?.length).toBe(40_000);
+			expect(
+				Buffer.from(stored ?? [])
+					.subarray(0, 3)
+					.toString(),
+			).toBe("ID3");
+			expect(stored?.[3]).toBe(7);
+			expect(stored?.[39_999]).toBe(7);
+			expect(server.closedEarly).toBe(false);
+			expect(transcript()?.status).toBe("complete");
+		});
+
+		it("still stores a media file shorter than the bytes it looks at", async () => {
+			await serve({ bytes: 1_000, chunks: 1, everyMs: 1, contentLength: true });
+			const seen = storedAtExtract();
+
+			await run();
+
+			expect(seen.bytes?.length).toBe(1_000);
+			expect(transcript()?.status).toBe("complete");
+		});
+
+		it("still says an empty file could not be downloaded", async () => {
+			await serve({ bytes: 0, chunks: 1, everyMs: 1, contentLength: false });
+
+			await run();
+
+			expect(transcript()).toMatchObject({
+				status: "error",
+				errorCode: "source_unreachable",
+			});
+		});
+	});
 });

apps/web/workflows/developer-transcribe.ts

@@ -57,7 +57,11 @@ import {
 	transcriptObjectKeys,
 	transcriptWordCount,
 } from "@/lib/developer-transcripts";
-import { urlIsSelfContainedMedia } from "@/lib/media-container";
+import {
+	MEDIA_CONTAINER_SNIFF_BYTES,
+	sniffMedia,
+	urlIsSelfContainedMedia,
+} from "@/lib/media-container";
 import { measureMp3DurationSeconds } from "@/lib/mp3-duration";
 import {
 	fetchPublicMedia,
@@ -317,6 +321,47 @@ async function spoolToFile(
 	return size;
 }
 
+async function readSourceHead(body: ReadableStream<Uint8Array>) {
+	const reader = body.getReader();
+	const chunks: Uint8Array[] = [];
+	let length = 0;
+	let finished = false;
+	for (; length < MEDIA_CONTAINER_SNIFF_BYTES; ) {
+		const { done, value } = await reader.read();
+		if (done) {
+			finished = true;
+			break;
+		}
+		chunks.push(value);
+		length += value.length;
+	}
+	const read = new Uint8Array(length);
+	let offset = 0;
+	for (const chunk of chunks) {
+		read.set(chunk, offset);
+		offset += chunk.length;
+	}
+	let sentHead = false;
+	return {
+		head: read.subarray(0, MEDIA_CONTAINER_SNIFF_BYTES),
+		cancel: () => reader.cancel().catch(() => {}),
+		body: new ReadableStream<Uint8Array>({
+			async pull(controller) {
+				if (!sentHead) {
+					sentHead = true;
+					if (read.length > 0) controller.enqueue(read);
+					if (finished) controller.close();
+					return;
+				}
+				const { done, value } = await reader.read();
+				if (done) controller.close();
+				else controller.enqueue(value);
+			},
+			cancel: (reason) => reader.cancel(reason),
+		}),
+	};
+}
+
 async function holdForSource(
 	transcriptId: string,
 	known: KnownSource,
@@ -400,20 +445,27 @@ async function prepareTranscriptSource(
 				await response.body.cancel().catch(() => {});
 				return stop(refused);
 			}
+			const source = await readSourceHead(response.body);
+			if (source.head.length > 0 && sniffMedia(source.head) !== "media") {
+				await source.cancel();
+				return stop(
+					failure("unsupported_format", FAILURE_MESSAGES.unsupported_format),
+				);
+			}
 			const contentType =
 				response.headers.get("content-type") ?? "application/octet-stream";
 			if (declared !== null && declared > TRANSCRIPT_MAX_SPOOLED_URL_BYTES) {
 				await putDeveloperObjectStream(
 					keys.source,
-					limitStreamBytes(response.body, declared),
+					limitStreamBytes(source.body, declared),
 					declared,
 					contentType,
 				);
 				sourceBytes = declared;
 			} else {
 				const bytes = await spoolToFile(
 					spoolPath,
-					response.body,
+					source.body,
 					declared ?? TRANSCRIPT_MAX_SPOOLED_URL_BYTES,
 				);
 				if (bytes === 0 || (declared !== null && bytes < declared)) {