fix(mux)!: refuse malformed container imports

Cargo.toml

@@ -107,6 +107,7 @@ criterion = "0.8"
 # (`fallback-dynamic-loading`), so nothing links against libcuda at build time.
 cudarc = { version = "0.19", default-features = false, features = ["driver", "fallback-dynamic-loading", "cuda-12020"] }
 dispatch2 = "0.3.1"
+ebml-iterable = "0.7.1"
 flate2 = { version = "1.1", default-features = false }
 futures = "0.3"
 getrandom = { version = "0.4", features = ["wasm_js"] }

quest/m1/cut-audit/README.md

@@ -65,7 +65,6 @@ leaving JS-only redesigns of `js/net` to [rs2ts](/quest/m2/rs2ts/README.md).
 - [A media hole in the player keeps the timeline](/quest/m1/cut-audit/watch-audio-hole.md) - a skipped group re-primes decoders without flushing audio or re-anchoring sync
 - [A camera broadcast stays announced across unmute](/quest/m1/cut-audit/publish-announce-latch.md) - viewers do not restart on every device flap
 - [A description-less AAC rendition means one thing](/quest/m1/cut-audit/aac-description-less.md) - every player and exporter agrees on the frame format
-- [A malformed MKV track fails the import](/quest/m1/cut-audit/mkv-import-errors.md) - no silently missing tracks or collapsed timestamps
 - [RTMP sessions follow one strict path](/quest/m1/cut-audit/rtmp-session-strict.md) - idle sockets, repeated requests, rewinds, and stalled servers are handled
 - [The export jitter buffer admits only frames due within its window](/quest/m1/cut-audit/jitter-admission.md) - a fast track is backpressured, not buffered without bound
 - [Peer-driven receive loops yield on a per-message budget](/quest/m1/cut-audit/peer-loop-budget.md) - small-message floods cannot monopolize a session driver

rs/moq-mux/Cargo.toml

@@ -17,6 +17,7 @@ anyhow = { workspace = true }
 base64 = { workspace = true }
 bytes = { workspace = true }
 crc = "3"
+ebml-iterable = { workspace = true }
 h264-parser = { version = "0.4.2" }
 h264-reader = "0.9.0"
 hang = { workspace = true }

rs/moq-mux/MIGRATE.md

@@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### MKV and FLV import refuse malformed tracks and unsupported codecs (#6908)
+
+MKV import fails on a malformed track, a duplicate or zero `TrackNumber`, a zero `TimestampScale`, a block timestamp outside the `u64` timeline, an unsupported codec or subtitle encoding, or input truncated at `finish()`, instead of skipping the track or wrapping the timestamp. Input after `finish()` fails with `mkv::Error::TracksAfterFinish`.
+FLV import fails with `Error::UnknownFormat` on an unsupported legacy or enhanced codec instead of dropping its tags.
+Supply valid metadata and supported codecs, or transcode or remove unsupported media tracks before import. Tracks of an unsupported role (such as MKV buttons) are still skipped.
+
 ### Rate following can disable a simulcast rung
 
 `rate::Follower::next` and `poll_next` return a `rate::Target`, and `rate::Control::update` returns `Option<rate::Target>`: `Target::Enabled(rate)` is the bitrate to encode at, and `Target::Disabled` means stop encoding and advertise the rendition disabled.

rs/moq-mux/benches/containers.rs

@@ -1,4 +1,4 @@
-//! Complete fMP4/CMAF and Annex-B imports and exports, with setup outside timing.
+//! Complete container and Annex-B imports and exports, with setup outside timing.
 use std::time::Duration;
 
 use bytes::{Bytes, BytesMut};
@@ -645,9 +645,43 @@ fn ts_si_table_history(c: &mut Criterion) {
 	);
 }
 
