From 5bdb9b21c92ee17a1b88adb246e1a2913f207586 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 30 Oct 2026 02:04:45 -0700 Subject: [PATCH 1/4] chore(quest): claim quest/m2/wire-fuzz-coverage.md Quest-Claim: 59a150c4dac89f40 From 537cc3e51ef709d3dda6233ca385281b83ce61c5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 30 Oct 2026 02:27:48 -0700 Subject: [PATCH 2/4] test(net): fuzz group ingest and request framing with invariants Two moq-net fuzz targets, plus seeds at every length cap: - ingest drives lite's FrameIngest and moq-transport's GroupIngest over a stream delivered whole, byte by byte, and in fuzzer-chosen chunks, and requires the same frames and ending from each, no stall, and payload and header accounting that matches the stream's length. - request reads a request stream through Next, checks the framing is exact, and requires Body::message to agree with the control stream's length-prefixed decode for every message type sharing the ID. The request target found Next::decode peeking the type on a clone, so a type cut short reported need() == 0. Let it fall through to T, which records what it needed on the real decoder. Rewrite quest/m2/wire-fuzz-coverage.md to the cmsg targets and Miri. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/nightly.yml | 2 +- quest/m2/README.md | 2 +- quest/m2/wire-fuzz-coverage.md | 27 +- rs/moq-net/fuzz/Cargo.toml | 14 + rs/moq-net/fuzz/README.md | 18 +- rs/moq-net/fuzz/fuzz_targets/ingest.rs | 7 + rs/moq-net/fuzz/fuzz_targets/request.rs | 7 + rs/moq-net/fuzz/regressions/README.md | 2 +- .../request/draft22-next-short-type | Bin 0 -> 11 bytes rs/moq-net/src/fuzz.rs | 58 +++ rs/moq-net/src/fuzz/ingest.rs | 448 ++++++++++++++++++ rs/moq-net/src/fuzz/request.rs | 233 +++++++++ rs/moq-net/src/ietf/mod.rs | 6 + rs/moq-net/src/ietf/request_stream.rs | 12 +- rs/moq-net/src/ietf/subscriber.rs | 14 +- rs/moq-net/src/lite/mod.rs | 3 + rs/moq-net/src/lite/subscriber.rs | 6 +- 17 files changed, 827 insertions(+), 32 deletions(-) create mode 100644 rs/moq-net/fuzz/fuzz_targets/ingest.rs create mode 100644 rs/moq-net/fuzz/fuzz_targets/request.rs create mode 100644 rs/moq-net/fuzz/regressions/request/draft22-next-short-type create mode 100644 rs/moq-net/src/fuzz/ingest.rs create mode 100644 rs/moq-net/src/fuzz/request.rs diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 6e4936f982c..0cad9078072 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -675,7 +675,7 @@ jobs: strategy: fail-fast: false matrix: - target: [lite, announce, ietf, varint, path, pattern, packet, params, streamid, streams, frame, connection] + target: [lite, announce, ietf, varint, path, pattern, ingest, request, packet, params, streamid, streams, frame, connection] steps: - name: Checkout uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 diff --git a/quest/m2/README.md b/quest/m2/README.md index 4f4f2452b0a..0ab2574c5a4 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -22,7 +22,7 @@ lite-07; bitrate enforcement is not part of this release. - [Release execution](/quest/m2/release/README.md) - publication, installed-artifact proof, cleanup, and downstream adoption after m1 - [Untrusted media parsers have fuzz targets](/quest/m2/media-fuzz.md) - fMP4, MKV, FLV, TS, Annex-B, catalog, and AMF0 parsing fuzzed with a bounded-work invariant -- [Wire fuzz coverage](/quest/m2/wire-fuzz-coverage.md) - the audited moq-net decoders have targets that assert an invariant, and the unsafe peer input reaches runs under Miri +- [Wire fuzz coverage](/quest/m2/wire-fuzz-coverage.md) - the cmsg walks in moq-sock and moq-uring have invariant fuzz targets, and the unsafe peer input reaches runs under Miri - [In-memory unsafe runs under Miri](/quest/m2/miri-memory.md) - kio's and moq-net's memory-only unsafe runs under a nightly Miri job - [VA-API renderer tests](/quest/m2/vaapi-render-tests.md) - the VA-API renderer, decode, and transcode tests run in `just rs gpu` on AMD and Intel and fail instead of skipping, and hosted CI reports them ignored - [Every opt-in test target has a CI caller](/quest/m2/test-inventory.md) - the merge gate flags ignored tests, fuzz and bench targets, recipes, and scripts no workflow runs diff --git a/quest/m2/wire-fuzz-coverage.md b/quest/m2/wire-fuzz-coverage.md index 2a3cf12d60c..f2a60fd67cb 100644 --- a/quest/m2/wire-fuzz-coverage.md +++ b/quest/m2/wire-fuzz-coverage.md @@ -1,30 +1,33 @@ -# [L] The audited moq-net decoders and peer-reachable unsafe have invariant fuzz targets +# [M] Peer-reachable unsafe has invariant fuzz targets and runs under Miri ## Goal -The moq-net decoders the audit found uncovered are fuzzed with an assertion stronger than "it did not panic", and the hand-written unsafe that peer input reaches in `moq-c`, `moq-uring`, and `moq-sock` runs under Miri or a target of its own. Other peer-facing decoders, such as the media importers and RTMP's AMF0, are outside this quest. +The hand-written unsafe that peer input reaches in `moq-c`, `moq-uring`, and `moq-sock` runs under Miri or a fuzz target of its own, asserting more than "it did not panic". Other peer-facing decoders, such as the media importers and RTMP's AMF0, are outside this quest. -`ietf_wire` already covers both SETUP envelopes and the Setup Options a session reads from them. The gaps that remain: - -- The incremental frame and group ingest (`lite::subscriber::FrameIngest`, `ietf::subscriber::GroupIngest`/`SubgroupRecv`, the `lite::ietf` state/object/property decoders) has no target. -- Request-stream framing (`Body`, `Framed`, `FollowUp`, `Next`, `whole_message`) is only unit-tested. -- Every seed body is six bytes or less, so no length-prefix or count rejection path is exercised near its cap. -- `rs/moq-c`, `rs/moq-uring`, and `rs/moq-sock` have no fuzz target and no Miri run. Their cmsg walks now bound every length by the buffer, with unit tests, but nothing searches for the next mistake. +The moq-net half is done: the `ingest` and `request` targets cover the incremental group ingest and request-stream framing, and the seeds include prefixes at and past every cap. What remains: `rs/moq-c`, `rs/moq-uring`, and `rs/moq-sock` have no fuzz target and no Miri run. Their cmsg walks bound every length by the buffer, with unit tests, but nothing searches for the next mistake. ## Plan Decided 2026-10-22 from the security audit, with the maintainer: -- Add one target for a stream of frame headers, records, and payloads through the incremental ingest, asserting the declared sizes and the stream position stay consistent. -- Add a target for request-stream framing. -- Add seeds that approach a limit (`0xffff` and 64 MiB prefixes) so the rejection paths run. -- Add a Miri job for the crates whose unsafe is reachable from a peer, and a target for `moq-sock`'s cmsg walk and `moq-uring`'s `RecvMeta::parse`. Both walks must stop at a malformed length and keep what came before it. +- Add a target for `moq-sock`'s cmsg walk (`cmsg::Iter` and `ControlMetadata::decode` in `rs/moq-sock/src/udp/unix.rs`) and for `moq-uring`'s `RecvMeta::parse` (`rs/moq-uring/src/udp.rs`). Both walks must stop at a malformed length and keep what came before it: build well-formed control messages from the input, append a malformed header and arbitrary bytes, and assert the walk reads the same metadata as the well-formed prefix, and the metadata a reference model expects from it. +- Add a Miri job for the crates whose unsafe is reachable from a peer. - Wire every target and the Miri run into the nightly gate, and keep the merge-gate replay of committed regressions. The point is the invariant, not the coverage number: a target that only checks for a panic would miss the bugs this audit is about. +Each crate follows moq-net's shape (`rs/moq-net/fuzz/README.md`): the target body in a hidden `fuzz` module so the crate's own tests replay `fuzz/regressions/` on stable, a one-line shim per target, an `--exclude` in the `no_fuzz` list of `rs/justfile`, and a row in `nightly.yml`'s fuzz matrix. Target names must stay unique across crates. + +Ordering: + +- `moq-sock`'s `CMsg::decode` still reads `size_of::()` past a short message (only a `debug_assert_eq!` checks it), which [Bound cmsg payload reads](/quest/m2/cmsg-decode-bound.md) fixes. A moq-sock target landing first would fail nightly on that known overread, so land it after, or with, that fix. The moq-uring target has no such dependency. +- [In-memory unsafe under Miri](/quest/m2/miri-memory.md) shares one `just rs miri` recipe and nightly job with this quest: whichever lands first creates it, and the other adds its crates. + Public API: none. Wire: none. ## Related - [CI cost and latency](/quest/m0/ci/README.md) - where the nightly budget lives +- [Bound cmsg payload reads](/quest/m2/cmsg-decode-bound.md) - fixes the moq-sock overread a cmsg target would hit +- [In-memory unsafe under Miri](/quest/m2/miri-memory.md) - shares the Miri recipe and nightly job +- [Untrusted media parsers have fuzz targets](/quest/m2/media-fuzz.md) - the media parser targets beside these diff --git a/rs/moq-net/fuzz/Cargo.toml b/rs/moq-net/fuzz/Cargo.toml index 0fdff664654..e9812d7ac2b 100644 --- a/rs/moq-net/fuzz/Cargo.toml +++ b/rs/moq-net/fuzz/Cargo.toml @@ -62,6 +62,20 @@ test = false doc = false bench = false +[[bin]] +name = "ingest" +path = "fuzz_targets/ingest.rs" +test = false +doc = false +bench = false + +[[bin]] +name = "request" +path = "fuzz_targets/request.rs" +test = false +doc = false +bench = false + [dependencies] libfuzzer-sys = "0.4" moq-net = { workspace = true, features = ["fuzz"] } diff --git a/rs/moq-net/fuzz/README.md b/rs/moq-net/fuzz/README.md index 5a5c75a5ed6..5b724d87ca6 100644 --- a/rs/moq-net/fuzz/README.md +++ b/rs/moq-net/fuzz/README.md @@ -1,7 +1,8 @@ # Fuzzing moq-net Coverage-guided fuzzing of the wire codecs: the moq-lite and moq-transport control -messages, stream and datagram headers, both varint codecs, and `Path`. These are the +messages, stream and datagram headers, group streams through the incremental ingest, +moq-transport request-stream framing, both varint codecs, and `Path`. These are the bytes a relay parses from an untrusted peer, so they are the surface worth fuzzing. ```bash @@ -9,7 +10,7 @@ cargo install --locked cargo-fuzz just rs fuzz lite ``` -The targets are `lite`, `announce`, `ietf`, `varint`, `path`, and `pattern`. Extra arguments pass through to +The targets are `lite`, `announce`, `ietf`, `varint`, `path`, `pattern`, `ingest`, and `request`. Extra arguments pass through to libFuzzer, so `just rs fuzz lite -- -max_total_time=300` bounds a run. The Nightly workflow runs every target for five minutes, resumes from the previous run's @@ -56,6 +57,19 @@ Beyond "does not panic": PATH and AUTHORITY are one-sided. - `Path::relative` inverts `Path::resolve`, and never produces a reference that walks above the root. +- A group stream ingests to the same frames and the same ending whether it arrives whole, + a byte at a time, or in chunks the input picks, and never stalls once complete. Every + frame matches its track's timeline, Object IDs only rise, and the frames hold no more + bytes than the stream carried. On a clean end, where no block length is hidden, every + byte is a frame, a header field, or a property block. +- A request-stream message is its type, a two-byte length, and exactly that body; at + most one GOAWAY migrates it. Its body decodes as each message its type names exactly + when the control stream's length-prefixed reading accepts it, as the same message, and + the REQUEST_UPDATE readers stop where the framing did. + +`seeds/` also holds length prefixes at and one past each cap (a moq-lite message's size, +a moq-transport length, a frame, a property or extension block), so every rejection path +starts with an input beside it. A refused decode is not a finding unless it asks for more bytes despite a complete body. An encode failure is legitimate: a value can decode at a version that cannot diff --git a/rs/moq-net/fuzz/fuzz_targets/ingest.rs b/rs/moq-net/fuzz/fuzz_targets/ingest.rs new file mode 100644 index 00000000000..9c2d1d3ef61 --- /dev/null +++ b/rs/moq-net/fuzz/fuzz_targets/ingest.rs @@ -0,0 +1,7 @@ +#![no_main] + +use libfuzzer_sys::fuzz_target; + +fuzz_target!(|data: &[u8]| { + moq_net::fuzz::ingest(data); +}); diff --git a/rs/moq-net/fuzz/fuzz_targets/request.rs b/rs/moq-net/fuzz/fuzz_targets/request.rs new file mode 100644 index 00000000000..061aa21be35 --- /dev/null +++ b/rs/moq-net/fuzz/fuzz_targets/request.rs @@ -0,0 +1,7 @@ +#![no_main] + +use libfuzzer_sys::fuzz_target; + +fuzz_target!(|data: &[u8]| { + moq_net::fuzz::request(data); +}); diff --git a/rs/moq-net/fuzz/regressions/README.md b/rs/moq-net/fuzz/regressions/README.md index 7b94df69899..61befbdbfa8 100644 --- a/rs/moq-net/fuzz/regressions/README.md +++ b/rs/moq-net/fuzz/regressions/README.md @@ -1,7 +1,7 @@ # Regressions Crash inputs the fuzzer found, one file per input, in a directory named after its -target (`lite/`, `ietf/`, `varint/`, `path/`). `fuzz::tests::regressions` replays them +target (`lite/`, `ietf/`, `request/`, and so on). `fuzz::tests::regressions` replays them on the stable toolchain as part of `just check`, so a bug found by fuzzing stays fixed without anyone running the fuzzer again. diff --git a/rs/moq-net/fuzz/regressions/request/draft22-next-short-type b/rs/moq-net/fuzz/regressions/request/draft22-next-short-type new file mode 100644 index 0000000000000000000000000000000000000000..73a61f0d2557c223998e07bf94409e4a477a821c GIT binary patch literal 11 ScmZP+U|?ckWM^P>*be{#F##L^ literal 0 HcmV?d00001 diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 45c1f9efea5..d79ad6b60f4 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -10,6 +10,12 @@ //! Compiled only under `cfg(test)` or the `fuzz` feature, so none of this is part of //! the published API. See `fuzz/README.md` for the workflow. +mod ingest; +mod request; + +pub use ingest::ingest; +pub use request::request; + use bytes::Bytes; use crate::{ @@ -93,6 +99,8 @@ pub const TARGETS: &[(&str, Target)] = &[ ("varint", varint), ("path", path), ("pattern", pattern), + ("ingest", ingest), + ("request", request), ]; /// The moq-lite versions a target decodes at, selected by the input's first byte. @@ -1478,9 +1486,46 @@ pub fn seeds() -> Vec { }); } + // Length prefixes at and one past each cap, ahead of a body far too short for them, so + // the fuzzer starts beside every rejection path instead of searching for its bound: a + // moq-lite message's size (0xffff for most, 64 MiB for ANNOUNCE_INIT) and a + // moq-transport control message's two-byte length. + for (index, version) in LITE_VERSIONS.iter().enumerate() { + for kind in 0..LITE_KINDS { + for size in limits(&[0xffff, 64 << 20]) { + let mut data = vec![index as u8, kind]; + Encoder::new(&mut data, (*version).into()).varint(size).unwrap(); + data.extend_from_slice(&[0; 4]); + seeds.push(Seed { + target: "lite", + kind, + data, + }); + } + } + } + for index in 0..IETF_VERSIONS.len() as u8 { + for kind in 0..IETF_KINDS { + seeds.push(Seed { + target: "ietf", + kind, + data: vec![index, kind, 0xff, 0xff, 0, 0, 0, 0], + }); + } + } + + seeds.extend(ingest::seeds()); + seeds.extend(ingest::limit_seeds()); + seeds.extend(request::seeds()); + seeds } +/// Each of `caps`, and one past it. +pub(crate) fn limits(caps: &[u64]) -> impl Iterator + '_ { + caps.iter().flat_map(|cap| [*cap, cap + 1]) +} + type FrameReader = coding::Reader, lite::Version>; /// Frames arriving on concurrent group streams, as the frame bench times them: one @@ -1689,6 +1734,19 @@ mod tests { assert!(missing.is_empty(), "no seed decodes for {missing:?}"); } + /// The stream seeds are built from our own framing, one per version and protocol, so + /// each must decode whole: one that does not leaves its version's path unseeded. Those + /// at a cap must not, or they sit beside no rejection path. + #[test] + fn stream_seeds_decode() { + for seed in ingest::seeds().into_iter().chain(request::seeds()) { + assert!(target(seed.target)(&seed.data), "{} seed {:x?} does not decode", seed.target, seed.data); + } + for seed in ingest::limit_seeds() { + assert!(!ingest(&seed.data), "ingest seed {:x?} decodes past its cap", seed.data); + } + } + /// Replay the crash inputs committed under `fuzz/regressions/`, so a bug the fuzzer /// found stays fixed on the stable toolchain, in the suite CI already runs. #[test] diff --git a/rs/moq-net/src/fuzz/ingest.rs b/rs/moq-net/src/fuzz/ingest.rs new file mode 100644 index 00000000000..a3f8ec70453 --- /dev/null +++ b/rs/moq-net/src/fuzz/ingest.rs @@ -0,0 +1,448 @@ +//! The `ingest` target: a group stream's bytes through the incremental ingest a session +//! runs, moq-lite's [`lite::FrameIngest`] or moq-transport's [`ietf::GroupIngest`]. +//! +//! The ingest is a state machine resumed wherever a read runs short, so what it builds +//! must not depend on where the transport split the stream. Each input runs whole, then +//! a byte at a time, then in chunks the input picks, and all three must leave the same +//! frames and the same ending. Every byte a frame holds came off the stream, and on a +//! clean end every byte of the stream is accounted for by a frame or its header. + +use bytes::Bytes; + +use super::Chunks; +use crate::{ + Timescale, broadcast, cache, + coding::{self, Decode}, + group, ietf, lite, + model::object::{Fields, TrackProperties}, + track, +}; + +/// The longest varint either codec writes: nine bytes for leading-ones, eight for QUIC. +const MAX_VARINT: usize = 9; + +/// One frame as a relay reads it back out of the group. +#[derive(Debug, PartialEq)] +struct Seen { + timestamp: Option, + payload: Bytes, + /// The Object ID and property block the frame kept, if it was not plain. + fields: Option<(u64, Option)>, + inflated: Option, +} + +/// What one run of the ingest left behind. +#[derive(Debug, PartialEq)] +struct Ingested { + frames: Vec, + /// How the stream ended: cleanly with what it said, or refused with the error. + end: Result, +} + +/// How a run reads the stream: the session's settings, decoded from the input's first +/// bytes, and the stream itself. +#[derive(Clone, Copy)] +enum Session { + Lite { + version: lite::Version, + /// The session negotiated the `ietf` extension, so each frame follows a record. + records: bool, + frame_start: u64, + }, + Ietf { + version: ietf::Version, + }, +} + +/// The track a stream's group belongs to. +fn track_info(timed: bool, compress: bool, draft: Option) -> track::Info { + let mut info = track::Info::default(); + if timed { + info = info.with_timescale(Timescale::MILLI); + } + info.compress = compress; + info.properties = draft.map(|version| { + std::sync::Arc::new(TrackProperties { + version, + values: Vec::new(), + }) + }); + info +} + +/// Decode a group stream from arbitrary bytes. +/// +/// Byte 0 picks the protocol and version, byte 1 the track's settings, byte 2 the chunk +/// size of the third run, and the rest is the stream. For moq-transport the stream opens +/// with the subgroup header, which is decoded first, as the session peeks it. +pub fn ingest(data: &[u8]) -> bool { + let [selector, settings, chunk, rest @ ..] = data else { + return false; + }; + let timed = settings & 0x1 != 0; + let compress = settings & 0x2 != 0; + let draft = super::IETF_VERSIONS.get(((settings >> 2) & 0x3) as usize).copied(); + let chunk = (*chunk as usize).max(1); + + let (session, info, body) = match selector % 2 { + 0 => { + let version = super::LITE_VERSIONS[(*selector as usize / 2) % super::LITE_VERSIONS.len()]; + // Lite05/06 timestamps may be send times, so their tracks arrive untimed. + let timed = timed && version.has_media_time(); + let session = Session::Lite { + version, + records: settings & 0x10 != 0 && version.has_ietf(), + frame_start: (settings >> 5) as u64, + }; + // Only lite-07 can mark a track compressed. + let compress = compress && version == lite::Version::Lite07; + (session, track_info(timed, compress, draft), rest) + } + _ => { + let version = super::IETF_VERSIONS[(*selector as usize / 2) % super::IETF_VERSIONS.len()]; + // A moq-transport track was always published over moq-transport. + (Session::Ietf { version }, track_info(timed, compress, Some(version)), rest) + } + }; + + let whole = run(session, &info, body, usize::MAX); + let bytewise = run(session, &info, body, 1); + assert_eq!(whole, bytewise, "a byte at a time ingests differently"); + if chunk != 1 { + let chunked = run(session, &info, body, chunk); + assert_eq!(whole, chunked, "{chunk}-byte chunks ingest differently"); + } + + let Some(whole) = whole else { + return false; + }; + check(session, &info, body, &whole); + whole.end.is_ok() +} + +/// Hold one run to what the stream could have said. +fn check(session: Session, info: &track::Info, body: &[u8], ingested: &Ingested) { + let mut ids = None; + let mut payload = 0; + for (index, frame) in ingested.frames.iter().enumerate() { + assert_eq!( + frame.timestamp.is_some(), + info.timescale.is_some(), + "frame {index} does not match its track's timeline" + ); + assert_eq!(frame.inflated.is_some(), info.compress, "frame {index} lost its inflated size"); + let id = frame.fields.as_ref().map_or(index as u64, |(id, _)| *id); + assert!(ids.is_none_or(|prior| id > prior), "Object ID {id} does not follow {ids:?}"); + ids = Some(id); + payload += frame.payload.len(); + } + + // Each frame has at least a one-byte size ahead of its payload. + let frames = ingested.frames.len(); + assert!( + payload + frames <= body.len(), + "{frames} frames hold {payload} bytes from a {}-byte stream", + body.len() + ); + + // On a clean end nothing was skipped: every byte is a frame, a header field, or a + // property block. A moq-transport extension block is decoded rather than kept, so its + // length is unknown and the bound only applies without one. + let Ok(_) = &ingested.end else { return }; + let fields = match session { + Session::Lite { records, .. } => { + // Timing, inflated size, and size; a record adds its code and block length. + let fields = 3 + 2 * records as usize; + // The Group State, and an END_OF_GROUP record with its block length. + let trailer = (1 + 2 * records as usize) * MAX_VARINT; + let blocks: usize = ingested + .frames + .iter() + .filter_map(|frame| frame.fields.as_ref()?.1.as_ref()) + .map(Bytes::len) + .sum(); + Some((fields, trailer, blocks)) + } + // The ID delta and the size, then an empty object's status. A status object that + // ends the group, or the track, repeats them. + Session::Ietf { .. } => match info.compress || info.timescale.is_some() { + true => None, + false => Some((3, 3 * MAX_VARINT, 0)), + }, + }; + if let Some((fields, trailer, blocks)) = fields { + let header = header_len(session, body); + let most = header + payload + blocks + frames * fields * MAX_VARINT + trailer; + assert!( + body.len() <= most, + "{frames} frames of {payload} bytes account for at most {most} of a {}-byte stream", + body.len() + ); + } +} + +/// The subgroup header's length, which a moq-transport run decodes before its objects. +fn header_len(session: Session, body: &[u8]) -> usize { + match session { + Session::Lite { .. } => 0, + Session::Ietf { version } => ietf::GroupHeader::decode_slice(body, version).map_or(0, |(_, used)| used), + } +} + +/// Run the ingest over `body`, delivered `chunk` bytes at a time, and read back the group. +/// +/// `None` when the stream does not open a group at all. +fn run(session: Session, info: &track::Info, body: &[u8], chunk: usize) -> Option { + let mut broadcast = broadcast::Info::new(); + // Unbounded, so eviction never aborts a group mid-frame. + broadcast.pool = cache::Pool::unbounded(); + let broadcast = broadcast.produce(); + let track = broadcast.create_track("ingest", info.clone()).ok()?; + + let (end, group) = match session { + Session::Lite { + version, + records, + frame_start, + } => { + let mut reader = reader(Bytes::copy_from_slice(body), chunk, version); + let producer = track.append_group().ok()?; + let consumer = producer.consume(); + let mut group = crate::recv::Group::new(producer); + let framing = lite::Framing::new(info, version); + let mut ingest = lite::FrameIngest::new(framing, version, records, frame_start); + let end = drive(body.len(), |waiter| ingest.poll(&mut reader, &mut group, waiter)); + (end.map(|()| "fin".to_string()), (group, consumer)) + } + Session::Ietf { version } => { + let (header, used) = ietf::GroupHeader::decode_slice(body, version).ok()?; + let mut reader = reader(Bytes::copy_from_slice(&body[used..]), chunk, version); + let producer = track.create_group(group::Info { sequence: header.group_id }).ok()?; + producer + .set_wire(group::Wire { + sparse: false, + first_object: header.flags.first_object, + first_id: 0, + extensions: header.flags.has_extensions, + has_end: header.flags.has_end, + fetchable: true, + last: None, + }) + .ok()?; + let consumer = producer.consume(); + let mut group = crate::recv::Group::new(producer); + let mut ingest = ietf::GroupIngest::new(version, &header, info.timescale); + let end = drive(body.len(), |waiter| ingest.poll(&mut reader, &mut group, waiter)); + (end.map(|ended| format!("{ended:?}")), (group, consumer)) + } + }; + + let (group, consumer) = group; + // Settle the group as the session does, so the reader below never waits. + match &end { + Ok(_) => { + let _ = group.finish(); + } + Err(err) => { + let _ = group.abort(err.clone()); + } + } + + Some(Ingested { + frames: read(consumer, info.compress), + end: end.map_err(|err| err.to_string()), + }) +} + +type Stream = coding::Reader, V>; + +fn reader(payload: Bytes, chunk: usize, version: V) -> Stream { + let stream = Chunks { + payload, + chunk, + ready: true, + }; + coding::Reader::new(crate::transport::poll::Buffered::new(stream), version) +} + +/// Poll the ingest until it settles. +/// +/// The stream hands over a chunk every other poll and the ingest yields every few frames, +/// so a stream of `len` bytes settles within a few polls per byte. One that does not is +/// waiting on something other than the stream, which a peer could hold forever. +fn drive( + len: usize, + mut poll: impl FnMut(&kio::Waiter) -> std::task::Poll>, +) -> Result { + let waiter = &kio::Waiter::noop(); + for _ in 0..4 * len + 64 { + if let std::task::Poll::Ready(res) = poll(waiter) { + return res; + } + } + panic!("the ingest stalled on a complete {len}-byte stream"); +} + +/// Read every frame the group holds, in its stored form. +fn read(consumer: group::Consumer, compress: bool) -> Vec { + let mut consumer = match compress { + true => consumer.compressed(), + false => consumer, + }; + let waiter = &kio::Waiter::noop(); + let mut frames = Vec::new(); + while let std::task::Poll::Ready(Ok(Some(object))) = consumer.poll_read_object(waiter) { + let fields = object.fields.as_deref().map(|Fields { id, properties }| { + let block = properties.block.as_ref().map(|block| block.bytes.clone()); + (*id, block) + }); + frames.push(Seen { + timestamp: object.frame.timestamp, + payload: object.frame.payload.contiguous(0), + fields, + inflated: object.inflated, + }); + } + frames +} + +/// The streams the seeds open with: one group per protocol, built from our own framing. +pub(super) fn seeds() -> Vec { + let mut seeds = Vec::new(); + let mut push = |data: Vec| { + seeds.push(super::Seed { + target: "ingest", + kind: 0, + data, + }) + }; + + // moq-lite: three plain frames, at every version, timed and untimed. The timing + // prefix is a zigzag delta, so 0x02 steps the clock forward by one. + for (index, version) in super::LITE_VERSIONS.iter().enumerate() { + for timed in [false, true] { + let mut data = vec![(index * 2) as u8, timed as u8, 3]; + let timing = version.wire_timescale(timed.then_some(Timescale::MILLI)).is_some(); + for (n, payload) in [&b"abc"[..], b"", b"defgh"].into_iter().enumerate() { + if timing { + data.push(if n == 0 { 0 } else { 2 }); + } + data.push(payload.len() as u8); + data.extend_from_slice(payload); + } + push(data); + } + } + + // moq-lite-07 with the `ietf` extension: a plain Group State, then records. + let lite07 = super::LITE_VERSIONS.len() - 1; + let mut data = vec![(lite07 * 2) as u8, 0x10, 2, 0]; + for payload in [&b"ab"[..], b"c"] { + // Object 0 is the first record's absolute ID, then each next one is a zero delta, + // and neither has a property block. + data.extend_from_slice(&[0, 0, payload.len() as u8]); + data.extend_from_slice(payload); + } + push(data); + + // moq-transport: a subgroup header, then objects, each ID delta, size, payload. + for (index, version) in super::IETF_VERSIONS.iter().enumerate() { + let Some(encoded) = subgroup(*version, false) else { + continue; + }; + let mut data = vec![(index * 2 + 1) as u8, 0, 3]; + data.extend_from_slice(&encoded); + // Object 0, then a zero delta to object 1. + for payload in [&b"abc"[..], b"de"] { + data.extend_from_slice(&[0, payload.len() as u8]); + data.extend_from_slice(payload); + } + // An empty object with status 0, a plain zero-length frame. + data.extend_from_slice(&[0, 0, 0]); + push(data); + } + + seeds +} + +/// Streams that declare a size at or one past a cap, ahead of far fewer bytes, so the +/// fuzzer starts beside each rejection path instead of searching for its bound: a +/// frame's size, a moq-lite record's property block, and a moq-transport object's +/// extension block. +pub(super) fn limit_seeds() -> Vec { + let mut seeds = Vec::new(); + let mut push = |data: Vec| { + seeds.push(super::Seed { + target: "ingest", + kind: 0, + data, + }) + }; + + for (index, version) in super::IETF_VERSIONS.iter().enumerate() { + let Some(encoded) = subgroup(*version, false) else { + continue; + }; + let varint = |value: u64| { + let mut out = Vec::new(); + coding::Encoder::new(&mut out, (*version).into()).varint(value).unwrap(); + out + }; + for size in super::limits(&[group::MAX_CACHE_BYTES]) { + let mut data = vec![(index * 2 + 1) as u8, 0, 3]; + data.extend_from_slice(&encoded); + data.push(0); + data.extend(varint(size)); + data.extend_from_slice(&[0; 4]); + push(data); + } + let Some(encoded) = subgroup(*version, true) else { + continue; + }; + for size in super::limits(&[ietf::MAX_OBJECT_EXTENSIONS as u64]) { + let mut data = vec![(index * 2 + 1) as u8, 0, 3]; + data.extend_from_slice(&encoded); + data.push(0); + data.extend(varint(size)); + data.extend_from_slice(&[0; 4]); + push(data); + } + } + + let lite07 = super::LITE_VERSIONS.len() - 1; + // The caps on moq-lite-07: a frame's size, and a record's property block. + let varint = |value: u64| { + let mut out = Vec::new(); + coding::Encoder::new(&mut out, lite::Version::Lite07.into()).varint(value).unwrap(); + out + }; + for size in super::limits(&[group::MAX_CACHE_BYTES]) { + let mut data = vec![(lite07 * 2) as u8, 0, 3]; + data.extend(varint(size)); + data.extend_from_slice(&[0; 4]); + push(data); + } + for size in super::limits(&[lite::ietf::MAX_PROPERTIES as u64]) { + let mut data = vec![(lite07 * 2) as u8, 0x10, 3, 0, 0]; + data.extend(varint(size)); + data.extend_from_slice(&[0; 4]); + push(data); + } + + seeds +} + +/// Group 0's subgroup header, with or without object extensions. +fn subgroup(version: ietf::Version, has_extensions: bool) -> Option> { + let header = ietf::GroupHeader { + track_alias: 1, + group_id: 0, + sub_group_id: 0, + publisher_priority: 0, + flags: ietf::GroupFlags { + has_extensions, + ..Default::default() + }, + }; + coding::Encode::encode_bytes(&header, version).ok().map(Vec::from) +} diff --git a/rs/moq-net/src/fuzz/request.rs b/rs/moq-net/src/fuzz/request.rs new file mode 100644 index 00000000000..28145604acb --- /dev/null +++ b/rs/moq-net/src/fuzz/request.rs @@ -0,0 +1,233 @@ +//! The `request` target: the messages on a moq-transport request stream, framed as the +//! session reads them. +//! +//! A request stream is read one whole `[type][length][body]` message at a time +//! ([`ietf::Framed`]), at most one GOAWAY migrating it ([`ietf::Next`]), and the body is +//! decoded once the type is known ([`ietf::Body::message`]). The control stream decodes +//! the same messages through their own length prefix, so both readings must agree. + +use crate::{ + coding::{Decode, DecodeError, Decoder, Encoder, Form}, + ietf, + ietf::Message, +}; + +/// Decode a request stream from arbitrary bytes. +/// +/// Byte 0 picks the draft and the rest is the stream. Returns whether any message decoded. +pub fn request(data: &[u8]) -> bool { + let Some((&selector, mut rest)) = data.split_first() else { + return false; + }; + let version = super::IETF_VERSIONS[selector as usize % super::IETF_VERSIONS.len()]; + + let mut any = false; + let mut goaway = false; + while !rest.is_empty() { + let mut reader = Decoder::new(rest, version.into()); + let next = match ietf::Next::::decode(&mut reader, version) { + Ok(next) => next, + Err(DecodeError::Short) => { + assert!(reader.need() > rest.len(), "short read inside a complete message"); + break; + } + Err(_) => break, + }; + let used = reader.position(); + any = true; + + match next { + // A second GOAWAY ends the session, so nothing after it is read. + ietf::Next::GoAway { .. } if std::mem::replace(&mut goaway, true) => break, + ietf::Next::GoAway { .. } => {} + ietf::Next::Message(framed) => { + framing(&rest[..used], &framed, version); + body(&framed, version); + } + } + rest = &rest[used..]; + } + any +} + +/// A whole message is its type, a two-byte length, and exactly that many bytes, and +/// nothing past it was read. +fn framing(consumed: &[u8], framed: &ietf::Framed, version: ietf::Version) { + let ietf::Framed(id, ietf::Body(body)) = framed; + let form: Form = version.into(); + let mut decoder = Decoder::new(consumed, form); + assert_eq!(decoder.varint().ok(), Some(*id), "message type {id:#x} was not read from the stream"); + let id_len = decoder.position(); + let length = u16::from_be_bytes([consumed[id_len], consumed[id_len + 1]]) as usize; + assert_eq!(length, body.len(), "message {id:#x} body is not its declared length"); + assert_eq!(consumed.len(), id_len + 2 + length, "message {id:#x} read past its body"); + assert_eq!(&consumed[id_len + 2..], &body[..], "message {id:#x} body is not the stream's bytes"); + + // Not a GOAWAY, or the stream would have migrated instead. + assert_ne!(*id, ietf::GoAway::ID, "a GOAWAY was read as a request message"); + + // The hand-written REQUEST_UPDATE readings frame the message themselves, and must + // stop where the generic framing did. The message is complete, so none may run short. + fn whole<'a, T: Decode<'a, ietf::Version>>(consumed: &'a [u8], version: ietf::Version) { + match T::decode_slice(consumed, version) { + Ok((_, used)) => assert_eq!(used, consumed.len(), "an update read past its frame"), + Err(DecodeError::Short) => panic!("an update ran short inside a complete frame"), + Err(_) => {} + } + } + if *id == ietf::SubscribeUpdate::ID { + whole::(consumed, version); + whole::>(consumed, version); + whole::>(consumed, version); + } +} + +/// Decode a body as every message its type names, and check the control stream's reading +/// agrees. Some types are shared (REQUEST_UPDATE, REQUEST_OK), told apart by the request. +fn body(framed: &ietf::Framed, version: ietf::Version) { + let ietf::Framed(id, body) = framed; + let mut prefixed = Vec::with_capacity(body.0.len() + 2); + prefixed.extend_from_slice(&(body.0.len() as u16).to_be_bytes()); + prefixed.extend_from_slice(&body.0); + + // A macro rather than a generic function, so each message borrows the body for as long + // as its own decode needs, not for one lifetime every message shares. + macro_rules! agree { + ($($ty:ty),* $(,)?) => {$( + if *id == <$ty>::ID { + let framed = body.message::<$ty>(version); + // The control stream's reading: behind its own two-byte length, which must + // bound it exactly. + let control = <$ty>::decode_slice(&prefixed, version).map(|(msg, used)| { + assert_eq!(used, prefixed.len(), "a control message stopped short of its length"); + format!("{msg:?}") + }); + match (&framed, &control) { + (Ok(msg), Ok(control)) => assert_eq!(&format!("{msg:?}"), control, "the two readings differ"), + (Err(_), Err(_)) => {} + _ => panic!("a request body reads {framed:?} but the control stream reads {control:?}"), + } + if let Ok(msg) = framed { + let mut encoded = Vec::new(); + // An encode failure is legitimate: a value can decode at a draft that + // cannot express it again. Emitting a body we refuse is not. + if msg.encode_msg(&mut Encoder::new(&mut encoded, version.into()), version).is_ok() { + let echo = ietf::Body(encoded.into()); + let reread = echo + .message::<$ty>(version) + .unwrap_or_else(|err| panic!("could not read our own {msg:?} back: {err}")); + let mut again = Vec::new(); + reread + .encode_msg(&mut Encoder::new(&mut again, version.into()), version) + .expect("could not re-encode what we just encoded"); + assert_eq!(&again[..], &echo.0[..], "encoding is not stable"); + } + } + } + )*}; + } + agree!( + ietf::Subscribe<'_>, + ietf::SubscribeOk, + ietf::SubscribeUpdate, + ietf::Publish<'_>, + ietf::PublishOk, + ietf::PublishDone<'_>, + ietf::Fetch<'_>, + ietf::FetchOk, + ietf::TrackStatus<'_>, + ietf::TrackStatusOk, + ietf::PublishNamespace, + ietf::PublishNamespaceUpdate, + ietf::SubscribeNamespace, + ietf::Namespace, + ietf::NamespaceDone, + ietf::SubscribeTracks, + ietf::PublishSkipped<'_>, + ietf::RequestOk, + ietf::RequestError<'_>, + ); +} + +/// The streams the seeds open with: one of each request message from our own encoder, +/// then the same behind a GOAWAY. +pub(super) fn seeds() -> Vec { + let mut seeds = Vec::new(); + for (index, version) in super::IETF_VERSIONS.iter().enumerate() { + let messages: Vec<(u64, Vec)> = vec![ + encode( + &ietf::RequestOk { + active: None, + largest: None, + }, + *version, + ), + encode( + &ietf::RequestError { + error_code: 1, + reason_phrase: "no".into(), + retry_interval: 0, + }, + *version, + ), + encode( + &ietf::SubscribeOk { + track_alias: 1, + largest: Some(ietf::Location { group: 2, object: 3 }), + properties: Default::default(), + }, + *version, + ), + encode( + &ietf::PublishDone { + status_code: ietf::PublishDoneStatus::TrackEnded.code(), + stream_count: 0, + reason_phrase: "done".into(), + }, + *version, + ), + ] + .into_iter() + .flatten() + .collect(); + + let goaway = encode( + &ietf::GoAway { + new_session_uri: "".into(), + timeout: 0, + request_id: None, + }, + *version, + ); + + for (_, message) in &messages { + let mut data = vec![index as u8]; + data.extend_from_slice(message); + seeds.push(super::Seed { + target: "request", + kind: 0, + data: data.clone(), + }); + if let Some((_, goaway)) = &goaway { + let mut data = vec![index as u8]; + data.extend_from_slice(goaway); + data.extend_from_slice(message); + seeds.push(super::Seed { + target: "request", + kind: 0, + data, + }); + } + } + } + seeds +} + +/// `msg` framed as a request stream carries it, or `None` if this draft cannot. +fn encode<'a, M: Message<'a>>(msg: &M, version: ietf::Version) -> Option<(u64, Vec)> { + let mut out = Vec::new(); + let mut w = Encoder::new(&mut out, version.into()); + w.varint(M::ID).ok()?; + crate::coding::Encode::encode(msg, &mut w, version).ok()?; + Some((M::ID, out)) +} diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index dd187e8f0bb..b01e9e9e1e4 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -80,6 +80,12 @@ pub use subscribe_namespace::*; pub use subscribe_tracks::*; #[cfg(feature = "async")] use subscriber::*; +// The fuzz harness drives a subgroup stream's ingest directly. +#[cfg(all(feature = "async", any(test, feature = "fuzz")))] +pub(crate) use { + request_stream::{Framed, NamespaceUpdate, Next, Update}, + subscriber::GroupIngest, +}; pub use track::*; pub use version::Version; diff --git a/rs/moq-net/src/ietf/request_stream.rs b/rs/moq-net/src/ietf/request_stream.rs index f154559aaad..025b138f842 100644 --- a/rs/moq-net/src/ietf/request_stream.rs +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -52,7 +52,7 @@ pub(super) fn poll_withdrawn( /// One whole `[type][length][body]` message off a request stream, decoded at once so a read /// dropped part-way consumes nothing. #[derive(Debug)] -pub(super) struct Framed(pub(super) u64, pub(super) super::Body); +pub(crate) struct Framed(pub(crate) u64, pub(crate) super::Body); impl<'a> Decode<'a, Version> for Framed { fn decode(r: &mut Decoder<'a>, version: Version) -> Result { @@ -68,7 +68,7 @@ impl<'a> Decode<'a, Version> for Framed { /// A message following the request on a request stream: `T`, or a GOAWAY migrating this /// one request (draft-22 section 9.2). #[derive(Debug)] -pub(super) enum Next { +pub(crate) enum Next { Message(T), /// `redirect` is whether it names a new session, which only a server may do. GoAway { @@ -79,7 +79,9 @@ pub(super) enum Next { impl<'a, T: Decode<'a, Version>> Decode<'a, Version> for Next { fn decode(r: &mut Decoder<'a>, version: Version) -> Result { let mut peek = r.clone(); - if peek.varint()? == super::GoAway::ID { + // A type cut short falls through to `T`, whose own read of it records what it + // needed on `r`, where the reader looks for it. + if peek.varint().is_ok_and(|id| id == super::GoAway::ID) { *r = peek; r.uncounted = r.position(); let msg = super::GoAway::decode(r, version)?; @@ -229,7 +231,7 @@ pub(super) async fn write_error( /// A framed subscription update, preserving omitted preferences. #[derive(Debug)] -pub(super) struct Update { +pub(crate) struct Update { /// The update's own Request ID. pub(super) request_id: super::RequestId, pub(super) priority: Option, @@ -331,7 +333,7 @@ impl<'a> Decode<'a, Version> for Update { /// SUBSCRIBE_TRACKS, which from draft-22 may also carry TRACK_PROPERTY_FILTER and /// FORWARD. #[derive(Debug)] -pub(super) struct NamespaceUpdate { +pub(crate) struct NamespaceUpdate { pub(super) request_id: super::RequestId, pub(super) prefix: Option, pub(super) refused: Option, diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index f70b2ec62b2..b17e75c3d1f 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -4557,7 +4557,7 @@ impl Subscriber { let mut producer = crate::recv::Group::new(producer); let observing = producer.handle().clone(); - let mut ingest = GroupIngest::new(self, &group, timescale); + let mut ingest = GroupIngest::new(self.version, &group, timescale); ingest.expected_first = Some(start); let res = { kio::wait(|waiter| { @@ -4731,7 +4731,7 @@ enum Opened { /// How a subgroup stream ended cleanly. #[derive(Debug, PartialEq, Eq)] -enum Ended { +pub(crate) enum Ended { /// The stream finished, or carried an explicit end of group. Group, /// It carried an END_OF_TRACK after the group's last object, and whether that @@ -4958,13 +4958,13 @@ impl Subscriber { /// Pumps moq-transport subgroup objects from a reader into a group producer: /// the id delta, the extension headers (carrying the timestamp), the size, the /// status for empty objects, and the streamed payload. -struct GroupIngest { +pub(crate) struct GroupIngest { has_extensions: bool, timescale: Option, version: Version, prior_object: Option, /// The live tail must begin exactly after the filled head, even if it is quiet. - expected_first: Option, + pub(crate) expected_first: Option, properties: crate::model::object::Properties, /// The property list `properties` was read from, for a status object, which has no frame. raw: bytes::Bytes, @@ -4991,11 +4991,11 @@ enum IngestPhase { } impl GroupIngest { - fn new(subscriber: &Subscriber, group: &ietf::GroupHeader, timescale: Option) -> Self { + pub(crate) fn new(version: Version, group: &ietf::GroupHeader, timescale: Option) -> Self { Self { has_extensions: group.flags.has_extensions, timescale, - version: subscriber.version, + version, prior_object: None, expected_first: None, properties: Default::default(), @@ -5983,7 +5983,7 @@ impl GroupIngest { /// `Ready(Ok(_))` once the stream FINs on an object boundary, or an explicit /// end-of-group or end-of-track status arrives. The caller finishes or aborts the /// group; an object cut short mid-payload was already aborted here with the reason. - fn poll( + pub(crate) fn poll( &mut self, reader: &mut Reader, group: &mut crate::recv::Group, diff --git a/rs/moq-net/src/lite/mod.rs b/rs/moq-net/src/lite/mod.rs index 04b514cf239..651b06abca6 100644 --- a/rs/moq-net/src/lite/mod.rs +++ b/rs/moq-net/src/lite/mod.rs @@ -65,6 +65,9 @@ pub use stream::*; pub use subscribe::*; #[cfg(feature = "async")] use subscriber::*; +// The fuzz harness drives a group stream's ingest directly. +#[cfg(all(feature = "async", any(test, feature = "fuzz")))] +pub(crate) use subscriber::FrameIngest; pub use track::*; pub use tracks::*; pub use version::Version; diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 8dd35ae830f..c08a9f75910 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -1031,7 +1031,7 @@ impl GroupRecv { /// Pumps bare FRAME messages from a reader into a group producer: the wire /// format shared by GROUP streams and FETCH responses. -struct FrameIngest { +pub(crate) struct FrameIngest { /// A timescale decodes the lite-05 zigzag-delta timestamp prefix; none leaves frames /// untimed (pre-lite-05). framing: lite::Framing, @@ -1092,7 +1092,7 @@ enum IngestPhase { impl FrameIngest { /// `ietf` when the session negotiated the extension, so the stream starts with a Group State. - fn new(framing: lite::Framing, version: Version, ietf: bool, frame_start: u64) -> Self { + pub(crate) fn new(framing: lite::Framing, version: Version, ietf: bool, frame_start: u64) -> Self { Self { framing, prev_ts: 0, @@ -1129,7 +1129,7 @@ impl FrameIngest { /// `Ready(Ok(()))` once the stream FINs on a frame boundary. The caller /// finishes or aborts the group; a frame cut short mid-payload was already /// aborted here with the reason. - fn poll( + pub(crate) fn poll( &mut self, reader: &mut Reader, group: &mut crate::recv::Group, From 7da6d34e3794ec25214517350689cec58fd10602 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 30 Oct 2026 02:32:53 -0700 Subject: [PATCH 3/4] style: just fix Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/fuzz/README.md | 2 +- rs/moq-net/src/fuzz.rs | 7 ++++++- rs/moq-net/src/fuzz/ingest.rs | 27 ++++++++++++++++++++++----- rs/moq-net/src/fuzz/request.rs | 18 +++++++++++++++--- rs/moq-net/src/ietf/mod.rs | 4 ++-- 5 files changed, 46 insertions(+), 12 deletions(-) diff --git a/rs/moq-net/fuzz/README.md b/rs/moq-net/fuzz/README.md index 5b724d87ca6..49ea5674b52 100644 --- a/rs/moq-net/fuzz/README.md +++ b/rs/moq-net/fuzz/README.md @@ -65,7 +65,7 @@ Beyond "does not panic": - A request-stream message is its type, a two-byte length, and exactly that body; at most one GOAWAY migrates it. Its body decodes as each message its type names exactly when the control stream's length-prefixed reading accepts it, as the same message, and - the REQUEST_UPDATE readers stop where the framing did. + the REQUEST\_UPDATE readers stop where the framing did. `seeds/` also holds length prefixes at and one past each cap (a moq-lite message's size, a moq-transport length, a frame, a property or extension block), so every rejection path diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index d79ad6b60f4..09f5d007881 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -1740,7 +1740,12 @@ mod tests { #[test] fn stream_seeds_decode() { for seed in ingest::seeds().into_iter().chain(request::seeds()) { - assert!(target(seed.target)(&seed.data), "{} seed {:x?} does not decode", seed.target, seed.data); + assert!( + target(seed.target)(&seed.data), + "{} seed {:x?} does not decode", + seed.target, + seed.data + ); } for seed in ingest::limit_seeds() { assert!(!ingest(&seed.data), "ingest seed {:x?} decodes past its cap", seed.data); diff --git a/rs/moq-net/src/fuzz/ingest.rs b/rs/moq-net/src/fuzz/ingest.rs index a3f8ec70453..8cd4d3925d7 100644 --- a/rs/moq-net/src/fuzz/ingest.rs +++ b/rs/moq-net/src/fuzz/ingest.rs @@ -101,7 +101,11 @@ pub fn ingest(data: &[u8]) -> bool { _ => { let version = super::IETF_VERSIONS[(*selector as usize / 2) % super::IETF_VERSIONS.len()]; // A moq-transport track was always published over moq-transport. - (Session::Ietf { version }, track_info(timed, compress, Some(version)), rest) + ( + Session::Ietf { version }, + track_info(timed, compress, Some(version)), + rest, + ) } }; @@ -130,9 +134,16 @@ fn check(session: Session, info: &track::Info, body: &[u8], ingested: &Ingested) info.timescale.is_some(), "frame {index} does not match its track's timeline" ); - assert_eq!(frame.inflated.is_some(), info.compress, "frame {index} lost its inflated size"); + assert_eq!( + frame.inflated.is_some(), + info.compress, + "frame {index} lost its inflated size" + ); let id = frame.fields.as_ref().map_or(index as u64, |(id, _)| *id); - assert!(ids.is_none_or(|prior| id > prior), "Object ID {id} does not follow {ids:?}"); + assert!( + ids.is_none_or(|prior| id > prior), + "Object ID {id} does not follow {ids:?}" + ); ids = Some(id); payload += frame.payload.len(); } @@ -217,7 +228,11 @@ fn run(session: Session, info: &track::Info, body: &[u8], chunk: usize) -> Optio Session::Ietf { version } => { let (header, used) = ietf::GroupHeader::decode_slice(body, version).ok()?; let mut reader = reader(Bytes::copy_from_slice(&body[used..]), chunk, version); - let producer = track.create_group(group::Info { sequence: header.group_id }).ok()?; + let producer = track + .create_group(group::Info { + sequence: header.group_id, + }) + .ok()?; producer .set_wire(group::Wire { sparse: false, @@ -413,7 +428,9 @@ pub(super) fn limit_seeds() -> Vec { // The caps on moq-lite-07: a frame's size, and a record's property block. let varint = |value: u64| { let mut out = Vec::new(); - coding::Encoder::new(&mut out, lite::Version::Lite07.into()).varint(value).unwrap(); + coding::Encoder::new(&mut out, lite::Version::Lite07.into()) + .varint(value) + .unwrap(); out }; for size in super::limits(&[group::MAX_CACHE_BYTES]) { diff --git a/rs/moq-net/src/fuzz/request.rs b/rs/moq-net/src/fuzz/request.rs index 28145604acb..cec97c4f209 100644 --- a/rs/moq-net/src/fuzz/request.rs +++ b/rs/moq-net/src/fuzz/request.rs @@ -56,12 +56,24 @@ fn framing(consumed: &[u8], framed: &ietf::Framed, version: ietf::Version) { let ietf::Framed(id, ietf::Body(body)) = framed; let form: Form = version.into(); let mut decoder = Decoder::new(consumed, form); - assert_eq!(decoder.varint().ok(), Some(*id), "message type {id:#x} was not read from the stream"); + assert_eq!( + decoder.varint().ok(), + Some(*id), + "message type {id:#x} was not read from the stream" + ); let id_len = decoder.position(); let length = u16::from_be_bytes([consumed[id_len], consumed[id_len + 1]]) as usize; assert_eq!(length, body.len(), "message {id:#x} body is not its declared length"); - assert_eq!(consumed.len(), id_len + 2 + length, "message {id:#x} read past its body"); - assert_eq!(&consumed[id_len + 2..], &body[..], "message {id:#x} body is not the stream's bytes"); + assert_eq!( + consumed.len(), + id_len + 2 + length, + "message {id:#x} read past its body" + ); + assert_eq!( + &consumed[id_len + 2..], + &body[..], + "message {id:#x} body is not the stream's bytes" + ); // Not a GOAWAY, or the stream would have migrated instead. assert_ne!(*id, ietf::GoAway::ID, "a GOAWAY was read as a request message"); diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index b01e9e9e1e4..9776a3c055d 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -81,13 +81,13 @@ pub use subscribe_tracks::*; #[cfg(feature = "async")] use subscriber::*; // The fuzz harness drives a subgroup stream's ingest directly. +pub use track::*; +pub use version::Version; #[cfg(all(feature = "async", any(test, feature = "fuzz")))] pub(crate) use { request_stream::{Framed, NamespaceUpdate, Next, Update}, subscriber::GroupIngest, }; -pub use track::*; -pub use version::Version; #[cfg(test)] mod tests { From 758e2b745e809a5f597fc00755e5ac393956b400 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 30 Oct 2026 03:17:21 -0700 Subject: [PATCH 4/4] test(net): bound ingest accounting by the subgroup's extension flag An untimed, uncompressed moq-transport track still carries extension blocks when its subgroup header says so, and their length is not kept, so the upper bound only holds without them. Commit the input that exceeded it. Co-Authored-By: Claude Opus 5.5 --- .../ingest/draft22-untimed-extension-block | Bin 0 -> 75 bytes rs/moq-net/src/fuzz/ingest.rs | 19 +++++++++--------- 2 files changed, 9 insertions(+), 10 deletions(-) create mode 100644 rs/moq-net/fuzz/regressions/ingest/draft22-untimed-extension-block diff --git a/rs/moq-net/fuzz/regressions/ingest/draft22-untimed-extension-block b/rs/moq-net/fuzz/regressions/ingest/draft22-untimed-extension-block new file mode 100644 index 0000000000000000000000000000000000000000..0cf89a4a7870e659d0904d0e03b0db302109fbe8 GIT binary patch literal 75 ZcmZQ(V2)&DU|_IQv { // Timing, inflated size, and size; a record adds its code and block length. @@ -176,13 +177,9 @@ fn check(session: Session, info: &track::Info, body: &[u8], ingested: &Ingested) } // The ID delta and the size, then an empty object's status. A status object that // ends the group, or the track, repeats them. - Session::Ietf { .. } => match info.compress || info.timescale.is_some() { - true => None, - false => Some((3, 3 * MAX_VARINT, 0)), - }, + Session::Ietf { .. } => (!extensions).then_some((3, 3 * MAX_VARINT, 0)), }; if let Some((fields, trailer, blocks)) = fields { - let header = header_len(session, body); let most = header + payload + blocks + frames * fields * MAX_VARINT + trailer; assert!( body.len() <= most, @@ -192,11 +189,13 @@ fn check(session: Session, info: &track::Info, body: &[u8], ingested: &Ingested) } } -/// The subgroup header's length, which a moq-transport run decodes before its objects. -fn header_len(session: Session, body: &[u8]) -> usize { +/// The subgroup header's length, which a moq-transport run decodes before its objects, +/// and whether its objects carry extension blocks. +fn subgroup_header(session: Session, body: &[u8]) -> (usize, bool) { match session { - Session::Lite { .. } => 0, - Session::Ietf { version } => ietf::GroupHeader::decode_slice(body, version).map_or(0, |(_, used)| used), + Session::Lite { .. } => (0, false), + Session::Ietf { version } => ietf::GroupHeader::decode_slice(body, version) + .map_or((0, false), |(header, used)| (used, header.flags.has_extensions)), } }