audit: byte caps on GOP buffers, opener-scan resume, honest video codec

Round-6 findings from the 10-phase release audit:

- Wire the documented MAX_PENDING_BYTES byte cap into the MPEG-2 GOP
  buffer (it was dead code) and add an equivalent MAX_GOP_BYTES cap to the
  sparse-PTS reorder, so a crafted stream of few-but-huge access units
  cannot over-allocate — both were bounded only by frame count before.
- probe_evo_streams defaulted an unsniffable HD-DVD video stream to H.264,
  which mis-parses a VC-1 (or still-encrypted) clip into a corrupt track.
  Emit the video stream only when the codec is actually identified — the
  honest outcome, matching the audio path (a real clear clip always carries
  its sequence header at the head).
- Resume the AU-opener search from a cursor (like the boundary search), so
  a long unsynced junk run is O(bytes), not O(buffer) per push.
- Mark mpeg2's now-dead MAX_AU_BUFFER test-only; restore #[doc(hidden)] on
  the aacs probe harness module.
- Add regression tests: the 0xFD video-routing guard, the FMTS-is-UHD key
  state, and the GOP byte caps.

src/mux/au_assembly.rs

@@ -121,6 +121,11 @@ pub(crate) struct AuAssembler {
     /// boundary is "the next opener after a frame is already seen"). Meaningless
     /// for `Mode::StartCode`. Reset with `scan_pos`.
     seen_unit: bool,
+    /// Pre-sync opener-search cursor: the offset up to which the buffer has been
+    /// searched for the FIRST AU opener with none found. Resumes the opener scan
+    /// so a long run of junk with no start code (hostile/corrupt input) costs
+    /// O(bytes) total, not O(buffer) per push. Reset when `buf[0]` moves.
+    opener_pos: usize,
 }
 
 impl AuAssembler {
@@ -143,6 +148,7 @@ impl AuAssembler {
             disc_marks: VecDeque::new(),
             scan_pos: 0,
             seen_unit: false,
+            opener_pos: 0,
         }
     }
 
@@ -159,6 +165,7 @@ impl AuAssembler {
             disc_marks: VecDeque::new(),
             scan_pos: 0,
             seen_unit: false,
+            opener_pos: 0,
         }
     }
 
