mirror of
https://github.com/rustdesk/rustdesk.git
synced 2026-09-11 06:51:00 +03:00
feat: decouple WebRTC from UDP punch, route controlled signaling over TCP
- the WebRTC offer now rides any punch request; only an offer-less request may close and reuse the rendezvous socket for TCP punching (request_allows_tcp_punch replaces the udp_port-based invariant), with a separate offer-less request racing as the TCP fallback - WebSocket mode no longer disables WebRTC — ws only tunnels the signaling/relay legs while ICE stays the only P2P path there; SOCKS proxy still disables it (ICE would bypass the proxy and leak the real IP) - controlled side: WebRTC-only punch replies and trickled ICE candidates go over dedicated TCP connections to the rendezvous server instead of the UDP mediator channel, for ws/TCP-only hbbs deployments; drop the now-redundant rz_sender plumbing and the 400ms candidate re-send on that leg - guard is_udp handling against responses to requests that advertised no udp_port; skip the IPv6 socket bind under force-relay - test_udp_uat: drop the STUN port race — the punch port must come from the rendezvous server's TestNatResponse observing this socket's mapping, a STUN probe from another socket can advertise an unreachable port - bump hbb_common (webrtc 0.13 MSRV pin rationale + upgrade checklist docs) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Submodule libs/hbb_common updated: 0952f18b8e...6aa8fbe46b
137
src/client.rs
137
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 {
|
fn request_allows_tcp_punch(webrtc_sdp_offer: &str) -> bool {
|
||||||
// A normal TCP punch request must close its rendezvous socket before reusing that local
|
// WebRTC trickle ICE retains the rendezvous socket as its signaling bridge. Only a request
|
||||||
// address, while WebRTC trickle ICE retains the socket as its signaling bridge. Therefore a
|
// without an offer may close that socket and reuse its local address for TCP punching.
|
||||||
// request with udp_port=0 must never carry WebRTC. UDP requests can carry it because their
|
webrtc_sdp_offer.is_empty()
|
||||||
// punch socket is separate; force-relay requests can too because they never enter TCP punching.
|
|
||||||
udp_port > 0 || force_relay
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Client {
|
impl Client {
|
||||||
@@ -456,18 +454,17 @@ impl Client {
|
|||||||
} else {
|
} else {
|
||||||
(None, None)
|
(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
|
crate::get_ipv6_socket().await
|
||||||
} else {
|
} else {
|
||||||
None
|
None
|
||||||
};
|
};
|
||||||
// Prepare WebRTC only for a possible UDP request, or for force-relay where TURN is the
|
// WebRTC uses its own ICE sockets and does not depend on the legacy UDP punch socket.
|
||||||
// only WebRTC path. `_start_inner` applies the stricter wire invariant after the UDP NAT
|
// When this request carries an offer, `_start_inner` keeps its rendezvous socket solely
|
||||||
// test: an actual request with udp_port=0 never includes the offer.
|
// for trickle signaling; a separate offer-less request owns any TCP punch attempt.
|
||||||
let may_prepare_webrtc = udp.0.is_some() || interface.is_force_relay();
|
let webrtc_offerer = if Self::should_create_webrtc_offerer(&interface) {
|
||||||
let webrtc_offerer = if may_prepare_webrtc
|
|
||||||
&& Self::should_create_webrtc_offerer(&interface)
|
|
||||||
{
|
|
||||||
match WebRTCStream::new("", interface.is_force_relay(), CONNECT_TIMEOUT).await {
|
match WebRTCStream::new("", interface.is_force_relay(), CONNECT_TIMEOUT).await {
|
||||||
Ok(stream) => Some(stream),
|
Ok(stream) => Some(stream),
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
@@ -493,14 +490,14 @@ impl Client {
|
|||||||
servers.clone(),
|
servers.clone(),
|
||||||
contained,
|
contained,
|
||||||
);
|
);
|
||||||
if udp.0.is_none() {
|
if interface.is_force_relay() || (udp.0.is_none() && !has_webrtc_offerer) {
|
||||||
return fut.await;
|
return fut.await;
|
||||||
}
|
}
|
||||||
let preferred_fut = fut.boxed();
|
let preferred_fut = fut.boxed();
|
||||||
// This is deliberately a pure TCP punch request: its WebRTC argument must stay `None`.
|
// 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
|
// 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.
|
// 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(
|
let fallback_fut = Self::_start_inner(
|
||||||
peer.to_owned(),
|
peer.to_owned(),
|
||||||
key.to_owned(),
|
key.to_owned(),
|
||||||
@@ -566,19 +563,23 @@ impl Client {
|
|||||||
|
|
||||||
/// Whether to build a WebRTC offerer for this 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:
|
/// Skips it when a SOCKS proxy is configured: WebRTC's ICE binds its own UDP sockets and
|
||||||
/// WebRTC's ICE binds its own UDP sockets and speaks STUN directly, which would bypass either
|
/// speaks STUN directly, which would bypass the proxy policy and can leak the real IP.
|
||||||
/// policy (and can leak the real IP through a proxy). Under force_relay the pc uses Relay-only
|
/// WebSocket mode does NOT skip it: ws only tunnels the signaling/relay legs to the server
|
||||||
/// ICE, which gathers nothing and can never connect unless a TURN server is configured, so
|
/// (and `connect_tcp` is ws-aware for them), while ICE remains the only viable P2P path in
|
||||||
/// skip building a guaranteed-dead pc + STUN/TURN gathering + answerer signaling in that case
|
/// ws deployments where classic punching is forced to relay. Independent of the udp-punch
|
||||||
/// too.
|
/// 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:
|
/// 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
|
/// `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
|
/// RelayResponse path races it without a P2P preference delay. Any request carrying an offer
|
||||||
/// requires either a usable UDP request or force_relay; a normal TCP punch request must never
|
/// keeps its rendezvous socket for trickle signaling and never reuses it for TCP punching; a
|
||||||
/// carry a WebRTC offer because trickle ICE retains the socket TCP punching needs to reuse.
|
/// separate offer-less request provides the TCP fallback when force-relay is not requested.
|
||||||
fn should_create_webrtc_offerer(interface: &impl Interface) -> bool {
|
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;
|
return false;
|
||||||
}
|
}
|
||||||
if interface.is_force_relay() && !WebRTCStream::has_turn_server() {
|
if interface.is_force_relay() && !WebRTCStream::has_turn_server() {
|
||||||
@@ -799,25 +800,25 @@ impl Client {
|
|||||||
.unwrap_or((None, None));
|
.unwrap_or((None, None));
|
||||||
let udp_nat_port = udp.1.map(|x| *x.lock().unwrap()).unwrap_or(0);
|
let udp_nat_port = udp.1.map(|x| *x.lock().unwrap()).unwrap_or(0);
|
||||||
let webrtc_sdp_offer =
|
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()) {
|
||||||
if let Some(stream) = webrtc_offerer.as_ref().and_then(|g| g.stream()) {
|
match stream.get_local_endpoint_trickle().await {
|
||||||
match stream.get_local_endpoint_trickle().await {
|
Ok(endpoint) => endpoint,
|
||||||
Ok(endpoint) => endpoint,
|
Err(err) => {
|
||||||
Err(err) => {
|
log::warn!("failed to read local WebRTC offer: {}", err);
|
||||||
log::warn!("failed to read local WebRTC offer: {}", err);
|
String::new()
|
||||||
String::new()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
} else {
|
|
||||||
String::new()
|
|
||||||
}
|
}
|
||||||
} else {
|
} 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()
|
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 {
|
msg_out.set_punch_hole_request(PunchHoleRequest {
|
||||||
id: peer.to_owned(),
|
id: peer.to_owned(),
|
||||||
token: token.to_owned(),
|
token: token.to_owned(),
|
||||||
@@ -893,7 +894,7 @@ impl Client {
|
|||||||
feedback = ph.feedback;
|
feedback = ph.feedback;
|
||||||
webrtc_sdp_answer = ph.webrtc_sdp_answer;
|
webrtc_sdp_answer = ph.webrtc_sdp_answer;
|
||||||
let s = udp.0.take();
|
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 {
|
if let Some(s) = s {
|
||||||
allow_err!(s.connect(peer_addr).await);
|
allow_err!(s.connect(peer_addr).await);
|
||||||
udp.0 = Some(s);
|
udp.0 = Some(s);
|
||||||
@@ -1224,6 +1225,7 @@ impl Client {
|
|||||||
ipv6.0,
|
ipv6.0,
|
||||||
webrtc_for_connect,
|
webrtc_for_connect,
|
||||||
webrtc_bridge_stop,
|
webrtc_bridge_stop,
|
||||||
|
allow_tcp_punch,
|
||||||
punch_type,
|
punch_type,
|
||||||
)
|
)
|
||||||
.await?,
|
.await?,
|
||||||
@@ -1252,6 +1254,7 @@ impl Client {
|
|||||||
udp_socket_v6: Option<Arc<UdpSocket>>,
|
udp_socket_v6: Option<Arc<UdpSocket>>,
|
||||||
webrtc_offerer: Option<WebRTCStream>,
|
webrtc_offerer: Option<WebRTCStream>,
|
||||||
webrtc_bridge_stop: Option<oneshot::Sender<()>>,
|
webrtc_bridge_stop: Option<oneshot::Sender<()>>,
|
||||||
|
allow_tcp_punch: bool,
|
||||||
punch_type: &str,
|
punch_type: &str,
|
||||||
) -> ResultType<(
|
) -> ResultType<(
|
||||||
Stream,
|
Stream,
|
||||||
@@ -1300,14 +1303,16 @@ impl Client {
|
|||||||
let start = std::time::Instant::now();
|
let start = std::time::Instant::now();
|
||||||
|
|
||||||
let mut connect_futures = Vec::new();
|
let mut connect_futures = Vec::new();
|
||||||
let fut = connect_tcp_local(peer, Some(local_addr), connect_timeout);
|
if allow_tcp_punch {
|
||||||
connect_futures.push(
|
let fut = connect_tcp_local(peer, Some(local_addr), connect_timeout);
|
||||||
async move {
|
connect_futures.push(
|
||||||
let conn = fut.await?;
|
async move {
|
||||||
Ok((conn, None, "TCP"))
|
let conn = fut.await?;
|
||||||
}
|
Ok((conn, None, "TCP"))
|
||||||
.boxed(),
|
}
|
||||||
);
|
.boxed(),
|
||||||
|
);
|
||||||
|
}
|
||||||
if let Some(udp_socket_nat) = udp_socket_nat {
|
if let Some(udp_socket_nat) = udp_socket_nat {
|
||||||
connect_futures.push(udp_nat_connect(udp_socket_nat, "UDP", connect_timeout).boxed());
|
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
|
// Run all connection attempts concurrently, return the first successful one
|
||||||
let (mut conn, kcp, mut typ) = match select_ok(connect_futures).await {
|
let direct_result = if connect_futures.is_empty() {
|
||||||
Ok(conn) => (Ok(conn.0 .0), conn.0 .1, conn.0 .2),
|
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, ""),
|
Err(e) => (Err(e), None, ""),
|
||||||
};
|
};
|
||||||
if let Some(stop) = webrtc_bridge_stop {
|
if let Some(stop) = webrtc_bridge_stop {
|
||||||
@@ -5025,13 +5035,10 @@ async fn test_udp_uat(
|
|||||||
udp_port: Arc<Mutex<u16>>,
|
udp_port: Arc<Mutex<u16>>,
|
||||||
mut stop_udp_rx: oneshot::Receiver<()>,
|
mut stop_udp_rx: oneshot::Receiver<()>,
|
||||||
) -> ResultType<()> {
|
) -> ResultType<()> {
|
||||||
let (tx, mut rx) = oneshot::channel::<_>();
|
// The punch port must come only from the rendezvous server's TestNatResponse, which
|
||||||
tokio::spawn(async {
|
// observes THIS socket's public mapping. A STUN probe binds a different socket and reports
|
||||||
if let Ok(v) = crate::test_nat_ipv4().await {
|
// a different NAT mapping, so racing it here could advertise a port the peer can never
|
||||||
tx.send(v).ok();
|
// reach (and, on symmetric NAT, silently poison the whole UDP punch).
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
let start = Instant::now();
|
let start = Instant::now();
|
||||||
let mut msg_out = RendezvousMessage::new();
|
let mut msg_out = RendezvousMessage::new();
|
||||||
msg_out.set_test_nat_request(TestNatRequest {
|
msg_out.set_test_nat_request(TestNatRequest {
|
||||||
@@ -5057,11 +5064,6 @@ async fn test_udp_uat(
|
|||||||
|
|
||||||
loop {
|
loop {
|
||||||
tokio::select! {
|
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 => {
|
_ = &mut stop_udp_rx => {
|
||||||
log::debug!("UDP NAT test received stop signal after {} packets", packets_sent);
|
log::debug!("UDP NAT test received stop signal after {} packets", packets_sent);
|
||||||
break;
|
break;
|
||||||
@@ -5144,7 +5146,7 @@ async fn udp_nat_connect(
|
|||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod webrtc_race_tests {
|
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::{
|
use hbb_common::{
|
||||||
anyhow::anyhow,
|
anyhow::anyhow,
|
||||||
futures::future::{BoxFuture, FutureExt},
|
futures::future::{BoxFuture, FutureExt},
|
||||||
@@ -5172,10 +5174,9 @@ mod webrtc_race_tests {
|
|||||||
const NOT_P2P: fn(&&'static str) -> bool = |_| false;
|
const NOT_P2P: fn(&&'static str) -> bool = |_| false;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn tcp_punch_request_never_carries_webrtc() {
|
fn webrtc_request_never_reuses_its_signaling_socket_for_tcp_punch() {
|
||||||
assert!(!request_can_carry_webrtc(0, false));
|
assert!(request_allows_tcp_punch(""));
|
||||||
assert!(request_can_carry_webrtc(1, false));
|
assert!(!request_allows_tcp_punch("webrtc://offer"));
|
||||||
assert!(request_can_carry_webrtc(0, true));
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
|
|||||||
@@ -38,7 +38,6 @@ use crate::{
|
|||||||
};
|
};
|
||||||
|
|
||||||
type Message = RendezvousMessage;
|
type Message = RendezvousMessage;
|
||||||
type RendezvousSender = mpsc::UnboundedSender<Message>;
|
|
||||||
|
|
||||||
fn connection_meta(
|
fn connection_meta(
|
||||||
control_permissions: Option<ControlPermissions>,
|
control_permissions: Option<ControlPermissions>,
|
||||||
@@ -112,7 +111,6 @@ pub struct RendezvousMediator {
|
|||||||
host: String,
|
host: String,
|
||||||
host_prefix: String,
|
host_prefix: String,
|
||||||
keep_alive: i32,
|
keep_alive: i32,
|
||||||
rz_sender: RendezvousSender,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RendezvousMediator {
|
impl RendezvousMediator {
|
||||||
@@ -219,13 +217,11 @@ impl RendezvousMediator {
|
|||||||
let host = check_port(&host, RENDEZVOUS_PORT);
|
let host = check_port(&host, RENDEZVOUS_PORT);
|
||||||
log::info!("start udp: {host}");
|
log::info!("start udp: {host}");
|
||||||
let (mut socket, mut addr) = new_udp_for(&host, CONNECT_TIMEOUT).await?;
|
let (mut socket, mut addr) = new_udp_for(&host, CONNECT_TIMEOUT).await?;
|
||||||
let (rz_sender, mut rz_out_rx) = mpsc::unbounded_channel::<Message>();
|
|
||||||
let mut rz = Self {
|
let mut rz = Self {
|
||||||
addr: addr.clone(),
|
addr: addr.clone(),
|
||||||
host: host.clone(),
|
host: host.clone(),
|
||||||
host_prefix: Self::get_host_prefix(&host),
|
host_prefix: Self::get_host_prefix(&host),
|
||||||
keep_alive: crate::DEFAULT_KEEP_ALIVE,
|
keep_alive: crate::DEFAULT_KEEP_ALIVE,
|
||||||
rz_sender,
|
|
||||||
};
|
};
|
||||||
|
|
||||||
let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT));
|
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() => {
|
_ = timer.tick() => {
|
||||||
if SHOULD_EXIT.load(Ordering::SeqCst) {
|
if SHOULD_EXIT.load(Ordering::SeqCst) {
|
||||||
break;
|
break;
|
||||||
@@ -446,13 +439,11 @@ impl RendezvousMediator {
|
|||||||
let mut conn = connect_tcp(host.clone(), CONNECT_TIMEOUT).await?;
|
let mut conn = connect_tcp(host.clone(), CONNECT_TIMEOUT).await?;
|
||||||
let key = crate::get_key(true).await;
|
let key = crate::get_key(true).await;
|
||||||
crate::secure_tcp(&mut conn, &key).await?;
|
crate::secure_tcp(&mut conn, &key).await?;
|
||||||
let (rz_sender, mut rz_out_rx) = mpsc::unbounded_channel::<Message>();
|
|
||||||
let mut rz = Self {
|
let mut rz = Self {
|
||||||
addr: conn.local_addr().into_target_addr()?,
|
addr: conn.local_addr().into_target_addr()?,
|
||||||
host: host.clone(),
|
host: host.clone(),
|
||||||
host_prefix: Self::get_host_prefix(&host),
|
host_prefix: Self::get_host_prefix(&host),
|
||||||
keep_alive: crate::DEFAULT_KEEP_ALIVE,
|
keep_alive: crate::DEFAULT_KEEP_ALIVE,
|
||||||
rz_sender,
|
|
||||||
};
|
};
|
||||||
let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT));
|
let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT));
|
||||||
let mut last_register_sent: Option<Instant> = None;
|
let mut last_register_sent: Option<Instant> = None;
|
||||||
@@ -480,9 +471,6 @@ impl RendezvousMediator {
|
|||||||
let msg = Message::parse_from_bytes(&bytes)?;
|
let msg = Message::parse_from_bytes(&bytes)?;
|
||||||
rz.handle_resp(msg.union, Sink::Stream(&mut conn), &server, &mut update_latency).await?
|
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() => {
|
_ = timer.tick() => {
|
||||||
if SHOULD_EXIT.load(Ordering::SeqCst) {
|
if SHOULD_EXIT.load(Ordering::SeqCst) {
|
||||||
break;
|
break;
|
||||||
@@ -679,8 +667,9 @@ impl RendezvousMediator {
|
|||||||
/// Build the WebRTC answerer for a punch-hole offer and return the SDP answer that rides in
|
/// Build the WebRTC answerer for a punch-hole offer and return the SDP answer that rides in
|
||||||
/// the punch reply (PunchHoleSent / RelayResponse).
|
/// the punch reply (PunchHoleSent / RelayResponse).
|
||||||
///
|
///
|
||||||
/// This is awaited inline on the punch-reply critical path (also serializing the mediator's
|
/// This is awaited inline on the punch-reply critical path (handle_punch_hole runs as its own
|
||||||
/// message loop), which is acceptable only because everything awaited here is local-only —
|
/// 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
|
/// 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
|
/// 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
|
/// 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 socket_addr = return_route.clone();
|
||||||
let session_key_for_ice = session_key.clone();
|
let session_key_for_ice = session_key.clone();
|
||||||
tokio::spawn(async move {
|
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 {
|
while let Some(candidate) = local_ice_rx.recv().await {
|
||||||
let mut msg = Message::new();
|
let mut msg = Message::new();
|
||||||
msg.set_ice_candidate(IceCandidate {
|
msg.set_ice_candidate(IceCandidate {
|
||||||
@@ -747,15 +744,34 @@ impl RendezvousMediator {
|
|||||||
candidate,
|
candidate,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
});
|
});
|
||||||
let _ = sender.send(msg.clone());
|
// One reconnect attempt per candidate: the first send after an hbbs
|
||||||
// The mediator channel to the rendezvous server is UDP in the default setup,
|
// restart or an idle-killed connection fails on the stale stream.
|
||||||
// so a candidate can be lost in flight; re-send once after a short delay (the
|
for _ in 0..2 {
|
||||||
// peer's ICE agent dedups repeats, so the second copy is free).
|
if conn.is_none() {
|
||||||
let sender = sender.clone();
|
match connect_tcp(&*host, CONNECT_TIMEOUT).await {
|
||||||
tokio::spawn(async move {
|
Ok(s) => conn = Some(s),
|
||||||
sleep(0.4).await;
|
Err(err) => {
|
||||||
let _ = sender.send(msg);
|
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.control_permissions.clone().into_option(),
|
||||||
ph.controlled_context.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 not run under a SOCKS proxy: candidates
|
||||||
// WebRTC opens its own ICE sockets, so it must never be used when local traffic is routed
|
// and STUN bypass the proxy and leak the real IP. WebSocket mode does NOT disable it —
|
||||||
// through SOCKS/WebSocket. force_relay is different: relay-only ICE is viable with TURN.
|
// 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()
|
let webrtc_viable = !ph.webrtc_sdp_offer.is_empty()
|
||||||
&& !local_proxy
|
&& !Config::is_proxy()
|
||||||
&& (!ph.force_relay || WebRTCStream::has_turn_server());
|
&& (!ph.force_relay || WebRTCStream::has_turn_server());
|
||||||
let webrtc_sdp_answer = if webrtc_viable {
|
let webrtc_sdp_answer = if webrtc_viable {
|
||||||
self.spawn_webrtc_answerer(
|
self.spawn_webrtc_answerer(
|
||||||
@@ -879,6 +897,23 @@ impl RendezvousMediator {
|
|||||||
.await?;
|
.await?;
|
||||||
return Ok(());
|
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);
|
log::debug!("Punch tcp hole to {:?}", peer_addr);
|
||||||
let mut socket = {
|
let mut socket = {
|
||||||
let socket = connect_tcp(&*self.host, CONNECT_TIMEOUT).await?;
|
let socket = connect_tcp(&*self.host, CONNECT_TIMEOUT).await?;
|
||||||
|
|||||||
Reference in New Issue
Block a user