diff --git a/src/client.rs b/src/client.rs index 84a1b4fc8..eeec1c4a8 100644 --- a/src/client.rs +++ b/src/client.rs @@ -302,12 +302,10 @@ async fn race_transports_prefer_webrtc<'a, T: 'a>( } } -fn request_can_carry_webrtc(udp_port: u16, force_relay: bool) -> bool { - // A normal TCP punch request must close its rendezvous socket before reusing that local - // address, while WebRTC trickle ICE retains the socket as its signaling bridge. Therefore a - // request with udp_port=0 must never carry WebRTC. UDP requests can carry it because their - // punch socket is separate; force-relay requests can too because they never enter TCP punching. - udp_port > 0 || force_relay +fn request_allows_tcp_punch(webrtc_sdp_offer: &str) -> bool { + // WebRTC trickle ICE retains the rendezvous socket as its signaling bridge. Only a request + // without an offer may close that socket and reuse its local address for TCP punching. + webrtc_sdp_offer.is_empty() } impl Client { @@ -456,18 +454,17 @@ impl Client { } else { (None, None) }; - let ipv6 = if crate::get_ipv6_punch_enabled() { + // Under force-relay a direct IPv6 path is not allowed, so don't bind the v6 socket; + // the controlled side likewise skips v6 when relaying. + let ipv6 = if crate::get_ipv6_punch_enabled() && !interface.is_force_relay() { crate::get_ipv6_socket().await } else { None }; - // Prepare WebRTC only for a possible UDP request, or for force-relay where TURN is the - // only WebRTC path. `_start_inner` applies the stricter wire invariant after the UDP NAT - // test: an actual request with udp_port=0 never includes the offer. - let may_prepare_webrtc = udp.0.is_some() || interface.is_force_relay(); - let webrtc_offerer = if may_prepare_webrtc - && Self::should_create_webrtc_offerer(&interface) - { + // WebRTC uses its own ICE sockets and does not depend on the legacy UDP punch socket. + // When this request carries an offer, `_start_inner` keeps its rendezvous socket solely + // for trickle signaling; a separate offer-less request owns any TCP punch attempt. + 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) => { @@ -493,14 +490,14 @@ impl Client { servers.clone(), contained, ); - if udp.0.is_none() { + if interface.is_force_relay() || (udp.0.is_none() && !has_webrtc_offerer) { return fut.await; } let preferred_fut = fut.boxed(); // This is deliberately a pure TCP punch request: its WebRTC argument must stay `None`. // TCP punching closes the rendezvous socket before binding a new connection to the same // local address; a WebRTC ICE bridge would retain that socket and break the port reuse. - // The preferred request above may carry WebRTC only because it owns a separate UDP socket. + // The preferred request retains its own socket for WebRTC signaling. let fallback_fut = Self::_start_inner( peer.to_owned(), key.to_owned(), @@ -566,19 +563,23 @@ impl Client { /// 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. + /// Skips it when a SOCKS proxy is configured: WebRTC's ICE binds its own UDP sockets and + /// speaks STUN directly, which would bypass the proxy policy and can leak the real IP. + /// WebSocket mode does NOT skip it: ws only tunnels the signaling/relay legs to the server + /// (and `connect_tcp` is ws-aware for them), while ICE remains the only viable P2P path in + /// ws deployments where classic punching is forced to relay. Independent of the udp-punch + /// option too — the offer rides any request, and a server or peer without WebRTC support + /// drops the field, so the race simply proceeds without it. + /// 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 races it without a P2P preference delay. The caller additionally - /// requires either a usable UDP request or force_relay; a normal TCP punch request must never - /// carry a WebRTC offer because trickle ICE retains the socket TCP punching needs to reuse. + /// RelayResponse path races it without a P2P preference delay. Any request carrying an offer + /// keeps its rendezvous socket for trickle signaling and never reuses it for TCP punching; a + /// separate offer-less request provides the TCP fallback when force-relay is not requested. fn should_create_webrtc_offerer(interface: &impl Interface) -> bool { - if !crate::get_udp_punch_enabled() || use_ws() || Config::is_proxy() { + if Config::is_proxy() { return false; } if interface.is_force_relay() && !WebRTCStream::has_turn_server() { @@ -799,25 +800,25 @@ impl Client { .unwrap_or((None, None)); let udp_nat_port = udp.1.map(|x| *x.lock().unwrap()).unwrap_or(0); let webrtc_sdp_offer = - if request_can_carry_webrtc(udp_nat_port, interface.is_force_relay()) { - if let Some(stream) = webrtc_offerer.as_ref().and_then(|g| g.stream()) { - match stream.get_local_endpoint_trickle().await { - Ok(endpoint) => endpoint, - Err(err) => { - log::warn!("failed to read local WebRTC offer: {}", err); - String::new() - } + if let Some(stream) = webrtc_offerer.as_ref().and_then(|g| g.stream()) { + match stream.get_local_endpoint_trickle().await { + Ok(endpoint) => endpoint, + Err(err) => { + log::warn!("failed to read local WebRTC offer: {}", err); + String::new() } - } else { - String::new() } } else { - // Hard protocol invariant: a normal request with udp_port=0 is a TCP punch - // request. It must not start WebRTC trickle on the rendezvous socket that TCP - // punching needs to close and reuse by local address. String::new() }; - let punch_type = if udp_nat_port > 0 { "UDP" } else { "TCP" }; + let allow_tcp_punch = request_allows_tcp_punch(&webrtc_sdp_offer); + let punch_type = if udp_nat_port > 0 { + "UDP" + } else if allow_tcp_punch { + "TCP" + } else { + "WebRTC" + }; msg_out.set_punch_hole_request(PunchHoleRequest { id: peer.to_owned(), token: token.to_owned(), @@ -893,7 +894,7 @@ impl Client { feedback = ph.feedback; webrtc_sdp_answer = ph.webrtc_sdp_answer; let s = udp.0.take(); - if ph.is_udp && s.is_some() { + if udp_nat_port > 0 && ph.is_udp && s.is_some() { if let Some(s) = s { allow_err!(s.connect(peer_addr).await); udp.0 = Some(s); @@ -1224,6 +1225,7 @@ impl Client { ipv6.0, webrtc_for_connect, webrtc_bridge_stop, + allow_tcp_punch, punch_type, ) .await?, @@ -1252,6 +1254,7 @@ impl Client { udp_socket_v6: Option>, webrtc_offerer: Option, webrtc_bridge_stop: Option>, + allow_tcp_punch: bool, punch_type: &str, ) -> ResultType<( Stream, @@ -1300,14 +1303,16 @@ impl Client { let start = std::time::Instant::now(); let mut connect_futures = Vec::new(); - let fut = connect_tcp_local(peer, Some(local_addr), connect_timeout); - connect_futures.push( - async move { - let conn = fut.await?; - Ok((conn, None, "TCP")) - } - .boxed(), - ); + if allow_tcp_punch { + let fut = connect_tcp_local(peer, Some(local_addr), connect_timeout); + connect_futures.push( + async move { + let conn = fut.await?; + Ok((conn, None, "TCP")) + } + .boxed(), + ); + } if let Some(udp_socket_nat) = udp_socket_nat { connect_futures.push(udp_nat_connect(udp_socket_nat, "UDP", connect_timeout).boxed()); } @@ -1333,8 +1338,13 @@ impl Client { ); } // Run all connection attempts concurrently, return the first successful one - let (mut conn, kcp, mut typ) = match select_ok(connect_futures).await { - Ok(conn) => (Ok(conn.0 .0), conn.0 .1, conn.0 .2), + let direct_result = if connect_futures.is_empty() { + Err(anyhow!("No direct transport available")) + } else { + select_ok(connect_futures).await.map(|conn| conn.0) + }; + let (mut conn, kcp, mut typ) = match direct_result { + Ok(conn) => (Ok(conn.0), conn.1, conn.2), Err(e) => (Err(e), None, ""), }; if let Some(stop) = webrtc_bridge_stop { @@ -5062,13 +5072,10 @@ async fn test_udp_uat( udp_port: Arc>, mut stop_udp_rx: oneshot::Receiver<()>, ) -> ResultType<()> { - let (tx, mut rx) = oneshot::channel::<_>(); - tokio::spawn(async { - if let Ok(v) = crate::test_nat_ipv4().await { - tx.send(v).ok(); - } - }); - + // The punch port must come only from the rendezvous server's TestNatResponse, which + // observes THIS socket's public mapping. A STUN probe binds a different socket and reports + // a different NAT mapping, so racing it here could advertise a port the peer can never + // reach (and, on symmetric NAT, silently poison the whole UDP punch). let start = Instant::now(); let mut msg_out = RendezvousMessage::new(); msg_out.set_test_nat_request(TestNatRequest { @@ -5094,11 +5101,6 @@ async fn test_udp_uat( loop { tokio::select! { - Ok((addr, server)) = &mut rx => { - *udp_port.lock().unwrap() = addr.port(); - log::debug!("UDP NAT test received response from {}: {}", addr, server); - break; - } _ = &mut stop_udp_rx => { log::debug!("UDP NAT test received stop signal after {} packets", packets_sent); break; @@ -5181,7 +5183,7 @@ async fn udp_nat_connect( #[cfg(test)] mod webrtc_race_tests { - use super::{race_transports_prefer_webrtc, request_can_carry_webrtc}; + use super::{race_transports_prefer_webrtc, request_allows_tcp_punch}; use hbb_common::{ anyhow::anyhow, futures::future::{BoxFuture, FutureExt}, @@ -5209,10 +5211,9 @@ mod webrtc_race_tests { const NOT_P2P: fn(&&'static str) -> bool = |_| false; #[test] - fn tcp_punch_request_never_carries_webrtc() { - assert!(!request_can_carry_webrtc(0, false)); - assert!(request_can_carry_webrtc(1, false)); - assert!(request_can_carry_webrtc(0, true)); + fn webrtc_request_never_reuses_its_signaling_socket_for_tcp_punch() { + assert!(request_allows_tcp_punch("")); + assert!(!request_allows_tcp_punch("webrtc://offer")); } #[tokio::test] diff --git a/src/rendezvous_mediator.rs b/src/rendezvous_mediator.rs index 94df47732..1351d75fc 100644 --- a/src/rendezvous_mediator.rs +++ b/src/rendezvous_mediator.rs @@ -38,7 +38,6 @@ use crate::{ }; type Message = RendezvousMessage; -type RendezvousSender = mpsc::UnboundedSender; fn connection_meta( control_permissions: Option, @@ -112,7 +111,6 @@ pub struct RendezvousMediator { host: String, host_prefix: String, keep_alive: i32, - rz_sender: RendezvousSender, } impl RendezvousMediator { @@ -219,13 +217,11 @@ impl RendezvousMediator { let host = check_port(&host, RENDEZVOUS_PORT); log::info!("start udp: {host}"); let (mut socket, mut addr) = new_udp_for(&host, CONNECT_TIMEOUT).await?; - let (rz_sender, mut rz_out_rx) = mpsc::unbounded_channel::(); let mut rz = Self { addr: addr.clone(), host: host.clone(), host_prefix: Self::get_host_prefix(&host), keep_alive: crate::DEFAULT_KEEP_ALIVE, - rz_sender, }; let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT)); @@ -285,9 +281,6 @@ impl RendezvousMediator { }, } }, - Some(msg_out) = rz_out_rx.recv() => { - Sink::Framed(&mut socket, &addr).send(&msg_out).await?; - }, _ = timer.tick() => { if SHOULD_EXIT.load(Ordering::SeqCst) { break; @@ -446,13 +439,11 @@ impl RendezvousMediator { let mut conn = connect_tcp(host.clone(), CONNECT_TIMEOUT).await?; let key = crate::get_key(true).await; crate::secure_tcp(&mut conn, &key).await?; - let (rz_sender, mut rz_out_rx) = mpsc::unbounded_channel::(); let mut rz = Self { addr: conn.local_addr().into_target_addr()?, host: host.clone(), host_prefix: Self::get_host_prefix(&host), keep_alive: crate::DEFAULT_KEEP_ALIVE, - rz_sender, }; let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT)); let mut last_register_sent: Option = None; @@ -480,9 +471,6 @@ impl RendezvousMediator { let msg = Message::parse_from_bytes(&bytes)?; rz.handle_resp(msg.union, Sink::Stream(&mut conn), &server, &mut update_latency).await? } - Some(msg_out) = rz_out_rx.recv() => { - Sink::Stream(&mut conn).send(&msg_out).await?; - } _ = timer.tick() => { if SHOULD_EXIT.load(Ordering::SeqCst) { break; @@ -679,8 +667,9 @@ impl RendezvousMediator { /// 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 — + /// This is awaited inline on the punch-reply critical path (handle_punch_hole runs as its own + /// spawned task, so only this reply is delayed), 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 @@ -735,10 +724,18 @@ impl RendezvousMediator { }); { - let sender = self.rz_sender.clone(); + let host = self.host.clone(); let socket_addr = return_route.clone(); let session_key_for_ice = session_key.clone(); tokio::spawn(async move { + // Candidates ride a dedicated TCP connection to the rendezvous server, like + // the answer, NOT the mediator channel: that channel is UDP in the default + // setup, and target deployments front hbbs with websocket/TCP only, where + // its UDP port is unreachable. The server keeps candidate-carrying TCP + // connections open, so one lazily-opened connection serves the whole + // trickle, and TCP reliability replaces the old 400ms duplicate re-send + // (the controller keeps its own re-send for the server->peer UDP downlink). + let mut conn = None; while let Some(candidate) = local_ice_rx.recv().await { let mut msg = Message::new(); msg.set_ice_candidate(IceCandidate { @@ -747,15 +744,34 @@ impl RendezvousMediator { candidate, ..Default::default() }); - 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); - }); + // One reconnect attempt per candidate: the first send after an hbbs + // restart or an idle-killed connection fails on the stale stream. + for _ in 0..2 { + if conn.is_none() { + match connect_tcp(&*host, CONNECT_TIMEOUT).await { + Ok(s) => conn = Some(s), + Err(err) => { + log::warn!( + "failed to connect for WebRTC ICE candidate: {}", + err + ); + break; + } + } + } + if let Some(s) = conn.as_mut() { + match s.send(&msg).await { + Ok(()) => break, + Err(err) => { + log::debug!( + "WebRTC ICE candidate send failed, reconnecting: {}", + err + ); + conn = None; + } + } + } + } } }); } @@ -813,11 +829,13 @@ impl RendezvousMediator { ph.control_permissions.clone().into_option(), ph.controlled_context.clone().into_option(), ); - 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. + // WebRTC opens its own ICE sockets, so it must not run under a SOCKS proxy: candidates + // and STUN bypass the proxy and leak the real IP. WebSocket mode does NOT disable it — + // ws only tunnels the signaling/relay legs to the server, classic punching stays forced + // to relay (`relay` above), and the answer rides the RelayResponse, leaving ICE as the + // only P2P path there. force_relay is different: relay-only ICE is viable with TURN. let webrtc_viable = !ph.webrtc_sdp_offer.is_empty() - && !local_proxy + && !Config::is_proxy() && (!ph.force_relay || WebRTCStream::has_turn_server()); let webrtc_sdp_answer = if webrtc_viable { self.spawn_webrtc_answerer( @@ -879,6 +897,23 @@ impl RendezvousMediator { .await?; return Ok(()); } + if !ph.webrtc_sdp_offer.is_empty() { + // WebRTC-only request (udp_port <= 0): return the answer over a short-lived TCP + // connection to the rendezvous server, like create_relay does. It must NOT ride + // the mediator channel: that channel is UDP in the default setup, and the answer + // is the largest message of the punch exchange — a single lost or fragmented + // datagram costs a whole 3s retry round; hbbs also applies UDP-punch semantics + // (source-address observation / is_udp) to PunchHoleSent received over UDP, + // which this request never asked for. + // No TCP punch connection is created or accepted; the controller retains its + // request socket for trickled ICE signaling. IPv6, when present, was started + // above and its address is carried in this same response. + let mut msg_out = Message::new(); + msg_out.set_punch_hole_sent(msg_punch); + let mut socket = connect_tcp(&*self.host, CONNECT_TIMEOUT).await?; + socket.send(&msg_out).await?; + return Ok(()); + } log::debug!("Punch tcp hole to {:?}", peer_addr); let mut socket = { let socket = connect_tcp(&*self.host, CONNECT_TIMEOUT).await?;