diff --git a/Cargo.lock b/Cargo.lock index 1746adc00..26e6456af 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1161,30 +1161,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" -[[package]] -name = "chacha20" -version = "0.9.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818" -dependencies = [ - "cfg-if 1.0.0", - "cipher", - "cpufeatures", -] - -[[package]] -name = "chacha20poly1305" -version = "0.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35" -dependencies = [ - "aead", - "chacha20", - "cipher", - "poly1305", - "zeroize", -] - [[package]] name = "chrono" version = "0.4.41" @@ -1234,7 +1210,6 @@ checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ "crypto-common", "inout", - "zeroize", ] [[package]] @@ -2442,42 +2417,6 @@ dependencies = [ "linux-raw-sys 0.6.5", ] -[[package]] -name = "dtls" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f531dd7c181beaf3cebab3716afa4d0d41ab888be85232583f56bbaf07ca208a" -dependencies = [ - "aes", - "aes-gcm", - "async-trait", - "bincode", - "byteorder", - "cbc", - "ccm", - "chacha20poly1305", - "der-parser", - "hmac", - "log", - "p256", - "p384", - "portable-atomic", - "rand 0.9.2", - "rand_core 0.6.4", - "rcgen", - "ring", - "rustls", - "sec1", - "serde 1.0.228", - "sha1", - "sha2", - "thiserror 1.0.61", - "tokio", - "webrtc-util", - "x25519-dalek", - "x509-parser", -] - [[package]] name = "dtoa" version = "0.4.8" @@ -4177,16 +4116,15 @@ dependencies = [ [[package]] name = "interceptor" -version = "0.15.0" +version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ea51375727680dc15f06e8ad90fa31df75d79dd030100e8ad60eef1c27fe2c98" +checksum = "1ac0781c825d602095113772e389ef0607afcb869ae0e68a590d8e0799cdcef8" dependencies = [ "async-trait", "bytes", - "futures", "log", "portable-atomic", - "rand 0.9.2", + "rand 0.8.5", "rtcp", "rtp", "thiserror 1.0.61", @@ -6277,17 +6215,6 @@ dependencies = [ "windows-sys 0.52.0", ] -[[package]] -name = "poly1305" -version = "0.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8159bd90725d2df49889a078b54f4f79e87f1f8a8444194cdca81d38f5393abf" -dependencies = [ - "cpufeatures", - "opaque-debug", - "universal-hash", -] - [[package]] name = "polyval" version = "0.6.2" @@ -7094,9 +7021,9 @@ dependencies = [ [[package]] name = "rtcp" -version = "0.14.0" +version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "81d30d1c4091644431c22acf9f8be6191b56805e0e977f15ca7104b4a6d6eaec" +checksum = "e9689528bf3a9eb311fd938d05516dd546412f9ce4fffc8acfc1db27cc3dbf72" dependencies = [ "bytes", "thiserror 1.0.61", @@ -7105,14 +7032,14 @@ dependencies = [ [[package]] name = "rtp" -version = "0.14.0" +version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f126f38ea84c02480e32e547c1459a939052f74fb92117ac3eef23fdac6b023" +checksum = "c54733451a67d76caf9caa07a7a2cec6871ea9dda92a7847f98063d459200f4b" dependencies = [ "bytes", "memchr", "portable-atomic", - "rand 0.9.2", + "rand 0.8.5", "serde 1.0.228", "thiserror 1.0.61", "webrtc-util", @@ -7547,11 +7474,11 @@ dependencies = [ [[package]] name = "sdp" -version = "0.10.0" +version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32c374dceda16965d541c8800ce9cc4e1c14acfd661ddf7952feeedc3411e5c6" +checksum = "4cd277015eada44a0bb810a4b84d3bf6e810573fa62fb442f457edf6a1087a69" dependencies = [ - "rand 0.9.2", + "rand 0.8.5", "substring", "thiserror 1.0.61", "url", @@ -8129,15 +8056,15 @@ dependencies = [ [[package]] name = "stun" -version = "0.9.0" +version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a512c5d501e3e3b5a4bb3e8e31462d56d54a66b95a28b8596e14422bf21c32b" +checksum = "7dbc2bab375524093c143dc362a03fb6a1fb79e938391cdb21665688f88a088a" dependencies = [ "base64 0.22.1", "crc", "lazy_static", "md-5", - "rand 0.9.2", + "rand 0.8.5", "ring", "subtle", "thiserror 1.0.61", @@ -9048,9 +8975,9 @@ dependencies = [ [[package]] name = "turn" -version = "0.11.0" +version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ed995882f66ab94238de77c62e5e778389698ab700afa4696f4754da8f457cb" +checksum = "3f5aea1116456e1da71c45586b87c72e3b43164fbf435eb93ff6aa475416a9a4" dependencies = [ "async-trait", "base64 0.22.1", @@ -9058,7 +8985,7 @@ dependencies = [ "log", "md-5", "portable-atomic", - "rand 0.9.2", + "rand 0.8.5", "ring", "stun", "thiserror 1.0.61", @@ -9184,12 +9111,6 @@ dependencies = [ "unic-common", ] -[[package]] -name = "unicase" -version = "2.8.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75b844d17643ee918803943289730bec8aac480150456169e647ed0b576ba539" - [[package]] name = "unicode-bidi" version = "0.3.15" @@ -9724,25 +9645,26 @@ dependencies = [ [[package]] name = "webrtc" -version = "0.14.0" +version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08fd686c0920ac08f3a57eacc48e31f0e4ca1ffefba4478784606f78c14e83ad" +checksum = "24bab7195998d605c862772f90a452ba655b90a2f463c850ac032038890e367a" dependencies = [ "arc-swap", "async-trait", "bytes", - "dtls", + "cfg-if 1.0.0", "hex", "interceptor", "lazy_static", "log", "portable-atomic", - "rand 0.9.2", + "rand 0.8.5", "rcgen", "regex", "ring", "rtcp", "rtp", + "rustls", "sdp", "serde 1.0.228", "serde_json 1.0.118", @@ -9750,12 +9672,13 @@ dependencies = [ "smol_str", "stun", "thiserror 1.0.61", + "time 0.3.36", "tokio", "turn", - "unicase", "url", "waitgroup", "webrtc-data", + "webrtc-dtls", "webrtc-ice", "webrtc-mdns", "webrtc-media", @@ -9766,9 +9689,9 @@ dependencies = [ [[package]] name = "webrtc-data" -version = "0.12.0" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "062a5438d63bb0756a221693d76cc0dd6119affee1dfdfe57abe3a2a8c8b3eea" +checksum = "4e97b932854da633a767eff0cc805425a2222fc6481e96f463e57b015d949d1d" dependencies = [ "bytes", "log", @@ -9780,17 +9703,54 @@ dependencies = [ ] [[package]] -name = "webrtc-ice" -version = "0.14.0" +name = "webrtc-dtls" +version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "69cb13fd1a373e68addc4bba0c8ca058627518e54342583d024bdcbb8ae5d97d" +checksum = "5ccbe4d9049390ab52695c3646c1395c877e16c15fb05d3bda8eee0c7351711c" +dependencies = [ + "aes", + "aes-gcm", + "async-trait", + "bincode", + "byteorder", + "cbc", + "ccm", + "der-parser", + "hkdf", + "hmac", + "log", + "p256", + "p384", + "portable-atomic", + "rand 0.8.5", + "rand_core 0.6.4", + "rcgen", + "ring", + "rustls", + "sec1", + "serde 1.0.228", + "sha1", + "sha2", + "subtle", + "thiserror 1.0.61", + "tokio", + "webrtc-util", + "x25519-dalek", + "x509-parser", +] + +[[package]] +name = "webrtc-ice" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb51bde0d790f109a15bfe4d04f1b56fb51d567da231643cb3f21bb74d678997" dependencies = [ "arc-swap", "async-trait", "crc", "log", "portable-atomic", - "rand 0.9.2", + "rand 0.8.5", "serde 1.0.228", "serde_json 1.0.118", "stun", @@ -9806,9 +9766,9 @@ dependencies = [ [[package]] name = "webrtc-mdns" -version = "0.10.0" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a17279a067e75df72ce923fdeb7f04cd808f6f5aa4910dc6bcb4fbe66b396ace" +checksum = "979cc85259c53b7b620803509d10d35e2546fa505d228850cbe3f08765ea6ea8" dependencies = [ "log", "socket2 0.5.10", @@ -9819,22 +9779,22 @@ dependencies = [ [[package]] name = "webrtc-media" -version = "0.11.0" +version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94a84c910fec0848fd5a0d8a5651e0ddbdedaf25a7d3ae3f0b15f71ac73a1773" +checksum = "80041211deccda758a3e19aa93d6b10bc1d37c9183b519054b40a83691d13810" dependencies = [ "byteorder", "bytes", - "rand 0.9.2", + "rand 0.8.5", "rtp", "thiserror 1.0.61", ] [[package]] name = "webrtc-sctp" -version = "0.13.0" +version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f985465467d8910c1f8ac4382cd64f83b1f6a1a75021a82b221546f6fb3b856f" +checksum = "07439c134425d51d2f10907aaf2f815fdfb587dce19fe94a4ae8b5faf2aae5ae" dependencies = [ "arc-swap", "async-trait", @@ -9842,7 +9802,7 @@ dependencies = [ "crc", "log", "portable-atomic", - "rand 0.9.2", + "rand 0.8.5", "thiserror 1.0.61", "tokio", "webrtc-util", @@ -9850,9 +9810,9 @@ dependencies = [ [[package]] name = "webrtc-srtp" -version = "0.16.0" +version = "0.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "66d8cdc33413f1d0192670a80ce93d17cb78d57fe3a2414be30d6f6dff121123" +checksum = "01e773f79b09b057ffbda6b03fe7b43403b012a240cf8d05d630674c3723b5bb" dependencies = [ "aead", "aes", @@ -9873,19 +9833,20 @@ dependencies = [ [[package]] name = "webrtc-util" -version = "0.12.0" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d1c0c7e0c8f280f2bbfae442701465777ac07adaf46ce0c5863cd58e13fe472a" +checksum = "64bfb10dbe6d762f80169ae07cf252bafa1f764b9594d140008a0231c0cdce58" dependencies = [ "async-trait", "bitflags 1.3.2", "bytes", "ipnet", "lazy_static", + "libc", "log", "nix 0.26.4", "portable-atomic", - "rand 0.9.2", + "rand 0.8.5", "thiserror 1.0.61", "tokio", "winapi 0.3.9", diff --git a/libs/hbb_common b/libs/hbb_common index d4e727306..f98f3e873 160000 --- a/libs/hbb_common +++ b/libs/hbb_common @@ -1 +1 @@ -Subproject commit d4e727306f28aded5f2e461b1aaaad741e9e35fa +Subproject commit f98f3e8732198aacbc3be04662a8f5d89600d5cf diff --git a/src/client.rs b/src/client.rs index 6be5434b9..bb1870fde 100644 --- a/src/client.rs +++ b/src/client.rs @@ -30,7 +30,7 @@ use uuid::Uuid; use crate::{ check_port, common::input::{MOUSE_BUTTON_LEFT, MOUSE_BUTTON_RIGHT, MOUSE_TYPE_DOWN, MOUSE_TYPE_UP}, - create_symmetric_key_msg, decode_id_pk, get_rs_pk, is_keyboard_mode_supported, + create_symmetric_key_msg, decode_id_pk, decode_id_pk_dtls, get_rs_pk, is_keyboard_mode_supported, kcp_stream::KcpStream, secure_tcp, ui_interface::{get_builtin_option, resolve_avatar_url, use_texture_render}, @@ -51,7 +51,7 @@ use hbb_common::{ CONNECT_TIMEOUT, READ_TIMEOUT, RELAY_PORT, RENDEZVOUS_PORT, RENDEZVOUS_SERVERS, }, fs::JobType, - futures::future::{select_ok, FutureExt}, + futures::future::{select_ok, BoxFuture, FutureExt}, get_version_number, log, message_proto::{option_message::BoolOption, *}, protobuf::{Message as _, MessageField}, @@ -175,6 +175,118 @@ pub fn get_key_state(key: enigo::Key) -> bool { ENIGO.lock().unwrap().get_key_state(key) } +/// RAII guard for a WebRTC offerer that has not yet been adopted into a connection. +/// +/// The offerer's `RTCPeerConnection` is created eagerly in `Client::start` and inserted into the +/// global `SESSIONS` map; if it never receives a remote answer it stays in ICE state `New` +/// forever, so its state-change handler never fires and it never self-removes. This guard closes +/// the pc on drop, covering the paths that would otherwise leak it: early `?`/`bail!` returns in +/// `_start_inner`, and `select_ok` cancelling the racing attempt that holds the offerer. Call +/// [`OffererGuard::into_inner`] to disarm when the stream is adopted into a live connection. +struct OffererGuard(Option); + +impl OffererGuard { + fn new(stream: WebRTCStream) -> Self { + Self(Some(stream)) + } + + fn stream(&self) -> Option<&WebRTCStream> { + self.0.as_ref() + } + + fn into_inner(mut self) -> Option { + self.0.take() + } +} + +impl Drop for OffererGuard { + fn drop(&mut self) { + if let Some(stream) = self.0.take() { + Client::spawn_close_webrtc(stream); + } + } +} + +/// Race the WebRTC connect attempt against the other transport futures, preferring P2P. +/// +/// A plain `select_ok` would almost always pick the relay: its TCP connect completes in one +/// round trip while WebRTC needs candidate trickle + ICE checks + DTLS + SCTP. Instead: +/// - a WebRTC success is committed immediately; +/// - an `is_p2p` result from `others` (e.g. IPv6 direct) is committed immediately; +/// - a relayed result inside the preference window is *held*, giving WebRTC until the window +/// expires to connect and take priority, while unfinished `others` keep racing so a slower +/// direct transport can still win; when the window ends (or WebRTC fails) the held connection +/// is committed; +/// - a failure on one side commits the survivor as soon as it succeeds; two failures compose +/// into one error. +/// +/// `others` must be non-empty (`select_ok` requires it). +async fn race_transports_prefer_webrtc<'a, T: 'a>( + webrtc_fut: BoxFuture<'a, ResultType>, + others: Vec>>, + window_ms: u64, + is_p2p: impl Fn(&T) -> bool, +) -> ResultType { + let mut webrtc_fut = Some(webrtc_fut); + let mut others_fut = Some(select_ok(others)); + let mut held: Option = None; + let mut webrtc_err: Option = None; + let mut others_err: Option = None; + let window = tokio::time::sleep(Duration::from_millis(window_ms)); + tokio::pin!(window); + let mut window_over = false; + loop { + tokio::select! { + res = async { webrtc_fut.as_mut().unwrap().await }, if webrtc_fut.is_some() => { + webrtc_fut = None; + match res { + // WebRTC connected: preferred outright; a held relay conn just drops. + Ok(conn) => return Ok(conn), + Err(e) => { + if let Some(conn) = held.take() { + return Ok(conn); + } + match others_err.take() { + Some(oe) => bail!("WebRTC failed: {}; fallback failed: {}", e, oe), + None if others_fut.is_none() => bail!("WebRTC failed: {}", e), + None => webrtc_err = Some(e), + } + } + } + } + res = async { others_fut.as_mut().unwrap().await }, if others_fut.is_some() => { + others_fut = None; + match res { + Ok((conn, unfinished)) => { + if is_p2p(&conn) || window_over || webrtc_fut.is_none() { + return Ok(conn); + } + // Hold the first relay, but keep polling unfinished alternatives: an IPv6 + // direct attempt may complete inside the same WebRTC preference window. + if held.is_none() { + held = Some(conn); + } + if !unfinished.is_empty() { + others_fut = Some(select_ok(unfinished)); + } + } + Err(e) => match webrtc_err.take() { + Some(we) => bail!("WebRTC failed: {}; fallback failed: {}", we, e), + None if webrtc_fut.is_none() => return Err(e), + None => others_err = Some(e), + }, + } + } + _ = &mut window, if !window_over => { + window_over = true; + if let Some(conn) = held.take() { + return Ok(conn); + } + } + } + } +} + impl Client { const CLIENT_CLIPBOARD_NAME: &'static str = "client-clipboard"; @@ -326,14 +438,17 @@ impl Client { } else { None }; - let webrtc_offerer = + let webrtc_offerer = if Self::should_create_webrtc_offerer(&interface) { match WebRTCStream::new("", interface.is_force_relay(), CONNECT_TIMEOUT).await { Ok(stream) => Some(stream), Err(err) => { log::warn!("webrtc offerer setup failed: {}", err); None } - }; + } + } else { + None + }; let fut = Self::_start_inner( peer.to_owned(), key.to_owned(), @@ -378,6 +493,70 @@ impl Client { !session_key.is_empty() && ice.session_key == session_key && !ice.candidate.is_empty() } + /// Tear down an abandoned WebRTC offerer off the connection hot path. + /// + /// The pc must be closed explicitly (a dropped handle leaves the `SESSIONS` clone alive), + /// but `close()` awaits internal ICE/DTLS shutdown and the result is pure cleanup with no + /// downstream dependency, so it must not block session establishment. + fn spawn_close_webrtc(webrtc: WebRTCStream) { + // Use the current runtime handle explicitly: this is also called from OffererGuard::drop, + // which can run during runtime teardown where a bare tokio::spawn would panic. If no + // runtime is available the pc cannot be closed here (it is released at process exit). + // Handle::spawn can also panic when the runtime is shutting down; catch that so Drop + // never panics (prefer a brief leak until process exit over aborting the process). + match tokio::runtime::Handle::try_current() { + Ok(handle) => { + let spawn_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + handle.spawn(async move { + webrtc.close().await; + }); + })); + if spawn_result.is_err() { + log::warn!("failed to spawn WebRTC close (runtime shutting down)"); + } + } + Err(_) => { + log::warn!("no tokio runtime available to close WebRTC peer connection"); + } + } + } + + /// Whether to build a WebRTC offerer for this connection. + /// + /// Skips it when UDP punching is disabled or a SOCKS proxy/websocket transport is configured: + /// WebRTC's ICE binds its own UDP sockets and speaks STUN directly, which would bypass either + /// policy (and can leak the real IP through a proxy). Under force_relay the pc uses Relay-only + /// ICE, which gathers nothing and can never connect unless a TURN server is configured, so + /// skip building a guaranteed-dead pc + STUN/TURN gathering + answerer signaling in that case + /// too. + /// When force_relay *and* TURN are configured, WebRTC via TURN is a valid "relayed" path: + /// `connect` keeps a WebRTC win instead of replacing it with the RustDesk relay, and the + /// RelayResponse path prefers it within the WebRTC preference window. + fn should_create_webrtc_offerer(interface: &impl Interface) -> bool { + if !crate::get_udp_punch_enabled() || use_ws() || Config::is_proxy() { + return false; + } + if interface.is_force_relay() && !WebRTCStream::has_turn_server() { + return false; + } + true + } + + /// Max ICE candidates buffered during the punch window before the answer is applied. + /// Bounds memory if a misbehaving rendezvous floods candidates. + const MAX_PENDING_WEBRTC_ICE: usize = 64; + + /// Prefer-P2P window: how long a WebRTC attempt outranks an already-established relay + /// result, and the floor for a punch-path WebRTC attempt whose race timeout is tuned for a + /// raw TCP SYN. Long enough for candidate trickle + ICE checks + DTLS on high-latency + /// links; short enough that UDP-blocked networks settle on relay without a noticeable wait. + const WEBRTC_PREFER_WINDOW_MS: u64 = 2500; + + /// Delay before re-sending an ICE candidate over the rendezvous route once. The hop to the + /// peer can be UDP (the controlled side's mediator channel), so a candidate can be lost in + /// flight; the remote ICE agent dedups repeats, so the second copy is free. + const WEBRTC_ICE_RESEND_DELAY: Duration = Duration::from_millis(400); + fn spawn_webrtc_ice_bridge( mut socket: Stream, mut local_ice_rx: Option>, @@ -387,6 +566,7 @@ impl Client { ) -> oneshot::Sender<()> { let (stop_tx, mut stop_rx) = oneshot::channel::<()>(); tokio::spawn(async move { + let mut pending_resend: Vec<(Instant, RendezvousMessage)> = Vec::new(); loop { match stop_rx.try_recv() { Ok(_) | Err(tokio::sync::oneshot::error::TryRecvError::Closed) => break, @@ -404,10 +584,20 @@ impl Client { candidate, ..Default::default() }); - if let Err(err) = socket.send(&msg).await { - log::warn!("failed to send WebRTC ICE candidate: {}", err); - return; + // Bound the send so a stalled rendezvous socket cannot block the + // bridge past its stop signal. + match timeout(3000, socket.send(&msg)).await { + Ok(Ok(())) => {} + Ok(Err(err)) => { + log::warn!("failed to send WebRTC ICE candidate: {}", err); + return; + } + Err(_) => { + log::warn!("WebRTC ICE candidate send timed out"); + return; + } } + pending_resend.push((Instant::now(), msg)); } Err(TryRecvError::Empty) => break, Err(TryRecvError::Disconnected) => { @@ -418,16 +608,53 @@ impl Client { } } - if let Some(msg_in) = - crate::get_next_nonkeyexchange_msg(&mut socket, Some(100)).await - { - if let Some(rendezvous_message::Union::IceCandidate(ice)) = msg_in.union { - if Self::is_expected_webrtc_ice_candidate(&ice, &session_key) { - if let Err(err) = webrtc.add_remote_ice_candidate(&ice.candidate).await - { - log::warn!("failed to add WebRTC ICE candidate: {}", err); + // Re-send each candidate once after a short delay: the rendezvous hop to the peer + // can be UDP, so the first copy may be lost; the remote ICE agent dedups repeats. + let mut i = 0; + while i < pending_resend.len() { + if pending_resend[i].0.elapsed() >= Self::WEBRTC_ICE_RESEND_DELAY { + let (_, msg) = pending_resend.swap_remove(i); + match timeout(3000, socket.send(&msg)).await { + Ok(Ok(())) => {} + Ok(Err(err)) => { + log::warn!("failed to re-send WebRTC ICE candidate: {}", err); + return; + } + Err(_) => { + log::warn!("WebRTC ICE candidate re-send timed out"); + return; } } + } else { + i += 1; + } + } + + // Read one incoming message, blocking up to the poll window. Distinguish a real + // timeout (keep looping to drain outbound candidates) from stream EOF/error: the + // rendezvous server closes the punch connection after the response, and an + // unrecognized-message server closes it on the first candidate we send. Without + // this, the collapsed None-on-EOF made the loop hot-spin a core until stop. + match timeout(100, socket.next()).await { + Err(_) => {} + Ok(None) => break, + Ok(Some(Ok(bytes))) => { + if let Ok(msg_in) = RendezvousMessage::parse_from_bytes(&bytes) { + if let Some(rendezvous_message::Union::IceCandidate(ice)) = msg_in.union + { + if Self::is_expected_webrtc_ice_candidate(&ice, &session_key) { + if let Err(err) = + webrtc.add_remote_ice_candidate(&ice.candidate).await + { + log::warn!("failed to add WebRTC ICE candidate: {}", err); + } + } + } + } + } + Ok(Some(Err(err))) => { + log::debug!("WebRTC ICE bridge socket read ended: {}", err); + break; } } } @@ -444,7 +671,7 @@ impl Client { mut udp: (Option>, Option>>), stop_udp_tx: Option>, mut ipv6: Option<(Arc, bytes::Bytes)>, - mut webrtc_offerer: Option, + webrtc_offerer: Option, mut rendezvous_server: String, servers: Vec, contained: bool, @@ -459,6 +686,10 @@ impl Client { (i32, String), bool, )> { + // Wrap the offerer so any early return below (?/bail) or cancellation of this future by + // the outer select_ok closes its pc instead of leaking it in SESSIONS. Disarmed via + // into_inner() once the stream is adopted into a connection attempt. + let mut webrtc_offerer = webrtc_offerer.map(OffererGuard::new); let mut start = Instant::now(); let mut socket = connect_tcp(&*rendezvous_server, CONNECT_TIMEOUT).await; debug_assert!(!servers.contains(&rendezvous_server)); @@ -521,8 +752,10 @@ impl Client { .take() .map(|(socket, addr)| (Some(socket), Some(addr))) .unwrap_or((None, None)); - let webrtc_sdp_offer = if let Some(webrtc) = webrtc_offerer.as_ref() { - match webrtc.get_local_endpoint().await { + let webrtc_sdp_offer = if let Some(stream) = + webrtc_offerer.as_ref().and_then(|g| g.stream()) + { + match stream.get_local_endpoint().await { Ok(endpoint) => endpoint, Err(err) => { log::warn!("failed to read local WebRTC offer: {}", err); @@ -550,7 +783,8 @@ impl Client { }); let webrtc_session_key = webrtc_offerer .as_ref() - .map(|webrtc| webrtc.session_key().to_owned()) + .and_then(|guard| guard.stream()) + .map(|stream| stream.session_key().to_owned()) .unwrap_or_default(); let mut webrtc_sdp_answer = String::new(); let mut pending_webrtc_ice = Vec::::new(); @@ -649,22 +883,34 @@ impl Client { let mut webrtc_bridge_stop = None; let mut webrtc_for_connect = None; if !rr.webrtc_sdp_answer.is_empty() { - if let Some(webrtc) = webrtc_offerer.take() { - if let Err(err) = - webrtc.set_remote_endpoint(&rr.webrtc_sdp_answer).await - { - log::warn!("failed to set WebRTC relay answer: {}", err); - } else { - for candidate in pending_webrtc_ice.drain(..) { - if let Err(err) = - webrtc.add_remote_ice_candidate(&candidate).await - { - log::warn!( - "failed to add buffered WebRTC ICE candidate: {}", - err - ); + if let Some(guard) = webrtc_offerer.take() { + // Run the awaited setup on the guard's borrowed stream so a + // cancellation during these awaits still closes the pc via the + // guard's drop; take ownership only at the synchronous handoff. + let setup_ok = if let Some(webrtc) = guard.stream() { + if let Err(err) = + webrtc.set_remote_endpoint(&rr.webrtc_sdp_answer).await + { + log::warn!("failed to set WebRTC relay answer: {}", err); + false + } else { + for candidate in pending_webrtc_ice.drain(..) { + if let Err(err) = + webrtc.add_remote_ice_candidate(&candidate).await + { + log::warn!( + "failed to add buffered WebRTC ICE candidate: {}", + err + ); + } } + true } + } else { + false + }; + if let Some(webrtc) = setup_ok.then(|| guard.into_inner()).flatten() + { let session_key = webrtc.session_key().to_owned(); let local_ice_rx = webrtc.take_local_ice_rx(); webrtc_bridge_stop = Some(Self::spawn_webrtc_ice_bridge( @@ -676,8 +922,16 @@ impl Client { )); webrtc_for_connect = Some(webrtc); } + // If setup failed, `guard` drops here and closes the pc. } } + // An offerer not adopted into the relay race (empty answer) is closed by + // the guard's drop here. + drop(webrtc_offerer.take()); + // Keep relay_server for a WebRTC secure-failure fallback: request_relay + // coordinates a FRESH uuid via the rendezvous server, so it works even if + // the raced create_relay already consumed the original uuid pairing. + let relay_server_rr = rr.relay_server.clone(); let fut = Self::create_relay( &peer, rr.uuid, @@ -693,38 +947,122 @@ impl Client { } .boxed(), ); - if let Some(mut webrtc) = webrtc_for_connect { - connect_futures.push( - async move { - webrtc.wait_connected(CONNECT_TIMEOUT).await?; - Ok((Stream::WebRTC(webrtc), None, "WebRTC")) - } - .boxed(), - ); - } - // Run all connection attempts concurrently, return the first successful one - let (conn, kcp, typ) = match select_ok(connect_futures).await { - Ok(conn) => (Ok(conn.0 .0), conn.0 .1, conn.0 .2), - - Err(e) => (Err(e), None, ""), + // Keep the adopted offerer in a guard that stays armed across the race AND + // secure_connection, so cancellation of this future by the outer race (or a + // secure-handshake failure) closes the pc instead of leaking it. It is + // disarmed only once WebRTC is confirmed the winning, secured transport. + let mut webrtc_guard = None; + let race_result = if let Some(webrtc) = webrtc_for_connect { + webrtc_guard = Some(OffererGuard::new(webrtc.clone())); + let mut raced = webrtc; + let webrtc_fut = async move { + raced.wait_connected(CONNECT_TIMEOUT).await?; + Ok((Stream::WebRTC(raced), None, "WebRTC")) + } + .boxed(); + // The peer answered WebRTC: prefer P2P. The relay result is held for + // the preference window so WebRTC can win even though a relay TCP + // connect completes much faster than ICE + DTLS setup. + race_transports_prefer_webrtc( + webrtc_fut, + connect_futures, + Self::WEBRTC_PREFER_WINDOW_MS, + |result| result.2 == "IPv6", + ) + .await + } else { + // Run all connection attempts concurrently, take the first success. + select_ok(connect_futures).await.map(|r| r.0) }; if let Some(stop) = webrtc_bridge_stop { let _ = stop.send(()); } - let mut conn = conn?; + // The ? / secure_connection failures below return early; webrtc_guard stays + // in scope and closes the offerer on any such exit (loss, error, cancellation). + let (mut conn, kcp, mut typ) = race_result?; feedback = rr.feedback; log::info!("{:?} used to establish {typ} connection", start.elapsed()); - let pk = - Self::secure_connection(&peer, signed_id_pk, &key, &mut conn).await?; + let pk = match Self::secure_connection( + &peer, + signed_id_pk.clone(), + &key, + &mut conn, + ) + .await + { + Ok(pk) => pk, + Err(e) if typ == "WebRTC" => { + // WebRTC won the race but identity/DTLS binding failed. Fall back + // to a freshly-coordinated relay (request_relay negotiates a new + // uuid, immune to the raced create_relay having consumed the + // original pairing) so a bad WebRTC handshake does not kill the + // whole session when relay is available. + log::warn!( + "WebRTC secure handshake failed ({}), falling back to relay", + e + ); + drop(webrtc_guard.take()); + let mut relay_conn = Self::request_relay( + &peer, + relay_server_rr, + &rendezvous_server, + !signed_id_pk.is_empty(), + &key, + &token, + conn_type, + ) + .await + .map_err(|relay_e| { + anyhow!( + "WebRTC secure failed ({}); relay fallback also failed: {}", + e, + relay_e + ) + })?; + let pk = Self::secure_connection( + &peer, + signed_id_pk, + &key, + &mut relay_conn, + ) + .await?; + conn = relay_conn; + typ = if use_ws() { "WebSocket" } else { "Relay" }; + pk + } + Err(e) => return Err(e), + }; + // Compute the direct/relayed flag (an await) BEFORE disarming the guard, so + // a cancellation of this future during webrtc_relayed() still closes the pc + // via the guard's drop. Matches connect()'s ordering. + let direct = match typ { + "IPv6" => true, + // WebRTC through a TURN server is relayed traffic; report it as such. + "WebRTC" => !conn.webrtc_relayed().await.unwrap_or(false), + _ => false, + }; + // Secured and WebRTC won: disarm so the returned conn keeps the pc alive. + if typ == "WebRTC" { + if let Some(guard) = webrtc_guard.take() { + let _ = guard.into_inner(); + } + } return Ok(( - (conn, typ == "IPv6" || typ == "WebRTC", pk, kcp, typ), + (conn, direct, pk, kcp, typ), (feedback, rendezvous_server), false, )); } Some(rendezvous_message::Union::IceCandidate(ice)) => { if Self::is_expected_webrtc_ice_candidate(&ice, &webrtc_session_key) { - pending_webrtc_ice.push(ice.candidate); + if pending_webrtc_ice.len() < Self::MAX_PENDING_WEBRTC_ICE { + pending_webrtc_ice.push(ice.candidate); + } else { + log::warn!( + "dropping WebRTC ICE candidate: pending buffer full ({})", + Self::MAX_PENDING_WEBRTC_ICE + ); + } } else { log::debug!( "dropping ICE candidate for unexpected WebRTC session key {}", @@ -741,16 +1079,26 @@ impl Client { let mut webrtc_bridge_stop = None; let mut webrtc_for_connect = None; if !webrtc_sdp_answer.is_empty() { - if let Some(webrtc) = webrtc_offerer.take() { - if let Err(err) = webrtc.set_remote_endpoint(&webrtc_sdp_answer).await { - log::warn!("failed to set WebRTC answer: {}", err); - drop(socket); - } else { - for candidate in pending_webrtc_ice.drain(..) { - if let Err(err) = webrtc.add_remote_ice_candidate(&candidate).await { - log::warn!("failed to add buffered WebRTC ICE candidate: {}", err); + if let Some(guard) = webrtc_offerer.take() { + // Run the awaited setup on the guard's borrowed stream so a cancellation during + // these awaits still closes the pc via the guard's drop; take ownership only at + // the synchronous handoff below. + let setup_ok = if let Some(webrtc) = guard.stream() { + if let Err(err) = webrtc.set_remote_endpoint(&webrtc_sdp_answer).await { + log::warn!("failed to set WebRTC answer: {}", err); + false + } else { + for candidate in pending_webrtc_ice.drain(..) { + if let Err(err) = webrtc.add_remote_ice_candidate(&candidate).await { + log::warn!("failed to add buffered WebRTC ICE candidate: {}", err); + } } + true } + } else { + false + }; + if let Some(webrtc) = setup_ok.then(|| guard.into_inner()).flatten() { let session_key = webrtc.session_key().to_owned(); let local_ice_rx = webrtc.take_local_ice_rx(); webrtc_bridge_stop = Some(Self::spawn_webrtc_ice_bridge( @@ -761,6 +1109,9 @@ impl Client { session_key, )); webrtc_for_connect = Some(webrtc); + } else { + // setup failed (guard dropped -> pc closed) or no stream: release the socket. + drop(socket); } } else { drop(socket); @@ -768,7 +1119,18 @@ impl Client { } else { drop(socket); } + // An offerer never adopted into a connection attempt (e.g. the peer returned no WebRTC + // answer) is closed by the guard's drop here, so its pc does not linger in SESSIONS. + drop(webrtc_offerer.take()); if peer_addr.port() == 0 { + // Bailing before connect(): an offerer already adopted into webrtc_for_connect was + // disarmed out of its guard, so close it (and stop its bridge) explicitly here. + if let Some(webrtc) = webrtc_for_connect.take() { + Self::spawn_close_webrtc(webrtc); + } + if let Some(stop) = webrtc_bridge_stop.take() { + let _ = stop.send(()); + } bail!("Failed to connect via rendezvous server"); } let time_used = start.elapsed().as_millis() as u64; @@ -839,6 +1201,10 @@ impl Client { Option, &'static str, )> { + // Guard the offerer for the whole of connect(): any early return — cancellation during the + // awaits below, the relay override, or a secure_connection failure — closes its pc via the + // guard's drop. Disarmed only once WebRTC is the confirmed winning, secured transport. + let mut webrtc_guard = webrtc_offerer.map(OffererGuard::new); let direct_failures = interface.get_lch().read().unwrap().direct_failures; let mut connect_timeout = 0; const MIN: u64 = 1000; @@ -889,11 +1255,20 @@ impl Client { if let Some(udp_socket_v6) = udp_socket_v6 { connect_futures.push(udp_nat_connect(udp_socket_v6, "IPv6", connect_timeout).boxed()); } - if let Some(mut webrtc) = webrtc_offerer { + // Race a clone of the offerer; the guard retains its own clone so a losing/cancelled race + // still closes the pc (select_ok drops the future's clone without closing). + if let Some(stream) = webrtc_guard.as_ref().and_then(|g| g.stream()) { + let mut raced = stream.clone(); + // The punch-tuned timeout can be as low as 1s — enough for a raw TCP SYN but not for + // candidate trickle + ICE checks + DTLS. Give WebRTC its own floor (prefer-P2P) so a + // viable P2P path is not abandoned before it can complete; TCP/UDP keep the tighter + // timeout, so a working direct connection still wins the race immediately, and the + // relay fallback below only waits the extra time when direct attempts all failed. + let webrtc_timeout = connect_timeout.max(Self::WEBRTC_PREFER_WINDOW_MS); connect_futures.push( async move { - webrtc.wait_connected(connect_timeout).await?; - Ok((Stream::WebRTC(webrtc), None, "WebRTC")) + raced.wait_connected(webrtc_timeout).await?; + Ok((Stream::WebRTC(raced), None, "WebRTC")) } .boxed(), ); @@ -906,9 +1281,14 @@ impl Client { if let Some(stop) = webrtc_bridge_stop { let _ = stop.send(()); } + // webrtc_guard stays armed across the relay override and secure_connection below; it is + // disarmed only at the successful return when WebRTC is the kept transport. let mut direct = !conn.is_err(); - if interface.is_force_relay() || conn.is_err() { + // A WebRTC win under force_relay only happens when the pc was built with Relay-only ICE + // (TURN configured), which already honors the relay requirement — keep it instead of + // replacing it with the RustDesk relay. + if (interface.is_force_relay() && typ != "WebRTC") || conn.is_err() { if !relay_server.is_empty() { let switch_code = interface.get_switch_code(); conn = Self::request_relay( @@ -939,15 +1319,71 @@ impl Client { start.elapsed(), punch_type ); - let res = Self::secure_connection(peer_id, signed_id_pk, key, &mut conn).await; + let res = Self::secure_connection(peer_id, signed_id_pk.clone(), key, &mut conn).await; let pk: Option> = match res { Ok(pk) => pk, + Err(e) if typ == "WebRTC" && !relay_server.is_empty() => { + // WebRTC won the race but identity/DTLS binding failed; fall back to a freshly + // coordinated relay instead of failing the whole attempt. The guard is dropped + // first so the bad pc is closed promptly. + log::warn!("WebRTC secure handshake failed ({}), falling back to relay", e); + drop(webrtc_guard.take()); + match Self::request_relay( + peer_id, + relay_server.to_owned(), + rendezvous_server, + !signed_id_pk.is_empty(), + key, + token, + conn_type, + ) + .await + { + Ok(mut relay_conn) => { + match Self::secure_connection(peer_id, signed_id_pk, key, &mut relay_conn) + .await + { + Ok(pk) => { + conn = relay_conn; + typ = "Relay"; + direct = false; + pk + } + Err(e) => { + interface.update_direct(Some(false)); + bail!(e); + } + } + } + Err(relay_e) => { + interface.update_direct(Some(direct)); + bail!( + "WebRTC secure failed ({}); relay fallback also failed: {}", + e, + relay_e + ); + } + } + } Err(e) => { // this direct is mainly used by on_establish_connection_error, so we update it here before bail interface.update_direct(Some(direct)); + // webrtc_guard is still armed here, so a WebRTC winner whose secure handshake + // failed is closed by the guard's drop as we bail (no explicit close needed). bail!(e); } }; + if typ == "WebRTC" { + // WebRTC through a TURN server (force_relay, or a TURN pair winning under All + // policy) is relayed traffic; report the direct flag accordingly. + if conn.webrtc_relayed().await.unwrap_or(false) { + direct = false; + } + // Secured: disarm so the returned conn keeps the pc alive. + if let Some(guard) = webrtc_guard.take() { + let _ = guard.into_inner(); + } + } log::debug!("{} punch secure_connection ok", punch_type); Ok((conn, direct, pk, kcp, typ)) } @@ -964,6 +1400,15 @@ impl Client { } else { key }); + // A WebRTC channel is peer-authenticated only once its DTLS fingerprint is bound to the + // verified peer identity below. Once a trusted identity IS established, every binding + // failure fails closed: any peer able to answer WebRTC also signs its fingerprint, so a + // mismatch is concrete evidence of a rendezvous/relay MITM (not a legacy peer), and + // callers fall back to a fresh relay connection rather than proceeding on that channel. + // Without a trusted identity (absent, or unverifiable under our configured root) no + // binding is possible at all; WebRTC then proceeds like TCP's non-secure fallback — + // DTLS-encrypted but reported unsecured. TCP/UDP/relay keep their existing behavior. + let is_webrtc = conn.is_webrtc(); let mut sign_pk = None; let mut option_pk = None; if !signed_id_pk.is_empty() { @@ -982,6 +1427,17 @@ impl Client { let sign_pk = match sign_pk { Some(v) => v, None => { + // No trusted peer identity: either the deployment provides none (empty + // signed_id_pk, key-less) or the blob does not verify under our configured root + // (typical benign cause: self-hosted server with the client not configured with + // its key, so verification runs against the built-in RS_PUB_KEY). Either way no + // fingerprint binding is possible, which is the same cryptographic state as + // key-less: DTLS-encrypted to an unauthenticated peer. Proceed like TCP's + // non-secure fallback with is_secured() left false. Bailing here instead would + // only push the session onto a relay where the same blob fails verification + // again and the payload then runs in plaintext — strictly worse than + // unauthenticated DTLS, while an active attacker reaches the same warned, + // unauthenticated endpoint either way. // send an empty message out in case server is setting up secure and waiting for first message conn.send(&Message::new()).await?; return Ok(option_pk); @@ -992,8 +1448,22 @@ impl Client { let bytes = res?; if let Ok(msg_in) = Message::parse_from_bytes(&bytes) { if let Some(message::Union::SignedId(si)) = msg_in.union { - if let Ok((id, their_pk_b)) = decode_id_pk(&si.id, &sign_pk) { + if let Ok((id, their_pk_b, signed_fp)) = decode_id_pk_dtls(&si.id, &sign_pk) { if id == peer_id { + // WebRTC only: bind the DTLS channel to the verified peer identity. + // webrtc-rs already verified the peer certificate matches the remote + // SDP fingerprint, so requiring the peer to have signed that same + // fingerprint defeats a rendezvous/relay MITM that swaps SDPs. Fail + // closed: an unreadable/empty/mismatched fingerprint aborts the + // WebRTC connection rather than silently accepting it. + if is_webrtc { + let actual_fp = conn.dtls_fingerprint(false).await.ok_or_else( + || anyhow!("WebRTC DTLS fingerprint unavailable"), + )?; + if signed_fp.is_empty() || signed_fp != actual_fp { + bail!("WebRTC DTLS fingerprint not bound to peer identity (possible MITM)"); + } + } let (asymmetric_value, symmetric_value, key) = create_symmetric_key_msg(their_pk_b); let mut msg_out = Message::new(); @@ -1005,10 +1475,16 @@ impl Client { timeout(CONNECT_TIMEOUT, conn.send(&msg_out)).await??; conn.set_key(key); } else { + if is_webrtc { + bail!("WebRTC handshake id mismatch (possible MITM)"); + } log::error!("Handshake failed: sign failure"); conn.send(&Message::new()).await?; } } else { + if is_webrtc { + bail!("WebRTC peer identity could not be verified (refusing unbound channel)"); + } // fall back to non-secure connection in case pk mismatch log::info!("pk mismatch, fall back to non-secure"); let mut msg_out = Message::new(); @@ -1016,10 +1492,16 @@ impl Client { conn.send(&msg_out).await?; } } else { + if is_webrtc { + bail!("WebRTC handshake received an unexpected message type"); + } log::error!("Handshake failed: invalid message type"); conn.send(&Message::new()).await?; } } else { + if is_webrtc { + bail!("WebRTC handshake received a malformed message"); + } log::error!("Handshake failed: invalid message format"); conn.send(&Message::new()).await?; } @@ -4608,3 +5090,146 @@ async fn udp_nat_connect( })?; Ok((res.1, Some(res.0), typ)) } + +#[cfg(test)] +mod webrtc_race_tests { + use super::race_transports_prefer_webrtc; + use hbb_common::{ + anyhow::anyhow, + futures::future::{BoxFuture, FutureExt}, + tokio, + tokio::time::{sleep, Duration, Instant}, + ResultType, + }; + + fn ok_after(ms: u64, tag: &'static str) -> BoxFuture<'static, ResultType<&'static str>> { + async move { + sleep(Duration::from_millis(ms)).await; + Ok(tag) + } + .boxed() + } + + fn err_after(ms: u64, what: &'static str) -> BoxFuture<'static, ResultType<&'static str>> { + async move { + sleep(Duration::from_millis(ms)).await; + Err(anyhow!(what)) + } + .boxed() + } + + const NOT_P2P: fn(&&'static str) -> bool = |_| false; + + #[tokio::test] + async fn webrtc_preferred_over_faster_relay_within_window() { + let got = race_transports_prefer_webrtc( + ok_after(120, "webrtc"), + vec![ok_after(10, "relay")], + 60_000, + NOT_P2P, + ) + .await + .unwrap(); + assert_eq!(got, "webrtc"); + } + + #[tokio::test] + async fn unfinished_ipv6_still_beats_held_relay() { + let start = Instant::now(); + let got = race_transports_prefer_webrtc( + ok_after(60_000, "webrtc"), + vec![ok_after(10, "relay"), ok_after(100, "ipv6")], + 1_000, + |t| *t == "ipv6", + ) + .await + .unwrap(); + assert_eq!(got, "ipv6"); + assert!(start.elapsed() < Duration::from_secs(5)); + } + + #[tokio::test] + async fn held_relay_committed_when_window_expires() { + let start = Instant::now(); + let got = race_transports_prefer_webrtc( + ok_after(60_000, "webrtc"), + vec![ok_after(10, "relay")], + 150, + NOT_P2P, + ) + .await + .unwrap(); + assert_eq!(got, "relay"); + assert!(start.elapsed() < Duration::from_secs(5)); + } + + #[tokio::test] + async fn held_relay_committed_when_webrtc_fails() { + let start = Instant::now(); + let got = race_transports_prefer_webrtc( + err_after(50, "webrtc dead"), + vec![ok_after(10, "relay")], + 60_000, + NOT_P2P, + ) + .await + .unwrap(); + assert_eq!(got, "relay"); + assert!(start.elapsed() < Duration::from_secs(5)); + } + + #[tokio::test] + async fn relay_committed_directly_after_webrtc_failed() { + let got = race_transports_prefer_webrtc( + err_after(5, "webrtc dead"), + vec![ok_after(100, "relay")], + 60_000, + NOT_P2P, + ) + .await + .unwrap(); + assert_eq!(got, "relay"); + } + + #[tokio::test] + async fn webrtc_still_wins_past_window_when_relay_dead() { + let got = race_transports_prefer_webrtc( + ok_after(300, "webrtc"), + vec![err_after(10, "relay dead")], + 50, + NOT_P2P, + ) + .await + .unwrap(); + assert_eq!(got, "webrtc"); + } + + #[tokio::test] + async fn p2p_transport_committed_immediately() { + let start = Instant::now(); + let got = race_transports_prefer_webrtc( + ok_after(60_000, "webrtc"), + vec![ok_after(10, "ipv6")], + 60_000, + |t| *t == "ipv6", + ) + .await + .unwrap(); + assert_eq!(got, "ipv6"); + assert!(start.elapsed() < Duration::from_secs(5)); + } + + #[tokio::test] + async fn both_failing_compose_error() { + let err = race_transports_prefer_webrtc( + err_after(10, "webrtc dead"), + vec![err_after(20, "relay dead")], + 1_000, + NOT_P2P, + ) + .await + .unwrap_err() + .to_string(); + assert!(err.contains("webrtc dead") && err.contains("relay dead"), "{}", err); + } +} diff --git a/src/client/io_loop.rs b/src/client/io_loop.rs index d7a4f570f..83a5656e4 100644 --- a/src/client/io_loop.rs +++ b/src/client/io_loop.rs @@ -196,6 +196,9 @@ impl Remote { tokio::time::sleep(KCP_CLOSE_REASON_FLUSH_DELAY).await; } self.handle_disconnected(round); + // Close the WebRTC pc on this decline path too (no-op for TCP/WS); otherwise its + // pc lingers in the global session cache until ICE decays on its own. + peer.close_webrtc().await; return; } self.handler.update_direct(Some(direct)); @@ -342,6 +345,10 @@ impl Remote { } } log::debug!("Exit io_loop of id={}", self.handler.get_id()); + // Close the WebRTC peer connection (if this session used it) so its pc is not left + // lingering in the global session cache after the session ends; dropping `peer` + // alone does not release it. No-op for TCP/WebSocket transports. + peer.close_webrtc().await; // Stop client audio server. if let Some(s) = self.stop_voice_call_sender.take() { s.send(()).ok(); diff --git a/src/common.rs b/src/common.rs index 648bc6b5c..25c4eae71 100644 --- a/src/common.rs +++ b/src/common.rs @@ -2126,11 +2126,21 @@ pub fn get_rs_pk(str_base64: &str) -> Option { } pub fn decode_id_pk(signed: &[u8], key: &sign::PublicKey) -> ResultType<(String, [u8; 32])> { + let (id, pk, _) = decode_id_pk_dtls(signed, key)?; + Ok((id, pk)) +} + +/// Like [`decode_id_pk`] but also returns the signed DTLS certificate fingerprint (empty string +/// for non-WebRTC peers), used to bind a WebRTC DTLS channel to the verified peer identity. +pub fn decode_id_pk_dtls( + signed: &[u8], + key: &sign::PublicKey, +) -> ResultType<(String, [u8; 32], String)> { let res = IdPk::parse_from_bytes( &sign::verify(signed, key).map_err(|_| anyhow!("Signature mismatch"))?, )?; if let Some(pk) = get_pk(&res.pk) { - Ok((res.id, pk)) + Ok((res.id, pk, res.dtls_fingerprint)) } else { bail!("Wrong their public length"); } diff --git a/src/port_forward.rs b/src/port_forward.rs index 0cc640663..f772ccdcd 100644 --- a/src/port_forward.rs +++ b/src/port_forward.rs @@ -156,6 +156,9 @@ async fn connect_and_login( if !stream.is_secured() && !crate::common::is_direct_ip_access(id) { if !confirm_insecure_connection(&interface, ui_receiver).await { *close_port_forward = true; + // Close the WebRTC pc on this decline path too (no-op for TCP/WS), matching every + // other exit in this function; a bare drop leaks it in the global session cache. + stream.close_webrtc().await; return Ok(None); } } @@ -170,6 +173,7 @@ async fn connect_and_login( tokio::select! { res = timeout(READ_TIMEOUT, stream.next()) => match res { Err(_) => { + stream.close_webrtc().await; bail!("Timeout"); } Ok(Some(Ok(bytes))) => { @@ -177,7 +181,13 @@ async fn connect_and_login( received = true; interface.update_received(true); } - let msg_in = Message::parse_from_bytes(&bytes)?; + let msg_in = match Message::parse_from_bytes(&bytes) { + Ok(msg) => msg, + Err(err) => { + stream.close_webrtc().await; + return Err(err.into()); + } + }; match msg_in.union { Some(message::Union::Hash(hash)) => { challenge = Some(hash.clone()); @@ -188,6 +198,7 @@ async fn connect_and_login( Some(message::Union::LoginResponse(lr)) => match lr.union { Some(login_response::Union::Error(err)) => { if !interface.handle_login_error(&err) { + stream.close_webrtc().await; return Ok(None); } } @@ -204,9 +215,11 @@ async fn connect_and_login( } } Ok(Some(Err(err))) => { + stream.close_webrtc().await; bail!("Connection closed: {}", err); } _ => { + stream.close_webrtc().await; bail!("Reset by the peer"); } }, @@ -226,6 +239,7 @@ async fn connect_and_login( if let Some(Ok(bytes)) = res { buffer.extend(bytes); } else { + stream.close_webrtc().await; return Ok(None); } }, @@ -330,6 +344,7 @@ async fn run_forward(forward: Framed, stream: Stream) -> }, } } + stream.close_webrtc().await; Ok(()) } diff --git a/src/rendezvous_mediator.rs b/src/rendezvous_mediator.rs index fb04d2c4a..d7f71be49 100644 --- a/src/rendezvous_mediator.rs +++ b/src/rendezvous_mediator.rs @@ -676,6 +676,17 @@ impl RendezvousMediator { Ok(()) } + /// Build the WebRTC answerer for a punch-hole offer and return the SDP answer that rides in + /// the punch reply (PunchHoleSent / RelayResponse). + /// + /// This is awaited inline on the punch-reply critical path (also serializing the mediator's + /// message loop), which is acceptable only because everything awaited here is local-only — + /// pc construction + DTLS cert keygen + SDP answer, sub-millisecond in practice. Trickle ICE + /// makes that possible: the answer carries no candidates; STUN/TURN gathering runs afterward + /// and trickles via IceCandidate messages. Keep network I/O out of this path — actual + /// connection setup (wait_connected + create_tcp_connection) belongs in the detached task + /// below. On error the caller degrades to an empty answer and the punch proceeds without + /// WebRTC. async fn spawn_webrtc_answerer( &self, ph: &PunchHole, @@ -686,10 +697,27 @@ impl RendezvousMediator { ) -> ResultType { let mut stream = WebRTCStream::new(&ph.webrtc_sdp_offer, force_relay, CONNECT_TIMEOUT).await?; - let answer = stream.get_local_endpoint().await?; + let answer = match stream.get_local_endpoint().await { + Ok(answer) => answer, + Err(e) => { + // Close the freshly-created pc so a failure here doesn't leak it in SESSIONS. + stream.close().await; + return Err(e); + } + }; let session_key = stream.session_key().to_owned(); let return_route = ph.socket_addr.clone(); + // A duplicate PunchHole (the offerer re-sends the same request across punch attempts) + // resolves to the SESSIONS-cached stream. `take_local_ice_rx` yields the receiver + // exactly once per stream instance, so `None` here means an answerer was already + // spawned for this offer: return the (identical) cached answer without spawning a + // second connect task. Otherwise two `create_tcp_connection` tasks would detach and + // read the same data channel, interleaving the handshake and corrupting the session. + let Some(mut local_ice_rx) = stream.take_local_ice_rx() else { + return Ok(answer); + }; + let (remote_ice_tx, mut remote_ice_rx) = mpsc::unbounded_channel::(); WEBRTC_ICE_TXS .lock() @@ -706,7 +734,7 @@ impl RendezvousMediator { } }); - if let Some(mut local_ice_rx) = stream.take_local_ice_rx() { + { let sender = self.rz_sender.clone(); let socket_addr = return_route.clone(); let session_key_for_ice = session_key.clone(); @@ -719,7 +747,15 @@ impl RendezvousMediator { candidate, ..Default::default() }); - let _ = sender.send(msg); + let _ = sender.send(msg.clone()); + // The mediator channel to the rendezvous server is UDP in the default setup, + // so a candidate can be lost in flight; re-send once after a short delay (the + // peer's ICE agent dedups repeats, so the second copy is free). + let sender = sender.clone(); + tokio::spawn(async move { + sleep(0.4).await; + let _ = sender.send(msg); + }); } }); } @@ -733,8 +769,17 @@ impl RendezvousMediator { .remove(&session_key_for_cleanup); if let Err(err) = result { log::warn!("webrtc wait_connected failed: {}", err); + // Release the pc now rather than waiting for the ICE agent to time out into a + // terminal state (~30s); this also drops the SESSIONS entry promptly. + stream.close().await; return; } + // create_tcp_connection takes ownership of the stream; keep a handle to close the pc + // once the session returns. It runs the whole session and returns Ok on normal end, + // Err on setup failure — either way the pc must be closed, else it lingers forever in + // SESSIONS (its state handler only fires on a terminal ICE state, which a cleanly + // closed session may never reach) leaking the pc, channels, and socket fds. + let stream_for_cleanup = stream.clone(); if let Err(err) = crate::server::create_tcp_connection( server, Stream::WebRTC(stream), @@ -746,6 +791,7 @@ impl RendezvousMediator { { log::warn!("failed to create WebRTC server connection: {}", err); } + stream_for_cleanup.close().await; }); Ok(answer) @@ -760,19 +806,32 @@ impl RendezvousMediator { return Ok(()); } let peer_addr_v6 = hbb_common::AddrMangle::decode(&ph.socket_addr_v6); - let relay = use_ws() || Config::is_proxy() || ph.force_relay; + let local_proxy = use_ws() || Config::is_proxy(); + let relay = local_proxy || ph.force_relay; let mut socket_addr_v6 = Default::default(); let meta = connection_meta( ph.control_permissions.clone().into_option(), ph.controlled_context.clone().into_option(), ); - let webrtc_sdp_answer = if !ph.webrtc_sdp_offer.is_empty() { - self.spawn_webrtc_answerer(&ph, relay, server.clone(), peer_addr, meta.clone()) - .await - .unwrap_or_else(|err| { - log::warn!("failed to create WebRTC answer: {}", err); - String::new() - }) + let control_permissions = ph.control_permissions.clone().into_option(); + // WebRTC opens its own ICE sockets, so it must never be used when local traffic is routed + // through SOCKS/WebSocket. force_relay is different: relay-only ICE is viable with TURN. + let webrtc_viable = !ph.webrtc_sdp_offer.is_empty() + && !local_proxy + && (!ph.force_relay || WebRTCStream::has_turn_server()); + let webrtc_sdp_answer = if webrtc_viable { + self.spawn_webrtc_answerer( + &ph, + ph.force_relay, + server.clone(), + peer_addr, + meta.clone(), + ) + .await + .unwrap_or_else(|err| { + log::warn!("failed to create WebRTC answer: {}", err); + String::new() + }) } else { String::new() }; diff --git a/src/server.rs b/src/server.rs index 5af982772..ff3f1a443 100644 --- a/src/server.rs +++ b/src/server.rs @@ -211,11 +211,21 @@ pub async fn create_tcp_connection( let sk = sign::SecretKey(sk_); let mut msg_out = Message::new(); let (our_pk_b, our_sk_b) = box_::gen_keypair(); + // On a WebRTC transport, bind our DTLS certificate fingerprint to our signed identity so + // the controller can verify the DTLS channel it negotiated actually terminates at us + // (not a rendezvous/relay that swapped the SDP fingerprint). Empty on other transports. + // Fail immediately on WebRTC if the local fingerprint is unavailable: signing "" would + // only make the client fail-closed after a wasted round-trip. + let dtls_fingerprint = stream.dtls_fingerprint(true).await.unwrap_or_default(); + if stream.is_webrtc() && dtls_fingerprint.is_empty() { + bail!("WebRTC local DTLS fingerprint unavailable"); + } msg_out.set_signed_id(SignedId { id: sign::sign( &IdPk { id: Config::get_id(), pk: Bytes::from(our_pk_b.0.to_vec()), + dtls_fingerprint, ..Default::default() } .write_to_bytes()