fix(net): refuse malformed complete announce bodies

quest/m1/cut-audit/README.md

@@ -59,7 +59,6 @@ leaving JS-only redesigns of `js/net` to [rs2ts](/quest/m2/rs2ts/README.md).
 - [A route too large to re-encode is skipped](/quest/m1/cut-audit/announce-reencode.md) - one oversized route never resets a downstream announce stream
 - [A reserved group survives a route returning](/quest/m1/cut-audit/reserved-route-return.md) - a route flip A to B to A no longer panics a peeked reader
 - [JS tracks end clean only on a deliberate finish](/quest/m1/cut-audit/js-close-cause.md) - local close, violations, and unanswered requests end with errors
-- [A malformed lite message closes the session](/quest/m1/cut-audit/announce-decode-complete.md) - a bad ANNOUNCE body no longer jams its stream
 - [Withdrawing a moq-transport FETCH or TRACK_STATUS ends it](/quest/m1/cut-audit/ietf-withdraw-resolve.md) - withdrawal during resolution releases the request
 - [JS moq-transport sessions close on peer violations](/quest/m1/cut-audit/js-ietf-violations.md) - right close codes, no unhandled rejection, bounded reflected namespaces
 - [A media hole in the player keeps the timeline](/quest/m1/cut-audit/watch-audio-hole.md) - a skipped group re-primes decoders without flushing audio or re-anchoring sync

rs/moq-net/fuzz/README.md

@@ -40,6 +40,8 @@ symbol stripping. Prefix direct calls with
 
 Beyond "does not panic":
 
+- A short decode requires more outer input. A field running short inside a complete
+  length-prefixed body is malformed and cannot ask for bytes from the next message.
 - Whatever we encode, we decode back, consuming every byte. The peer's decoder is this
   same code, so emitting something we would reject is a bug on the wire.
 - Encoding is stable: encode, decode, encode again yields identical bytes. Skipped
@@ -53,8 +55,9 @@ Beyond "does not panic":
 - `Path::relative` inverts `Path::resolve`, and never produces a reference that walks
   above the root.
 
-A decode that fails is not a finding, and neither is an encode that fails: a value can
-decode at a version that cannot express it again.
+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
+express it again.
 
 ## When something crashes
 

rs/moq-net/fuzz/regressions/lite/complete-announce-lite05

@@ -0,0 +1 @@
+

rs/moq-net/src/fuzz.rs

@@ -105,7 +105,7 @@ const IETF_VERSIONS: &[ietf::Version] = &[
 ];
 
 /// How many types [`lite_wire`] dispatches over.
-const LITE_KINDS: u8 = 22;
+const LITE_KINDS: u8 = 28;
 
 /// How many types [`ietf_wire`] dispatches over.
 const IETF_KINDS: u8 = 44;
