feat(webrtc): negotiate data channel message limits

misc/webrtc-utils/CHANGELOG.md

@@ -1,5 +1,10 @@
 ## 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.
+  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).
 

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

@@ -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");
@@ -91,7 +97,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,14 +120,22 @@ 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.
+///
+/// `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();
@@ -141,6 +155,7 @@ pub fn render_description(
         // NOTE: ufrag is equal to pwd.
         ufrag: ufrag.to_owned(),
         pwd: ufrag.to_owned(),
+        max_message_size: config.max_message_size(),
     };
     tt.render("description", &context).unwrap()
 }
@@ -156,3 +171,49 @@ pub fn random_ufrag() -> String {
             .collect::<String>()
     )
 }
+
+#[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}");
+    }
+}