fix(webrtc): reject malformed fragment framing, correct receive-path docs

`next()` read every non-FRAG_END header as "more fragments", so a peer whose
framing had diverged was only caught by the MAX_FRAME_LENGTH cap — and a
FRAG_MORE carrying no payload was never caught at all: it adds nothing to the
accumulator, so the cap never trips and the loop spins for as long as the peer
keeps writing, with no error and no teardown. Decide the header's meaning in one
match, so a future header kind cannot be handled in one place and missed in the
other. Neither case is reachable from send_bytes_inner, which emits FRAG_MORE
only for a full MAX_FRAGMENT_PAYLOAD chunk.

Release the accumulator on the error paths rather than truncating it: at the cap
that is ~1 GiB still referenced through the SESSIONS clone.

Doc corrections, all of them overclaims in the previous pass:

- the cancel-safety entry held only for the successful read path.
  read_data_channel does await after dequeuing on its ErrShortBuffer and DCEP
  branches, and next() awaits pc.close() on its error paths — where
  RTCPeerConnection::close latches is_closed before its first await, so a
  cancelled close silently turns every later close into a no-op and leaves the
  pc in SESSIONS.
- recv_state: cancellation drops the guard mid-message, so it is the
  single-reader assumption, not the mutex, that ultimately keeps two readers
  from splicing into one accumulator.
- is_relayed: stream.rs promised None before pair selection while webrtc.rs
  documented Some(true) under Relay policy; align both.
- get_local_endpoint: examples/webrtc.rs calls it too, not only the tests.
- PunchHole.reserved 11: named the wrong writer — PunchHole is written by the
  rendezvous server, not by peers. Reserve the name as well as the tag.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

protos/rendezvous.proto

