TheLanc3/rustdesk
1
0
Fork 0
mirror of https://github.com/rustdesk/rustdesk.git synced 2026-09-05 23:51:04 +03:00
Code Issues Packages Projects Releases Wiki Activity

webrtc: judge the race by the resolved path, not the label; bound the ICE queue

Browse Source 
Third review round. Two of these are regressions from the previous one.

- The RelayResponse race predicate was `is_direct_transport(result.2)`,
  which answers true for the label "WebRTC" - but WebRTC is only a
  direct path when ICE nominated a non-TURN pair. A TURN-relayed WebRTC
  result therefore committed instantly and cancelled the IPv6 attempt
  racing beside it, which is the same inversion the previous fix removed
  in the other direction. (That fix was also argued from a wrong premise:
  the site does carry an IPv6 future, pushed ~50 lines earlier than the
  relay one.) Each future now resolves whether its path is direct and
  the predicate reads that bool, matching the outer race, and the
  downstream recomputation goes away.

- policy_relay still folded in Config::is_proxy(), and that is what gets
  persisted into the peer's config as force-always-relay - so one
  session through a proxy pinned the peer to relay forever and disabled
  WebRTC for it, exactly the latch the previous round fixed for
  WebSocket. Split out peer_relay: the saved option or an explicit
  request for THIS peer, and the only part written back.

- The controlled side buffered remote ICE candidates in an unbounded
  channel while the controller caps the same buffer at 64, and draining
  one costs a JSON parse plus the ICE agent's lock. Whoever can reach a
  session's route could grow it without limit inside the long-lived
  service process. Bounded, with the overflow logged through the
  existing throttle.

- That route was also removed by key alone when an answerer finished, so
  a punch retry that built a fresh answerer under the same fingerprint
  had its live sender deleted by the previous one's cleanup - after
  which it received no candidates at all. Evict only our own sender, the
  way the session cache already guards the analogous case.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ExUfAkYbq8UC9pQCiLy8TQ
This commit is contained in:
rustdesk
2026-08-08 19:14:29 +08:00
parent ea7407b73b
commit ccf9afd069
3 changed files with 81 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
src/client.rs
View File