@@ -131,8 +131,14 @@ where
 	T: Decode<V> + Encode<V>,
 	V: Copy + Into<Form>,
 {
-	let Ok((decoded, _)) = T::decode_slice(data, version) else {
-		return false;
+	let mut reader = Decoder::new(data, version.into());
+	let decoded = match T::decode(&mut reader, version) {
+		Ok(decoded) => decoded,
+		Err(coding::DecodeError::Short) => {
+			assert!(reader.need() > data.len(), "short read inside a complete body");
+			return false;
+		}
+		Err(_) => return false,
 	};
 
 	let Ok(first) = decoded.encode_bytes(version) else {
@@ -224,6 +230,12 @@ pub fn lite_wire(data: &[u8]) -> bool {
 		19 => datagram(rest, version),
 		20 => roundtrip::<lite::Parameters, _>(rest, version),
 		21 => roundtrip::<lite::ietf::TrackProperties, _>(rest, version),
+		22 => roundtrip::<lite::Tracks<'static>, _>(rest, version),
+		23 => roundtrip::<lite::TracksEvent<'static>, _>(rest, version),
+		24 => roundtrip::<lite::Auth, _>(rest, version),
+		25 => roundtrip::<lite::AuthOk, _>(rest, version),
+		26 => roundtrip::<lite::AuthError, _>(rest, version),
+		27 => roundtrip::<lite::AuthReply, _>(rest, version),
 		_ => unreachable!("kind is taken modulo LITE_KINDS"),
 	}
 }
@@ -1074,6 +1086,17 @@ pub fn seeds() -> Vec<Seed> {
 		}
 	}
 
+	// A typed empty TRACKS_SYNCED and AUTH_OK reply need nonuniform envelope bytes.
+	for (kind, body) in [(23, &[4, 0][..]), (27, &[0, 3, 0, 0, 0][..])] {
+		let mut data = vec![(LITE_VERSIONS.len() - 1) as u8, kind];
+		data.extend_from_slice(body);
+		seeds.push(Seed {
+			target: "lite",
+			kind,
+			data,
+		});
+	}
+
 	for version in 0..IETF_VERSIONS.len() as u8 {
 		for kind in 0..IETF_KINDS {
 			for fill in FILLS {

rs/moq-net/src/lite/announce.rs

@@ -339,33 +339,36 @@ impl Decode<Version> for AnnounceBroadcast<'_> {
 			let typ = buf.varint()?;
 			let size = decode_size(buf, MAX_MESSAGE_SIZE)?;
 			let mut body = buf.sub(size)?;
-			let msg = match typ {
-				ANNOUNCE_START => Self::Active {
-					suffix: PathRef::decode(&mut body, version)?,
-					epoch: super::epoch::decode_epoch(&mut body, version)?,
-					hops: HopsRef::decode(&mut body, version)?,
-					cost: Cost::decode(&mut body, version)?,
-				},
-				SYNCED if version.has_synced() => Self::Synced,
-				ANNOUNCE_END => Self::EndedId { id: body.varint()? },
-				ANNOUNCE_UPDATE => Self::Update {
-					id: body.varint()?,
-					hops: HopsRef::decode(&mut body, version)?,
-					cost: Cost::decode(&mut body, version)?,
-				},
-				ANNOUNCE_RESTART if version.has_announce_restart() => Self::Restart {
-					id: body.varint()?,
-					epoch: super::epoch::decode_epoch(&mut body, version)?,
-					hops: HopsRef::decode(&mut body, version)?,
-					cost: Cost::decode(&mut body, version)?,
-				},
-				// Unknown types are skipped by length so an earlier Lite06 build
-				// negotiating the same ALPN does not kill the announce stream.
-				_ => {
-					body.rest();
-					Self::Skipped
-				}
-			};
+			let msg = (|| {
+				Ok(match typ {
+					ANNOUNCE_START => Self::Active {
+						suffix: PathRef::decode(&mut body, version)?,
+						epoch: super::epoch::decode_epoch(&mut body, version)?,
+						hops: HopsRef::decode(&mut body, version)?,
+						cost: Cost::decode(&mut body, version)?,
+					},
+					SYNCED if version.has_synced() => Self::Synced,
+					ANNOUNCE_END => Self::EndedId { id: body.varint()? },
+					ANNOUNCE_UPDATE => Self::Update {
+						id: body.varint()?,
+						hops: HopsRef::decode(&mut body, version)?,
+						cost: Cost::decode(&mut body, version)?,
+					},
+					ANNOUNCE_RESTART if version.has_announce_restart() => Self::Restart {
+						id: body.varint()?,
+						epoch: super::epoch::decode_epoch(&mut body, version)?,
+						hops: HopsRef::decode(&mut body, version)?,
+						cost: Cost::decode(&mut body, version)?,
+					},
+					// Unknown types are skipped by length so an earlier Lite06 build
+					// negotiating the same ALPN does not kill the announce stream.
+					_ => {
+						body.rest();
+						Self::Skipped
+					}
+				})
+			})()
+			.map_err(DecodeError::complete)?;
 			if !body.is_empty() {
 				return Err(DecodeError::Long);
 			}
@@ -375,7 +378,7 @@ impl Decode<Version> for AnnounceBroadcast<'_> {
 		// Older versions: a single size-prefixed ANNOUNCE_BROADCAST with an inner status.
 		let size = decode_size(buf, MAX_MESSAGE_SIZE)?;
 		let mut body = buf.sub(size)?;
-		let msg = Self::decode_legacy(&mut body, version)?;
+		let msg = Self::decode_legacy(&mut body, version).map_err(DecodeError::complete)?;
 		if !body.is_empty() {
 			return Err(DecodeError::Long);
 		}

rs/moq-net/src/lite/subscriber.rs

@@ -5930,6 +5930,47 @@ mod tests {
 		assert!(cursor.is_synced());
 	}
 
+	/// A complete body cannot be repaired by bytes from the next announcement.
+	/// Keep the peer stream open: FIN must not be what discovers the malformed body.
+	#[moq_clock::test]
+	async fn malformed_complete_announce_closes_before_more_bytes() {
+		for version in [Version::Lite05, Version::Lite06, Version::Lite07] {
+			let mut script = announce_script(version, &[&announce_ok(version, 1)]);
+			script.extend_from_slice(match version {
+				Version::Lite05 => &[1, 1],
+				_ => &[0, 1, 5],
+			});
+			let session = crate::lite::test_transport::ScriptedSession::new(script);
+			let log = session.log.clone();
+			let (mut prefix, _consumer) = root_prefix(version, session);
+			let result = prefix.poll(&kio::Waiter::noop());
+			assert!(
+				matches!(result, Poll::Ready(Err(Error::ProtocolViolation))),
+				"{version:?}: {result:?}"
+			);
+			assert_eq!(log.closes()[0].0, crate::SessionError::ProtocolViolation.to_code());
+		}
+	}
+
+	/// Only an incomplete outer frame waits, then resumes on its missing byte.
+	#[moq_clock::test]
+	async fn an_incomplete_announce_waits_for_its_body() {
+		for version in [Version::Lite05, Version::Lite06, Version::Lite07] {
+			let mut script = announce_script(version, &[&announce_ok(version, 1), &announce_start(version, "room")]);
+			let last = script.pop().unwrap();
+			let session = crate::lite::test_transport::ScriptedSession::new(script);
+			let log = session.log.clone();
+			let (mut prefix, consumer) = root_prefix(version, session.clone());
+			let mut cursor = consumer.announced();
+			assert!(prefix.poll(&kio::Waiter::noop()).is_pending());
+			assert!(log.closes().is_empty());
+			session.push(&[last]);
+			assert!(prefix.poll(&kio::Waiter::noop()).is_pending());
+			cursor.assert_next_active("room");
+			assert!(log.closes().is_empty());
+		}
+	}
+
 	#[moq_clock::test]
 	async fn a_duplicate_synced_is_refused() {
 		const VERSION: Version = Version::Lite07;