+/// Live WebM framing across transport chunk sizes, including byte-at-a-time metadata.
+fn mkv_chunks(c: &mut Criterion) {
+	let runtime = tokio::runtime::Builder::new_current_thread()
+		.enable_all()
+		.build()
+		.unwrap();
+	let _enter = runtime.enter();
+	let data = include_bytes!("../src/container/mkv/test_data/opus.webm");
+	let mut group = c.benchmark_group("mkv-chunks");
+	group.throughput(Throughput::Bytes(data.len() as u64));
+	for chunk in [1, 7, 64, data.len()] {
+		group.bench_function(BenchmarkId::from_parameter(chunk), |b| {
+			b.iter_batched(
+				|| {
+					let mut broadcast = moq_net::broadcast::Info::new().produce();
+					let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap();
+					let import = moq_mux::container::mkv::Import::new(broadcast, catalog.reserve());
+					(catalog, import)
+				},
+				|(catalog, mut import)| {
+					for bytes in std::hint::black_box(data).chunks(chunk) {
+						import.decode(bytes).unwrap();
+					}
+					import.finish().unwrap();
+					(catalog, import)
+				},
+				BatchSize::PerIteration,
+			);
+		});
+	}
+	group.finish();
+}
+
 criterion_group!(
 	benches,
 	containers,
+	mkv_chunks,
 	interleaved_runs,
 	queued_groups,
 	ts_si_renewal,

rs/moq-mux/src/codec/vp9/import.rs

@@ -53,11 +53,11 @@ impl Import {
 		Ok(())
 	}
 
-	fn init(&mut self, vp9: hang::catalog::VP9, width: u16, height: u16) -> crate::Result<()> {
+	fn init(&mut self, vp9: hang::catalog::VP9, width: u32, height: u32) -> crate::Result<()> {
 		let mut config = hang::catalog::VideoConfig::new(vp9.clone());
 		config.color = super::color(&vp9);
-		config.coded_width = Some(width as u32);
-		config.coded_height = Some(height as u32);
+		config.coded_width = Some(width);
+		config.coded_height = Some(height);
 
 		self.apply_config(config)?;
 		Ok(())

rs/moq-mux/src/codec/vp9/mod.rs

@@ -53,8 +53,8 @@ pub(crate) struct FrameHeader {
 /// The parts of a VP9 key-frame header we surface to the catalog.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) struct KeyFrame {
-	pub width: u16,
-	pub height: u16,
+	pub width: u32,
+	pub height: u32,
 	pub profile: u8,
 	pub bit_depth: u8,
 	/// `vpcC` chroma subsampling enum (0 = 4:2:0 vertical, 1 = 4:2:0 colocated,
@@ -137,8 +137,8 @@ impl FrameHeader {
 		}
 
 		// frame_size (VP9 spec §6.2.3): each field is `value - 1`.
-		let width = (r.read(16)? + 1) as u16;
-		let height = (r.read(16)? + 1) as u16;
+		let width = r.read(16)? + 1;
+		let height = r.read(16)? + 1;
 
 		Ok(Self {
 			keyframe: true,
@@ -202,8 +202,8 @@ pub(crate) fn config_from_keyframe(data: &[u8]) -> Result<Option<hang::catalog::
 	let vp9 = key.to_catalog();
 	let mut config = hang::catalog::VideoConfig::new(vp9.clone());
 	config.color = color(&vp9);
-	config.coded_width = Some(width as u32);
-	config.coded_height = Some(height as u32);
+	config.coded_width = Some(width);
+	config.coded_height = Some(height);
 	Ok(Some(config))
 }
 
@@ -236,7 +236,7 @@ fn matrix_coefficients(color_space: u8) -> u8 {
 /// The lowest VP9 level whose `MaxLumaPictureSize` fits `width * height` (VP9
 /// spec Annex A). Framerate and bitrate aren't known from the header, so this
 /// is a picture-size lower bound rather than the exact encoded level.
-fn level_for(width: u16, height: u16) -> u8 {
+fn level_for(width: u32, height: u32) -> u8 {
 	let area = width as u64 * height as u64;
 	// (level, MaxLumaPictureSize), ascending.
 	const LEVELS: &[(u8, u64)] = &[
@@ -326,6 +326,28 @@ mod tests {
 		assert!(!key.full_range);
 	}
 
+	#[test]
+	fn dimensions_keep_the_complete_sixteen_bit_plus_one_range() {
+		for (width, height) in [
+			(1u32, 65_536u32),
+			(65_536, 1),
+			(65_535, 65_536),
+			(65_536, 65_535),
+			(65_536, 65_536),
+		] {
+			let mut frame = KEYFRAME_320X240.to_vec();
+			for (start, value) in [(36, width - 1), (52, height - 1)] {
+				for bit in 0..16 {
+					let position = start + bit;
+					let mask = 1 << (7 - position % 8);
+					frame[position / 8] = (frame[position / 8] & !mask) | (((value >> (15 - bit)) & 1) as u8 * mask);
+				}
+			}
+			let config = config_from_keyframe(&frame).unwrap().unwrap();
+			assert_eq!((config.coded_width, config.coded_height), (Some(width), Some(height)));
+		}
+	}
+
 	#[test]
 	fn keyframe_to_catalog() {
 		let key = FrameHeader::parse(KEYFRAME_320X240).unwrap().key.unwrap();

rs/moq-mux/src/container/flv/import.rs

@@ -18,7 +18,7 @@
 //! codecs (MP3 / AC-3 / E-AC-3) carry their config in band, so they configure from
 //! the first frame instead. Sample bytes already match the codec payload, so no codec
 //! transform is needed before the selected media container wraps them. FLAC (`fLaC`)
-//! enhanced audio, and any other codec, are logged and dropped.
+//! enhanced audio, and any other unsupported codec, fail the import.
 
 use std::collections::BTreeMap;
 
@@ -51,7 +51,7 @@ const MAX_DATA_OFFSET: usize = 64 * 1024;
 ///
 /// Supports legacy H.264 + AAC and the enhanced-RTMP FourCC codecs (HEVC, AV1,
 /// VP9, Opus, AC-3, E-AC-3), the payloads produced by RTMP encoders and
-/// `ffmpeg -f flv`. Unsupported codecs, plus `onMetaData` script tags, are logged
+/// `ffmpeg -f flv`. Unsupported codecs fail the import; `onMetaData` script tags are logged
 /// and dropped.
 ///
 /// Legacy and single-track enhanced tags carry one video and one audio track;
@@ -212,8 +212,7 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 		let frame_type = first >> 4;
 		let codec_id = first & 0x0f;
 		if codec_id != VIDEO_CODEC_AVC {
-			tracing::warn!(codec_id, "unsupported FLV video codec, dropping");
-			return Ok(());
+			return Err(crate::Error::UnknownFormat(format!("FLV video CodecID {codec_id}")).into());
 		}
 
 		anyhow::ensure!(body.len() >= 5, "AVC video tag too short");
@@ -289,6 +288,9 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 		payload: &[u8],
 		timestamp: u64,
 	) -> anyhow::Result<()> {
+		if !matches!(fourcc, b"avc1" | b"hvc1" | b"av01" | b"vp09") {
+			return Err(crate::Error::UnknownFormat(format!("FLV video FourCC {fourcc:?}")).into());
+		}
 		match packet_type {
 			VIDEO_PACKET_SEQUENCE_START => {
 				let config = match fourcc {
@@ -299,8 +301,7 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 					// redundant with the key-frame header we configure from.
 					b"vp09" => return Ok(()),
 					other => {
-						tracing::warn!(fourcc = ?other, "unsupported enhanced FLV video codec, dropping");
-						return Ok(());
+						return Err(crate::Error::UnknownFormat(format!("FLV video FourCC {other:?}")).into());
 					}
 				};
 				self.init_video(track_id, config)
@@ -356,8 +357,7 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 			return self.write_audio(DEFAULT_TRACK_ID, frame, timestamp);
 		}
 		if sound_format != AUDIO_FORMAT_AAC {
-			tracing::warn!(sound_format, "unsupported FLV audio format, dropping");
-			return Ok(());
+			return Err(crate::Error::UnknownFormat(format!("FLV audio SoundFormat {sound_format}")).into());
 		}
 
 		anyhow::ensure!(body.len() >= 2, "AAC audio tag too short");
@@ -420,6 +420,9 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 		payload: &[u8],
 		timestamp: u64,
 	) -> anyhow::Result<()> {
+		if !matches!(fourcc, b"Opus" | b"mp4a" | b".mp3" | b"ac-3" | b"ec-3") {
+			return Err(crate::Error::UnknownFormat(format!("FLV audio FourCC {fourcc:?}")).into());
+		}
 		match packet_type {
 			AUDIO_PACKET_SEQUENCE_START => {
 				let config = match fourcc {
@@ -428,8 +431,7 @@ impl<E: crate::catalog::hang::CatalogExt> Import<E> {
 					// MP3 / AC-3 / E-AC-3 are verbatim with no sequence header; they
 					// configure from the first frame. Anything else is unsupported.
 					other => {
-						tracing::warn!(fourcc = ?other, "unsupported enhanced FLV audio codec, dropping");
-						return Ok(());
+						return Err(crate::Error::UnknownFormat(format!("FLV audio FourCC {other:?}")).into());
 					}
 				};
 				self.init_audio(track_id, config)

rs/moq-mux/src/container/flv/import_test.rs

@@ -875,3 +875,49 @@ fn script_metadata_drops_warn_once() {
 		}
 	});
 }
+
+#[test]
+fn unsupported_codecs_fail_at_the_public_importer_boundary() {
+	let mut cases = vec![(super::TAG_VIDEO, vec![0x12, 0]), (super::TAG_AUDIO, vec![0xb0, 0])];
+	for (tag, packets) in [
+		(super::TAG_VIDEO, &[0, 1, 2, 3, 4][..]),
+		(super::TAG_AUDIO, &[0, 1, 2, 4][..]),
+	] {
+		for packet in packets {
+			cases.push((tag, vec![0x90 | packet, b'n', b'o', b'p', b'e']));
+			// OneTrack enhanced framing carries the same codec after its framing byte.
+			cases.push((tag, vec![0x95, *packet, b'n', b'o', b'p', b'e', 0]));
+		}
+	}
+
+	for (tag, payload) in cases {
+		let mut data = synth_flv();
+		write_tag(&mut data, tag, 0, &payload);
+		for initial in [false, true] {
+			let mut broadcast = moq_net::broadcast::Info::new().produce();
+			let catalog = crate::catalog::Producer::new(&mut broadcast, Default::default()).unwrap();
+			let init = crate::import::ContainerInit::new(
+				crate::import::ContainerFormat::Flv,
+				if initial {
+					bytes::Bytes::copy_from_slice(&data)
+				} else {
+					bytes::Bytes::new()
+				},
+			);
+			let result = (|| -> crate::Result<()> {
+				let mut import: crate::import::Container =
+					crate::import::Container::new(broadcast, catalog.reserve(), &init)?;
+				if !initial {
+					for chunk in data.chunks(7) {
+						import.decode(chunk)?;
+					}
+				}
+				import.finish()
+			})();
+			assert!(
+				matches!(result, Err(crate::Error::UnknownFormat(_))),
+				"tag {tag}, initial {initial}: {result:?}"
+			);
+		}
+	}
+}

rs/moq-mux/src/container/flv/mod.rs

@@ -18,7 +18,7 @@
 //! from the frame), and the sample bytes already match the
 //! [`Legacy`](crate::catalog::hang::Container) container, so no codec transform is
 //! needed. Other legacy codecs (VP6, Speex, …) and the E-RTMP FLAC audio are
-//! logged and dropped on import, and rejected on export.
+//! rejected on import and export.
 
 mod export;
 mod import;