From c7d37a8dc047259fa35084946e1790487387ef9e Mon Sep 17 00:00:00 2001 From: yexiyue Date: Fri, 24 Jul 2026 23:18:59 +0800 Subject: [PATCH 1/5] feat(webrtc): negotiate data channel message limits --- misc/webrtc-utils/src/lib.rs | 4 +- misc/webrtc-utils/src/noise.rs | 129 +++++++++++++++++- misc/webrtc-utils/src/sdp.rs | 4 +- misc/webrtc-utils/src/stream.rs | 94 +++++++++++-- misc/webrtc-utils/src/stream/framed_dc.rs | 12 +- transports/webrtc-websys/src/connection.rs | 12 +- transports/webrtc-websys/src/stream.rs | 9 +- .../src/stream/poll_data_channel.rs | 12 +- transports/webrtc-websys/src/transport.rs | 19 ++- transports/webrtc-websys/src/upgrade.rs | 27 +++- transports/webrtc/src/tokio/connection.rs | 9 +- transports/webrtc/src/tokio/sdp.rs | 2 +- transports/webrtc/src/tokio/stream.rs | 13 +- transports/webrtc/src/tokio/transport.rs | 12 ++ transports/webrtc/src/tokio/upgrade.rs | 22 ++- 15 files changed, 316 insertions(+), 64 deletions(-) diff --git a/misc/webrtc-utils/src/lib.rs b/misc/webrtc-utils/src/lib.rs index 330cebee3b1..43c3cbbbc76 100644 --- a/misc/webrtc-utils/src/lib.rs +++ b/misc/webrtc-utils/src/lib.rs @@ -11,5 +11,7 @@ mod stream; mod transport; pub use fingerprint::{Fingerprint, SHA256}; -pub use stream::{DropListener, MAX_MSG_LEN, Stream}; +pub use stream::{ + DEFAULT_MAX_MESSAGE_SIZE, DropListener, MAX_MSG_LEN, MIN_MESSAGE_SIZE, Stream, StreamConfig, +}; pub use transport::parse_webrtc_dial_addr; diff --git a/misc/webrtc-utils/src/noise.rs b/misc/webrtc-utils/src/noise.rs index bcb5102120f..249e1da70e9 100644 --- a/misc/webrtc-utils/src/noise.rs +++ b/misc/webrtc-utils/src/noise.rs @@ -18,7 +18,9 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. -use futures::{AsyncRead, AsyncWrite, AsyncWriteExt}; +use std::num::NonZeroUsize; + +use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use libp2p_core::{ UpgradeInfo, upgrade::{InboundConnectionUpgrade, OutboundConnectionUpgrade}, @@ -28,7 +30,10 @@ use libp2p_identity::PeerId; use libp2p_noise as noise; pub use noise::Error; -use crate::fingerprint::Fingerprint; +use crate::{ + fingerprint::Fingerprint, + stream::{DEFAULT_MAX_MESSAGE_SIZE, MIN_MESSAGE_SIZE, StreamConfig}, +}; pub async fn inbound( id_keys: identity::Keypair, @@ -36,6 +41,28 @@ pub async fn inbound( client_fingerprint: Fingerprint, server_fingerprint: Fingerprint, ) -> Result +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + inbound_with_message_size( + id_keys, + stream, + client_fingerprint, + server_fingerprint, + StreamConfig::default(), + ) + .await + .map(|(peer_id, _)| peer_id) +} + +/// Authenticates the connection and negotiates its encoded message-size limit. +pub async fn inbound_with_message_size( + id_keys: identity::Keypair, + stream: T, + client_fingerprint: Fingerprint, + server_fingerprint: Fingerprint, + stream_config: StreamConfig, +) -> Result<(PeerId, StreamConfig), Error> where T: AsyncRead + AsyncWrite + Unpin + Send + 'static, { @@ -47,9 +74,10 @@ where // send application data 0.5 RTT earlier. let (peer_id, mut channel) = noise.upgrade_outbound(stream, info).await?; - channel.close().await?; + let stream_config = negotiate_message_size(&mut channel, stream_config).await; + let _ = channel.close().await; - Ok(peer_id) + Ok((peer_id, stream_config)) } pub async fn outbound( @@ -58,6 +86,28 @@ pub async fn outbound( server_fingerprint: Fingerprint, client_fingerprint: Fingerprint, ) -> Result +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + outbound_with_message_size( + id_keys, + stream, + server_fingerprint, + client_fingerprint, + StreamConfig::default(), + ) + .await + .map(|(peer_id, _)| peer_id) +} + +/// Authenticates the connection and negotiates its encoded message-size limit. +pub async fn outbound_with_message_size( + id_keys: identity::Keypair, + stream: T, + server_fingerprint: Fingerprint, + client_fingerprint: Fingerprint, + stream_config: StreamConfig, +) -> Result<(PeerId, StreamConfig), Error> where T: AsyncRead + AsyncWrite + Unpin + Send + 'static, { @@ -69,9 +119,52 @@ where // send application data 0.5 RTT earlier. let (peer_id, mut channel) = noise.upgrade_inbound(stream, info).await?; - channel.close().await?; + let stream_config = negotiate_message_size(&mut channel, stream_config).await; + let _ = channel.close().await; - Ok(peer_id) + Ok((peer_id, stream_config)) +} + +/// Exchanges the local limit after the authenticated Noise handshake. +/// +/// Older peers close the reserved data channel immediately after Noise. Treat that as their +/// historical 16 KiB limit, preserving compatibility while newer peers use the smaller limit. +async fn negotiate_message_size(channel: &mut T, local: StreamConfig) -> StreamConfig +where + T: AsyncRead + AsyncWrite + Unpin, +{ + let fallback = local.limited_by(DEFAULT_MAX_MESSAGE_SIZE); + let advertised = local.max_message_size() as u64; + + if channel.write_all(&advertised.to_be_bytes()).await.is_err() || channel.flush().await.is_err() + { + return fallback; + } + + let mut remote = [0; std::mem::size_of::()]; + if channel.read_exact(&mut remote).await.is_err() { + return fallback; + } + + effective_message_size(local, Some(u64::from_be_bytes(remote))) +} + +fn effective_message_size(local: StreamConfig, remote: Option) -> StreamConfig { + let fallback = local.limited_by(DEFAULT_MAX_MESSAGE_SIZE); + let Some(remote) = remote else { + return fallback; + }; + let Ok(remote) = usize::try_from(remote) else { + return fallback; + }; + let Some(remote) = NonZeroUsize::new(remote) else { + return fallback; + }; + if remote < MIN_MESSAGE_SIZE { + return fallback; + } + + local.limited_by(remote) } pub(crate) fn noise_prologue( @@ -115,4 +208,28 @@ mod tests { "6c69627032702d7765627274632d6e6f6973653a122030fc9f469c207419dfdd0aab5f27a86c973c94e40548db9375cca2e915973b9912203e79af40d6059617a0d83b83a52ce73b0c1f37a72c6043ad2969e2351bdca870" ); } + + #[test] + fn message_size_negotiation_uses_the_smaller_valid_limit() { + let local = StreamConfig::new(NonZeroUsize::new(16 * 1024).unwrap()); + + assert_eq!( + effective_message_size(local, Some(8 * 1024)).max_message_size(), + 8 * 1024 + ); + } + + #[test] + fn message_size_negotiation_falls_back_for_legacy_or_invalid_peers() { + let local = StreamConfig::new(NonZeroUsize::new(8 * 1024).unwrap()); + + assert_eq!( + effective_message_size(local, None).max_message_size(), + 8 * 1024 + ); + assert_eq!( + effective_message_size(local, Some(0)).max_message_size(), + 8 * 1024 + ); + } } diff --git a/misc/webrtc-utils/src/sdp.rs b/misc/webrtc-utils/src/sdp.rs index 2be0ed76f29..5eb7dac5e42 100644 --- a/misc/webrtc-utils/src/sdp.rs +++ b/misc/webrtc-utils/src/sdp.rs @@ -91,7 +91,7 @@ a=ice-pwd:{pwd} a=fingerprint:{fingerprint_algorithm} {fingerprint_value} a=setup:passive a=sctp-port:5000 -a=max-message-size:16384 +a=max-message-size:{max_message_size} a=candidate:1467250027 1 UDP 1467250027 {target_ip} {target_port} typ host a=end-of-candidates "; @@ -114,6 +114,7 @@ struct DescriptionContext { pub(crate) fingerprint_value: String, pub(crate) ufrag: String, pub(crate) pwd: String, + pub(crate) max_message_size: usize, } /// Renders a [`TinyTemplate`] description using the provided arguments. @@ -141,6 +142,7 @@ pub fn render_description( // NOTE: ufrag is equal to pwd. ufrag: ufrag.to_owned(), pwd: ufrag.to_owned(), + max_message_size: 16 * 1024, }; tt.render("description", &context).unwrap() } diff --git a/misc/webrtc-utils/src/stream.rs b/misc/webrtc-utils/src/stream.rs index 96ad82564c1..fad21372b48 100644 --- a/misc/webrtc-utils/src/stream.rs +++ b/misc/webrtc-utils/src/stream.rs @@ -21,6 +21,7 @@ use std::{ io, + num::NonZeroUsize, pin::Pin, task::{Context, Poll}, }; @@ -41,18 +42,67 @@ mod drop_listener; mod framed_dc; mod state; -/// Maximum length of a message. -/// -/// "As long as message interleaving is not supported, the sender SHOULD limit the maximum message -/// size to 16 KB to avoid monopolization." -/// Source: -pub const MAX_MSG_LEN: usize = 16 * 1024; /// Length of varint, in bytes. const VARINT_LEN: usize = 2; /// Overhead of the protobuf encoding, in bytes. const PROTO_OVERHEAD: usize = 5; -/// Maximum length of data, in bytes. -const MAX_DATA_LEN: usize = MAX_MSG_LEN - VARINT_LEN - PROTO_OVERHEAD; + +/// Default maximum length of a WebRTC data-channel message. +/// +/// "As long as message interleaving is not supported, the sender SHOULD limit the maximum message +/// size to 16 KB to avoid monopolization." +/// Source: +pub const DEFAULT_MAX_MESSAGE_SIZE: NonZeroUsize = + NonZeroUsize::new(16 * 1024).expect("constant is non-zero"); +/// Smallest encoded message that can carry one byte of application data. +pub const MIN_MESSAGE_SIZE: NonZeroUsize = + NonZeroUsize::new(VARINT_LEN + PROTO_OVERHEAD + 1).expect("constant is non-zero"); +/// Backwards-compatible default message-size value. +pub const MAX_MSG_LEN: usize = DEFAULT_MAX_MESSAGE_SIZE.get(); + +/// Per-connection WebRTC message-size policy. +/// +/// The same value must be used by the framing codec, its write high-water mark and the +/// transport's data-channel backpressure accounting. Keeping it in one value prevents those +/// layers from silently disagreeing about a valid frame size. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct StreamConfig { + max_message_size: NonZeroUsize, +} + +impl StreamConfig { + /// Creates a stream configuration with the provided maximum encoded message size. + pub const fn new(max_message_size: NonZeroUsize) -> Self { + assert!(max_message_size.get() >= MIN_MESSAGE_SIZE.get()); + Self { max_message_size } + } + + /// Returns the smaller of two independently advertised limits. + pub const fn limited_by(self, remote_max_message_size: NonZeroUsize) -> Self { + Self::new( + if self.max_message_size.get() <= remote_max_message_size.get() { + self.max_message_size + } else { + remote_max_message_size + }, + ) + } + + /// Returns the maximum encoded data-channel message size. + pub const fn max_message_size(self) -> usize { + self.max_message_size.get() + } + + pub(crate) const fn max_data_size(self) -> usize { + self.max_message_size.get() - VARINT_LEN - PROTO_OVERHEAD + } +} + +impl Default for StreamConfig { + fn default() -> Self { + Self::new(DEFAULT_MAX_MESSAGE_SIZE) + } +} pub use drop_listener::DropListener; /// A stream backed by a WebRTC data channel. @@ -63,6 +113,7 @@ pub struct Stream { io: FramedDc, state: State, read_buffer: Bytes, + config: StreamConfig, /// Dropping this will close the oneshot and notify the receiver by emitting `Canceled`. drop_notifier: Option>, } @@ -74,15 +125,21 @@ where /// Returns a new [`Stream`] and a [`DropListener`], /// which will notify the receiver when/if the stream is dropped. pub fn new(data_channel: T) -> (Self, DropListener) { + Self::with_config(data_channel, StreamConfig::default()) + } + + /// Returns a new stream and drop listener using the supplied message-size policy. + pub fn with_config(data_channel: T, config: StreamConfig) -> (Self, DropListener) { let (sender, receiver) = oneshot::channel(); let stream = Self { - io: framed_dc::new(data_channel.clone()), + io: framed_dc::new(data_channel.clone(), config), state: State::Open, read_buffer: Bytes::default(), + config, drop_notifier: Some(sender), }; - let listener = DropListener::new(framed_dc::new(data_channel), receiver); + let listener = DropListener::new(framed_dc::new(data_channel, config), receiver); (stream, listener) } @@ -205,7 +262,7 @@ where ready!(self.io.poll_ready_unpin(cx))?; - let n = usize::min(buf.len(), MAX_DATA_LEN); + let n = usize::min(buf.len(), self.config.max_data_size()); Pin::new(&mut self.io).start_send(Message { flag: None, @@ -284,21 +341,30 @@ mod tests { #[test] fn max_data_len() { + let config = StreamConfig::default(); // Largest possible message. - let message = [0; MAX_DATA_LEN]; + let message = vec![0; config.max_data_size()]; let protobuf = Message { flag: Some(Flag::Fin as i32), message: Some(message.to_vec()), }; - let mut codec = codec(); + let mut codec = codec(config); let mut dst = BytesMut::new(); codec.encode(protobuf, &mut dst).unwrap(); // Ensure the varint prefixed and protobuf encoded largest message is no longer than the // maximum limit specified in the libp2p WebRTC specification. - assert_eq!(dst.len(), MAX_MSG_LEN); + assert_eq!(dst.len(), config.max_message_size()); + } + + #[test] + fn effective_limit_is_the_smaller_advertised_limit() { + let local = StreamConfig::new(NonZeroUsize::new(16 * 1024).unwrap()); + let remote = NonZeroUsize::new(8 * 1024).unwrap(); + + assert_eq!(local.limited_by(remote).max_message_size(), remote.get()); } } diff --git a/misc/webrtc-utils/src/stream/framed_dc.rs b/misc/webrtc-utils/src/stream/framed_dc.rs index f0bbda84ca8..5e040416623 100644 --- a/misc/webrtc-utils/src/stream/framed_dc.rs +++ b/misc/webrtc-utils/src/stream/framed_dc.rs @@ -23,21 +23,21 @@ use futures::{AsyncRead, AsyncWrite}; use crate::{ proto::Message, - stream::{MAX_DATA_LEN, MAX_MSG_LEN, VARINT_LEN}, + stream::{StreamConfig, VARINT_LEN}, }; pub(crate) type FramedDc = Framed>; -pub(crate) fn new(inner: T) -> FramedDc +pub(crate) fn new(inner: T, config: StreamConfig) -> FramedDc where T: AsyncRead + AsyncWrite, { - let mut framed = Framed::new(inner, codec()); + let mut framed = Framed::new(inner, codec(config)); // If not set, `Framed` buffers up to 131kB of data before sending, which leads to "outbound // packet larger than maximum message size" error in webrtc-rs. - framed.set_send_high_water_mark(MAX_DATA_LEN); + framed.set_send_high_water_mark(config.max_data_size()); framed } -pub(crate) fn codec() -> prost_codec::Codec { - prost_codec::Codec::new(MAX_MSG_LEN - VARINT_LEN) +pub(crate) fn codec(config: StreamConfig) -> prost_codec::Codec { + prost_codec::Codec::new(config.max_message_size() - VARINT_LEN) } diff --git a/transports/webrtc-websys/src/connection.rs b/transports/webrtc-websys/src/connection.rs index acd84cfe313..794016eb347 100644 --- a/transports/webrtc-websys/src/connection.rs +++ b/transports/webrtc-websys/src/connection.rs @@ -8,7 +8,7 @@ use std::{ use futures::{StreamExt, channel::mpsc, stream::FuturesUnordered}; use js_sys::{Object, Reflect}; use libp2p_core::muxing::{StreamMuxer, StreamMuxerEvent}; -use libp2p_webrtc_utils::Fingerprint; +use libp2p_webrtc_utils::{Fingerprint, StreamConfig}; use send_wrapper::SendWrapper; use wasm_bindgen::prelude::*; use wasm_bindgen_futures::JsFuture; @@ -38,13 +38,14 @@ pub struct Connection { /// A list of futures, which, once completed, signal that a [`Stream`] has been dropped. drop_listeners: FuturesUnordered, no_drop_listeners_waker: Option, + stream_config: StreamConfig, _ondatachannel_closure: SendWrapper>, } impl Connection { /// Create a new inner WebRTC Connection - pub(crate) fn new(peer_connection: RtcPeerConnection) -> Self { + pub(crate) fn new(peer_connection: RtcPeerConnection, stream_config: StreamConfig) -> Self { // An ondatachannel Future enables us to poll for incoming data channel events in // poll_incoming let (mut tx_ondatachannel, rx_ondatachannel) = mpsc::channel(4); // we may get more than one data channel opened on a single peer connection @@ -72,13 +73,14 @@ impl Connection { closed: false, drop_listeners: FuturesUnordered::default(), no_drop_listeners_waker: None, + stream_config, inbound_data_channels: SendWrapper::new(rx_ondatachannel), _ondatachannel_closure: SendWrapper::new(ondatachannel_closure), } } fn new_stream_from_data_channel(&mut self, data_channel: RtcDataChannel) -> Stream { - let (stream, drop_listener) = Stream::new(data_channel); + let (stream, drop_listener) = Stream::new(data_channel, self.stream_config); self.drop_listeners.push(drop_listener); if let Some(waker) = self.no_drop_listeners_waker.take() { @@ -204,8 +206,8 @@ impl RtcPeerConnection { /// Creates the stream for the initial noise handshake. /// /// The underlying data channel MUST have `negotiated` set to `true` and carry the ID 0. - pub(crate) fn new_handshake_stream(&self) -> (Stream, DropListener) { - Stream::new(self.new_data_channel(true)) + pub(crate) fn new_handshake_stream(&self, config: StreamConfig) -> (Stream, DropListener) { + Stream::new(self.new_data_channel(true), config) } /// Creates a regular data channel for when the connection is already established. diff --git a/transports/webrtc-websys/src/stream.rs b/transports/webrtc-websys/src/stream.rs index ee0183b07f0..ca444570fd5 100644 --- a/transports/webrtc-websys/src/stream.rs +++ b/transports/webrtc-websys/src/stream.rs @@ -5,6 +5,7 @@ use std::{ }; use futures::{AsyncRead, AsyncWrite}; +use libp2p_webrtc_utils::StreamConfig; use send_wrapper::SendWrapper; use web_sys::RtcDataChannel; @@ -23,9 +24,11 @@ pub struct Stream { pub(crate) type DropListener = SendWrapper>; impl Stream { - pub(crate) fn new(data_channel: RtcDataChannel) -> (Self, DropListener) { - let (inner, drop_listener) = - libp2p_webrtc_utils::Stream::new(PollDataChannel::new(data_channel)); + pub(crate) fn new(data_channel: RtcDataChannel, config: StreamConfig) -> (Self, DropListener) { + let (inner, drop_listener) = libp2p_webrtc_utils::Stream::with_config( + PollDataChannel::new(data_channel, config), + config, + ); ( Self { diff --git a/transports/webrtc-websys/src/stream/poll_data_channel.rs b/transports/webrtc-websys/src/stream/poll_data_channel.rs index eaeec4d6789..07a14d6e135 100644 --- a/transports/webrtc-websys/src/stream/poll_data_channel.rs +++ b/transports/webrtc-websys/src/stream/poll_data_channel.rs @@ -12,7 +12,7 @@ use std::{ use bytes::BytesMut; use futures::{AsyncRead, AsyncWrite, task::AtomicWaker}; -use libp2p_webrtc_utils::MAX_MSG_LEN; +use libp2p_webrtc_utils::StreamConfig; use wasm_bindgen::prelude::*; use web_sys::{Event, MessageEvent, RtcDataChannel, RtcDataChannelEvent, RtcDataChannelState}; @@ -44,6 +44,7 @@ pub(crate) struct PollDataChannel { /// Failing these will (very likely), /// cause the application developer to drop the stream which resets it. overloaded: Rc, + max_message_size: usize, // Store the closures for proper garbage collection. // These are wrapped in an [`Rc`] so we can implement [`Clone`]. @@ -54,7 +55,7 @@ pub(crate) struct PollDataChannel { } impl PollDataChannel { - pub(crate) fn new(inner: RtcDataChannel) -> Self { + pub(crate) fn new(inner: RtcDataChannel, config: StreamConfig) -> Self { let open_waker = Rc::new(AtomicWaker::new()); let on_open_closure = Closure::new({ let open_waker = open_waker.clone(); @@ -105,7 +106,7 @@ impl PollDataChannel { let mut read_buffer = read_buffer.lock().unwrap(); - if read_buffer.len() + data.length() as usize > MAX_MSG_LEN { + if read_buffer.len() + data.length() as usize > config.max_message_size() { overloaded.store(true, Ordering::SeqCst); tracing::warn!("Remote is overloading us with messages, resetting stream",); return; @@ -125,6 +126,7 @@ impl PollDataChannel { write_waker, close_waker, overloaded, + max_message_size: config.max_message_size(), _on_open_closure: Rc::new(on_open_closure), _on_write_closure: Rc::new(on_write_closure), _on_close_closure: Rc::new(on_close_closure), @@ -207,8 +209,8 @@ impl AsyncWrite for PollDataChannel { futures::ready!(this.poll_ready(cx))?; - debug_assert!(this.buffered_amount() <= MAX_MSG_LEN); - let remaining_space = MAX_MSG_LEN - this.buffered_amount(); + debug_assert!(this.buffered_amount() <= this.max_message_size); + let remaining_space = this.max_message_size - this.buffered_amount(); if remaining_space == 0 { this.write_waker.register(cx.waker()); diff --git a/transports/webrtc-websys/src/transport.rs b/transports/webrtc-websys/src/transport.rs index 365216f35c2..8052467d7b6 100644 --- a/transports/webrtc-websys/src/transport.rs +++ b/transports/webrtc-websys/src/transport.rs @@ -1,5 +1,7 @@ +use libp2p_webrtc_utils::StreamConfig; use std::{ future::Future, + num::NonZeroUsize, pin::Pin, task::{Context, Poll}, }; @@ -18,6 +20,7 @@ use super::{Connection, Error, upgrade}; #[derive(Clone)] pub struct Config { keypair: Keypair, + stream_config: StreamConfig, } /// A WebTransport [`Transport`](libp2p_core::Transport) that works with `web-sys`. @@ -30,8 +33,15 @@ impl Config { pub fn new(keypair: &Keypair) -> Self { Config { keypair: keypair.to_owned(), + stream_config: StreamConfig::default(), } } + + /// Limits encoded WebRTC data-channel messages for connections created by this transport. + pub fn with_max_message_size(mut self, max_message_size: NonZeroUsize) -> Self { + self.stream_config = StreamConfig::new(max_message_size); + self + } } impl Transport { @@ -93,8 +103,13 @@ impl libp2p_core::Transport for Transport { let config = self.config.clone(); Ok(async move { - let (peer_id, connection) = - upgrade::outbound(sock_addr, server_fingerprint, config.keypair.clone()).await?; + let (peer_id, connection) = upgrade::outbound( + sock_addr, + server_fingerprint, + config.keypair.clone(), + config.stream_config, + ) + .await?; Ok((peer_id, connection)) } diff --git a/transports/webrtc-websys/src/upgrade.rs b/transports/webrtc-websys/src/upgrade.rs index d1e322a0838..1702e2d037d 100644 --- a/transports/webrtc-websys/src/upgrade.rs +++ b/transports/webrtc-websys/src/upgrade.rs @@ -1,7 +1,7 @@ use std::net::SocketAddr; use libp2p_identity::{Keypair, PeerId}; -use libp2p_webrtc_utils::{Fingerprint, noise}; +use libp2p_webrtc_utils::{Fingerprint, StreamConfig, noise}; use send_wrapper::SendWrapper; use super::Error; @@ -13,8 +13,14 @@ pub(crate) async fn outbound( sock_addr: SocketAddr, remote_fingerprint: Fingerprint, id_keys: Keypair, + stream_config: StreamConfig, ) -> Result<(PeerId, Connection), Error> { - let fut = SendWrapper::new(outbound_inner(sock_addr, remote_fingerprint, id_keys)); + let fut = SendWrapper::new(outbound_inner( + sock_addr, + remote_fingerprint, + id_keys, + stream_config, + )); fut.await } @@ -23,12 +29,13 @@ async fn outbound_inner( sock_addr: SocketAddr, remote_fingerprint: Fingerprint, id_keys: Keypair, + stream_config: StreamConfig, ) -> Result<(PeerId, Connection), Error> { let rtc_peer_connection = RtcPeerConnection::new(remote_fingerprint.algorithm()).await?; // Create stream for Noise handshake // Must create data channel before Offer is created for it to be included in the SDP - let (channel, listener) = rtc_peer_connection.new_handshake_stream(); + let (channel, listener) = rtc_peer_connection.new_handshake_stream(stream_config); drop(listener); let ufrag = libp2p_webrtc_utils::sdp::random_ufrag(); @@ -47,11 +54,17 @@ async fn outbound_inner( tracing::trace!(?local_fingerprint); tracing::trace!(?remote_fingerprint); - let peer_id = noise::outbound(id_keys, channel, remote_fingerprint, local_fingerprint) - .await - .map_err(AuthenticationError)?; + let (peer_id, stream_config) = noise::outbound_with_message_size( + id_keys, + channel, + remote_fingerprint, + local_fingerprint, + stream_config, + ) + .await + .map_err(AuthenticationError)?; tracing::debug!(peer=%peer_id, "Remote peer identified"); - Ok((peer_id, Connection::new(rtc_peer_connection))) + Ok((peer_id, Connection::new(rtc_peer_connection, stream_config))) } diff --git a/transports/webrtc/src/tokio/connection.rs b/transports/webrtc/src/tokio/connection.rs index 3009e7ed33c..03062a6ad7e 100644 --- a/transports/webrtc/src/tokio/connection.rs +++ b/transports/webrtc/src/tokio/connection.rs @@ -36,6 +36,7 @@ use futures::{ stream::FuturesUnordered, }; use libp2p_core::muxing::{StreamMuxer, StreamMuxerEvent}; +use libp2p_webrtc_utils::StreamConfig; use webrtc::{ data::data_channel::DataChannel as DetachedDataChannel, data_channel::RTCDataChannel, peer_connection::RTCPeerConnection, @@ -66,13 +67,14 @@ pub struct Connection { /// A list of futures, which, once completed, signal that a [`Stream`] has been dropped. drop_listeners: FuturesUnordered, no_drop_listeners_waker: Option, + stream_config: StreamConfig, } impl Unpin for Connection {} impl Connection { /// Creates a new connection. - pub(crate) async fn new(rtc_conn: RTCPeerConnection) -> Self { + pub(crate) async fn new(rtc_conn: RTCPeerConnection, stream_config: StreamConfig) -> Self { let (data_channel_tx, data_channel_rx) = mpsc::channel(MAX_DATA_CHANNELS_IN_FLIGHT); Connection::register_incoming_data_channels_handler( @@ -88,6 +90,7 @@ impl Connection { close_fut: None, drop_listeners: FuturesUnordered::default(), no_drop_listeners_waker: None, + stream_config, } } @@ -159,7 +162,7 @@ impl StreamMuxer for Connection { Some(detached) => { tracing::trace!(stream=%detached.stream_identifier(), "Incoming stream"); - let (stream, drop_listener) = Stream::new(detached); + let (stream, drop_listener) = Stream::new(detached, self.stream_config); self.drop_listeners.push(drop_listener); if let Some(waker) = self.no_drop_listeners_waker.take() { waker.wake() @@ -233,7 +236,7 @@ impl StreamMuxer for Connection { tracing::trace!(stream=%detached.stream_identifier(), "Outbound stream"); - let (stream, drop_listener) = Stream::new(detached); + let (stream, drop_listener) = Stream::new(detached, self.stream_config); self.drop_listeners.push(drop_listener); if let Some(waker) = self.no_drop_listeners_waker.take() { waker.wake() diff --git a/transports/webrtc/src/tokio/sdp.rs b/transports/webrtc/src/tokio/sdp.rs index f28c5c33105..8d233b34ef5 100644 --- a/transports/webrtc/src/tokio/sdp.rs +++ b/transports/webrtc/src/tokio/sdp.rs @@ -140,5 +140,5 @@ a=ice-pwd:{pwd} a=fingerprint:{fingerprint_algorithm} {fingerprint_value} a=setup:actpass a=sctp-port:5000 -a=max-message-size:16384 +a=max-message-size:{max_message_size} "; diff --git a/transports/webrtc/src/tokio/stream.rs b/transports/webrtc/src/tokio/stream.rs index 9d5a9faf440..3b98ab79bac 100644 --- a/transports/webrtc/src/tokio/stream.rs +++ b/transports/webrtc/src/tokio/stream.rs @@ -25,7 +25,7 @@ use std::{ }; use futures::prelude::*; -use libp2p_webrtc_utils::MAX_MSG_LEN; +use libp2p_webrtc_utils::StreamConfig; use tokio_util::compat::{Compat, TokioAsyncReadCompatExt}; use webrtc::data::data_channel::{DataChannel, PollDataChannel}; @@ -42,11 +42,16 @@ pub(crate) type DropListener = libp2p_webrtc_utils::DropListener) -> (Self, DropListener) { + pub(crate) fn new( + data_channel: Arc, + config: StreamConfig, + ) -> (Self, DropListener) { let mut data_channel = PollDataChannel::new(data_channel).compat(); - data_channel.get_mut().set_read_buf_capacity(MAX_MSG_LEN); + data_channel + .get_mut() + .set_read_buf_capacity(config.max_message_size()); - let (inner, drop_listener) = libp2p_webrtc_utils::Stream::new(data_channel); + let (inner, drop_listener) = libp2p_webrtc_utils::Stream::with_config(data_channel, config); (Self { inner }, drop_listener) } diff --git a/transports/webrtc/src/tokio/transport.rs b/transports/webrtc/src/tokio/transport.rs index d5857a42c11..f6eca7ca2ce 100644 --- a/transports/webrtc/src/tokio/transport.rs +++ b/transports/webrtc/src/tokio/transport.rs @@ -18,9 +18,11 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. +use libp2p_webrtc_utils::StreamConfig; use std::{ io, net::{IpAddr, SocketAddr}, + num::NonZeroUsize, pin::Pin, task::{Context, Poll, Waker}, }; @@ -71,6 +73,12 @@ impl Transport { listeners: SelectAll::new(), } } + + /// Limits encoded WebRTC data-channel messages for connections created by this transport. + pub fn with_max_message_size(mut self, max_message_size: NonZeroUsize) -> Self { + self.config.stream_config = StreamConfig::new(max_message_size); + self + } } impl libp2p_core::Transport for Transport { @@ -150,6 +158,7 @@ impl libp2p_core::Transport for Transport { let (peer_id, connection) = upgrade::outbound( sock_addr, config.inner, + config.stream_config, udp_mux, client_fingerprint.into_inner(), server_fingerprint, @@ -331,6 +340,7 @@ impl Stream for ListenStream { let upgrade = upgrade::inbound( new_addr.addr, self.config.inner.clone(), + self.config.stream_config, self.udp_mux.udp_mux_handle(), self.config.fingerprint.into_inner(), new_addr.ufrag, @@ -365,6 +375,7 @@ struct Config { inner: RTCConfiguration, fingerprint: Fingerprint, id_keys: identity::Keypair, + stream_config: StreamConfig, } impl Config { @@ -379,6 +390,7 @@ impl Config { ..RTCConfiguration::default() }, fingerprint, + stream_config: StreamConfig::default(), } } } diff --git a/transports/webrtc/src/tokio/upgrade.rs b/transports/webrtc/src/tokio/upgrade.rs index 5333dda5c94..f2cb1b73e00 100644 --- a/transports/webrtc/src/tokio/upgrade.rs +++ b/transports/webrtc/src/tokio/upgrade.rs @@ -31,7 +31,7 @@ use futures::{channel::oneshot, future::Either}; use futures_timer::Delay; use libp2p_identity as identity; use libp2p_identity::PeerId; -use libp2p_webrtc_utils::{Fingerprint, noise}; +use libp2p_webrtc_utils::{Fingerprint, StreamConfig, noise}; use webrtc::{ api::{APIBuilder, setting_engine::SettingEngine}, data::data_channel::DataChannel, @@ -48,6 +48,7 @@ use crate::tokio::{Connection, error::Error, sdp, sdp::random_ufrag, stream::Str pub(crate) async fn outbound( addr: SocketAddr, config: RTCConfiguration, + stream_config: StreamConfig, udp_mux: Arc, client_fingerprint: Fingerprint, server_fingerprint: Fingerprint, @@ -67,15 +68,19 @@ pub(crate) async fn outbound( peer_connection.set_remote_description(answer).await?; // This will start the gathering of ICE candidates. let data_channel = await_noise_data_channel_open(noise_channel_open_rx).await?; - let peer_id = noise::outbound( + let (peer_id, stream_config) = noise::outbound_with_message_size( id_keys, data_channel, server_fingerprint, client_fingerprint, + stream_config, ) .await?; - Ok((peer_id, Connection::new(peer_connection).await)) + Ok(( + peer_id, + Connection::new(peer_connection, stream_config).await, + )) } /// Creates a new inbound WebRTC connection. @@ -83,6 +88,7 @@ pub(crate) async fn outbound( pub(crate) async fn inbound( addr: SocketAddr, config: RTCConfiguration, + stream_config: StreamConfig, udp_mux: Arc, server_fingerprint: Fingerprint, remote_ufrag: String, @@ -103,15 +109,19 @@ pub(crate) async fn inbound( let data_channel = await_noise_data_channel_open(noise_channel_open_rx).await?; let client_fingerprint = get_remote_fingerprint(&peer_connection).await; - let peer_id = noise::inbound( + let (peer_id, stream_config) = noise::inbound_with_message_size( id_keys, data_channel, client_fingerprint, server_fingerprint, + stream_config, ) .await?; - Ok((peer_id, Connection::new(peer_connection).await)) + Ok(( + peer_id, + Connection::new(peer_connection, stream_config).await, + )) } #[allow(clippy::result_large_err)] @@ -248,7 +258,7 @@ async fn await_noise_data_channel_open( } }; - let (substream, drop_listener) = Stream::new(channel); + let (substream, drop_listener) = Stream::new(channel, StreamConfig::default()); drop(drop_listener); // Don't care about cancelled substreams during initial handshake. Ok(substream) From 9e3bcd9b2b30ebc33a1d35c10bd64e3ff14dffd5 Mon Sep 17 00:00:00 2001 From: yexiyue Date: Fri, 24 Jul 2026 23:25:53 +0800 Subject: [PATCH 2/5] docs: add WebRTC message limit changelogs --- misc/webrtc-utils/CHANGELOG.md | 2 ++ transports/webrtc-websys/CHANGELOG.md | 2 ++ transports/webrtc/CHANGELOG.md | 2 ++ 3 files changed, 6 insertions(+) diff --git a/misc/webrtc-utils/CHANGELOG.md b/misc/webrtc-utils/CHANGELOG.md index 600101a55e7..12ca527e4c0 100644 --- a/misc/webrtc-utils/CHANGELOG.md +++ b/misc/webrtc-utils/CHANGELOG.md @@ -1,5 +1,7 @@ ## 0.5.0 +- Negotiate WebRTC data-channel message limits after Noise authentication. + - Revert migration to `quick-protobuf`, migrate back to `prost`. See [PR 6363](https://github.com/libp2p/rust-libp2p/pull/6363). diff --git a/transports/webrtc-websys/CHANGELOG.md b/transports/webrtc-websys/CHANGELOG.md index 4e9f05ae892..ca9bbf39d99 100644 --- a/transports/webrtc-websys/CHANGELOG.md +++ b/transports/webrtc-websys/CHANGELOG.md @@ -1,5 +1,7 @@ ## 0.5.0 +- Add `Config::with_max_message_size` to configure the WebRTC data-channel framing limit. + - Require `getrandom/js` feature only under `wasm` target. See [PR 6102](https://github.com/libp2p/rust-libp2p/pull/6102) diff --git a/transports/webrtc/CHANGELOG.md b/transports/webrtc/CHANGELOG.md index b090ebd3efe..d22446f89be 100644 --- a/transports/webrtc/CHANGELOG.md +++ b/transports/webrtc/CHANGELOG.md @@ -1,5 +1,7 @@ ## 0.10.0-alpha +- Add `Transport::with_max_message_size` to configure the WebRTC data-channel framing limit. + - Update webrtc-rs to `v0.17` and fix libp2p noise data channel negotiation. See [PR 6429](https://github.com/libp2p/rust-libp2p/pull/6429) From 5984c716b5cd0d81c2ca4dc71e222d7b700b888d Mon Sep 17 00:00:00 2001 From: yexiyue Date: Sat, 25 Jul 2026 11:27:49 +0800 Subject: [PATCH 3/5] fix(webrtc-websys): configure receive buffer limit --- transports/webrtc-websys/CHANGELOG.md | 1 + transports/webrtc-websys/src/connection.rs | 20 ++++- transports/webrtc-websys/src/stream.rs | 9 ++- .../src/stream/poll_data_channel.rs | 73 ++++++++++++++++++- transports/webrtc-websys/src/transport.rs | 16 ++++ transports/webrtc-websys/src/upgrade.rs | 13 +++- 6 files changed, 120 insertions(+), 12 deletions(-) diff --git a/transports/webrtc-websys/CHANGELOG.md b/transports/webrtc-websys/CHANGELOG.md index ca9bbf39d99..c2c4b6dbfa4 100644 --- a/transports/webrtc-websys/CHANGELOG.md +++ b/transports/webrtc-websys/CHANGELOG.md @@ -1,6 +1,7 @@ ## 0.5.0 - Add `Config::with_max_message_size` to configure the WebRTC data-channel framing limit. +- Add `Config::with_max_read_buffer_size` to configure the browser callback receive buffer. - Require `getrandom/js` feature only under `wasm` target. See [PR 6102](https://github.com/libp2p/rust-libp2p/pull/6102) diff --git a/transports/webrtc-websys/src/connection.rs b/transports/webrtc-websys/src/connection.rs index 794016eb347..fa0854bf90c 100644 --- a/transports/webrtc-websys/src/connection.rs +++ b/transports/webrtc-websys/src/connection.rs @@ -1,6 +1,7 @@ //! A libp2p connection backed by an [RtcPeerConnection](https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection). use std::{ + num::NonZeroUsize, pin::Pin, task::{Context, Poll, Waker, ready}, }; @@ -39,13 +40,18 @@ pub struct Connection { drop_listeners: FuturesUnordered, no_drop_listeners_waker: Option, stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, _ondatachannel_closure: SendWrapper>, } impl Connection { /// Create a new inner WebRTC Connection - pub(crate) fn new(peer_connection: RtcPeerConnection, stream_config: StreamConfig) -> Self { + pub(crate) fn new( + peer_connection: RtcPeerConnection, + stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, + ) -> Self { // An ondatachannel Future enables us to poll for incoming data channel events in // poll_incoming let (mut tx_ondatachannel, rx_ondatachannel) = mpsc::channel(4); // we may get more than one data channel opened on a single peer connection @@ -74,13 +80,15 @@ impl Connection { drop_listeners: FuturesUnordered::default(), no_drop_listeners_waker: None, stream_config, + max_read_buffer_size, inbound_data_channels: SendWrapper::new(rx_ondatachannel), _ondatachannel_closure: SendWrapper::new(ondatachannel_closure), } } fn new_stream_from_data_channel(&mut self, data_channel: RtcDataChannel) -> Stream { - let (stream, drop_listener) = Stream::new(data_channel, self.stream_config); + let (stream, drop_listener) = + Stream::new(data_channel, self.stream_config, self.max_read_buffer_size); self.drop_listeners.push(drop_listener); if let Some(waker) = self.no_drop_listeners_waker.take() { @@ -206,8 +214,12 @@ impl RtcPeerConnection { /// Creates the stream for the initial noise handshake. /// /// The underlying data channel MUST have `negotiated` set to `true` and carry the ID 0. - pub(crate) fn new_handshake_stream(&self, config: StreamConfig) -> (Stream, DropListener) { - Stream::new(self.new_data_channel(true), config) + pub(crate) fn new_handshake_stream( + &self, + config: StreamConfig, + max_read_buffer_size: NonZeroUsize, + ) -> (Stream, DropListener) { + Stream::new(self.new_data_channel(true), config, max_read_buffer_size) } /// Creates a regular data channel for when the connection is already established. diff --git a/transports/webrtc-websys/src/stream.rs b/transports/webrtc-websys/src/stream.rs index ca444570fd5..dc0c28bb0a1 100644 --- a/transports/webrtc-websys/src/stream.rs +++ b/transports/webrtc-websys/src/stream.rs @@ -1,5 +1,6 @@ //! The WebRTC [Stream] over the Connection use std::{ + num::NonZeroUsize, pin::Pin, task::{Context, Poll}, }; @@ -24,9 +25,13 @@ pub struct Stream { pub(crate) type DropListener = SendWrapper>; impl Stream { - pub(crate) fn new(data_channel: RtcDataChannel, config: StreamConfig) -> (Self, DropListener) { + pub(crate) fn new( + data_channel: RtcDataChannel, + config: StreamConfig, + max_read_buffer_size: NonZeroUsize, + ) -> (Self, DropListener) { let (inner, drop_listener) = libp2p_webrtc_utils::Stream::with_config( - PollDataChannel::new(data_channel, config), + PollDataChannel::new(data_channel, config, max_read_buffer_size), config, ); diff --git a/transports/webrtc-websys/src/stream/poll_data_channel.rs b/transports/webrtc-websys/src/stream/poll_data_channel.rs index 07a14d6e135..80de9212735 100644 --- a/transports/webrtc-websys/src/stream/poll_data_channel.rs +++ b/transports/webrtc-websys/src/stream/poll_data_channel.rs @@ -1,6 +1,7 @@ use std::{ - cmp::min, + cmp::{max, min}, io, + num::NonZeroUsize, pin::Pin, rc::Rc, sync::{ @@ -16,6 +17,14 @@ use libp2p_webrtc_utils::StreamConfig; use wasm_bindgen::prelude::*; use web_sys::{Event, MessageEvent, RtcDataChannel, RtcDataChannelEvent, RtcDataChannelState}; +fn effective_max_read_buffer_size(config: StreamConfig, configured: NonZeroUsize) -> usize { + max(configured.get(), config.max_message_size()) +} + +fn read_buffer_overflows(buffered: usize, incoming: usize, max_read_buffer_size: usize) -> bool { + buffered.saturating_add(incoming) > max_read_buffer_size +} + /// [`PollDataChannel`] is a wrapper around [`RtcDataChannel`] which implements [`AsyncRead`] and /// [`AsyncWrite`]. #[derive(Debug, Clone)] @@ -55,7 +64,13 @@ pub(crate) struct PollDataChannel { } impl PollDataChannel { - pub(crate) fn new(inner: RtcDataChannel, config: StreamConfig) -> Self { + pub(crate) fn new( + inner: RtcDataChannel, + config: StreamConfig, + configured_max_read_buffer_size: NonZeroUsize, + ) -> Self { + let max_read_buffer_size = + effective_max_read_buffer_size(config, configured_max_read_buffer_size); let open_waker = Rc::new(AtomicWaker::new()); let on_open_closure = Closure::new({ let open_waker = open_waker.clone(); @@ -106,7 +121,11 @@ impl PollDataChannel { let mut read_buffer = read_buffer.lock().unwrap(); - if read_buffer.len() + data.length() as usize > config.max_message_size() { + if read_buffer_overflows( + read_buffer.len(), + data.length() as usize, + max_read_buffer_size, + ) { overloaded.store(true, Ordering::SeqCst); tracing::warn!("Remote is overloading us with messages, resetting stream",); return; @@ -169,6 +188,54 @@ impl PollDataChannel { } } +#[cfg(test)] +mod tests { + use std::num::NonZeroUsize; + + use libp2p_webrtc_utils::StreamConfig; + + use super::{effective_max_read_buffer_size, read_buffer_overflows}; + + #[test] + fn read_buffer_accepts_multiple_valid_messages() { + let message_size = NonZeroUsize::new(8 * 1024).unwrap(); + let max_read_buffer_size = effective_max_read_buffer_size( + StreamConfig::new(message_size), + NonZeroUsize::new(256 * 1024).unwrap(), + ); + + assert!(!read_buffer_overflows( + message_size.get(), + message_size.get(), + max_read_buffer_size, + )); + } + + #[test] + fn configured_read_buffer_remains_bounded() { + let max_read_buffer_size = 16 * 1024; + + assert!(read_buffer_overflows( + max_read_buffer_size, + 1, + max_read_buffer_size, + )); + } + + #[test] + fn read_buffer_can_hold_one_larger_valid_message() { + let message_size = NonZeroUsize::new(512 * 1024).unwrap(); + + assert_eq!( + effective_max_read_buffer_size( + StreamConfig::new(message_size), + NonZeroUsize::new(256 * 1024).unwrap(), + ), + message_size.get(), + ); + } +} + impl AsyncRead for PollDataChannel { fn poll_read( self: Pin<&mut Self>, diff --git a/transports/webrtc-websys/src/transport.rs b/transports/webrtc-websys/src/transport.rs index 8052467d7b6..ec409d0469a 100644 --- a/transports/webrtc-websys/src/transport.rs +++ b/transports/webrtc-websys/src/transport.rs @@ -16,11 +16,16 @@ use libp2p_identity::{Keypair, PeerId}; use super::{Connection, Error, upgrade}; +/// Default maximum number of bytes browser callbacks may queue before the Rust task polls them. +const DEFAULT_MAX_READ_BUFFER_SIZE: NonZeroUsize = + NonZeroUsize::new(256 * 1024).expect("constant is non-zero"); + /// Config for the [`Transport`]. #[derive(Clone)] pub struct Config { keypair: Keypair, stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, } /// A WebTransport [`Transport`](libp2p_core::Transport) that works with `web-sys`. @@ -34,6 +39,7 @@ impl Config { Config { keypair: keypair.to_owned(), stream_config: StreamConfig::default(), + max_read_buffer_size: DEFAULT_MAX_READ_BUFFER_SIZE, } } @@ -42,6 +48,15 @@ impl Config { self.stream_config = StreamConfig::new(max_message_size); self } + + /// Limits bytes queued by browser data-channel callbacks before the stream is polled. + /// + /// The effective limit is raised as needed to hold one valid encoded message. This limit is + /// local to the browser transport and is not negotiated with the remote peer. + pub fn with_max_read_buffer_size(mut self, max_read_buffer_size: NonZeroUsize) -> Self { + self.max_read_buffer_size = max_read_buffer_size; + self + } } impl Transport { @@ -108,6 +123,7 @@ impl libp2p_core::Transport for Transport { server_fingerprint, config.keypair.clone(), config.stream_config, + config.max_read_buffer_size, ) .await?; diff --git a/transports/webrtc-websys/src/upgrade.rs b/transports/webrtc-websys/src/upgrade.rs index 1702e2d037d..abf5165b128 100644 --- a/transports/webrtc-websys/src/upgrade.rs +++ b/transports/webrtc-websys/src/upgrade.rs @@ -1,4 +1,4 @@ -use std::net::SocketAddr; +use std::{net::SocketAddr, num::NonZeroUsize}; use libp2p_identity::{Keypair, PeerId}; use libp2p_webrtc_utils::{Fingerprint, StreamConfig, noise}; @@ -14,12 +14,14 @@ pub(crate) async fn outbound( remote_fingerprint: Fingerprint, id_keys: Keypair, stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, ) -> Result<(PeerId, Connection), Error> { let fut = SendWrapper::new(outbound_inner( sock_addr, remote_fingerprint, id_keys, stream_config, + max_read_buffer_size, )); fut.await } @@ -30,12 +32,14 @@ async fn outbound_inner( remote_fingerprint: Fingerprint, id_keys: Keypair, stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, ) -> Result<(PeerId, Connection), Error> { let rtc_peer_connection = RtcPeerConnection::new(remote_fingerprint.algorithm()).await?; // Create stream for Noise handshake // Must create data channel before Offer is created for it to be included in the SDP - let (channel, listener) = rtc_peer_connection.new_handshake_stream(stream_config); + let (channel, listener) = + rtc_peer_connection.new_handshake_stream(stream_config, max_read_buffer_size); drop(listener); let ufrag = libp2p_webrtc_utils::sdp::random_ufrag(); @@ -66,5 +70,8 @@ async fn outbound_inner( tracing::debug!(peer=%peer_id, "Remote peer identified"); - Ok((peer_id, Connection::new(rtc_peer_connection, stream_config))) + Ok(( + peer_id, + Connection::new(rtc_peer_connection, stream_config, max_read_buffer_size), + )) } From acaaf0ba0bc241fa6ffd6fc3d40438c55fb9f9bd Mon Sep 17 00:00:00 2001 From: yexiyue Date: Tue, 11 Aug 2026 21:59:54 +0900 Subject: [PATCH 4/5] fix(webrtc): advertise the configured message size, not a constant `render_description` took a `{max_message_size}` placeholder but the context still filled in `16 * 1024`, so `with_max_message_size` reached the framing layer while SDP kept announcing 16 KiB regardless. That only stays harmless while the configured limit is *below* 16 KiB, as it is by default: the endpoint then sends less than it advertised. Configure anything larger and the peer's SCTP is told to expect 16 KiB while messages up to the new limit arrive. Both `sdp::answer` and `sdp::render_description` now take the `StreamConfig` the framing layer is built from, so the advertised limit and the enforced one cannot drift apart. Both call sites already had it in scope. The test asserts four different sizes: a single one would also pass against a hard-coded value that happens to match it, which is how this survived review. --- misc/webrtc-utils/CHANGELOG.md | 2 + misc/webrtc-utils/src/sdp.rs | 65 +++++++++++++++++++++++-- transports/webrtc-websys/src/sdp.rs | 4 +- transports/webrtc-websys/src/upgrade.rs | 2 +- transports/webrtc/src/tokio/sdp.rs | 11 ++++- transports/webrtc/src/tokio/upgrade.rs | 4 +- 6 files changed, 79 insertions(+), 9 deletions(-) diff --git a/misc/webrtc-utils/CHANGELOG.md b/misc/webrtc-utils/CHANGELOG.md index 12ca527e4c0..377eebbf549 100644 --- a/misc/webrtc-utils/CHANGELOG.md +++ b/misc/webrtc-utils/CHANGELOG.md @@ -1,6 +1,8 @@ ## 0.5.0 - Negotiate WebRTC data-channel message limits after Noise authentication. + `sdp::answer` and `sdp::render_description` now take the `StreamConfig` they advertise + `a=max-message-size` from, instead of always announcing 16 KiB. - Revert migration to `quick-protobuf`, migrate back to `prost`. See [PR 6363](https://github.com/libp2p/rust-libp2p/pull/6363). diff --git a/misc/webrtc-utils/src/sdp.rs b/misc/webrtc-utils/src/sdp.rs index 5eb7dac5e42..f8181fbe0e1 100644 --- a/misc/webrtc-utils/src/sdp.rs +++ b/misc/webrtc-utils/src/sdp.rs @@ -24,14 +24,20 @@ use rand::{Rng, distributions::Alphanumeric, thread_rng}; use serde::Serialize; use tinytemplate::TinyTemplate; -use crate::fingerprint::Fingerprint; +use crate::{fingerprint::Fingerprint, stream::StreamConfig}; -pub fn answer(addr: SocketAddr, server_fingerprint: Fingerprint, client_ufrag: &str) -> String { +pub fn answer( + addr: SocketAddr, + server_fingerprint: Fingerprint, + client_ufrag: &str, + config: StreamConfig, +) -> String { let answer = render_description( SERVER_SESSION_DESCRIPTION, addr, server_fingerprint, client_ufrag, + config, ); tracing::trace!(%answer, "Created SDP answer"); @@ -118,11 +124,18 @@ struct DescriptionContext { } /// Renders a [`TinyTemplate`] description using the provided arguments. +/// +/// `config` supplies `a=max-message-size`, which tells the remote how large an SCTP user +/// message this endpoint is willing to receive (RFC 8841). It is deliberately the same +/// [`StreamConfig`] the framing layer is built from rather than a separate number: the two +/// must agree, and a peer that sends up to what we advertised has to find the framing layer +/// able to accept it. pub fn render_description( description: &str, addr: SocketAddr, fingerprint: Fingerprint, ufrag: &str, + config: StreamConfig, ) -> String { let mut tt = TinyTemplate::new(); tt.add_template("description", description).unwrap(); @@ -142,7 +155,7 @@ pub fn render_description( // NOTE: ufrag is equal to pwd. ufrag: ufrag.to_owned(), pwd: ufrag.to_owned(), - max_message_size: 16 * 1024, + max_message_size: config.max_message_size(), }; tt.render("description", &context).unwrap() } @@ -158,3 +171,49 @@ pub fn random_ufrag() -> String { .collect::() ) } + +#[cfg(test)] +mod tests { + use std::num::NonZeroUsize; + + use super::*; + + fn config(bytes: usize) -> StreamConfig { + StreamConfig::new(NonZeroUsize::new(bytes).expect("non-zero")) + } + + fn addr() -> SocketAddr { + "127.0.0.1:1234".parse().expect("valid address") + } + + /// `a=max-message-size` must follow the configured limit rather than a constant. + /// + /// Several sizes on purpose: asserting a single one would also pass against a hard-coded + /// value that happens to match it, which is exactly how this went unnoticed — the template + /// took a `{max_message_size}` placeholder while the context still filled in `16 * 1024`. + #[test] + fn advertised_message_size_follows_the_config() { + for bytes in [8 * 1024, 16 * 1024, 64 * 1024, 256 * 1024] { + let sdp = render_description( + SERVER_SESSION_DESCRIPTION, + addr(), + Fingerprint::FF, + "ufrag", + config(bytes), + ); + + assert!( + sdp.contains(&format!("a=max-message-size:{bytes}")), + "a {bytes} B limit was not advertised; rendered SDP was:\n{sdp}" + ); + } + } + + /// The answer helper must forward the config too, not just `render_description`. + #[test] + fn answer_advertises_the_configured_message_size() { + let sdp = answer(addr(), Fingerprint::FF, "ufrag", config(64 * 1024)); + + assert!(sdp.contains("a=max-message-size:65536"), "{sdp}"); + } +} diff --git a/transports/webrtc-websys/src/sdp.rs b/transports/webrtc-websys/src/sdp.rs index 7919ce6b091..ef482325677 100644 --- a/transports/webrtc-websys/src/sdp.rs +++ b/transports/webrtc-websys/src/sdp.rs @@ -1,6 +1,6 @@ use std::net::SocketAddr; -use libp2p_webrtc_utils::Fingerprint; +use libp2p_webrtc_utils::{Fingerprint, StreamConfig}; use web_sys::{RtcSdpType, RtcSessionDescriptionInit}; /// Creates the SDP answer used by the client. @@ -8,12 +8,14 @@ pub(crate) fn answer( addr: SocketAddr, server_fingerprint: Fingerprint, client_ufrag: &str, + config: StreamConfig, ) -> RtcSessionDescriptionInit { let answer_obj = RtcSessionDescriptionInit::new(RtcSdpType::Answer); answer_obj.set_sdp(&libp2p_webrtc_utils::sdp::answer( addr, server_fingerprint, client_ufrag, + config, )); answer_obj } diff --git a/transports/webrtc-websys/src/upgrade.rs b/transports/webrtc-websys/src/upgrade.rs index abf5165b128..9aabbcf885a 100644 --- a/transports/webrtc-websys/src/upgrade.rs +++ b/transports/webrtc-websys/src/upgrade.rs @@ -50,7 +50,7 @@ async fn outbound_inner( .set_local_description(munged_offer) .await?; - let answer = sdp::answer(sock_addr, remote_fingerprint, &ufrag); + let answer = sdp::answer(sock_addr, remote_fingerprint, &ufrag, stream_config); rtc_peer_connection.set_remote_description(answer).await?; let local_fingerprint = rtc_peer_connection.local_fingerprint()?; diff --git a/transports/webrtc/src/tokio/sdp.rs b/transports/webrtc/src/tokio/sdp.rs index 8d233b34ef5..404f109c039 100644 --- a/transports/webrtc/src/tokio/sdp.rs +++ b/transports/webrtc/src/tokio/sdp.rs @@ -21,7 +21,7 @@ use std::net::SocketAddr; pub(crate) use libp2p_webrtc_utils::sdp::random_ufrag; -use libp2p_webrtc_utils::{Fingerprint, sdp::render_description}; +use libp2p_webrtc_utils::{Fingerprint, StreamConfig, sdp::render_description}; use webrtc::peer_connection::sdp::session_description::RTCSessionDescription; /// Creates the SDP answer used by the client. @@ -29,11 +29,13 @@ pub(crate) fn answer( addr: SocketAddr, server_fingerprint: Fingerprint, client_ufrag: &str, + config: StreamConfig, ) -> RTCSessionDescription { RTCSessionDescription::answer(libp2p_webrtc_utils::sdp::answer( addr, server_fingerprint, client_ufrag, + config, )) .unwrap() } @@ -41,12 +43,17 @@ pub(crate) fn answer( /// Creates the SDP offer used by the server. /// /// Certificate verification is disabled which is why we hardcode a dummy fingerprint here. -pub(crate) fn offer(addr: SocketAddr, client_ufrag: &str) -> RTCSessionDescription { +pub(crate) fn offer( + addr: SocketAddr, + client_ufrag: &str, + config: StreamConfig, +) -> RTCSessionDescription { let offer = render_description( CLIENT_SESSION_DESCRIPTION, addr, Fingerprint::FF, client_ufrag, + config, ); tracing::trace!(offer=%offer, "Created SDP offer"); diff --git a/transports/webrtc/src/tokio/upgrade.rs b/transports/webrtc/src/tokio/upgrade.rs index f2cb1b73e00..d9f86a846cb 100644 --- a/transports/webrtc/src/tokio/upgrade.rs +++ b/transports/webrtc/src/tokio/upgrade.rs @@ -63,7 +63,7 @@ pub(crate) async fn outbound( tracing::debug!(offer=%offer.sdp, "created SDP offer for outbound connection"); peer_connection.set_local_description(offer).await?; - let answer = sdp::answer(addr, server_fingerprint, &ufrag); + let answer = sdp::answer(addr, server_fingerprint, &ufrag, stream_config); tracing::debug!(?answer, "calculated SDP answer for outbound connection"); peer_connection.set_remote_description(answer).await?; // This will start the gathering of ICE candidates. @@ -99,7 +99,7 @@ pub(crate) async fn inbound( let peer_connection = new_inbound_connection(addr, config, udp_mux, &remote_ufrag).await?; let noise_channel_open_rx = create_noise_data_channel(&peer_connection).await?; - let offer = sdp::offer(addr, &remote_ufrag); + let offer = sdp::offer(addr, &remote_ufrag, stream_config); tracing::debug!(?offer, "calculated SDP offer for inbound connection"); peer_connection.set_remote_description(offer).await?; From 07e564827b0d68bdaa62ce9021e6a91bbaf32c10 Mon Sep 17 00:00:00 2001 From: yexiyue Date: Tue, 11 Aug 2026 22:51:15 +0900 Subject: [PATCH 5/5] fix(webrtc): send one frame per data-channel message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The high-water mark is a *lower* bound on when to flush, not an upper bound on the buffer: `poll_ready` flushes while `buffer.len() >= hwm`, then `start_send` appends a whole frame. Setting it to `max_data_size()` therefore allowed a short frame — one that leaves the buffer below the mark — to be written out together with the full-size frame that followed it. The layer below turns one write into exactly one SCTP user message, so that coalesced write becomes a message larger than the negotiated `max_message_size`. webrtc-rs rejects it with "outbound packet larger than maximum message size" and the frame is simply lost; the byte stream above never re-syncs. Measured on a 1 MiB transfer with an 8 KiB limit: 125 writes of 8190 B and three of 8419 B. SCTP rejected exactly those three, and the receiver ended up 49,467 B short. This was previously masked: the SDP always advertised a hard-coded 16 KiB while this repo's framing used 8 KiB, so the oversized writes still fit under the advertised limit. Advertising the configured size honestly (previous commit) exposed it, and configuring anything above 8 KiB exposed it even before that. The regression test mixes frame sizes on purpose — a run of full-size frames never reproduces it, because each one lands the buffer above the mark and gets flushed on its own, leaving nothing to coalesce with. --- misc/webrtc-utils/CHANGELOG.md | 1 + misc/webrtc-utils/src/stream/framed_dc.rs | 124 +++++++++++++++++++++- 2 files changed, 122 insertions(+), 3 deletions(-) diff --git a/misc/webrtc-utils/CHANGELOG.md b/misc/webrtc-utils/CHANGELOG.md index 377eebbf549..96e8fcd2893 100644 --- a/misc/webrtc-utils/CHANGELOG.md +++ b/misc/webrtc-utils/CHANGELOG.md @@ -3,6 +3,7 @@ - Negotiate WebRTC data-channel message limits after Noise authentication. `sdp::answer` and `sdp::render_description` now take the `StreamConfig` they advertise `a=max-message-size` from, instead of always announcing 16 KiB. + Send one frame per data-channel message, so no write can exceed the negotiated limit. - Revert migration to `quick-protobuf`, migrate back to `prost`. See [PR 6363](https://github.com/libp2p/rust-libp2p/pull/6363). diff --git a/misc/webrtc-utils/src/stream/framed_dc.rs b/misc/webrtc-utils/src/stream/framed_dc.rs index 5e040416623..71d9f940672 100644 --- a/misc/webrtc-utils/src/stream/framed_dc.rs +++ b/misc/webrtc-utils/src/stream/framed_dc.rs @@ -18,6 +18,8 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. +use std::num::NonZeroUsize; + use asynchronous_codec::Framed; use futures::{AsyncRead, AsyncWrite}; @@ -32,12 +34,128 @@ where T: AsyncRead + AsyncWrite, { let mut framed = Framed::new(inner, codec(config)); - // If not set, `Framed` buffers up to 131kB of data before sending, which leads to "outbound - // packet larger than maximum message size" error in webrtc-rs. - framed.set_send_high_water_mark(config.max_data_size()); + // One encoded frame per write, because the layer below turns every write into exactly one + // SCTP user message — and that message must not exceed `max_message_size`. + // + // The high-water mark is a *lower* bound on when to flush, not an upper bound on the + // buffer: `poll_ready` flushes while `buffer.len() >= hwm` and `start_send` then appends a + // whole frame, so the buffer peaks at `hwm - 1 + one frame`. Any `hwm` above 1 therefore + // lets two frames coalesce into a single message of up to `2 * max_message_size`, which + // webrtc-rs rejects with "outbound packet larger than maximum message size" — silently + // losing that frame. + // + // This used to be `config.max_data_size()`, which only ever worked because the SDP always + // advertised a hard-coded 16 KiB while the framing layer used 8 KiB: the doubled buffer + // landed exactly on the advertised limit. Raising the configured size, or advertising the + // configured size honestly, broke it in both directions. + // + // Sending one frame per message is also what the spec describes; coalescing happened to + // decode correctly only because each frame carries its own length prefix. + framed.set_send_high_water_mark(1); framed } pub(crate) fn codec(config: StreamConfig) -> prost_codec::Codec { prost_codec::Codec::new(config.max_message_size() - VARINT_LEN) } + +#[cfg(test)] +mod tests { + use std::{ + io, + pin::Pin, + sync::{Arc, Mutex}, + task::{Context, Poll}, + }; + + use futures::{AsyncRead, AsyncWrite, SinkExt}; + + use super::*; + + /// Records the length of every individual write. + /// + /// The layer this sits on top of in production (`PollDataChannel`) turns one write into one + /// SCTP user message, so these lengths *are* the message sizes the peer's SCTP will police. + #[derive(Clone, Default)] + struct RecordingWriter(Arc>>); + + impl AsyncRead for RecordingWriter { + fn poll_read( + self: Pin<&mut Self>, + _: &mut Context<'_>, + _: &mut [u8], + ) -> Poll> { + Poll::Ready(Ok(0)) + } + } + + impl AsyncWrite for RecordingWriter { + fn poll_write( + self: Pin<&mut Self>, + _: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + self.0.lock().unwrap().push(buf.len()); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + + /// No single write may exceed `max_message_size`, whatever mix of frame sizes is queued. + /// + /// **The sizes must be mixed.** A run of full-size frames alone never reproduces the bug: + /// each one lands the buffer above the high-water mark, so the next `poll_ready` flushes it + /// and nothing is left behind. It takes a *small* frame — one that leaves the buffer below + /// the mark — followed by a full-size one for the two to be written out together. Measured + /// on a real transfer before the fix: 125 writes of 8190 B and three of **8419 B** against + /// an 8192 B limit; SCTP rejected exactly those three and the stream lost them. + #[test] + fn no_write_exceeds_the_configured_message_size() { + for bytes in [8 * 1024usize, 16 * 1024, 64 * 1024] { + let config = StreamConfig::new(NonZeroUsize::new(bytes).expect("non-zero")); + let writer = RecordingWriter::default(); + let mut framed = new(writer.clone(), config); + + futures::executor::block_on(async { + // `feed`, not `send`: `send` flushes after every item, which empties the + // buffer between frames and hides the coalescing entirely. The production + // path (`Stream::poll_write`) does not flush per frame either. + for _ in 0..4 { + // A short frame first: it leaves the buffer below the high-water mark. + framed + .feed(Message { + flag: Some(0), + message: Some(vec![0u8; 200]), + }) + .await + .expect("feed"); + framed + .feed(Message { + flag: None, + message: Some(vec![0u8; config.max_data_size()]), + }) + .await + .expect("feed"); + } + framed.close().await.expect("close"); + }); + + let writes = writer.0.lock().unwrap().clone(); + assert!(!writes.is_empty(), "nothing was written"); + for len in writes { + assert!( + len <= bytes, + "a {len} B write exceeds the {bytes} B limit; SCTP would reject it and the \ + frame would be lost" + ); + } + } + } +}