diff --git a/Cargo.toml b/Cargo.toml index 6ad3012fa..6010c0a84 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,7 +65,7 @@ rustls-pki-types = "1.11" rustls-native-certs = "0.8" webpki-roots = "1.0.4" async-recursion = "1.1" -webrtc = { version = "0.14.0", optional = true } +webrtc = { version = "0.13.0", optional = true } libloading = "0.8" [target.'cfg(not(any(target_os = "android", target_os = "ios")))'.dependencies] @@ -78,7 +78,7 @@ protobuf-codegen = { version = "3.7" } [dev-dependencies] clap = "4.5.51" -webrtc = "0.14.0" +webrtc = "0.13.0" [target.'cfg(target_os = "windows")'.dependencies] winapi = { version = "0.3", features = [ diff --git a/protos/message.proto b/protos/message.proto index 8b2136811..2024634cf 100644 --- a/protos/message.proto +++ b/protos/message.proto @@ -38,6 +38,10 @@ message VideoFrame { message IdPk { string id = 1; bytes pk = 2; + // DTLS certificate fingerprint of the signer's WebRTC endpoint, signed together with id/pk so + // a WebRTC peer's DTLS channel can be bound to its verified identity (defeats a rendezvous/relay + // that swaps SDP fingerprints). Empty for non-WebRTC handshakes. + string dtls_fingerprint = 3; } message DisplayInfo { diff --git a/src/bytes_codec.rs b/src/bytes_codec.rs index cbd53d918..5bd5590c1 100644 --- a/src/bytes_codec.rs +++ b/src/bytes_codec.rs @@ -4,6 +4,8 @@ use tokio_util::codec::{Decoder, Encoder}; // Bound speculative allocation from untrusted frame headers. const MAX_PREALLOCATED_PAYLOAD_LEN: usize = 256 * 1024; +/// Largest payload representable by the four-byte RustDesk frame header. +pub const MAX_FRAME_LENGTH: usize = 0x3FFF_FFFF; #[derive(Debug, Clone, Copy)] pub struct BytesCodec { @@ -132,7 +134,7 @@ impl Encoder for BytesCodec { let h = (data.len() << 2) as u32 | 0x2; buf.put_u16_le((h & 0xFFFF) as u16); buf.put_u8((h >> 16) as u8); - } else if data.len() <= 0x3FFFFFFF { + } else if data.len() <= MAX_FRAME_LENGTH { buf.put_u32_le((data.len() << 2) as u32 | 0x3); } else { return Err(io::Error::new(io::ErrorKind::InvalidInput, "Overflow")); @@ -290,7 +292,7 @@ mod tests { fn decode_large_frame_header_caps_preallocation() { let mut codec = BytesCodec::new(); let mut buf = BytesMut::new(); - let n = 0x3FFFFFFFusize; + let n = MAX_FRAME_LENGTH; const MAX_REASONABLE_CAPACITY: usize = MAX_PREALLOCATED_PAYLOAD_LEN * 4; buf.put_u32_le((n << 2) as u32 | 0x3); diff --git a/src/stream.rs b/src/stream.rs index a8e6b6c2d..1c9f381fd 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -74,6 +74,64 @@ impl Stream { } } + /// Whether this is a WebRTC transport. Used to enforce the WebRTC-only DTLS fingerprint + /// binding (fail closed) in the secure handshake. + #[inline] + pub fn is_webrtc(&self) -> bool { + match self { + #[cfg(feature = "webrtc")] + Stream::WebRTC(_) => true, + #[allow(unreachable_patterns)] + _ => false, + } + } + + /// Close the underlying WebRTC peer connection if this is a WebRTC stream; no-op otherwise. + /// A WebRTC pc is kept alive by the global session cache and its cleanup handler only fires on + /// a terminal ICE state, so it must be closed explicitly at session end. TCP/WebSocket streams + /// release their resources on drop and need nothing here. + #[inline] + pub async fn close_webrtc(&self) { + match self { + #[cfg(feature = "webrtc")] + Stream::WebRTC(s) => s.close().await, + #[allow(unreachable_patterns)] + _ => {} + } + } + + /// 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. + #[inline] + pub async fn webrtc_relayed(&self) -> Option { + match self { + #[cfg(feature = "webrtc")] + Stream::WebRTC(s) => s.is_relayed().await, + #[allow(unreachable_patterns)] + _ => None, + } + } + + /// DTLS certificate fingerprint for a WebRTC stream (`local`=true for this endpoint's own + /// cert, false for the peer's), used to bind the channel to the signed peer identity. + /// Returns None for non-WebRTC transports, which authenticate via the secretbox key exchange. + #[inline] + #[cfg_attr(not(feature = "webrtc"), allow(unused_variables))] + pub async fn dtls_fingerprint(&self, local: bool) -> Option { + match self { + #[cfg(feature = "webrtc")] + Stream::WebRTC(s) => { + if local { + s.local_dtls_fingerprint().await.ok() + } else { + s.remote_dtls_fingerprint().await.ok() + } + } + #[allow(unreachable_patterns)] + _ => None, + } + } + #[inline] pub async fn next_timeout( &mut self, diff --git a/src/webrtc.rs b/src/webrtc.rs index 79af7bcce..e05ae04f7 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -1,28 +1,33 @@ use std::collections::HashMap; use std::io::{Error, ErrorKind}; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex as StdMutex}; use std::time::Duration; use webrtc::api::setting_engine::SettingEngine; use webrtc::api::APIBuilder; +use webrtc::data::data_channel::DataChannel as DetachedDataChannel; use webrtc::data_channel::RTCDataChannel; use webrtc::ice::mdns::MulticastDnsMode; use webrtc::ice_transport::ice_candidate::RTCIceCandidateInit; +use webrtc::ice_transport::ice_candidate_type::RTCIceCandidateType; use webrtc::ice_transport::ice_server::RTCIceServer; use webrtc::peer_connection::configuration::RTCConfiguration; use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState; use webrtc::peer_connection::policy::ice_transport_policy::RTCIceTransportPolicy; use webrtc::peer_connection::sdp::session_description::RTCSessionDescription; use webrtc::peer_connection::RTCPeerConnection; +use webrtc::stats::StatsReportType; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use base64::Engine; -use bytes::{Bytes, BytesMut}; +use bytes::{BufMut, Bytes, BytesMut}; use tokio::sync::{mpsc, watch, Mutex}; use tokio::time::timeout; use url::Url; +use crate::bytes_codec::MAX_FRAME_LENGTH; use crate::config; use crate::protobuf::Message; use crate::sodiumoxide::crypto::secretbox::Key; @@ -42,12 +47,41 @@ pub struct WebRTCStream { local_ice_rx: Arc>>>, session_key: String, send_timeout: u64, + // Built with Relay-only ICE policy (force_relay): every selected pair goes through TURN. + relay_only: bool, + // Detached data channel, cached after the first `detach()` so send/recv do not re-lock and + // re-fetch it per message. Shared across clones; `detach()` is idempotent. + detached: Arc>>>, + // Receive-side reassembly state, guarded by a single mutex so the fragment accumulator + // survives `next()` cancellation (e.g. `next_timeout`) instead of losing already-read + // fragments mid-message. Assumes a single reader, consistent with the rest of the stream API. + recv_state: Arc>, + // 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 + // "secured after key exchange" so key-less / unbound WebRTC is not shown as peer-authenticated. + peer_verified: Arc, } -/// Standard maximum message size for WebRTC data channels (RFC 8831, 65535 bytes). -/// Most browsers, including Chromium, enforce this protocol limit. -const DATA_CHANNEL_BUFFER_SIZE: u16 = u16::MAX; +#[derive(Default)] +struct RecvState { + // Accumulated payload of the logical message currently being reassembled. + acc: BytesMut, + // Reused read scratch buffer, avoiding a per-message allocation. + scratch: Vec, +} +// The SCTP data channel's 65536-byte max message size is handled by +// splitting a logical message into fragments carrying a 1-byte header. Fragment payload is kept +// well under the limit so header+payload never reaches the exact-65536 boundary that the +// receiver's reassembly would truncate with data loss. +const MAX_FRAGMENT_PAYLOAD: usize = 60000; +/// Receive scratch size: must be >= 1 (fragment header) + `MAX_FRAGMENT_PAYLOAD` and fit the +/// negotiated SCTP max message size. +const RECV_BUF_SIZE: usize = 64 * 1024; +/// Fragment header byte: more fragments follow for this logical message. +const FRAG_MORE: u8 = 1; +/// Fragment header byte: final (or only) fragment of a logical message. +const FRAG_END: u8 = 0; // use 3 public STUN servers to find out the NAT type, 2 must be the same address but different ports // https://stackoverflow.com/questions/72805316/determine-nat-mapping-behaviour-using-two-stun-servers // luckily nextcloud supports two ports for STUN @@ -71,6 +105,10 @@ impl Clone for WebRTCStream { local_ice_rx: self.local_ice_rx.clone(), session_key: self.session_key.clone(), send_timeout: self.send_timeout, + relay_only: self.relay_only, + detached: self.detached.clone(), + recv_state: self.recv_state.clone(), + peer_verified: self.peer_verified.clone(), } } } @@ -128,12 +166,25 @@ impl WebRTCStream { Ok(fingerprint.to_string()) } + /// Process-local SESSIONS-map key: the DTLS fingerprint prefixed by role. An offerer and an + /// answerer that share a fingerprint (a single process connecting to its own id) would + /// otherwise collide, handing the offerer back as the answerer. The wire-level `session_key` + /// used for ICE-candidate routing stays the bare fingerprint so both peers still match. + #[inline] + fn cache_key(fingerprint: &str, is_offerer: bool) -> String { + format!( + "{}:{}", + if is_offerer { "offer" } else { "answer" }, + fingerprint + ) + } + #[inline] fn get_key_for_sdp_json(sdp_json: &str) -> ResultType { if sdp_json.is_empty() { return Ok("".to_string()); } - let sdp = serde_json::from_str::(&sdp_json)?; + let sdp = serde_json::from_str::(sdp_json)?; Self::get_key_for_sdp(&sdp) } @@ -177,6 +228,17 @@ impl WebRTCStream { } } + /// Whether the ICE configuration contains a usable TURN server. A Relay-policy peer + /// connection (force_relay) can only gather relay candidates, so without a TURN server it can + /// never connect — callers use this to skip building a guaranteed-dead pc. + pub fn has_turn_server() -> bool { + Self::get_ice_servers().iter().any(|s| { + s.urls + .iter() + .any(|u| u.starts_with("turn:") || u.starts_with("turns:")) + }) + } + #[inline] fn get_ice_servers() -> Vec { let mut ice_servers = Vec::new(); @@ -217,7 +279,12 @@ impl WebRTCStream { force_relay: bool, ms_timeout: u64, ) -> ResultType { - log::debug!("New webrtc stream to endpoint: {}", remote_endpoint); + // The endpoint contains a Base64-encoded SDP with host addresses and live ICE + // credentials. Log only its size so debug logs cannot disclose that information. + log::debug!( + "New webrtc stream (remote endpoint: {} bytes)", + remote_endpoint.len() + ); let remote_offer = if remote_endpoint.is_empty() { "".into() } else { @@ -225,16 +292,16 @@ impl WebRTCStream { }; let mut key = Self::get_key_for_sdp_json(&remote_offer)?; - let sessions_lock = SESSIONS.lock().await; - if let Some(cached_stream) = sessions_lock.get(&key) { - if !key.is_empty() { + let start_local_offer = remote_offer.is_empty(); + if !key.is_empty() { + let sessions_lock = SESSIONS.lock().await; + if let Some(cached_stream) = + sessions_lock.get(&Self::cache_key(&key, start_local_offer)) + { log::debug!("Start webrtc with cached peer"); return Ok(cached_stream.clone()); } } - drop(sessions_lock); - - let start_local_offer = remote_offer.is_empty(); // Create a SettingEngine and enable Detach let mut s = SettingEngine::default(); s.detach_data_channels(); @@ -284,7 +351,14 @@ impl WebRTCStream { let bootstrap_dc = if start_local_offer { let dc_open_notify = notify_tx.clone(); // Create a data channel with label "bootstrap" - let dc = pc.create_data_channel("bootstrap", None).await?; + let dc = match pc.create_data_channel("bootstrap", None).await { + Ok(dc) => dc, + Err(e) => { + // Close before propagating: the pc is live and would otherwise leak. + pc.close().await.ok(); + return Err(e.into()); + } + }; dc.on_open(Box::new(move || { log::debug!("Local data channel bootstrap open."); let _ = dc_open_notify.send(WebRTCConnectionState::Open); @@ -321,7 +395,12 @@ impl WebRTCStream { // This will notify you when the peer has connected/disconnected let stream_for_close = stream.clone(); - let pc_for_close = pc.clone(); + // Weak, not strong: a handler stored inside the pc that captured a strong + // `Arc` forms a pc -> internal -> handler -> pc cycle that `close()` + // never breaks (it only fires the handler) and no `Drop` clears, permanently leaking every + // pc and the ICE-candidate sender's forwarding task. Upgrade inside the handler; if the pc + // is already gone there is nothing left in SESSIONS to evict. + let pc_for_close = Arc::downgrade(&pc); pc.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| { let stream_for_close2 = stream_for_close.clone(); let on_connection_notify = notify_tx.clone(); @@ -329,21 +408,35 @@ impl WebRTCStream { Box::pin(async move { log::debug!("WebRTC session peer connection state: {}", s); match s { - RTCPeerConnectionState::Disconnected - | RTCPeerConnectionState::Failed - | RTCPeerConnectionState::Closed => { + // `Disconnected` is a transient, recoverable ICE state (webrtc-ice fires it + // after ~5s without consent and returns to `Connected` when traffic resumes). + // Only tear down on the terminal states so a short network blip (Wi-Fi roam, + // sleep/wake, cell handover) does not permanently kill an established session. + RTCPeerConnectionState::Failed | RTCPeerConnectionState::Closed => { let _ = on_connection_notify.send(WebRTCConnectionState::Closed( s.to_string(), )); - log::debug!("WebRTC session closing due to disconnected"); + log::debug!("WebRTC session closing due to {}", s); let _ = stream_for_close2.lock().await.close().await; log::debug!("WebRTC session stream closed"); + let Some(pc_for_close2) = pc_for_close2.upgrade() else { + return; + }; let mut sessions_lock = SESSIONS.lock().await; match Self::get_key_for_peer(&pc_for_close2, start_local_offer).await { - Ok(k) => { - sessions_lock.remove(&k); - log::debug!("WebRTC session removed key: {}", k); + Ok(fingerprint) => { + let k = Self::cache_key(&fingerprint, start_local_offer); + // Only evict if the cached entry IS this pc: a duplicate offer + // resolves to the same key, and closing the discarded duplicate pc + // must not remove the live winner sharing that key. + if sessions_lock + .get(&k) + .is_some_and(|s| Arc::ptr_eq(&s.pc, &pc_for_close2)) + { + sessions_lock.remove(&k); + log::debug!("WebRTC session removed key: {}", k); + } } Err(e) => { log::error!( @@ -374,31 +467,45 @@ impl WebRTCStream { })); // process offer/answer - if start_local_offer { - let sdp = pc.create_offer(None).await?; - pc.set_local_description(sdp.clone()).await?; - - log::debug!("local offer:\n{}", sdp.sdp); - // get local sdp key - key = Self::get_key_for_sdp(&sdp)?; - log::debug!("Start webrtc with local key: {}", key); - } else { - let sdp = serde_json::from_str::(&remote_offer)?; - pc.set_remote_description(sdp.clone()).await?; - let answer = pc.create_answer(None).await?; - pc.set_local_description(answer).await?; - - log::debug!("remote offer:\n{}", sdp.sdp); - // get remote sdp key - key = Self::get_key_for_sdp(&sdp)?; - log::debug!("Start webrtc with remote key: {}", key); - } - - let mut final_lock = SESSIONS.lock().await; - if let Some(session) = final_lock.get(&key) { - pc.close().await.ok(); - return Ok(session.clone()); + // + // Trickle ICE: the local description is returned WITHOUT waiting for candidate gathering + // (candidates stream out via `take_local_ice_rx` afterwards), so this block is local-only + // work — pc construction, DTLS cert keygen, SDP marshal — at sub-millisecond cost. The + // controlled side awaits answer creation inline on its punch-reply critical path and + // relies on that: adding any gathering/network wait here would delay the TCP/UDP + // hole-punch reply for every connection. + // 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 = async { + if start_local_offer { + let sdp = pc.create_offer(None).await?; + pc.set_local_description(sdp.clone()).await?; + // SDP carries host/srflx IPs and ICE ufrag/pwd; log only its size, not the body. + log::debug!("local offer SDP built ({} bytes)", sdp.sdp.len()); + let k = Self::get_key_for_sdp(&sdp)?; + log::debug!("Start webrtc with local key: {}", k); + Ok(k) + } else { + let sdp = serde_json::from_str::(&remote_offer)?; + pc.set_remote_description(sdp.clone()).await?; + let answer = pc.create_answer(None).await?; + pc.set_local_description(answer).await?; + log::debug!("remote offer SDP received ({} bytes)", sdp.sdp.len()); + let k = Self::get_key_for_sdp(&sdp)?; + log::debug!("Start webrtc with remote key: {}", k); + Ok(k) + } } + .await; + key = match offer_answer { + Ok(k) => k, + Err(e) => { + pc.close().await.ok(); + return Err(e); + } + }; let webrtc_stream = Self { pc, @@ -407,8 +514,30 @@ impl WebRTCStream { local_ice_rx: Arc::new(StdMutex::new(Some(ice_rx))), session_key: key.clone(), send_timeout: ms_timeout, + relay_only: force_relay, + detached: Arc::new(Mutex::new(None)), + recv_state: Arc::new(Mutex::new(RecvState::default())), + peer_verified: Arc::new(AtomicBool::new(false)), }; - final_lock.insert(key, webrtc_stream.clone()); + // Insert into the session cache, but never `await pc.close()` while holding this lock: + // `close()` fires the peer-connection-state handler inline, which itself locks SESSIONS, + // self-deadlocking the whole process. Resolve any duplicate off-lock. + let cache_key = Self::cache_key(&key, start_local_offer); + let duplicate = { + let mut final_lock = SESSIONS.lock().await; + if let Some(session) = final_lock.get(&cache_key) { + Some(session.clone()) + } else { + final_lock.insert(cache_key, webrtc_stream.clone()); + None + } + }; + if let Some(session) = duplicate { + // A concurrent `new()` already cached an equivalent stream; discard this pc's + // resources (off-lock) and return the cached one. + webrtc_stream.close().await; + return Ok(session); + } Ok(webrtc_stream) } @@ -426,12 +555,58 @@ impl WebRTCStream { #[inline] pub async fn set_remote_endpoint(&self, endpoint: &str) -> ResultType<()> { let offer = Self::get_remote_offer(endpoint)?; - log::debug!("WebRTC set remote sdp: {}", offer); + log::debug!("WebRTC set remote sdp ({} bytes)", offer.len()); let sdp = serde_json::from_str::(&offer)?; self.pc.set_remote_description(sdp).await?; Ok(()) } + /// DTLS certificate fingerprint of the local description (this endpoint's own cert). + #[inline] + pub async fn local_dtls_fingerprint(&self) -> ResultType { + Self::get_key_for_peer(&self.pc, true).await + } + + /// DTLS certificate fingerprint of the remote description (the peer's cert). webrtc-rs + /// verifies the negotiated peer certificate against this fingerprint during the DTLS + /// handshake, so once the channel is open a matching fingerprint identifies the peer's cert. + #[inline] + pub async fn remote_dtls_fingerprint(&self) -> ResultType { + Self::get_key_for_peer(&self.pc, false).await + } + + /// Whether the established connection runs through a TURN relay: `Some(true)` when the pc is + /// Relay-policy (TURN is the only possibility) or the selected ICE candidate pair uses a + /// relay candidate; `None` before a pair is selected. Feeds the UI's direct/relayed flag. + pub async fn is_relayed(&self) -> Option { + if self.relay_only { + return Some(true); + } + let dtls = self.pc.sctp().transport(); + dtls.ice_transport().get_selected_candidate_pair().await?; + + // webrtc 0.13 keeps RTCIceCandidatePair's candidates private. Its stats report exposes + // the selected (nominated) pair and the corresponding candidate types instead. + let stats = self.pc.get_stats().await; + let pair = stats.reports.values().find_map(|report| match report { + StatsReportType::CandidatePair(pair) if pair.nominated => Some(pair), + _ => None, + })?; + let is_relay = |candidate_id: &str| { + matches!( + stats.reports.get(candidate_id), + Some( + StatsReportType::LocalCandidate(candidate) + | StatsReportType::RemoteCandidate(candidate) + ) if RTCIceCandidateType::from(candidate.candidate_type) + == RTCIceCandidateType::Relay + ) + }; + Some( + is_relay(&pair.local_candidate_id) || is_relay(&pair.remote_candidate_id), + ) + } + #[inline] pub fn take_local_ice_rx(&self) -> Option> { self.local_ice_rx.lock().ok().and_then(|mut rx| rx.take()) @@ -464,6 +639,19 @@ impl WebRTCStream { Ok(()) } + /// Explicitly tear down the peer connection. + /// + /// Dropping a `WebRTCStream` handle is not enough to release the underlying + /// `RTCPeerConnection`: the global `SESSIONS` map holds a clone, so the pc (and its + /// 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(); + } + #[inline] pub fn set_raw(&mut self) { // not-supported @@ -481,14 +669,16 @@ impl WebRTCStream { #[inline] pub fn set_key(&mut self, _key: Key) { - // not-supported - // WebRTC uses built-in DTLS encryption for secure communication. - // DTLS handles key exchange and encryption automatically, so explicit key management is not required. + // WebRTC traffic is DTLS-encrypted regardless; the secretbox key is unused. + // Callers invoke set_key only after the controller has bound the DTLS fingerprint to the + // verified peer identity (or the controlled side has completed the matching handshake). + // Mark peer-verified so is_secured() matches TCP's post-key-exchange meaning. + self.peer_verified.store(true, Ordering::Release); } #[inline] pub fn is_secured(&self) -> bool { - true + self.peer_verified.load(Ordering::Acquire) } #[inline] @@ -515,20 +705,73 @@ impl WebRTCStream { } } - pub async fn send_bytes(&mut self, bytes: Bytes) -> ResultType<()> { - if let Err(err) = self.wait_connected(self.send_timeout).await { - self.pc.close().await.ok(); - let kind = if err.to_string().contains("deadline") - || err.to_string().contains("timeout") - { - ErrorKind::TimedOut - } else { - ErrorKind::Other - }; - return Err(Error::new(kind, err.to_string()).into()); + /// Fetch (and cache) the detached data channel. `detach()` is idempotent and returns a + /// clone of the same underlying channel, so caching it just avoids re-locking per message. + async fn detached_dc(&self) -> ResultType> { + { + let cache = self.detached.lock().await; + if let Some(dc) = cache.as_ref() { + return Ok(dc.clone()); + } + } + let raw = self.stream.lock().await.clone(); + let dc = raw.detach().await?; + let mut cache = self.detached.lock().await; + // Another task may have cached it while we were detaching. + if let Some(existing) = cache.as_ref() { + return Ok(existing.clone()); + } + *cache = Some(dc.clone()); + Ok(dc) + } + + pub async fn send_bytes(&mut self, bytes: Bytes) -> ResultType<()> { + let send_timeout = self.send_timeout; + // Bound the WHOLE data-channel send (wait-for-open + every write) by send_timeout, + // mirroring FramedStream. Without this a write can park indefinitely on SCTP + // pending-queue backpressure and connection.rs's timeout timer never runs. + if send_timeout > 0 { + match timeout( + Duration::from_millis(send_timeout), + self.send_bytes_inner(bytes), + ) + .await + { + Ok(res) => res, + Err(_) => { + self.pc.close().await.ok(); + Err(Error::new(ErrorKind::TimedOut, "WebRTC send timeout").into()) + } + } + } else { + self.send_bytes_inner(bytes).await + } + } + + async fn send_bytes_inner(&mut self, bytes: Bytes) -> ResultType<()> { + if bytes.len() > MAX_FRAME_LENGTH { + return Err(Error::new(ErrorKind::InvalidInput, "Overflow").into()); + } + self.wait_for_connect_result().await?; + let dc = self.detached_dc().await?; + let data = bytes.as_ref(); + let mut offset = 0; + // Always emit at least one fragment (a lone FRAG_END header for an empty message), so a + // zero-length data-channel message — which the receiver cannot distinguish from EOF — is + // never sent. + loop { + let end = (offset + MAX_FRAGMENT_PAYLOAD).min(data.len()); + let is_last = end >= data.len(); + let chunk = &data[offset..end]; + let mut framed = BytesMut::with_capacity(1 + chunk.len()); + framed.put_u8(if is_last { FRAG_END } else { FRAG_MORE }); + framed.put_slice(chunk); + dc.write(&framed.freeze()).await?; + offset = end; + if is_last { + break; + } } - let stream = self.stream.lock().await.clone(); - stream.send(&bytes).await?; Ok(()) } @@ -538,30 +781,53 @@ impl WebRTCStream { self.pc.close().await.ok(); return Some(Err(Error::new(ErrorKind::Other, err.to_string()))); } - let stream = self.stream.lock().await.clone(); - - // TODO reuse buffer? - let mut buffer = BytesMut::zeroed(DATA_CHANNEL_BUFFER_SIZE as usize); - let dc = stream.detach().await.ok()?; - let n = match dc.read(&mut buffer).await { - Ok(n) => n, + let dc = match self.detached_dc().await { + Ok(dc) => dc, Err(err) => { + self.pc.close().await.ok(); + return Some(Err(Error::new(ErrorKind::Other, err.to_string()))); + } + }; + // Hold recv_state across the reassembly loop: the accumulator must survive `next()` + // cancellation (e.g. next_timeout) so already-read fragments are not lost mid-message. + let mut st = self.recv_state.lock().await; + if st.scratch.len() < RECV_BUF_SIZE { + st.scratch.resize(RECV_BUF_SIZE, 0); + } + loop { + let RecvState { acc, scratch } = &mut *st; + let n = match dc.read(scratch.as_mut_slice()).await { + Ok(n) => n, + Err(err) => { + self.pc.close().await.ok(); + return Some(Err(Error::new( + ErrorKind::Other, + format!("data channel read error: {}", err), + ))); + } + }; + if n == 0 { + // Clean EOF: the remote reset the stream or shut its write half. An empty logical + // message is represented by a 1-byte header, so it is never confused with EOF. + self.pc.close().await.ok(); + return None; + } + 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(); self.pc.close().await.ok(); return Some(Err(Error::new( ErrorKind::Other, - format!("data channel read error: {}", err), + "WebRTC reassembled message exceeded maximum frame size", ))); } - }; - if n == 0 { - self.pc.close().await.ok(); - return Some(Err(Error::new( - ErrorKind::Other, - "data channel read exited with 0 bytes", - ))); + if scratch[0] == FRAG_END { + let msg = std::mem::take(acc); + return Some(Ok(msg)); + } } - buffer.truncate(n); - Some(Ok(buffer)) } #[inline] @@ -583,6 +849,8 @@ mod tests { use crate::config; use crate::webrtc::WebRTCStream; use crate::webrtc::DEFAULT_ICE_SERVERS; + use std::time::Duration; + use tokio::time::timeout; use webrtc::peer_connection::sdp::session_description::RTCSessionDescription; #[test] @@ -841,4 +1109,81 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc let err = stream.wait_connected(10).await.unwrap_err(); assert!(err.to_string().contains("timeout")); } + + async fn connect_loopback() -> (WebRTCStream, WebRTCStream) { + let mut offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + let offer = offerer.get_local_endpoint().await.unwrap(); + let answerer = WebRTCStream::new(&offer, false, 20000).await.unwrap(); + let answer = answerer.get_local_endpoint().await.unwrap(); + offerer.set_remote_endpoint(&answer).await.unwrap(); + + // Bridge trickle candidates directly between the two peers, both directions. + let mut off_ice = offerer.take_local_ice_rx().unwrap(); + let mut ans_ice = answerer.take_local_ice_rx().unwrap(); + let answerer_for_ice = answerer.clone(); + let offerer_for_ice = offerer.clone(); + tokio::spawn(async move { + while let Some(c) = off_ice.recv().await { + let _ = answerer_for_ice.add_remote_ice_candidate(&c).await; + } + }); + tokio::spawn(async move { + while let Some(c) = ans_ice.recv().await { + let _ = offerer_for_ice.add_remote_ice_candidate(&c).await; + } + }); + + offerer.wait_connected(20000).await.unwrap(); + let mut answerer = answerer; + answerer.wait_connected(20000).await.unwrap(); + (offerer, answerer) + } + + // In-process offerer<->answerer loopback exercising the send/next data plane that the framing, + // empty-message, and EOF fixes live in. Connects over host candidates (works offline; any + // configured/default STUN just fails in the background without blocking the host pair). + #[tokio::test] + async fn test_webrtc_loopback_roundtrip() { + let connect = async { + let (mut offerer, mut answerer) = connect_loopback().await; + + // Host-candidate loopback is direct, never TURN-relayed. + assert_eq!(offerer.is_relayed().await, Some(false)); + + // Small message. + offerer.send_raw(b"hello".to_vec()).await.unwrap(); + let got = answerer.next().await.unwrap().unwrap(); + assert_eq!(&got[..], b"hello"); + + // Empty message: must round-trip as an empty frame, not be seen as EOF. + offerer.send_raw(Vec::new()).await.unwrap(); + let got = answerer.next().await.unwrap().unwrap(); + assert_eq!(got.len(), 0, "empty message must not be treated as EOF"); + + // Payload far above the 64KB single-message cap: must be fragmented and reassembled. + let big = vec![0xABu8; 200_000]; + offerer.send_raw(big.clone()).await.unwrap(); + let got = answerer.next().await.unwrap().unwrap(); + assert_eq!(got.len(), big.len(), "large message must survive fragmentation"); + assert_eq!(&got[..], &big[..]); + + // Reverse direction. + answerer.send_raw(b"world".to_vec()).await.unwrap(); + let got = offerer.next().await.unwrap().unwrap(); + assert_eq!(&got[..], b"world"); + + // Peer close: the other side observes a clean EOF (None) or a close error, never a hang. + offerer.close().await; + match timeout(Duration::from_secs(10), answerer.next()).await { + Ok(None) | Ok(Some(Err(_))) => {} + Ok(Some(Ok(b))) => panic!("expected EOF after peer close, got {} bytes", b.len()), + Err(_) => panic!("answerer.next() hung after peer close"), + } + answerer.close().await; + }; + timeout(Duration::from_secs(40), connect) + .await + .expect("webrtc loopback did not complete in time"); + } + }