Negotiate SCTP max message size (algesten#852)

* Init SCTP with max msg size constants

* Extract and apply max-message-size from SDP

* Consume self in SdpApi::accept_offer

* Use correct const for max_receive_message_size

LOCAL_MAX_MESSAGE_SIZE is what we are willing to
receive.

DEFAULT_REMOTE_MAX_MESSAGE_SIZE is what the remote
is willing to receive, hence we are able to send.

* Set remote_max_message_size in SCTP initialization

* Set max_send_message_size on newly connected Association

* Fix linter issues

* cargo fmt

* Use upstream sctp-proto at a fixed commit

* Revert "cargo fmt"

This reverts commit d54639b.

* cargo fmt

* Revert unrelated changes in apply_offer & apply_answer

* Revert formatting changes in sdp/data.rs

* fix formatting even more

* clean up imports

* correctly handle zero value for max_message_size

* set max send message size in SNAP init case

* Fix max-message-size lint

* Add changelog entry for max-message-size

* Resolve changelog conflict for max-message-size

---------

Co-authored-by: Martin Algesten <martin@algesten.se>

Author: haikoschol

CHANGELOG.md

@@ -1,5 +1,6 @@
 # Unreleased
 
+  * Negotiate SCTP max message size for data channels #852
   * Expose the negotiated DTLS protocol version #979
   * Make MTU size configurable #967
   * Add optional `drv` feature for using public identity types as `drv::Input` #974 #975

src/change/direct.rs

@@ -126,7 +126,7 @@ impl<'a> DirectApi<'a> {
     /// connecting party.
     pub fn start_sctp(&mut self, client: bool) {
         self.rtc
-            .try_init_sctp(client, None)
+            .try_init_sctp(client, None, None)
             .expect("starting SCTP should be infallible")
     }
 
@@ -162,7 +162,7 @@ impl<'a> DirectApi<'a> {
         client: bool,
         sctp_init_data: SctpInitData,
     ) -> Result<(), RtcError> {
-        self.rtc.try_init_sctp(client, Some(sctp_init_data))
+        self.rtc.try_init_sctp(client, Some(sctp_init_data), None)
     }
 
     /// Create a new data channel.

src/change/sdp.rs

@@ -2,6 +2,7 @@
 
 use std::fmt;
 use std::ops::{Deref, DerefMut};
+use std::slice::Iter;
 
 use crate::Rtc;
 use crate::RtcError;
@@ -100,6 +101,8 @@ impl<'a> SdpApi<'a> {
         // Ensure setup=active/passive is corresponding remote and init dtls.
         init_dtls(self.rtc, &offer)?;
 
+        let remote_max_message_size = extract_max_message_size(offer.media_lines.iter());
+
         // Extract a=sctp-init from remote offer before apply_offer consumes it.
         let remote_sctp_init = offer.sctp_init().map(|v| v.to_owned());
 
@@ -119,7 +122,8 @@ impl<'a> SdpApi<'a> {
 
         if self.rtc.session.app().is_some() {
             let init_data = self.rtc.sctp.build_snap_init_data();
-            self.rtc.try_init_sctp(client, init_data)?;
+            self.rtc
+                .try_init_sctp(client, init_data, remote_max_message_size)?;
         }
 
         let params = AsSdpParams::new(self.rtc, None);
@@ -198,6 +202,8 @@ impl<'a> SdpApi<'a> {
         // Split out new channels, since that is not handled by the Session.
         let new_channels = pending.changes.take_new_channels();
 
+        let remote_max_message_size = extract_max_message_size(answer.media_lines.iter());
+
         // Modify session with answer
         apply_answer(&mut self.rtc.session, pending.changes, answer)?;
 
@@ -211,7 +217,8 @@ impl<'a> SdpApi<'a> {
 
         if self.rtc.session.app().is_some() {
             let init_data = self.rtc.sctp.build_snap_init_data();
-            self.rtc.try_init_sctp(client, init_data)?;
+            self.rtc
+                .try_init_sctp(client, init_data, remote_max_message_size)?;
         }
 
         for (id, config) in new_channels {
@@ -1392,6 +1399,23 @@ fn update_media(
     }
 }
 
+fn extract_max_message_size(mut media_lines: Iter<MediaLine>) -> Option<u32> {
+    if let Some(app_line) = media_lines.find(|m| m.typ.is_channel()) {
+        if let Some(max_size) = app_line.max_message_size() {
+            // RFC 8841 §6.1: a value of 0 means the peer can receive a message of
+            // any size, subject to local capacity. Represent that as the maximum.
+            let max_size = if max_size == 0 {
+                u32::MAX
+            } else {
+                u32::try_from(max_size).unwrap_or(u32::MAX)
+            };
+            return Some(max_size);
+        }
+    }
+
+    None
+}
+
 trait AsSdpMediaLine {
     fn mid(&self) -> Mid;
     fn msid(&self) -> Option<&Msid>;
@@ -1428,7 +1452,9 @@ impl AsSdpMediaLine for (Mid, usize) {
     ) -> MediaLine {
         attrs.push(MediaAttribute::Mid(self.0));
         attrs.push(MediaAttribute::SctpPort(5000));
-        attrs.push(MediaAttribute::MaxMessageSize(262144));
+        attrs.push(MediaAttribute::MaxMessageSize(
+            crate::sctp::LOCAL_MAX_MESSAGE_SIZE as usize,
+        ));
 
         MediaLine {
             typ: sdp::MediaType::Application,
@@ -2477,4 +2503,197 @@ mod test {
         // No space at the end
         assert_eq!(count_lines(&line_string, "a=rid:no_attrs send"), 1);
     }
+
+    #[test]
+    fn test_local_max_message_size_advertised() {
+        crate::init_crypto_default();
+
+        let now = Instant::now();
+        let mut rtc = Rtc::new(now);
+
+        // Create an offer with a data channel
+        let mut change = rtc.sdp_api();
+        change.add_channel("test-channel".into());
+        let (offer, _) = change.apply().unwrap();
+
+        // Find the application m-line in the offer
+        let app_line = offer
+            .media_lines
+            .iter()
+            .find(|m| m.typ.is_channel())
+            .expect("should have application m-line");
+
+        // Verify that max-message-size attribute is present and matches LOCAL_MAX_MESSAGE_SIZE
+        let max_size = app_line
+            .max_message_size()
+            .expect("max-message-size attribute should be present");
+
+        assert_eq!(
+            max_size,
+            crate::sctp::LOCAL_MAX_MESSAGE_SIZE as usize,
+            "max-message-size should match LOCAL_MAX_MESSAGE_SIZE constant"
+        );
+
+        // Also verify it's in the SDP string output
+        let sdp_string = offer.to_sdp_string();
+        let expected_line = format!("a=max-message-size:{}", crate::sctp::LOCAL_MAX_MESSAGE_SIZE);
+        assert!(
+            sdp_string.contains(&expected_line),
+            "SDP should contain max-message-size attribute with LOCAL_MAX_MESSAGE_SIZE value"
+        );
+    }
+
+    #[test]
+    fn test_remote_max_message_size_parsing() {
+        // Parse SDP with max-message-size attribute and verify value is extracted correctly
+        let sdp = "v=0\r\n\
+            o=- 0 0 IN IP4 172.17.0.1\r\n\
+            s=-\r\n\
+            c=IN IP4 172.17.0.1\r\n\
+            t=0 0\r\n\
+            a=group:BUNDLE 0\r\n\
+            a=fingerprint:sha-256 B4:12:1C:7C:7D:ED:F1:FA:61:07:57:9C:29:BE:58:E3:BC:41:E7:13:8E:7D\
+            :D3:9D:1F:94:6E:A5:23:46:94:23\r\n\
+            m=application 9999 UDP/DTLS/SCTP webrtc-datachannel\r\n\
+            a=mid:0\r\n\
+            a=ice-ufrag:test\r\n\
+            a=ice-pwd:testpassword1234\r\n\
+            a=setup:actpass\r\n\
+            a=sctp-port:5000\r\n\
+            a=max-message-size:131072\r\n\
+            ";
+
+        let offer = SdpOffer::from_sdp_string(sdp).expect("should parse");
+
+        // Find the application m-line
+        let app_line = offer
+            .media_lines
+            .iter()
+            .find(|m| m.typ.is_channel())
+            .expect("should have application m-line");
+
+        assert_eq!(
+            app_line.max_message_size(),
+            Some(131072),
+            "max-message-size should be parsed as 131072"
+        );
+    }
+
+    #[test]
+    fn test_remote_max_message_size_applied() {
+        crate::init_crypto_default();
+
+        let now = Instant::now();
+        let mut rtc1 = Rtc::new(now);
+        let mut rtc2 = Rtc::new(now);
+
+        // Create an offer from rtc1 with a channel
+        let mut change1 = rtc1.sdp_api();
+        change1.add_channel("test-channel".into());
+        let (offer1, pending1) = change1.apply().unwrap();
+
+        // Get the offer SDP string and modify it to have a custom max-message-size
+        let custom_max_size1 = 98304u32;
+        let sdp_string = offer1.to_sdp_string();
+        let modified_sdp = sdp_string.replace(
+            &format!("a=max-message-size:{}", crate::sctp::LOCAL_MAX_MESSAGE_SIZE),
+            &format!("a=max-message-size:{}", custom_max_size1),
+        );
+
+        let modified_offer =
+            SdpOffer::from_sdp_string(&modified_sdp).expect("modified SDP should parse");
+
+        let answer = rtc2.sdp_api().accept_offer(modified_offer).unwrap();
+
+        // Verify that rtc2's SCTP send limit is set to the custom value from rtc1's offer
+        assert_eq!(
+            rtc2.sctp.remote_max_message_size(),
+            custom_max_size1,
+            "rtc2 should have remote max message size set to rtc1's advertised value"
+        );
+
+        // Now verify the reverse: rtc1 accepts rtc2's answer and applies its max-message-size
+        let custom_max_size2 = 131072u32;
+        let sdp_string = answer.to_sdp_string();
+        let modified_sdp = sdp_string.replace(
+            &format!("a=max-message-size:{}", crate::sctp::LOCAL_MAX_MESSAGE_SIZE),
+            &format!("a=max-message-size:{}", custom_max_size2),
+        );
+
+        let modified_answer =
+            SdpAnswer::from_sdp_string(&modified_sdp).expect("modified SDP should parse");
+
+        rtc1.sdp_api()
+            .accept_answer(pending1, modified_answer)
+            .unwrap();
+
+        assert_eq!(
+            rtc1.sctp.remote_max_message_size(),
+            custom_max_size2,
+            "rtc1 should have remote max message size set to rtc2's advertised value"
+        );
+    }
+
+    #[test]
+    fn test_remote_max_message_size_zero_means_unbounded() {
+        // RFC 8841 §6.1: a=max-message-size:0 means the peer can receive a message
+        // of any size, so we represent it as u32::MAX rather than passing 0 through
+        // (which would reject every non-empty outbound message in sctp-proto).
+        crate::init_crypto_default();
+
+        let now = Instant::now();
+        let mut rtc1 = Rtc::new(now);
+        let mut rtc2 = Rtc::new(now);
+
+        let mut change1 = rtc1.sdp_api();
+        change1.add_channel("test-channel".into());
+        let (offer1, _pending1) = change1.apply().unwrap();
+
+        let sdp_string = offer1.to_sdp_string();
+        let modified_sdp = sdp_string.replace(
+            &format!("a=max-message-size:{}", crate::sctp::LOCAL_MAX_MESSAGE_SIZE),
+            "a=max-message-size:0",
+        );
+
+        let modified_offer =
+            SdpOffer::from_sdp_string(&modified_sdp).expect("modified SDP should parse");
+
+        rtc2.sdp_api().accept_offer(modified_offer).unwrap();
+
+        assert_eq!(
+            rtc2.sctp.remote_max_message_size(),
+            u32::MAX,
+            "a=max-message-size:0 should be treated as unbounded (u32::MAX)"
+        );
+    }
+
+    #[test]
+    fn test_remote_max_message_size_too_large_clamps_to_unbounded() {
+        crate::init_crypto_default();
+
+        let now = Instant::now();
+        let mut rtc1 = Rtc::new(now);
+        let mut rtc2 = Rtc::new(now);
+
+        let mut change1 = rtc1.sdp_api();
+        change1.add_channel("test-channel".into());
+        let (offer1, _pending1) = change1.apply().unwrap();
+
+        let sdp_string = offer1.to_sdp_string();
+        let modified_sdp = sdp_string.replace(
+            &format!("a=max-message-size:{}", crate::sctp::LOCAL_MAX_MESSAGE_SIZE),
+            "a=max-message-size:4294967296",
+        );
+
+        let modified_offer =
+            SdpOffer::from_sdp_string(&modified_sdp).expect("modified SDP should parse");
+
+        rtc2.sdp_api().accept_offer(modified_offer).unwrap();
+
+        assert_eq!(
+            rtc2.sctp.remote_max_message_size(),
+            u32::MAX,
+            "a=max-message-size larger than u32::MAX should be treated as unbounded"
+        );
+    }
 }

src/lib.rs

@@ -1480,14 +1480,20 @@ impl Rtc {
         &mut self,
         client: bool,
         sctp_init_data: Option<SctpInitData>,
+        remote_max_message_size: Option<u32>,
     ) -> Result<(), RtcError> {
         // If we got an m=application line, ensure we have negotiated the
         // SCTP association with the other side.
         if self.sctp.is_inited() {
             return Ok(());
         }
 
-        self.sctp.init(client, self.last_now, sctp_init_data)?;
+        self.sctp.init(
+            client,
+            self.last_now,
+            sctp_init_data,
+            remote_max_message_size,
+        )?;
         Ok(())
     }