diff --git a/Cargo.toml b/Cargo.toml index 11a49653a2..09d2850952 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,7 +68,11 @@ 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 } +# Pinned to 0.13: webrtc >=0.14 pulls sdp 0.10 / webrtc-util 0.12, which use +# usize::is_multiple_of (needs rustc >=1.87), while rustdesk CI builds with Rust 1.75 +# (sciter i128 ABI pin, flutter-build.yml). Bump only after CI's Rust moves past 1.87, +# and work through the upgrade checklist at the top of src/webrtc.rs first. +webrtc = { version = "0.13.0", optional = true } libloading = "0.8" [target.'cfg(not(any(target_os = "android", target_os = "ios")))'.dependencies] @@ -81,7 +85,7 @@ protobuf-codegen = { version = "3.7" } [dev-dependencies] clap = "4.5.51" -webrtc = "0.14.0" +webrtc = "0.13.0" # keep in lockstep with [dependencies] webrtc (rustc 1.75 pin, see above) [target.'cfg(target_os = "windows")'.dependencies] winapi = { version = "0.3", features = [ diff --git a/protos/message.proto b/protos/message.proto index 8b21368114..2024634cf5 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/protos/rendezvous.proto b/protos/rendezvous.proto index b2e5c01572..8f796502fc 100644 --- a/protos/rendezvous.proto +++ b/protos/rendezvous.proto @@ -29,6 +29,10 @@ message PunchHoleRequest { int32 upnp_port = 9; bytes socket_addr_v6 = 10; string switch_code = 11; + // The offer's envelope declares its own ICE transport policy (`ice_policy` key inside + // the webrtc:// payload): under force_relay it tells the peer whether the relay is + // transport-forced (WebSocket — answer may use full ICE) or policy (Relay-only + TURN). + string webrtc_sdp_offer = 12; } message ControlPermissions { @@ -64,6 +68,7 @@ message PunchHole { bytes socket_addr_v6 = 7; ControlPermissions control_permissions = 8; ControlledContext controlled_context = 9; + string webrtc_sdp_offer = 10; } message TestNatRequest { @@ -90,6 +95,7 @@ message PunchHoleSent { string version = 5; int32 upnp_port = 6; bytes socket_addr_v6 = 7; + string webrtc_sdp_answer = 8; } message RegisterPk { @@ -135,6 +141,7 @@ message PunchHoleResponse { bool is_udp = 9; int32 upnp_port = 10; bytes socket_addr_v6 = 11; + string webrtc_sdp_answer = 12; } message ConfigUpdate { @@ -169,6 +176,7 @@ message RelayResponse { int32 feedback = 9; bytes socket_addr_v6 = 10; int32 upnp_port = 11; + string webrtc_sdp_answer = 12; } message SoftwareUpdate { string url = 1; } @@ -240,6 +248,13 @@ message HttpProxyResponse { string error = 4; } +message IceCandidate { + string id = 1; + bytes socket_addr = 2; + string session_key = 3; + string candidate = 4; +} + message RendezvousMessage { oneof union { RegisterPeer register_peer = 6; @@ -265,5 +280,6 @@ message RendezvousMessage { HealthCheck hc = 26; HttpProxyRequest http_proxy_request = 27; HttpProxyResponse http_proxy_response = 28; + IceCandidate ice_candidate = 29; } } diff --git a/src/bytes_codec.rs b/src/bytes_codec.rs index cbd53d918f..5bd5590c14 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/config.rs b/src/config.rs index e968ec469a..793ad47779 100644 --- a/src/config.rs +++ b/src/config.rs @@ -2993,8 +2993,11 @@ pub mod keys { "allow-command-line-settings-when-settings-disabled"; // Connection punch-through options + pub const OPTION_ENABLE_TCP_PUNCH: &str = "enable-tcp-punch"; pub const OPTION_ENABLE_UDP_PUNCH: &str = "enable-udp-punch"; pub const OPTION_ENABLE_IPV6_PUNCH: &str = "enable-ipv6-punch"; + pub const OPTION_ENABLE_WEBRTC: &str = "enable-webrtc"; + pub const OPTION_ALLOW_KCP_CC: &str = "allow-kcp-congestion-control"; pub const OPTION_HIDE_USERNAME_ON_CARD: &str = "hide-username-on-card"; pub const OPTION_HIDE_HELP_CARDS: &str = "hide-help-cards"; pub const OPTION_DEFAULT_CONNECT_PASSWORD: &str = "default-connect-password"; @@ -3124,8 +3127,10 @@ pub mod keys { OPTION_ALLOW_AUTO_RECORD_OUTGOING, OPTION_HIDE_RECORDING_BUTTON, OPTION_VIDEO_SAVE_DIRECTORY, + OPTION_ENABLE_TCP_PUNCH, OPTION_ENABLE_UDP_PUNCH, OPTION_ENABLE_IPV6_PUNCH, + OPTION_ENABLE_WEBRTC, OPTION_TOUCH_MODE, OPTION_SHOW_VIRTUAL_MOUSE, OPTION_SHOW_VIRTUAL_JOYSTICK, @@ -3192,6 +3197,7 @@ pub mod keys { OPTION_ALLOW_INSECURE_TLS_FALLBACK, OPTION_KEEP_AWAKE_DURING_INCOMING_SESSIONS, OPTION_ALLOW_AUTO_UPDATE, + OPTION_ALLOW_KCP_CC, ]; // BUILDIN_SETTINGS diff --git a/src/lib.rs b/src/lib.rs index 2b35642190..3e7435e577 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -57,6 +57,7 @@ pub use toml; pub use uuid; pub mod fingerprint; pub use flexi_logger; +pub mod log_throttle; pub mod stream; pub mod websocket; #[cfg(feature = "webrtc")] @@ -457,8 +458,18 @@ pub fn init_log(_is_async: bool, _name: &str) -> Option, +} + +struct ThrottleState { + suppressed: u64, + last: Option, +} + +impl LogThrottle { + pub const fn new(interval: Duration) -> Self { + Self { + interval, + state: Mutex::new(ThrottleState { + suppressed: 0, + last: None, + }), + } + } + + /// Record one occurrence. Returns the number of occurrences to report (including this one) + /// when a line is due, or `None` while still inside the interval. + /// + /// The first occurrence after a quiet period always reports, so an isolated fault is not + /// delayed by the interval. + pub fn due(&self) -> Option { + // A poisoned lock only means some other thread panicked while holding it; the guarded + // data is two plain counters that are still usable, and going silent for the rest of + // the process would be worse than a stale count. + let mut state = self + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + state.suppressed += 1; + // `map_or(true, ..)` rather than clippy's preferred `is_none_or`: that was stabilized in + // Rust 1.82 and this crate builds on the 1.75 pinned by CI. + #[allow(clippy::unnecessary_map_or)] + let due = state + .last + .map_or(true, |last| last.elapsed() >= self.interval); + if !due { + return None; + } + state.last = Some(Instant::now()); + Some(std::mem::replace(&mut state.suppressed, 0)) + } +} + +/// Log at most one line per interval from this call site, suffixed with the number of +/// occurrences it stands for. +/// +/// Each expansion declares its own hidden static, so two sites never share a count and +/// adding one is a single line: +/// +/// ```ignore +/// throttled_log!(Duration::from_secs(5), warn, "rejected ipc peer {peer_pid:?}"); +/// ``` +/// +/// An isolated event logs unchanged; a burst collapses to `... (x47)`. The count includes +/// the occurrence being reported, so it reads as a total rather than as "and N more". +#[macro_export] +macro_rules! throttled_log { + ($interval:expr, $level:ident, $($arg:tt)+) => {{ + static THROTTLE: $crate::log_throttle::LogThrottle = + $crate::log_throttle::LogThrottle::new($interval); + if let Some(n) = THROTTLE.due() { + if n > 1 { + $crate::log::$level!("{} (x{})", format_args!($($arg)+), n); + } else { + $crate::log::$level!("{}", format_args!($($arg)+)); + } + } + }}; +} + +#[cfg(test)] +mod tests { + use super::*; + + // Two directions of one socket need two throttles: an ICMP error on a connected socket is + // reported once and cleared, so the steady state alternates (send succeeds, the next recv + // reports it) and anything shared between them is reset by the succeeding side every cycle. + #[test] + fn separate_throttles_do_not_reset_each_other() { + let send = LogThrottle::new(Duration::from_secs(60)); + let recv = LogThrottle::new(Duration::from_secs(60)); + assert_eq!(recv.due(), Some(1)); + for _ in 0..1_000 { + // The send side succeeding must not hand the recv side a fresh emit slot. + assert_eq!(recv.due(), None); + } + assert_eq!( + send.due(), + Some(1), + "the other direction keeps its own slot" + ); + } + + #[test] + fn first_call_reports_immediately() { + let t = LogThrottle::new(Duration::from_secs(60)); + assert_eq!(t.due(), Some(1)); + } + + #[test] + fn calls_inside_the_interval_are_counted_not_reported() { + let t = LogThrottle::new(Duration::from_secs(60)); + assert_eq!(t.due(), Some(1)); + for _ in 0..100 { + assert_eq!(t.due(), None); + } + } + + #[test] + fn the_next_due_line_carries_everything_suppressed() { + let t = LogThrottle::new(Duration::ZERO); + assert_eq!(t.due(), Some(1)); + // A zero interval is always due, so each call reports exactly itself. + assert_eq!(t.due(), Some(1)); + + let t = LogThrottle::new(Duration::from_millis(30)); + assert_eq!(t.due(), Some(1)); + assert_eq!(t.due(), None); + assert_eq!(t.due(), None); + std::thread::sleep(Duration::from_millis(40)); + // The two suppressed calls plus this one. + assert_eq!(t.due(), Some(3)); + } +} diff --git a/src/stream.rs b/src/stream.rs index a8e6b6c2d1..2df8269154 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -74,6 +74,87 @@ 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. + /// + /// Deliberately not `async`: most callers sit in a `select!` arm or a future the UI can + /// abandon, and an awaited close that loses that race is unretryable — `close()` latches + /// `is_closed` before its first await, so every later attempt early-returns while the state + /// handler that would evict the session never runs. With no await point here there is + /// nothing to cancel; the teardown runs to completion on the WebRTC I/O runtime + /// (see `webrtc::WEBRTC_RT`) that owns the pc's sockets and pump tasks, so it reaches the + /// wire even after a caller's own runtime dies on return. + #[inline] + pub fn close_webrtc(&self) { + match self { + #[cfg(feature = "webrtc")] + Stream::WebRTC(s) => s.close_detached(), + #[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, 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 { + match self { + #[cfg(feature = "webrtc")] + Stream::WebRTC(s) => s.is_relayed().await, + #[allow(unreachable_patterns)] + _ => None, + } + } + + /// Whether an established WebRTC transport reaches the peer over IPv6 (used to name the + /// transport in the UI). `None` for non-WebRTC transports and before ICE selects a pair — + /// every other transport already carries the family in the label it was raced under. + #[inline] + pub async fn webrtc_remote_ipv6(&self) -> Option { + match self { + #[cfg(feature = "webrtc")] + Stream::WebRTC(s) => s.is_remote_ipv6().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, @@ -137,13 +218,21 @@ impl Stream { pub fn from(stream: TcpStream, stream_addr: SocketAddr) -> Self { Self::Tcp(tcp::FramedStream::from(stream, stream_addr)) } +} - #[inline] - #[cfg(feature = "webrtc")] - pub fn get_webrtc_stream(&self) -> Option { - match self { - Self::WebRTC(s) => Some(s.clone()), - _ => None, - } +/// Owning the stream owns the transport, WebRTC included. +/// +/// A peer connection outlives its handle — the session cache holds a clone, and the handler that +/// evicts it only fires on a terminal ICE state — so it has to be closed explicitly. Doing that +/// at each exit path made it an obligation every `return`, `break` and `?` had to remember, and +/// the long-lived side never did: `server::connection` ends its ~15 exits by dropping the stream. +/// Nothing warned, because a missed close leaks silently and only under WebRTC. +/// +/// `Stream` is not `Clone`, so dropping it really is the end of the transport and there is no +/// second owner to surprise. Explicit `close_webrtc()` calls remain valid — they close sooner +/// than scope end — but they are now an optimization rather than the thing correctness rests on. +impl Drop for Stream { + fn drop(&mut self) { + self.close_webrtc(); } } diff --git a/src/webrtc.rs b/src/webrtc.rs index 8f3c410cc7..41adfef5d7 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -1,26 +1,48 @@ +//! WebRTC transport for RustDesk streams. +//! +//! # Bumping the webrtc crate +//! +//! This module depends on webrtc-rs internals its public API does not promise. Re-verify each +//! against the new sources (last checked: webrtc 0.13, -data 0.11, -sctp 0.12), then run +//! `cargo test --features webrtc webrtc::tests`. +//! +//! - `DataChannel::write` parks on a full PendingQueue instead of buffering without bound. +//! - SCTP max message size is 65536: `MAX_FRAGMENT_PAYLOAD` + 1 must stay below it. +//! - `read_sctp` does not await after dequeuing, so a dropped `next()` cannot lose a fragment. +//! - `detach()` is an idempotent Arc clone that does not close on drop. +//! - `on_*` handlers live in the pc: one holding a strong `Arc` leaks it. +//! - `close()` latches `is_closed` before its first await, so a cancelled close is unretryable. +//! - `Disconnected` is transient; only `Failed`/`Closed` are terminal. +//! - `RTCIceCandidatePair`'s `Display` layout — `is_relayed()` parses it because 0.13 keeps the +//! candidates private. 0.17+ exposes them; switch and delete the parse. + use std::collections::HashMap; use std::io::{Error, ErrorKind}; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; -use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, 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::sdp_type::RTCSdpType; use webrtc::peer_connection::sdp::session_description::RTCSessionDescription; use webrtc::peer_connection::RTCPeerConnection; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use base64::Engine; -use bytes::{Bytes, BytesMut}; -use tokio::sync::watch; -use tokio::sync::Mutex; -use tokio::time::timeout; +use bytes::{Buf, BufMut, Bytes, BytesMut}; +use tokio::sync::{mpsc, watch, Mutex, Semaphore}; +use tokio::time::{timeout, timeout_at, Instant}; use url::Url; use crate::config; @@ -28,29 +50,177 @@ use crate::protobuf::Message; use crate::sodiumoxide::crypto::secretbox::Key; use crate::ResultType; +#[derive(Clone, Debug, PartialEq, Eq)] +enum WebRTCConnectionState { + Pending, + Open, + Closed(String), +} + +/// A shared handle, not an owner: every clone points at the same `pc`, and `SESSIONS` holds one +/// of those clones. Hence no `Drop` — closing per clone would kill a live session as soon as a +/// losing race future is dropped, and closing on the last clone would wait on the very close +/// that evicts the cache entry. Ownership is `OffererGuard` until a connection attempt adopts +/// the stream and `Stream::WebRTC` after; holding one outside those two means closing it by hand. pub struct WebRTCStream { pc: Arc, stream: Arc>>, - state_notify: watch::Receiver, + state_notify: watch::Receiver, + 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>>>, + // 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, + // Behind the Arc, not in the future, so a cancelled `next()` keeps the partial message. + // SINGLE READER ONLY: reassembly spans calls and the fragment header carries no length or + // sequence number, so two readers splice unrelated fragments into one wrong-but-valid message. + 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, + // Whether anyone still wants this pc — see `HandoffState`. Shared by every clone, the + // `SESSIONS` entry included, because that question is about the pc, not about one handle. + handoff: Arc, + // Taken once at construction: `pc.local_description()` grows with every gathered candidate + // and this must not (why: `UDP_ENDPOINT_BUDGET`). Shared so cloning the struct stays cheap. + local_endpoint: 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; +/// Whether a pc `new_inner` built is still wanted, shared by every `NewStreamHandoff` handed out +/// for it. `new_inner` caches the pc before its `new()` returns, so the builder's handoff and any +/// number of cache hits are in flight at once; a pc is abandoned only when the last of them is +/// dropped with none adopted, and closing on any single Drop would kill a live caller's session. +#[derive(Default)] +struct HandoffState { + // Handoffs handed out and neither adopted nor dropped yet. Incremented under the `SESSIONS` + // lock so a hit that is still arriving cannot be overtaken by the abandon-close below. + pending: AtomicUsize, + // Latched by the first `into_inner`: some caller owns the stream, and since `WebRTCStream` + // has no `Drop` there is nothing that could un-own it. Abandoning any later hit is then a + // no-op, which is what keeps a cached pc alive across a cancelled caller. + adopted: AtomicBool, + // Set under the `SESSIONS` lock once the abandon-close commits, so a hit that locks the map + // afterwards (eviction only happens later, from the state handler) rejects the entry instead + // of adopting a pc that is already going away. + closing: AtomicBool, +} + +#[derive(Default)] +struct RecvState { + // Accumulated payload of the logical message currently being reassembled. Only multi-fragment + // messages reach it; a message that arrives whole is split straight out of `scratch`. + acc: BytesMut, + // Read buffer, refilled in `SCRATCH_REFILL` chunks rather than per read. A whole-message + // fragment is handed to the caller with `split_to`, which shares this allocation instead of + // copying, so the buffer is consumed from the front and re-initialized only when what is left + // can no longer hold one fragment — amortizing that cost over many small messages. + scratch: BytesMut, +} -// 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 -// unluckily webrtc-rs does not use the same port to do the STUN request -static DEFAULT_ICE_SERVERS: [&str; 3] = [ +// 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; +/// How much read buffer to initialize at a time. Whole messages are split out of it without +/// copying, so it is consumed rather than reused; refilling in chunks spreads the one cost that +/// remains — zeroing — across every small message that fits. Slices handed to the caller keep the +/// whole chunk alive, so this trades a bounded amount of retention for the copy. +const SCRATCH_REFILL: usize = 4 * RECV_BUF_SIZE; +/// 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; +/// Largest logical message this transport will send or reassemble, and the memory an +/// unauthenticated peer can make a receiver hold. Kept at parity with TCP on purpose: one session +/// moves between both paths, so a transport-specific ceiling would kill it on the other one. +const MAX_RECV_MESSAGE: usize = crate::bytes_codec::MAX_FRAME_LENGTH; +// Four networks, not four names: webrtc-ice queries each URL from its own socket, so entries +// sharing a host buy no redundancy - the two this list used to carry failed together, in the same +// millisecond, on a peer whose route to that one host was down. Two anycast, two unicast; the 443 +// entry is for networks that pass no other UDP port. +static DEFAULT_ICE_SERVERS: [&str; 4] = [ "stun:stun.cloudflare.com:3478", - "stun:stun.nextcloud.com:3478", + "stun:stun.l.google.com:19302", + "stun:stun.antisip.com:3478", "stun:stun.nextcloud.com:443", ]; lazy_static::lazy_static! { static ref SESSIONS: Arc::>> = Default::default(); + + // The process-lifetime runtime that owns ALL WebRTC I/O. A pc's UDP sockets register with + // the reactor — and its ICE/DTLS/SCTP pump tasks spawn on the runtime — that is current + // while `new_inner` and the handlers it installs execute; if that were a session's + // `#[tokio::main]` runtime (dropped the moment io_loop returns), the sockets and pumps + // would die with it and even a close driven elsewhere would "succeed" without ever putting + // close_notify on the wire — the peer then waits out ICE decay. So `new()` runs its body + // here, and detached closes run here too: as independent tasks (a deadline-free `close()` + // must not head-of-line block other sessions' teardown) that are never cancelled (a close + // cancelled after `RTCPeerConnection::close` latches `is_closed` is unretryable, stranding + // the pc in SESSIONS). Data-plane futures (send/recv/close) may be polled from any caller + // runtime — cross-runtime polling is fine; only the OWNING driver must stay alive. Two + // workers so one busy session cannot starve the rest. This is the one deliberate exception + // to AGENTS.md's no-runtime-in-libraries rule: a single lazily-built runtime for I/O whose + // lifetime exceeds any caller's. + static ref WEBRTC_RT: Option = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_name("webrtc-io") + .enable_all() + .build() + .map_err(|err| log::error!("failed to build the WebRTC I/O runtime: {}", err)) + .ok(); +} + +/// Hands a stream from `new_inner` (running detached on `WEBRTC_RT`) back to its `new()` +/// caller. If that caller was cancelled while awaiting, the runtime drops the task's output — +/// this guard — and an unanswered offerer never reaches a terminal ICE state on its own, so +/// without the Drop-close its pc, sockets and pump tasks would sit in SESSIONS forever. Claims +/// are counted in the shared `HandoffState`, so only the last one dropped with the stream still +/// unadopted closes it; a hit another caller is already using is left alone. +struct NewStreamHandoff { + stream: Option, +} + +impl NewStreamHandoff { + /// Take a claim on `stream`. Call under the `SESSIONS` lock: a claim registered after the map + /// is unlocked can land behind the abandon-close of the claim that was last outstanding. + fn claim(stream: WebRTCStream) -> Self { + stream.handoff.pending.fetch_add(1, Ordering::AcqRel); + Self { + stream: Some(stream), + } + } + + fn into_inner(mut self) -> Option { + let stream = self.stream.take()?; + // Before releasing the claim, or a concurrent Drop can see the count reach zero without + // seeing that this caller took ownership. + stream.handoff.adopted.store(true, Ordering::Release); + stream.handoff.pending.fetch_sub(1, Ordering::AcqRel); + Some(stream) + } +} + +impl Drop for NewStreamHandoff { + fn drop(&mut self) { + if let Some(stream) = self.stream.take() { + let last = stream.handoff.pending.fetch_sub(1, Ordering::AcqRel) == 1; + if last && !stream.handoff.adopted.load(Ordering::Acquire) { + stream.close_if_abandoned(); + } + } + } } impl Clone for WebRTCStream { @@ -59,7 +229,16 @@ impl Clone for WebRTCStream { pc: self.pc.clone(), stream: self.stream.clone(), state_notify: self.state_notify.clone(), + 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(), + send_gate: self.send_gate.clone(), + recv_state: self.recv_state.clone(), + peer_verified: self.peer_verified.clone(), + handoff: self.handoff.clone(), + local_endpoint: self.local_endpoint.clone(), } } } @@ -94,6 +273,13 @@ impl WebRTCStream { format!("webrtc://{}", encoded_sdp) } + // Envelope JSON key carrying the local description's ICE transport policy, alongside the + // RTCSessionDescription fields (see `local_endpoint`). + // `'static` spelled out: an elided lifetime here is a warn-by-default future hard error on + // the 1.75 toolchain CI pins (elided_lifetimes_in_associated_constant). + const ICE_POLICY_KEY: &'static str = "ice_policy"; + const ICE_POLICY_ALL: &'static str = "all"; + #[inline] fn get_key_for_sdp(sdp: &RTCSessionDescription) -> ResultType { let binding = sdp.unmarshal()?; @@ -117,12 +303,52 @@ 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 + ) + } + + /// Reject a data channel from inside `on_data_channel`, where closing it directly does not + /// work: webrtc-rs runs that handler to completion BEFORE `handle_open` binds the SCTP stream, + /// so `close()` only flips the ready state and `handle_open` then sets it back to Open. + fn close_unbound_channel(dc: Arc) { + let dc_for_close = dc.clone(); + dc.on_open(Box::new(move || { + let dc = dc_for_close.clone(); + Box::pin(async move { + if let Err(err) = dc.close().await { + log::debug!("failed to close rejected data channel: {}", err); + } + }) + })); + } + + /// Whether a cached `SESSIONS` entry may be handed to a caller asking for `force_relay`: it + /// was built for the first caller, so its ICE policy may not match, and the state handler + /// evicts late enough that dead entries linger. Both the lookup and the insert must use it. + fn is_reusable_for(&self, force_relay: bool) -> bool { + self.relay_only == force_relay + && !self.handoff.closing.load(Ordering::Acquire) + && !matches!( + *self.state_notify.borrow(), + WebRTCConnectionState::Closed(_) + ) + } + #[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) } @@ -139,38 +365,118 @@ impl WebRTCStream { #[inline] fn get_ice_server_from_url(url: &str) -> Option { - // standard url format with turn scheme: turn://user:pass@host:port - match Url::parse(url) { - Ok(u) => { - if u.scheme() == "turn" - || u.scheme() == "turns" - || u.scheme() == "stun" - || u.scheme() == "stuns" - { - Some(RTCIceServer { - urls: vec![format!( - "{}:{}:{}", - u.scheme(), - u.host_str().unwrap_or_default(), - u.port().unwrap_or(3478) - )], - username: u.username().to_string(), - credential: u.password().unwrap_or_default().to_string(), - ..Default::default() - }) - } else { + let u = Url::parse(url).ok()?; + if !matches!(u.scheme(), "turn" | "turns" | "stun" | "stuns") { + return None; + } + // Two accepted spellings parse differently: `turn://user:pass@host:port` has an authority + // and is the only form that can carry credentials, while the RFC 7065 `turn:host:port` is + // cannot-be-a-base, so `host_str()` is None and the whole `host:port` lands in the path. + let (host, port) = match u.host_str() { + Some(host) => (host.to_owned(), u.port().unwrap_or(3478)), + None => Self::split_host_port(u.path())?, + }; + if host.is_empty() { + return None; + } + let username = u.username().to_string(); + let credential = u.password().unwrap_or_default().to_string(); + // A TURN server without credentials is not merely useless: webrtc-rs validates every + // configured server in `new_peer_connection`, so one credential-less entry makes EVERY + // peer connection fail — including plain non-relay ones that never wanted TURN. The + // RFC 7065 spelling has nowhere to put credentials, so drop such entries here rather + // than let them poison the whole configuration; `turn://user:pass@host:port` carries them. + if matches!(u.scheme(), "turn" | "turns") && (username.is_empty() || credential.is_empty()) + { + log::warn!( + "Ignoring TURN server without credentials: {}:{}:{} (use {}://user:pass@host:port)", + u.scheme(), + host, + port, + u.scheme() + ); + return None; + } + Some(RTCIceServer { + urls: vec![format!("{}:{}:{}", u.scheme(), host, port)], + username, + credential, + ..Default::default() + }) + } + + /// Split `host[:port]` from an RFC 7065 URL path, defaulting the port. Bracketed IPv6 + /// literals keep their brackets, which is the form webrtc-rs re-parses. + fn split_host_port(path: &str) -> Option<(String, u16)> { + let rest = path.split(['?', '#']).next().unwrap_or_default(); + if rest.is_empty() { + return None; + } + if let Some(after_open) = rest.strip_prefix('[') { + let (host, tail) = after_open.split_once(']')?; + let port = tail + .strip_prefix(':') + .and_then(|p| p.parse().ok()) + .unwrap_or(3478); + return Some((format!("[{host}]"), port)); + } + // An unbracketed IPv6 literal has no unambiguous split point — `2001:db8::1` would be cut + // at its last colon into host `2001:db8:` port `1` — so require the bracketed form for + // those rather than emit a host webrtc-ice cannot resolve. + if rest.matches(':').count() > 1 { + log::warn!("Ignoring ICE server {rest}: bracket IPv6 literals as [addr]:port"); + return None; + } + match rest.rsplit_once(':') { + Some((host, port)) if !host.is_empty() => match port.parse() { + Ok(port) => Some((host.to_owned(), port)), + // A port that is present but unusable is a typo, not a host: folding it back in + // would produce `host:99999:3478`, which webrtc-ice rejects outright — taking + // every peer connection down with it, not just this server. + Err(_) => { + log::warn!("Ignoring ICE server {rest}: invalid port"); None } - } - Err(_) => None, + }, + _ => Some((rest.to_owned(), 3478)), } } + /// 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 { + // `get_ice_server_from_url` is what makes a bare scheme test sufficient: it drops entries + // with no host and TURN entries with no credentials, i.e. exactly the ones that would + // answer `true` here while being unusable — which is worse than answering `false`, since + // the caller skips its "don't build a guaranteed-dead Relay-only pc" guard on our word. + 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(); - let cfg = config::Config::get_option(config::keys::OPTION_ICE_SERVERS); + Self::parse_ice_servers(&config::Config::get_option( + config::keys::OPTION_ICE_SERVERS, + )) + } + /// The default UDP STUN servers as bare `host:port`, for application-level address probes. + pub fn default_stun_servers() -> Vec { + DEFAULT_ICE_SERVERS + .iter() + .filter_map(|url| url.strip_prefix("stun:").map(str::to_owned)) + .collect() + } + + /// Split out from `get_ice_servers` so parsing can be exercised without touching the + /// process-global, on-disk-persisted option — the tests run in parallel threads of one + /// process, so a test that rewrote it raced every peer connection another test was building. + fn parse_ice_servers(cfg: &str) -> Vec { + let mut ice_servers = Vec::new(); let mut has_stun = false; for url in cfg.split(',').map(str::trim) { @@ -201,33 +507,81 @@ impl WebRTCStream { ice_servers } + /// Built and driven on `WEBRTC_RT`: every socket and background task the pc creates must + /// belong to a runtime that outlives the session (see `WEBRTC_RT`). A caller that abandons + /// this future mid-await leaves the setup task to finish there; the abandoned result then + /// closes a freshly built pc (and leaves a shared cache hit alone) — see `NewStreamHandoff`. pub async fn new( remote_endpoint: &str, force_relay: bool, ms_timeout: u64, ) -> ResultType { - log::debug!("New webrtc stream to endpoint: {}", remote_endpoint); + let Some(rt) = WEBRTC_RT.as_ref() else { + return Err(anyhow::anyhow!("WebRTC I/O runtime unavailable")); + }; + let remote_endpoint = remote_endpoint.to_owned(); + rt.spawn(Self::new_inner(remote_endpoint, force_relay, ms_timeout)) + .await + .map_err(|err| anyhow::anyhow!("WebRTC setup task failed: {}", err))?? + .into_inner() + .ok_or_else(|| anyhow::anyhow!("WebRTC setup handoff was empty")) + } + + async fn new_inner( + remote_endpoint: String, + force_relay: bool, + ms_timeout: u64, + ) -> ResultType { + // 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 { - Self::get_remote_offer(remote_endpoint)? + Self::get_remote_offer(&remote_endpoint)? }; 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() { + // Claim inside the lock: `close_if_abandoned` commits under it too, so this hit + // either registers in time to call that close off, or finds the entry already rejected. + let cached = { + let sessions_lock = SESSIONS.lock().await; + match sessions_lock.get(&Self::cache_key(&key, start_local_offer)) { + Some(session) if session.is_reusable_for(force_relay) => { + Some(NewStreamHandoff::claim(session.clone())) + } + Some(session) => { + log::debug!( + "Ignoring cached webrtc peer (relay_only {}, wanted {})", + session.relay_only, + force_relay + ); + None + } + None => None, + } + }; + if let Some(handoff) = cached { log::debug!("Start webrtc with cached peer"); - return Ok(cached_stream.clone()); + return Ok(handoff); } } - 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(); s.set_ice_multicast_dns_mode(MulticastDnsMode::Disabled); + // fe80::/10 can only be bound together with a scope id, which `IpAddr` cannot carry, so + // gathering one never yields a candidate - only a failed bind and a warning per address. + // Spelled out because `is_unicast_link_local` is not stable on our MSRV. + s.set_ip_filter(Box::new(|ip: IpAddr| match ip { + IpAddr::V6(v6) => v6.segments()[0] & 0xffc0 != 0xfe80, + IpAddr::V4(_) => true, + })); // Create the API object let api = APIBuilder::new().with_setting_engine(s).build(); @@ -243,16 +597,50 @@ impl WebRTCStream { ..Default::default() }; - let (notify_tx, notify_rx) = watch::channel(false); + let (notify_tx, notify_rx) = watch::channel(WebRTCConnectionState::Pending); + let (ice_tx, ice_rx) = mpsc::unbounded_channel::(); // Create a new RTCPeerConnection let pc = Arc::new(api.new_peer_connection(config).await?); + // `on_*` handlers are never cleared — not by `close()`, not by `SESSIONS` eviction — so + // this sender outlives the pc and would park a caller's ICE forwarder on `recv()` forever, + // holding the pc with it. The terminal-state handler below drops this closure to break it. + let local_ice_tx = ice_tx.clone(); + pc.on_ice_candidate(Box::new(move |candidate| { + let local_ice_tx = local_ice_tx.clone(); + Box::pin(async move { + let Some(candidate) = candidate else { + return; + }; + match candidate.to_json() { + Ok(candidate) => match serde_json::to_string(&candidate) { + Ok(candidate_json) => { + let _ = local_ice_tx.send(candidate_json); + } + Err(err) => { + log::warn!("failed to serialize local ICE candidate: {}", err); + } + }, + Err(err) => { + log::warn!("failed to convert local ICE candidate to JSON: {}", err); + } + } + }) + })); + 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(true); + let _ = dc_open_notify.send(WebRTCConnectionState::Open); Box::pin(async {}) })); dc @@ -267,17 +655,44 @@ impl WebRTCStream { // Register data channel creation handling let dc_open_notify = notify_tx.clone(); let stream_for_dc = stream.clone(); + // The remote may open any number of channels; only the first is ever bound. + let dc_bound = Arc::new(AtomicBool::new(false)); pc.on_data_channel(Box::new(move |dc: Arc| { let d_label = dc.label().to_owned(); let dc_open_notify2 = dc_open_notify.clone(); let stream_for_dc_clone = stream_for_dc.clone(); - log::debug!("Remote data channel {} ready", d_label); + let dc_bound = dc_bound.clone(); Box::pin(async move { + // Reassembly is sound only on an ordered, fully-reliable channel: the fragment + // header has no sequence number, so a reorder or a loss silently merges + // messages. webrtc-rs takes these from the unauthenticated REMOTE's DCEP OPEN. + if !dc.ordered() + || dc.max_retransmits().is_some() + || dc.max_packet_lifetime().is_some() + { + log::warn!( + "Rejecting WebRTC data channel {}: not ordered and fully reliable", + d_label + ); + Self::close_unbound_channel(dc); + return; + } + // Bind the first channel only. `detached` caches the first detached handle + // for the life of the stream, so rebinding would leave teardown closing a + // channel that send/recv no longer use; worse, a second channel's `on_open` + // would push Open onto the same watch and re-arm a session already latched + // Closed — and that watch gates both `send_bytes_inner` and `next()`. + if dc_bound.swap(true, Ordering::SeqCst) { + log::warn!("Ignoring extra WebRTC data channel {}", d_label); + Self::close_unbound_channel(dc); + return; + } + log::debug!("Remote data channel {} ready", d_label); let mut stream_lock = stream_for_dc_clone.lock().await; *stream_lock = dc.clone(); drop(stream_lock); dc.on_open(Box::new(move || { - let _ = dc_open_notify2.send(true); + let _ = dc_open_notify2.send(WebRTCConnectionState::Open); Box::pin(async {}) })); }) @@ -286,7 +701,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(); @@ -294,19 +714,40 @@ impl WebRTCStream { Box::pin(async move { log::debug!("WebRTC session peer connection state: {}", s); match s { - RTCPeerConnectionState::Disconnected - | RTCPeerConnectionState::Failed - | RTCPeerConnectionState::Closed => { - let _ = on_connection_notify.send(true); - log::debug!("WebRTC session closing due to disconnected"); + // `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 {}", 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; + }; + + // Nothing else ever drops this sender (`close()` clears no handler), so + // replace the handler to close the channel: a caller's forwarder loop then + // ends instead of parking on `recv()` and holding this pc alive. + pc_for_close2.on_ice_candidate(Box::new(|_| Box::pin(async {}))); + 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!( @@ -336,67 +777,364 @@ impl WebRTCStream { }) })); - // process offer/answer - if start_local_offer { - let sdp = pc.create_offer(None).await?; - let mut gather_complete = pc.gathering_complete_promise().await; - pc.set_local_description(sdp.clone()).await?; - let _ = gather_complete.recv().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?; - let mut gather_complete = pc.gathering_complete_promise().await; - pc.set_local_description(answer).await?; - let _ = gather_complete.recv().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); + // Trickle ICE: local-only work, no gathering wait — the controlled side awaits answer + // creation inline on its punch-reply critical path. A failure below leaves a live pc whose + // state handler only fires on a terminal ICE state, so close before propagating. + // Encode each endpoint before `set_local_description`, which is what starts gathering, so + // there is nothing for `trickle_endpoint` to strip in the first place. + let offer_answer: ResultType<(String, String)> = async { + if start_local_offer { + let sdp = pc.create_offer(None).await?; + let endpoint = Self::trickle_endpoint(&sdp, force_relay)?; + 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, endpoint {} bytes)", + sdp.sdp.len(), + endpoint.len() + ); + let k = Self::get_key_for_sdp(&sdp)?; + log::debug!("Start webrtc with local key: {}", k); + Ok((k, endpoint)) + } else { + let sdp = serde_json::from_str::(&remote_offer)?; + pc.set_remote_description(sdp.clone()).await?; + let answer = pc.create_answer(None).await?; + let endpoint = Self::trickle_endpoint(&answer, force_relay)?; + pc.set_local_description(answer).await?; + log::debug!( + "remote offer SDP received ({} bytes), local answer endpoint {} bytes", + sdp.sdp.len(), + endpoint.len() + ); + let k = Self::get_key_for_sdp(&sdp)?; + log::debug!("Start webrtc with remote key: {}", k); + Ok((k, endpoint)) + } } - - let mut final_lock = SESSIONS.lock().await; - if let Some(session) = final_lock.get(&key) { - pc.close().await.ok(); - return Ok(session.clone()); + .await; + let (new_key, local_endpoint) = match offer_answer { + Ok(x) => x, + Err(e) => { + pc.close().await.ok(); + return Err(e); + } + }; + // Only the UDP leg has a ceiling, and which leg carries this is the rendezvous server's + // to decide, so this reports rather than refuses: over a TCP/WS route a larger endpoint is + // delivered fine, and failing here would cost WebRTC for no reason. + if local_endpoint.len() > Self::UDP_ENDPOINT_BUDGET { + log::warn!( + "WebRTC endpoint is {} bytes, over the {} byte UDP budget; a peer reached over UDP \ + may never receive it", + local_endpoint.len(), + Self::UDP_ENDPOINT_BUDGET + ); } + key = new_key; let webrtc_stream = Self { pc, stream, state_notify: notify_rx, + 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)), + send_gate: Arc::new(Semaphore::new(1)), + recv_state: Arc::new(Mutex::new(RecvState::default())), + peer_verified: Arc::new(AtomicBool::new(false)), + handoff: Default::default(), + local_endpoint: Arc::new(local_endpoint), + }; + // 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); + // Claim inside the lock, the insert included: with the pc cached but unclaimed, a hit + // could take the only claim on it and release it again, closing it under this caller. + let mut duplicate = None; + let handoff = { + let mut final_lock = SESSIONS.lock().await; + // Same admissibility test as the lookup above, or that lookup is dead code: an entry + // rejected there is still in the map when we get here, so returning it unconditionally + // would discard the pc we just built precisely because the cached one was unusable. + match final_lock.get(&cache_key) { + Some(session) if session.is_reusable_for(force_relay) => { + duplicate = Some(webrtc_stream.clone()); + NewStreamHandoff::claim(session.clone()) + } + _ => { + final_lock.insert(cache_key, webrtc_stream.clone()); + NewStreamHandoff::claim(webrtc_stream) + } + } }; - final_lock.insert(key, webrtc_stream.clone()); - Ok(webrtc_stream) + if let Some(loser) = duplicate { + // A concurrent `new()` already cached an equivalent stream; discard this pc's + // resources (on the closer thread — awaiting here would strand the pc if this + // `new()` were cancelled mid-close) and return the cached one. + loser.close_detached(); + } + Ok(handoff) } + /// 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 { - if let Some(local_desc) = self.pc.local_description().await { - let sdp = serde_json::to_string(&local_desc)?; - let endpoint = Self::sdp_to_endpoint(&sdp); - Ok(endpoint) - } else { - Err(anyhow::anyhow!("Local desc is not set")) + // Preserve the original one-shot endpoint contract: callers that only exchange this SDP + // do not have a separate path for `take_local_ice_rx`, so their endpoint must contain the + // gathered host/srflx/relay candidates. + let mut gather_complete = self.pc.gathering_complete_promise().await; + let _gathering_channel_closed = gather_complete.recv().await; + let Some(local_desc) = self.pc.local_description().await else { + return Err(anyhow::anyhow!("Local desc is not set")); + }; + Self::encode_endpoint(local_desc.sdp_type, &local_desc.sdp, self.relay_only) + } + + /// The offer/answer exactly as it was created, for callers that signal candidates via + /// `take_local_ice_rx`. Taken at construction, so this cannot fail and cannot block — unlike + /// `get_local_endpoint`, which reads the live description and waits for gathering. + /// + /// Deliberately not `pc.local_description()`, which runs `populate_local_candidates` and so + /// returns the SDP plus every candidate gathered up to that moment — and callers read this a + /// network round trip after `new`, by which time gathering has filled it in. Why the size + /// matters: `UDP_ENDPOINT_BUDGET`. + #[inline] + pub fn local_endpoint(&self) -> &str { + self.local_endpoint.as_str() + } + + /// The size a trickle endpoint must stay under for the UDP leg of its route to be safe. It is + /// the one signaling message that can grow: a peer registered over UDP receives it inside + /// `PunchHole` — with the mangled addresses and the permission blobs — as a single datagram, + /// and one that needs IP fragmentation is dropped without a trace on paths that discard + /// fragments. The candidates that follow are one small message each and never approach it. + /// + /// Not an MTU calculation: `PunchHole`'s other fields are not bounded here, so a value near + /// this number could still fragment. It is a tripwire on a quantity `trickle_endpoint` pins at + /// ~700 bytes, sized to leave that headroom while still catching a regression that lets + /// candidates back in — as a log line rather than another silent drop. Nothing enforces it: a + /// TCP/WS leg has no packet ceiling. + const UDP_ENDPOINT_BUDGET: usize = 1024; + + /// The endpoint a trickling peer signals: the session parameters needed to start ICE and DTLS + /// (ice-ufrag, ice-pwd, fingerprint, setup, the sctp m-line), without the candidates — those + /// ride `take_local_ice_rx` and arrive as individual `IceCandidate` messages. + /// + /// Candidates are the only part of a local description that grows. Stripping them makes the + /// size a property of the value rather than of when it was taken, so it cannot drift with + /// gathering however this is later refactored. + fn trickle_endpoint(local_desc: &RTCSessionDescription, relay_only: bool) -> ResultType { + let mut sdp = String::with_capacity(local_desc.sdp.len()); + for line in local_desc.sdp.lines() { + // `end-of-candidates` would tell the remote agent to stop waiting for the trickle. + if line.starts_with("a=candidate:") || line.starts_with("a=end-of-candidates") { + continue; + } + sdp.push_str(line); + sdp.push_str("\r\n"); } + Self::encode_endpoint(local_desc.sdp_type, &sdp, relay_only) + } + + /// Base64 envelope of a local description, carrying its ICE transport policy unless the pc is + /// Relay-only (see `ICE_POLICY_KEY`). + /// Built from the two fields a session description serializes rather than from the value, so + /// a caller that rewrote the SDP cannot leave a stale `parsed` tree riding along with it. + fn encode_endpoint(sdp_type: RTCSdpType, sdp: &str, relay_only: bool) -> ResultType { + let mut envelope = serde_json::Map::new(); + envelope.insert("type".to_owned(), serde_json::to_value(sdp_type)?); + envelope.insert("sdp".to_owned(), serde_json::Value::from(sdp)); + if !relay_only { + // Rides in the envelope, not a proto field the rendezvous server would forward, + // because it is the offer's own property: the receiver must know whether + // force_relay was policy (stay Relay-only) or transport. Older peers ignore it. + envelope.insert( + Self::ICE_POLICY_KEY.to_owned(), + serde_json::Value::from(Self::ICE_POLICY_ALL), + ); + } + Ok(Self::sdp_to_endpoint(&serde_json::to_string(&envelope)?)) + } + + /// Whether the peer's endpoint declares it was built with ICE transport policy `all` + /// (W3C RTCIceTransportPolicy), i.e. it gathers host/srflx/relay candidates and a direct + /// pair may form even though the request carries force_relay. Absent key, foreign format + /// or parse failure all mean "not declared" — the old Relay-only reading. + pub fn endpoint_declares_all_ice(endpoint: &str) -> bool { + let Ok(sdp_json) = Self::get_remote_offer(endpoint) else { + return false; + }; + serde_json::from_str::(&sdp_json) + .ok() + .and_then(|v| { + v.get(Self::ICE_POLICY_KEY)? + .as_str() + .map(|p| p == Self::ICE_POLICY_ALL) + }) + .unwrap_or(false) } #[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 connection runs through a TURN relay; feeds the UI's direct/relayed flag. + /// Under Relay policy the answer is known by construction, so that arm returns `Some(true)` + /// 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 { + if self.relay_only { + return Some(true); + } + let dtls = self.pc.sctp().transport(); + let pair = dtls.ice_transport().get_selected_candidate_pair().await?; + + // Not the stats report's `nominated` flag: webrtc-ice never clears it per checklist entry, + // so after a pair switch several entries carry it and map order decides the answer. + // 0.13 keeps the candidates private, so read the types out of `Display` by position. + let relay = RTCIceCandidateType::Relay.to_string(); + Some( + pair.to_string() + .split(" <-> ") + .filter_map(|side| side.split_whitespace().nth(2)) + .any(|typ| typ == relay), + ) + } + + /// Whether the nominated pair reaches the peer over IPv6 — `None` before one is selected. + /// The remote side on purpose: it is the address the peer is actually reached at, which the + /// rendezvous-observed address the session is otherwise identified by cannot report. + pub async fn is_remote_ipv6(&self) -> Option { + let dtls = self.pc.sctp().transport(); + let pair = dtls.ice_transport().get_selected_candidate_pair().await?; + // Same `Display` parse as `is_relayed`, one token over: a side renders as + // `(remote)
:`, and `protocol` is only udp/tcp, so the + // family has to come from the address — an IPv6 literal is the only one with two colons. + let pair = pair.to_string(); + let remote = pair.split(" <-> ").nth(1)?; + Some(remote.split_whitespace().nth(3)?.matches(':').count() > 1) + } + + #[inline] + pub fn take_local_ice_rx(&self) -> Option> { + self.local_ice_rx.lock().ok().and_then(|mut rx| rx.take()) + } + + #[inline] + pub async fn add_remote_ice_candidate(&self, candidate_json: &str) -> ResultType<()> { + if candidate_json.is_empty() { + return Ok(()); + } + let candidate = serde_json::from_str::(candidate_json)?; + self.pc.add_ice_candidate(candidate).await?; + Ok(()) + } + + #[inline] + pub fn session_key(&self) -> &str { + &self.session_key + } + + pub async fn wait_connected(&mut self, ms: u64) -> ResultType<()> { + if ms > 0 { + match timeout(Duration::from_millis(ms), self.wait_for_connect_result()).await { + Ok(result) => result?, + Err(_) => return Err(anyhow::anyhow!("WebRTC wait_connected timeout")), + } + } else { + self.wait_for_connect_result().await?; + } + Ok(()) + } + + /// 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 + /// terminal ICE state. Closing fires the state handler, which evicts the `SESSIONS` entry — + /// callers that abandon a stream (e.g. an offerer that lost the transport race) must call it. + #[inline] + pub async fn close(&self) { + self.pc.close().await.ok(); + } + + /// Tear the pc down as an independent task on `WEBRTC_RT` — the runtime that owns the + /// pc's sockets and pump tasks, so the close can actually reach the wire no matter which + /// caller runtimes have died — keeping `keep` alive until the teardown finishes. Callable + /// from any context, `Drop` on a runtime-less thread included, and never cancelled, so + /// `close()`'s `is_closed` latch is only ever set by a close that finishes. The send path + /// passes its permit as `keep`. + pub fn close_detached_with(&self, keep: T) { + // If the runtime never built, no pc exists either (`new` fails first); nothing to close. + let Some(rt) = WEBRTC_RT.as_ref() else { + return; + }; + let pc = self.pc.clone(); + rt.spawn(async move { + if let Err(err) = pc.close().await { + log::debug!("WebRTC close failed: {}", err); + } + drop(keep); + }); + } + + #[inline] + pub fn close_detached(&self) { + self.close_detached_with(()); + } + + /// Close a pc no caller ever took. Spawned on `WEBRTC_RT` like `close_detached`, but the + /// decision is remade under the `SESSIONS` lock that a cache hit claims under: either the hit + /// registers first and this call stands down, or `closing` commits first and + /// `is_reusable_for` rejects the entry, which the state handler evicts only later. + fn close_if_abandoned(&self) { + let Some(rt) = WEBRTC_RT.as_ref() else { + return; + }; + let stream = self.clone(); + rt.spawn(async move { + { + let _sessions_lock = SESSIONS.lock().await; + if stream.handoff.pending.load(Ordering::Acquire) != 0 + || stream.handoff.adopted.load(Ordering::Acquire) + { + return; + } + stream.handoff.closing.store(true, Ordering::Release); + } + // Off-lock: `close()` fires the state handler inline and that handler locks SESSIONS. + if let Err(err) = stream.pc.close().await { + log::debug!("WebRTC close of an abandoned peer failed: {}", err); + } + }); + } + #[inline] pub fn set_raw(&mut self) { // not-supported @@ -414,14 +1152,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] @@ -435,66 +1175,254 @@ impl WebRTCStream { } #[inline] - async fn wait_for_connect_result(&mut self) { - if *self.state_notify.borrow() { - return; + async fn wait_for_connect_result(&mut self) -> ResultType<()> { + loop { + match self.state_notify.borrow().clone() { + WebRTCConnectionState::Open => return Ok(()), + WebRTCConnectionState::Closed(reason) => { + return Err(anyhow::anyhow!("WebRTC connection closed: {}", reason)); + } + WebRTCConnectionState::Pending => {} + } + self.state_notify.changed().await?; + } + } + + /// 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()); } - let _ = self.state_notify.changed().await; + *cache = Some(dc.clone()); + Ok(dc) } + /// NOT cancel-safe: dropping this future mid-message (e.g. wrapping it in `select!`/`timeout`) + /// can leave a partial fragment sequence on the wire, corrupting reassembly of every later + /// message on this stream. A caller that abandons a send must treat the stream as dead and + /// close it; the built-in `send_timeout` path below already does (it closes the pc). pub async fn send_bytes(&mut self, bytes: Bytes) -> ResultType<()> { - if self.send_timeout > 0 { - match timeout( - Duration::from_millis(self.send_timeout), + let send_timeout = self.send_timeout; + let send_gate = self.send_gate.clone(); + // Bound the WHOLE send (wait-for-open, queueing behind another clone, every write) by + // send_timeout: otherwise a write parks indefinitely on SCTP backpressure and + // connection.rs's timer never runs. That parking is also what bounds sender memory — + // PendingQueue admits 128 KiB and inflight is cwnd/rwnd-capped — so this timeout is the + // TCP-send-timeout equivalent. See the checklist entry on send backpressure. + if send_timeout > 0 { + // Waiting for the connection to open is not this message's progress. Charging it to + // the send clock meant that on the first send after signaling, a slow ICE/DTLS + // completion ate most of the budget and the write inherited the remainder — then + // timed out and closed a peer connection that was an RTT from working. Give it its + // own budget and start the send clock once the channel is actually usable. + timeout( + Duration::from_millis(send_timeout), self.wait_for_connect_result(), ) .await - { - Ok(_) => {} - Err(_) => { - self.pc.close().await.ok(); + .map_err(|_| Error::new(ErrorKind::TimedOut, "WebRTC connect timeout"))??; + + let gate_deadline = Instant::now() + Duration::from_millis(send_timeout); + let send_permit = match timeout_at(gate_deadline, send_gate.acquire_owned()).await { + Ok(Ok(permit)) => permit, + Ok(Err(err)) => { return Err(Error::new( - ErrorKind::TimedOut, - "WebRTC send wait for connect timeout", + ErrorKind::BrokenPipe, + format!("WebRTC send gate closed: {}", err), ) .into()); } + Err(_) => { + // Deliberately no teardown: this task never held the permit, so all that + // happened is that another clone is legitimately mid-message. Closing from + // here would abort a healthy sender's fragment sequence — precisely the + // corruption the permit exists to prevent. + return Err(Error::new(ErrorKind::TimedOut, "WebRTC send gate timeout").into()); + } + }; + // Fresh budget once the permit is held: queueing behind another clone's message is + // not this message's progress, so charging it here meant a caller that only just lost + // the gate race got a few milliseconds to write in — and then tore the whole peer + // connection down for missing them. The narrower the miss, the more certain the + // teardown, which is precisely backwards. + let write_deadline = Instant::now() + Duration::from_millis(send_timeout); + let mut desynced = false; + match timeout_at(write_deadline, self.send_bytes_inner(bytes, &mut desynced)).await { + // A write that failed part-way leaves the same orphaned prefix a timeout does, so + // it needs the same teardown — and the same handoff of the permit, or a waiting + // clone appends its message to that prefix while the close is still in flight. + Ok(res) => { + if desynced { + self.close_detached_with(send_permit); + } + res + } + Err(_) => { + // Hand the logical-message permit to the teardown so no waiting clone can + // append a new message after a partially-written fragment sequence. Holding + // it across an awaited close would drop both on cancellation. + self.close_detached_with(send_permit); + Err(Error::new(ErrorKind::TimedOut, "WebRTC send timeout").into()) + } } } else { - self.wait_for_connect_result().await; + let send_permit = send_gate.acquire_owned().await.map_err(|err| { + Error::new( + ErrorKind::BrokenPipe, + format!("WebRTC send gate closed: {}", err), + ) + })?; + let mut desynced = false; + let res = self.send_bytes_inner(bytes, &mut desynced).await; + if desynced { + self.close_detached_with(send_permit); + } + res + } + } + + /// `desynced` is set when the failure left a partial fragment sequence on the wire, i.e. when + /// the stream can no longer be made consistent and the caller must close it. + async fn send_bytes_inner(&mut self, bytes: Bytes, desynced: &mut bool) -> ResultType<()> { + // Same bound the receiver enforces, so we never emit a message a same-version peer + // would have to kill the connection over. + if bytes.len() > MAX_RECV_MESSAGE { + 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. + let mut wrote_any = false; + 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); + if let Err(err) = dc.write(&framed.freeze()).await { + // Fragments already on the wire leave the peer an unterminated prefix, and this + // framing has no length for it to notice with, so the next message is appended + // to it. Callers wrap sends in `allow_err!`, so the error alone is not enough. + *desynced = wrote_any; + return Err(err.into()); + } + wrote_any = true; + offset = end; + if is_last { + break; + } } - let stream = self.stream.lock().await.clone(); - stream.send(&bytes).await?; Ok(()) } #[inline] pub async fn next(&mut self) -> Option> { - self.wait_for_connect_result().await; - 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, + if let Err(err) = self.wait_for_connect_result().await { + self.close_detached(); + return Some(Err(Error::new(ErrorKind::Other, err.to_string()))); + } + let dc = match self.detached_dc().await { + Ok(dc) => dc, Err(err) => { - self.pc.close().await.ok(); + self.close_detached(); + return Some(Err(Error::new(ErrorKind::Other, err.to_string()))); + } + }; + // Held across `dc.read().await` on purpose, against the usual "no locks across await" + // rule: it is what makes the single reader this reassembly requires (see the field) an + // exclusion rather than a convention. Releasing it around the read would admit exactly + // the second reader that corrupts `acc`. The cost — a would-be second reader blocking + // until a packet arrives — is a state the design does not permit anyway. + let mut st = self.recv_state.lock().await; + loop { + let RecvState { acc, scratch } = &mut *st; + if scratch.len() < RECV_BUF_SIZE { + scratch.resize(SCRATCH_REFILL, 0); + } + let n = match dc.read(&mut scratch[..RECV_BUF_SIZE]).await { + Ok(n) => n, + Err(err) => { + // Release the partial message, as the framing-violation paths below do: the + // buffer can hold up to the cap and `recv_state` outlives this call through + // the `SESSIONS` clone. + *acc = BytesMut::new(); + self.close_detached(); + return Some(Err(Error::new( + ErrorKind::Other, + format!("data channel read error: {}", err), + ))); + } + }; + if n == 0 { + // Not exclusively a stream reset: webrtc-data maps the empty-message PPIDs to + // n == 0 as well and `read` drops the flag that would separate them. Both mean + // nothing frameable follows, so treat them alike but not as a clean remote close. + log::debug!("WebRTC data channel ended (reset or empty message)"); + *acc = BytesMut::new(); + self.close_detached(); + return None; + } + // Both would otherwise read as "more fragments". The payload-less FRAG_MORE is the + // nastier one: it adds nothing to `acc`, so the cap below never trips and the loop + // spins for as long as the peer keeps sending. + 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.close_detached(); return Some(Err(Error::new( - ErrorKind::Other, - format!("data channel read error: {}", err), + ErrorKind::InvalidData, + format!("WebRTC {why}"), ))); } - }; - if n == 0 { - self.pc.close().await.ok(); - return Some(Err(Error::new( - ErrorKind::Other, - "data channel read exited with 0 bytes", - ))); + // Bound BEFORE growing: this framing carries no length, so an oversize message is only + // discovered by accumulating it, and checking after the append would let `BytesMut`'s + // reallocate-and-copy hold the old and the new buffer at once. + if acc.len() + (n - 1) > MAX_RECV_MESSAGE { + // Release the buffer, don't just truncate it: by definition it is near the cap + // here, and `recv_state` outlives this call through the `SESSIONS` clone. + *acc = BytesMut::new(); + self.close_detached(); + return Some(Err(Error::new( + ErrorKind::InvalidData, + "WebRTC reassembled message exceeded maximum frame size", + ))); + } + // A message that arrived whole is handed over as a slice of the read buffer: no + // allocation and no copy, which is every input event and most audio packets. + if header == FRAG_END && acc.is_empty() { + scratch.advance(1); + return Some(Ok(scratch.split_to(n - 1))); + } + acc.extend_from_slice(&scratch[1..n]); + if header == FRAG_END { + let msg = std::mem::take(acc); + return Some(Ok(msg)); + } } - buffer.truncate(n); - Some(Ok(buffer)) } #[inline] @@ -513,48 +1441,47 @@ pub fn is_webrtc_endpoint(endpoint: &str) -> bool { #[cfg(test)] mod tests { - use crate::config; use crate::webrtc::WebRTCStream; - use crate::webrtc::DEFAULT_ICE_SERVERS; + use crate::webrtc::{NewStreamHandoff, DEFAULT_ICE_SERVERS, FRAG_MORE, SESSIONS, WEBRTC_RT}; + use bytes::{BufMut, Bytes, BytesMut}; + use std::{sync::Arc, time::Duration}; + use tokio::sync::Barrier; + use tokio::time::timeout; use webrtc::peer_connection::sdp::session_description::RTCSessionDescription; #[test] fn test_webrtc_ice_url() { - assert_eq!( - WebRTCStream::get_ice_server_from_url("turn://example.com:3478") - .unwrap_or_default() - .urls[0], - "turn:example.com:3478" - ); + let turn = WebRTCStream::get_ice_server_from_url("turn://123:321@example.com:3478") + .expect("credentialed turn is usable"); + assert_eq!(turn.urls[0], "turn:example.com:3478"); + assert_eq!(turn.username, "123"); + assert_eq!(turn.credential, "321"); assert_eq!( - WebRTCStream::get_ice_server_from_url("turn://example.com") + WebRTCStream::get_ice_server_from_url("turn://123:321@example.com") .unwrap_or_default() .urls[0], "turn:example.com:3478" ); - assert_eq!( - WebRTCStream::get_ice_server_from_url("turn://123@example.com") - .unwrap_or_default() - .username, - "123" - ); - - assert_eq!( - WebRTCStream::get_ice_server_from_url("turn://123@example.com") - .unwrap_or_default() - .credential, - "" - ); - - assert_eq!( - WebRTCStream::get_ice_server_from_url("turn://123:321@example.com") - .unwrap_or_default() - .credential, - "321" - ); + // TURN without both halves of the credential is dropped rather than passed on: + // webrtc-rs validates every configured server when the peer connection is built, so + // one such entry fails EVERY connection, including those that never wanted TURN. + for missing in [ + "turn://example.com:3478", + "turn://example.com", + "turn://123@example.com", + "turns://example.com:5349", + "turn:example.com:3478", + ] { + assert_eq!( + WebRTCStream::get_ice_server_from_url(missing), + None, + "credential-less {missing} must not reach the configuration" + ); + } + // STUN needs no credentials, so both spellings stay usable. assert_eq!( WebRTCStream::get_ice_server_from_url("stun://example.com:3478") .unwrap_or_default() @@ -567,28 +1494,199 @@ mod tests { None ); - config::Config::set_option("ice-servers".to_string(), "".to_string()); + // RFC 7065 spelling (`scheme:host:port`, no authority) — what STUN/TURN docs hand out. + // `url` cannot-be-a-base's it, so host and port come out of the path; getting this wrong + // produced a hostless "stun::3478" no ICE agent can resolve. + for (input, expected) in [ + ("stun:example.com:19302", "stun:example.com:19302"), + ("stun:example.com", "stun:example.com:3478"), + ("stun:[2001:db8::1]:19302", "stun:[2001:db8::1]:19302"), + ("stun:[2001:db8::1]", "stun:[2001:db8::1]:3478"), + ( + "stun:example.com:19302?transport=udp", + "stun:example.com:19302", + ), + ] { + assert_eq!( + WebRTCStream::get_ice_server_from_url(input) + .unwrap_or_default() + .urls[0], + expected, + "parsing {input}" + ); + } + + // Malformed entries are dropped, never repaired into something webrtc-ice will choke + // on: a folded-in bad port ("host:99999:3478") or an unbracketed IPv6 split at its last + // colon fails peer-connection construction outright, taking every server down with it. + for bad in [ + "stun:", + "stun:example.com:99999", + "stun:example.com:abc", + "stun:2001:db8::1", + ] { + assert_eq!( + WebRTCStream::get_ice_server_from_url(bad), + None, + "malformed {bad} must be dropped" + ); + } + } + + // Parsing is exercised through `parse_ice_servers`, never by rewriting the global + // `ice-servers` option: `set_option` persists to the real config file and the tests share + // one process, so mutating it raced every peer connection the loopback tests were building. + #[test] + fn test_webrtc_ice_server_list() { assert_eq!( - WebRTCStream::get_ice_servers()[0].urls[0], + WebRTCStream::parse_ice_servers("")[0].urls[0], DEFAULT_ICE_SERVERS[0].to_string() ); - config::Config::set_option( - "ice-servers".to_string(), - ",stun://example.com,turn://example.com,sdf".to_string(), + // Unusable entries drop out of the list; the rest of the config still applies. + let parsed = WebRTCStream::parse_ice_servers( + ",stun://example.com,turn://u:p@example.com,turn://nocreds.example.com,sdf", ); - assert_eq!( - WebRTCStream::get_ice_servers()[0].urls[0], - "stun:example.com:3478" + assert_eq!(parsed[0].urls[0], "stun:example.com:3478"); + assert_eq!(parsed[1].urls[0], "turn:example.com:3478"); + assert_eq!(parsed.len(), 2); + + // TURN-only config still gets the default STUN servers prepended. + let turn_only = WebRTCStream::parse_ice_servers("turn://u:p@example.com:3478"); + assert_eq!(turn_only[0].urls[0], DEFAULT_ICE_SERVERS[0].to_string()); + assert_eq!(turn_only[1].urls[0], "turn:example.com:3478"); + } + + #[test] + fn test_endpoint_ice_policy_declaration() { + // An envelope with the marker declares full ICE; everything else — no marker, + // wrong value, foreign scheme, garbage — reads as the old Relay-only semantics. + let marked = + WebRTCStream::sdp_to_endpoint(r#"{"type":"offer","sdp":"v=0","ice_policy":"all"}"#); + assert!(WebRTCStream::endpoint_declares_all_ice(&marked)); + + let unmarked = WebRTCStream::sdp_to_endpoint(r#"{"type":"offer","sdp":"v=0"}"#); + assert!(!WebRTCStream::endpoint_declares_all_ice(&unmarked)); + + let wrong = + WebRTCStream::sdp_to_endpoint(r#"{"type":"offer","sdp":"v=0","ice_policy":"relay"}"#); + assert!(!WebRTCStream::endpoint_declares_all_ice(&wrong)); + + assert!(!WebRTCStream::endpoint_declares_all_ice("")); + assert!(!WebRTCStream::endpoint_declares_all_ice( + "webrtc://not-base64!" + )); + assert!(!WebRTCStream::endpoint_declares_all_ice( + "https://example.com" + )); + + // The marker must be invisible to the plain RTCSessionDescription parse old peers do. + let sdp_json = WebRTCStream::get_remote_offer(&marked).unwrap(); + serde_json::from_str::< + webrtc::peer_connection::sdp::session_description::RTCSessionDescription, + >(&sdp_json) + .expect("extra envelope key must not break RTCSessionDescription parsing"); + } + + /// The live description once it has gathered something. Polls instead of awaiting + /// `gathering_complete_promise`: a stream sits in the global `SESSIONS` for the whole wait, and + /// waiting out gathering there is long enough to trip `test_cancelled_new_does_not_leak_the_pc` + /// when the suite runs its tests in parallel. One candidate is all these tests need. + /// Returns `None` on timeout rather than panicking: the caller must get to `close()` before + /// it asserts, or a stranded pc fails `test_cancelled_new_does_not_leak_the_pc` as well and + /// one root failure is reported as two. + async fn first_gathered_description(stream: &WebRTCStream) -> Option { + timeout(Duration::from_secs(10), async { + loop { + if let Some(desc) = stream.pc.local_description().await { + if desc.sdp.contains("a=candidate:") { + return desc; + } + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .ok() + } + + /// The signalled endpoint and the live description must diverge. `pc.local_description()` + /// appends every candidate gathered so far; the endpoint the peer is handed must not grow with + /// them, because the rendezvous server delivers it to a UDP-registered peer as one datagram + /// and a fragmented one is dropped without a trace on paths that discard fragments. + #[tokio::test] + async fn test_stored_endpoint_does_not_grow_with_the_live_description() { + let offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + let live = first_gathered_description(&offerer).await; + let endpoint = offerer.local_endpoint().to_owned(); + offerer.close().await; + + // Some(_) is the assertion that the live description grew: the helper only returns once it + // carries a candidate, so the endpoint below really had something to grow by. + live.expect("no ICE candidate gathered within 10s"); + assert!( + !WebRTCStream::get_remote_offer(&endpoint) + .unwrap() + .contains("a=candidate"), + "the signalled endpoint grew candidates alongside the live description" ); - assert_eq!( - WebRTCStream::get_ice_servers()[1].urls[0], - "turn:example.com:3478" + assert!( + endpoint.len() <= WebRTCStream::UDP_ENDPOINT_BUDGET, + "signalled endpoint is {} bytes, over the {} byte budget", + endpoint.len(), + WebRTCStream::UDP_ENDPOINT_BUDGET + ); + } + + /// The split itself: candidates out, everything ICE and DTLS need to start in. A stripped + /// endpoint the peer cannot consume would trade a silent drop for a silent failure. + #[tokio::test] + async fn test_trickle_endpoint_splits_candidates_off() { + let offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + let gathered = first_gathered_description(&offerer).await; + offerer.close().await; + let gathered = gathered.expect("no ICE candidate gathered within 10s"); + + let endpoint = WebRTCStream::trickle_endpoint(&gathered, false).unwrap(); + let json = WebRTCStream::get_remote_offer(&endpoint).unwrap(); + assert!(!json.contains("a=candidate:"), "candidate survived the split"); + assert!( + !json.contains("a=end-of-candidates"), + "end-of-candidates would stop the remote agent waiting for the trickle" + ); + for keep in [ + "m=application", + "a=ice-ufrag:", + "a=ice-pwd:", + "a=fingerprint:", + "a=setup:", + "a=mid:", + "a=sctp-port:", + ] { + assert!(json.contains(keep), "trickle endpoint dropped {}", keep); + } + assert!( + endpoint.len() <= WebRTCStream::UDP_ENDPOINT_BUDGET, + "stripped endpoint is {} bytes", + endpoint.len() ); - assert_eq!(WebRTCStream::get_ice_servers().len(), 2); - config::Config::set_option( - "ice-servers".to_string(), - "".to_string(), + // `parsed` is #[serde(skip)], so deserializing alone would accept any string as the SDP + // body. Unmarshal it, which is also what the peer does, and require the fingerprint the + // session key is derived from to have survived. + let desc: RTCSessionDescription = + serde_json::from_str(&json).expect("a stripped endpoint must deserialize"); + WebRTCStream::get_key_for_sdp(&desc) + .expect("a stripped endpoint must still parse as SDP and keep its fingerprint"); + + // `encode_endpoint`, which `get_local_endpoint` also goes through, holds the opposite + // contract: carry whatever it is handed, candidates included. (What `get_local_endpoint` + // adds on top — the gathering wait — is covered by test_webrtc_loopback_gathered_endpoints.) + let kept = WebRTCStream::encode_endpoint(gathered.sdp_type, &gathered.sdp, false).unwrap(); + assert!( + WebRTCStream::get_remote_offer(&kept) + .unwrap() + .contains("a=candidate:"), + "encode_endpoint dropped the candidates it was handed" ); } @@ -747,10 +1845,12 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc "invalid webrtc endpoint should error" ); - assert!( - WebRTCStream::new("", false, 10000).await.is_ok(), - "local webrtc endpoint should ok" - ); + let stream = WebRTCStream::new("", false, 10000) + .await + .expect("local webrtc endpoint should ok"); + // A raw WebRTCStream has no Drop: close it, or its session stays cached in SESSIONS + // and pollutes cross-test assertions about the cache. + stream.close().await; endpoint = "webrtc://eyJ0eXBlIjoiYW5zd2VyIiwic2RwIjoidj0wXHJcbm89LSA0MTA1NDk3NTY2NDgyMTQzODEwIDYwMzk1NzQw\ MCBJTiBJUDQgMC4wLjAuMFxyXG5zPS1cclxudD0wIDBcclxuYT1maW5nZXJwcmludDpzaGEtMjU2IDYxOjYwOjc0OjQwOjI4OkNFOjBCOjBDOjc1OjRCOj\ @@ -767,4 +1867,646 @@ IHR5cCBzcmZseCByYWRkciAwLjAuMC4wIHJwb3J0IDY0MDA4XHJcbmE9ZW5kLW9mLWNhbmRpZGF0ZXNc "connect to an 'answer' webrtc endpoint should error" ); } + + #[tokio::test] + async fn test_webrtc_wait_connected_timeout() { + let mut stream = WebRTCStream::new("", false, 100).await.unwrap(); + let err = stream.wait_connected(10).await.unwrap_err(); + assert!(err.to_string().contains("timeout")); + stream.close().await; // see test_webrtc_new_stream: no Drop on a raw WebRTCStream + } + + async fn connect_loopback() -> (WebRTCStream, WebRTCStream) { + let mut offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + let offer = offerer.local_endpoint().to_owned(); + let answerer = WebRTCStream::new(&offer, false, 20000).await.unwrap(); + let answer = answerer.local_endpoint().to_owned(); + 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) + } + + // next()'s teardown must not be cancellable: an awaited close that loses the select! race + // against a consumer's timer strands the pc (is_closed already latched) and its SESSIONS + // entry. `close_detached` is non-async, so what needs proving is that the handoff completes. + #[tokio::test] + async fn test_close_detached_completes_without_the_caller() { + let (offerer, answerer) = connect_loopback().await; + let key = format!("offer:{}", offerer.session_key()); + assert!( + SESSIONS.lock().await.contains_key(&key), + "offerer should be cached while live" + ); + + offerer.close_detached(); + drop(offerer); + + for _ in 0..200 { + if !SESSIONS.lock().await.contains_key(&key) { + answerer.close().await; + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("detached close never evicted the peer connection from SESSIONS"); + } + + // Whole messages are split out of the read buffer instead of copied, so the buffer is + // consumed from the front and periodically re-initialized. Alternating whole and fragmented + // messages exercises that against the accumulator path, and running well past one refill + // exercises the refill itself — a boundary the fixed-size buffer never had. + #[tokio::test] + async fn test_webrtc_mixed_message_sizes_survive_scratch_refill() { + let (mut offerer, mut answerer) = connect_loopback().await; + + // Sized either side of MAX_FRAGMENT_PAYLOAD, and enough rounds to consume several + // SCRATCH_REFILL chunks. + let sizes = [1usize, 64, 60_000, 60_001, 130_000, 7]; + for round in 0..40u8 { + for &len in &sizes { + let payload = Bytes::from(vec![round; len]); + offerer.send_bytes(payload.clone()).await.unwrap(); + let got = timeout(Duration::from_secs(10), answerer.next()) + .await + .expect("receiver starved") + .expect("stream ended") + .expect("read failed"); + assert_eq!(got.len(), len, "round {round}, len {len}"); + assert!( + got.iter().all(|&b| b == round), + "round {}, len {}: content or boundary corrupted", + round, + len + ); + } + } + + offerer.close().await; + answerer.close().await; + } + + // Owning the Stream owns the peer connection: dropping it must close the pc and evict the + // session, without the owner having to remember to. Every exit path used to carry that + // obligation, and the controlled side never honoured it. + #[tokio::test] + async fn dropping_the_stream_closes_the_peer_connection() { + let offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + let key = format!("offer:{}", offerer.session_key()); + assert!( + SESSIONS.lock().await.contains_key(&key), + "offerer should be cached while live" + ); + + // No explicit close anywhere: the stream simply goes out of scope, as it does on the + // exits that forget. + drop(crate::Stream::WebRTC(offerer)); + + for _ in 0..200 { + if !SESSIONS.lock().await.contains_key(&key) { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("dropping the stream left the peer connection in SESSIONS"); + } + + // A replayed offer asking for a different ICE policy must not be handed the cached peer + // connection. The lookup and the insert-time duplicate check read the same key, so the test + // fails if either one stops applying `is_reusable_for` — which is how the guard was dead + // code: the lookup rejected the entry, and the insert handed back the very same one. + #[tokio::test] + async fn test_cached_peer_is_not_reused_across_ice_policies() { + let offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + let offer = offerer.local_endpoint().to_owned(); + + let all_ice = WebRTCStream::new(&offer, false, 20000).await.unwrap(); + assert!(!all_ice.relay_only); + + let relay_only = WebRTCStream::new(&offer, true, 20000).await.unwrap(); + assert!( + relay_only.relay_only, + "a Relay-only request was answered with the cached All-policy peer connection" + ); + + relay_only.close().await; + all_ice.close().await; + offerer.close().await; + } + + // The local-candidate channel must close when the pc does. Its only sender lives inside the + // on_ice_candidate handler, which close() does not clear, so without the teardown the + // receiver stays open forever — and the forwarder loop this API asks callers to write holds + // a stream clone while parked on recv(), keeping the pc alive past its own close. + #[tokio::test] + async fn test_local_ice_channel_closes_with_the_peer_connection() { + let offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + let mut ice_rx = offerer.take_local_ice_rx().unwrap(); + + // Exactly the loop every consumer writes, holding a clone of the stream. + let forwarder_stream = offerer.clone(); + let forwarder = tokio::spawn(async move { + while let Some(candidate) = ice_rx.recv().await { + let _ = forwarder_stream.add_remote_ice_candidate(&candidate).await; + } + }); + + offerer.close().await; + drop(offerer); + + tokio::time::timeout(Duration::from_secs(20), forwarder) + .await + .expect("forwarder task outlived the peer connection") + .unwrap(); + } + + // Extra channels must not displace the bound one: the newcomer's on_open would push Open onto + // the watch that gates send and recv, re-arming a session already latched Closed. Only the + // bind-once guard is exercised; the ordered+reliable check cannot decide anything here. + #[tokio::test] + async fn test_webrtc_answerer_binds_only_the_first_data_channel() { + use webrtc::data_channel::data_channel_init::RTCDataChannelInit; + + let mut offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + // Replace the bootstrap channel's siblings: an unordered one and a duplicate. + let unordered = offerer + .pc + .create_data_channel( + "unordered", + Some(RTCDataChannelInit { + ordered: Some(false), + ..Default::default() + }), + ) + .await + .unwrap(); + let duplicate = offerer + .pc + .create_data_channel("duplicate", None) + .await + .unwrap(); + + let offer = offerer.local_endpoint().to_owned(); + let answerer = WebRTCStream::new(&offer, false, 20000).await.unwrap(); + let answer = answerer.local_endpoint().to_owned(); + offerer.set_remote_endpoint(&answer).await.unwrap(); + + 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(); + + // The bootstrap channel still carries data: the rejected siblings did not displace it. + let payload = Bytes::from_static(b"bootstrap still bound"); + offerer.send_bytes(payload.clone()).await.unwrap(); + let got = tokio::time::timeout(Duration::from_secs(10), answerer.next()) + .await + .expect("answerer starved") + .expect("stream ended") + .expect("read failed"); + assert_eq!(&got[..], &payload[..]); + + drop(unordered); + drop(duplicate); + offerer.close().await; + answerer.close().await; + } + + // One-shot callers exchange only the endpoints and never consume `take_local_ice_rx`. + #[tokio::test] + async fn test_webrtc_loopback_gathered_endpoints() { + let connect = async { + let mut offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + let offer = offerer.get_local_endpoint().await.unwrap(); + let mut answerer = WebRTCStream::new(&offer, false, 20000).await.unwrap(); + let answer = answerer.get_local_endpoint().await.unwrap(); + offerer.set_remote_endpoint(&answer).await.unwrap(); + + offerer.wait_connected(20000).await.unwrap(); + answerer.wait_connected(20000).await.unwrap(); + offerer.close().await; + answerer.close().await; + }; + timeout(Duration::from_secs(40), connect) + .await + .expect("gathered-endpoint WebRTC loopback did not complete in time"); + } + + // 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"); + } + + // A client session runs on its own `#[tokio::main(flavor = "current_thread")]` runtime that + // is dropped the moment io_loop returns, so a close spawned or awaited there can be killed + // before (or worse, after) `close()` latches `is_closed`. Pin the production sequence over + // `Stream::WebRTC`: `close_webrtc()` from a thread with no runtime at all, then Drop; the + // peer must still see EOF — the closer thread, not any caller runtime, delivers it. + #[tokio::test(flavor = "multi_thread")] + async fn test_session_end_close_reaches_the_peer() { + let (offerer, mut answerer) = connect_loopback().await; + let stream = crate::Stream::WebRTC(offerer); + tokio::task::spawn_blocking(move || { + stream.close_webrtc(); + drop(stream); + }) + .await + .expect("session thread"); + + match timeout(Duration::from_secs(5), answerer.next()).await { + Ok(None) | Ok(Some(Err(_))) => {} + Ok(Some(Ok(b))) => panic!("expected EOF after session close, got {} bytes", b.len()), + Err(_) => panic!("peer never observed the session-end close"), + } + answerer.close().await; + } + + // The shared count, not any one handoff, decides: with a second claim outstanding on a pc + // no one has adopted, dropping the first must leave it open for that caller, and dropping + // the last must close it — an unanswered offerer never reaches a terminal ICE state itself. + #[tokio::test(flavor = "multi_thread")] + async fn test_last_abandoned_claim_closes_an_unadopted_pc() { + let builder = WEBRTC_RT + .as_ref() + .expect("WebRTC I/O runtime") + .spawn(WebRTCStream::new_inner(String::new(), false, 20000)) + .await + .expect("setup task") + .expect("offerer builds"); + let stream = builder + .stream + .as_ref() + .expect("a fresh handoff carries its stream") + .clone(); + let key = format!("offer:{}", stream.session_key()); + let hit = NewStreamHandoff::claim(stream); + + drop(builder); + tokio::time::sleep(Duration::from_millis(300)).await; + assert!( + SESSIONS.lock().await.contains_key(&key), + "a claim was still outstanding, so the pc had to stay open" + ); + + drop(hit); + for _ in 0..200 { + if !SESSIONS.lock().await.contains_key(&key) { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("the last abandoned claim never closed the unadopted pc"); + } + + // `new_inner` caches the pc BEFORE its own `new()` returns, so a concurrent caller can hit + // that cache and adopt the stream while the builder's handoff is still outstanding. Closing + // on the builder's Drop then tears down a connection another caller is already using, so + // abandonment has to be a property of the pc, not of one handoff. + #[tokio::test(flavor = "multi_thread")] + async fn test_abandoned_builder_handoff_spares_an_adopted_pc() { + // Exactly what `new()` does, minus the adoption: the setup task runs on WEBRTC_RT and + // hands back a fresh, still-unadopted handoff. + let builder = WEBRTC_RT + .as_ref() + .expect("WebRTC I/O runtime") + .spawn(WebRTCStream::new_inner(String::new(), false, 20000)) + .await + .expect("setup task") + .expect("offerer builds"); + let stream = builder + .stream + .as_ref() + .expect("a fresh handoff carries its stream") + .clone(); + let key = format!("offer:{}", stream.session_key()); + assert!( + SESSIONS.lock().await.contains_key(&key), + "new_inner must cache the pc it built" + ); + + // The concurrent `new()`: a cache hit that takes the stream and goes on using it. + let adopted = NewStreamHandoff::claim(stream) + .into_inner() + .expect("a cache hit hands the stream back"); + // ...and now the builder's `new()` is cancelled, so the runtime drops its handoff. + drop(builder); + + tokio::time::sleep(Duration::from_millis(500)).await; + assert!( + SESSIONS.lock().await.contains_key(&key), + "abandoning the builder's handoff closed a pc another caller had already adopted" + ); + adopted.close_detached(); + } + + // End-to-end: cancel `new()` itself after its first poll (the spawn is already in flight) + // and every session it transiently created must be closed and evicted again. + // + // The cancelled attempt's own key is unknowable here — its DTLS fingerprint is generated + // inside the task that was abandoned — so what is watched for is a key that OUTLASTS the + // window, not an instant with no new keys at all. Concurrent tests each close what they + // create, so their keys come and go and drop out of the running intersection; a leaked pc + // never does. Waiting for an empty difference instead made this test depend on the suite + // having an idle moment, which `--test-threads=2` never gives it: one lane is this test for + // the whole wait while the other keeps starting sessions. + #[tokio::test(flavor = "multi_thread")] + async fn test_cancelled_new_does_not_leak_the_pc() { + use std::collections::HashSet; + let before: HashSet = SESSIONS.lock().await.keys().cloned().collect(); + // Zero timeout: polls the future exactly once (spawning new_inner), then cancels it — + // usually. The setup task runs on its own runtime and can finish inside that single poll, + // and then `new()` hands back a live stream instead. `WebRTCStream` has no `Drop`, so + // discarding that one is itself a leak, and this test would go on to report it as the + // cancelled attempt's. Close what comes back and try for a real cancellation. Fewer test + // threads make the setup task likelier to win, which is why this surfaced under + // `--test-threads=2` and not at the default. + let mut cancelled = false; + for _ in 0..20 { + match timeout(Duration::ZERO, WebRTCStream::new("", false, 20000)).await { + Err(_) => { + cancelled = true; + break; + } + Ok(Ok(stream)) => stream.close().await, + Ok(Err(_)) => {} + } + } + assert!(cancelled, "new() never lost the race with its own cancellation"); + // Let the detached setup task finish (and insert its session) before sampling, or the + // first sample is taken before the leak has formed and every later one intersects to + // nothing. + tokio::time::sleep(Duration::from_millis(500)).await; + const ATTEMPTS: usize = 600; // 30s + let mut persisted: Option> = None; + for attempt in 1..=ATTEMPTS { + let now: HashSet = SESSIONS.lock().await.keys().cloned().collect(); + let new_keys: HashSet = now.difference(&before).cloned().collect(); + persisted = Some(match persisted { + None => new_keys, + Some(prev) => prev.intersection(&new_keys).cloned().collect(), + }); + let persisted_keys = persisted.as_ref().map_or(0, HashSet::len); + if persisted_keys == 0 { + return; + } + // Positional, not an inline `{persisted:?}`: on edition 2018 a lone-literal `panic!` + // does not go through format_args and would print the placeholder verbatim. + assert!( + attempt < ATTEMPTS, + "cancelled new() left a pc cached in SESSIONS (persisted: {:?})", + persisted + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + + // The production shape of a controller session end: the offerer was created on a session's + // own current-thread runtime, the session queues its detached close and the runtime is + // destroyed immediately after (io_loop returning drops it). The pc's sockets and its + // ICE/DTLS/SCTP pump tasks must not die with that runtime, or the queued close completes + // without ever putting close_notify on the wire and the peer waits out ICE decay. Gathered + // (non-trickle) endpoints keep the cross-runtime signaling to two string handoffs. + #[tokio::test(flavor = "multi_thread")] + async fn test_close_survives_creator_runtime_destruction() { + // tokio oneshots for every handoff: their sends are synchronous, and the session side + // awaits them INSIDE block_on — a current_thread runtime only drives its tasks while + // being block_on-driven, and the offerer's SCTP/ICE pumps must stay live until the + // answerer has confirmed the channel is open. + let (offer_tx, offer_rx) = tokio::sync::oneshot::channel::(); + let (answer_tx, answer_rx) = tokio::sync::oneshot::channel::(); + let (go_tx, go_rx) = tokio::sync::oneshot::channel::<()>(); + let session = std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("session runtime"); + let offerer = rt.block_on(async { + let mut offerer = WebRTCStream::new("", false, 20000).await.unwrap(); + offer_tx + .send(offerer.get_local_endpoint().await.unwrap()) + .unwrap(); + let answer = answer_rx.await.unwrap(); + offerer.set_remote_endpoint(&answer).await.unwrap(); + offerer.wait_connected(20000).await.unwrap(); + // Keep driving the runtime until the answerer confirms open, so the EOF + // assertion — not connection setup — is what discriminates. + go_rx.await.unwrap(); + offerer + }); + // The session ends: queue the close, then destroy the runtime the pc was created + // on — exactly what io_loop returning does. + drop(crate::Stream::WebRTC(offerer)); + drop(rt); + }); + + let offer = offer_rx.await.expect("offer from session thread"); + let mut answerer = WebRTCStream::new(&offer, false, 20000).await.unwrap(); + answer_tx + .send(answerer.get_local_endpoint().await.unwrap()) + .expect("answer to session thread"); + answerer.wait_connected(20000).await.unwrap(); + go_tx.send(()).expect("release session thread"); + session.join().expect("session thread"); + + match timeout(Duration::from_secs(5), answerer.next()).await { + Ok(None) | Ok(Some(Err(_))) => {} + Ok(Some(Ok(b))) => panic!("expected EOF after session close, got {} bytes", b.len()), + Err(_) => panic!("peer never observed the close after the creator runtime died"), + } + answerer.close().await; + } + + // The peer-initiated end: `next()` hits EOF and fires its own `close_detached` before the + // session's final close (here via `Stream`'s Drop) is issued. Both land on the serialized + // closer thread, so the first runs to completion and the second is a benign no-op — the pc + // must end up closed and evicted, never stranded by one closer cancelling the other. + #[tokio::test(flavor = "multi_thread")] + async fn test_eof_close_then_drop_still_evicts() { + let (mut offerer, answerer) = connect_loopback().await; + let key = format!("offer:{}", offerer.session_key()); + answerer.close().await; + match timeout(Duration::from_secs(5), offerer.next()).await { + Ok(None) | Ok(Some(Err(_))) => {} // EOF path fired close_detached internally + Ok(Some(Ok(b))) => panic!("expected EOF after peer close, got {} bytes", b.len()), + Err(_) => panic!("offerer.next() hung after peer close"), + } + drop(crate::Stream::WebRTC(offerer)); // second close via Drop + + for _ in 0..200 { + if !SESSIONS.lock().await.contains_key(&key) { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("pc still cached after EOF close + Drop close"); + } + + // 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 { + let (offerer, mut answerer) = connect_loopback().await; + let mut sender_a = offerer.clone(); + let mut sender_b = offerer.clone(); + let expected_a = vec![0xAA; 200_000]; + let expected_b = vec![0xBB; 200_000]; + let payload_a = expected_a.clone(); + let payload_b = expected_b.clone(); + let barrier = Arc::new(Barrier::new(3)); + + let barrier_a = barrier.clone(); + let send_a = tokio::spawn(async move { + barrier_a.wait().await; + sender_a.send_raw(payload_a).await + }); + let barrier_b = barrier.clone(); + let send_b = tokio::spawn(async move { + barrier_b.wait().await; + sender_b.send_raw(payload_b).await + }); + + barrier.wait().await; + let receive = async { + let first = answerer.next().await.unwrap().unwrap(); + let second = answerer.next().await.unwrap().unwrap(); + (first, second) + }; + let (send_a, send_b, (first, second)) = tokio::join!(send_a, send_b, receive); + send_a.unwrap().unwrap(); + send_b.unwrap().unwrap(); + + let boundaries_preserved = (first.as_ref() == expected_a.as_slice() + && second.as_ref() == expected_b.as_slice()) + || (first.as_ref() == expected_b.as_slice() + && second.as_ref() == expected_a.as_slice()); + assert!(boundaries_preserved, "concurrent messages were interleaved"); + + offerer.close().await; + answerer.close().await; + }; + timeout(Duration::from_secs(40), connect) + .await + .expect("concurrent WebRTC sends did not complete in time"); + } }