@@ -977,8 +977,14 @@ impl Client {
let addr = AddrMangle::decode(&rr.socket_addr_v6);
if addr.port() > 0 {
if s.connect(addr).await.is_ok() {
connect_futures
.push(udp_nat_connect(s, "IPv6", CONNECT_TIMEOUT).boxed());
connect_futures.push(
async move {
let (conn, kcp, typ) =
udp_nat_connect(s, "IPv6", CONNECT_TIMEOUT).await?;
Ok((conn, kcp, typ, true))
}
.boxed(),
);
}
}
}
@@ -1046,7 +1052,12 @@ impl Client {
connect_futures.push(
async move {
let conn = fut.await?;
Ok((conn, None, if use_ws() { "WebSocket" } else { "Relay" }))
Ok((
conn,
None,
if use_ws() { "WebSocket" } else { "Relay" },
false,
))
}
.boxed(),
);
@@ -1060,7 +1071,13 @@ impl Client {
let mut raced = webrtc;
let webrtc_fut = async move {
raced.wait_connected(CONNECT_TIMEOUT).await?;
Ok((Stream::WebRTC(raced), None, "WebRTC"))
// Resolve relayed-ness here, not from the label: WebRTC is only a
// P2P path when ICE nominated a non-TURN pair, and the race has to
// know which it got. Committing a TURN pair as if it were direct
// cancels a genuine direct attempt still in flight — the same
// inversion the preference window exists to prevent, one level up.
let relayed = raced.is_relayed().await.unwrap_or(true);
Ok((Stream::WebRTC(raced), None, "WebRTC", !relayed))
}
.boxed();
if interface.is_policy_relay() {
@@ -1079,7 +1096,7 @@ impl Client {
webrtc_fut,
connect_futures,
Self::WEBRTC_PREFER_WINDOW_MS,
|result| is_direct_transport(result.2),
|result| result.3,
)
.await
}
@@ -1092,7 +1109,7 @@ impl Client {
}
// 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?;
let (mut conn, kcp, mut typ, direct) = race_result?;
feedback = rr.feedback;
log::info!("{:?} used to establish {typ} connection", start.elapsed());
let pk = match Self::secure_connection(
@@ -1146,18 +1163,8 @@ impl Client {
}
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.
// An unknown answer (no selected pair yet, or the pc closed under a
// concurrent teardown) must not be read as "direct": claiming a P2P
// path needs evidence of one.
"WebRTC" => !conn.webrtc_relayed().await.unwrap_or(true),
_ => false,
};
// `direct` came from the winning future, which resolved it while the pc was
// definitely alive — the race needed it to pick a winner at all.
// Secured and WebRTC won: disarm so the returned conn keeps the pc alive.
if typ == "WebRTC" {
if let Some(guard) = webrtc_guard.take() {
@@ -2598,13 +2605,15 @@ pub struct LoginConfigHandler {
// reconnect before the real reboot disconnect.
restart_remote_device_at: Option<Instant>,
pub force_relay: bool,
// The policy component of force_relay: the user's relay choice (option or explicit
// request) plus proxy, WITHOUT the WebSocket-transport component. ws kills classic
// TCP/UDP punching (force_relay stays set for those paths) but says nothing about
// ICE, so WebRTC decisions - offer ICE policy, prefer-P2P racing - key off this
// instead: relay-by-policy must stay Relay-only ICE, relay-by-transport may go
// direct over full ICE.
// The policy component of force_relay: peer_relay plus proxy, WITHOUT the WebSocket
// transport. ws kills classic TCP/UDP punching (force_relay stays set for those paths)
// but says nothing about ICE, so the WebRTC decisions - offer ICE policy, prefer-P2P
// racing - key off this instead: relay-by-policy must stay Relay-only ICE, while
// relay-by-transport may still go direct over full ICE.
pub policy_relay: bool,
// The peer-scoped component: this peer's saved force-always-relay option, or an explicit
// relay request for it. The only part that may be written back into the peer's config.
pub peer_relay: bool,
pub direct: Option<bool>,
pub received: bool,
switch_uuid: Option<String>,
@@ -2717,10 +2726,16 @@ impl LoginConfigHandler {
self.session_id = sid;
self.supported_encoding = Default::default();
self.clear_restarting_remote_device();
self.policy_relay =
// Three scopes, and only the first belongs in the peer's saved config: what was decided
// about THIS PEER (its saved option, or an explicit relay request such as an `/r` id or
// a retry-via-relay), versus what is true of this client right now (a proxy, WebSocket).
// Persisting either of the latter turns a transient local setup into a permanent property
// of the peer — and relay-by-policy means Relay-only ICE, so WebRTC never goes direct to
// it again.
self.peer_relay =
config::option2bool("force-always-relay", &self.get_option("force-always-relay"))
|| force_relay
|| Config::is_proxy();
|| force_relay;
self.policy_relay = self.peer_relay || Config::is_proxy();
self.force_relay = self.policy_relay || use_ws();
if let Some((real_id, server, key)) = &self.other_server {
let other_server_key = self.get_option("other-server-key");
@@ -3438,12 +3453,9 @@ impl LoginConfigHandler {
.insert("other-server-key".to_owned(), c.clone());
}
}
// policy_relay, not force_relay: this writes the user's relay CHOICE back into the peer's
// saved config, and force_relay also carries the WebSocket transport, which is a property
// of this client's current setup rather than of the peer. Persisting that turned one ws
// session into a permanent relay-by-policy peer — and since relay-by-policy means
// Relay-only ICE, WebRTC could never go direct to it again.
if self.policy_relay {
// peer_relay only — see `initialize`: neither the proxy nor the WebSocket transport is a
// fact about this peer, and writing one here makes it permanent.
if self.peer_relay {
config
.options
.insert("force-always-relay".to_owned(), "Y".to_owned());

39
src/rendezvous_mediator.rs
View File

@@ -53,8 +53,12 @@ lazy_static::lazy_static! {
static ref SOLVING_PK_MISMATCH: Mutex<String> = Default::default();
static ref LAST_MSG: Mutex<(SocketAddr, Instant)> = Mutex::new((SocketAddr::new([0; 4].into(), 0), Instant::now()));
static ref LAST_RELAY_MSG: Mutex<(SocketAddr, Instant)> = Mutex::new((SocketAddr::new([0; 4].into(), 0), Instant::now()));
static ref WEBRTC_ICE_TXS: Mutex<HashMap<String, mpsc::UnboundedSender<String>>> = Default::default();
static ref WEBRTC_ICE_TXS: Mutex<HashMap<String, mpsc::Sender<String>>> = Default::default();
}
/// Remote ICE candidates buffered per session while the answerer applies them. Mirrors the
/// controller's own cap: gathering yields host, then srflx, then relay, so a real peer sends
/// well under this, and anything past it is someone deciding how much memory this process holds.
const MAX_PENDING_REMOTE_ICE: usize = 64;
// The rendezvous ICE route is reachable without a prior punch and the peer decides how many
// candidates it sends, so these sites would let someone else set how much this machine writes to
// its log file. One line a minute each, carrying the suppressed count.
@@ -63,6 +67,8 @@ static UNKNOWN_ICE_SESSION_LOG: hbb_common::log_throttle::LogThrottle =
hbb_common::log_throttle::LogThrottle::new(ICE_LOG_INTERVAL);
static REJECTED_REMOTE_ICE_LOG: hbb_common::log_throttle::LogThrottle =
hbb_common::log_throttle::LogThrottle::new(ICE_LOG_INTERVAL);
static FULL_ICE_QUEUE_LOG: hbb_common::log_throttle::LogThrottle =
hbb_common::log_throttle::LogThrottle::new(ICE_LOG_INTERVAL);
static SHOULD_EXIT: AtomicBool = AtomicBool::new(false);
static MANUAL_RESTARTED: AtomicBool = AtomicBool::new(false);
@@ -418,7 +424,11 @@ impl RendezvousMediator {
Some(rendezvous_message::Union::IceCandidate(ice)) => {
let tx = WEBRTC_ICE_TXS.lock().await.get(&ice.session_key).cloned();
if let Some(tx) = tx {
let _ = tx.send(ice.candidate);
if tx.try_send(ice.candidate).is_err() {
if let Some(n) = FULL_ICE_QUEUE_LOG.due() {
log::debug!("dropped {} ICE candidate(s): queue full or closed", n);
}
}
} else if let Some(n) = UNKNOWN_ICE_SESSION_LOG.due() {
log::debug!(
"dropped {} ICE candidate(s) for unknown WebRTC session key, last: {}",
@@ -717,7 +727,13 @@ impl RendezvousMediator {
return Ok(answer);
};
let (remote_ice_tx, mut remote_ice_rx) = mpsc::unbounded_channel::<String>();
// Bounded, like the controller's own candidate buffer: how many candidates arrive is the
// sender's choice, while draining one costs a JSON parse and the ICE agent's lock, so an
// unbounded queue lets whoever can reach this session's route grow it without limit inside
// a long-lived service process. A full queue drops the newest candidate, which costs at
// most one path; ICE keeps whatever pairs it already has.
let (remote_ice_tx, mut remote_ice_rx) = mpsc::channel::<String>(MAX_PENDING_REMOTE_ICE);
let own_ice_tx = remote_ice_tx.clone();
WEBRTC_ICE_TXS
.lock()
.await
@@ -795,10 +811,19 @@ impl RendezvousMediator {
let session_key_for_cleanup = session_key.clone();
tokio::spawn(async move {
let result = stream.wait_connected(CONNECT_TIMEOUT).await;
WEBRTC_ICE_TXS
.lock()
.await
.remove(&session_key_for_cleanup);
// Only evict our own route. The key is the offer's DTLS fingerprint, identical across
// the controller's punch retries, so a retry that built a fresh answerer has already
// replaced this entry — removing it blindly would delete the live session's sender and
// leave it receiving no candidates at all.
{
let mut txs = WEBRTC_ICE_TXS.lock().await;
if txs
.get(&session_key_for_cleanup)
.is_some_and(|tx| tx.same_channel(&own_ice_tx))
{
txs.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

6
src/ui_session_interface.rs
View File

@@ -1296,9 +1296,11 @@ impl<T: InvokeUiSession> Session<T> {
if true == force_relay {
let mut lc = self.lc.write().unwrap();
lc.force_relay = true;
// An explicit retry-via-relay is user policy, not transport necessity: keep
// WebRTC on Relay-only ICE for this round like any force-always-relay session.
// An explicit retry-via-relay is a decision about this peer, not transport
// necessity: Relay-only ICE for this round like any force-always-relay session,
// and it is the one kind of relay that belongs in the peer's saved config.
lc.policy_relay = true;
lc.peer_relay = true;
}
self.lc.write().unwrap().peer_info = None;
self.reconnect_count.fetch_add(1, Ordering::SeqCst);