diff --git a/Cargo.lock b/Cargo.lock index 6a26796f90..aeb39fca67 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5443,6 +5443,7 @@ dependencies = [ "bytes", "crc", "criterion", + "ebml-iterable", "futures", "h264-parser", "h264-reader 0.9.0", diff --git a/Cargo.toml b/Cargo.toml index 99b1a01da5..6a7db7b687 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/quest/m1/cut-audit/README.md b/quest/m1/cut-audit/README.md index 641ce105f6..d705a245e2 100644 --- a/quest/m1/cut-audit/README.md +++ b/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 diff --git a/quest/m1/cut-audit/mkv-import-errors.md b/quest/m1/cut-audit/mkv-import-errors.md deleted file mode 100644 index e78f21e638..0000000000 --- a/quest/m1/cut-audit/mkv-import-errors.md +++ /dev/null @@ -1,37 +0,0 @@ -# [S] A malformed MKV track fails the import instead of vanishing - -## Goal - -The MKV importer fails loud on a malformed track, a duplicate track -number, or an invalid timestamp, instead of publishing a catalog that -silently lacks the track or carries collapsed timestamps. - -## Plan - -Found by the 2026-10-24 pre-cut audit at 151a529670, source-inferred. - -- `handle_tracks` (`rs/moq-mux/src/container/mkv/import.rs:338`) logs and - skips every `add_track` error except an allowlist (`InvalidSubtitle`, - `InvalidAspect`, `ValueOutOfRange`). `InvalidColour` - ([moq-dev/moq#5846](https://github.com/moq-dev/moq/pull/5846)), a malformed - avcC/hvcC, or a missing AAC CodecPrivate drops the video track and the - catalog goes out audio-only. Every validation fix has had to extend the - allowlist by hand ([moq-dev/moq#6532](https://github.com/moq-dev/moq/pull/6532)). - Return `Ok` only for an unsupported role, decided where the error is - raised, and propagate everything else, an unsupported codec included; - delete the allowlist. -- A duplicate `TrackNumber` (`:452`) replaces the earlier track. Refuse it. -- `cluster_timestamp as i64 + block.timestamp` (`:533`) can overflow (a - wrap in release builds, misreported as `NegativeBlockTimestamp`), and - `TimestampScale = 0` (`:289`) collapses every timestamp to 0. Use - `checked_add_signed` and refuse a zero scale. -- While in moq-mux codecs: `codec/vp9/mod.rs:140` computes - `(r.read(16)? + 1) as u16`, so a 65536 dimension becomes 0 in the catalog. - Use `u32`, like the catalog field. - -Decided 2026-10-25 (maintainer): reject an unsupported codec everywhere, as -`rs/moq-mux/AGENTS.md` says. MKV and FLV (`flv/import.rs` about lines 215, -302, 359, 431) stop dropping such tracks and fail the import with a typed -Error. - -Public API: none. Wire: none. diff --git a/rs/moq-mux/Cargo.toml b/rs/moq-mux/Cargo.toml index 5d74669dbb..ff9da615d9 100644 --- a/rs/moq-mux/Cargo.toml +++ b/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 } diff --git a/rs/moq-mux/MIGRATE.md b/rs/moq-mux/MIGRATE.md index a36c142fdb..0b4e5326dc 100644 --- a/rs/moq-mux/MIGRATE.md +++ b/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`: `Target::Enabled(rate)` is the bitrate to encode at, and `Target::Disabled` means stop encoding and advertise the rendition disabled. diff --git a/rs/moq-mux/benches/containers.rs b/rs/moq-mux/benches/containers.rs index 33e240bf4a..b844fd55ba 100644 --- a/rs/moq-mux/benches/containers.rs +++ b/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, diff --git a/rs/moq-mux/src/codec/vp9/import.rs b/rs/moq-mux/src/codec/vp9/import.rs index f65b476430..e01ec13f99 100644 --- a/rs/moq-mux/src/codec/vp9/import.rs +++ b/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(()) diff --git a/rs/moq-mux/src/codec/vp9/mod.rs b/rs/moq-mux/src/codec/vp9/mod.rs index bb7c92b0bb..76e899a4e0 100644 --- a/rs/moq-mux/src/codec/vp9/mod.rs +++ b/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 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(); diff --git a/rs/moq-mux/src/container/flv/import.rs b/rs/moq-mux/src/container/flv/import.rs index 08c07376cc..4d17afcedf 100644 --- a/rs/moq-mux/src/container/flv/import.rs +++ b/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 Import { 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 Import { 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 Import { // 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 Import { 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 Import { 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 Import { // 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) diff --git a/rs/moq-mux/src/container/flv/import_test.rs b/rs/moq-mux/src/container/flv/import_test.rs index 1ab72a52c0..b6765917ed 100644 --- a/rs/moq-mux/src/container/flv/import_test.rs +++ b/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:?}" + ); + } + } +} diff --git a/rs/moq-mux/src/container/flv/mod.rs b/rs/moq-mux/src/container/flv/mod.rs index c6c3f9463c..1c7d117177 100644 --- a/rs/moq-mux/src/container/flv/mod.rs +++ b/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; diff --git a/rs/moq-mux/src/container/mkv/import.rs b/rs/moq-mux/src/container/mkv/import.rs index 2dcede5dc1..35c882ab3e 100644 --- a/rs/moq-mux/src/container/mkv/import.rs +++ b/rs/moq-mux/src/container/mkv/import.rs @@ -1,15 +1,12 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::convert::TryFrom; use std::io::Cursor; use crate::Result; -use bytes::{Buf, Bytes, BytesMut}; +use bytes::{Bytes, BytesMut}; use hang::catalog::{AAC, AudioCodec, AudioConfig, H265, VP9, VideoCodec, VideoConfig}; use moq_net::Timestamp; use mp4_atom::Atom; -use webm_iterable::WebmIterator; -use webm_iterable::errors::TagIteratorError; -use webm_iterable::iterator::AllowableErrors; use webm_iterable::matroska_spec::{BlockLacing, Master, MatroskaSpec, SimpleBlock}; use super::Error; @@ -41,7 +38,7 @@ const DEFAULT_TIMESTAMP_SCALE_NS: u64 = 1_000_000; /// /// Laced blocks require a positive TrackEntry DefaultDuration to timestamp each frame. /// -/// Unsupported track roles and subtitle encodings are logged and dropped. +/// Unsupported track roles are skipped; malformed tracks and unsupported codecs fail the import. pub struct Import { broadcast: moq_net::broadcast::Producer, catalog: crate::catalog::Producer, @@ -58,11 +55,10 @@ pub struct Import { /// Accumulated unparsed input. buffer: BytesMut, - /// Bytes already dropped from the front of `buffer`, so a tag's offset in it is absolute. - consumed: u64, - /// Where the latest handled tag starts, as an absolute offset. A drain pass restarts from a - /// replay point, so any tag starting at or before it was already handled. - handled: Option, + /// Incremental element state; an incomplete master retains its own children across chunks. + decoder: ebml_iterable::TagDecoder, + /// Whether `finish()` ended the input, even if it failed before the decoder drained. + finished: bool, /// Whether the Tracks element has been processed. tracks_seen: bool, @@ -172,8 +168,16 @@ impl Import { timebase: reserved.timebase(), initial_reservation: Some(reserved), buffer: BytesMut::new(), - consumed: 0, - handled: None, + decoder: ebml_iterable::TagDecoder::new(&[ + MatroskaSpec::Ebml(Master::Start), + MatroskaSpec::Info(Master::Start), + MatroskaSpec::Tracks(Master::Start), + MatroskaSpec::TrackEntry(Master::Start), + MatroskaSpec::Audio(Master::Start), + MatroskaSpec::Video(Master::Start), + MatroskaSpec::BlockGroup(Master::Start), + ]), + finished: false, tracks_seen: false, timestamp_scale_ns: DEFAULT_TIMESTAMP_SCALE_NS, cluster_timestamp: 0, @@ -196,81 +200,23 @@ impl Import { /// scratch). Bytes that cannot yet form a complete top-level tag are retained /// for the next call. pub fn decode(&mut self, data: &[u8]) -> Result<()> { - // Move the input into our scratch buffer. + if self.finished && !data.is_empty() { + return Err(Error::TracksAfterFinish.into()); + } self.buffer.extend_from_slice(data); self.drain() } - /// Run the iterator over the buffered bytes, processing every fully-parsed top-level tag. - /// - /// On each call, the iterator restarts from the beginning of the retained buffer, so tags - /// already handled are skipped by their absolute offset. After parsing stops (UnexpectedEOF or end of buffer), bytes up to the start - /// of the most-recently emitted top-level tag are discarded so memory does not grow - /// unboundedly. + /// Decode complete elements, leaving incomplete bytes and master state for the next chunk. fn drain(&mut self) -> Result<()> { - // Buffer master tags that are bounded and convenient to handle atomically. - let buffered = [ - MatroskaSpec::Ebml(Master::Start), - MatroskaSpec::Info(Master::Start), - MatroskaSpec::Tracks(Master::Start), - MatroskaSpec::TrackEntry(Master::Start), - MatroskaSpec::Audio(Master::Start), - MatroskaSpec::Video(Master::Start), - MatroskaSpec::BlockGroup(Master::Start), - ]; - - if self.buffer.is_empty() { - return Ok(()); - } - - let snapshot = self.buffer.clone().freeze(); - let mut cursor = Cursor::new(snapshot.as_ref()); - let mut iter = WebmIterator::new(&mut cursor, &buffered); - // We restart the iterator from the beginning of the retained buffer on every - // drain pass. Once data is replayed mid-Segment, ebml-iterable would otherwise - // reject Segment children (Cluster, Tracks, etc.) as appearing without their - // parent. Allowing hierarchy problems plus skipping tags by offset gives us - // idempotent streaming behavior. - iter.allow_errors(&[AllowableErrors::HierarchyProblems]); - // Don't synthesize Master::End tags when the buffer ends mid-element. - iter.emit_master_end_when_eof(false); - - let mut last_offset: usize = 0; - - loop { - match iter.next() { - Some(Ok(tag)) => { - last_offset = iter.last_emitted_tag_offset(); - // A Master::End reports its master's start, so it is skipped here too. - let start = self.consumed + last_offset as u64; - if self.handled.is_some_and(|handled| start <= handled) { - continue; - } - self.handled = Some(start); - self.handle_tag(tag)?; - } - Some(Err(TagIteratorError::UnexpectedEOF { .. })) => break, - Some(Err(_e)) => { - return Err(Error::MatroskaParse.into()); - } - None => { - last_offset = snapshot.len(); - break; - } - } + while let Some(tag) = self + .decoder + .decode(&mut self.buffer) + .map_err(|_| Error::MatroskaParse)? + { + self.handle_tag(tag.tag)?; } - - drop(iter); - - // Retain bytes from the start of the last emitted tag (safe replay point) onward. - // At the very least, this lets us reuse partially-read tags as more data arrives. - // If we never emitted anything (very first call with too few bytes), keep everything. - if last_offset > 0 { - self.buffer.advance(last_offset); - self.consumed += last_offset as u64; - } - Ok(()) } @@ -286,6 +232,9 @@ impl Import { MatroskaSpec::Info(Master::Full(children)) => { for c in &children { if let MatroskaSpec::TimestampScale(v) = c { + if *v == 0 { + return Err(Error::ValueOutOfRange("TimestampScale").into()); + } self.timestamp_scale_ns = *v; } } @@ -336,23 +285,21 @@ impl Import { } fn handle_tracks(&mut self, reserved: &crate::catalog::Reserved, entries: Vec) -> Result<()> { + let mut numbers = HashSet::new(); for entry in entries { - if let MatroskaSpec::TrackEntry(Master::Full(children)) = entry - && let Err(e) = self.add_track(reserved, children) - { - if matches!( - e, - crate::Error::Mkv(Error::InvalidSubtitle(_) | Error::InvalidAspect(_) | Error::ValueOutOfRange(_)) - ) { - return Err(e); - } - tracing::warn!(error = ?e, "skipping MKV track"); + if let MatroskaSpec::TrackEntry(Master::Full(children)) = entry { + self.add_track(reserved, children, &mut numbers)?; } } Ok(()) } - fn add_track(&mut self, reserved: &crate::catalog::Reserved, children: Vec) -> Result<()> { + fn add_track( + &mut self, + reserved: &crate::catalog::Reserved, + children: Vec, + numbers: &mut HashSet, + ) -> Result<()> { let mut track_number: Option = None; let mut track_type: Option = None; let mut codec_id: Option = None; @@ -369,7 +316,11 @@ impl Import { MatroskaSpec::DefaultDuration(v) if v > 0 => { default_duration = Some(std::time::Duration::from_nanos(v)) } - MatroskaSpec::TrackNumber(v) => track_number = Some(v), + MatroskaSpec::TrackNumber(v) => { + if track_number.replace(v).is_some() { + return Err(moq_net::Error::Duplicate.into()); + } + } MatroskaSpec::TrackType(v) => track_type = Some(v), MatroskaSpec::CodecID(v) => codec_id = Some(v), MatroskaSpec::Language(v) => language = Some(v), @@ -383,6 +334,12 @@ impl Import { } let track_number = track_number.ok_or(Error::MissingTrackNumber)?; + if track_number == 0 { + return Err(Error::ValueOutOfRange("TrackNumber").into()); + } + if !numbers.insert(track_number) { + return Err(moq_net::Error::Duplicate.into()); + } let track_type = track_type.ok_or(Error::MissingTrackType)?; let codec_id = codec_id.ok_or(Error::MissingCodecId)?; @@ -391,10 +348,7 @@ impl Import { 1 => (TrackKind::Video, ".mkv-v"), 2 => (TrackKind::Audio, ".mkv-a"), 17 if matches!(codec_id.as_str(), "S_TEXT/UTF8" | "S_TEXT/WEBVTT") => (TrackKind::Text, ".mkv-t"), - 17 => { - tracing::warn!(codec_id, "unsupported MKV subtitle encoding, skipping"); - return Ok(()); - } + 17 => return Err(crate::Error::UnknownFormat(format!("MKV subtitle codec {codec_id}"))), other => { tracing::warn!(track_type = other, codec_id, "unsupported MKV track type, skipping"); return Ok(()); @@ -530,12 +484,16 @@ impl Import { // Compute PTS in MKV's native nanosecond units and stamp it on the // timestamp at NANO scale so a passthrough re-emit preserves precision. - let block_ticks = (self.cluster_timestamp as i64) + (block.timestamp as i64); - if block_ticks < 0 { - return Err(Error::NegativeBlockTimestamp.into()); - } + let block_ticks = self + .cluster_timestamp + .checked_add_signed(i64::from(block.timestamp)) + .ok_or(if block.timestamp < 0 { + Error::NegativeBlockTimestamp + } else { + Error::TimestampOverflow + })?; - let pts_ns = (block_ticks as u64) + let pts_ns = block_ticks .checked_mul(self.timestamp_scale_ns) .ok_or(Error::TimestampOverflow)?; if track.kind == TrackKind::Text { @@ -635,7 +593,15 @@ impl Import { /// Finish all tracks, flushing current groups. pub fn finish(&mut self) -> Result<()> { - // No input follows, so the buffer that waited for the rest of an element is done. + self.finished = true; + while let Some(tag) = self + .decoder + .decode_eof(&mut self.buffer) + .map_err(|_| Error::MatroskaParse)? + { + self.handle_tag(tag.tag)?; + } + // Finished sources can remain owned after EOF; release the last element's allocation. self.buffer = BytesMut::new(); // No frame follows to anchor the clock, so publish the declared track set now. self.initial_reservation = None; @@ -874,7 +840,7 @@ fn build_audio_config( codec_private: Option<&Bytes>, audio_children: Option<&[MatroskaSpec]>, ) -> Result { - // Identify the codec first: unsupported tracks retain their skip behavior even with bad fields. + // Reject an unsupported codec before interpreting its codec-specific fields. let mut config = match codec_id { "A_OPUS" => { // The catalog describes the decoder's output: Opus always decodes at 48 kHz, whatever the @@ -1116,3 +1082,105 @@ fn split_lacing<'a>(block: &'a SimpleBlock<'_>) -> std::result::Result Imported { } } -/// Every drain pass restarts from a replay point, so a small chunk replays tags; each block -/// still publishes exactly once. +/// Incomplete elements span chunks, and each completed block publishes exactly once. #[tokio::test] async fn chunked_decode_publishes_each_block_once() { let data = MkvBuilder::new() @@ -1205,7 +1210,6 @@ async fn subtitle_languages_and_empty_tracks_are_published() { text_entry(1, "S_TEXT/UTF8", "en", "English"), text_entry(2, "S_TEXT/UTF8", "es", "Spanish"), text_entry(3, "S_TEXT/UTF8", "fr", "French"), - text_entry(4, "S_TEXT/ASS", "de", "Unsupported"), ]) .cluster(0, || { let MatroskaSpec::SimpleBlock(block) = simple_block(1, 250, true, b"Hello") else { @@ -1734,3 +1738,317 @@ fn unrepresentable_colour_is_refused() { assert!(super::import::matroska_color(&[child]).is_err()); } } + +/// Container entry points must preserve importer failures, both in the constructor's +/// initial data and in arbitrarily split subsequent chunks. +fn import_result(data: &[u8], initial: bool) -> crate::Result<()> { + 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::Mkv, + if initial { + bytes::Bytes::copy_from_slice(data) + } else { + bytes::Bytes::new() + }, + ); + 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() +} + +#[test] +fn malformed_track_metadata_fails_through_the_container() { + use MatroskaSpec::*; + let cases = [ + TrackEntry(Master::Full(vec![TrackType(2), CodecID("A_OPUS".into())])), + TrackEntry(Master::Full(vec![ + TrackNumber(2), + TrackType(2), + CodecID("A_AAC".into()), + ])), + TrackEntry(Master::Full(vec![ + TrackNumber(2), + TrackType(1), + CodecID("V_MPEG4/ISO/AVC".into()), + CodecPrivate(vec![0]), + ])), + TrackEntry(Master::Full(vec![ + TrackNumber(2), + TrackType(1), + CodecID("V_VP9".into()), + Video(Master::Full(vec![ + PixelWidth(320), + PixelHeight(240), + Colour(Master::Full(vec![MaxCLL(70_000)])), + ])), + ])), + ]; + for (index, invalid) in cases.into_iter().enumerate() { + for reverse in [false, true] { + let mut tracks = vec![track_entry_audio_opus(1, 48_000.0, 2), invalid.clone()]; + if reverse { + tracks.reverse(); + } + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(tracks) + .segment_end() + .build(); + for initial in [false, true] { + assert!( + import_result(&data, initial).is_err(), + "case {index}, reverse {reverse}, initial {initial}" + ); + } + } + } +} + +#[test] +fn duplicate_track_numbers_fail_even_for_skipped_roles() { + use MatroskaSpec::*; + for second in [ + track_entry_audio_opus(1, 44_100.0, 1), + TrackEntry(Master::Full(vec![ + TrackNumber(1), + TrackType(18), + CodecID("B_BUTTONS".into()), + ])), + ] { + for reverse in [false, true] { + let mut tracks = vec![track_entry_audio_opus(1, 48_000.0, 2), second.clone()]; + if reverse { + tracks.reverse(); + } + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(tracks) + .segment_end() + .build(); + for initial in [false, true] { + let result = import_result(&data, initial); + assert!( + matches!(&result, Err(crate::Error::Moq(moq_net::Error::Duplicate))), + "reverse {reverse}, initial {initial}: {result:?}" + ); + } + } + } +} + +#[test] +fn invalid_timestamp_scales_are_refused() { + for initial in [false, true] { + let zero = MkvBuilder::new() + .header("webm") + .segment_start() + .info(0) + .tracks(vec![track_entry_audio_opus(1, 48_000.0, 2)]) + .cluster(0, || vec![simple_block(1, 0, true, b"opus")]) + .segment_end() + .build(); + assert!(matches!( + import_result(&zero, initial), + Err(crate::Error::Mkv(super::Error::ValueOutOfRange("TimestampScale"))) + )); + } +} + +#[test] +fn block_tick_overflow_is_refused_before_timestamp_conversion() { + for (cluster, relative, scale) in [(u64::MAX, 1, 1), (u64::MAX, 0, 2)] { + for initial in [false, true] { + let data = MkvBuilder::new() + .header("webm") + .segment_start() + .info(scale) + .tracks(vec![track_entry_audio_opus(1, 48_000.0, 2)]) + .cluster(cluster, || vec![simple_block(1, relative, true, b"opus")]) + .segment_end() + .build(); + let outcome = std::panic::catch_unwind(|| import_result(&data, initial)); + assert!(matches!( + outcome, + Ok(Err(crate::Error::Mkv(super::Error::TimestampOverflow))) + )); + } + } + let negative = MkvBuilder::new() + .header("webm") + .segment_start() + .info(1) + .tracks(vec![track_entry_audio_opus(1, 48_000.0, 2)]) + .cluster(0, || vec![simple_block(1, -1, true, b"opus")]) + .segment_end() + .build(); + for initial in [false, true] { + assert!(matches!( + import_result(&negative, initial), + Err(crate::Error::Mkv(super::Error::NegativeBlockTimestamp)) + )); + } +} + +#[test] +fn unsupported_subtitle_codec_is_refused() { + for reverse in [false, true] { + let mut tracks = vec![ + track_entry_audio_opus(1, 48_000.0, 2), + text_entry(2, "S_TEXT/ASS", "de", "Unsupported"), + ]; + if reverse { + tracks.reverse(); + } + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(tracks) + .segment_end() + .build(); + for initial in [false, true] { + assert!( + matches!(import_result(&data, initial), Err(crate::Error::UnknownFormat(codec)) if codec.contains("S_TEXT/ASS")) + ); + } + } +} + +#[tokio::test(start_paused = true)] +async fn tiny_chunks_preserve_tracks_and_skip_only_unsupported_roles() { + use MatroskaSpec::*; + let data = MkvBuilder::new() + .header("matroska") + .segment_start() + .info(1_000_000) + .tracks(vec![ + track_entry_audio_opus(1, 48_000.0, 2), + track_entry_audio_opus(2, 48_000.0, 1), + TrackEntry(Master::Full(vec![ + TrackNumber(3), + TrackType(18), + CodecID("B_BUTTONS".into()), + Video(Master::Full(vec![PixelWidth(u64::MAX)])), + ])), + ]) + .cluster(0, || { + vec![ + simple_block(1, 0, true, b"stereo"), + simple_block(2, 20, true, b"mono"), + simple_block(3, 0, true, b"ignored"), + ] + }) + .segment_end() + .build(); + for chunk in [1, 2, 7, 16, data.len()] { + let imported = import_chunked(&data, chunk).await; + assert!(imported.refused.is_none(), "chunk {chunk}: {:?}", imported.refused); + assert_eq!(imported.published.len(), 2, "chunk {chunk}"); + let mut timestamps: Vec<_> = imported.published.into_values().collect(); + timestamps.sort(); + assert_eq!(timestamps, [vec![0], vec![20_000]], "chunk {chunk}"); + } +} + +#[test] +fn partial_track_metadata_is_refused_at_eof() { + let data = MkvBuilder::new() + .header("webm") + .segment_start() + .info(1_000_000) + .tracks(vec![track_entry_audio_opus(1, 48_000.0, 2)]) + .segment_end() + .build(); + for missing in 1..=8 { + for initial in [false, true] { + assert!( + matches!( + import_result(&data[..data.len() - missing], initial), + Err(crate::Error::Mkv(super::Error::MatroskaParse)) + ), + "missing {missing}, initial {initial}" + ); + } + } + for initial in [false, true] { + import_result(&data, initial).unwrap(); + } +} + +#[test] +fn block_ticks_above_the_signed_range_are_not_misreported_as_negative() { + for (cluster, relative) in [(i64::MAX as u64, 1i16), (u64::MAX, -1)] { + let data = MkvBuilder::new() + .header("webm") + .segment_start() + .info(1) + .tracks(vec![track_entry_audio_opus(1, 48_000.0, 2)]) + .cluster(cluster, || vec![simple_block(1, relative, true, b"frame")]) + .segment_end() + .build(); + // The addition fits u64, but exceeds MoQ's existing timestamp value range. + for initial in [false, true] { + assert!( + matches!(import_result(&data, initial), Err(crate::Error::TimestampOverflow(_))), + "{cluster}+{relative}, initial {initial}" + ); + } + } +} + +#[test] +fn invalid_track_number_fields_are_refused() { + use MatroskaSpec::*; + for (fields, duplicate) in [ + (vec![TrackNumber(0)], false), + (vec![TrackNumber(1), TrackNumber(2)], true), + ] { + let mut entry = fields; + entry.extend([TrackType(2), CodecID("A_OPUS".into())]); + let data = MkvBuilder::new() + .header("webm") + .segment_start() + .info(1_000_000) + .tracks(vec![TrackEntry(Master::Full(entry))]) + .segment_end() + .build(); + for initial in [false, true] { + let error = import_result(&data, initial).unwrap_err(); + if duplicate { + assert!(matches!(error, crate::Error::Moq(moq_net::Error::Duplicate))); + } else { + assert!(matches!( + error, + crate::Error::Mkv(super::Error::ValueOutOfRange("TrackNumber")) + )); + } + } + } +} + +/// ffmpeg's live WebM uses an unknown-size Segment and real Opus packet metadata. +/// The encoded fixture needs no ffmpeg installation when this gate runs. +#[tokio::test(start_paused = true)] +async fn real_live_webm_imports_across_chunk_boundaries() { + let data = include_bytes!("test_data/opus.webm"); + for initial in [false, true] { + import_result(data, initial).unwrap(); + } + for chunk in [1, 7, 64, data.len()] { + let imported = import_chunked(data, chunk).await; + assert!(imported.refused.is_none(), "chunk {chunk}: {:?}", imported.refused); + assert_eq!(imported.published.len(), 1); + let timestamps = imported.published.into_values().next().unwrap(); + assert_eq!(timestamps.len(), 6); + assert!(timestamps.windows(2).all(|pair| pair[0] < pair[1])); + } +} diff --git a/rs/moq-mux/src/container/mkv/mod.rs b/rs/moq-mux/src/container/mkv/mod.rs index 0b27efeeed..5d9b8acfb5 100644 --- a/rs/moq-mux/src/container/mkv/mod.rs +++ b/rs/moq-mux/src/container/mkv/mod.rs @@ -55,7 +55,7 @@ pub enum Error { #[error("timestamp overflow")] TimestampOverflow, - /// A Tracks element arrived after `finish()`, so the tracks it declares could never finish. + /// Input arrived after `finish()`, so any tracks it declares could never finish. #[error("Tracks after finish")] TracksAfterFinish, diff --git a/rs/moq-mux/src/container/mkv/test_data/README.md b/rs/moq-mux/src/container/mkv/test_data/README.md new file mode 100644 index 0000000000..1523d75d76 --- /dev/null +++ b/rs/moq-mux/src/container/mkv/test_data/README.md @@ -0,0 +1,13 @@ +# WebM fixture + +`opus.webm` is 0.1 seconds of generated 440 Hz mono audio at 48 kHz, +encoded as six Opus packets in a live WebM Segment with unknown size. +It was generated with ffmpeg 9.0.2: + +```sh +ffmpeg -f lavfi -i sine=frequency=440:sample_rate=48000 -t 0.1 \ + -c:a libopus -f webm -live 1 -map_metadata -1 -fflags +bitexact opus.webm +``` + +The importer test checks constructor and streaming error propagation, +then reads all six emitted frames across one-byte and larger chunk boundaries. diff --git a/rs/moq-mux/src/container/mkv/test_data/opus.webm b/rs/moq-mux/src/container/mkv/test_data/opus.webm new file mode 100644 index 0000000000..18c3c562ad Binary files /dev/null and b/rs/moq-mux/src/container/mkv/test_data/opus.webm differ