@@ -66,7 +66,11 @@ message PunchHole {
   ControlPermissions control_permissions = 8;
   ControlledContext controlled_context = 9;
   string webrtc_sdp_offer = 10;
+  // Was `string requester_id = 11`, dropped once ICE routing stopped needing it. Reserved so the
+  // tag is never reassigned: PunchHole is written by the rendezvous server, and an hbbs built
+  // against the earlier field still puts a string here.
   reserved 11;
+  reserved "requester_id";
 }
 
 message TestNatRequest {

src/stream.rs

@@ -101,7 +101,9 @@ impl Stream {
     }
 
     /// Whether an established WebRTC transport runs through a TURN relay (used for the UI's
-    /// direct/relayed flag). None for non-WebRTC transports or before ICE selects a pair.
+    /// direct/relayed flag). `None` for non-WebRTC transports, and for a non-relay-policy pc
+    /// before ICE selects a pair; a Relay-policy pc answers `Some(true)` straight away — see
+    /// `WebRTCStream::is_relayed`.
     #[inline]
     pub async fn webrtc_relayed(&self) -> Option<bool> {
         match self {

src/webrtc.rs

@@ -14,6 +14,14 @@
 //!   send_timeout-then-close semantics. If a new version buffers unboundedly instead, video can
 //!   OOM a slow session and the send timeout never fires.
 //! - **Max SCTP message size 65536**: `MAX_FRAGMENT_PAYLOAD` + 1 header byte must stay below it.
+//! - **The successful read path is cancel-safe**: `read_sctp` dequeues synchronously and returns
+//!   with no `.await` after it, and `read_data_channel` adds none on the user-data path. So
+//!   `next_timeout`, which drops that future routinely, cannot lose a fragment — a version that
+//!   awaited after dequeuing would, undetectably, since the header carries no length or checksum.
+//!   Scope: the *read*. `read_data_channel` does await after dequeuing on its ErrShortBuffer and
+//!   DCEP branches, and `next()` itself awaits `pc.close()` on its error paths — and
+//!   `RTCPeerConnection::close` latches `is_closed` before its first await, so a cancelled close
+//!   silently makes every later close a no-op and leaves the pc in `SESSIONS`.
 //! - **`detach()` is an idempotent Arc clone with no close-on-drop** (`detached_dc` caches it and
 //!   clones are shared across `WebRTCStream` clones).
 //! - **`on_*` handlers are stored inside the pc**: a handler capturing a strong
@@ -84,9 +92,11 @@ pub struct WebRTCStream {
     // Serialize a complete logical message across clones. Each fragment is a separate SCTP
     // message, so serializing only individual writes would allow two large messages to interleave.
     send_gate: Arc<Semaphore>,
-    // Receive-side reassembly state, guarded by a single mutex so the fragment accumulator
+    // Receive-side reassembly state. The accumulator survives a cancelled `next()` because it
-    // survives `next()` cancellation (e.g. `next_timeout`) instead of losing already-read
+    // lives here behind the Arc, not in the future. The mutex excludes a concurrent reader only
-    // fragments mid-message. Assumes a single reader, consistent with the rest of the stream API.
+    // while `next()` is actually running — a cancellation drops the guard mid-message, so it is
+    // the single-reader assumption, not the lock, that ultimately prevents two readers splicing
+    // into one `acc`. Single reader assumed, like the rest of the stream API.
     recv_state: Arc<Mutex<RecvState>>,
     // True once the controller has completed the RustDesk identity binding (DTLS fingerprint
     // matched to the signed peer id, via `set_key`). DTLS always encrypts; this flag mirrors TCP's
@@ -501,16 +511,12 @@ impl WebRTCStream {
 
         // process offer/answer
         //
-        // Trickle ICE: the local description is returned WITHOUT waiting for candidate gathering
+        // Trickle ICE: this block is local-only work (pc construction, DTLS keygen, SDP marshal),
-        // (candidates stream out via `take_local_ice_rx` afterwards), so this block is local-only
+        // no gathering wait. The controlled side awaits answer creation inline on its punch-reply
-        // work — pc construction, DTLS cert keygen, SDP marshal — at sub-millisecond cost. The
+        // critical path, so adding a network wait here would delay every hole punch.
-        // controlled side awaits answer creation inline on its punch-reply critical path and
+        // A failure below leaves a live pc whose state handler only fires on a terminal ICE state,
-        // relies on that: adding any gathering/network wait here would delay the TCP/UDP
+        // so a bare `?`-drop leaks it — remotely triggerable via a crafted `type:"answer"` offer
-        // hole-punch reply for every connection.
+        // that passes the pre-check but fails `set_remote_description`. Close before propagating.
-        // Any failure below leaves a live pc with handlers already registered; its state handler
-        // only fires on a terminal ICE state, so a bare `?`-drop would leak it (remotely
-        // triggerable: a crafted `type:"answer"` offer passes the JSON+fingerprint pre-check but
-        // fails `set_remote_description`). Close the pc before propagating any such error.
         let offer_answer: ResultType<String> = async {
             if start_local_offer {
                 let sdp = pc.create_offer(None).await?;
@@ -575,6 +581,11 @@ impl WebRTCStream {
         Ok(webrtc_stream)
     }
 
+    /// One-shot endpoint: waits for ICE gathering so the SDP already carries the candidates.
+    /// The wait is deliberately unbounded — a deadline here would return a half-gathered SDP and
+    /// defeat the contract; callers needing one should wrap the call or use the trickle variant,
+    /// which both rustdesk signaling paths do. Remaining callers are the loopback tests and
+    /// `examples/webrtc.rs`, neither of which bounds it.
     #[inline]
     pub async fn get_local_endpoint(&self) -> ResultType<String> {
         // Preserve the original one-shot endpoint contract: callers that only exchange this SDP
@@ -621,9 +632,11 @@ impl WebRTCStream {
         Self::get_key_for_peer(&self.pc, false).await
     }
 
-    /// Whether the established connection runs through a TURN relay: `Some(true)` when the pc is
+    /// Whether the connection runs through a TURN relay; feeds the UI's direct/relayed flag.
-    /// Relay-policy (TURN is the only possibility) or the selected ICE candidate pair uses a
+    /// Under Relay policy the answer is known by construction, so that arm returns `Some(true)`
-    /// relay candidate; `None` before a pair is selected. Feeds the UI's direct/relayed flag.
+    /// without consulting ICE — including before a pair is selected, unlike the `None` the other
+    /// arm returns then. Both callers ask post-connection, where the arms agree; making the relay
+    /// arm await a pair would only add a stats round trip.
     pub async fn is_relayed(&self) -> Option<bool> {
         if self.relay_only {
             return Some(true);
@@ -685,14 +698,10 @@ impl WebRTCStream {
         Ok(())
     }
 
-    /// Explicitly tear down the peer connection.
+    /// Explicitly tear down the peer connection. Dropping the handle is not enough: `SESSIONS`
-    ///
+    /// holds a clone, so the pc and its ICE/DTLS resources survive until it happens to reach a
-    /// Dropping a `WebRTCStream` handle is not enough to release the underlying
+    /// terminal ICE state. Closing fires the state handler, which evicts the `SESSIONS` entry —
-    /// `RTCPeerConnection`: the global `SESSIONS` map holds a clone, so the pc (and its
+    /// callers that abandon a stream (e.g. an offerer that lost the transport race) must call it.
-    /// ICE/DTLS/STUN resources) would stay alive until it happens to reach a terminal ICE
-    /// state. Closing here fires `on_peer_connection_state_change`, which removes the
-    /// `SESSIONS` entry, so callers that abandon a stream (e.g. a raced offerer that lost to
-    /// another transport) should call this to avoid leaking it.
     #[inline]
     pub async fn close(&self) {
         self.pc.close().await.ok();
@@ -778,13 +787,11 @@ impl WebRTCStream {
     pub async fn send_bytes(&mut self, bytes: Bytes) -> ResultType<()> {
         let send_timeout = self.send_timeout;
         let send_gate = self.send_gate.clone();
-        // Bound the WHOLE data-channel send (wait-for-open + every write) by send_timeout,
+        // Bound the WHOLE send (wait-for-open, queueing behind another clone, every write) by
-        // including time queued behind another clone. Without this a write can park indefinitely
+        // send_timeout: otherwise a write parks indefinitely on SCTP backpressure and
-        // on SCTP pending-queue backpressure and connection.rs's timeout timer never runs.
+        // connection.rs's timer never runs. That parking is also what bounds sender memory —
-        // That parking is also what bounds sender memory: webrtc-sctp's PendingQueue admits at
+        // PendingQueue admits 128 KiB and inflight is cwnd/rwnd-capped — so this timeout is the
-        // most 128 KiB (byte-counting semaphore) and inflight data is cwnd/rwnd-capped, so a slow
+        // TCP-send-timeout equivalent. See the checklist entry on send backpressure.
-        // link parks the write here until this timeout closes the pc — TCP-send-timeout
-        // equivalent. Verified against webrtc-sctp 0.12; see the module-level upgrade checklist.
         if send_timeout > 0 {
             let deadline = Instant::now() + Duration::from_millis(send_timeout);
             let _send_permit = match timeout_at(deadline, send_gate.acquire_owned()).await {
@@ -865,8 +872,9 @@ impl WebRTCStream {
                 return Some(Err(Error::new(ErrorKind::Other, err.to_string())));
             }
         };
-        // Hold recv_state across the reassembly loop: the accumulator must survive `next()`
+        // Held across the whole loop for exclusion, not for accumulator survival (see the field).
-        // cancellation (e.g. next_timeout) so already-read fragments are not lost mid-message.
+        // Cancelling mid-`dc.read()` loses no data; cancelling mid-`pc.close()` on an error path
+        // below does leak the pc — see the checklist entry on cancel-safety.
         let mut st = self.recv_state.lock().await;
         if st.scratch.len() < RECV_BUF_SIZE {
             st.scratch.resize(RECV_BUF_SIZE, 0);
@@ -889,18 +897,41 @@ impl WebRTCStream {
                 self.pc.close().await.ok();
                 return None;
             }
+            // Two framing violations, both of which would otherwise be read as "more fragments":
+            // an unrecognized header, and a FRAG_MORE carrying no payload. The latter is the
+            // nastier one — it adds nothing to `acc`, so the MAX_FRAME_LENGTH cap below never
+            // trips and the loop spins for as long as the peer keeps sending. `send_bytes_inner`
+            // emits FRAG_MORE only for a full MAX_FRAGMENT_PAYLOAD chunk, so neither is reachable
+            // from our own sender.
+            let header = scratch[0];
+            let bad = match header {
+                FRAG_END => None,
+                FRAG_MORE if n > 1 => None,
+                FRAG_MORE => Some("FRAG_MORE fragment carries no payload".to_owned()),
+                other => Some(format!("fragment header {other} is neither FRAG_END nor FRAG_MORE")),
+            };
+            if let Some(why) = bad {
+                *acc = BytesMut::new();
+                self.pc.close().await.ok();
+                return Some(Err(Error::new(
+                    ErrorKind::InvalidData,
+                    format!("WebRTC {why}"),
+                )));
+            }
             acc.extend_from_slice(&scratch[1..n]);
             // Match TCP's maximum frame size while preventing an unbounded FRAG_MORE stream from
             // exhausting memory.
             if acc.len() > MAX_FRAME_LENGTH {
-                acc.clear();
+                // Release the buffer, don't just truncate it: by definition it is at the cap here,
+                // and `recv_state` outlives this call through the `SESSIONS` clone.
+                *acc = BytesMut::new();
                 self.pc.close().await.ok();
                 return Some(Err(Error::new(
-                    ErrorKind::Other,
+                    ErrorKind::InvalidData,
                     "WebRTC reassembled message exceeded maximum frame size",
                 )));
             }
-            if scratch[0] == FRAG_END {
+            if header == FRAG_END {
                 let msg = std::mem::take(acc);
                 return Some(Ok(msg));
             }
@@ -925,7 +956,8 @@ pub fn is_webrtc_endpoint(endpoint: &str) -> bool {
 mod tests {
     use crate::config;
     use crate::webrtc::WebRTCStream;
-    use crate::webrtc::DEFAULT_ICE_SERVERS;
+    use crate::webrtc::{DEFAULT_ICE_SERVERS, FRAG_MORE};
+    use bytes::{BufMut, BytesMut};
     use std::{sync::Arc, time::Duration};
     use tokio::sync::Barrier;
     use tokio::time::timeout;
@@ -1284,6 +1316,48 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc
             .expect("webrtc loopback did not complete in time");
     }
 
+    // Both framing violations must end the stream. The empty-FRAG_MORE case is the one that
+    // cannot be caught downstream: it adds nothing to the accumulator, so the MAX_FRAME_LENGTH
+    // cap never trips and `next()` would otherwise spin for as long as the peer keeps writing.
+    // The bad frame is injected mid-message so the branch's `acc` reset is exercised too.
+    #[tokio::test]
+    async fn test_webrtc_rejects_bad_fragment_framing() {
+        // (raw frame, expected substring)
+        let cases: [(&[u8], &str); 2] = [
+            (&[0x2A, b'x'], "fragment header 42"),
+            (&[FRAG_MORE], "carries no payload"),
+        ];
+        for (frame, want) in cases {
+            let connect = async {
+                let (offerer, mut answerer) = connect_loopback().await;
+                let dc = offerer.detached_dc().await.unwrap();
+
+                // Leave a partial message in the accumulator first, so the rejection has
+                // something to discard. `send_bytes` only ever emits valid headers, so the bad
+                // frame itself has to be written through the raw channel below it.
+                let mut lead = BytesMut::with_capacity(1 + 8);
+                lead.put_u8(FRAG_MORE);
+                lead.put_slice(b"leading!");
+                dc.write(&lead.freeze()).await.unwrap();
+                dc.write(&bytes::Bytes::copy_from_slice(frame)).await.unwrap();
+
+                let err = answerer
+                    .next()
+                    .await
+                    .expect("bad framing must surface as an error, not EOF")
+                    .expect_err("bad framing must be rejected");
+                assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+                assert!(err.to_string().contains(want), "unexpected error: {}", err);
+
+                offerer.close().await;
+                answerer.close().await;
+            };
+            timeout(Duration::from_secs(40), connect)
+                .await
+                .expect("webrtc loopback did not complete in time");
+        }
+    }
+
     #[tokio::test]
     async fn test_webrtc_concurrent_large_sends_preserve_boundaries() {
         let connect = async {