src/client.rs
Rama master · réplica de lectura del repositorio original.
#[cfg(not(any(target_os = "android", target_os = "ios")))]
use crate::clipboard::clipboard_listener::{self, ClipboardEvent};
use async_trait::async_trait;
use bytes::Bytes;
#[cfg(not(target_os = "linux"))]
use cpal::{
traits::{DeviceTrait, HostTrait, StreamTrait},
Device, Host, StreamConfig,
};
use crossbeam_queue::ArrayQueue;
use magnum_opus::{Channels::*, Decoder as AudioDecoder};
#[cfg(not(target_os = "linux"))]
use ringbuf::{ring_buffer::RbBase, Rb};
use serde::{Deserialize, Serialize};
use std::{
collections::HashMap,
ffi::c_void,
net::SocketAddr,
ops::Deref,
str::FromStr,
sync::{
mpsc::{self, RecvTimeoutError},
Arc, Mutex, RwLock,
},
};
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, decode_id_pk_dtls, dtls_fingerprint_bound, get_rs_pk,
is_keyboard_mode_supported,
kcp_stream::KcpStream,
secure_tcp, secure_tcp_required,
ui_interface::{get_builtin_option, resolve_avatar_url, use_texture_render},
ui_session_interface::{InvokeUiSession, Session},
};
#[cfg(feature = "unix-file-copy-paste")]
use crate::{clipboard::check_clipboard_files, clipboard_file::unix_file_clip};
pub use file_trait::FileManager;
#[cfg(not(feature = "flutter"))]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
use hbb_common::tokio::sync::mpsc::UnboundedSender;
use hbb_common::{
allow_err,
anyhow::{anyhow, Context},
bail,
config::{
self, use_ws, Config, LocalConfig, PeerConfig, PeerInfoSerde, Resolution,
CONNECT_TIMEOUT, READ_TIMEOUT, RELAY_PORT, RENDEZVOUS_PORT, RENDEZVOUS_SERVERS,
},
futures::future::{select_ok, BoxFuture, FutureExt},
get_version_number, log,
protobuf::{Message as _, MessageField},
rand,
rendezvous_proto::*,
sha2::{Digest, Sha256},
socket_client::{connect_tcp, connect_tcp_local, ipv4_to_ipv6, new_direct_udp_for_unverified},
sodiumoxide::{base64, crypto::sign},
timeout,
tokio::{
self,
net::UdpSocket,
sync::{
mpsc::{error::TryRecvError, unbounded_channel, UnboundedReceiver},
oneshot,
},
time::{interval, Duration, Instant},
},
webrtc::WebRTCStream,
websocket::{check_ws, is_ws_endpoint},
AddrMangle, ResultType, Stream,
};
use base::{
config::keys,
fs::JobType,
message_proto::{option_message::BoolOption, *},
};
pub use helper::*;
use scrap::{
codec::Decoder,
record::{Recorder, RecorderContext},
CodecFormat, ImageFormat, ImageRgb, ImageTexture,
};
#[cfg(not(target_os = "ios"))]
use crate::clipboard::CLIPBOARD_INTERVAL;
#[cfg(not(any(target_os = "android", target_os = "ios")))]
use crate::clipboard::{check_clipboard, ClipboardSide};
#[cfg(not(feature = "flutter"))]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
use crate::ui_session_interface::SessionPermissionConfig;
pub use super::lang::*;
mod login_method;
#[cfg(not(target_os = "linux"))]
mod audio_playback;
#[cfg(target_os = "windows")]
mod audio_playback_recovery;
#[cfg(all(test, not(target_os = "linux")))]
#[path = "client/tests/audio_state_tests.rs"]
mod audio_state_tests;
pub mod file_trait;
pub mod helper;
pub mod io_loop;
pub mod screenshot;
pub const MILLI1: Duration = Duration::from_millis(1);
pub const SEC30: Duration = Duration::from_secs(30);
// Empirical restart reconnect grace window.
const RESTART_REMOTE_DEVICE_GRACE: Duration = Duration::from_secs(5 * 60);
pub const VIDEO_QUEUE_SIZE: usize = 120;
const MAX_DECODE_FAIL_COUNTER: usize = 3;
pub const LOGIN_MSG_PASSWORD_EMPTY: &str = "Empty Password";
pub const LOGIN_MSG_PASSWORD_WRONG: &str = "Wrong Password";
pub const LOGIN_MSG_2FA_WRONG: &str = "Wrong 2FA Code";
pub const REQUIRE_2FA: &'static str = "2FA Required";
pub const LOGIN_MSG_NO_PASSWORD_ACCESS: &str = "No Password Access";
pub const LOGIN_MSG_OFFLINE: &str = "Offline";
pub const LOGIN_SCREEN_WAYLAND: &str = "Wayland login screen is not supported";
#[cfg(target_os = "linux")]
pub const SCRAP_UBUNTU_HIGHER_REQUIRED: &str = "ubuntu-21-04-required";
#[cfg(target_os = "linux")]
pub const SCRAP_OTHER_VERSION_OR_X11_REQUIRED: &str =
"wayland-requires-higher-linux-version";
#[cfg(target_os = "linux")]
pub const SCRAP_XDP_PORTAL_UNAVAILABLE: &str =
"xdp-portal-unavailable";
pub const SCRAP_X11_REQUIRED: &str = "x11 expected";
pub const SCRAP_X11_REF_URL: &str = "https://rustdesk.com/docs/en/manual/linux/#x11-required";
#[cfg(not(target_os = "linux"))]
pub const AUDIO_BUFFER_MS: usize = 3000;
#[cfg(feature = "flutter")]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
pub(crate) struct ClientClipboardContext;
#[cfg(not(feature = "flutter"))]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
pub(crate) struct ClientClipboardContext {
pub cfg: SessionPermissionConfig,
pub tx: UnboundedSender<Data>,
#[cfg(feature = "unix-file-copy-paste")]
pub is_file_supported: bool,
}
/// Client of the remote desktop.
pub struct Client;
#[cfg(not(target_os = "ios"))]
struct ClipboardState {
#[cfg(feature = "flutter")]
is_text_required: bool,
#[cfg(all(feature = "flutter", feature = "unix-file-copy-paste"))]
is_file_required: bool,
running: bool,
}
#[cfg(not(target_os = "linux"))]
lazy_static::lazy_static! {
static ref AUDIO_HOST: Host = cpal::default_host();
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
lazy_static::lazy_static! {
static ref ENIGO: Arc<Mutex<enigo::Enigo>> = Arc::new(Mutex::new(enigo::Enigo::new()));
}
#[cfg(not(target_os = "ios"))]
lazy_static::lazy_static! {
static ref CLIPBOARD_STATE: Arc<Mutex<ClipboardState>> = Arc::new(Mutex::new(ClipboardState::new()));
}
const PUBLIC_SERVER: &str = "public";
#[cfg(not(any(target_os = "android", target_os = "ios")))]
pub fn get_key_state(key: enigo::Key) -> bool {
use enigo::KeyboardControllable;
#[cfg(target_os = "macos")]
if key == enigo::Key::NumLock {
return true;
}
ENIGO.lock().unwrap().get_key_state(key)
}
/// Closes an unadopted WebRTC offerer's pc on drop. Without an answer it stays in ICE `New`
/// forever, so its state handler never fires to self-remove it from `SESSIONS`; this covers the
/// early returns and cancelled races that would leak it. `into_inner` disarms on adoption.
struct OffererGuard(Option<WebRTCStream>);
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<WebRTCStream> {
self.0.take()
}
}
impl Drop for OffererGuard {
fn drop(&mut self) {
if let Some(stream) = self.0.take() {
stream.close_detached();
}
}
}
/// Race WebRTC against the other transports, preferring P2P: `select_ok` would always pick the
/// relay, whose TCP connect beats ICE + DTLS + SCTP by an order of magnitude. An `is_p2p` result
/// wins outright; a relayed one — from either side, since `webrtc_fut` is a whole punch attempt
/// that can also end in a relay — is held for `window_ms` to give the other side a chance.
///
/// `others` must be non-empty (`select_ok` requires it).
async fn race_transports_prefer_webrtc<'a, T: 'a>(
webrtc_fut: BoxFuture<'a, ResultType<T>>,
others: Vec<BoxFuture<'a, ResultType<T>>>,
window_ms: u64,
is_p2p: impl Fn(&T) -> bool,
) -> ResultType<T> {
let mut webrtc_fut = Some(webrtc_fut);
let mut others_fut = Some(select_ok(others));
let mut held: Option<T> = None;
let mut webrtc_err: Option<hbb_common::anyhow::Error> = None;
let mut others_err: Option<hbb_common::anyhow::Error> = None;
let window = tokio::time::sleep(Duration::from_millis(window_ms));
tokio::pin!(window);
let mut window_started = false;
loop {
tokio::select! {
res = async {
match webrtc_fut.as_mut() {
Some(fut) => fut.await,
None => std::future::pending().await,
}
}, if webrtc_fut.is_some() => {
webrtc_fut = None;
match res {
// A direct connection is the outcome this race exists to protect: commit it
// outright, and a held relay conn just drops.
Ok(conn) if is_p2p(&conn) || others_fut.is_none() => return Ok(conn),
// `webrtc_fut` is a whole punch attempt, not just the WebRTC connect, so it
// can end in a relay of its own. Committing that immediately would preempt a
// direct punch still in flight on the other branch — the exact inversion this
// function exists to prevent — so hold it on the same terms as any relay.
Ok(conn) => {
if held.is_none() {
held = Some(conn);
window.as_mut().reset(
Instant::now() + Duration::from_millis(window_ms),
);
window_started = true;
}
}
Err(e) => {
// Commit a held relay only when nothing direct is still racing; otherwise
// keep it and let the survivor (or the window) decide.
if others_fut.is_none() {
if let Some(conn) = held.take() {
return Ok(conn);
}
}
match others_err.take() {
// Report one error, as `select_ok` does: the UI matches and translates
// the text whole (e.g. the offline auto-retry), and both legs often
// carry the same rendezvous refusal anyway.
Some(oe) => {
log::warn!("fallback leg also failed: {}", oe);
return Err(e);
}
None if others_fut.is_none() => bail!("WebRTC failed: {}", e),
None => webrtc_err = Some(e),
}
}
}
}
res = async {
match others_fut.as_mut() {
Some(fut) => fut.await,
None => std::future::pending().await,
}
}, if others_fut.is_some() => {
others_fut = None;
match res {
Ok((conn, unfinished)) => {
if is_p2p(&conn) {
return Ok(conn);
}
// Relayed: commit now only if nothing direct can still arrive. If a
// direct attempt is still in flight (here or in `unfinished`), hold it
// and keep racing for the preference window instead of discarding them.
if webrtc_fut.is_none() && unfinished.is_empty() {
return Ok(conn);
}
if held.is_none() {
held = Some(conn);
window.as_mut().reset(
Instant::now() + Duration::from_millis(window_ms),
);
window_started = true;
}
if !unfinished.is_empty() {
others_fut = Some(select_ok(unfinished));
}
}
Err(e) => match webrtc_err.take() {
// Nothing more can win, but a parked relay is still a valid outcome — take
// it before failing the connection.
Some(we) => match held.take() {
Some(conn) => return Ok(conn),
None => {
log::warn!("WebRTC leg also failed: {}", we);
return Err(e);
}
},
None if webrtc_fut.is_none() => match held.take() {
Some(conn) => return Ok(conn),
None => return Err(e),
},
None => others_err = Some(e),
},
}
}
_ = &mut window, if window_started => {
if let Some(conn) = held.take() {
return Ok(conn);
}
window_started = false;
}
}
}
}
// A peer decides how many ICE candidates it sends, and the rendezvous route that carries them is
// reachable without a prior punch, so these sites would otherwise let someone else set how much
// this machine writes to its log file. One line a minute each, carrying the suppressed count.
use hbb_common::log_throttle::LogThrottle;
const ICE_LOG_INTERVAL: Duration = Duration::from_secs(60);
static REJECTED_ICE_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL);
static UDP_UAT_ERR_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL);
static UNEXPECTED_ICE_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL);
static PENDING_ICE_FULL_LOG: LogThrottle = LogThrottle::new(ICE_LOG_INTERVAL);
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()
}
/// TCP punch is a user option like the other direct transports, but it is also the backstop:
/// with every direct transport switched off there would be nothing left to punch with, so it
/// runs regardless. Only the switches decide that — a transport that is enabled but fails to
/// materialize (no public v6 address, offerer setup error) leaves this alone, because the
/// relay fallback already covers a round that ends up with no usable direct transport.
fn tcp_punch_allowed() -> bool {
crate::get_tcp_punch_enabled()
|| !(crate::get_udp_punch_enabled()
|| crate::get_ipv6_punch_enabled()
|| crate::get_webrtc_enabled())
}
impl Client {
const CLIENT_CLIPBOARD_NAME: &'static str = "client-clipboard";
/// Start a new connection.
pub async fn start(
peer: &str,
key: &str,
token: &str,
conn_type: ConnType,
interface: impl Interface,
) -> ResultType<(
(
Stream,
bool,
Option<Vec<u8>>,
Option<KcpStream>,
&'static str,
),
(i32, String),
)> {
debug_assert!(peer == interface.get_id());
interface.update_direct(None);
interface.update_received(false);
match Self::_start(peer, key, token, conn_type, interface.clone()).await {
Err(err) => {
let err_str = err.to_string();
if err_str.starts_with("Failed") {
bail!(err_str + ": Please try later");
} else {
return Err(err);
}
}
Ok(x) => {
// Set x.2 to true only in the connect() function to indicate that direct_failures needs to be updated; everywhere else it should be set to false.
if x.2 {
let direct_failures = interface.get_lch().read().unwrap().direct_failures;
let direct = x.0 .1;
if !interface.is_force_relay() && (direct_failures == 0) != direct {
let n = if direct { 0 } else { 1 };
log::info!("direct_failures updated to {}", n);
interface.get_lch().write().unwrap().set_direct_failure(n);
}
}
Ok((x.0, x.1))
}
}
}
/// Start a new connection.
async fn _start(
peer: &str,
key: &str,
token: &str,
conn_type: ConnType,
interface: impl Interface,
) -> ResultType<(
(
Stream,
bool,
Option<Vec<u8>>,
Option<KcpStream>,
&'static str,
),
(i32, String),
bool,
)> {
if config::is_incoming_only() && !is_switch_sides_back(conn_type, &interface).await {
bail!("Incoming only mode");
}
// to-do: remember the port for each peer, so that we can retry easier
if hbb_common::is_ip_str(peer) {
return Ok((
(
connect_tcp_local(check_port(peer, RELAY_PORT + 1), None, CONNECT_TIMEOUT)
.await?,
true,
None,
None,
"TCP",
),
(0, "".to_owned()),
false,
));
}
// Allow connect to {domain}:{port}
if hbb_common::is_domain_port_str(peer) {
return Ok((
(
connect_tcp_local(peer, None, CONNECT_TIMEOUT).await?,
true,
None,
None,
"TCP",
),
(0, "".to_owned()),
false,
));
}
let other_server = interface.get_lch().read().unwrap().other_server.clone();
let (peer, other_server, key, token) = if let Some((a, b, c)) = other_server.as_ref() {
(a.as_ref(), b.as_ref(), c.as_ref(), "")
} else {
(peer, "", key, token)
};
let (rendezvous_server, servers, contained) = if other_server.is_empty() {
crate::get_rendezvous_server(1_000).await
} else {
if other_server == PUBLIC_SERVER {
(
check_port(RENDEZVOUS_SERVERS[0], RENDEZVOUS_PORT),
RENDEZVOUS_SERVERS[1..]
.iter()
.map(|x| x.to_string())
.collect(),
true,
)
} else {
(check_port(other_server, RENDEZVOUS_PORT), Vec::new(), true)
}
};
// Same relay gate as the v6 socket below: under any forced relay the v6 punch cannot
// be used, so probing v6 reachability is wasted work on every such connection.
if crate::get_ipv6_punch_enabled() && !interface.is_force_relay() {
crate::test_ipv6().await;
}
let (stop_udp_tx, stop_udp_rx) = oneshot::channel::<()>();
let udp =
// no need to care about multiple rendezvous servers case, since it is acutally not used any more.
// Shared state for UDP NAT test result
if crate::get_udp_punch_enabled() && !interface.is_force_relay() {
if let Ok((socket, addr)) = new_direct_udp_for_unverified(&rendezvous_server).await {
let udp_port = Arc::new(Mutex::new(0));
let up_cloned = udp_port.clone();
let socket_cloned = socket.clone();
let func = async move {
allow_err!(test_udp_uat(socket_cloned, addr, up_cloned, stop_udp_rx).await);
};
tokio::spawn(func);
(Some(socket), Some(udp_port))
} else {
(None, None)
}
} else {
(None, None)
};
// 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
};
// 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) {
// ICE policy follows relay-by-POLICY, not force_relay: under WebSocket the
// latter is set for the classic paths, but ICE opens its own sockets and may
// still go direct - that is the only P2P path ws deployments have.
match WebRTCStream::new("", interface.is_policy_relay(), CONNECT_TIMEOUT).await {
Ok(stream) => Some(stream),
Err(err) => {
log::warn!("webrtc offerer setup failed: {}", err);
None
}
}
} else {
None
};
let has_webrtc_offerer = webrtc_offerer.is_some();
let fut = Self::_start_inner(
peer.to_owned(),
key.to_owned(),
token.to_owned(),
conn_type,
interface.clone(),
udp.clone(),
Some(stop_udp_tx),
ipv6,
webrtc_offerer,
rendezvous_server.clone(),
servers.clone(),
contained,
);
// The fallback request exists only to carry a TCP punch, so it is pointless once TCP
// punch is off — it would reach `connect()` with nothing to try and just open a second
// relay.
if interface.is_force_relay()
|| (udp.0.is_none() && !has_webrtc_offerer)
|| !tcp_punch_allowed()
{
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 retains its own socket for WebRTC signaling.
let fallback_fut = Self::_start_inner(
peer.to_owned(),
key.to_owned(),
token.to_owned(),
conn_type,
interface,
(None, None),
None,
None,
None,
rendezvous_server,
servers,
contained,
)
.boxed();
if has_webrtc_offerer {
return race_transports_prefer_webrtc(
preferred_fut,
vec![fallback_fut],
Self::relay_fallback_delay_ms(),
|result| result.0 .1,
)
.await;
}
let connect_futures = vec![preferred_fut, fallback_fut];
match select_ok(connect_futures).await {
Ok(conn) => Ok((conn.0 .0, conn.0 .1, conn.0 .2)),
Err(e) => Err(e),
}
}
fn is_expected_webrtc_ice_candidate(ice: &IceCandidate, session_key: &str) -> bool {
!session_key.is_empty() && ice.session_key == session_key && !ice.candidate.is_empty()
}
/// Whether to build a WebRTC offerer for this connection.
///
/// A SOCKS proxy rules it out: ICE binds its own UDP sockets and speaks STUN directly, past
/// the proxy and with the real IP. Relay-by-policy without TURN rules it out too, since
/// Relay-only ICE can then gather nothing. WebSocket does not: it tunnels only the signaling
/// and relay legs, so the offer keeps full ICE and direct is exactly what it is there for.
fn should_create_webrtc_offerer(interface: &impl Interface) -> bool {
if !crate::get_webrtc_enabled() {
return false;
}
if Config::is_proxy() {
return false;
}
if interface.is_policy_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. On overflow the oldest is
/// evicted: gathering order is host, then srflx, then relay, so the newest arrivals are the
/// ones that traverse NAT.
const MAX_PENDING_WEBRTC_ICE: usize = 64;
/// Default relay fallback delay: how long an already-established relay result is held back
/// while a WebRTC attempt is still in flight, 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. The same role RFC 8305 calls a connection attempt delay.
const RELAY_FALLBACK_DELAY_MS: u64 = 2500;
/// The delay as the user configured it, falling back to `RELAY_FALLBACK_DELAY_MS`. The
/// settings field holds seconds, which is what a user reasons about; everything here is
/// milliseconds. Unparseable, zero or negative all mean "unset", so clearing the field
/// restores the default instead of collapsing the delay and handing every race to the
/// relay.
fn relay_fallback_delay_ms() -> u64 {
match LocalConfig::get_option(keys::OPTION_RELAY_FALLBACK_DELAY)
.trim()
.parse::<f64>()
{
Ok(secs) if secs.is_finite() && secs > 0.0 => (secs * 1000.0).round() as u64,
_ => Self::RELAY_FALLBACK_DELAY_MS,
}
}
/// UDP-NAT-test wait when the TCP clock is implausible (see TCP_RTT_PLAUSIBLE_MIN). The
/// normal bound is `rtt / 2`: the test has been running since before the TCP connect, so on
/// a network where TCP RTT ~ UDP RTT its response has already landed. A transparent TCP
/// proxy — a TUN-mode VPN on the host, or a redirect-mode proxy on the LAN gateway (soft
/// router), which fakes the handshake for every device behind it — breaks that by answering
/// in ~3ms while the real UDP round trip is hundreds of ms: the window collapsed to ~1.5ms,
/// udp_port stayed 0, and UDP punch never ran on such networks. The wait still exits the instant the port
/// arrives, so a genuinely nearby server (LAN hbbs) pays nothing; only UDP-dead networks
/// wait out the full grace, and only on this round — the pure-TCP fallback round never
/// waits. Sized as a ceiling on real-world rendezvous RTTs plus one 20ms retransmit
/// (intercontinental ~300ms); paths slower than that lose UDP punch under a proxy, which
/// is today's behavior, not a regression.
const UDP_NAT_TEST_GRACE: Duration = Duration::from_millis(400);
/// Below this, the measured TCP connect time is not a believable WAN round trip — it was
/// answered inside the LAN (host TUN proxy, gateway transparent proxy, or a genuinely local
/// server) — and must not be used to size the UDP window.
/// Generous on purpose: over-triggering costs nothing (the wait exits on arrival, and a
/// UDP-dead round loses to the racing fallback anyway), while a tight bound would let a
/// busy proxy's occasional ~80ms handshake slip through and silently drop UDP punch.
const TCP_RTT_PLAUSIBLE_MIN: Duration = Duration::from_millis(100);
/// 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);
/// Bridge local ICE candidates to the peer over the punch socket, and feed the peer's back
/// into the pc.
///
/// Never reconnect this socket: its address *is* the return route (the server mangles it into
/// `PunchHole.socket_addr` and resolves the echo through `tcp_punch`), so a new address is one
/// nothing points at. Once it dies, both directions are dead — abandon WebRTC, do not retry.
fn spawn_webrtc_ice_bridge(
mut socket: Stream,
mut local_ice_rx: Option<UnboundedReceiver<String>>,
webrtc: WebRTCStream,
peer: String,
session_key: String,
) -> 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,
Err(tokio::sync::oneshot::error::TryRecvError::Empty) => {}
}
if let Some(rx) = local_ice_rx.as_mut() {
loop {
match rx.try_recv() {
Ok(candidate) => {
let mut msg = RendezvousMessage::new();
msg.set_ice_candidate(IceCandidate {
id: peer.clone(),
session_key: session_key.clone(),
candidate,
..Default::default()
});
// 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) => {
local_ice_rx = None;
break;
}
}
}
}
// 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
{
if let Some(n) = REJECTED_ICE_LOG.due() {
log::warn!(
"failed to add {} WebRTC ICE candidate(s), last: {}",
n,
err
);
}
}
}
}
}
}
Ok(Some(Err(err))) => {
log::debug!("WebRTC ICE bridge socket read ended: {}", err);
break;
}
}
}
});
stop_tx
}
async fn _start_inner(
peer: String,
key: String,
token: String,
conn_type: ConnType,
interface: impl Interface,
mut udp: (Option<Arc<UdpSocket>>, Option<Arc<Mutex<u16>>>),
stop_udp_tx: Option<oneshot::Sender<()>>,
mut ipv6: Option<(Arc<UdpSocket>, bytes::Bytes)>,
webrtc_offerer: Option<WebRTCStream>,
mut rendezvous_server: String,
servers: Vec<String>,
contained: bool,
) -> ResultType<(
(
Stream,
bool,
Option<Vec<u8>>,
Option<KcpStream>,
&'static str,
),
(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));
let rtt = start.elapsed();
log::debug!("TCP connection establishment time used: {:?}", rtt);
if socket.is_err() && !servers.is_empty() {
log::info!("try the other servers: {:?}", servers);
for server in servers {
let server = check_port(server, RENDEZVOUS_PORT);
socket = connect_tcp(&*server, CONNECT_TIMEOUT).await;
if socket.is_ok() {
rendezvous_server = server;
break;
}
}
crate::refresh_rendezvous_server();
} else if !contained {
crate::refresh_rendezvous_server();
}
log::info!("rendezvous server: {}", rendezvous_server);
let mut socket = socket?;
let mut my_addr = socket.local_addr();
let mut signed_id_pk = Vec::new();
let mut relay_server = "".to_owned();
let mut peer_addr = Config::get_any_listen_addr(true);
let mut peer_nat_type = NatType::UNKNOWN_NAT;
let my_nat_type = crate::get_nat_type(100).await;
let mut is_local = false;
let mut feedback = 0;
use hbb_common::protobuf::Enum;
let nat_type = if interface.is_force_relay() {
NatType::SYMMETRIC
} else {
NatType::from_i32(my_nat_type).unwrap_or(NatType::UNKNOWN_NAT)
};
let switch_code = interface.get_switch_code();
let legacy_secure = !key.is_empty() && (!token.is_empty() || !switch_code.is_empty());
let carries_offer = webrtc_offerer.as_ref().and_then(|g| g.stream()).is_some();
// Counted from before the key exchange, so the exchange spends the UDP NAT test's own
// wait rather than replacing it: the test runs beside both.
let udp_nat_wait_from = Instant::now();
let mut exchanged = false;
if carries_offer {
// An offer puts both sides' ICE candidates, every interface address of both
// machines, on this socket, so it goes out only once the server's key exchange has
// encrypted it. When the server does not complete one, an hbbs from before the
// exchange, the offer is dropped and this becomes a punch without WebRTC, on a fresh
// socket since the failed exchange may have consumed a message on this one. Degrade
// to no WebRTC, never to WebRTC signalling in the clear.
match secure_tcp_required(&mut socket, &key).await {
Ok(()) => exchanged = true,
Err(err) => {
log::warn!(
"WebRTC signalling to {} cannot be encrypted, punching without WebRTC: {}",
rendezvous_server,
err
);
webrtc_offerer = None;
socket = connect_tcp(&*rendezvous_server, CONNECT_TIMEOUT).await?;
my_addr = socket.local_addr();
}
}
}
if !exchanged && legacy_secure {
secure_tcp(&mut socket, &key)
.await
.map_err(|e| anyhow!("Failed to secure tcp: {}", e))?;
}
// A token or switch code has always taken this socket straight to the punch without
// waiting for the UDP NAT test. The WebRTC exchange does not replace that wait, it only
// spends part of the same budget, so what is left of it is waited out here and a result
// that has already arrived is taken at once.
if let Some(udp) = udp.1.as_ref().filter(|_| !legacy_secure) {
let tm = udp_nat_wait_from;
// rtt is the TCP connect time. When it is too short to be a real WAN round trip it
// says nothing about the UDP path (a TUN VPN or the LAN gateway answered the
// handshake, not the server), so fall back to the flat grace; otherwise trust it.
let udp_nat_wait = if rtt < Self::TCP_RTT_PLAUSIBLE_MIN {
Self::UDP_NAT_TEST_GRACE
} else {
rtt / 2
};
loop {
let port = *udp.lock().unwrap();
if port > 0 {
break;
}
if tm.elapsed() > udp_nat_wait {
break;
}
hbb_common::sleep(0.001).await;
}
}
// Stop UDP NAT test task if still running
stop_udp_tx.map(|tx| tx.send(()));
let mut msg_out = RendezvousMessage::new();
let mut ipv6 = ipv6
.take()
.map(|(socket, addr)| (Some(socket), Some(addr)))
.unwrap_or((None, None));
let udp_nat_port = udp.1.map(|x| *x.lock().unwrap()).unwrap_or(0);
let webrtc_sdp_offer = webrtc_offerer
.as_ref()
.and_then(|g| g.stream())
.map(|stream| stream.local_endpoint().to_owned())
.unwrap_or_default();
let allow_tcp_punch = tcp_punch_allowed() && request_allows_tcp_punch(&webrtc_sdp_offer);
// Every direct transport this round carries, not one of them: a round can carry several
// at once (a NAT port and an offer and a v6 address), and since the TCP punch became a
// switch it can carry none — a single name had to misreport both. `relay` is not a punch,
// it is what a round with nothing to punch with can still end as.
let mut transports = Vec::new();
if udp_nat_port > 0 {
transports.push("UDP");
}
if allow_tcp_punch {
transports.push("TCP");
}
if ipv6.1.is_some() {
transports.push("IPv6");
}
if !webrtc_sdp_offer.is_empty() {
transports.push("WebRTC");
}
let punch_type = if transports.is_empty() {
"Relay".to_owned()
} else {
transports.join("+")
};
msg_out.set_punch_hole_request(PunchHoleRequest {
id: peer.to_owned(),
token: token.to_owned(),
nat_type: nat_type.into(),
licence_key: key.to_owned(),
conn_type: conn_type.into(),
version: crate::VERSION.to_owned(),
udp_port: udp_nat_port as _,
force_relay: interface.is_force_relay(),
socket_addr_v6: ipv6.1.unwrap_or_default(),
switch_code,
// The offer's envelope itself declares its ICE policy (`ice_policy: "all"` under
// pure ws), telling the controlled side its answer may gather every candidate
// type despite force_relay instead of requiring TURN.
webrtc_sdp_offer,
..Default::default()
});
let webrtc_session_key = webrtc_offerer
.as_ref()
.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::<String>::new();
'punch_attempts: for i in 1..=3 {
log::info!(
"#{} {} punch attempt with {}, id: {}",
i,
punch_type,
my_addr,
peer
);
socket.send(&msg_out).await?;
// below timeout should not bigger than hbbs's connection timeout.
let attempt_deadline = Instant::now() + Duration::from_millis((i * 3000) as u64);
loop {
let remaining = attempt_deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
break;
}
let timeout_ms = remaining
.as_millis()
.clamp(1, u64::MAX as u128) as u64;
let Some(msg_in) =
crate::get_next_nonkeyexchange_msg(&mut socket, Some(timeout_ms)).await
else {
break;
};
match msg_in.union {
Some(rendezvous_message::Union::PunchHoleResponse(ph)) => {
if ph.socket_addr.is_empty() {
if !ph.other_failure.is_empty() {
bail!(ph.other_failure);
}
match ph.failure.enum_value() {
Ok(punch_hole_response::Failure::ID_NOT_EXIST) => {
bail!("ID does not exist");
}
Ok(punch_hole_response::Failure::OFFLINE) => {
bail!("Remote desktop is offline");
}
Ok(punch_hole_response::Failure::LICENSE_MISMATCH) => {
bail!("Key mismatch");
}
Ok(punch_hole_response::Failure::LICENSE_OVERUSE) => {
bail!("Key overuse");
}
_ => bail!("other punch hole failure"),
}
} else {
peer_nat_type = ph.nat_type();
is_local = ph.is_local();
signed_id_pk = ph.pk.into();
relay_server = ph.relay_server;
peer_addr = AddrMangle::decode(&ph.socket_addr);
feedback = ph.feedback;
webrtc_sdp_answer = ph.webrtc_sdp_answer;
let s = udp.0.take();
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);
}
}
let s = ipv6.0.take();
if !ph.socket_addr_v6.is_empty() && s.is_some() {
let addr = AddrMangle::decode(&ph.socket_addr_v6);
if addr.port() > 0 {
if let Some(s) = s {
allow_err!(s.connect(addr).await);
ipv6.0 = Some(s);
}
}
}
log::info!("{} Hole Punched {} = {}", punch_type, peer, peer_addr);
break 'punch_attempts;
}
}
Some(rendezvous_message::Union::RelayResponse(rr)) => {
log::info!(
"relay requested from peer, time used: {:?}, relay_server: {}",
start.elapsed(),
rr.relay_server
);
start = Instant::now();
let mut connect_futures = Vec::new();
if let Some(s) = ipv6.0 {
let addr = AddrMangle::decode(&rr.socket_addr_v6);
if addr.port() > 0 {
if s.connect(addr).await.is_ok() {
connect_futures.push(
async move {
let (conn, kcp, typ) =
udp_nat_connect(s, "IPv6", CONNECT_TIMEOUT).await?;
Ok((conn, kcp, typ, true))
}
.boxed(),
);
}
}
}
signed_id_pk = rr.pk().into();
let mut webrtc_bridge_stop = None;
let mut webrtc_for_connect = None;
if !rr.webrtc_sdp_answer.is_empty() {
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(
socket,
local_ice_rx,
webrtc.clone(),
peer.clone(),
session_key,
));
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,
rr.relay_server,
&key,
conn_type,
my_addr.is_ipv4(),
);
connect_futures.push(
async move {
let conn = fut.await?;
Ok((
conn,
None,
if use_ws() { "WebSocket" } else { "Relay" },
false,
))
}
.boxed(),
);
// 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?;
// 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() {
// Relay-only WebRTC can use only TURN, so it has no P2P advantage
// over the RustDesk relay. Take the first successful relay instead
// of delaying an already-ready result for the preference window.
// Policy, not force_relay: under ws the offer is full ICE and a
// direct path is exactly what the preference window exists for.
connect_futures.push(webrtc_fut);
select_ok(connect_futures).await.map(|r| r.0)
} else {
// 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::relay_fallback_delay_ms(),
|result| result.3,
)
.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(());
}
// 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, mut direct) = race_result?;
feedback = rr.feedback;
log::info!("{:?} used to establish {typ} connection", start.elapsed());
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,
&interface.get_switch_code(),
)
.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" };
// The transport is now a relay: the WebRTC win it replaced must
// not carry its direct flag into the return, or the relay is
// reported P2P and the outer race treats it as one.
direct = false;
pk
}
Err(e) => return Err(e),
};
// `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() {
let _ = guard.into_inner();
}
}
return Ok((
(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) {
// Evict the oldest, not the newest. Candidates arrive in gathering
// order — host first, then srflx, then relay — so dropping arrivals
// would discard exactly the ones that traverse NAT and keep the
// host ones that only work on a shared LAN.
if pending_webrtc_ice.len() >= Self::MAX_PENDING_WEBRTC_ICE {
if let Some(n) = PENDING_ICE_FULL_LOG.due() {
log::warn!(
"WebRTC ICE pending buffer full ({}), evicted {} oldest",
Self::MAX_PENDING_WEBRTC_ICE,
n
);
}
pending_webrtc_ice.remove(0);
}
pending_webrtc_ice.push(ice.candidate);
} else if let Some(n) = UNEXPECTED_ICE_LOG.due() {
log::debug!(
"dropped {} ICE candidate(s) for unexpected WebRTC session key, last: {}",
n,
ice.session_key,
);
}
}
_ => {
log::error!("Unexpected protobuf msg received: {:?}", msg_in);
}
}
}
}
let mut webrtc_bridge_stop = None;
let mut webrtc_for_connect = None;
if !webrtc_sdp_answer.is_empty() {
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(
socket,
local_ice_rx,
webrtc.clone(),
peer.clone(),
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);
}
} 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() {
webrtc.close_detached();
}
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;
log::info!(
"{} ms used to {} punch hole, relay_server: {}, {}",
time_used,
punch_type,
relay_server,
if is_local {
"is_local: true".to_owned()
} else {
format!("nat_type: {:?}", peer_nat_type)
}
);
Ok((
Self::connect(
my_addr,
peer_addr,
&peer,
signed_id_pk,
&relay_server,
&rendezvous_server,
time_used,
peer_nat_type,
my_nat_type,
is_local,
&key,
&token,
conn_type,
interface,
udp.0,
ipv6.0,
webrtc_for_connect,
webrtc_bridge_stop,
allow_tcp_punch,
&punch_type,
)
.await?,
(feedback, rendezvous_server),
true,
))
}
/// Connect to the peer.
async fn connect(
local_addr: SocketAddr,
peer: SocketAddr,
peer_id: &str,
signed_id_pk: Vec<u8>,
relay_server: &str,
rendezvous_server: &str,
punch_time_used: u64,
peer_nat_type: NatType,
my_nat_type: i32,
is_local: bool,
key: &str,
token: &str,
conn_type: ConnType,
interface: impl Interface,
udp_socket_nat: Option<Arc<UdpSocket>>,
udp_socket_v6: Option<Arc<UdpSocket>>,
webrtc_offerer: Option<WebRTCStream>,
webrtc_bridge_stop: Option<oneshot::Sender<()>>,
allow_tcp_punch: bool,
punch_type: &str,
) -> ResultType<(
Stream,
bool,
Option<Vec<u8>>,
Option<KcpStream>,
&'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;
if is_local || peer_nat_type == NatType::SYMMETRIC {
connect_timeout = MIN;
} else {
if relay_server.is_empty() {
connect_timeout = CONNECT_TIMEOUT;
} else {
if peer_nat_type == NatType::ASYMMETRIC {
let mut my_nat_type = my_nat_type;
if my_nat_type == NatType::UNKNOWN_NAT as i32 {
my_nat_type = crate::get_nat_type(100).await;
}
if my_nat_type == NatType::ASYMMETRIC as i32 {
connect_timeout = CONNECT_TIMEOUT;
if direct_failures > 0 {
connect_timeout = punch_time_used * 6;
}
} else if my_nat_type == NatType::SYMMETRIC as i32 {
connect_timeout = MIN;
}
}
if connect_timeout == 0 {
let n = if direct_failures > 0 { 3 } else { 6 };
connect_timeout = punch_time_used * (n as u64);
}
}
if connect_timeout < MIN {
connect_timeout = MIN;
}
}
log::info!("peer address: {}, timeout: {}", peer, connect_timeout);
let start = std::time::Instant::now();
// Each attempt carries whether its path is direct (4th field). TCP/UDP/IPv6 punch are
// always direct; WebRTC is direct only when ICE nominated a non-TURN pair.
let mut direct_futures = Vec::new();
if allow_tcp_punch {
let fut = connect_tcp_local(peer, Some(local_addr), connect_timeout);
direct_futures.push(
async move {
let conn = fut.await?;
Ok((conn, None, "TCP", true))
}
.boxed(),
);
}
if let Some(udp_socket_nat) = udp_socket_nat {
direct_futures.push(
async move {
let (conn, kcp, typ) =
udp_nat_connect(udp_socket_nat, "UDP", connect_timeout).await?;
Ok((conn, kcp, typ, true))
}
.boxed(),
);
}
if let Some(udp_socket_v6) = udp_socket_v6 {
direct_futures.push(
async move {
let (conn, kcp, typ) =
udp_nat_connect(udp_socket_v6, "IPv6", connect_timeout).await?;
Ok((conn, kcp, typ, true))
}
.boxed(),
);
}
// 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).
let webrtc_fut = webrtc_guard
.as_ref()
.and_then(|g| g.stream())
.map(|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 immediately, and the
// relay fallback only waits the extra time when direct attempts all failed.
let webrtc_timeout = connect_timeout.max(Self::relay_fallback_delay_ms());
async move {
raced.wait_connected(webrtc_timeout).await?;
// Resolve the pair here: a TURN win is relayed, not direct, and must be held
// behind still-racing direct attempts rather than committed as P2P.
let relayed = raced.is_relayed().await.unwrap_or(true);
Ok((Stream::WebRTC(raced), None, "WebRTC", !relayed))
}
.boxed()
});
// Prefer P2P: a direct result wins outright, a relayed WebRTC (TURN) is held for the
// window so a direct punch can still land. Falls back to plain select_ok when only one
// kind is present.
let direct_result = match (webrtc_fut, direct_futures.is_empty()) {
(Some(webrtc_fut), false) => {
race_transports_prefer_webrtc(
webrtc_fut,
direct_futures,
Self::relay_fallback_delay_ms(),
|r| r.3,
)
.await
}
(Some(webrtc_fut), true) => webrtc_fut.await,
(None, false) => select_ok(direct_futures).await.map(|c| c.0),
(None, true) => Err(anyhow!("No direct transport available")),
};
let (mut conn, kcp, mut typ, mut direct) = match direct_result {
Ok((conn, kcp, typ, direct)) => (Ok(conn), kcp, typ, direct),
Err(e) => (Err(e), None, "", false),
};
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.
// Keep a WebRTC win instead of replacing it with the RustDesk relay: under relay-by-
// policy the pc was built with Relay-only ICE (TURN configured), which already honors
// the relay requirement, and under ws-forced relay a direct full-ICE connection is the
// preferred outcome, not a violation.
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(
peer_id,
relay_server.to_owned(),
rendezvous_server,
!signed_id_pk.is_empty(),
key,
token,
conn_type,
&switch_code,
)
.await;
if let Err(e) = conn {
// this direct is mainly used by on_establish_connection_error, so we update it here before bail
interface.update_direct(Some(false));
bail!("Failed to connect via relay server: {}", e);
}
typ = "Relay";
direct = false;
} else {
bail!("Failed to make direct connection to remote desktop");
}
}
let mut conn = conn?;
log::info!(
"{:?} used to establish {typ} connection with {} punch",
start.elapsed(),
punch_type
);
let res = Self::secure_connection(peer_id, signed_id_pk.clone(), key, &mut conn).await;
let pk: Option<Vec<u8>> = 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,
&interface.get_switch_code(),
)
.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. An unknown answer
// counts as relayed — claiming a P2P path needs evidence of one.
if conn.webrtc_relayed().await.unwrap_or(true) {
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))
}
/// Establish secure connection with the server.
async fn secure_connection(
peer_id: &str,
signed_id_pk: Vec<u8>,
key: &str,
conn: &mut Stream,
) -> ResultType<Option<Vec<u8>>> {
let rs_pk = get_rs_pk(if key.is_empty() {
config::RS_PUB_KEY
} 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() {
if let Some(rs_pk) = rs_pk {
if let Ok((id, pk)) = decode_id_pk(&signed_id_pk, &rs_pk) {
if id == peer_id {
sign_pk = Some(sign::PublicKey(pk));
option_pk = Some(pk.to_vec());
}
}
}
if sign_pk.is_none() {
log::error!("Handshake failed: invalid public key from rendezvous server");
}
}
let sign_pk = match sign_pk {
Some(v) => v,
None => {
// No trusted peer identity (key-less deployment, or a blob that does not verify
// under our root), so no fingerprint binding is possible. Fall through like TCP's
// non-secure path with is_secured() false: bailing would only move the session to
// a relay that fails the same check and then runs in plaintext.
// 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);
}
};
match timeout(READ_TIMEOUT, conn.next()).await? {
Some(res) => {
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, signed_fp, kx_version)) =
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 bound the certificate to the remote SDP, so
// requiring the peer to have SIGNED that fingerprint defeats a
// rendezvous/relay MITM that swaps SDPs. Fail closed.
if is_webrtc {
let actual_fp = conn.dtls_fingerprint(false).await.ok_or_else(
|| anyhow!("WebRTC DTLS fingerprint unavailable"),
)?;
if !dtls_fingerprint_bound(&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);
// A WebRTC stream is encrypted by DTLS and takes no stream
// key of its own, split or not, so the pick says what runs: 0.
let picked = if is_webrtc {
0
} else {
hbb_common::tcp::kx_version_for(kx_version)
};
let mut msg_out = Message::new();
msg_out.set_public_key(PublicKey {
asymmetric_value: asymmetric_value.clone(),
symmetric_value,
kx_version: picked,
..Default::default()
});
timeout(CONNECT_TIMEOUT, conn.send(&msg_out)).await??;
conn.set_negotiated_key(
key,
true,
&hbb_common::tcp::KxTranscript {
initiator_pk: &asymmetric_value,
responder_pk: &their_pk_b,
advertised: kx_version,
picked,
},
)?;
} 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();
msg_out.set_public_key(PublicKey::new());
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?;
}
}
None => {
bail!("Reset by the peer");
}
}
Ok(option_pk)
}
/// Request a relay connection to the server.
async fn request_relay(
peer: &str,
relay_server: String,
rendezvous_server: &str,
secure: bool,
key: &str,
token: &str,
conn_type: ConnType,
switch_code: &str,
) -> ResultType<Stream> {
let mut succeed = false;
let mut uuid = "".to_owned();
let mut ipv4 = true;
for i in 1..=3 {
// use different socket due to current hbbs implementation requiring different nat address for each attempt
let mut socket = connect_tcp(rendezvous_server, CONNECT_TIMEOUT)
.await
.with_context(|| "Failed to connect to rendezvous server")?;
if !key.is_empty() && (!token.is_empty() || !switch_code.is_empty()) {
secure_tcp(&mut socket, key).await?;
}
ipv4 = socket.local_addr().is_ipv4();
let mut msg_out = RendezvousMessage::new();
uuid = Uuid::new_v4().to_string();
log::info!(
"#{} request relay attempt, id: {}, uuid: {}, relay_server: {}, secure: {}",
i,
peer,
uuid,
relay_server,
secure,
);
msg_out.set_request_relay(RequestRelay {
id: peer.to_owned(),
token: token.to_owned(),
uuid: uuid.clone(),
relay_server: relay_server.clone(),
secure,
switch_code: switch_code.to_owned(),
..Default::default()
});
socket.send(&msg_out).await?;
if let Some(msg_in) =
crate::get_next_nonkeyexchange_msg(&mut socket, Some(CONNECT_TIMEOUT)).await
{
if let Some(rendezvous_message::Union::RelayResponse(rs)) = msg_in.union {
if !rs.refuse_reason.is_empty() {
bail!(rs.refuse_reason);
}
succeed = true;
break;
}
}
}
if !succeed {
bail!("Timeout");
}
Self::create_relay(peer, uuid, relay_server, key, conn_type, ipv4).await
}
/// Create a relay connection to the server.
async fn create_relay(
peer: &str,
uuid: String,
relay_server: String,
key: &str,
conn_type: ConnType,
ipv4: bool,
) -> ResultType<Stream> {
crate::common::check_relay_uuid(&uuid)?;
// The proxy's address family does not constrain the relay target.
let relay_server = check_port(relay_server, RELAY_PORT);
let use_ws_proxy = Config::is_proxy() && is_ws_endpoint(&check_ws(&relay_server));
let mut conn = connect_tcp(
ipv4_to_ipv6(relay_server, ipv4 || use_ws_proxy),
CONNECT_TIMEOUT,
)
.await
.with_context(|| "Failed to connect to relay server")?;
let mut msg_out = RendezvousMessage::new();
msg_out.set_request_relay(RequestRelay {
licence_key: key.to_owned(),
id: peer.to_owned(),
uuid,
conn_type: conn_type.into(),
..Default::default()
});
conn.send(&msg_out).await?;
Ok(conn)
}
#[inline]
#[cfg(feature = "flutter")]
#[cfg(not(target_os = "ios"))]
pub fn set_is_text_clipboard_required(b: bool) {
CLIPBOARD_STATE.lock().unwrap().is_text_required = b;
}
#[inline]
#[cfg(all(feature = "flutter", feature = "unix-file-copy-paste"))]
pub fn set_is_file_clipboard_required(b: bool) {
CLIPBOARD_STATE.lock().unwrap().is_file_required = b;
}
#[cfg(not(target_os = "ios"))]
fn try_stop_clipboard() {
// Disconnected Flutter sessions may keep UI handlers alive, so only connected sessions
// should block clipboard cleanup.
#[cfg(feature = "flutter")]
if crate::flutter::sessions::has_connected_sessions_running(ConnType::DEFAULT_CONN) {
return;
}
#[cfg(not(target_os = "android"))]
clipboard_listener::unsubscribe(Self::CLIENT_CLIPBOARD_NAME);
CLIPBOARD_STATE.lock().unwrap().running = false;
#[cfg(all(feature = "unix-file-copy-paste", target_os = "linux"))]
if let Err(e) = crate::clipboard::try_empty_clipboard_files_sync(
crate::clipboard::ClipboardSide::Client,
0,
) {
log::error!("Failed to empty client clipboard files: {}", e);
}
#[cfg(all(feature = "unix-file-copy-paste", target_os = "linux"))]
clipboard::platform::unix::fuse::uninit_fuse_context(true);
}
// `try_start_clipboard` is called by all session when connection is established. (When handling peer info).
// This function only create one thread with a loop, the loop is shared by all sessions.
// After all sessions are end, the loop exists.
//
// If clipboard update is detected, the text will be sent to all sessions by `send_clipboard_msg`.
#[cfg(not(any(target_os = "android", target_os = "ios")))]
fn try_start_clipboard(
_client_clip_ctx: Option<ClientClipboardContext>,
) -> Option<UnboundedReceiver<()>> {
let mut clipboard_lock = CLIPBOARD_STATE.lock().unwrap();
if clipboard_lock.running {
return None;
}
let (tx_cb_result, rx_cb_result) = mpsc::channel();
if let Err(e) =
clipboard_listener::subscribe(Self::CLIENT_CLIPBOARD_NAME.to_owned(), tx_cb_result)
{
log::error!("Failed to subscribe clipboard listener: {}", e);
return None;
}
clipboard_lock.running = true;
let (tx_started, rx_started) = unbounded_channel();
log::info!("Start client clipboard loop");
std::thread::spawn(move || {
let mut handler = ClientClipboardHandler {
ctx: None,
#[cfg(not(feature = "flutter"))]
client_clip_ctx: _client_clip_ctx,
};
tx_started.send(()).ok();
loop {
if !CLIPBOARD_STATE.lock().unwrap().running {
break;
}
match rx_cb_result.recv_timeout(Duration::from_millis(CLIPBOARD_INTERVAL)) {
Ok(ClipboardEvent::Changed) => {
handler.check_clipboard(true);
}
#[cfg(all(target_os = "linux", feature = "unix-file-copy-paste"))]
Ok(ClipboardEvent::InitialSelection) => {
// Preserve file startup sync without bypassing text initial-sync settings.
handler.check_clipboard(false);
}
Ok(ClipboardEvent::Stop) => {
log::debug!("Clipboard listener stopped");
break;
}
Ok(ClipboardEvent::StopWithError(err)) => {
log::error!("Clipboard listener stopped with error: {}", err);
break;
}
Err(RecvTimeoutError::Timeout) => {}
Err(RecvTimeoutError::Disconnected) => {
log::error!("Clipboard listener disconnected");
break;
}
}
}
log::info!("Stop client clipboard loop");
CLIPBOARD_STATE.lock().unwrap().running = false;
});
Some(rx_started)
}
#[cfg(target_os = "android")]
fn try_start_clipboard(_p: Option<()>) -> Option<UnboundedReceiver<()>> {
let mut clipboard_lock = CLIPBOARD_STATE.lock().unwrap();
if clipboard_lock.running {
return None;
}
clipboard_lock.running = true;
log::info!("Start client clipboard loop");
std::thread::spawn(move || {
loop {
if !CLIPBOARD_STATE.lock().unwrap().running {
break;
}
if !CLIPBOARD_STATE.lock().unwrap().is_text_required {
std::thread::sleep(Duration::from_millis(CLIPBOARD_INTERVAL));
continue;
}
if let Some(msg) = crate::clipboard::get_clipboards_msg(true) {
crate::flutter::send_clipboard_msg(msg, false);
}
std::thread::sleep(Duration::from_millis(CLIPBOARD_INTERVAL));
}
log::info!("Stop client clipboard loop");
CLIPBOARD_STATE.lock().unwrap().running = false;
});
None
}
}
#[cfg(not(target_os = "ios"))]
impl ClipboardState {
fn new() -> Self {
Self {
#[cfg(feature = "flutter")]
is_text_required: true,
#[cfg(all(feature = "flutter", feature = "unix-file-copy-paste"))]
is_file_required: true,
running: false,
}
}
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
struct ClientClipboardHandler {
ctx: Option<crate::clipboard::ClipboardContext>,
#[cfg(not(feature = "flutter"))]
client_clip_ctx: Option<ClientClipboardContext>,
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
impl ClientClipboardHandler {
fn is_text_required(&self) -> bool {
#[cfg(feature = "flutter")]
{
CLIPBOARD_STATE.lock().unwrap().is_text_required
}
#[cfg(not(feature = "flutter"))]
{
self.client_clip_ctx
.as_ref()
.map(|ctx| ctx.cfg.is_text_clipboard_required())
.unwrap_or(false)
}
}
#[cfg(feature = "unix-file-copy-paste")]
fn is_file_required(&self) -> bool {
#[cfg(feature = "flutter")]
{
CLIPBOARD_STATE.lock().unwrap().is_file_required
}
#[cfg(not(feature = "flutter"))]
{
self.client_clip_ctx
.as_ref()
.map(|ctx| ctx.cfg.is_file_clipboard_required())
.unwrap_or(false)
}
}
fn check_clipboard(&mut self, check_text: bool) {
if CLIPBOARD_STATE.lock().unwrap().running {
#[cfg(feature = "unix-file-copy-paste")]
if self.is_file_required() {
if let Some(urls) = check_clipboard_files(&mut self.ctx, ClipboardSide::Client, false) {
if !urls.is_empty() {
#[cfg(target_os = "macos")]
if crate::clipboard::is_file_url_set_by_rustdesk(&urls) {
return;
}
match clipboard::platform::unix::serv_files::sync_files(&urls) {
Ok(()) => {
let msg = crate::clipboard_file::clip_2_msg(
unix_file_clip::get_format_list(),
);
self.send_msg(msg, true);
}
Err(e) => {
log::error!("Failed to sync clipboard files: {}", e);
}
}
return;
}
}
}
if check_text && self.is_text_required() {
if let Some(msg) = check_clipboard(&mut self.ctx, ClipboardSide::Client, false) {
self.send_msg(msg, false);
}
}
}
}
#[inline]
#[cfg(feature = "flutter")]
fn send_msg(&self, msg: Message, _is_file: bool) {
crate::flutter::send_clipboard_msg(msg, _is_file);
}
#[cfg(not(feature = "flutter"))]
fn send_msg(&self, msg: Message, _is_file: bool) {
if let Some(ctx) = &self.client_clip_ctx {
#[cfg(feature = "unix-file-copy-paste")]
if _is_file {
if ctx.is_file_supported {
let _ = ctx.tx.send(Data::Message(msg));
}
return;
}
let pi = ctx.cfg.lc.read().unwrap().peer_info.clone();
if let Some(pi) = pi.as_ref() {
if let Some(message::Union::MultiClipboards(multi_clipboards)) = &msg.union {
if let Some(msg_out) = crate::clipboard::get_msg_if_not_support_multi_clip(
&pi.version,
&pi.platform,
multi_clipboards,
) {
let _ = ctx.tx.send(Data::Message(msg_out));
return;
}
}
}
let _ = ctx.tx.send(Data::Message(msg));
}
}
}
/// Audio handler for the [`Client`].
#[derive(Default)]
pub struct AudioHandler {
audio_decoder: Option<(AudioDecoder, Vec<f32>)>,
#[cfg(target_os = "linux")]
simple: Option<psimple::Simple>,
#[cfg(not(target_os = "linux"))]
audio_buffer: AudioBuffer,
#[cfg(not(target_os = "linux"))]
audio_resampler: Option<crate::audio_resampler::AudioResampler>,
sample_rate: (u32, u32),
#[cfg(not(target_os = "linux"))]
audio_stream: Option<Box<dyn StreamTrait>>,
channels: u16,
#[cfg(not(target_os = "linux"))]
device_channel: u16,
#[cfg(not(target_os = "linux"))]
playback_status: Arc<audio_playback::AudioPlaybackStatus>,
#[cfg(target_os = "windows")]
playback_recovery: audio_playback_recovery::PlaybackRecovery,
}
#[cfg(not(target_os = "linux"))]
#[derive(Clone, Copy)]
struct DecodedAudioConfig {
sample_rate: u32,
input_channels: u16,
output_channels: u16,
}
#[cfg(not(target_os = "linux"))]
fn create_audio_resampler(
input_rate: u32,
output_rate: u32,
channels: u16,
) -> ResultType<Option<crate::audio_resampler::AudioResampler>> {
if input_rate == output_rate {
return Ok(None);
}
Ok(Some(crate::audio_resampler::AudioResampler::new(
crate::audio_resampler::AudioResamplerConfig {
input_rate,
output_rate,
channels,
},
)?))
}
#[cfg(not(target_os = "linux"))]
fn prepare_decoded_audio(
input: &[f32],
resampler: Option<&mut crate::audio_resampler::AudioResampler>,
config: DecodedAudioConfig,
) -> Result<Vec<f32>, crate::audio_resampler::AudioResamplerError> {
let mut output = match resampler {
Some(resampler) => resampler.process(input)?,
None => input.to_owned(),
};
if config.input_channels != config.output_channels {
output = crate::audio_rechannel(
output,
config.sample_rate,
config.sample_rate,
config.input_channels,
config.output_channels,
);
}
Ok(output)
}
#[cfg(not(target_os = "linux"))]
struct AudioBuffer(
pub Arc<std::sync::Mutex<ringbuf::HeapRb<f32>>>,
usize,
[usize; 30],
Arc<std::sync::atomic::AtomicUsize>,
);
#[cfg(not(target_os = "linux"))]
impl Default for AudioBuffer {
fn default() -> Self {
Self(
Arc::new(std::sync::Mutex::new(
ringbuf::HeapRb::<f32>::new(48000 * 2 * AUDIO_BUFFER_MS / 1000), // 48000hz, 2 channel
)),
48000 * 2,
[0; 30],
Arc::new(std::sync::atomic::AtomicUsize::new(0)),
)
}
}
#[cfg(not(target_os = "linux"))]
impl AudioBuffer {
pub fn resize(&mut self, sample_rate: usize, channels: usize) {
let capacity = sample_rate * channels * AUDIO_BUFFER_MS / 1000;
let old_capacity = self.0.lock().unwrap().capacity();
if capacity != old_capacity {
*self.0.lock().unwrap() = ringbuf::HeapRb::<f32>::new(capacity);
self.1 = sample_rate * channels;
log::info!("Audio buffer resized from {old_capacity} to {capacity}");
}
}
fn try_shrink(&mut self, having: usize) {
extern crate chrono;
use chrono::prelude::*;
let mut i = (having * 10) / self.1;
if i > 29 {
i = 29;
}
self.2[i] += 1;
#[allow(non_upper_case_globals)]
static mut tms: i64 = 0;
let dt = Local::now().timestamp_millis();
unsafe {
if tms == 0 {
tms = dt;
return;
} else if dt < tms + 12000 {
return;
}
tms = dt;
}
// the safer water mark to drop
let mut zero = 0;
// the water mark taking most of time
let mut max = 0;
for i in 0..30 {
if self.2[i] == 0 && zero == i {
zero += 1;
}
if self.2[i] > self.2[max] {
self.2[max] = 0;
max = i;
} else {
self.2[i] = 0;
}
}
zero = zero * 2 / 3;
// how many data can be dropped:
// 1. will not drop if buffered data is less than 600ms
// 2. choose based on min(zero, max)
const N: usize = 4;
self.2[max] = 0;
if max < 6 {
return;
} else if max > zero * N {
max = zero * N;
}
let mut lock = self.0.lock().unwrap();
let cap = lock.capacity();
let having = lock.occupied_len();
let skip = (cap * max / (30 * N) + 1) & (!1);
if (having > skip * 3) && (skip > 0) {
lock.skip(skip);
let generation = self.signal_discontinuity();
drop(lock);
log::info!("skip {skip}, based {max} {zero}, generation={generation}");
}
}
/// The caller must hold the PCM buffer lock while signaling the discard.
fn signal_discontinuity(&self) -> usize {
self.3
.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
.wrapping_add(1)
}
/// append pcm to audio buffer, if buffered data
/// exceeds AUDIO_BUFFER_MS, only AUDIO_BUFFER_MS
/// will be kept.
fn append_pcm2(&self, buffer: &[f32]) -> usize {
let mut lock = self.0.lock().unwrap();
let cap = lock.capacity();
let having = lock.occupied_len() + buffer.len();
lock.push_slice_overwrite(buffer);
let discard = (having > cap).then(|| (having - cap, self.signal_discontinuity()));
let occupied = lock.occupied_len();
drop(lock);
if let Some((discarded, generation)) = discard {
hbb_common::throttled_log!(
audio_playback::AUDIO_PLAYBACK_LOG_INTERVAL,
debug,
"Audio buffer capacity discard: samples={discarded}, generation={generation}"
);
}
occupied
}
/// append pcm to audio buffer, trying to drop data
/// when data is too much (per 12 seconds) based
/// statistics.
pub fn append_pcm(&mut self, buffer: &[f32]) {
let having = self.append_pcm2(buffer);
self.try_shrink(having);
}
}
impl AudioHandler {
#[cfg(target_os = "linux")]
fn start_audio(&mut self, format0: AudioFormat) -> ResultType<()> {
use psimple::Simple;
use pulse::sample::{Format, Spec};
use pulse::stream::Direction;
let spec = Spec {
format: Format::F32le,
channels: format0.channels as _,
rate: format0.sample_rate as _,
};
if !spec.is_valid() {
bail!("Invalid audio format");
}
self.simple = Some(Simple::new(
None, // Use the default server
&crate::get_app_name(), // Our application’s name
Direction::Playback, // We want a playback stream
None, // Use the default device
"playback", // Description of our stream
&spec, // Our sample format
None, // Use default channel map
None, // Use default buffering attributes
)?);
self.sample_rate = (format0.sample_rate, format0.sample_rate);
Ok(())
}
/// Start the audio playback.
#[cfg(not(target_os = "linux"))]
fn start_audio(&mut self, format0: AudioFormat) -> ResultType<()> {
let device = AUDIO_HOST
.default_output_device()
.with_context(|| "Failed to get default output device")?;
log::info!(
"Using default output device: \"{}\"",
device.name().unwrap_or("".to_owned())
);
let config = device.default_output_config().map_err(|e| anyhow!(e))?;
let sample_format = config.sample_format();
log::info!("Default output format: {:?}", config);
log::info!("Remote input format: {:?}", format0);
#[allow(unused_mut)]
let mut config: StreamConfig = config.into();
#[cfg(not(any(target_os = "android", target_os = "ios")))]
{
// this makes ios and android audio output not work
config.buffer_size = cpal::BufferSize::Fixed(64);
}
self.sample_rate = (format0.sample_rate, config.sample_rate.0);
let audio_resampler = create_audio_resampler(
format0.sample_rate, config.sample_rate.0, format0.channels as _,
)?;
let mut build_output_stream = |config: StreamConfig| match sample_format {
cpal::SampleFormat::I8 => self.build_output_stream::<i8>(&config, &device),
cpal::SampleFormat::I16 => self.build_output_stream::<i16>(&config, &device),
cpal::SampleFormat::I32 => self.build_output_stream::<i32>(&config, &device),
cpal::SampleFormat::I64 => self.build_output_stream::<i64>(&config, &device),
cpal::SampleFormat::U8 => self.build_output_stream::<u8>(&config, &device),
cpal::SampleFormat::U16 => self.build_output_stream::<u16>(&config, &device),
cpal::SampleFormat::U32 => self.build_output_stream::<u32>(&config, &device),
cpal::SampleFormat::U64 => self.build_output_stream::<u64>(&config, &device),
cpal::SampleFormat::F32 => self.build_output_stream::<f32>(&config, &device),
cpal::SampleFormat::F64 => self.build_output_stream::<f64>(&config, &device),
f => bail!("unsupported audio format: {:?}", f),
};
if config.channels > format0.channels as _ {
let no_rechannel_config = StreamConfig {
channels: format0.channels as _,
..config.clone()
};
if let Err(_) = build_output_stream(no_rechannel_config) {
build_output_stream(config)?;
}
} else {
build_output_stream(config)?;
}
self.audio_resampler = audio_resampler;
Ok(())
}
/// Handle audio format and create an audio decoder.
pub fn handle_format(&mut self, f: AudioFormat) {
self.handle_format_with_start(f, Self::start_audio);
}
fn handle_format_with_start(
&mut self,
f: AudioFormat,
start: impl FnOnce(&mut Self, AudioFormat) -> ResultType<()>,
) {
if !is_supported_audio_channel_count(f.channels) {
log::error!("Unsupported audio channel count: {}", f.channels);
return;
}
match AudioDecoder::new(f.sample_rate, if f.channels > 1 { Stereo } else { Mono }) {
Ok(d) => {
#[cfg(target_os = "windows")]
let playback_failed = self.cancel_pending_playback();
#[cfg(target_os = "linux")]
let keep_existing_stream = self.simple.is_some()
&& self.sample_rate.0 == f.sample_rate
&& u32::from(self.channels) == f.channels;
#[cfg(not(target_os = "linux"))]
let keep_existing_stream = self.audio_stream.is_some()
&& self.sample_rate.0 == f.sample_rate
&& u32::from(self.channels) == f.channels;
let buffer = vec![0.; f.sample_rate as usize * f.channels as usize];
#[cfg(not(target_os = "linux"))]
let mut previous = std::mem::take(self);
#[cfg(target_os = "windows")]
self.prepare_playback(&f);
self.audio_decoder = Some((d, buffer));
self.channels = f.channels as _;
let result = start(self, f);
#[cfg(target_os = "windows")]
let keep_existing_stream = keep_existing_stream
&& !playback_failed
&& !previous.playback_recovery.report_pending();
#[cfg(not(target_os = "linux"))]
if result.is_err() && keep_existing_stream {
// The restarted capture has new Opus history even when output startup fails.
previous.audio_decoder = self.audio_decoder.take();
*self = previous;
self.handle_audio_start_result(result, true);
return;
}
#[cfg(target_os = "windows")]
self.finish_playback_replacement(result, keep_existing_stream.then_some(previous));
#[cfg(not(target_os = "windows"))]
self.handle_audio_start_result(result, keep_existing_stream);
}
Err(err) => {
log::error!("Failed to create audio decoder: {}", err);
}
}
}
fn handle_audio_start_result(&mut self, result: ResultType<()>, keep_existing_stream: bool) {
if let Err(error) = result {
if keep_existing_stream {
log::error!(
"Failed to replace audio playback stream; keeping the existing compatible stream: {error:#}"
);
} else {
*self = Self::default();
log::error!("Failed to start audio playback: {error:#}");
}
}
}
/// Handle audio frame and play it.
#[inline]
pub fn handle_frame(&mut self, frame: AudioFrame) {
#[cfg(not(target_os = "linux"))]
self.playback_status.report_errors();
#[cfg(not(target_os = "linux"))]
if self.audio_stream.is_none()
|| !self
.playback_status
.ready
.load(std::sync::atomic::Ordering::Acquire)
{
return;
}
#[cfg(target_os = "linux")]
if self.simple.is_none() {
log::trace!("PulseAudio simple binding does not exists");
return;
}
self.audio_decoder.as_mut().map(|(d, buffer)| {
let decoded_frames = match d.decode_float(&frame.data, buffer, false) {
Ok(decoded_frames) => decoded_frames,
Err(error) => {
log::warn!("Failed to decode audio frame: {error:?}");
return;
}
};
let channels = self.channels;
let n = decoded_frames * channels as usize;
#[cfg(not(target_os = "linux"))]
{
let config = DecodedAudioConfig {
sample_rate: self.sample_rate.1,
input_channels: self.channels,
output_channels: self.device_channel,
};
let buffer = match prepare_decoded_audio(
&buffer[0..n],
self.audio_resampler.as_mut(),
config,
) {
Ok(output) => output,
Err(error) => {
log::error!("Failed to resample decoded audio: {error:#}");
return;
}
};
self.audio_buffer.append_pcm(&buffer);
}
#[cfg(target_os = "linux")]
{
let data_u8 =
unsafe { std::slice::from_raw_parts::<u8>(buffer.as_ptr() as _, n * 4) };
self.simple.as_mut().map(|x| x.write(data_u8));
}
});
}
/// Build audio output stream for current device.
#[cfg(not(target_os = "linux"))]
fn build_output_stream<T: cpal::Sample + cpal::SizedSample + cpal::FromSample<f32>>(
&mut self,
config: &StreamConfig,
device: &Device,
) -> ResultType<()> {
self.device_channel = config.channels;
#[cfg(target_os = "windows")]
let err_fn = self.playback_recovery.new_error_callback();
#[cfg(not(target_os = "windows"))]
let err_fn = move |err| {
// too many errors, will improve later
log::trace!("an error occurred on stream: {}", err);
};
self.audio_buffer
.resize(config.sample_rate.0 as _, config.channels as _);
let audio_buffer = self.audio_buffer.0.clone();
let discontinuity_generation = self.audio_buffer.3.clone();
let mut playback_writer = audio_playback::AudioPlaybackWriter::new(
audio_playback::AudioPlaybackConfig {
sample_rate: config.sample_rate.0,
channels: config.channels as usize,
},
audio_buffer,
discontinuity_generation,
)?;
let playback_status = playback_writer.status.clone();
let timeout = None;
let stream = device.build_output_stream(
config,
move |data: &mut [T], _: &cpal::OutputCallbackInfo| {
playback_writer.write_output(data);
},
err_fn,
timeout,
)?;
stream.play()?;
self.audio_stream = Some(Box::new(stream));
self.playback_status = playback_status;
Ok(())
}
}
fn is_supported_audio_channel_count(channels: u32) -> bool {
(1..=2).contains(&channels)
}
#[cfg(test)]
mod audio_format_tests {
use super::is_supported_audio_channel_count;
#[test]
fn only_mono_and_stereo_are_supported() {
assert!(is_supported_audio_channel_count(1));
assert!(is_supported_audio_channel_count(2));
assert!(!is_supported_audio_channel_count(0));
assert!(!is_supported_audio_channel_count(u32::MAX));
}
#[test]
fn failed_audio_start_discards_format_state() {
use super::{anyhow, AudioDecoder, AudioHandler, Stereo};
const SAMPLE_RATE: u32 = 48_000;
const CHANNELS: u16 = 2;
let decoder = AudioDecoder::new(SAMPLE_RATE, Stereo).unwrap();
let mut handler = AudioHandler {
audio_decoder: Some((decoder, Vec::new())),
sample_rate: (SAMPLE_RATE, SAMPLE_RATE),
channels: CHANNELS,
..Default::default()
};
handler.handle_audio_start_result(Err(anyhow!("Injected playback startup failure")), false);
assert!(handler.audio_decoder.is_none());
assert_eq!(handler.channels, 0);
assert_eq!(handler.sample_rate, (0, 0));
}
}
/// Video handler for the [`Client`].
pub struct VideoHandler {
decoder: Decoder,
pub rgb: ImageRgb,
pub texture: ImageTexture,
recorder: Arc<Mutex<Option<Recorder>>>,
record: bool,
_display: usize, // useful for debug
fail_counter: usize,
first_frame: bool,
}
impl VideoHandler {
#[cfg(feature = "flutter")]
pub fn get_adapter_luid() -> Option<i64> {
crate::flutter::get_adapter_luid()
}
#[cfg(not(feature = "flutter"))]
pub fn get_adapter_luid() -> Option<i64> {
None
}
/// Create a new video handler.
pub fn new(format: CodecFormat, _display: usize) -> Self {
let luid = Self::get_adapter_luid();
log::info!("new video handler for display #{_display}, format: {format:?}, luid: {luid:?}");
let rgba_format =
if cfg!(feature = "flutter") && (cfg!(windows) || cfg!(target_os = "linux")) {
ImageFormat::ABGR
} else {
ImageFormat::ARGB
};
VideoHandler {
decoder: Decoder::new(format, luid),
rgb: ImageRgb::new(rgba_format, crate::get_dst_align_rgba()),
texture: Default::default(),
recorder: Default::default(),
record: false,
_display,
fail_counter: 0,
first_frame: true,
}
}
/// Handle a new video frame.
#[inline]
pub fn handle_frame(
&mut self,
vf: VideoFrame,
pixelbuffer: &mut bool,
chroma: &mut Option<Chroma>,
) -> ResultType<bool> {
let format = CodecFormat::from(&vf);
if format != self.decoder.format() {
self.reset(Some(format));
}
match &vf.union {
Some(frame) => {
let res = self.decoder.handle_video_frame(
frame,
&mut self.rgb,
&mut self.texture,
pixelbuffer,
chroma,
);
if res.as_ref().is_ok_and(|x| *x) {
self.fail_counter = 0;
} else {
if self.fail_counter < usize::MAX {
if self.first_frame && self.fail_counter < MAX_DECODE_FAIL_COUNTER {
log::error!("decode first frame failed");
self.fail_counter = MAX_DECODE_FAIL_COUNTER;
} else {
self.fail_counter += 1;
}
log::error!(
"Failed to handle video frame, fail counter: {}",
self.fail_counter
);
}
}
self.first_frame = false;
if self.record {
self.recorder.lock().unwrap().as_mut().map(|r| {
let (w, h) = if *pixelbuffer {
(self.rgb.w, self.rgb.h)
} else {
(self.texture.w, self.texture.h)
};
r.write_frame(frame, w, h).ok();
});
}
res
}
_ => Ok(false),
}
}
/// Reset the decoder, change format if it is Some
pub fn reset(&mut self, format: Option<CodecFormat>) {
log::info!(
"reset video handler for display #{}, format: {format:?}",
self._display
);
#[cfg(target_os = "macos")]
self.rgb.set_align(crate::get_dst_align_rgba());
let luid = Self::get_adapter_luid();
let format = format.unwrap_or(self.decoder.format());
self.decoder = Decoder::new(format, luid);
self.fail_counter = 0;
self.first_frame = true;
}
/// Start or stop screen record.
pub fn record_screen(&mut self, start: bool, id: String, display_idx: usize, camera: bool) {
self.record = false;
if start {
self.recorder = Recorder::new(RecorderContext {
server: false,
id,
dir: crate::ui_interface::video_save_directory(false),
display_idx,
camera,
tx: None,
})
.map_or(Default::default(), |r| Arc::new(Mutex::new(Some(r))));
} else {
self.recorder = Default::default();
}
self.record = start;
}
}
// The source of sent password
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
enum PasswordSource {
PersonalAb(Vec<u8>),
SharedAb(String),
Undefined,
}
impl Default for PasswordSource {
fn default() -> Self {
PasswordSource::Undefined
}
}
impl PasswordSource {
// Whether the password is personal ab password
pub fn is_personal_ab(&self, password: &[u8]) -> bool {
if password.is_empty() {
return false;
}
match self {
PasswordSource::PersonalAb(p) => p == password,
_ => false,
}
}
// Whether the password is shared ab password
pub fn is_shared_ab(&self, password: &[u8], hash: &Hash) -> bool {
if password.is_empty() {
return false;
}
match self {
PasswordSource::SharedAb(p) => Self::equal(p, password, hash),
_ => false,
}
}
// Whether the password equals to the connected password
fn equal(password: &str, connected_password: &[u8], hash: &Hash) -> bool {
let mut hasher = Sha256::new();
hasher.update(password);
hasher.update(&hash.salt);
let res = hasher.finalize();
connected_password[..] == res[..]
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct ConnToken {
password: Vec<u8>,
password_source: PasswordSource,
session_id: u64,
}
/// Login config handler for [`Client`].
#[derive(Default)]
pub struct LoginConfigHandler {
id: String,
pub conn_type: ConnType,
pub is_terminal_admin: bool,
hash: Hash,
password: Vec<u8>, // remember password for reconnect
pub remember: bool,
login_method: login_method::LoginMethod,
config: PeerConfig,
pub port_forward: (String, i32),
/// This login's `multiplex`, filled with `port_forward` under the turn
/// lock. `port_forward_mux` says whether a mapping probes for the tunnel;
/// one the probe latched to the raw pipe logs in without asking, so an
/// upgraded peer keeps giving it the raw pipe.
pub(crate) port_forward_multiplex: bool,
/// Set once per window, before its mappings start: every accept's claim
/// reads it.
pub(crate) port_forward_mux: bool,
/// Held by a port-forward mapping from filling `port_forward` and `hash`
/// until its login is built from them; a window's mappings log in
/// concurrently.
pub(crate) port_forward_login_turn: Arc<hbb_common::tokio::sync::Mutex<()>>,
pub version: i64,
features: Option<Features>,
pub session_id: u64, // used for local <-> server communication
pub supported_encoding: SupportedEncoding,
restarting_remote_device: bool,
// Start time of the restart grace window. On Windows the peer may briefly
// reconnect before the real reboot disconnect.
restart_remote_device_at: Option<Instant>,
pub force_relay: bool,
// force_relay minus the WebSocket transport. ws forces relay for classic punching but says
// nothing about ICE, so every WebRTC decision keys off this: policy means Relay-only ICE,
// transport may still go direct.
pub policy_relay: bool,
// The peer-scoped part of it, and the only part that may be written back to the peer.
pub peer_relay: bool,
pub direct: Option<bool>,
pub received: bool,
switch_uuid: Option<String>,
#[cfg(feature = "flutter")]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
switch_back_allowed: bool,
pub save_ab_password_to_recent: bool, // true: connected with ab password
pub other_server: Option<(String, String, String)>,
pub custom_fps: Arc<Mutex<Option<usize>>>,
pub adapter_luid: Option<i64>,
pub mark_unsupported: Vec<CodecFormat>,
pub selected_windows_session_id: Option<u32>,
pub peer_info: Option<PeerInfo>,
password_source: PasswordSource, // where the sent password comes from
shared_password: Option<String>, // Store the shared password
pub enable_trusted_devices: bool,
pub record_state: bool,
pub record_permission: bool,
}
impl Deref for LoginConfigHandler {
type Target = PeerConfig;
fn deref(&self) -> &Self::Target {
&self.config
}
}
impl LoginConfigHandler {
pub(crate) fn set_hash(&mut self, hash: Hash) {
self.hash = hash;
}
/// Initialize the login config handler.
///
/// # Arguments
///
/// * `id` - id of peer
/// * `conn_type` - Connection type enum.
pub fn initialize(
&mut self,
id: String,
conn_type: ConnType,
switch_uuid: Option<String>,
mut force_relay: bool,
adapter_luid: Option<i64>,
shared_password: Option<String>,
conn_token: Option<String>,
login_method: &str,
) {
let mut id = id;
if id.contains("@") {
let mut v = id.split("@");
let raw_id: &str = v.next().unwrap_or_default();
let mut server_key = v.next().unwrap_or_default().split('?');
let server = server_key.next().unwrap_or_default();
let args = server_key.next().unwrap_or_default();
let key = if server == PUBLIC_SERVER {
config::RS_PUB_KEY.to_owned()
} else {
let mut args_map: HashMap<String, &str> = HashMap::new();
for arg in args.split('&') {
if let Some(kv) = arg.find('=') {
let k = arg[0..kv].to_lowercase();
let v = &arg[kv + 1..];
args_map.insert(k, v);
}
}
let key = args_map.remove("key").unwrap_or_default();
key.to_owned()
};
// here we can check <id>/r@server
let real_id = crate::ui_interface::handle_relay_id(raw_id).to_string();
if real_id != raw_id {
force_relay = true;
}
self.other_server = Some((real_id.clone(), server.to_owned(), key));
id = format!("{real_id}@{server}");
} else {
let real_id = crate::ui_interface::handle_relay_id(&id);
if real_id != id {
force_relay = true;
id = real_id.to_owned();
}
}
self.id = id;
self.conn_type = conn_type;
self.login_method = Default::default();
if !login_method.is_empty() && !self.set_login_method(login_method) {
log::warn!("Failed to initialize login method: {}", login_method);
}
let config = self.load_config();
self.remember = !config.password.is_empty();
self.config = config;
let conn_token = conn_token
.map(|x| serde_json::from_str::<ConnToken>(&x).ok())
.flatten();
let mut sid = 0;
if let Some(token) = conn_token {
sid = token.session_id;
self.password = token.password; // use as last password
self.password_source = token.password_source;
}
if sid == 0 {
sid = rand::random();
if sid == 0 {
// you won the lottery
sid = 1;
}
}
self.session_id = sid;
self.supported_encoding = Default::default();
self.clear_restarting_remote_device();
// Three scopes: what was decided about this PEER, what this CLIENT is set up as (proxy),
// and what its TRANSPORT forces (ws). Only the first may be written back to the peer's
// config — persisting the others would make a local setup a permanent peer property.
self.peer_relay =
config::option2bool("force-always-relay", &self.get_option("force-always-relay"))
|| 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");
if !other_server_key.is_empty() && key.is_empty() {
self.other_server = Some((real_id.to_owned(), server.to_owned(), other_server_key));
}
}
self.direct = None;
self.received = false;
#[cfg(feature = "flutter")]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
{
self.switch_back_allowed = false;
}
self.switch_uuid = switch_uuid;
self.adapter_luid = adapter_luid;
self.selected_windows_session_id = None;
self.shared_password = shared_password;
self.record_state = false;
self.record_permission = true;
// `std::env::remove_var("IS_TERMINAL_ADMIN");` is called in `session_add_sync()` - `flutter_ffi.rs`.
let is_terminal_admin = conn_type == ConnType::TERMINAL
&& std::env::var("IS_TERMINAL_ADMIN").map_or(false, |v| v == "Y");
self.is_terminal_admin = is_terminal_admin;
}
#[cfg(feature = "flutter")]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
pub fn allow_switch_back_once(&mut self) {
self.switch_back_allowed = true;
}
#[cfg(feature = "flutter")]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
pub fn consume_switch_back_permission(&mut self) -> bool {
if self.switch_back_allowed {
self.switch_back_allowed = false;
true
} else {
false
}
}
/// Check if the client should auto login.
/// Return password if the client should auto login, otherwise return empty string.
pub fn should_auto_login(&self) -> String {
let l = self.lock_after_session_end.v;
let a = !self.get_option("auto-login").is_empty();
let p = self.get_option("os-password");
if !p.is_empty() && l && a {
p
} else {
"".to_owned()
}
}
/// Load [`PeerConfig`].
pub fn load_config(&self) -> PeerConfig {
debug_assert!(self.id.len() > 0);
PeerConfig::load(&self.id)
}
/// Save a [`PeerConfig`] into the handler.
///
/// # Arguments
///
/// * `config` - [`PeerConfig`] to save.
pub fn save_config(&mut self, config: PeerConfig) {
config.store(&self.id);
self.config = config;
}
/// Set an option for handler's [`PeerConfig`].
///
/// # Arguments
///
/// * `k` - key of option
/// * `v` - value of option
pub fn set_option(&mut self, k: String, v: String) {
let mut config = self.load_config();
if v == self.get_option(&k) {
return;
}
config.options.insert(k, v);
self.save_config(config);
}
//to-do: too many dup code below.
/// Save view style to the current config.
///
/// # Arguments
///
/// * `value` - The view style to be saved.
pub fn save_view_style(&mut self, value: String) {
let mut config = self.load_config();
config.view_style = value;
self.save_config(config);
}
/// Save keyboard mode to the current config.
///
/// # Arguments
///
/// * `value` - The view style to be saved.
pub fn save_keyboard_mode(&mut self, value: String) {
let mut config = self.load_config();
config.keyboard_mode = value;
self.save_config(config);
}
/// Save reverse mouse wheel ("", "Y") to the current config.
///
/// # Arguments
///
/// * `value` - The reverse mouse wheel ("", "Y").
pub fn save_reverse_mouse_wheel(&mut self, value: String) {
let mut config = self.load_config();
config.reverse_mouse_wheel = value;
self.save_config(config);
}
/// Save "displays_as_individual_windows" ("", "Y") to the current config.
///
/// # Arguments
///
/// * `value` - The "displays_as_individual_windows" value ("", "Y").
pub fn save_displays_as_individual_windows(&mut self, value: String) {
let mut config = self.load_config();
config.displays_as_individual_windows = value;
self.save_config(config);
}
/// Save "use_all_my_displays_for_the_remote_session" ("", "Y") to the current config.
///
/// # Arguments
///
/// * `value` - The "use_all_my_displays_for_the_remote_session" value ("", "Y").
pub fn save_use_all_my_displays_for_the_remote_session(&mut self, value: String) {
let mut config = self.load_config();
config.use_all_my_displays_for_the_remote_session = value;
self.save_config(config);
}
/// Save scroll style to the current config.
///
/// # Arguments
///
/// * `value` - The scroll style to be saved.
pub fn save_scroll_style(&mut self, value: String) {
let mut config = self.load_config();
config.scroll_style = value;
self.save_config(config);
}
/// Save edge scroll edge thickness to the current config.
///
/// # Arguments
///
/// * `value` - The edge thickness to be saved.
pub fn save_edge_scroll_edge_thickness(&mut self, value: i32) {
let mut config = self.load_config();
config.edge_scroll_edge_thickness = value;
self.save_config(config);
}
/// Set a ui config of flutter for handler's [`PeerConfig`].
///
/// # Arguments
///
/// * `k` - key of option
/// * `v` - value of option
pub fn save_ui_flutter(&mut self, k: String, v: String) {
let mut config = self.load_config();
if v.is_empty() {
config.ui_flutter.remove(&k);
} else {
config.ui_flutter.insert(k, v);
}
self.save_config(config);
}
pub fn set_direct_failure(&mut self, value: i32) {
let mut config = self.load_config();
config.direct_failures = value;
self.save_config(config);
}
/// Get a ui config of flutter for handler's [`PeerConfig`].
/// Return String if the option is found, otherwise return "".
///
/// # Arguments
///
/// * `k` - key of option
pub fn get_ui_flutter(&self, k: &str) -> String {
if let Some(v) = self.config.ui_flutter.get(k) {
v.clone()
} else {
"".to_owned()
}
}
/// Toggle an option in the handler.
///
/// # Arguments
///
/// * `name` - The name of the option to toggle.
///
// It's Ok to check the option empty in this function.
// `toggle_option()` is only called in a session.
// Custom client advanced settings will not effect this function.
pub fn toggle_option(&mut self, name: String) -> Option<Message> {
let mut option = OptionMessage::default();
let mut config = self.load_config();
if name == "show-remote-cursor" {
config.show_remote_cursor.v = !config.show_remote_cursor.v;
option.show_remote_cursor = (if config.show_remote_cursor.v {
BoolOption::Yes
} else {
BoolOption::No
})
.into();
} else if name == "follow-remote-cursor" {
config.follow_remote_cursor.v = !config.follow_remote_cursor.v;
option.follow_remote_cursor = (if config.follow_remote_cursor.v {
BoolOption::Yes
} else {
BoolOption::No
})
.into();
} else if name == "follow-remote-window" {
config.follow_remote_window.v = !config.follow_remote_window.v;
option.follow_remote_window = (if config.follow_remote_window.v {
BoolOption::Yes
} else {
BoolOption::No
})
.into();
} else if name == "disable-audio" {
config.disable_audio.v = !config.disable_audio.v;
option.disable_audio = (if config.disable_audio.v {
BoolOption::Yes
} else {
BoolOption::No
})
.into();
} else if name == "disable-clipboard" {
config.disable_clipboard.v = !config.disable_clipboard.v;
option.disable_clipboard = (if config.disable_clipboard.v {
BoolOption::Yes
} else {
BoolOption::No
})
.into();
} else if name == "lock-after-session-end" {
config.lock_after_session_end.v = !config.lock_after_session_end.v;
option.lock_after_session_end = (if config.lock_after_session_end.v {
BoolOption::Yes
} else {
BoolOption::No
})
.into();
} else if name == keys::OPTION_TERMINAL_PERSISTENT {
config.terminal_persistent.v = !config.terminal_persistent.v;
option.terminal_persistent = (if config.terminal_persistent.v {
BoolOption::Yes
} else {
BoolOption::No
})
.into();
} else if name == "privacy-mode" {
// try toggle privacy mode
option.privacy_mode = (if config.privacy_mode.v {
BoolOption::No
} else {
BoolOption::Yes
})
.into();
} else if name == "enable-file-copy-paste" {
config.enable_file_copy_paste.v = !config.enable_file_copy_paste.v;
option.enable_file_transfer = (if config.enable_file_copy_paste.v {
BoolOption::Yes
} else {
BoolOption::No
})
.into();
} else if name == "block-input" {
option.block_input = BoolOption::Yes.into();
} else if name == "unblock-input" {
option.block_input = BoolOption::No.into();
} else if name == "show-quality-monitor" {
config.show_quality_monitor.v = !config.show_quality_monitor.v;
} else if name == "allow_swap_key" {
config.allow_swap_key.v = !config.allow_swap_key.v;
} else if name == "view-only" {
config.view_only.v = !config.view_only.v;
let f = |b: bool| {
if b {
BoolOption::Yes.into()
} else {
BoolOption::No.into()
}
};
if config.view_only.v {
option.disable_keyboard = f(true);
option.disable_clipboard = f(true);
option.show_remote_cursor = f(true);
option.enable_file_transfer = f(false);
option.lock_after_session_end = f(false);
} else {
option.disable_keyboard = f(false);
option.disable_clipboard = f(self.get_toggle_option("disable-clipboard"));
option.show_remote_cursor = f(self.get_toggle_option("show-remote-cursor"));
option.enable_file_transfer = f(self.config.enable_file_copy_paste.v);
option.lock_after_session_end = f(self.config.lock_after_session_end.v);
if config.show_my_cursor.v {
config.show_my_cursor.v = false;
option.show_my_cursor = BoolOption::No.into();
}
}
} else if name == "show-my-cursor" {
config.show_my_cursor.v = !config.show_my_cursor.v;
option.show_my_cursor = if config.show_my_cursor.v {
BoolOption::Yes
} else {
BoolOption::No
}
.into();
} else {
let is_set = self
.options
.get(&name)
.map(|o| !o.is_empty())
.unwrap_or(false);
if is_set {
self.config.options.remove(&name);
} else {
self.config.options.insert(name, "Y".to_owned());
}
self.config.store(&self.id);
return None;
}
#[cfg(feature = "unix-file-copy-paste")]
if option.enable_file_transfer.enum_value() == Ok(BoolOption::No) {
crate::clipboard::try_empty_clipboard_files(crate::clipboard::ClipboardSide::Client, 0);
}
if !name.contains("block-input") {
self.save_config(config);
}
let mut misc = Misc::new();
misc.set_option(option);
let mut msg_out = Message::new();
msg_out.set_misc(misc);
Some(msg_out)
}
/// Get [`PeerConfig`] of the current [`LoginConfigHandler`].
///
/// # Arguments
pub fn get_config(&mut self) -> &mut PeerConfig {
&mut self.config
}
/// Get [`OptionMessage`] of the current [`LoginConfigHandler`].
/// Return `None` if there's no option, for example, when the session is only for file transfer.
///
/// # Arguments
///
/// * `ignore_default` - If `true`, ignore the default value of the option.
fn get_option_message(&self, ignore_default: bool) -> Option<OptionMessage> {
if self.conn_type.eq(&ConnType::PORT_FORWARD)
|| self.conn_type.eq(&ConnType::RDP)
|| self.conn_type.eq(&ConnType::FILE_TRANSFER)
{
return None;
}
let mut msg = OptionMessage::new();
if self.conn_type.eq(&ConnType::TERMINAL) {
if self.get_toggle_option(keys::OPTION_TERMINAL_PERSISTENT) {
msg.terminal_persistent = BoolOption::Yes.into();
return Some(msg);
} else {
return None;
}
}
let q = self.image_quality.clone();
if let Some(q) = self.get_image_quality_enum(&q, ignore_default) {
msg.image_quality = q.into();
} else if q == "custom" {
let config = self.load_config();
let allow_more = !crate::using_public_server() || self.direct == Some(true);
let quality = if config.custom_image_quality.is_empty() {
50
} else {
let mut quality = config.custom_image_quality[0];
if !allow_more && quality > 100 {
quality = 50;
}
quality
};
msg.custom_image_quality = quality << 8;
#[cfg(feature = "flutter")]
if let Some(custom_fps) = self.options.get("custom-fps") {
let mut custom_fps = custom_fps.parse().unwrap_or(30);
if !allow_more && custom_fps > 30 {
custom_fps = 30;
}
msg.custom_fps = custom_fps;
*self.custom_fps.lock().unwrap() = Some(custom_fps as _);
}
}
let view_only = self.get_toggle_option("view-only");
if view_only {
msg.disable_keyboard = BoolOption::Yes.into();
}
if view_only || self.get_toggle_option("show-remote-cursor") {
msg.show_remote_cursor = BoolOption::Yes.into();
}
if view_only && self.get_toggle_option("show-my-cursor") {
msg.show_my_cursor = BoolOption::Yes.into();
}
if self.get_toggle_option("follow-remote-cursor") {
msg.follow_remote_cursor = BoolOption::Yes.into();
}
if self.get_toggle_option("follow-remote-window") {
msg.follow_remote_window = BoolOption::Yes.into();
}
if !view_only && self.get_toggle_option("lock-after-session-end") {
msg.lock_after_session_end = BoolOption::Yes.into();
}
if self.get_toggle_option("disable-audio") {
msg.disable_audio = BoolOption::Yes.into();
}
if !view_only && self.get_toggle_option(keys::OPTION_ENABLE_FILE_COPY_PASTE) {
msg.enable_file_transfer = BoolOption::Yes.into();
}
if view_only || self.get_toggle_option("disable-clipboard") {
msg.disable_clipboard = BoolOption::Yes.into();
}
msg.supported_decoding = MessageField::some(self.get_supported_decoding());
Some(msg)
}
pub fn get_supported_decoding(&self) -> SupportedDecoding {
Decoder::supported_decodings(
Some(&self.id),
use_texture_render(),
self.adapter_luid,
&self.mark_unsupported,
)
}
/// Parse the image quality option.
/// Return [`ImageQuality`] if the option is valid, otherwise return `None`.
///
/// # Arguments
///
/// * `q` - The image quality option.
/// * `ignore_default` - Ignore the default value.
fn get_image_quality_enum(&self, q: &str, ignore_default: bool) -> Option<ImageQuality> {
if q == "low" {
Some(ImageQuality::Low)
} else if q == "best" {
Some(ImageQuality::Best)
} else if q == "balanced" {
if ignore_default {
None
} else {
Some(ImageQuality::Balanced)
}
} else {
None
}
}
/// Get the status of a toggle option.
///
/// # Arguments
///
/// * `name` - The name of the toggle option.
///
// It's Ok to check the option empty in this function.
// `get_toggle_option()` is only called in a session.
// Custom client advanced settings will not effect this function.
pub fn get_toggle_option(&self, name: &str) -> bool {
if name == "show-remote-cursor" {
self.config.show_remote_cursor.v
} else if name == "lock-after-session-end" {
self.config.lock_after_session_end.v
} else if name == keys::OPTION_TERMINAL_PERSISTENT {
self.config.terminal_persistent.v
} else if name == "privacy-mode" {
self.config.privacy_mode.v
} else if name == keys::OPTION_ENABLE_FILE_COPY_PASTE {
self.config.enable_file_copy_paste.v
} else if name == "disable-audio" {
self.config.disable_audio.v
} else if name == "disable-clipboard" {
self.config.disable_clipboard.v
} else if name == "show-quality-monitor" {
self.config.show_quality_monitor.v
} else if name == "allow_swap_key" {
self.config.allow_swap_key.v
} else if name == "view-only" {
self.config.view_only.v
} else if name == "show-my-cursor" {
self.config.show_my_cursor.v
} else if name == "follow-remote-cursor" {
self.config.follow_remote_cursor.v
} else if name == "follow-remote-window" {
self.config.follow_remote_window.v
} else {
!self.get_option(name).is_empty()
}
}
pub fn is_privacy_mode_supported(&self) -> bool {
if let Some(features) = &self.features {
features.privacy_mode
} else {
false
}
}
/// Create a [`Message`] for refreshing video.
pub fn refresh() -> Message {
let mut misc = Misc::new();
misc.set_refresh_video(true);
let mut msg_out = Message::new();
msg_out.set_misc(misc);
msg_out
}
/// Create a [`Message`] for refreshing video.
pub fn refresh_display(display: usize) -> Message {
let mut misc = Misc::new();
misc.set_refresh_video_display(display as _);
let mut msg_out = Message::new();
msg_out.set_misc(misc);
msg_out
}
/// Create a [`Message`] for saving custom image quality.
///
/// # Arguments
///
/// * `bitrate` - The given bitrate.
/// * `quantizer` - The given quantizer.
pub fn save_custom_image_quality(&mut self, image_quality: i32) -> Message {
let mut misc = Misc::new();
misc.set_option(OptionMessage {
custom_image_quality: image_quality << 8,
..Default::default()
});
let mut msg_out = Message::new();
msg_out.set_misc(misc);
let mut config = self.load_config();
config.image_quality = "custom".to_owned();
config.custom_image_quality = vec![image_quality as _];
self.save_config(config);
msg_out
}
/// Save the given image quality to the config.
/// Return a [`Message`] that contains image quality, or `None` if the image quality is not valid.
/// # Arguments
///
/// * `value` - The image quality.
pub fn save_image_quality(&mut self, value: String) -> Option<Message> {
let mut res = None;
if let Some(q) = self.get_image_quality_enum(&value, false) {
let mut misc = Misc::new();
misc.set_option(OptionMessage {
image_quality: q.into(),
..Default::default()
});
let mut msg_out = Message::new();
msg_out.set_misc(misc);
res = Some(msg_out);
}
let mut config = self.load_config();
config.image_quality = value;
self.save_config(config);
res
}
pub fn save_trackpad_speed(&mut self, speed: i32) {
let mut config = self.load_config();
config.trackpad_speed = speed;
self.save_config(config);
}
/// Create a [`Message`] for saving custom fps.
///
/// # Arguments
///
/// * `fps` - The given fps.
/// * `save_config` - Save the config.
pub fn set_custom_fps(&mut self, fps: i32, save_config: bool) -> Message {
let mut misc = Misc::new();
misc.set_option(OptionMessage {
custom_fps: fps,
..Default::default()
});
let mut msg_out = Message::new();
msg_out.set_misc(misc);
if save_config {
let mut config = self.load_config();
config
.options
.insert("custom-fps".to_owned(), fps.to_string());
self.save_config(config);
}
*self.custom_fps.lock().unwrap() = Some(fps as _);
msg_out
}
pub fn get_option(&self, k: &str) -> String {
if let Some(v) = self.config.options.get(k) {
v.clone()
} else {
"".to_owned()
}
}
#[inline]
pub fn get_custom_resolution(&self, display: i32) -> Option<(i32, i32)> {
self.config
.custom_resolutions
.get(&display.to_string())
.map(|r| (r.w, r.h))
}
#[inline]
pub fn set_custom_resolution(&mut self, display: i32, wh: Option<(i32, i32)>) {
let display = display.to_string();
let mut config = self.load_config();
match wh {
Some((w, h)) => {
config
.custom_resolutions
.insert(display, Resolution { w, h });
}
None => {
config.custom_resolutions.remove(&display);
}
}
self.save_config(config);
}
/// Get user name.
/// Return the name of the given peer. If the peer has no name, return the name in the config.
///
/// # Arguments
///
/// * `pi` - peer info.
pub fn get_username(&self, pi: &PeerInfo) -> String {
return if pi.username.is_empty() {
self.info.username.clone()
} else {
pi.username.clone()
};
}
/// Handle peer info.
///
/// # Arguments
///
/// * `username` - The name of the peer.
/// * `pi` - The peer info.
pub fn handle_peer_info(&mut self, pi: &PeerInfo) {
if !pi.version.is_empty() {
self.version = hbb_common::get_version_number(&pi.version);
}
self.features = pi.features.clone().into_option();
let serde = PeerInfoSerde {
username: pi.username.clone(),
hostname: pi.hostname.clone(),
platform: pi.platform.clone(),
};
let mut config = self.load_config();
config.info = serde;
let password = self.password.clone();
let password0 = config.password.clone();
let remember = self.remember;
let hash = self.hash.clone();
if remember && !self.skips_saved_password() {
// remember is true: use PeerConfig password or ui login
// not sync shared password to recent
if !password.is_empty()
&& password != password0
&& !self.password_source.is_shared_ab(&password, &hash)
{
config.password = password.clone();
log::debug!("remember password of {}", self.id);
}
} else if !self.skips_saved_password() {
if self.password_source.is_personal_ab(&password) {
// sync personal ab password to recent automatically
config.password = password.clone();
log::debug!("save ab password of {} to recent", self.id);
} else if !password0.is_empty() {
config.password = Default::default();
log::debug!("remove password of {}", self.id);
}
}
if let Some((_, b, c)) = self.other_server.as_ref() {
if b != PUBLIC_SERVER {
config
.options
.insert("other-server-key".to_owned(), c.clone());
}
}
// 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());
}
#[cfg(feature = "flutter")]
{
// sync connected password to personal ab automatically if it is not shared password
if !self.skips_saved_password()
&& !config.password.is_empty()
&& !self.password_source.is_shared_ab(&password, &hash)
&& !self.password_source.is_personal_ab(&password)
{
let hash = base64::encode(config.password.clone(), base64::Variant::Original);
let evt: HashMap<&str, String> = HashMap::from([
("name", "sync_peer_hash_password_to_personal_ab".to_string()),
("id", self.id.clone()),
("hash", hash),
]);
let evt = serde_json::ser::to_string(&evt).unwrap_or("".to_owned());
crate::flutter::push_global_event(crate::flutter::APP_TYPE_MAIN, evt);
}
}
if config.keyboard_mode.is_empty() {
if is_keyboard_mode_supported(
&KeyboardMode::Map,
get_version_number(&pi.version),
&pi.platform,
) {
config.keyboard_mode = KeyboardMode::Map.to_string();
} else {
config.keyboard_mode = KeyboardMode::Legacy.to_string();
}
} else {
let keyboard_modes =
crate::get_supported_keyboard_modes(get_version_number(&pi.version), &pi.platform);
let current_mode = &KeyboardMode::from_str(&config.keyboard_mode).unwrap_or_default();
if !keyboard_modes.contains(current_mode) {
config.keyboard_mode = KeyboardMode::Legacy.to_string();
}
}
// no matter if change, for update file time
self.save_config(config);
self.supported_encoding = pi.encoding.clone().unwrap_or_default();
log::info!("peer info supported_encoding:{:?}", self.supported_encoding);
}
pub fn get_remote_dir(&self) -> String {
serde_json::from_str::<HashMap<String, String>>(&self.get_option("remote_dir"))
.unwrap_or_default()
.remove(&self.info.username)
.unwrap_or_default()
}
pub fn get_all_remote_dir(&self, path: String) -> String {
let d = self.get_option("remote_dir");
let user = self.info.username.clone();
let mut x = serde_json::from_str::<HashMap<String, String>>(&d).unwrap_or_default();
if path.is_empty() {
x.remove(&user);
} else {
x.insert(user, path);
}
serde_json::to_string::<HashMap<String, String>>(&x).unwrap_or_default()
}
/// Create a [`Message`] for login.
fn create_login_msg(
&self,
os_username: String,
os_password: String,
password: Vec<u8>,
) -> Message {
let my_id = Config::get_id();
let (my_id, pure_id) = if let Some((id, _, _)) = self.other_server.as_ref() {
let server = Config::get_rendezvous_server();
(format!("{my_id}@{server}"), id.clone())
} else {
(my_id, self.id.clone())
};
let mut avatar = get_builtin_option(keys::OPTION_AVATAR);
if avatar.is_empty() {
avatar = serde_json::from_str::<serde_json::Value>(&LocalConfig::get_option(
"user_info",
))
.ok()
.and_then(|x| {
x.get("avatar")
.and_then(|x| x.as_str())
.map(|x| x.trim().to_owned())
})
.unwrap_or_default();
}
avatar = resolve_avatar_url(avatar);
let mut display_name = get_builtin_option(keys::OPTION_DISPLAY_NAME);
if display_name.is_empty() {
display_name =
serde_json::from_str::<serde_json::Value>(&LocalConfig::get_option("user_info"))
.map(|x| {
x.get("display_name")
.and_then(|x| x.as_str())
.map(|x| x.trim())
.filter(|x| !x.is_empty())
.or_else(|| x.get("name").and_then(|x| x.as_str()))
.map(|x| x.to_owned())
.unwrap_or_default()
})
.unwrap_or_default();
}
if display_name.is_empty() {
display_name = crate::username();
}
let display_name = display_name
.split_whitespace()
.map(|word| {
word.chars()
.enumerate()
.map(|(i, c)| {
if i == 0 {
c.to_uppercase().to_string()
} else {
c.to_string()
}
})
.collect::<String>()
})
.collect::<Vec<_>>()
.join(" ");
#[cfg(not(target_os = "android"))]
let my_platform = hbb_common::whoami::platform().to_string();
#[cfg(target_os = "android")]
let my_platform = "Android".into();
let hwid = if self.get_option("trust-this-device") == "Y" {
crate::get_hwid()
} else {
Bytes::new()
};
let os_login: MessageField<OSLogin> = if self.conn_type == ConnType::TERMINAL {
Some(OSLogin {
username: os_username,
password: os_password,
..Default::default()
})
.into()
} else {
Default::default()
};
let mut lr = LoginRequest {
username: pure_id,
password: password.into(),
my_id,
my_name: display_name,
my_platform,
option: self.get_option_message(true).into(),
session_id: self.session_id,
version: crate::VERSION.to_string(),
os_login,
hwid,
avatar,
..Default::default()
};
if self.skips_saved_password() {
lr.password.clear();
}
match self.conn_type {
ConnType::FILE_TRANSFER => lr.set_file_transfer(FileTransfer {
dir: self.get_remote_dir(),
show_hidden: !self.get_option("remote_show_hidden").is_empty(),
..Default::default()
}),
ConnType::VIEW_CAMERA => lr.set_view_camera(Default::default()),
ConnType::PORT_FORWARD | ConnType::RDP => lr.set_port_forward(PortForward {
host: self.port_forward.0.clone(),
port: self.port_forward.1,
multiplex: self.port_forward_multiplex,
..Default::default()
}),
ConnType::TERMINAL => {
let mut terminal = Terminal::new();
terminal.service_id = self.get_option(self.get_key_terminal_service_id());
lr.set_terminal(terminal);
}
_ => {}
}
let mut msg_out = Message::new();
msg_out.set_login_request(lr);
msg_out
}
pub fn update_supported_decodings(&self) -> Message {
let decoding = scrap::codec::Decoder::supported_decodings(
Some(&self.id),
use_texture_render(),
self.adapter_luid,
&self.mark_unsupported,
);
let mut misc = Misc::new();
misc.set_option(OptionMessage {
supported_decoding: hbb_common::protobuf::MessageField::some(decoding),
..Default::default()
});
let mut msg_out = Message::new();
msg_out.set_misc(misc);
msg_out
}
pub fn restart_remote_device(&self) -> Message {
let mut misc = Misc::new();
misc.set_restart_remote_device(true);
let mut msg_out = Message::new();
msg_out.set_misc(misc);
msg_out
}
pub fn mark_restarting_remote_device(&mut self) {
self.restarting_remote_device = true;
self.restart_remote_device_at = Some(Instant::now());
}
pub fn clear_restarting_remote_device(&mut self) {
self.restarting_remote_device = false;
self.restart_remote_device_at = None;
}
pub fn is_restarting_remote_device(&self) -> bool {
if !self.restarting_remote_device {
return false;
}
// Keep this flag alive for a short grace window instead of clearing it on
// connection_ready or the first peer bytes. During OS restart the peer can
// briefly reconnect before the real reboot disconnect, and clearing it too
// early would let the next disconnect escape the restart flow and fall back
// to the normal error dialog / manual reconnect path.
self.restart_remote_device_at
.map(|started_at| started_at.elapsed() < RESTART_REMOTE_DEVICE_GRACE)
.unwrap_or(false)
}
pub fn get_conn_token(&self) -> Option<String> {
if self.password.is_empty() {
return None;
}
serde_json::to_string(&ConnToken {
password: self.password.clone(),
password_source: self.password_source.clone(),
session_id: self.session_id,
})
.ok()
}
pub fn get_id(&self) -> &str {
&self.id
}
pub fn get_key_terminal_service_id(&self) -> &'static str {
if self.is_terminal_admin {
"terminal-admin-service-id"
} else {
"terminal-service-id"
}
}
}
/// Media data.
pub enum MediaData {
VideoQueue,
VideoFrame(Box<VideoFrame>),
AudioFrame(Box<AudioFrame>),
AudioFormat(AudioFormat),
Reset,
RecordScreen(bool),
}
pub type MediaSender = mpsc::Sender<MediaData>;
/// Start video thread.
///
/// # Arguments
///
/// * `video_callback` - The callback for video frame. Being called when a video frame is ready.
pub fn start_video_thread<F, T>(
session: Session<T>,
display: usize,
video_receiver: mpsc::Receiver<MediaData>,
video_queue: Arc<RwLock<ArrayQueue<VideoFrame>>>,
fps: Arc<RwLock<Option<usize>>>,
chroma: Arc<RwLock<Option<Chroma>>>,
discard_queue: Arc<RwLock<bool>>,
video_callback: F,
) where
F: 'static + FnMut(usize, &mut scrap::ImageRgb, *mut c_void, bool) + Send,
T: InvokeUiSession,
{
let mut video_callback = video_callback;
let mut last_chroma = None;
let is_view_camera = session.is_view_camera();
std::thread::spawn(move || {
#[cfg(windows)]
sync_cpu_usage();
get_hwcodec_config();
let mut video_handler = None;
let mut count = 0;
let mut duration = std::time::Duration::ZERO;
let mut skip_beginning = 0;
loop {
if let Ok(data) = video_receiver.recv() {
match data {
MediaData::VideoFrame(_) | MediaData::VideoQueue => {
let vf = match data {
MediaData::VideoFrame(vf) => {
*discard_queue.write().unwrap() = false;
*vf
}
MediaData::VideoQueue => {
if let Some(vf) = video_queue.read().unwrap().pop() {
if discard_queue.read().unwrap().clone() {
continue;
}
vf
} else {
continue;
}
}
_ => {
// unreachable!();
continue;
}
};
let display = vf.display as usize;
let start = std::time::Instant::now();
let format = CodecFormat::from(&vf);
if video_handler.is_none() {
let mut handler = VideoHandler::new(format, display);
let record_state = session.lc.read().unwrap().record_state;
let record_permission = session.lc.read().unwrap().record_permission;
let id = session.lc.read().unwrap().id.clone();
if record_state && record_permission {
handler.record_screen(true, id, display, is_view_camera);
}
video_handler = Some(handler);
}
if let Some(handler) = video_handler.as_mut() {
let mut pixelbuffer = true;
let mut tmp_chroma = None;
let format_changed = handler.decoder.format() != format;
match handler.handle_frame(vf, &mut pixelbuffer, &mut tmp_chroma) {
Ok(true) => {
video_callback(
display,
&mut handler.rgb,
handler.texture.texture,
pixelbuffer,
);
// chroma
if tmp_chroma.is_some() && last_chroma != tmp_chroma {
last_chroma = tmp_chroma;
*chroma.write().unwrap() = tmp_chroma;
}
// fps calculation
fps_calculate(
&mut skip_beginning,
&fps,
format_changed,
start.elapsed(),
&mut count,
&mut duration,
);
}
Err(e) => {
// This is a simple workaround.
//
// I only see the following error:
// FailedCall("errcode=1 scrap::common::vpxcodec:libs\\scrap\\src\\common\\vpxcodec.rs:433:9")
// When switching from all displays to one display, the error occurs.
// eg:
// 1. Connect to a device with two displays (A and B).
// 2. Switch to display A. The error occurs.
// 3. If the error does not occur. Switch from A to display B. The error occurs.
//
// to-do: fix the error
log::error!("handle video frame error, {}", e);
session.refresh_video(display as _);
}
_ => {}
}
}
// check invalid decoders
let mut should_update_supported = false;
if let Some(handler) = video_handler.as_mut() {
if !handler.decoder.valid()
|| handler.fail_counter >= MAX_DECODE_FAIL_COUNTER
{
let mut lc = session.lc.write().unwrap();
let format = handler.decoder.format();
if !lc.mark_unsupported.contains(&format) {
lc.mark_unsupported.push(format);
should_update_supported = true;
log::info!("mark {format:?} decoder as unsupported, valid:{}, fail_counter:{}, all unsupported:{:?}", handler.decoder.valid(), handler.fail_counter, lc.mark_unsupported);
}
}
}
if should_update_supported {
session.send(Data::Message(
session.lc.read().unwrap().update_supported_decodings(),
));
}
}
MediaData::Reset => {
if let Some(handler) = video_handler.as_mut() {
handler.reset(None);
}
}
MediaData::RecordScreen(start) => {
let id = session.lc.read().unwrap().id.clone();
if let Some(handler) = video_handler.as_mut() {
handler.record_screen(start, id, display, is_view_camera);
}
}
_ => {}
}
} else {
break;
}
}
log::info!("Video decoder loop exits");
});
}
/// Start an audio thread
/// Return a audio [`MediaSender`]
pub fn start_audio_thread() -> MediaSender {
let (audio_sender, audio_receiver) = mpsc::channel::<MediaData>();
std::thread::spawn(move || {
let mut audio_handler = AudioHandler::default();
loop {
#[cfg(target_os = "windows")]
let received = audio_handler.receive_audio(&audio_receiver);
#[cfg(not(target_os = "windows"))]
let received = audio_receiver.recv();
if let Ok(data) = received {
match data {
MediaData::AudioFrame(af) => {
audio_handler.handle_frame(*af);
}
MediaData::AudioFormat(f) => {
log::debug!("recved audio format, sample rate={}", f.sample_rate);
audio_handler.handle_format(f);
}
_ => {}
}
} else {
break;
}
}
log::info!("Audio decoder loop exits");
});
audio_sender
}
#[inline]
fn fps_calculate(
skip_beginning: &mut usize,
fps: &Arc<RwLock<Option<usize>>>,
format_changed: bool,
elapsed: std::time::Duration,
count: &mut usize,
duration: &mut std::time::Duration,
) {
if format_changed {
*count = 0;
*duration = std::time::Duration::ZERO;
*skip_beginning = 0;
}
// // The first frame will be very slow
if *skip_beginning < 3 {
*skip_beginning += 1;
return;
}
*duration += elapsed;
*count += 1;
let ms = duration.as_millis();
if *count % 10 == 0 && ms > 0 {
*fps.write().unwrap() = Some((*count as usize) * 1000 / (ms as usize));
}
// Clear to get real-time fps
if *count >= 30 {
*count = 0;
*duration = Duration::ZERO;
}
}
fn get_hwcodec_config() {
// for sciter and unilink
#[cfg(feature = "hwcodec")]
#[cfg(any(target_os = "windows", target_os = "linux"))]
{
use std::sync::Once;
static ONCE: Once = Once::new();
ONCE.call_once(|| {
let start = std::time::Instant::now();
if let Err(e) = crate::ipc::get_hwcodec_config_from_server() {
log::error!(
"Failed to get hwcodec config: {e:?}, elapsed: {:?}",
start.elapsed()
);
} else {
log::info!("{:?} used to get hwcodec config", start.elapsed());
}
});
}
}
#[cfg(windows)]
fn sync_cpu_usage() {
use std::sync::Once;
static ONCE: Once = Once::new();
ONCE.call_once(|| {
let t = std::thread::spawn(do_sync_cpu_usage);
t.join().ok();
});
}
#[cfg(windows)]
#[tokio::main(flavor = "current_thread")]
async fn do_sync_cpu_usage() {
use crate::ipc::{connect, Data};
let start = std::time::Instant::now();
match connect(50, "").await {
Ok(mut conn) => {
if conn.send(&&Data::SyncWinCpuUsage(None)).await.is_ok() {
if let Ok(Some(data)) = conn.next_timeout(50).await {
match data {
Data::SyncWinCpuUsage(cpu_usage) => {
base::platform::windows::sync_cpu_usage(cpu_usage);
}
_ => {}
}
}
}
}
_ => {}
}
log::info!("{:?} used to sync cpu usage", start.elapsed());
}
/// Handle latency test.
///
/// # Arguments
///
/// * `t` - The latency test message.
/// * `peer` - The peer.
pub async fn handle_test_delay(t: TestDelay, peer: &mut Stream) {
if !t.from_client {
let mut msg_out = Message::new();
msg_out.set_test_delay(t);
allow_err!(peer.send(&msg_out).await);
}
}
/// Whether is track pad scrolling.
#[inline]
#[cfg(all(target_os = "macos", not(feature = "flutter")))]
fn check_scroll_on_mac(mask: i32, x: i32, y: i32) -> bool {
// flutter version we set mask type bit to 4 when track pad scrolling.
if mask & 7 == crate::input::MOUSE_TYPE_TRACKPAD {
return true;
}
if mask & 3 != crate::input::MOUSE_TYPE_WHEEL {
return false;
}
let btn = mask >> 3;
if y == -1 {
btn != 0xff88 && btn != -0x780000
} else if y == 1 {
btn != 0x78 && btn != 0x780000
} else if x != 0 {
// No mouse support horizontal scrolling.
true
} else {
false
}
}
/// Send mouse data.
///
/// # Arguments
///
/// * `mask` - Mouse event.
/// * mask = buttons << 3 | type
/// * type, 1: down, 2: up, 3: wheel, 4: trackpad
/// * buttons, 1: left, 2: right, 4: middle
/// * `x` - X coordinate.
/// * `y` - Y coordinate.
/// * `alt` - Whether the alt key is pressed.
/// * `ctrl` - Whether the ctrl key is pressed.
/// * `shift` - Whether the shift key is pressed.
/// * `command` - Whether the command key is pressed.
/// * `interface` - The interface for sending data.
#[inline]
pub fn send_mouse(
mask: i32,
x: i32,
y: i32,
alt: bool,
ctrl: bool,
shift: bool,
command: bool,
interface: &impl Interface,
) {
let mut msg_out = Message::new();
let mut mouse_event = MouseEvent {
mask,
x,
y,
..Default::default()
};
if alt {
mouse_event.modifiers.push(ControlKey::Alt.into());
}
if shift {
mouse_event.modifiers.push(ControlKey::Shift.into());
}
if ctrl {
mouse_event.modifiers.push(ControlKey::Control.into());
}
if command {
mouse_event.modifiers.push(ControlKey::Meta.into());
}
#[cfg(all(target_os = "macos", not(feature = "flutter")))]
if check_scroll_on_mac(mask, x, y) {
let factor = 3;
mouse_event.mask = crate::input::MOUSE_TYPE_TRACKPAD;
mouse_event.x *= factor;
mouse_event.y *= factor;
}
interface.swap_modifier_mouse(&mut mouse_event);
msg_out.set_mouse_event(mouse_event);
interface.send(Data::Message(msg_out));
}
#[inline]
pub fn send_pointer_device_event(
mut evt: PointerDeviceEvent,
alt: bool,
ctrl: bool,
shift: bool,
command: bool,
interface: &impl Interface,
) {
let mut msg_out = Message::new();
if alt {
evt.modifiers.push(ControlKey::Alt.into());
}
if shift {
evt.modifiers.push(ControlKey::Shift.into());
}
if ctrl {
evt.modifiers.push(ControlKey::Control.into());
}
if command {
evt.modifiers.push(ControlKey::Meta.into());
}
msg_out.set_pointer_device_event(evt);
interface.send(Data::Message(msg_out));
}
/// Activate OS by sending mouse movement.
///
/// # Arguments
///
/// * `interface` - The interface for sending data.
/// * `send_left_click` - Whether to send a click event.
fn activate_os(interface: &impl Interface, send_left_click: bool) {
let left_down = MOUSE_BUTTON_LEFT << 3 | MOUSE_TYPE_DOWN;
let left_up = MOUSE_BUTTON_LEFT << 3 | MOUSE_TYPE_UP;
let right_down = MOUSE_BUTTON_RIGHT << 3 | MOUSE_TYPE_DOWN;
let right_up = MOUSE_BUTTON_RIGHT << 3 | MOUSE_TYPE_UP;
send_mouse(left_up, 0, 0, false, false, false, false, interface);
std::thread::sleep(Duration::from_millis(50));
send_mouse(0, 0, 0, false, false, false, false, interface);
std::thread::sleep(Duration::from_millis(50));
send_mouse(0, 3, 3, false, false, false, false, interface);
let (click_down, click_up) = if send_left_click {
(left_down, left_up)
} else {
(right_down, right_up)
};
std::thread::sleep(Duration::from_millis(50));
send_mouse(click_down, 0, 0, false, false, false, false, interface);
send_mouse(click_up, 0, 0, false, false, false, false, interface);
/*
let mut key_event = KeyEvent::new();
// do not use Esc, which has problem with Linux
key_event.set_control_key(ControlKey::RightArrow);
key_event.press = true;
let mut msg_out = Message::new();
msg_out.set_key_event(key_event.clone());
interface.send(Data::Message(msg_out.clone()));
*/
}
/// Input the OS's password.
///
/// # Arguments
///
/// * `p` - The password.
/// * `activate` - Whether to activate OS.
/// * `interface` - The interface for sending data.
pub fn input_os_password(p: String, activate: bool, interface: impl Interface) {
std::thread::spawn(move || {
_input_os_password(p, activate, interface);
});
}
/// Input the OS's password.
///
/// # Arguments
///
/// * `p` - The password.
/// * `activate` - Whether to activate OS.
/// * `interface` - The interface for sending data.
fn _input_os_password(p: String, activate: bool, interface: impl Interface) {
let input_password = !p.is_empty();
if activate {
// Click event is used to bring up the password input box.
activate_os(&interface, input_password);
std::thread::sleep(Duration::from_millis(1200));
}
if !input_password {
return;
}
let mut key_event = KeyEvent::new();
key_event.mode = KeyboardMode::Legacy.into();
key_event.press = true;
let mut msg_out = Message::new();
key_event.set_seq(p);
msg_out.set_key_event(key_event.clone());
interface.send(Data::Message(msg_out.clone()));
key_event.set_control_key(ControlKey::Return);
msg_out.set_key_event(key_event);
interface.send(Data::Message(msg_out));
}
/// Auto Login. A peer that can check for its logon or lock screen gets the password and types it
/// itself, only if it is there right then. Keys typed from here land in whatever has focus if the
/// screen is not locked by the time they arrive.
pub fn auto_login(p: String, pi: &PeerInfo, interface: impl Interface) {
if !is_peer_handling_os_login(pi) {
input_os_password(p, true, interface);
return;
}
std::thread::spawn(move || {
activate_os(&interface, true);
std::thread::sleep(Duration::from_millis(1200));
let mut msg_out = Message::new();
msg_out.set_os_login(OSLogin {
password: p,
..Default::default()
});
interface.send(Data::Message(msg_out));
});
}
/// Windows and macOS peers from the 1.5.1 release handle `os_login`.
fn is_peer_handling_os_login(pi: &PeerInfo) -> bool {
(pi.platform == crate::common::PLATFORM_WINDOWS || pi.platform == crate::common::PLATFORM_MACOS)
&& hbb_common::get_version_number(&pi.version) >= hbb_common::get_version_number("1.5.1")
}
#[derive(Copy, Clone)]
struct LoginErrorMsgBox {
msgtype: &'static str,
title: &'static str,
text: &'static str,
link: &'static str,
try_again: bool,
}
lazy_static::lazy_static! {
static ref LOGIN_ERROR_MAP: Arc<HashMap<&'static str, LoginErrorMsgBox>> = {
let map = HashMap::from([(LOGIN_SCREEN_WAYLAND, LoginErrorMsgBox{
msgtype: "error",
title: "Login Error",
text: "Login screen using Wayland is not supported",
link: "https://rustdesk.com/docs/en/manual/linux/#login-screen",
try_again: true,
}), (LOGIN_MSG_NO_PASSWORD_ACCESS, LoginErrorMsgBox{
msgtype: "wait-remote-accept-nook",
title: "Prompt",
text: "Please wait for the remote side to accept your session request...",
link: "",
try_again: true,
})]);
Arc::new(map)
};
}
/// Handle login error.
/// Return true if the password is wrong, return false if there's an actual error.
pub fn handle_login_error(
lc: Arc<RwLock<LoginConfigHandler>>,
err: &str,
interface: &impl Interface,
) -> bool {
if err == LOGIN_MSG_PASSWORD_EMPTY {
lc.write().unwrap().password = Default::default();
interface.msgbox("input-password", "Password Required", "", "");
true
} else if err == LOGIN_MSG_PASSWORD_WRONG {
lc.write().unwrap().password = Default::default();
interface.msgbox("re-input-password", err, "Do you want to enter again?", "");
true
} else if err == LOGIN_MSG_2FA_WRONG || err == REQUIRE_2FA {
let enabled = lc.read().unwrap().get_option("trust-this-device") == "Y";
if enabled {
lc.write()
.unwrap()
.set_option("trust-this-device".to_string(), "".to_string());
}
interface.msgbox("input-2fa", err, "", "");
true
} else if LOGIN_ERROR_MAP.contains_key(err) {
if let Some(msgbox_info) = LOGIN_ERROR_MAP.get(err) {
interface.msgbox(
msgbox_info.msgtype,
msgbox_info.title,
msgbox_info.text,
msgbox_info.link,
);
msgbox_info.try_again
} else {
// unreachable!
false
}
} else {
if err.contains(SCRAP_X11_REQUIRED) {
interface.msgbox("error", "Login Error", err, SCRAP_X11_REF_URL);
} else {
interface.msgbox("error", "Login Error", err, "");
}
false
}
}
// "Switch sides" requires the incoming-only client to connect back to its
// controlling peer; verify the local pending uuid before opening the connection.
#[cfg(feature = "flutter")]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
async fn is_switch_sides_back(conn_type: ConnType, interface: &impl Interface) -> bool {
if conn_type != ConnType::DEFAULT_CONN {
return false;
}
let (id, uuid) = {
let lch = interface.get_lch();
let lc = lch.read().unwrap();
let Some(uuid) = lc.switch_uuid.as_deref() else {
return false;
};
let Ok(uuid) = Uuid::parse_str(uuid) else {
return false;
};
(lc.id.clone(), uuid)
};
if !request_local_switch_sides_uuid(
&id,
&uuid,
crate::ipc::SwitchSidesUuidAction::Check,
)
.await
{
return false;
}
let lch = interface.get_lch();
let lc = lch.read().unwrap();
let current_uuid = lc
.switch_uuid
.as_deref()
.and_then(|value| Uuid::parse_str(value).ok());
lc.id == id && current_uuid.as_ref() == Some(&uuid)
}
#[cfg(not(all(feature = "flutter", not(any(target_os = "android", target_os = "ios")))))]
async fn is_switch_sides_back(_conn_type: ConnType, _interface: &impl Interface) -> bool {
false
}
#[cfg(feature = "flutter")]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
async fn request_local_switch_sides_uuid(
id: &str,
uuid: &Uuid,
action: crate::ipc::SwitchSidesUuidAction,
) -> bool {
let Ok(mut conn) = crate::ipc::connect(1000, "").await else {
return false;
};
let uuid = uuid.to_string();
if conn
.send(&crate::ipc::Data::SwitchSidesUuid(
uuid.clone(),
id.to_owned(),
action,
None,
))
.await
.is_err()
{
return false;
}
match conn.next_timeout(1000).await {
Ok(Some(crate::ipc::Data::SwitchSidesUuid(
returned_uuid,
returned_id,
returned_action,
Some(true),
))) => {
returned_uuid == uuid && returned_id == id && returned_action == action
}
_ => false,
}
}
/// Handle hash message sent by peer.
/// Hash will be used for login.
///
/// # Arguments
///
/// * `lc` - Login config.
/// * `hash` - Hash sent by peer.
/// * `interface` - [`Interface`] for sending data.
/// * `peer` - [`Stream`] for communicating with peer.
pub async fn handle_hash(
lc: Arc<RwLock<LoginConfigHandler>>,
password_preset: &str,
hash: Hash,
interface: &impl Interface,
trust: &mut crate::trusted_devices::client::Session,
peer: &mut Stream,
) -> bool {
lc.write().unwrap().hash = hash.clone();
// Take care of password application order
// switch_uuid
#[cfg(feature = "flutter")]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
{
let uuid = lc.write().unwrap().switch_uuid.take();
if let Some(uuid) = uuid {
if let Ok(uuid) = uuid::Uuid::from_str(&uuid) {
let id = lc.read().unwrap().id.clone();
if !request_local_switch_sides_uuid(
&id,
&uuid,
crate::ipc::SwitchSidesUuidAction::Consume,
)
.await
{
log::warn!("Ignored untrusted switch_uuid");
} else {
lc.write().unwrap().allow_switch_back_once();
send_switch_login_request(lc.clone(), peer, uuid).await;
lc.write().unwrap().password_source = Default::default();
return true;
}
}
}
// Incoming-only may connect out solely for a verified switch-back;
// never fall through to password login, including on repeated hashes.
if config::is_incoming_only() {
interface.msgbox("error", "Connection Error", "Incoming only mode", "");
let mut misc = Misc::new();
misc.set_close_reason(
"Connection not allowed in incoming-only mode".to_owned(),
);
let mut msg = Message::new();
msg.set_misc(misc);
allow_err!(peer.send(&msg).await);
return false;
}
}
// last password
let mut password = lc.read().unwrap().password.clone();
// preset password
if password.is_empty() {
if !password_preset.is_empty() {
let mut hasher = Sha256::new();
hasher.update(password_preset);
hasher.update(&hash.salt);
let res = hasher.finalize();
password = res[..].into();
lc.write().unwrap().password_source = Default::default();
}
}
// shared password
// Currently it's used only when click shared ab peer card
let shared_password = lc.write().unwrap().shared_password.take();
if let Some(shared_password) = shared_password {
if !shared_password.is_empty() {
let mut hasher = Sha256::new();
hasher.update(shared_password.clone());
hasher.update(&hash.salt);
let res = hasher.finalize();
password = res[..].into();
lc.write().unwrap().password_source = PasswordSource::SharedAb(shared_password);
}
}
// peer config password
if password.is_empty() {
password = lc.read().unwrap().config.password.clone();
if !password.is_empty() {
lc.write().unwrap().password_source = Default::default();
}
}
// personal ab password
if password.is_empty() {
try_get_password_from_personal_ab(lc.clone(), &mut password);
}
if password.is_empty() {
let p = crate::ui_interface::get_builtin_option(keys::OPTION_DEFAULT_CONNECT_PASSWORD);
if !p.is_empty() {
let mut hasher = Sha256::new();
hasher.update(p.clone());
hasher.update(&hash.salt);
let res = hasher.finalize();
password = res[..].into();
lc.write().unwrap().password_source = PasswordSource::SharedAb(p); // reuse SharedAb here
}
}
lc.write().unwrap().password = password.clone();
let request_approval = lc.read().unwrap().skips_saved_password();
let is_terminal_admin = lc.read().unwrap().is_terminal_admin;
let is_terminal = lc.read().unwrap().conn_type.eq(&ConnType::TERMINAL);
if is_terminal && is_terminal_admin {
if request_approval {
interface.msgbox("terminal-admin-login-request-approval", "", "", "");
} else if password.is_empty() {
interface.msgbox("terminal-admin-login-password", "", "", "");
} else {
interface.msgbox("terminal-admin-login", "", "", "");
}
lc.write().unwrap().hash = hash;
return true;
}
let password = if password.is_empty() {
// login without password, the remote side can click accept
if !request_approval {
interface.msgbox("input-password", "Password Required", "", "");
}
Vec::new()
} else {
let mut hasher = Sha256::new();
hasher.update(&password);
hasher.update(&hash.challenge);
hasher.finalize()[..].into()
};
if request_approval {
interface.msgbox(
"wait-remote-accept-nook",
"Prompt",
"Please wait for the remote side to accept your session request...",
"",
);
}
if let Err(error) = send_login(
lc.clone(),
String::new(),
String::new(),
password,
trust,
peer,
)
.await
{
interface.on_error(&error.to_string());
return false;
}
lc.write().unwrap().hash = hash;
true
}
#[inline]
fn try_get_password_from_personal_ab(lc: Arc<RwLock<LoginConfigHandler>>, password: &mut Vec<u8>) {
let access_token = LocalConfig::get_option("access_token");
let ab = config::Ab::load();
if !access_token.is_empty() && access_token == ab.access_token {
let id = lc.read().unwrap().id.clone();
if let Some(ab) = ab.ab_entries.iter().find(|a| a.personal()) {
if let Some(p) = ab
.peers
.iter()
.find_map(|p| if p.id == id { Some(p) } else { None })
{
if let Ok(hash_password) = base64::decode(p.hash.clone(), base64::Variant::Original)
{
if !hash_password.is_empty() {
*password = hash_password.clone();
lc.write().unwrap().password_source =
PasswordSource::PersonalAb(hash_password);
}
}
}
}
}
}
/// Send login message to peer.
///
/// # Arguments
///
/// * `lc` - Login config.
/// * `os_username` - OS username.
/// * `os_password` - OS password.
/// * `password` - Password.
/// * `peer` - [`Stream`] for communicating with peer.
async fn send_login(
lc: Arc<RwLock<LoginConfigHandler>>,
os_username: String,
os_password: String,
password: Vec<u8>,
trust: &mut crate::trusted_devices::client::Session,
peer: &mut Stream,
) -> ResultType<()> {
let msg_out = lc
.read()
.unwrap()
.create_login_msg(os_username, os_password, password);
let msg_out = trust.login(msg_out).await?;
peer.send(&msg_out).await
}
/// Handle login request made from ui.
///
/// # Arguments
///
/// * `lc` - Login config.
/// * `os_username` - OS username.
/// * `os_password` - OS password.
/// * `password` - Password.
/// * `remember` - Whether to remember password.
/// * `peer` - [`Stream`] for communicating with peer.
pub async fn handle_login_from_ui(
lc: Arc<RwLock<LoginConfigHandler>>,
os_username: String,
os_password: String,
password: String,
remember: bool,
trust: &mut crate::trusted_devices::client::Session,
peer: &mut Stream,
) -> ResultType<()> {
let mut hash_password = if password.is_empty() {
let mut password2 = lc.read().unwrap().password.clone();
if password2.is_empty() {
password2 = lc.read().unwrap().config.password.clone();
if !password2.is_empty() {
lc.write().unwrap().password_source = Default::default();
}
}
password2
} else {
lc.write().unwrap().password_source = Default::default();
let mut hasher = Sha256::new();
hasher.update(password);
hasher.update(&lc.read().unwrap().hash.salt);
let res = hasher.finalize();
lc.write().unwrap().remember = remember;
res[..].into()
};
lc.write().unwrap().password = hash_password.clone();
let mut hasher2 = Sha256::new();
hasher2.update(&hash_password[..]);
hasher2.update(&lc.read().unwrap().hash.challenge);
hash_password = hasher2.finalize()[..].to_vec();
send_login(
lc.clone(),
os_username,
os_password,
hash_password,
trust,
peer,
)
.await
}
async fn send_switch_login_request(
lc: Arc<RwLock<LoginConfigHandler>>,
peer: &mut Stream,
uuid: Uuid,
) {
let mut msg_out = Message::new();
msg_out.set_switch_sides_response(SwitchSidesResponse {
uuid: Bytes::from(uuid.as_bytes().to_vec()),
lr: hbb_common::protobuf::MessageField::some(
lc.read()
.unwrap()
.create_login_msg("".to_owned(), "".to_owned(), vec![])
.login_request()
.to_owned(),
),
..Default::default()
});
allow_err!(peer.send(&msg_out).await);
}
/// Interface for client to send data and commands.
#[async_trait]
pub trait Interface: Send + Clone + 'static + Sized {
/// Send message data to remote peer.
fn send(&self, data: Data);
fn msgbox(&self, msgtype: &str, title: &str, text: &str, link: &str);
fn handle_login_error(&self, err: &str) -> bool;
fn handle_peer_info(&self, pi: PeerInfo);
fn set_multiple_windows_session(&self, sessions: Vec<WindowsSession>);
fn on_error(&self, err: &str) {
self.msgbox("error", "Error", err, "");
}
async fn handle_hash(
&self,
pass: &str,
hash: Hash,
trust: &mut crate::trusted_devices::client::Session,
peer: &mut Stream,
) -> bool;
async fn handle_login_from_ui(
&self,
os_username: String,
os_password: String,
password: String,
remember: bool,
trust: &mut crate::trusted_devices::client::Session,
peer: &mut Stream,
);
async fn handle_test_delay(&self, t: TestDelay, peer: &mut Stream);
fn get_lch(&self) -> Arc<RwLock<LoginConfigHandler>>;
fn get_id(&self) -> String {
self.get_lch().read().unwrap().id.clone()
}
fn is_force_relay(&self) -> bool {
self.get_lch().read().unwrap().force_relay
}
fn is_policy_relay(&self) -> bool {
self.get_lch().read().unwrap().policy_relay
}
fn get_switch_code(&self) -> String {
match self.get_lch().read().unwrap().switch_uuid.clone() {
Some(u) if !u.is_empty() => {
use hbb_common::sodiumoxide::crypto::hash::sha256;
crate::encode64(sha256::hash(u.as_bytes()).0)
}
_ => String::new(),
}
}
fn swap_modifier_mouse(&self, _msg: &mut base::protos::message::MouseEvent) {}
fn update_direct(&self, direct: Option<bool>) {
self.get_lch().write().unwrap().direct = direct;
}
fn update_received(&self, received: bool) {
self.get_lch().write().unwrap().received = received;
}
fn on_establish_connection_error(&self, err: String) {
let title = "Connection Error";
let text = err.to_string();
let lch = self.get_lch();
let (is_restarting, direct, received) = {
let lc = lch.read().unwrap();
(lc.is_restarting_remote_device(), lc.direct, lc.received)
};
if is_restarting {
log::info!("Restart remote device, suppress connection error: {err}");
// Flutter treats this as a reconnect control event. The text is kept
// for legacy UI and existing translation reuse.
self.msgbox("restarting", "Restarting remote device", "Connection in progress. Please wait.", "");
return;
}
let mut relay_hint = false;
let mut relay_hint_type = "relay-hint";
// force relay
let errno = errno::errno().0;
log::error!("Connection closed: {err}({errno})");
if direct == Some(true)
&& ((cfg!(windows) && (errno == 10054 || err.contains("10054")))
|| (!cfg!(windows) && (errno == 104 || err.contains("104")))
|| (!err.contains("Failed") && err.contains("deadline")))
// deadline: https://github.com/rustdesk/rustdesk-server-pro/discussions/325, most likely comes from secure tcp timeout
{
relay_hint = true;
if !received {
relay_hint_type = "relay-hint2"
}
}
// relay-hint
if cfg!(feature = "flutter") && relay_hint {
self.msgbox(relay_hint_type, title, &text, "");
} else {
self.msgbox("error", title, &text, "");
}
}
}
/// Data used by the client interface.
#[derive(Clone)]
pub enum Data {
Close,
RejectInsecureConnection,
Login((String, String, String, bool)),
Message(Message),
#[cfg(not(any(target_os = "android", target_os = "ios")))]
InitialClipboard(usize, Option<Message>),
SendFiles((i32, JobType, String, String, i32, bool, bool)),
RemoveDirAll((i32, String, bool, bool)),
ConfirmDeleteFiles((i32, i32)),
SetNoConfirm(i32),
RemoveDir((i32, String)),
RemoveFile((i32, String, i32, bool)),
CreateDir((i32, String, bool)),
CancelJob(i32),
RemovePortForward(i32),
AddPortForward((i32, String, i32)),
#[cfg(all(target_os = "windows", not(feature = "flutter")))]
ToggleClipboardFile,
NewRDP,
SetConfirmOverrideFile((i32, i32, bool, bool, bool)),
AddJob((i32, JobType, String, String, i32, bool, bool)),
ResumeJob((i32, bool)),
RecordScreen(bool),
ElevateDirect,
ElevateWithLogon(String, String),
NewVoiceCall,
CloseVoiceCall,
ContinueInsecureConnection,
ResetDecoder(Option<usize>),
RenameFile((i32, String, String, bool)),
TakeScreenshot((i32, String)),
}
pub async fn confirm_insecure_connection(
interface: &impl Interface,
receiver: &mut UnboundedReceiver<Data>,
) -> bool {
interface.msgbox(
"insecure-connection-nocancel-hasclose",
"Insecure Connection",
"conn-e2ee-unavailable-tip",
"",
);
while let Some(data) = receiver.recv().await {
match data {
Data::ContinueInsecureConnection => return true,
Data::RejectInsecureConnection => return false,
Data::Close => return false,
_ => {}
}
}
false
}
/// Keycode for key events.
#[derive(Clone, Debug)]
pub enum Key {
ControlKey(ControlKey),
Chr(u32),
_Raw(u32),
}
lazy_static::lazy_static! {
pub static ref KEY_MAP: HashMap<&'static str, Key> =
[
("VK_A", Key::Chr('a' as _)),
("VK_B", Key::Chr('b' as _)),
("VK_C", Key::Chr('c' as _)),
("VK_D", Key::Chr('d' as _)),
("VK_E", Key::Chr('e' as _)),
("VK_F", Key::Chr('f' as _)),
("VK_G", Key::Chr('g' as _)),
("VK_H", Key::Chr('h' as _)),
("VK_I", Key::Chr('i' as _)),
("VK_J", Key::Chr('j' as _)),
("VK_K", Key::Chr('k' as _)),
("VK_L", Key::Chr('l' as _)),
("VK_M", Key::Chr('m' as _)),
("VK_N", Key::Chr('n' as _)),
("VK_O", Key::Chr('o' as _)),
("VK_P", Key::Chr('p' as _)),
("VK_Q", Key::Chr('q' as _)),
("VK_R", Key::Chr('r' as _)),
("VK_S", Key::Chr('s' as _)),
("VK_T", Key::Chr('t' as _)),
("VK_U", Key::Chr('u' as _)),
("VK_V", Key::Chr('v' as _)),
("VK_W", Key::Chr('w' as _)),
("VK_X", Key::Chr('x' as _)),
("VK_Y", Key::Chr('y' as _)),
("VK_Z", Key::Chr('z' as _)),
("VK_0", Key::Chr('0' as _)),
("VK_1", Key::Chr('1' as _)),
("VK_2", Key::Chr('2' as _)),
("VK_3", Key::Chr('3' as _)),
("VK_4", Key::Chr('4' as _)),
("VK_5", Key::Chr('5' as _)),
("VK_6", Key::Chr('6' as _)),
("VK_7", Key::Chr('7' as _)),
("VK_8", Key::Chr('8' as _)),
("VK_9", Key::Chr('9' as _)),
("VK_COMMA", Key::Chr(',' as _)),
("VK_SLASH", Key::Chr('/' as _)),
("VK_SEMICOLON", Key::Chr(';' as _)),
("VK_QUOTE", Key::Chr('\'' as _)),
("VK_LBRACKET", Key::Chr('[' as _)),
("VK_RBRACKET", Key::Chr(']' as _)),
("VK_BACKSLASH", Key::Chr('\\' as _)),
("VK_MINUS", Key::Chr('-' as _)),
("VK_PLUS", Key::Chr('=' as _)), // it is =, but sciter return VK_PLUS
("VK_DIVIDE", Key::ControlKey(ControlKey::Divide)), // numpad
("VK_MULTIPLY", Key::ControlKey(ControlKey::Multiply)), // numpad
("VK_SUBTRACT", Key::ControlKey(ControlKey::Subtract)), // numpad
("VK_ADD", Key::ControlKey(ControlKey::Add)), // numpad
("VK_DECIMAL", Key::ControlKey(ControlKey::Decimal)), // numpad
("VK_F1", Key::ControlKey(ControlKey::F1)),
("VK_F2", Key::ControlKey(ControlKey::F2)),
("VK_F3", Key::ControlKey(ControlKey::F3)),
("VK_F4", Key::ControlKey(ControlKey::F4)),
("VK_F5", Key::ControlKey(ControlKey::F5)),
("VK_F6", Key::ControlKey(ControlKey::F6)),
("VK_F7", Key::ControlKey(ControlKey::F7)),
("VK_F8", Key::ControlKey(ControlKey::F8)),
("VK_F9", Key::ControlKey(ControlKey::F9)),
("VK_F10", Key::ControlKey(ControlKey::F10)),
("VK_F11", Key::ControlKey(ControlKey::F11)),
("VK_F12", Key::ControlKey(ControlKey::F12)),
("VK_ENTER", Key::ControlKey(ControlKey::Return)),
("VK_CANCEL", Key::ControlKey(ControlKey::Cancel)),
("VK_BACK", Key::ControlKey(ControlKey::Backspace)),
("VK_TAB", Key::ControlKey(ControlKey::Tab)),
("VK_CLEAR", Key::ControlKey(ControlKey::Clear)),
("VK_RETURN", Key::ControlKey(ControlKey::Return)),
("VK_SHIFT", Key::ControlKey(ControlKey::Shift)),
("VK_CONTROL", Key::ControlKey(ControlKey::Control)),
("VK_MENU", Key::ControlKey(ControlKey::Alt)),
("VK_PAUSE", Key::ControlKey(ControlKey::Pause)),
("VK_CAPITAL", Key::ControlKey(ControlKey::CapsLock)),
("VK_KANA", Key::ControlKey(ControlKey::Kana)),
("VK_HANGUL", Key::ControlKey(ControlKey::Hangul)),
("VK_JUNJA", Key::ControlKey(ControlKey::Junja)),
("VK_FINAL", Key::ControlKey(ControlKey::Final)),
("VK_HANJA", Key::ControlKey(ControlKey::Hanja)),
("VK_KANJI", Key::ControlKey(ControlKey::Kanji)),
("VK_ESCAPE", Key::ControlKey(ControlKey::Escape)),
("VK_CONVERT", Key::ControlKey(ControlKey::Convert)),
("VK_SPACE", Key::ControlKey(ControlKey::Space)),
("VK_PRIOR", Key::ControlKey(ControlKey::PageUp)),
("VK_NEXT", Key::ControlKey(ControlKey::PageDown)),
("VK_END", Key::ControlKey(ControlKey::End)),
("VK_HOME", Key::ControlKey(ControlKey::Home)),
("VK_LEFT", Key::ControlKey(ControlKey::LeftArrow)),
("VK_UP", Key::ControlKey(ControlKey::UpArrow)),
("VK_RIGHT", Key::ControlKey(ControlKey::RightArrow)),
("VK_DOWN", Key::ControlKey(ControlKey::DownArrow)),
("VK_SELECT", Key::ControlKey(ControlKey::Select)),
("VK_PRINT", Key::ControlKey(ControlKey::Print)),
("VK_EXECUTE", Key::ControlKey(ControlKey::Execute)),
("VK_SNAPSHOT", Key::ControlKey(ControlKey::Snapshot)),
("VK_SCROLL", Key::ControlKey(ControlKey::Scroll)),
("VK_INSERT", Key::ControlKey(ControlKey::Insert)),
("VK_DELETE", Key::ControlKey(ControlKey::Delete)),
("VK_HELP", Key::ControlKey(ControlKey::Help)),
("VK_SLEEP", Key::ControlKey(ControlKey::Sleep)),
("VK_SEPARATOR", Key::ControlKey(ControlKey::Separator)),
("VK_NUMPAD0", Key::ControlKey(ControlKey::Numpad0)),
("VK_NUMPAD1", Key::ControlKey(ControlKey::Numpad1)),
("VK_NUMPAD2", Key::ControlKey(ControlKey::Numpad2)),
("VK_NUMPAD3", Key::ControlKey(ControlKey::Numpad3)),
("VK_NUMPAD4", Key::ControlKey(ControlKey::Numpad4)),
("VK_NUMPAD5", Key::ControlKey(ControlKey::Numpad5)),
("VK_NUMPAD6", Key::ControlKey(ControlKey::Numpad6)),
("VK_NUMPAD7", Key::ControlKey(ControlKey::Numpad7)),
("VK_NUMPAD8", Key::ControlKey(ControlKey::Numpad8)),
("VK_NUMPAD9", Key::ControlKey(ControlKey::Numpad9)),
("Apps", Key::ControlKey(ControlKey::Apps)),
("Meta", Key::ControlKey(ControlKey::Meta)),
("RAlt", Key::ControlKey(ControlKey::RAlt)),
("RWin", Key::ControlKey(ControlKey::RWin)),
("RControl", Key::ControlKey(ControlKey::RControl)),
("RShift", Key::ControlKey(ControlKey::RShift)),
("CTRL_ALT_DEL", Key::ControlKey(ControlKey::CtrlAltDel)),
("LOCK_SCREEN", Key::ControlKey(ControlKey::LockScreen)),
].iter().cloned().collect();
}
/// Check if the given message is an error and can be retried.
///
/// # Arguments
///
/// * `msgtype` - The message type.
/// * `title` - The title of the message.
/// * `text` - The text of the message.
#[inline]
pub fn check_if_retry(msgtype: &str, title: &str, text: &str, retry_for_relay: bool) -> bool {
msgtype == "error"
&& title == "Connection Error"
&& ((text.contains("10054") || text.contains("104")) && retry_for_relay
|| (!text.to_lowercase().contains("offline")
&& !text.to_lowercase().contains("not exist")
&& (!text.to_lowercase().contains("handshake")
// https://github.com/snapview/tungstenite-rs/blob/e7e060a89a72cb08e31c25a6c7284dc1bd982e23/src/error.rs#L248
|| text
.to_lowercase()
.contains("connection reset without closing handshake") && use_ws())
&& !text.to_lowercase().contains("failed")
&& !text.to_lowercase().contains("resolve")
&& !text.to_lowercase().contains("mismatch")
&& !text.to_lowercase().contains("manually")
&& !text.to_lowercase().contains("restricted")
&& !text.to_lowercase().contains("incoming only")
&& !text.to_lowercase().contains("not allowed")))
}
#[cfg(test)]
mod retry_tests {
use super::check_if_retry;
#[test]
fn incoming_only_error_is_not_retryable() {
assert!(!check_if_retry(
"error",
"Connection Error",
"Incoming only mode",
false,
));
}
}
pub async fn hc_connection(
feedback: i32,
rendezvous_server: String,
token: &str,
) -> Option<tokio::sync::mpsc::UnboundedSender<()>> {
if feedback == 0 || rendezvous_server.is_empty() || token.is_empty() {
return None;
}
let (tx, rx) = unbounded_channel::<()>();
let token = token.to_owned();
tokio::spawn(async move {
allow_err!(hc_connection_(rendezvous_server, rx, token).await);
});
Some(tx)
}
async fn hc_connection_(
rendezvous_server: String,
mut rx: UnboundedReceiver<()>,
token: String,
) -> ResultType<()> {
let mut timer = crate::rustdesk_interval(interval(crate::TIMER_OUT));
let mut last_recv_msg = Instant::now();
let mut keep_alive = crate::DEFAULT_KEEP_ALIVE;
let host = check_port(&rendezvous_server, RENDEZVOUS_PORT);
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 mut msg_out = RendezvousMessage::new();
msg_out.set_hc(HealthCheck {
token,
..Default::default()
});
conn.send(&msg_out).await?;
loop {
tokio::select! {
res = rx.recv() => {
if res.is_none() {
log::debug!("HC connection is closed as controlling connection exits");
break;
}
}
res = conn.next() => {
last_recv_msg = Instant::now();
let bytes = res.ok_or_else(|| anyhow!("Rendezvous connection is reset by the peer"))??;
if bytes.is_empty() {
conn.send_bytes(bytes::Bytes::new()).await?;
continue; // heartbeat
}
let msg = RendezvousMessage::parse_from_bytes(&bytes)?;
match msg.union {
Some(rendezvous_message::Union::RegisterPkResponse(rpr)) => {
if rpr.keep_alive > 0 {
keep_alive = rpr.keep_alive * 1000;
log::info!("keep_alive: {}ms", keep_alive);
}
}
_ => {}
}
}
_ = timer.tick() => {
// https://www.emqx.com/en/blog/mqtt-keep-alive
if last_recv_msg.elapsed().as_millis() as u64 > keep_alive as u64 * 3 / 2 {
bail!("HC connection is timeout");
}
}
}
}
Ok(())
}
pub mod peer_online {
use hbb_common::{
anyhow::bail,
config::{Config, CONNECT_TIMEOUT, READ_TIMEOUT},
log,
rendezvous_proto::*,
sleep,
socket_client::connect_tcp,
ResultType, Stream,
};
pub async fn query_online_states<F: FnOnce(Vec<String>, Vec<String>)>(ids: Vec<String>, f: F) {
let test = false;
if test {
sleep(1.5).await;
let mut onlines = ids;
let offlines = onlines.drain((onlines.len() / 2)..).collect();
f(onlines, offlines)
} else {
let query_timeout = std::time::Duration::from_millis(3_000);
match query_online_states_(&ids, query_timeout).await {
Ok((onlines, offlines)) => {
f(onlines, offlines);
}
Err(e) => {
log::debug!("query onlines, {}", &e);
}
}
}
}
async fn create_online_stream() -> ResultType<Stream> {
let (rendezvous_server, _servers, _contained) =
crate::get_rendezvous_server(READ_TIMEOUT).await;
let tmp: Vec<&str> = rendezvous_server.split(":").collect();
if tmp.len() != 2 {
bail!("Invalid server address: {}", rendezvous_server);
}
let port: u16 = tmp[1].parse()?;
if port == 0 {
bail!("Invalid server address: {}", rendezvous_server);
}
let online_server = format!("{}:{}", tmp[0], port - 1);
connect_tcp(online_server, CONNECT_TIMEOUT).await
}
async fn query_online_states_(
ids: &Vec<String>,
timeout: std::time::Duration,
) -> ResultType<(Vec<String>, Vec<String>)> {
let mut msg_out = RendezvousMessage::new();
msg_out.set_online_request(OnlineRequest {
id: Config::get_id(),
peers: ids.clone(),
..Default::default()
});
let mut socket = match create_online_stream().await {
Ok(s) => s,
Err(e) => {
log::debug!("Failed to create peers online stream, {e}");
return Ok((vec![], ids.clone()));
}
};
// TODO: Use long connections to avoid socket creation
// If we use a Arc<Mutex<Option<FramedStream>>> to hold and reuse the previous socket,
// we may face the following error:
// An established connection was aborted by the software in your host machine. (os error 10053)
if let Err(e) = socket.send(&msg_out).await {
log::debug!("Failed to send peers online states query, {e}");
return Ok((vec![], ids.clone()));
}
// Retry for 2 times to get the online response
for _ in 0..2 {
if let Some(msg_in) =
crate::get_next_nonkeyexchange_msg(&mut socket, Some(timeout.as_millis() as _))
.await
{
match msg_in.union {
Some(rendezvous_message::Union::OnlineResponse(online_response)) => {
let states = online_response.states;
// One bit per ID, rounded up to whole bytes:
// 0 IDs -> 0 bytes; 1..=8 -> 1; 9..=16 -> 2; 17..=24 -> 3.
let required_len = ids.len().div_ceil(u8::BITS as usize);
if states.len() < required_len {
bail!(
"Invalid online response: expected at least {required_len} state bytes, got {}",
states.len()
);
}
let mut onlines = Vec::new();
let mut offlines = Vec::new();
for i in 0..ids.len() {
// bytes index from left to right
let bit_value = 0x01 << (7 - i % 8);
if (states[i / 8] & bit_value) == bit_value {
onlines.push(ids[i].clone());
} else {
offlines.push(ids[i].clone());
}
}
return Ok((onlines, offlines));
}
_ => {
// ignore
}
}
} else {
// TODO: Make sure socket closed?
bail!("Online stream receives None");
}
}
bail!("Failed to query online states, no online response");
}
#[cfg(test)]
mod tests {
use crate::client::Client;
use hbb_common::rendezvous_proto::IceCandidate;
#[test]
fn accepts_webrtc_ice_by_session_key_only() {
let ice = IceCandidate {
session_key: "session-a".to_owned(),
candidate: "candidate-json".to_owned(),
..Default::default()
};
assert!(Client::is_expected_webrtc_ice_candidate(&ice, "session-a"));
assert!(!Client::is_expected_webrtc_ice_candidate(&ice, "session-b"));
}
}
}
async fn test_udp_uat(
udp_socket: Arc<UdpSocket>,
server_addr: SocketAddr,
udp_port: Arc<Mutex<u16>>,
mut stop_udp_rx: oneshot::Receiver<()>,
) -> ResultType<()> {
// 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 {
..Default::default()
});
// Adaptive retry strategy that works within TCP RTT constraints
// Start with aggressive sending, then back off
let mut retry_interval = Duration::from_millis(20); // Start fast
const MAX_INTERVAL: Duration = Duration::from_millis(200);
let mut packets_sent = 0;
// Send initial burst to improve reliability
let data = msg_out.write_to_bytes()?;
for _ in 0..2 {
if let Err(e) = udp_socket.send_to(&data, server_addr).await {
log::warn!("Failed to send initial UDP NAT test packet: {}", e);
} else {
packets_sent += 1;
}
}
let mut last_send_time = Instant::now();
let mut buf = [0u8; 1500];
loop {
tokio::select! {
_ = &mut stop_udp_rx => {
log::debug!("UDP NAT test received stop signal after {} packets", packets_sent);
break;
}
_ = hbb_common::sleep(retry_interval.as_secs_f32()) => {
// Adaptive retry: send fewer packets as time goes on
let elapsed = last_send_time.elapsed();
if elapsed >= retry_interval {
// Send single packet (not double) to reduce network load
if let Err(e) = udp_socket.send_to(&data, server_addr).await {
log::warn!("Failed to send UDP NAT test retry packet: {}", e);
} else {
packets_sent += 1;
}
// Exponentially increase interval to reduce network pressure
retry_interval = std::cmp::min(
Duration::from_millis((retry_interval.as_millis() as f64 * 1.5) as u64),
MAX_INTERVAL
);
last_send_time = Instant::now();
}
}
res = udp_socket.recv(&mut buf[..]) => {
match res {
Ok(n) => {
match RendezvousMessage::parse_from_bytes(&buf[0..n]) {
Ok(msg_in) => {
if let Some(rendezvous_message::Union::TestNatResponse(response)) = msg_in.union {
*udp_port.lock().unwrap() = response.port as u16;
break;
}
}
Err(e) => {
log::warn!("Failed to parse UDP NAT test response: {}", e);
}
}
}
Err(e) => {
// Same ICMP-driven errors as punch_udp sees. Without a pause this arm
// re-arms recv immediately and spins the loop at CPU speed.
if let Some(n) = UDP_UAT_ERR_LOG.due() {
log::warn!("UDP NAT test socket error x{n}, last: {e}");
}
hbb_common::sleep(0.01).await;
}
}
}
}
}
let final_port = *udp_port.lock().unwrap();
log::debug!(
"UDP NAT test to {:?} finished: time={:?}, port={}, packets_sent={}, success={}",
server_addr,
start.elapsed(),
final_port,
packets_sent,
final_port > 0
);
Ok(())
}
#[inline]
async fn udp_nat_connect(
socket: Arc<UdpSocket>,
typ: &'static str,
ms_timeout: u64,
) -> ResultType<(Stream, Option<KcpStream>, &'static str)> {
crate::punch_udp(socket.clone(), false)
.await
.map_err(|err| {
log::debug!("{err}");
anyhow!(err)
})?;
let res = KcpStream::connect(socket, Duration::from_millis(ms_timeout))
.await
.map_err(|err| {
log::debug!("Failed to connect KCP stream: {}", err);
anyhow!(err)
})?;
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;
// A direct result must win even when it lands first — the LAN ordering, where ICE beats the
// relay's TCP connect. Parking it as if it were a relay commits the relay on arrival.
#[tokio::test]
async fn direct_result_wins_even_when_it_arrives_first() {
let got = race_transports_prefer_webrtc(
ok_after(10, "direct"),
vec![ok_after(120, "relay")],
60_000,
|result| *result == "direct",
)
.await
.unwrap();
assert_eq!(got, "direct");
}
#[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");
}
// The preferred branch runs a whole punch attempt, so it can itself end in a relay. That must
// not preempt a direct punch still in flight on the fallback branch.
#[tokio::test]
async fn relay_from_preferred_branch_does_not_preempt_a_direct_fallback() {
let got = race_transports_prefer_webrtc(
ok_after(10, "preferred-relay"),
vec![ok_after(120, "direct")],
60_000,
|tag| *tag == "direct",
)
.await
.unwrap();
assert_eq!(got, "direct");
}
// ...but it is still committed once nothing direct can arrive.
#[tokio::test]
async fn relay_from_preferred_branch_committed_when_fallback_fails() {
let got = race_transports_prefer_webrtc(
ok_after(10, "preferred-relay"),
vec![err_after(50, "punch failed")],
60_000,
NOT_P2P,
)
.await
.unwrap();
assert_eq!(got, "preferred-relay");
}
#[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));
}
// WebRTC fails first (others still racing), then a relay arrives with a direct attempt still
// unfinished behind it. The relay must not be committed while that direct attempt can win.
#[tokio::test]
async fn relay_does_not_preempt_unfinished_direct_after_webrtc_fails() {
let got = race_transports_prefer_webrtc(
err_after(5, "webrtc dead"),
vec![ok_after(10, "relay"), ok_after(100, "ipv6")],
60_000,
|t| *t == "ipv6",
)
.await
.unwrap();
assert_eq!(got, "ipv6");
}
#[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));
}
// Both legs get the same rendezvous refusal; the UI needs the text verbatim, e.g. the
// offline auto-retry matches it whole. Either leg may fail last.
#[tokio::test]
async fn same_failure_on_both_legs_reported_once() {
for (webrtc_ms, fallback_ms) in [(10, 20), (20, 10)] {
let err = race_transports_prefer_webrtc(
err_after(webrtc_ms, "Remote desktop is offline"),
vec![err_after(fallback_ms, "Remote desktop is offline")],
60_000,
NOT_P2P,
)
.await
.unwrap_err();
assert_eq!(err.to_string(), "Remote desktop is offline");
}
}
}
#[cfg(test)]
#[path = "client/tests/trusted_devices_tests.rs"]
mod trusted_devices_tests;
#[cfg(test)]
mod kx_tests {
use super::*;
use hbb_common::{
sodiumoxide::crypto::box_,
tcp::{Encrypt, FramedStream, KxTranscript, KX_VERSION_LATEST},
};
const PEER_ID: &str = "123456789";
/// What the stand-in controlled peer saw: the version the controller picked, and whether the
/// controller's first application message decrypted.
struct Seen {
picked: u32,
decrypted: bool,
}
fn test_delay(time: i64) -> Message {
let mut msg = Message::new();
msg.set_test_delay(TestDelay {
time,
..Default::default()
});
msg
}
fn delay_in(bytes: &[u8]) -> Option<i64> {
match Message::parse_from_bytes(bytes).ok()?.union? {
message::Union::TestDelay(t) => Some(t.time),
_ => None,
}
}
/// A stand-in controlled peer on loopback. It advertises `advertised` in its signed identity
/// and runs the key under `run`, or under the version the controller picked when `run` is
/// `None`, then sends one message and reads one. Returns the host, its identity as the
/// rendezvous server would sign it, and what it saw.
async fn controlled_stub(
advertised: u32,
run: Option<u32>,
rs_sk: sign::SecretKey,
) -> (String, Vec<u8>, oneshot::Receiver<Seen>) {
let (sign_pk, sign_sk) = sign::gen_keypair();
let id_pk = |pk: &[u8], kx_version| {
IdPk {
id: PEER_ID.to_owned(),
pk: pk.to_vec().into(),
kx_version,
..Default::default()
}
.write_to_bytes()
.unwrap()
};
let signed_id_pk = sign::sign(&id_pk(&sign_pk.0, 0), &rs_sk);
let listener = hbb_common::tcp::new_listener("127.0.0.1:0", false)
.await
.unwrap();
let host = listener.local_addr().unwrap().to_string();
let (tx, rx) = oneshot::channel();
tokio::spawn(async move {
let (stream, addr) = listener.accept().await.unwrap();
let mut s = FramedStream::from(stream, addr);
let (our_pk_b, our_sk_b) = box_::gen_keypair();
let mut msg = Message::new();
msg.set_signed_id(SignedId {
id: sign::sign(&id_pk(&our_pk_b.0, advertised), &sign_sk).into(),
..Default::default()
});
s.send(&msg).await.unwrap();
let bytes = s.next().await.unwrap().unwrap();
let Some(message::Union::PublicKey(pk)) =
Message::parse_from_bytes(&bytes).unwrap().union
else {
panic!("expected the controller's public key");
};
let key =
Encrypt::decode(&pk.symmetric_value, &pk.asymmetric_value, &our_sk_b).unwrap();
let t = KxTranscript {
initiator_pk: &pk.asymmetric_value,
responder_pk: &our_pk_b.0,
advertised,
picked: run.unwrap_or(pk.kx_version),
};
if t.picked == 0 {
s.set_key(key);
} else {
s.set_key_split(key, false, &t).unwrap();
}
s.send(&test_delay(2)).await.unwrap();
let decrypted = matches!(s.next().await, Some(Ok(b)) if delay_in(&b) == Some(1));
tx.send(Seen {
picked: pk.kx_version,
decrypted,
})
.ok();
});
(host, signed_id_pk, rx)
}
/// The controller's handshake against the stub, then one application message each way.
/// Returns what the stub saw and whether the stub's message decrypted on this side.
async fn handshake(advertised: u32, run: Option<u32>) -> (Seen, bool) {
let (rs_pk, rs_sk) = sign::gen_keypair();
let (host, signed_id_pk, seen) = controlled_stub(advertised, run, rs_sk).await;
let mut conn = connect_tcp(host, 3000).await.unwrap();
let pk =
Client::secure_connection(PEER_ID, signed_id_pk, &crate::encode64(rs_pk.0), &mut conn)
.await
.unwrap();
assert!(pk.is_some() && conn.is_secured());
conn.send(&test_delay(1)).await.unwrap();
let decrypted = matches!(conn.next().await, Some(Ok(b)) if delay_in(&b) == Some(2));
(seen.await.unwrap(), decrypted)
}
#[tokio::test]
async fn test_new_peers_pick_version_1_and_exchange_application_data() {
let (seen, decrypted) = handshake(KX_VERSION_LATEST, None).await;
assert_eq!(seen.picked, 1);
assert!(seen.decrypted && decrypted);
}
#[tokio::test]
async fn test_a_peer_without_versions_gets_version_0() {
let (seen, decrypted) = handshake(0, Some(0)).await;
assert_eq!(seen.picked, 0);
assert!(seen.decrypted && decrypted);
}
#[tokio::test]
async fn test_a_pick_lowered_in_transit_decrypts_nothing() {
// The controlled side takes 0 from a controller without versions, so a lowered pick is
// caught by the keys disagreeing on the first application message, not before.
let (seen, decrypted) = handshake(KX_VERSION_LATEST, Some(0)).await;
assert_eq!(seen.picked, 1);
assert!(!seen.decrypted && !decrypted);
}
}