diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 6b9d9e56ed2..284e731dde9 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -691,7 +691,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 0c1b094de59..7487872c3a7 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..49ea5674b52 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/ingest/draft22-untimed-extension-block b/rs/moq-net/fuzz/regressions/ingest/draft22-untimed-extension-block new file mode 100644 index 00000000000..0cf89a4a787 Binary files /dev/null and b/rs/moq-net/fuzz/regressions/ingest/draft22-untimed-extension-block differ 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 00000000000..73a61f0d255 Binary files /dev/null and b/rs/moq-net/fuzz/regressions/request/draft22-next-short-type differ diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 45c1f9efea5..09f5d007881 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,24 @@ 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..f44e7a4aabd --- /dev/null +++ b/rs/moq-net/src/fuzz/ingest.rs @@ -0,0 +1,464 @@ +//! 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 to a subgroup without them. + let Ok(_) = &ingested.end else { return }; + let (header, extensions) = subgroup_header(session, body); + 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 { .. } => (!extensions).then_some((3, 3 * MAX_VARINT, 0)), + }; + if let Some((fields, trailer, blocks)) = fields { + 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, +/// and whether its objects carry extension blocks. +fn subgroup_header(session: Session, body: &[u8]) -> (usize, bool) { + match session { + Session::Lite { .. } => (0, false), + Session::Ietf { version } => ietf::GroupHeader::decode_slice(body, version) + .map_or((0, false), |(header, used)| (used, header.flags.has_extensions)), + } +} + +/// 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..cec97c4f209 --- /dev/null +++ b/rs/moq-net/src/fuzz/request.rs @@ -0,0 +1,245 @@ +//! 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..9776a3c055d 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -80,8 +80,14 @@ pub use subscribe_namespace::*; 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, +}; #[cfg(test)] mod tests { 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 f6c48b1bcf3..6180c4b6b91 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -4603,7 +4603,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| { @@ -4777,7 +4777,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 @@ -5004,13 +5004,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, @@ -5037,11 +5037,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(), @@ -6030,7 +6030,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,