feat(webrtc): negotiate data channel message limits

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).
 

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;

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,14 +30,39 @@ 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<T>(
     id_keys: identity::Keypair,
     stream: T,
     client_fingerprint: Fingerprint,
     server_fingerprint: Fingerprint,
 ) -> Result<PeerId, Error>
+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<T>(
+    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<T>(
@@ -58,6 +86,28 @@ pub async fn outbound<T>(
     server_fingerprint: Fingerprint,
     client_fingerprint: Fingerprint,
 ) -> Result<PeerId, Error>
+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<T>(
+    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<T>(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::<u64>()];
+    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<u64>) -> 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
+        );
+    }
 }

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()
 }

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: <https://www.rfc-editor.org/rfc/rfc8831#name-transferring-user-data-on-a>
-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: <https://www.rfc-editor.org/rfc/rfc8831#name-transferring-user-data-on-a>
+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<T> {
     io: FramedDc<T>,
     state: State,
     read_buffer: Bytes,
+    config: StreamConfig,
     /// Dropping this will close the oneshot and notify the receiver by emitting `Canceled`.
     drop_notifier: Option<oneshot::Sender<GracefullyClosed>>,
 }
@@ -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<T>) {
+        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<T>) {
         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());
     }
 }