@@ -245,11 +252,11 @@ impl AuAssembler {
         if matches!(self.mode, Mode::Passthrough) {
             return Vec::new();
         }
-        let mode = self.mode;
         let mut out = Vec::new();
         loop {
-            // Locate the AU start code that opens the buffered run.
-            let Some(a0) = au_opener(mode, &self.buf) else {
+            // Locate the AU start code that opens the buffered run (resumes from
+            // opener_pos so an unsynced junk run is scanned once, not per push).
+            let Some(a0) = self.au_opener_resumable() else {
                 // No AU boundary buffered. Bound memory: drop all but a 3-byte
                 // tail (enough to catch a start-code prefix straddling the cut)
                 // once over the cap; otherwise wait for more data.
@@ -329,6 +336,22 @@ impl AuAssembler {
     fn reset_scan(&mut self) {
         self.scan_pos = 0;
         self.seen_unit = false;
+        self.opener_pos = 0;
+    }
+
+    /// Locate the first AU opener in `buf`, resuming the search from `opener_pos`
+    /// (bytes already searched with no opener) so a long unsynced run costs
+    /// O(bytes) total, not O(buffer) per push. Advances `opener_pos` on a miss.
+    fn au_opener_resumable(&mut self) -> Option<usize> {
+        match au_opener_from(self.mode, &self.buf, self.opener_pos) {
+            Some(o) => Some(o),
+            None => {
+                // Nothing yet; next call resumes here (back up 3 for a straddling
+                // start-code prefix). Never advance past what is searchable.
+                self.opener_pos = self.buf.len().saturating_sub(3).max(self.opener_pos);
+                None
+            }
+        }
     }
 
     /// Find the end of the AU that opens at `buf[0]`, resuming from `scan_pos`
@@ -401,13 +424,13 @@ impl AuAssembler {
 
 /// Offset of the start code that opens the next AU in `buf` (at or after 0), or
 /// `None` if no AU-opening start code is buffered yet.
-fn au_opener(mode: Mode, buf: &[u8]) -> Option<usize> {
+fn au_opener_from(mode: Mode, buf: &[u8], from: usize) -> Option<usize> {
     match mode {
-        Mode::StartCode(marker) => find_start_code(buf, 0, marker),
+        Mode::StartCode(marker) => find_start_code(buf, from, marker),
         // Any of the three AU-opening BDU types opens a VC-1 access unit.
-        Mode::Vc1 => find_vc1_start(buf, 0),
+        Mode::Vc1 => find_vc1_start(buf, from),
         // A sequence header, GOP header, or picture opens an MPEG-2 access unit.
-        Mode::Mpeg2 => find_mpeg2_start(buf, 0),
+        Mode::Mpeg2 => find_mpeg2_start(buf, from),
         Mode::Passthrough => None,
     }
 }

src/mux/codec/mpeg2.rs

@@ -46,10 +46,10 @@ const PICTURE_CODE: u8 = 0x00;
 /// Picture coding type: I-frame.
 const PICTURE_TYPE_I: u8 = 1;
 
-/// Hard cap on the access-unit reassembly buffer. A real MPEG-2 frame is well
-/// under 1 MiB (DVD I-frames ~100 KB); past this cap a corrupt stream that
-/// never produces a second access-unit boundary is force-flushed as a single
-/// frame rather than driving unbounded allocation.
+/// The access-unit reassembly cap now lives in [`crate::mux::au_assembly`] (the
+/// `AuAssembler` owns cross-PES buffering); this mirror exists only so the
+/// force-flush test below can size an over-cap fixture against the same bound.
+#[cfg(test)]
 const MAX_AU_BUFFER: usize = 8 * 1024 * 1024;
 
 /// Cap on frames held awaiting the first PES PTS anchor. A DVD stamps a PTS in
@@ -259,8 +259,11 @@ impl Mpeg2Parser {
             },
         });
         // Safety cap: a stream with no GOP/sequence boundaries would buffer
-        // unbounded. Force-flush a pathologically long run as its own GOP.
-        if self.gop_buf.len() >= MAX_PENDING_FRAMES {
+        // unbounded. Force-flush a pathologically long run as its own GOP —
+        // bounded by BOTH the frame count and the total buffered bytes, so a
+        // crafted stream of few-but-huge pictures cannot over-allocate either.
+        let gop_bytes: usize = self.gop_buf.iter().map(|p| p.frame.data.len()).sum();
+        if self.gop_buf.len() >= MAX_PENDING_FRAMES || gop_bytes >= MAX_PENDING_BYTES {
             self.flush_gop(out);
         }
     }
@@ -1402,13 +1405,12 @@ mod tests {
         let mut data = make_picture_header(PICTURE_TYPE_I);
         // > MAX_AU_BUFFER of slice bytes with no following picture/seq/GOP.
         data.extend(std::iter::repeat_n(0xAA, MAX_AU_BUFFER + 1024));
-        let frames = parser.parse(&make_pes(data, Some(0)));
-        assert!(
-            frames.is_empty(),
-            "over-cap AU is force-COMPLETED (bounded) but buffered in its GOP"
-        );
-        let frames = parser.flush();
-        assert_eq!(frames.len(), 1, "force-flushed at EOF, not dropped");
+        // The AU assembler force-completes the ~8 MiB AU (no boundary), and the
+        // GOP byte cap (MAX_PENDING_BYTES) then force-flushes that oversized GOP
+        // during parse rather than buffering it unbounded.
+        let mut frames = parser.parse(&make_pes(data, Some(0)));
+        frames.extend(parser.flush());
+        assert_eq!(frames.len(), 1, "over-cap AU force-flushed, not dropped");
         assert!(frames[0].keyframe);
     }
 

src/mux/codec/reorder.rs

@@ -44,6 +44,13 @@ const FALLBACK_FRAME_DUR_NS: i64 = 1_001_000_000 / 24;
 /// reassembly buffer grows unbounded on disc-controlled input.
 const MAX_GOP_FRAMES: usize = 600;
 
+/// Byte cap on the buffered GOP, complementing [`MAX_GOP_FRAMES`]. A GOP holds a
+/// couple hundred MB at most in practice; this force-completes a run of
+/// few-but-huge access units so a crafted/corrupt stream cannot over-allocate
+/// (the AU assembler caps each frame at 8 MiB, so 600 frames alone could reach
+/// ~5 GiB without this).
+const MAX_GOP_BYTES: usize = 64 * 1024 * 1024;
+
 /// One buffered coded picture awaiting its GOP's completion.
 struct Pending {
     /// Explicit PES PTS (ns) for this AU, or `None` when the source omitted it.
@@ -72,6 +79,9 @@ struct Gop {
 pub(crate) struct SparsePtsReorder {
     /// Frames of the GOP currently accumulating, in decode order.
     cur: Vec<Pending>,
+    /// Total `data` bytes buffered in `cur` — the byte-cap counter, reset each
+    /// time `cur` is drained into a completed GOP.
+    cur_bytes: usize,
     /// The previously-completed GOP, held one step so its duration can be
     /// calibrated from the next GOP's anchor before it is emitted.
     held: Option<Gop>,
@@ -86,6 +96,7 @@ impl SparsePtsReorder {
     pub(crate) fn new() -> Self {
         Self {
             cur: Vec::new(),
+            cur_bytes: 0,
             held: None,
             dur_ns: 0,
             next_start_ns: 0,
@@ -102,11 +113,15 @@ impl SparsePtsReorder {
         // A keyframe opens a new GOP: the picture already accumulated in `cur` is
         // a complete GOP. Complete it (this frame belongs to the NEW GOP). Also
         // force-complete a pathologically long run that never signalled a
-        // keyframe, so a crafted/corrupt stream cannot buffer without bound.
+        // keyframe — bounded by BOTH frame count and total buffered bytes, so a
+        // crafted/corrupt stream of few-but-huge access units cannot buffer
+        // without bound.
         let mut out = Vec::new();
-        if (frame.keyframe || self.cur.len() >= MAX_GOP_FRAMES) && !self.cur.is_empty() {
+        let over_cap = self.cur.len() >= MAX_GOP_FRAMES || self.cur_bytes >= MAX_GOP_BYTES;
+        if (frame.keyframe || over_cap) && !self.cur.is_empty() {
             out = self.complete_current_gop();
         }
+        self.cur_bytes += frame.data.len();
         self.cur.push(Pending {
             explicit,
             ctype,
@@ -131,6 +146,7 @@ impl SparsePtsReorder {
             return Vec::new();
         }
         let pend = std::mem::take(&mut self.cur);
+        self.cur_bytes = 0;
         let dispidx = display_indices(pend.iter().map(|p| p.ctype));
         let count = pend.len() as i64;
         let anchor = pend
@@ -310,6 +326,29 @@ mod tests {
         );
     }
 
+    #[test]
+    fn force_flushes_a_gop_that_exceeds_the_byte_cap() {
+        use CodingType::*;
+        // Few-but-huge access units with no keyframe must not accumulate past the
+        // byte cap: a handful of ~MAX_GOP_BYTES/4-sized frames force-completes the
+        // GOP well before the frame-count cap, bounding memory.
+        let big = MAX_GOP_BYTES / 4 + 1;
+        let mut r = SparsePtsReorder::new();
+        let mut emitted = 0usize;
+        // Enough huge frames to trigger several byte-cap completions (a GOP is
+        // held one step for duration calibration, so the first emit lands after
+        // the second cap fires) — well under the 600-frame count cap.
+        for i in 0..16 {
+            let mut f = frame(P, false);
+            f.data = vec![0u8; big];
+            emitted += r.push((i == 0).then_some(0), f).len();
+        }
+        assert!(
+            emitted >= 1,
+            "byte cap force-flushed (emitted {emitted}) before the frame-count cap"
+        );
+    }
+
     #[test]
     fn force_flushes_a_gop_that_never_signals_a_keyframe() {
         use CodingType::*;