moq-net: cap frame size at 16 MiB on the receive path (#1521)

Co-authored-by: Claude <noreply@anthropic.com>

Author: kixelated

rs/moq-net/src/error.rs

@@ -86,6 +86,10 @@ pub enum Error {
 	#[error("cache full")]
 	CacheFull,
 
+	/// A frame declared a payload size larger than the receiver accepts.
+	#[error("frame too large")]
+	FrameTooLarge,
+
 	/// A remote error received via a stream/session reset code.
 	#[error("remote error: code={0}")]
 	Remote(u32),
@@ -118,6 +122,7 @@ impl Error {
 			Self::Dropped => 24,
 			Self::Closed => 25,
 			Self::CacheFull => 26,
+			Self::FrameTooLarge => 27,
 			Self::App(app) => *app as u32 + 64,
 			Self::Remote(code) => *code,
 		}

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

@@ -1,8 +1,8 @@
 use std::collections::{HashMap, hash_map::Entry};
 
 use crate::{
-	Broadcast, BroadcastDynamic, Error, Frame, FrameProducer, Group, GroupProducer, OriginProducer, Path, PathOwned,
-	Track, TrackProducer,
+	Broadcast, BroadcastDynamic, Error, Frame, FrameProducer, Group, GroupProducer, MAX_FRAME_SIZE, OriginProducer,
+	Path, PathOwned, Track, TrackProducer,
 	coding::{Reader, Stream},
 	ietf::{self, Control, FilterType, GroupOrder, RequestId},
 	model::BroadcastProducer,
@@ -744,6 +744,9 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 					return Err(Error::Unsupported);
 				}
 			} else {
+				if size > MAX_FRAME_SIZE {
+					return Err(Error::FrameTooLarge);
+				}
 				let mut frame = producer.create_frame(Frame { size })?;
 
 				if let Err(err) = self.run_frame(stream, frame.clone()).await {

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

@@ -7,7 +7,7 @@ use futures::{StreamExt, stream::FuturesUnordered};
 
 use crate::{
 	AsPath, BandwidthProducer, Broadcast, BroadcastDynamic, Error, Frame, FrameProducer, Group, GroupProducer,
-	OriginProducer, Path, PathOwned, StatsHandle, SubscriberStats, SubscriberTrack, TrackProducer,
+	MAX_FRAME_SIZE, OriginProducer, Path, PathOwned, StatsHandle, SubscriberStats, SubscriberTrack, TrackProducer,
 	coding::{Reader, Stream},
 	lite,
 	model::BroadcastProducer,
@@ -457,6 +457,9 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 		track_stats: Arc<SubscriberTrack>,
 	) -> Result<(), Error> {
 		while let Some(size) = stream.decode_maybe::<u64>().await? {
+			if size > MAX_FRAME_SIZE {
+				return Err(Error::FrameTooLarge);
+			}
 			let mut frame = group.create_frame(Frame { size })?;
 			track_stats.frame();
 

rs/moq-net/src/model/frame.rs

@@ -7,6 +7,17 @@ use bytes::{BufMut, Bytes};
 
 use crate::{Error, Result};
 
+/// Maximum payload size accepted for a single frame on the wire.
+///
+/// The receive path preallocates a buffer from the declared frame size, so an
+/// untrusted peer could otherwise request a multi-gigabyte allocation with a
+/// single varint. Subscribers reject frames whose declared size exceeds this.
+///
+// TODO enforce this in [Frame::produce] / [FrameProducer::new] so the limit is
+// guaranteed for every caller, not just the wire decode paths. Blocked on
+// making the constructor fallible (returning [Result]), which is an API break.
+pub(crate) const MAX_FRAME_SIZE: u64 = 16 * 1024 * 1024;
+
 /// A chunk of data with an upfront size.
 ///
 /// Note that this is just the header.