github/rustdesk
4
0 Fork 0
mirror of https://github.com/rustdesk/rustdesk.git synced 2026-11-04 00:33:12 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
4d37f00af964e8a43fb2239d8599ff8690e4d906
rustdesk/src/client/io_loop.rs
T
21pages 4d37f00af9 fix(client): invalidate auto FPS state when updating FPS limits (#16283) 
* fix(client): sync FPS control and reset limits on quality changes

  Track the last sent FPS across manual and automatic updates.
  Reset the FPS limit to 30 when leaving custom image quality,
  while preserving lower automatic limits when already capped at 30.

 fix(client): reset last sent FPS state on reconnect

Signed-off-by: 21pages <sunboeasy@gmail.com>

fix(client): invalidate auto FPS state when updating FPS limits

  Clear last_auto_fps when setting custom FPS so the controller recomputes
  on its next tick. Restore the 30 FPS cap when switching to preset quality
  while preserving the saved custom FPS setting.

Signed-off-by: 21pages <sunboeasy@gmail.com>

fix(client): reset auto FPS state after successful login

  Move last_auto_fps invalidation from login send paths to
  handle_peer_info, covering both normal and switch-side logins.

Signed-off-by: 21pages <sunboeasy@gmail.com>

fix(client): scope auto FPS state to each connection

Keep automatic FPS state in the connection loop and invalidate it when processing manual FPS updates or successful login.

Validate queued automatic updates against the current FPS cap and record the value only after a successful send.

Signed-off-by: 21pages <sunboeasy@gmail.com>

* test(client): cover FPS adjustment after manual updates

Signed-off-by: 21pages <sunboeasy@gmail.com>

---------

Signed-off-by: 21pages <sunboeasy@gmail.com>
2026-10-23 19:13:39 +08:00

3134 lines
142 KiB
Rust
Raw Blame History

#[cfg(not(any(target_os = "android", target_os = "ios")))]
use crate::clipboard::{clipboard_listener, update_clipboard, ClipboardSide};
#[cfg(not(any(target_os = "ios")))]
use crate::{audio_service, clipboard::CLIPBOARD_INTERVAL, ConnInner, CLIENT_SERVER};
use crate::{
client::{
self, new_voice_call_request, Client, Data, Interface, MediaData, MediaSender,
QualityStatus, MILLI1, SEC30,
},
common::get_default_sound_input,
ui_session_interface::{InvokeUiSession, Session},
};
// Empirical no-data window before exposing the restart reconnect state to the UI.
// Restart msgbox text is kept as a legacy UI fallback; Flutter handles the type as a control event.
const RESTART_REMOTE_DEVICE_NO_DATA_TIMEOUT: Duration = Duration::from_secs(5);
const KCP_CLOSE_REASON_FLUSH_DELAY: Duration = Duration::from_millis(30);
// Deadline for the parting close-reason send once the peer is presumed gone; KCP waits for send
// capacity with no deadline of its own.
const KCP_CLOSE_REASON_GONE_DEADLINE: Duration = Duration::from_millis(500);
// Grace after ICE reports Disconnected, which it does ~5s after it stops hearing from the peer,
// for ~8s in total. Disconnected is transient by design, so this waits out a Wi-Fi roam or a
// sleep/wake rather than acting on the first hint.
const WEBRTC_SUSPECT_GRACE: Duration = Duration::from_secs(3);
// KCP gets no such hint, only how long since a packet arrived; its endpoint pings an idle peer
// about every 2s, so this is several missed pings, and matches the 8s WebRTC arrives at.
const KCP_PEER_SILENCE_LIMIT: Duration = Duration::from_secs(8);
#[cfg(feature = "unix-file-copy-paste")]
use crate::{clipboard::try_empty_clipboard_files, clipboard_file::unix_file_clip};
use base::{
config::keys,
fs::{
self, can_enable_overwrite_detection, get_job, get_string, new_send_confirm,
DigestCheckResult, RemoveJobMeta,
},
message_proto::{permission_info::Permission, *},
};
#[cfg(any(
target_os = "windows",
all(target_os = "macos", feature = "unix-file-copy-paste")
))]
use clipboard::ContextSend;
use crossbeam_queue::ArrayQueue;
#[cfg(not(target_os = "ios"))]
use hbb_common::tokio::sync::mpsc::error::TryRecvError;
use hbb_common::{
allow_err,
config::{self, LocalConfig, PeerConfig, TransferSerde},
get_time, log,
protobuf::Message as _,
rendezvous_proto::ConnType,
timeout,
tokio::{
self,
sync::mpsc,
time::{self, Duration, Instant},
},
Stream,
};
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
use hbb_common::{tokio::sync::Mutex as TokioMutex, ResultType};
use scrap::CodecFormat;
use std::{
collections::{HashMap, HashSet},
ffi::c_void,
num::NonZeroI64,
path::PathBuf,
sync::{
atomic::{AtomicUsize, Ordering},
Arc, RwLock,
},
};
pub struct Remote<T: InvokeUiSession> {
handler: Session<T>,
audio_sender: MediaSender,
receiver: mpsc::UnboundedReceiver<Data>,
sender: mpsc::UnboundedSender<Data>,
// Stop sending local audio to remote client.
stop_voice_call_sender: Option<std::sync::mpsc::Sender<()>>,
voice_call_request_timestamp: Option<NonZeroI64>,
read_jobs: Vec<fs::TransferJob>,
write_jobs: Vec<fs::TransferJob>,
remove_jobs: HashMap<i32, RemoveJob>,
timer: crate::RustDeskInterval,
last_update_jobs_status: (Instant, HashMap<i32, u64>),
// Set after PeerInfo for this round, not when the transport connects.
is_connected: bool,
// Whether the scheduled initial snapshot may still be sent.
// The connection loop clears this when handling its result or a live clipboard update.
#[cfg(not(any(target_os = "android", target_os = "ios")))]
initial_clipboard_pending: bool,
first_frame: bool,
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
client_conn_id: i32, // used for file clipboard
data_count: Arc<AtomicUsize>,
video_format: CodecFormat,
elevation_requested: bool,
peer_info: ParsedPeerInfo,
video_threads: HashMap<usize, VideoThread>,
last_auto_fps: Option<usize>,
chroma: Arc<RwLock<Option<Chroma>>>,
last_record_state: bool,
sent_close_reason: bool,
cursor_dedupe: CursorDedupe,
}
#[derive(Default)]
struct ParsedPeerInfo {
platform: String,
is_installed: bool,
idd_impl: String,
support_view_camera: bool,
support_terminal: bool,
}
impl ParsedPeerInfo {
fn is_support_virtual_display(&self) -> bool {
self.is_installed
&& self.platform == "Windows"
&& (self.idd_impl == "rustdesk_idd" || self.idd_impl == "amyuni_idd")
}
}
impl<T: InvokeUiSession> Remote<T> {
pub fn new(
handler: Session<T>,
receiver: mpsc::UnboundedReceiver<Data>,
sender: mpsc::UnboundedSender<Data>,
) -> Self {
Self {
handler,
audio_sender: crate::client::start_audio_thread(),
receiver,
sender,
read_jobs: Vec::new(),
write_jobs: Vec::new(),
remove_jobs: Default::default(),
timer: crate::rustdesk_interval(time::interval(SEC30)),
last_update_jobs_status: (Instant::now(), Default::default()),
is_connected: false,
#[cfg(not(any(target_os = "android", target_os = "ios")))]
initial_clipboard_pending: false,
first_frame: false,
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
client_conn_id: 0,
data_count: Arc::new(AtomicUsize::new(0)),
video_format: CodecFormat::Unknown,
stop_voice_call_sender: None,
voice_call_request_timestamp: None,
elevation_requested: false,
peer_info: Default::default(),
video_threads: Default::default(),
last_auto_fps: None,
chroma: Default::default(),
last_record_state: false,
sent_close_reason: false,
cursor_dedupe: Default::default(),
}
}
pub async fn io_loop(&mut self, key: &str, token: &str, round: u32) {
#[cfg(target_os = "windows")]
let _file_clip_context_holder = {
// `is_port_forward()` will not reach here, but we still check it for clarity.
if self.handler.is_default() {
// It is ok to call this function multiple times.
ContextSend::enable(true);
Some(crate::SimpleCallOnReturn {
b: true,
f: Box::new(|| {
// No need to call `enable(false)` for sciter version, because each client of sciter version is a new process.
// It's better to check if the peers are windows(support file copy&paste), but it's not necessary.
#[cfg(feature = "flutter")]
if !crate::flutter::sessions::has_sessions_running(ConnType::DEFAULT_CONN) {
ContextSend::enable(false);
};
}),
})
} else {
None
}
};
let mut received = false;
let conn_type = if self.handler.is_file_transfer() {
ConnType::FILE_TRANSFER
} else if self.handler.is_view_camera() {
ConnType::VIEW_CAMERA
} else if self.handler.is_terminal() {
ConnType::TERMINAL
} else {
ConnType::default()
};
match Client::start(
&self.handler.get_id(),
key,
token,
conn_type,
self.handler.clone(),
)
.await
{
Ok(((mut peer, direct, pk, kcp, stream_type), (feedback, rendezvous_server))) => {
self.handler
.connection_round_state
.lock()
.unwrap()
.set_connected();
let is_secured = peer.is_secured();
// Only WebRTC needs refining: its label names the transport that won the race,
// not the family ICE ended up nominating, and it is the one path where the two
// can disagree with the address the rendezvous observed.
let stream_type = if peer.webrtc_remote_ipv6().await.unwrap_or(false) {
"WebRTC/IPv6"
} else {
stream_type
};
self.handler
.set_connection_type(is_secured, direct, stream_type); // flutter -> connection_ready
if !is_secured
&& !crate::common::is_direct_ip_access(&self.handler.get_id())
&& !client::confirm_insecure_connection(&self.handler, &mut self.receiver).await
{
self.send_close_reason(&mut peer, "").await;
if kcp.is_some() {
tokio::time::sleep(KCP_CLOSE_REASON_FLUSH_DELAY).await;
}
self.handle_disconnected(round);
return;
}
self.handler.update_direct(Some(direct));
if conn_type == ConnType::DEFAULT_CONN || conn_type == ConnType::VIEW_CAMERA {
self.handler
.set_fingerprint(crate::common::pk_to_fingerprint(pk.unwrap_or_default()));
}
// just build for now
#[cfg(not(any(target_os = "windows", feature = "unix-file-copy-paste")))]
let (_tx_holder, mut rx_clip_client) = mpsc::unbounded_channel::<i32>();
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
let (_tx_holder, rx) = mpsc::unbounded_channel();
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
let mut rx_clip_client_holder = (Arc::new(TokioMutex::new(rx)), None);
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
{
if self.handler.is_default() {
(self.client_conn_id, rx_clip_client_holder.0) =
clipboard::get_rx_cliprdr_client(&self.handler.get_id());
log::debug!("get cliprdr client for conn_id {}", self.client_conn_id);
let client_conn_id = self.client_conn_id;
rx_clip_client_holder.1 = Some(crate::SimpleCallOnReturn {
b: true,
f: Box::new(move || {
clipboard::remove_channel_by_conn_id(client_conn_id);
}),
});
};
}
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
let mut rx_clip_client = rx_clip_client_holder.0.lock().await;
let mut status_timer =
crate::rustdesk_interval(time::interval(Duration::new(1, 0)));
let mut fps_instant = Instant::now();
let _keep_it = client::hc_connection(feedback, rendezvous_server, token).await;
let mut last_recv_time = Instant::now();
let mut webrtc_suspect_since: Option<Instant> = None;
let mut last_rx_progress = peer.rx_progress();
let mut peer_gone = false;
loop {
tokio::select! {
res = peer.next() => {
if let Some(res) = res {
match res {
Err(err) => {
self.handler.on_establish_connection_error(err.to_string());
break;
}
Ok(ref bytes) => {
last_recv_time = Instant::now();
if !received {
received = true;
self.handler.update_received(true);
}
self.data_count.fetch_add(bytes.len(), Ordering::Relaxed);
if !self.handle_msg_from_peer(bytes, &mut peer).await {
break
}
}
}
} else {
if self.handler.is_restarting_remote_device() {
log::info!("Restart remote device");
self.handler.msgbox("restarting", "Restarting remote device", "Connection in progress. Please wait.", "");
} else {
log::info!("Reset by the peer");
self.handler.msgbox("error", "Connection Error", "Reset by the peer", "");
}
break;
}
}
d = self.receiver.recv() => {
if let Some(d) = d {
if !self.handle_msg_from_ui(d, &mut peer).await {
break;
}
}
}
_msg = rx_clip_client.recv() => {
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
self.handle_local_clipboard_msg(&mut peer, _msg).await;
}
_ = self.timer.tick() => {
if last_recv_time.elapsed() >= SEC30 {
self.handler.msgbox("error", "Connection Error", "Timeout", "");
break;
}
if !self.read_jobs.is_empty() {
if let Err(err) = fs::handle_read_jobs(&mut self.read_jobs, &mut peer).await {
self.handler.msgbox("error", "Connection Error", &err.to_string(), "");
break;
}
self.update_jobs_status();
} else {
self.timer = crate::rustdesk_interval(time::interval_at(Instant::now() + SEC30, SEC30));
}
}
_ = status_timer.tick() => {
if self.handler.is_restarting_remote_device()
&& last_recv_time.elapsed() >= RESTART_REMOTE_DEVICE_NO_DATA_TIMEOUT
{
self.handler.msgbox("restarting-show", "Restarting remote device", "Connection in progress. Please wait.", "");
break;
}
let rx_progress = peer.rx_progress();
// `None` for transports that report none, and it never changes for a
// given one, so they are inert here.
let progressed = rx_progress != last_rx_progress;
last_rx_progress = rx_progress;
if peer.webrtc_disconnected() && !progressed {
webrtc_suspect_since.get_or_insert_with(Instant::now);
} else {
webrtc_suspect_since = None;
}
// Neither limit is a hard upper bound. A send is awaited inline in
// this loop, so one in progress delays this tick - bounded on WebRTC
// by the timeout the stream was built with, not bounded at all on
// KCP. The 30s watchdog above shares the loop and the same delay.
peer_gone = webrtc_suspect_since
.map_or(false, |since| since.elapsed() >= WEBRTC_SUSPECT_GRACE)
|| kcp
.as_ref()
.and_then(|k| k.peer_silent_for())
.map_or(false, |silent| silent >= KCP_PEER_SILENCE_LIMIT);
if peer_gone {
log::info!("Peer stopped answering, reconnecting");
#[cfg(feature = "flutter")]
self.handler.msgbox("restarting-show", "Connecting...", "Connection in progress. Please wait.", "");
// Sciter knows no `restarting-show` and would show a dialog that
// waits for a click, where the timeout this arrives ahead of is
// retryable and reconnects on its own. Keep that message for it.
#[cfg(not(feature = "flutter"))]
self.handler.msgbox("error", "Connection Error", "Timeout", "");
break;
}
let elapsed = fps_instant.elapsed().as_millis();
if elapsed < 1000 {
continue;
}
fps_instant = Instant::now();
let mut speed = self.data_count.swap(0, Ordering::Relaxed);
speed = speed * 1000 / elapsed as usize;
let speed = format!("{:.2}kB/s", speed as f32 / 1024 as f32);
let fps = self.video_threads.iter().map(|(k, v)| {
// Correcting the inaccuracy of status_timer
(k.clone(), (*v.frame_count.read().unwrap() as i32) * 1000 / elapsed as i32)
}).collect::<HashMap<usize, i32>>();
self.video_threads.iter().for_each(|(_, v)| {
*v.frame_count.write().unwrap() = 0;
});
self.fps_control(direct, fps.clone(), &mut peer).await;
let chroma = self.chroma.read().unwrap().clone();
let chroma = match chroma {
Some(Chroma::I444) => "4:4:4",
Some(Chroma::I420) => "4:2:0",
None => "-",
};
let chroma = Some(chroma.to_string());
let codec_format = if self.video_format == CodecFormat::Unknown {
None
} else {
Some(self.video_format.clone())
};
self.handler.update_quality_status(QualityStatus {
speed: Some(speed),
fps,
chroma,
codec_format,
..Default::default()
});
}
}
}
log::debug!("Exit io_loop of id={}", self.handler.get_id());
// Stop client audio server.
if let Some(s) = self.stop_voice_call_sender.take() {
s.send(()).ok();
}
if kcp.is_some() {
// Attempted rather than skipped even here: if the loss was one-way the peer
// does get it, and drops its side instead of waiting out its own timeout.
if peer_gone {
peer.set_send_timeout(KCP_CLOSE_REASON_GONE_DEADLINE.as_millis() as u64);
}
// Send the close reason if it hasn't been sent yet, as KCP cannot detect the socket close event.
self.send_close_reason(&mut peer, "kcp").await;
// KCP does not send messages immediately, so wait to ensure the last message is sent.
// 1ms works in my test, but 30ms is more reliable.
tokio::time::sleep(KCP_CLOSE_REASON_FLUSH_DELAY).await;
}
}
Err(err) => {
self.handler.on_establish_connection_error(err.to_string());
}
}
self.handle_disconnected(round);
}
fn handle_disconnected(&self, round: u32) {
// set_disconnected_ok is used to check if new connection round is started.
let _set_disconnected_ok = self
.handler
.connection_round_state
.lock()
.unwrap()
.set_disconnected(round);
#[cfg(not(target_os = "ios"))]
if self.handler.is_default() && _set_disconnected_ok {
// Other sessions may keep the listener running after this one disconnects.
#[cfg(feature = "flutter")]
crate::flutter::update_text_clipboard_required();
#[cfg(all(feature = "flutter", feature = "unix-file-copy-paste"))]
crate::flutter::update_file_clipboard_required();
Client::try_stop_clipboard();
}
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
if self.handler.is_default() && _set_disconnected_ok {
// Linux client cleanup runs synchronously in try_stop_clipboard() before FUSE is
// unmounted. Keep this async path for other file-clipboard platforms.
crate::clipboard::try_empty_clipboard_files(ClipboardSide::Client, self.client_conn_id);
}
}
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
async fn handle_local_clipboard_msg(
&mut self,
peer: &mut Stream,
msg: Option<clipboard::ClipboardFile>,
) {
match msg {
Some(clip) => match clip {
clipboard::ClipboardFile::NotifyCallback {
r#type,
title,
text,
} => {
self.handler.msgbox(&r#type, &title, &text, "");
}
// File-data responses bypass is_stopping_allowed, but still require login.
_ if !self.is_connected => {
log::debug!("Discarding local file clipboard message before login");
}
_ => {
let is_stopping_allowed = clip.is_stopping_allowed();
let server_file_transfer_enabled =
*self.handler.server_file_transfer_enabled.read().unwrap();
let file_transfer_enabled =
self.handler.lc.read().unwrap().enable_file_copy_paste.v;
let view_only = self.handler.lc.read().unwrap().view_only.v;
let stop = is_stopping_allowed
&& (view_only || !(server_file_transfer_enabled && file_transfer_enabled));
log::debug!(
"Process clipboard message from system, view_only: {}, stop: {}, is_stopping_allowed: {}, server_file_transfer_enabled: {}, file_transfer_enabled: {}",
view_only, stop, is_stopping_allowed, server_file_transfer_enabled, file_transfer_enabled
);
if stop {
#[cfg(target_os = "windows")]
{
ContextSend::set_is_stopped();
}
} else {
#[cfg(target_os = "windows")]
if let Err(e) = ContextSend::make_sure_enabled() {
log::error!("failed to restart clipboard context: {}", e);
// to-do: Show msgbox with "Don't show again" option
};
log::debug!("Send system clipboard message to remote");
#[cfg(not(any(target_os = "android", target_os = "ios")))]
if matches!(&clip, clipboard::ClipboardFile::FormatList { .. }) {
self.initial_clipboard_pending = false;
}
let msg = crate::clipboard_file::clip_2_msg(clip);
allow_err!(peer.send(&msg).await);
}
}
},
None => {
// unreachable!()
}
}
}
fn handle_job_status(&mut self, id: i32, file_num: i32, err: Option<String>) {
if let Some(job) = self.remove_jobs.get_mut(&id) {
if job.no_confirm {
let file_num = (file_num + 1) as usize;
if file_num < job.files.len() {
let path = format!("{}{}{}", job.path, job.sep, job.files[file_num].name);
self.sender
.send(Data::RemoveFile((id, path, file_num as i32, job.is_remote)))
.ok();
let elapsed = job.last_update_job_status.elapsed().as_millis() as i32;
if elapsed >= 1000 {
job.last_update_job_status = Instant::now();
} else {
return;
}
} else {
self.remove_jobs.remove(&id);
}
}
}
if let Some(err) = err {
self.handler.job_error(id, err, file_num);
} else {
self.handler.job_done(id, file_num);
}
}
fn stop_voice_call(&mut self) {
let voice_call_sender = std::mem::replace(&mut self.stop_voice_call_sender, None);
if let Some(stopper) = voice_call_sender {
let _ = stopper.send(());
}
}
// Start a voice call recorder, records audio and send to remote
fn start_voice_call(&mut self) -> Option<std::sync::mpsc::Sender<()>> {
if self.handler.is_file_transfer()
|| self.handler.is_port_forward()
|| self.handler.is_terminal()
{
return None;
}
// iOS does not have this server.
#[cfg(not(any(target_os = "ios")))]
{
// NOTE:
// The client server and --server both use the same sound input device.
// It's better to distinguish the server side and client side.
// But it' not necessary for now, because it's not a common case.
// And it is immediately known when the input device is changed.
crate::audio_service::set_voice_call_input_device(get_default_sound_input(), false);
// Create a channel to receive error or closed message
let (tx, rx) = std::sync::mpsc::channel();
let (tx_audio_data, mut rx_audio_data) =
hbb_common::tokio::sync::mpsc::unbounded_channel();
// Create a stand-alone inner, add subscribe to audio service
let conn_id = CLIENT_SERVER.write().unwrap().get_new_id();
let client_conn_inner = ConnInner::new(conn_id.clone(), Some(tx_audio_data), None);
// now we subscribe
CLIENT_SERVER.write().unwrap().subscribe(
audio_service::NAME,
client_conn_inner.clone(),
true,
);
let tx_audio = self.sender.clone();
std::thread::spawn(move || {
loop {
// check if client is closed
match rx.try_recv() {
Ok(_) | Err(std::sync::mpsc::TryRecvError::Disconnected) => {
log::debug!("Exit voice call audio service of client");
// unsubscribe
CLIENT_SERVER.write().unwrap().subscribe(
audio_service::NAME,
client_conn_inner,
false,
);
crate::audio_service::set_voice_call_input_device(None, true);
break;
}
_ => {}
}
match rx_audio_data.try_recv() {
Ok((_instant, msg)) => match &msg.union {
Some(message::Union::AudioFrame(frame)) => {
let mut msg = Message::new();
msg.set_audio_frame(frame.clone());
tx_audio.send(Data::Message(msg)).ok();
}
Some(message::Union::Misc(misc)) => {
let mut msg = Message::new();
msg.set_misc(misc.clone());
tx_audio.send(Data::Message(msg)).ok();
}
_ => {}
},
Err(err) => {
if err == TryRecvError::Empty {
// ignore
} else {
log::debug!("Failed to record local audio channel: {}", err);
}
// Both arms fall through with nothing else in this loop blocking, so
// without a pause the thread spun a core for the whole voice call.
std::thread::sleep(std::time::Duration::from_millis(1));
}
}
}
});
return Some(tx);
}
#[cfg(target_os = "ios")]
{
None
}
}
async fn send_close_reason(&mut self, peer: &mut Stream, reason: &str) {
if self.sent_close_reason {
return;
}
let mut misc = Misc::new();
misc.set_close_reason(reason.to_owned());
let mut msg = Message::new();
msg.set_misc(misc);
allow_err!(peer.send(&msg).await);
self.sent_close_reason = true;
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
fn start_initial_clipboard_sync(&mut self) {
let peer_info = {
let lc = self.handler.lc.read().unwrap();
lc.peer_info
.as_ref()
.map(|pi| (pi.version.clone(), pi.platform.clone()))
};
let Some((peer_version, peer_platform)) = peer_info else {
log::error!("Cannot start initial clipboard sync without peer information");
return;
};
self.initial_clipboard_pending = true;
let sender = self.sender.clone();
let permission_config = self.handler.get_permission_config();
// Clipboard access and encoding must not block the connection loop.
let read_clipboard = move || {
// Capture after readiness, before reading; later changes invalidate this snapshot.
let generation = clipboard_listener::current_generation();
let msg_out = if permission_config.is_text_clipboard_required() {
crate::clipboard::get_current_clipboard_msg(
&peer_version,
&peer_platform,
crate::clipboard::ClipboardSide::Client,
)
} else {
None
};
let msg_out = msg_out.filter(|_| permission_config.is_text_clipboard_required());
// Empty or failed reads must also finish the pending initial-sync attempt.
if let Err(err) = sender.send(Data::InitialClipboard(generation, msg_out)) {
log::debug!("Failed to send initial clipboard result: {}", err);
}
};
#[cfg(target_os = "linux")]
self.spawn_initial_clipboard_read_after_ready(read_clipboard);
#[cfg(not(target_os = "linux"))]
tokio::task::spawn_blocking(read_clipboard);
}
// Linux listener creation returns before X11/Wayland has subscribed to changes.
// Capture after readiness so changes during the read can invalidate the snapshot.
// A separate task keeps a stalled backend from blocking the connection loop.
#[cfg(target_os = "linux")]
fn spawn_initial_clipboard_read_after_ready(&self, read_clipboard: impl FnOnce() + Send + 'static) {
// Initial-sync wait budget, not a protocol-defined startup deadline.
const CLIPBOARD_READY_TIMEOUT: Duration = Duration::from_secs(5);
let sender = self.sender.clone();
tokio::spawn(async move {
let ready = tokio::select! {
result = time::timeout(CLIPBOARD_READY_TIMEOUT, clipboard_listener::wait_for_ready()) => {
match result {
Ok(ready) => ready,
Err(err) => Err(err.into()),
}
}
_ = sender.closed() => return,
};
if let Err(err) = ready {
log::error!(
"Failed to wait for clipboard listener readiness (limit {:?}): {}",
CLIPBOARD_READY_TIMEOUT,
err,
);
let generation = clipboard_listener::current_generation();
if let Err(err) = sender.send(Data::InitialClipboard(generation, None)) {
log::debug!("Failed to send initial clipboard result: {}", err);
}
return;
}
tokio::task::spawn_blocking(read_clipboard);
});
}
async fn handle_msg_from_ui(&mut self, data: Data, peer: &mut Stream) -> bool {
match data {
Data::Close => {
self.send_close_reason(peer, "").await;
return false;
}
Data::Login((os_username, os_password, password, remember)) => {
self.handler
.handle_login_from_ui(os_username, os_password, password, remember, peer)
.await;
}
#[cfg(all(target_os = "windows", not(feature = "flutter")))]
Data::ToggleClipboardFile => {
self.check_clipboard_file_context();
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
Data::InitialClipboard(generation, msg) => {
// A live update supersedes any initial snapshot still being prepared.
if !self.initial_clipboard_pending {
return true;
}
self.initial_clipboard_pending = false;
if !self.handler.is_text_clipboard_required() {
return true;
}
if generation != clipboard_listener::current_generation() {
// A clipboard change or listener restart can invalidate the snapshot
// without sending a live update that supersedes it.
drop(msg);
self.start_initial_clipboard_sync();
return true;
}
if let Some(msg) = msg {
allow_err!(peer.send(&msg).await);
}
}
Data::Message(msg) => {
// The Flutter clipboard broadcast is process-wide, so a clipboard can reach this
// round's queue before the round has logged in; it is dropped here, on the round
// itself.
#[cfg(feature = "flutter")]
if !self.is_connected
&& matches!(
msg.union.as_ref(),
Some(message::Union::Clipboard(_))
| Some(message::Union::MultiClipboards(_))
)
{
return true;
}
match &msg.union {
Some(message::Union::Cliprdr(_)) if !self.is_connected => {
log::debug!("Discarding outgoing file clipboard message before login");
return true;
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
Some(message::Union::Clipboard(_)) | Some(message::Union::MultiClipboards(_)) => {
self.initial_clipboard_pending = false;
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
Some(message::Union::Cliprdr(clip))
if matches!(clip.union.as_ref(), Some(cliprdr::Union::FormatList(_))) =>
{
self.initial_clipboard_pending = false;
}
Some(message::Union::Misc(misc)) => match misc.union {
Some(misc::Union::Option(ref option)) => {
if option.custom_fps > 0 {
self.last_auto_fps = None;
}
}
Some(misc::Union::RefreshVideo(_)) => {
self.video_threads.iter().for_each(|(_, v)| {
*v.discard_queue.write().unwrap() = true;
});
}
Some(misc::Union::RefreshVideoDisplay(display)) => {
if let Some(v) = self.video_threads.get_mut(&(display as usize)) {
*v.discard_queue.write().unwrap() = true;
}
}
_ => {}
},
_ => {}
}
allow_err!(peer.send(&msg).await);
}
Data::SendFiles((id, r#type, path, to, file_num, include_hidden, is_remote)) => {
log::info!("send files, is remote {}", is_remote);
let od = can_enable_overwrite_detection(self.handler.lc.read().unwrap().version);
if is_remote {
log::debug!("New job {}, write to {} from remote {}", id, to, path);
let to = match r#type {
fs::JobType::Generic => {
if let Err(err) = fs::download::validate_download_name(
&path,
self.handler.peer_platform() == "Windows",
) {
self.handle_job_status(id, -1, Some(err.to_string()));
return true;
}
fs::DataSource::FilePath(PathBuf::from(&to))
}
fs::JobType::Printer => {
fs::DataSource::MemoryCursor(std::io::Cursor::new(Vec::new()))
}
};
self.write_jobs.push(fs::TransferJob::new_write(
id,
r#type,
path.clone(),
to,
file_num,
include_hidden,
is_remote,
od,
));
allow_err!(
peer.send(&fs::new_send(id, r#type, path, file_num, include_hidden))
.await
);
} else {
match fs::TransferJob::new_read(
id,
r#type,
to.clone(),
fs::DataSource::FilePath(PathBuf::from(&path)),
file_num,
include_hidden,
is_remote,
od,
) {
Err(err) => {
self.handle_job_status(id, -1, Some(err.to_string()));
}
Ok(job) => {
log::debug!(
"New job {}, read {} to remote {}, {} files",
id,
path,
to,
job.files().len()
);
self.handler.update_folder_files(
job.id(),
job.files(),
path,
!is_remote,
true,
);
#[cfg(not(windows))]
let files = job.files().clone();
#[cfg(windows)]
let mut files = job.files().clone();
#[cfg(windows)]
if self.handler.peer_platform() != "Windows" {
// peer is not windows, need transform \ to /
fs::transform_windows_path(&mut files);
}
let total_size = job.total_size();
self.read_jobs.push(job);
self.timer = crate::rustdesk_interval(time::interval(MILLI1));
allow_err!(
peer.send(&fs::new_receive(id, to, file_num, files, total_size))
.await
);
}
}
}
}
Data::AddJob((id, r#type, path, to, file_num, include_hidden, is_remote)) => {
let od = can_enable_overwrite_detection(self.handler.lc.read().unwrap().version);
if is_remote {
log::debug!(
"new write waiting job {}, write to {} from remote {}",
id,
to,
path
);
let mut job = fs::TransferJob::new_write(
id,
r#type,
path.clone(),
fs::DataSource::FilePath(PathBuf::from(&to)),
file_num,
include_hidden,
is_remote,
od,
);
job.is_last_job = true;
self.write_jobs.push(job);
} else {
match fs::TransferJob::new_read(
id,
r#type,
to.clone(),
fs::DataSource::FilePath(PathBuf::from(&path)),
file_num,
include_hidden,
is_remote,
od,
) {
Err(err) => {
self.handle_job_status(id, -1, Some(err.to_string()));
}
Ok(mut job) => {
log::debug!(
"new read waiting job {}, read {} to remote {}, {} files",
id,
path,
to,
job.files().len()
);
self.handler.update_folder_files(
job.id(),
job.files(),
path,
!is_remote,
true,
);
job.is_last_job = true;
self.read_jobs.push(job);
self.timer = crate::rustdesk_interval(time::interval(MILLI1));
}
}
}
}
Data::ResumeJob((id, is_remote)) => {
if is_remote {
if let Some(job) = get_job(id, &mut self.write_jobs) {
job.is_last_job = false;
job.is_resume = true;
allow_err!(
peer.send(&fs::new_send(
id,
fs::JobType::Generic,
job.remote.clone(),
job.file_num,
job.show_hidden
))
.await
);
}
} else {
if let Some(job) = get_job(id, &mut self.read_jobs) {
match &job.data_source {
fs::DataSource::FilePath(_p) => {
job.is_last_job = false;
job.is_resume = true;
job.set_finished_size_on_resume();
#[cfg(not(windows))]
let files = job.files().clone();
#[cfg(windows)]
let mut files = job.files().clone();
#[cfg(windows)]
if self.handler.peer_platform() != "Windows" {
// peer is not windows, need transform \ to /
fs::transform_windows_path(&mut files);
}
allow_err!(
peer.send(&fs::new_receive(
id,
job.remote.clone(),
job.file_num,
files,
job.total_size(),
))
.await
);
}
fs::DataSource::MemoryCursor(_) => {
// unreachable!()
log::error!("Resume job with memory cursor");
}
}
}
}
}
Data::SetNoConfirm(id) => {
if let Some(job) = self.remove_jobs.get_mut(&id) {
job.no_confirm = true;
}
}
Data::ConfirmDeleteFiles((id, file_num)) => {
if let Some(job) = self.remove_jobs.get_mut(&id) {
let i = file_num as usize;
if i < job.files.len() {
self.handler.ui_handler.confirm_delete_files(
id,
file_num,
job.files[i].name.clone(),
);
}
}
}
Data::SetConfirmOverrideFile((id, file_num, need_override, remember, is_upload)) => {
if is_upload {
if let Some(job) = fs::get_job(id, &mut self.read_jobs) {
if remember {
job.set_overwrite_strategy(Some(need_override));
}
job.confirm(&FileTransferSendConfirmRequest {
id,
file_num,
union: if need_override {
Some(file_transfer_send_confirm_request::Union::OffsetBlk(0))
} else {
Some(file_transfer_send_confirm_request::Union::Skip(true))
},
..Default::default()
})
.await;
}
} else {
if let Some(job) = fs::get_job(id, &mut self.write_jobs) {
if remember {
job.set_overwrite_strategy(Some(need_override));
}
let mut msg = Message::new();
let mut file_action = FileAction::new();
let req = FileTransferSendConfirmRequest {
id,
file_num,
union: if need_override {
Some(file_transfer_send_confirm_request::Union::OffsetBlk(0))
} else {
Some(file_transfer_send_confirm_request::Union::Skip(true))
},
..Default::default()
};
job.confirm(&req).await;
file_action.set_send_confirm(req);
msg.set_file_action(file_action);
allow_err!(peer.send(&msg).await);
}
}
}
Data::RemoveDirAll((id, path, is_remote, include_hidden)) => {
let sep = self.handler.get_path_sep(is_remote);
if is_remote {
let mut msg_out = Message::new();
let mut file_action = FileAction::new();
file_action.set_all_files(ReadAllFiles {
id,
path: path.clone(),
include_hidden,
..Default::default()
});
msg_out.set_file_action(file_action);
allow_err!(peer.send(&msg_out).await);
self.remove_jobs
.insert(id, RemoveJob::new(Vec::new(), path, sep, is_remote));
} else {
match fs::get_recursive_files(&path, include_hidden) {
Ok(entries) => {
self.handler.update_folder_files(
id,
&entries,
path.clone(),
!is_remote,
false,
);
self.remove_jobs
.insert(id, RemoveJob::new(entries, path, sep, is_remote));
}
Err(err) => {
self.handle_job_status(id, -1, Some(err.to_string()));
}
}
}
}
Data::CancelJob(id) => {
self.cancel_transfer_job(id, peer).await;
}
Data::RemoveDir((id, path)) => {
let mut msg_out = Message::new();
let mut file_action = FileAction::new();
file_action.set_remove_dir(FileRemoveDir {
id,
path,
recursive: true,
..Default::default()
});
msg_out.set_file_action(file_action);
allow_err!(peer.send(&msg_out).await);
}
Data::RemoveFile((id, path, file_num, is_remote)) => {
if is_remote {
let mut msg_out = Message::new();
let mut file_action = FileAction::new();
file_action.set_remove_file(FileRemoveFile {
id,
path,
file_num,
..Default::default()
});
msg_out.set_file_action(file_action);
allow_err!(peer.send(&msg_out).await);
} else {
match fs::remove_file(&path) {
Err(err) => {
self.handle_job_status(id, file_num, Some(err.to_string()));
}
Ok(()) => {
self.handle_job_status(id, file_num, None);
}
}
}
}
Data::CreateDir((id, path, is_remote)) => {
if is_remote {
let mut msg_out = Message::new();
let mut file_action = FileAction::new();
file_action.set_create(FileDirCreate {
id,
path,
..Default::default()
});
msg_out.set_file_action(file_action);
allow_err!(peer.send(&msg_out).await);
} else {
match fs::create_dir(&path) {
Err(err) => {
self.handle_job_status(id, -1, Some(err.to_string()));
}
Ok(()) => {
self.handle_job_status(id, -1, None);
}
}
}
}
Data::RenameFile((id, path, new_name, is_remote)) => {
if is_remote {
let mut msg_out = Message::new();
let mut file_action = FileAction::new();
file_action.set_rename(FileRename {
id,
path,
new_name,
..Default::default()
});
msg_out.set_file_action(file_action);
allow_err!(peer.send(&msg_out).await);
} else {
let err = fs::rename_file(&path, &new_name)
.err()
.map(|e| e.to_string());
self.handle_job_status(id, -1, err);
}
}
Data::RecordScreen(start) => {
self.handler.lc.write().unwrap().record_state = start;
self.update_record_state();
}
Data::ElevateDirect => {
let mut request = ElevationRequest::new();
request.set_direct(true);
let mut misc = Misc::new();
misc.set_elevation_request(request);
let mut msg = Message::new();
msg.set_misc(misc);
allow_err!(peer.send(&msg).await);
self.elevation_requested = true;
}
Data::ElevateWithLogon(username, password) => {
let mut request = ElevationRequest::new();
request.set_logon(ElevationRequestWithLogon {
username,
password,
..Default::default()
});
let mut misc = Misc::new();
misc.set_elevation_request(request);
let mut msg = Message::new();
msg.set_misc(misc);
allow_err!(peer.send(&msg).await);
self.elevation_requested = true;
}
Data::NewVoiceCall => {
let msg = new_voice_call_request(true);
// Save the voice call request timestamp for the further validation.
self.voice_call_request_timestamp = Some(
NonZeroI64::new(msg.voice_call_request().req_timestamp)
.unwrap_or(NonZeroI64::new(get_time()).unwrap()),
);
allow_err!(peer.send(&msg).await);
self.handler.on_voice_call_waiting();
}
Data::CloseVoiceCall => {
self.stop_voice_call();
let msg = new_voice_call_request(false);
self.handler
.on_voice_call_closed("Closed manually by the peer");
allow_err!(peer.send(&msg).await);
}
Data::ResetDecoder(display) => match display {
Some(display) => {
if let Some(v) = self.video_threads.get_mut(&display) {
v.video_sender.send(MediaData::Reset).ok();
}
}
None => {
for (_, v) in self.video_threads.iter_mut() {
v.video_sender.send(MediaData::Reset).ok();
}
}
},
Data::TakeScreenshot((display, sid)) => {
let mut msg = Message::new();
msg.set_screenshot_request(ScreenshotRequest {
display,
sid,
..Default::default()
});
allow_err!(peer.send(&msg).await);
}
_ => {}
}
true
}
#[inline]
fn update_job_status(
job: &fs::TransferJob,
elapsed: i32,
last_update_jobs_status: &mut (Instant, HashMap<i32, u64>),
handler: &Session<T>,
) {
if elapsed <= 0 {
return;
}
let transferred = job.transferred();
let last_transferred = {
if let Some(v) = last_update_jobs_status.1.get(&job.id()) {
v.to_owned()
} else {
0
}
};
last_update_jobs_status.1.insert(job.id(), transferred);
let speed = (transferred - last_transferred) as f64 / (elapsed as f64 / 1000.);
let file_num = job.file_num() - 1;
handler.job_progress(job.id(), file_num, speed, job.finished_size() as f64);
}
fn update_jobs_status(&mut self) {
let elapsed = self.last_update_jobs_status.0.elapsed().as_millis() as i32;
if elapsed >= 1000 {
for job in self.read_jobs.iter() {
Self::update_job_status(
job,
elapsed,
&mut self.last_update_jobs_status,
&self.handler,
);
}
for job in self.write_jobs.iter() {
Self::update_job_status(
job,
elapsed,
&mut self.last_update_jobs_status,
&mut self.handler,
);
}
self.last_update_jobs_status.0 = Instant::now();
}
}
async fn cancel_transfer_job(&mut self, id: i32, peer: &mut Stream) {
let mut msg_out = Message::new();
let mut file_action = FileAction::new();
file_action.set_cancel(FileTransferCancel {
id,
..Default::default()
});
msg_out.set_file_action(file_action);
allow_err!(peer.send(&msg_out).await);
if let Some(job) = fs::remove_job(id, &mut self.write_jobs) {
job.remove_download_file();
}
let _ = fs::remove_job(id, &mut self.read_jobs);
self.remove_jobs.remove(&id);
}
pub async fn sync_jobs_status_to_local(&mut self) -> bool {
if !self.is_connected {
return false;
}
let mut config: PeerConfig = self.handler.load_config();
let mut transfer_metas = TransferSerde::default();
for job in self.read_jobs.iter() {
let json_str = serde_json::to_string(&job.gen_meta()).unwrap_or_default();
transfer_metas.read_jobs.push(json_str);
}
for job in self.write_jobs.iter() {
let json_str = serde_json::to_string(&job.gen_meta()).unwrap_or_default();
transfer_metas.write_jobs.push(json_str);
}
log::info!("meta: {:?}", transfer_metas);
if config.transfer != transfer_metas {
config.transfer = transfer_metas;
self.handler.save_config(config);
}
true
}
async fn send_toggle_virtual_display_msg(&self, peer: &mut Stream) {
if self.handler.is_view_camera() {
return;
}
if !self.peer_info.is_support_virtual_display() {
return;
}
let lc = self.handler.lc.read().unwrap();
let displays = lc.get_option("virtual-display");
for d in displays.split(',') {
if let Ok(index) = d.parse::<i32>() {
let mut misc = Misc::new();
misc.set_toggle_virtual_display(ToggleVirtualDisplay {
display: index,
on: true,
..Default::default()
});
let mut msg_out = Message::new();
msg_out.set_misc(misc);
allow_err!(peer.send(&msg_out).await);
}
}
}
async fn send_toggle_privacy_mode_msg(&self, peer: &mut Stream) {
if self.handler.is_view_camera() {
return;
}
let lc = self.handler.lc.read().unwrap();
if lc.version >= hbb_common::get_version_number("1.2.4")
&& lc.get_toggle_option("privacy-mode")
{
let impl_key = lc.get_option("privacy-mode-impl-key");
if impl_key == crate::privacy_mode::PRIVACY_MODE_IMPL_WIN_VIRTUAL_DISPLAY
&& !self.peer_info.is_support_virtual_display()
{
return;
}
let mut misc = Misc::new();
misc.set_toggle_privacy_mode(TogglePrivacyMode {
impl_key,
on: true,
..Default::default()
});
let mut msg_out = Message::new();
msg_out.set_misc(misc);
allow_err!(peer.send(&msg_out).await);
}
}
fn contains_key_frame(vf: &VideoFrame) -> bool {
use video_frame::Union::*;
match &vf.union {
Some(vf) => match vf {
Vp8s(f) | Vp9s(f) | Av1s(f) | H264s(f) | H265s(f) => f.frames.iter().any(|e| e.key),
_ => false,
},
None => false,
}
}
// Currently, this function only considers decoding speed and queue length, not network delay.
// The controlled end can consider auto fps as the maximum decoding fps.
#[inline]
async fn fps_control(
&mut self,
direct: bool,
real_fps_map: HashMap<usize, i32>,
peer: &mut Stream,
) {
self.video_threads.iter_mut().for_each(|(k, v)| {
let real_fps = real_fps_map.get(k).cloned().unwrap_or_default();
if real_fps == 0 {
v.fps_control.inactive_counter += 1;
} else {
v.fps_control.inactive_counter = 0;
}
});
let custom_fps = self.handler.lc.read().unwrap().custom_fps.clone();
let custom_fps = custom_fps.lock().unwrap().clone();
let mut custom_fps = custom_fps.unwrap_or(30);
if custom_fps < 5 || custom_fps > 120 {
custom_fps = 30;
}
let inactive_threshold = 15;
let max_queue_len = self
.video_threads
.iter()
.map(|v| v.1.video_queue.read().unwrap().len())
.max()
.unwrap_or_default();
let min_decode_fps = self
.video_threads
.iter()
.filter(|v| v.1.fps_control.inactive_counter < inactive_threshold)
.map(|v| *v.1.decode_fps.read().unwrap())
.min()
.flatten();
let Some(min_decode_fps) = min_decode_fps else {
return;
};
let mut limited_fps = if direct {
min_decode_fps * 9 / 10 // 30 got 27
} else {
min_decode_fps * 4 / 5 // 30 got 24
};
if limited_fps > custom_fps {
limited_fps = custom_fps;
}
let last_auto_fps = self.last_auto_fps;
let displays = self.video_threads.keys().cloned().collect::<Vec<_>>();
let mut fps_trending = |display: usize| {
let thread = self.video_threads.get_mut(&display)?;
let ctl = &mut thread.fps_control;
let len = thread.video_queue.read().unwrap().len();
let decode_fps = thread.decode_fps.read().unwrap().clone()?;
let last_auto_fps = last_auto_fps.clone().unwrap_or(custom_fps as _);
if ctl.inactive_counter > inactive_threshold {
return None;
}
if len > 1 && last_auto_fps > limited_fps || len > std::cmp::max(1, decode_fps / 2) {
ctl.idle_counter = 0;
return Some(false);
}
if len <= 1 {
ctl.idle_counter += 1;
if ctl.idle_counter > 3 && last_auto_fps + 3 <= limited_fps {
return Some(true);
}
}
if len > 1 {
ctl.idle_counter = 0;
}
None
};
let trendings: Vec<_> = displays.iter().map(|k| fps_trending(*k)).collect();
let should_decrease = trendings.iter().any(|v| *v == Some(false));
let should_increase = !should_decrease && trendings.iter().any(|v| *v == Some(true));
if last_auto_fps.is_none() || should_decrease || should_increase {
// limited_fps to ensure decoding is faster than encoding
let mut auto_fps = limited_fps;
if should_decrease && limited_fps < max_queue_len {
auto_fps = limited_fps / 2;
}
if auto_fps < 1 {
auto_fps = 1;
}
if Some(auto_fps) != last_auto_fps {
let mut misc = Misc::new();
misc.set_option(OptionMessage {
custom_fps: auto_fps as _,
..Default::default()
});
let mut msg = Message::new();
msg.set_misc(misc);
// Send inline so a queued manual FPS update is applied after this decision.
allow_err!(peer.send(&msg).await);
log::info!("Set fps to {}", auto_fps);
self.last_auto_fps = Some(auto_fps);
}
}
// send refresh
for (display, thread) in self.video_threads.iter_mut() {
let ctl = &mut thread.fps_control;
let video_queue = thread.video_queue.read().unwrap();
let tolerable = std::cmp::min(min_decode_fps, video_queue.capacity() / 2);
if ctl.refresh_times < 20 // enough
&& (video_queue.len() > tolerable
&& (ctl.refresh_times == 0 || ctl.last_refresh_instant.map(|t|t.elapsed().as_secs() > 10).unwrap_or(false)))
{
// Refresh causes client set_display, left frames cause flickering.
drop(video_queue);
self.handler.refresh_video(*display as _);
log::info!("Refresh display {} to reduce delay", display);
ctl.refresh_times += 1;
ctl.last_refresh_instant = Some(Instant::now());
}
}
}
fn check_view_camera_support(&self, peer_version: &str, peer_platform: &str) -> bool {
if self.peer_info.support_view_camera {
return true;
}
if hbb_common::get_version_number(&peer_version) < hbb_common::get_version_number("1.3.9")
&& (peer_platform == "Windows" || peer_platform == "Linux")
{
self.handler.msgbox(
"error",
"Download new version",
"upgrade_remote_rustdesk_client_to_{1.3.9}_tip",
"",
);
} else {
self.handler.on_error("view_camera_unsupported_tip");
}
return false;
}
fn check_terminal_support(&self, peer_version: &str) -> bool {
if self.peer_info.support_terminal {
return true;
}
if hbb_common::get_version_number(&peer_version) < hbb_common::get_version_number("1.4.1") {
self.handler.msgbox(
"error",
"Remote terminal not supported",
"Remote terminal is not supported by the remote side. Please upgrade to version 1.4.1 or higher.",
"",
);
} else {
self.handler
.on_error("Remote terminal is not supported by the remote side");
}
return false;
}
async fn handle_msg_from_peer(&mut self, data: &[u8], peer: &mut Stream) -> bool {
if let Ok(msg_in) = Message::parse_from_bytes(&data) {
match msg_in.union {
Some(message::Union::VideoFrame(vf)) => {
if !self.first_frame {
self.first_frame = true;
self.handler.close_success();
self.handler.adapt_size();
self.send_toggle_virtual_display_msg(peer).await;
self.send_toggle_privacy_mode_msg(peer).await;
}
self.video_format = CodecFormat::from(&vf);
let display = vf.display as usize;
if !self.video_threads.contains_key(&display) {
self.new_video_thread(display);
}
let Some(thread) = self.video_threads.get_mut(&display) else {
return true;
};
if Self::contains_key_frame(&vf) {
thread
.video_sender
.send(MediaData::VideoFrame(Box::new(vf)))
.ok();
} else {
let video_queue = thread.video_queue.read().unwrap();
if video_queue.force_push(vf).is_some() {
drop(video_queue);
self.handler.refresh_video(display as _);
} else {
thread.video_sender.send(MediaData::VideoQueue).ok();
}
}
}
Some(message::Union::Hash(hash)) => {
if !self
.handler
.handle_hash(&self.handler.password.clone(), hash, peer)
.await
{
return false;
}
}
Some(message::Union::LoginResponse(lr)) => match lr.union {
Some(login_response::Union::Error(err)) => {
if err == client::REQUIRE_2FA {
self.handler.lc.write().unwrap().enable_trusted_devices =
lr.enable_trusted_devices;
}
if !self.handler.handle_login_error(&err) {
return false;
}
}
Some(login_response::Union::PeerInfo(pi)) => {
let peer_version = pi.version.clone();
let peer_platform = pi.platform.clone();
self.set_peer_info(&pi);
if self.handler.is_view_camera() {
if !self.check_view_camera_support(&peer_version, &peer_platform) {
self.handler.lc.write().unwrap().handle_peer_info(&pi);
return false;
}
}
if self.handler.is_terminal() {
if !self.check_terminal_support(&peer_version) {
self.handler.lc.write().unwrap().handle_peer_info(&pi);
return false;
}
}
self.handler.handle_peer_info(pi);
#[cfg(all(target_os = "windows", not(feature = "flutter")))]
self.check_clipboard_file_context();
if self.handler.is_default() {
// Startup notifications must see the current sessions' clipboard requirements.
#[cfg(feature = "flutter")]
#[cfg(not(target_os = "ios"))]
crate::flutter::update_text_clipboard_required();
#[cfg(all(feature = "flutter", feature = "unix-file-copy-paste"))]
crate::flutter::update_file_clipboard_required();
#[cfg(feature = "flutter")]
#[cfg(not(target_os = "ios"))]
let rx = Client::try_start_clipboard(None);
#[cfg(not(feature = "flutter"))]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
let rx = Client::try_start_clipboard(Some(
crate::client::ClientClipboardContext {
cfg: self.handler.get_permission_config(),
tx: self.sender.clone(),
#[cfg(feature = "unix-file-copy-paste")]
is_file_supported: crate::is_support_file_copy_paste(
&peer_version,
),
},
));
// To make sure current text clipboard data is updated.
#[cfg(not(target_os = "ios"))]
if let Some(mut rx) = rx {
timeout(CLIPBOARD_INTERVAL, rx.recv()).await.ok();
}
// `is_connected`` becomes true after the first PeerInfo.
// Reconnects create a new Remote. Refreshes after a generation change
// use a separate entry point and are not blocked by this guard.
#[cfg(not(any(target_os = "android", target_os = "ios")))]
if !self.is_connected
&& self.handler.is_text_clipboard_required()
&& self.handler.lc.read().unwrap().sync_init_clipboard.v
{
self.start_initial_clipboard_sync();
}
// to-do: Android, is `sync_init_clipboard` really needed?
// https://github.com/rustdesk/rustdesk/discussions/9010
}
if self.handler.is_file_transfer() {
self.handler.load_last_jobs();
}
self.is_connected = true;
}
_ => {}
},
Some(message::Union::CursorData(cd)) if self.dedupes_cursors() => {
self.set_cursor_data_by_content(cd);
}
Some(message::Union::CursorId(id)) if self.dedupes_cursors() => {
self.handler
.set_cursor_id(self.cursor_dedupe.id(id).to_string());
}
Some(message::Union::CursorData(cd)) => {
let id = cd.id;
#[cfg(feature = "flutter")]
let compressed = cd.colors.clone();
match decode_cursor_data(cd) {
Ok(cd) => {
#[cfg(feature = "flutter")]
self.keep_cursor_shape(cursor_shape(id, &cd, &compressed));
self.handler.set_cursor_data(cd)
}
Err(err) => {
log::warn!("Rejected cursor {id}: {err}");
// The peer shows it now: selected, the UI treats it as a shape it
// lacks, as it will when the peer selects it again.
self.handler.set_cursor_id(id.to_string());
}
}
}
Some(message::Union::CursorId(id)) => {
self.handler.set_cursor_id(id.to_string());
}
Some(message::Union::CursorPosition(cp)) => {
self.handler.set_cursor_position(cp);
}
Some(message::Union::Clipboard(cb)) => {
let clipboard_allowed = {
let lc = self.handler.lc.read().unwrap();
!lc.disable_clipboard.v && !lc.view_only.v
};
if clipboard_allowed {
#[cfg(all(
feature = "flutter",
not(any(target_os = "android", target_os = "ios"))
))]
if self.handler.is_text_clipboard_required()
&& crate::clipboard::is_sync_clipboard_between_sessions_enabled()
{
let mut msg = Message::new();
msg.set_clipboard(cb.clone());
let session_id = self.handler.lc.read().unwrap().session_id;
crate::flutter::send_clipboard_msg_to_other_sessions(msg, session_id);
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
{
self.initial_clipboard_pending = false;
update_clipboard(vec![cb], ClipboardSide::Client);
}
#[cfg(target_os = "ios")]
{
let content = if cb.compress {
hbb_common::compress::decompress(&cb.content)
} else {
cb.content.into()
};
if let Ok(content) = String::from_utf8(content) {
self.handler.clipboard(content);
}
}
#[cfg(target_os = "android")]
crate::clipboard::handle_msg_clipboard(cb);
}
}
Some(message::Union::MultiClipboards(_mcb)) => {
let clipboard_allowed = {
let lc = self.handler.lc.read().unwrap();
!lc.disable_clipboard.v && !lc.view_only.v
};
if clipboard_allowed {
#[cfg(all(
feature = "flutter",
not(any(target_os = "android", target_os = "ios"))
))]
if self.handler.is_text_clipboard_required()
&& crate::clipboard::is_sync_clipboard_between_sessions_enabled()
{
let mut msg = Message::new();
msg.set_multi_clipboards(_mcb.clone());
let session_id = self.handler.lc.read().unwrap().session_id;
crate::flutter::send_clipboard_msg_to_other_sessions(msg, session_id);
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
{
self.initial_clipboard_pending = false;
update_clipboard(_mcb.clipboards, ClipboardSide::Client);
}
#[cfg(target_os = "ios")]
{
if let Some(cb) = _mcb
.clipboards
.iter()
.find(|c| c.format.enum_value() == Ok(ClipboardFormat::Text))
{
let content = if cb.compress {
hbb_common::compress::decompress(&cb.content)
} else {
cb.content.to_vec()
};
if let Ok(content) = String::from_utf8(content) {
self.handler.clipboard(content);
}
}
}
#[cfg(target_os = "android")]
crate::clipboard::handle_msg_multi_clipboards(_mcb);
}
}
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
Some(message::Union::Cliprdr(clip)) => {
self.handle_cliprdr_msg(clip, peer).await;
}
Some(message::Union::FileResponse(fr)) => {
match fr.union {
Some(file_response::Union::EmptyDirs(res)) => {
self.handler.update_empty_dirs(res);
}
Some(file_response::Union::Dir(fd)) => {
#[cfg(windows)]
let entries = fd.entries.to_vec();
#[cfg(not(windows))]
let mut entries = fd.entries.to_vec();
#[cfg(not(windows))]
{
if self.handler.peer_platform() == "Windows" {
fs::transform_windows_path(&mut entries);
}
}
if fd.id <= 0 {
if let Err(err) = fs::download::validate_listing(
&entries,
self.handler.peer_platform() == "Windows",
) {
self.handle_job_status(fd.id, -1, Some(err.to_string()));
return true;
}
}
// We cannot call cancel_transfer_job/handle_job_status while holding
// a mutable borrow from fs::get_job(&mut self.write_jobs), so defer
// the error handling until after the borrow scope ends.
let mut set_files_err = None;
if let Some(job) = fs::get_job(fd.id, &mut self.write_jobs) {
log::info!("job set_files: {:?}", entries);
if let Err(err) = job.set_files(entries) {
set_files_err = Some(err.to_string());
} else {
job.set_finished_size_on_resume();
self.handler.update_folder_files(
fd.id,
job.files(),
fd.path,
false,
false,
);
}
} else if let Some(job) = self.remove_jobs.get_mut(&fd.id) {
// Intentionally keep raw entries here:
// - remote remove flow executes deletions on peer side;
// - local remove flow is populated from local get_recursive_files().
job.files = entries;
self.handler
.update_folder_files(fd.id, &job.files, fd.path, false, false);
} else {
self.handler
.update_folder_files(fd.id, &entries, fd.path, false, false);
}
if let Some(err) = set_files_err {
log::warn!(
"Rejected unsafe file list from remote peer for job {}: {}",
fd.id,
err
);
self.cancel_transfer_job(fd.id, peer).await;
self.handle_job_status(fd.id, -1, Some(err));
}
}
Some(file_response::Union::Digest(digest)) => {
if digest.is_upload {
if let Some(job) = fs::get_job(digest.id, &mut self.read_jobs) {
if let Some(file) = job.files().get(digest.file_num as usize) {
if let fs::DataSource::FilePath(p) = &job.data_source {
let read_path =
get_string(&fs::TransferJob::join(p, &file.name));
let mut overwrite_strategy =
job.default_overwrite_strategy();
let mut offset = 0;
if digest.is_identical && job.is_resume {
if digest.transferred_size > 0 {
overwrite_strategy = Some(true);
offset = digest.transferred_size as _;
}
}
if let Some(overwrite) = overwrite_strategy {
let req = FileTransferSendConfirmRequest {
id: digest.id,
file_num: digest.file_num,
union: Some(if overwrite {
file_transfer_send_confirm_request::Union::OffsetBlk(offset)
} else {
file_transfer_send_confirm_request::Union::Skip(
true,
)
}),
..Default::default()
};
job.confirm(&req).await;
let msg = new_send_confirm(req);
allow_err!(peer.send(&msg).await);
} else {
self.handler.override_file_confirm(
digest.id,
digest.file_num,
read_path,
true,
digest.is_identical,
);
}
}
}
}
} else {
if let Some(job) = fs::get_job(digest.id, &mut self.write_jobs) {
if let Some(file) = job.files().get(digest.file_num as usize) {
if let fs::DataSource::FilePath(p) = &job.data_source {
let write_path =
get_string(&fs::TransferJob::join(p, &file.name));
job.set_digest(digest.file_size, digest.last_modified);
let peer_ver = self.handler.lc.read().unwrap().version;
let is_support_resume =
crate::is_support_file_transfer_resume_num(
peer_ver,
);
match job.download_digest(is_support_resume, &digest) {
Ok(res) => match res {
DigestCheckResult::IsSame => {
let req = FileTransferSendConfirmRequest {
id: digest.id,
file_num: digest.file_num,
union: Some(file_transfer_send_confirm_request::Union::Skip(true)),
..Default::default()
};
job.confirm(&req).await;
let msg = new_send_confirm(req);
allow_err!(peer.send(&msg).await);
}
DigestCheckResult::NeedConfirm(digest) => {
let mut overwrite_strategy =
job.default_overwrite_strategy();
let mut offset = 0;
if digest.is_identical
&& job.is_resume
&& digest.transferred_size > 0
{
overwrite_strategy = Some(true);
offset = digest.transferred_size as _;
}
if let Some(overwrite) = overwrite_strategy
{
let req =
FileTransferSendConfirmRequest {
id: digest.id,
file_num: digest.file_num,
union: Some(if overwrite {
file_transfer_send_confirm_request::Union::OffsetBlk(offset)
} else {
file_transfer_send_confirm_request::Union::Skip(true)
}),
..Default::default()
};
job.confirm(&req).await;
let msg = new_send_confirm(req);
allow_err!(peer.send(&msg).await);
} else {
self.handler.override_file_confirm(
digest.id,
digest.file_num,
write_path,
false,
digest.is_identical,
);
}
}
DigestCheckResult::NoSuchFile => {
let req = FileTransferSendConfirmRequest {
id: digest.id,
file_num: digest.file_num,
union: Some(file_transfer_send_confirm_request::Union::OffsetBlk(0)),
..Default::default()
};
// This did not approve replacing an existing local file.
let msg = new_send_confirm(req);
allow_err!(peer.send(&msg).await);
}
},
Err(err) => {
self.cancel_transfer_job(digest.id, peer).await;
self.handle_job_status(
digest.id,
-1,
Some(err.to_string()),
);
return true;
}
}
}
}
}
}
}
Some(file_response::Union::Block(block)) => {
let id = block.id;
let mut write_error = None;
if let Some(job) = fs::get_job(block.id, &mut self.write_jobs) {
let result = job.write(block).await;
if job.r#type == fs::JobType::Generic {
write_error = result.err().map(|err| err.to_string());
self.update_jobs_status();
}
}
if let Some(err) = write_error {
self.cancel_transfer_job(id, peer).await;
self.handle_job_status(id, -1, Some(err));
}
}
Some(file_response::Union::Done(d)) => {
let mut err: Option<String> = None;
let mut job_type = fs::JobType::Generic;
let mut printer_data = None;
if let Some(job) = fs::remove_job(d.id, &mut self.write_jobs) {
err = job
.finish_download()
.err()
.map(|err| err.to_string())
.or_else(|| job.job_error());
job_type = job.r#type;
printer_data = match job.get_buf_data().await {
Ok(d) => d,
Err(e) => {
log::error!("Failed to get the printer data: {}", e);
None
}
};
}
match job_type {
fs::JobType::Generic => {
self.handle_job_status(d.id, d.file_num, err);
}
fs::JobType::Printer => {
if let Some(err) = err {
log::error!("Receive print job failed, error {err}");
} else {
log::info!(
"Receive print job done, data len: {:?}",
printer_data.as_ref().map(|d| d.len()).unwrap_or(0)
);
#[cfg(target_os = "windows")]
if let Some(data) = printer_data {
let printer_name = self
.handler
.printer_names
.write()
.unwrap()
.remove(&d.id);
// Spawn a new thread to handle the print job.
// Or print job will block the ui thread.
std::thread::spawn(move || {
if let Err(e) =
crate::platform::send_raw_data_to_printer(
printer_name,
data,
)
{
log::error!("Print job error: {}", e);
}
});
}
}
}
}
}
Some(file_response::Union::Error(e)) => {
let job_type = fs::remove_job(e.id, &mut self.write_jobs)
.or_else(|| fs::remove_job(e.id, &mut self.read_jobs))
.map(|j| j.r#type)
.unwrap_or(fs::JobType::Generic);
match job_type {
fs::JobType::Generic => {
self.handle_job_status(e.id, e.file_num, Some(e.error));
}
fs::JobType::Printer => {
log::error!("Printer job error: {}", e.error);
}
}
}
_ => {}
}
}
Some(message::Union::Misc(misc)) => match misc.union {
Some(misc::Union::AudioFormat(f)) => {
self.audio_sender.send(MediaData::AudioFormat(f)).ok();
}
Some(misc::Union::ChatMessage(c)) => {
self.handler.new_message(c.text);
}
Some(misc::Union::PermissionInfo(p)) => {
log::info!("Change permission {:?} -> {}", p.permission, p.enabled);
// https://github.com/rustdesk/rustdesk/issues/3703#issuecomment-1474734754
match p.permission.enum_value() {
Ok(Permission::Keyboard) => {
*self.handler.server_keyboard_enabled.write().unwrap() = p.enabled;
#[cfg(feature = "flutter")]
#[cfg(not(target_os = "ios"))]
crate::flutter::update_text_clipboard_required();
#[cfg(all(feature = "flutter", feature = "unix-file-copy-paste"))]
crate::flutter::update_file_clipboard_required();
self.handler.set_permission("keyboard", p.enabled);
}
Ok(Permission::Clipboard) => {
*self.handler.server_clipboard_enabled.write().unwrap() = p.enabled;
#[cfg(feature = "flutter")]
#[cfg(not(target_os = "ios"))]
crate::flutter::update_text_clipboard_required();
self.handler.set_permission("clipboard", p.enabled);
}
Ok(Permission::Audio) => {
self.handler.set_permission("audio", p.enabled);
}
Ok(Permission::File) => {
*self.handler.server_file_transfer_enabled.write().unwrap() =
p.enabled;
if !p.enabled && self.handler.is_file_transfer() {
return true;
}
#[cfg(all(feature = "flutter", feature = "unix-file-copy-paste"))]
crate::flutter::update_file_clipboard_required();
self.handler.set_permission("file", p.enabled);
#[cfg(feature = "unix-file-copy-paste")]
if !p.enabled {
try_empty_clipboard_files(
ClipboardSide::Client,
self.client_conn_id,
);
}
}
Ok(Permission::Restart) => {
self.handler.set_permission("restart", p.enabled);
}
Ok(Permission::Recording) => {
self.handler.lc.write().unwrap().record_permission = p.enabled;
self.update_record_state();
self.handler.set_permission("recording", p.enabled);
}
Ok(Permission::BlockInput) => {
self.handler.set_permission("block_input", p.enabled);
}
Ok(Permission::PrivacyMode) => {
self.handler.set_permission("privacy_mode", p.enabled);
}
_ => {}
}
}
Some(misc::Union::SwitchDisplay(s)) => {
self.handler.handle_peer_switch_display(&s);
if let Some(thread) = self.video_threads.get_mut(&(s.display as usize)) {
thread.video_sender.send(MediaData::Reset).ok();
}
let mut scale = 1.0;
if let Some(pi) = &self.handler.lc.read().unwrap().peer_info {
if let Some(d) = pi.displays.get(s.display as usize) {
scale = d.scale;
}
}
if s.width > 0 && s.height > 0 {
self.handler.set_display(
s.x,
s.y,
s.width,
s.height,
s.cursor_embedded,
scale,
);
}
}
Some(misc::Union::CloseReason(c)) => {
self.sent_close_reason = true; // The controlled end will close, no need to send close reason
self.handler.msgbox("error", "Connection Error", &c, "");
return false;
}
Some(misc::Union::BackNotification(notification)) => {
if !self.handle_back_notification(notification).await {
return false;
}
}
Some(misc::Union::Uac(uac)) => {
let keyboard = self.handler.server_keyboard_enabled.read().unwrap().clone();
#[cfg(feature = "flutter")]
{
if uac && keyboard {
self.handler.msgbox(
"on-uac",
"Prompt",
"Please wait for confirmation of UAC...",
"",
);
} else {
self.handler.cancel_msgbox("on-uac");
self.handler.cancel_msgbox("wait-uac");
self.handler.cancel_msgbox("elevation-error");
}
}
#[cfg(not(feature = "flutter"))]
{
let msgtype = "custom-uac-nocancel";
let title = "Prompt";
let text = "Please wait for confirmation of UAC...";
let link = "";
if uac && keyboard {
self.handler.msgbox(msgtype, title, text, link);
} else {
self.handler.cancel_msgbox(&format!(
"{}-{}-{}-{}",
msgtype, title, text, link,
));
}
}
}
Some(misc::Union::ForegroundWindowElevated(elevated)) => {
let keyboard = self.handler.server_keyboard_enabled.read().unwrap().clone();
#[cfg(feature = "flutter")]
{
if elevated && keyboard {
self.handler.msgbox(
"on-foreground-elevated",
"Prompt",
"elevated_foreground_window_tip",
"",
);
} else {
self.handler.cancel_msgbox("on-foreground-elevated");
self.handler.cancel_msgbox("wait-uac");
self.handler.cancel_msgbox("elevation-error");
}
}
#[cfg(not(feature = "flutter"))]
{
let msgtype = "custom-elevated-foreground-nocancel";
let title = "Prompt";
let text = "elevated_foreground_window_tip";
let link = "";
if elevated && keyboard {
self.handler.msgbox(msgtype, title, text, link);
} else {
self.handler.cancel_msgbox(&format!(
"{}-{}-{}-{}",
msgtype, title, text, link,
));
}
}
}
Some(misc::Union::ElevationResponse(err)) => {
if err.is_empty() {
self.handler.msgbox("wait-uac", "", "", "");
} else {
self.handler.cancel_msgbox("wait-uac");
self.handler
.msgbox("elevation-error", "Elevation Error", &err, "");
}
}
Some(misc::Union::PortableServiceRunning(b)) => {
self.handler.portable_service_running(b);
if self.elevation_requested && b {
self.handler.msgbox(
"custom-nocancel-success",
"Successful",
"Elevate successfully",
"",
);
}
}
#[cfg(feature = "flutter")]
#[cfg(not(any(target_os = "android", target_os = "ios")))]
Some(misc::Union::SwitchBack(_)) => {
let allow_switch_back = self
.handler
.lc
.write()
.unwrap()
.consume_switch_back_permission();
if allow_switch_back {
self.handler.switch_back(&self.handler.get_id());
} else {
log::warn!(
"Ignored unsolicited SwitchBack from {}",
self.handler.get_id()
);
}
}
Some(misc::Union::SupportedEncoding(e)) => {
log::info!("update supported encoding:{:?}", e);
self.handler.lc.write().unwrap().supported_encoding = e;
}
Some(misc::Union::FollowCurrentDisplay(d_idx)) => {
self.handler.set_current_display(d_idx);
}
_ => {}
},
Some(message::Union::TestDelay(t)) => {
self.handler.handle_test_delay(t, peer).await;
}
Some(message::Union::AudioFrame(frame)) => {
if !self.handler.lc.read().unwrap().disable_audio.v {
self.audio_sender
.send(MediaData::AudioFrame(Box::new(frame)))
.ok();
}
}
Some(message::Union::FileAction(action)) => match action.union {
Some(file_action::Union::Send(_s)) => match _s.file_type.enum_value() {
#[cfg(target_os = "windows")]
Ok(file_transfer_send_request::FileType::Printer) => {
#[cfg(feature = "flutter")]
let action =
LocalConfig::get_option(keys::OPTION_PRINTER_INCOMING_JOB_ACTION);
#[cfg(not(feature = "flutter"))]
let action = "";
if action == "dismiss" {
// Just ignore the incoming print job.
} else {
let id = fs::get_next_job_id();
#[cfg(feature = "flutter")]
let allow_auto_print = LocalConfig::get_bool_option(
keys::OPTION_PRINTER_ALLOW_AUTO_PRINT,
);
#[cfg(not(feature = "flutter"))]
let allow_auto_print = false;
if allow_auto_print {
let printer_name = if action == "" {
"".to_string()
} else {
LocalConfig::get_option(keys::OPTION_PRINTER_SELECTED_NAME)
};
self.handler.printer_response(id, _s.path, printer_name);
} else {
self.handler.printer_request(id, _s.path);
}
}
}
_ => {}
},
Some(file_action::Union::SendConfirm(c)) => {
if let Some(job) = fs::get_job(c.id, &mut self.read_jobs) {
job.confirm(&c).await;
}
}
_ => {}
},
Some(message::Union::MessageBox(msgbox)) => {
let mut link = msgbox.link;
if let Some(v) = config::HELPER_URL.get(&link as &str) {
link = v.to_string();
} else {
log::warn!("Message box ignore link {} for security", &link);
link = "".to_string();
}
self.handler
.msgbox(&msgbox.msgtype, &msgbox.title, &msgbox.text, &link);
}
Some(message::Union::VoiceCallRequest(request)) => {
if request.is_connect {
// TODO: maybe we will do a voice call from the peer in the future.
} else {
log::debug!("The remote has requested to close the voice call");
if let Some(sender) = self.stop_voice_call_sender.take() {
allow_err!(sender.send(()));
self.handler.on_voice_call_closed("");
}
}
}
Some(message::Union::VoiceCallResponse(response)) => {
let ts = std::mem::replace(&mut self.voice_call_request_timestamp, None);
if let Some(ts) = ts {
if response.req_timestamp != ts.get() {
log::debug!("Possible encountering a voice call attack.");
} else {
if response.accepted {
// The peer accepted the voice call.
self.handler.on_voice_call_started();
self.stop_voice_call_sender = self.start_voice_call();
} else {
// The peer refused the voice call.
self.handler.on_voice_call_closed("");
}
}
}
}
Some(message::Union::PeerInfo(pi)) => {
self.handler.set_displays(&pi.displays);
self.handler.set_platform_additions(&pi.platform_additions);
}
Some(message::Union::ScreenshotResponse(response)) => {
crate::client::screenshot::set_screenshot(response.data);
self.handler
.handle_screenshot_resp(response.sid, response.msg);
}
Some(message::Union::TerminalResponse(response)) => {
use base::message_proto::terminal_response::Union;
if let Some(Union::Opened(opened)) = &response.union {
if opened.success && !opened.service_id.is_empty() {
let mut lc = self.handler.lc.write().unwrap();
let key = lc.get_key_terminal_service_id().to_owned();
lc.set_option(key, opened.service_id.clone());
}
}
self.handler.handle_terminal_response(response);
}
_ => {}
}
}
true
}
fn set_peer_info(&mut self, pi: &PeerInfo) {
self.last_auto_fps = None;
self.peer_info.platform = pi.platform.clone();
// Check features field for terminal support
if let Some(features) = pi.features.as_ref() {
self.peer_info.support_terminal = features.terminal;
}
if let Ok(platform_additions) =
serde_json::from_str::<HashMap<String, serde_json::Value>>(&pi.platform_additions)
{
self.peer_info.is_installed = platform_additions
.get("is_installed")
.map(|v| v.as_bool())
.flatten()
.unwrap_or(false);
self.peer_info.idd_impl = platform_additions
.get("idd_impl")
.map(|v| v.as_str())
.flatten()
.unwrap_or_default()
.to_string();
self.peer_info.support_view_camera = platform_additions
.get("support_view_camera")
.map(|v| v.as_bool())
.flatten()
.unwrap_or(false);
}
}
async fn handle_back_notification(&mut self, notification: BackNotification) -> bool {
match notification.union {
Some(back_notification::Union::BlockInputState(state)) => {
self.handle_back_msg_block_input(
state.enum_value_or(back_notification::BlockInputState::BlkStateUnknown),
notification.details,
)
.await;
}
Some(back_notification::Union::PrivacyModeState(state)) => {
if !self
.handle_back_msg_privacy_mode(
state.enum_value_or(back_notification::PrivacyModeState::PrvStateUnknown),
notification.details,
notification.impl_key,
)
.await
{
return false;
}
}
_ => {}
}
true
}
#[inline(always)]
fn update_block_input_state(&mut self, on: bool) {
self.handler.update_block_input_state(on);
}
async fn handle_back_msg_block_input(
&mut self,
state: back_notification::BlockInputState,
details: String,
) {
match state {
back_notification::BlockInputState::BlkOnSucceeded => {
self.update_block_input_state(true);
}
back_notification::BlockInputState::BlkOnFailed => {
self.handler.msgbox(
"custom-error",
"Block user input",
if details.is_empty() {
"Failed"
} else {
&details
},
"",
);
self.update_block_input_state(false);
}
back_notification::BlockInputState::BlkOffSucceeded => {
self.update_block_input_state(false);
}
back_notification::BlockInputState::BlkOffFailed => {
self.handler.msgbox(
"custom-error",
"Unblock user input",
if details.is_empty() {
"Failed"
} else {
&details
},
"",
);
}
_ => {}
}
}
#[inline(always)]
fn update_privacy_mode(&mut self, impl_key: String, on: bool) {
let mut config = self.handler.load_config();
config.privacy_mode.v = on;
if on {
// For compatibility, version < 1.2.4, the default value is 'privacy_mode_impl_mag'.
let impl_key = if impl_key.is_empty() {
"privacy_mode_impl_mag".to_string()
} else {
impl_key
};
config
.options
.insert("privacy-mode-impl-key".to_string(), impl_key);
}
self.handler.save_config(config);
self.handler.update_privacy_mode();
}
async fn handle_back_msg_privacy_mode(
&mut self,
state: back_notification::PrivacyModeState,
details: String,
impl_key: String,
) -> bool {
match state {
back_notification::PrivacyModeState::PrvOnByOther => {
self.handler.msgbox(
"error",
"Connecting...",
"Someone turns on privacy mode, exit",
"",
);
return false;
}
back_notification::PrivacyModeState::PrvNotSupported => {
self.handler
.msgbox("custom-error", "Privacy mode", "Unsupported", "");
self.update_privacy_mode(impl_key, false);
}
back_notification::PrivacyModeState::PrvOnSucceeded => {
self.handler
.msgbox("custom-nocancel", "Privacy mode", "Enter privacy mode", "");
self.update_privacy_mode(impl_key, true);
}
back_notification::PrivacyModeState::PrvOnFailedDenied => {
self.handler
.msgbox("custom-error", "Privacy mode", "Peer denied", "");
self.update_privacy_mode(impl_key, false);
}
back_notification::PrivacyModeState::PrvOnFailedPlugin
| back_notification::PrivacyModeState::PrvOnFailed => {
self.handler.msgbox(
"custom-error",
"Privacy mode",
if details.is_empty() {
"Failed"
} else {
&details
},
"",
);
self.update_privacy_mode(impl_key, false);
}
back_notification::PrivacyModeState::PrvOffSucceeded => {
self.handler
.msgbox("custom-nocancel", "Privacy mode", "Exit privacy mode", "");
self.update_privacy_mode(impl_key, false);
}
back_notification::PrivacyModeState::PrvOffByPeer => {
self.handler
.msgbox("custom-error", "Privacy mode", "Peer exit", "");
self.update_privacy_mode(impl_key, false);
}
back_notification::PrivacyModeState::PrvOffFailed => {
self.handler.msgbox(
"custom-error",
"Privacy mode",
if details.is_empty() {
"Failed to turn off"
} else {
&details
},
"",
);
}
back_notification::PrivacyModeState::PrvOffUnknown => {
self.handler
.msgbox("custom-error", "Privacy mode", "Turned off", "");
// log::error!("Privacy mode is turned off with unknown reason");
self.update_privacy_mode(impl_key, false);
}
_ => {}
}
true
}
#[cfg(all(target_os = "windows", not(feature = "flutter")))]
fn check_clipboard_file_context(&self) {
let enabled = *self.handler.server_file_transfer_enabled.read().unwrap()
&& self.handler.lc.read().unwrap().enable_file_copy_paste.v;
ContextSend::enable(enabled);
}
#[cfg(any(target_os = "windows", feature = "unix-file-copy-paste"))]
async fn handle_cliprdr_msg(&mut self, clip: base::message_proto::Cliprdr, _peer: &mut Stream) {
if !self.is_connected {
log::debug!("Discarding incoming file clipboard message before login");
return;
}
log::debug!("handling cliprdr msg from server peer");
#[cfg(feature = "flutter")]
if let Some(base::message_proto::cliprdr::Union::FormatList(_)) = &clip.union {
if self.client_conn_id
!= clipboard::get_client_conn_id(&crate::flutter::get_cur_peer_id()).unwrap_or(0)
{
return;
}
}
let Some(clip) = crate::clipboard_file::msg_2_clip(clip) else {
log::warn!("failed to decode cliprdr msg from server peer");
return;
};
let is_stopping_allowed = clip.is_beginning_message();
let file_transfer_enabled = self.handler.is_file_clipboard_required();
let stop = is_stopping_allowed && !file_transfer_enabled;
log::debug!(
"Process clipboard message from server peer, stop: {}, is_stopping_allowed: {}, file_transfer_enabled: {}",
stop, is_stopping_allowed, file_transfer_enabled);
if !stop {
#[cfg(any(
target_os = "windows",
all(target_os = "macos", feature = "unix-file-copy-paste")
))]
if let Err(e) = ContextSend::make_sure_enabled() {
log::error!("failed to restart clipboard context: {}", e);
};
#[cfg(target_os = "windows")]
{
if matches!(&clip, clipboard::ClipboardFile::FormatList { .. }) {
self.initial_clipboard_pending = false;
}
let _ = ContextSend::proc(|context| -> ResultType<()> {
context
.server_clip_file(self.client_conn_id, clip)
.map_err(|e| e.into())
});
}
#[cfg(feature = "unix-file-copy-paste")]
if crate::is_support_file_copy_paste_num(self.handler.lc.read().unwrap().version) {
#[cfg(not(any(target_os = "android", target_os = "ios")))]
if matches!(&clip, clipboard::ClipboardFile::FormatList { .. }) {
self.initial_clipboard_pending = false;
}
let mut out_msgs = vec![];
#[cfg(target_os = "macos")]
if clipboard::platform::unix::macos::should_handle_msg(&clip) {
if let Err(e) = ContextSend::proc(|context| -> ResultType<()> {
context
.server_clip_file(self.client_conn_id, clip)
.map_err(|e| e.into())
}) {
log::error!("failed to handle cliprdr msg: {}", e);
}
} else {
out_msgs = unix_file_clip::serve_clip_messages(
ClipboardSide::Client,
clip,
self.client_conn_id,
);
}
#[cfg(not(target_os = "macos"))]
{
out_msgs = unix_file_clip::serve_clip_messages(
ClipboardSide::Client,
clip,
self.client_conn_id,
);
}
for msg in out_msgs.into_iter() {
allow_err!(_peer.send(&msg).await);
}
}
}
}
fn new_video_thread(&mut self, display: usize) {
let video_queue = Arc::new(RwLock::new(ArrayQueue::new(client::VIDEO_QUEUE_SIZE)));
let (video_sender, video_receiver) = std::sync::mpsc::channel::<MediaData>();
let decode_fps = Arc::new(RwLock::new(None));
let frame_count = Arc::new(RwLock::new(0));
let discard_queue = Arc::new(RwLock::new(false));
let video_thread = VideoThread {
video_queue: video_queue.clone(),
video_sender,
decode_fps: decode_fps.clone(),
frame_count: frame_count.clone(),
fps_control: Default::default(),
discard_queue: discard_queue.clone(),
};
let handler = self.handler.ui_handler.clone();
crate::client::start_video_thread(
self.handler.clone(),
display,
video_receiver,
video_queue,
decode_fps,
self.chroma.clone(),
discard_queue,
move |display: usize,
data: &mut scrap::ImageRgb,
_texture: *mut c_void,
pixelbuffer: bool| {
*frame_count.write().unwrap() += 1;
if pixelbuffer {
handler.on_rgba(display, data);
} else {
#[cfg(all(feature = "vram", feature = "flutter"))]
handler.on_texture(display, _texture);
}
},
);
self.video_threads.insert(display, video_thread);
if self.video_threads.len() == 1 {
let auto_record = LocalConfig::get_bool_option(keys::OPTION_ALLOW_AUTO_RECORD_OUTGOING);
self.handler.lc.write().unwrap().record_state = auto_record;
self.update_record_state();
}
}
fn update_record_state(&mut self) {
// state
let permission = self.handler.lc.read().unwrap().record_permission;
if !permission {
self.handler.lc.write().unwrap().record_state = false;
}
let state = self.handler.lc.read().unwrap().record_state;
let start = state && permission;
if self.last_record_state == start {
return;
}
self.last_record_state = start;
log::info!("record screen start: {start}");
// update local
for (_, v) in self.video_threads.iter_mut() {
v.video_sender.send(MediaData::RecordScreen(start)).ok();
}
self.handler.update_record_status(start);
// update remote
let mut misc = Misc::new();
misc.set_client_record_status(start);
let mut msg = Message::new();
msg.set_misc(misc);
self.sender.send(Data::Message(msg)).ok();
}
/// A shape that decoded, for the UI to ask for again; see `Session::cursor_shapes`.
#[cfg(feature = "flutter")]
fn keep_cursor_shape(&self, shape: CursorData) {
self.handler
.cursor_shapes
.write()
.unwrap()
.insert(shape.id, shape);
}
fn dedupes_cursors(&self) -> bool {
!crate::is_peer_naming_cursors_by_content(self.handler.lc.read().unwrap().version)
}
/// A shape the UI already has, under whatever handle, is only selected.
fn set_cursor_data_by_content(&mut self, cd: CursorData) {
let peer_id = cd.id;
let id = CursorDedupe::name(&cd);
if self.cursor_dedupe.has_decoded(id) {
self.cursor_dedupe.record(peer_id, id);
self.handler.set_cursor_id(id.to_string());
return;
}
#[cfg(feature = "flutter")]
let compressed = cd.colors.clone();
match decode_cursor_data(cd) {
Ok(mut cd) => {
cd.id = id;
self.cursor_dedupe.record(peer_id, id);
#[cfg(feature = "flutter")]
self.keep_cursor_shape(cursor_shape(id, &cd, &compressed));
self.handler.set_cursor_data(cd);
}
Err(err) => {
log::warn!("Rejected cursor {peer_id}: {err}");
self.handler
.set_cursor_id(self.cursor_dedupe.rejected(peer_id, id).to_string());
}
}
}
}
/// A shape that decoded, still compressed, under the id the UI knows it by. Copied once it
/// decoded: the message's colors may be a slice of a larger received buffer, which a clone
/// would keep alive.
#[cfg(feature = "flutter")]
fn cursor_shape(id: u64, cd: &CursorData, compressed: &[u8]) -> CursorData {
CursorData {
id,
hotx: cd.hotx,
hoty: cd.hoty,
width: cd.width,
height: cd.height,
colors: compressed.to_vec().into(),
..Default::default()
}
}
/// The RGBA of a shape kept by `Session::cursor_shapes`, for the UI to draw it again.
#[cfg(any(feature = "flutter", test))]
pub(crate) fn kept_cursor_rgba(shape: CursorData) -> hbb_common::ResultType<CursorData> {
decode_cursor_data(shape)
}
// Both UI handlers receive validated, uncompressed RGBA from the receive loop.
fn decode_cursor_data(data: CursorData) -> hbb_common::ResultType<CursorData> {
use hbb_common::{anyhow::anyhow, bail};
// Limit decoded cursor data to 1 MiB before JSON serialization.
const MAX_CURSOR_SIZE: i32 = 512;
const RGBA_CHANNELS: usize = 4;
let mut cd = data;
if !(1..=MAX_CURSOR_SIZE).contains(&cd.width) || !(1..=MAX_CURSOR_SIZE).contains(&cd.height) {
bail!("invalid source size {}x{}", cd.width, cd.height);
}
if !(0..cd.width).contains(&cd.hotx) || !(0..cd.height).contains(&cd.hoty) {
bail!(
"hotspot ({},{}) is outside the cursor image",
cd.hotx,
cd.hoty
);
}
let expected = (cd.width as usize)
.checked_mul(cd.height as usize)
.and_then(|pixels| pixels.checked_mul(RGBA_CHANNELS))
.ok_or_else(|| anyhow!("cursor RGBA size overflow"))?;
let max_compressed_size = zstd::zstd_safe::compress_bound(expected);
if cd.colors.len() > max_compressed_size {
bail!(
"compressed cursor data too large: {} bytes (limit {max_compressed_size})",
cd.colors.len()
);
}
let colors = zstd::bulk::decompress(&cd.colors, expected)?;
if colors.len() != expected {
bail!(
"invalid RGBA length: expected {expected}, got {}",
colors.len()
);
}
cd.colors = colors.into();
Ok(cd)
}
/// Gives the UI one id per shape for a peer that names shapes by handle, so the UI decodes and
/// keeps a shape once however many handles it arrives under. Nothing is dropped for the
/// connection: the peer sends a shape once and may select any handle it named again.
#[derive(Default)]
struct CursorDedupe {
ids: HashMap<u64, u64>,
// The shapes given to the UI, by content id: whether one that arrives needs decoding.
decoded: HashSet<u64>,
}
impl CursorDedupe {
/// Hashes the compressed colors: one peer compresses the same pixels to the same bytes, and
/// a shape seen before is then never decompressed again.
fn name(cd: &CursorData) -> u64 {
crate::cursor_content_id(cd.width, cd.height, cd.hotx, cd.hoty, &cd.colors)
}
fn id(&self, peer_id: u64) -> u64 {
self.ids.get(&peer_id).copied().unwrap_or(peer_id)
}
fn has_decoded(&self, id: u64) -> bool {
self.decoded.contains(&id)
}
/// Names the handle once its shape decoded, now or before.
fn record(&mut self, peer_id: u64, id: u64) {
self.ids.insert(peer_id, id);
self.decoded.insert(id);
}
/// Names a known handle by the content of a shape that did not decode, which is never
/// marked decoded: the UI has no shape under it, now or when the handle is selected again,
/// and the shape the handle named before is not shown in its place. A handle never seen is
/// not filed, so what a peer sends that does not decode takes no room.
fn rejected(&mut self, peer_id: u64, id: u64) -> u64 {
if let Some(named) = self.ids.get_mut(&peer_id) {
*named = id;
}
id
}
}
struct RemoveJob {
files: Vec<FileEntry>,
path: String,
sep: &'static str,
is_remote: bool,
no_confirm: bool,
last_update_job_status: Instant,
}
impl RemoveJob {
fn new(files: Vec<FileEntry>, path: String, sep: &'static str, is_remote: bool) -> Self {
Self {
files,
path,
sep,
is_remote,
no_confirm: false,
last_update_job_status: Instant::now(),
}
}
pub fn _gen_meta(&self) -> RemoveJobMeta {
RemoveJobMeta {
path: self.path.clone(),
is_remote: self.is_remote,
no_confirm: self.no_confirm,
}
}
}
#[derive(Debug, Default)]
struct FpsControl {
refresh_times: usize,
last_refresh_instant: Option<Instant>,
idle_counter: usize,
inactive_counter: usize,
}
struct VideoThread {
video_queue: Arc<RwLock<ArrayQueue<VideoFrame>>>,
video_sender: MediaSender,
decode_fps: Arc<RwLock<Option<usize>>>,
frame_count: Arc<RwLock<usize>>,
discard_queue: Arc<RwLock<bool>>,
fps_control: FpsControl,
}
impl Drop for VideoThread {
fn drop(&mut self) {
// since channels are buffered, messages sent before the disconnect will still be properly received.
*self.discard_queue.write().unwrap() = true;
}
}
#[cfg(test)]
#[cfg(feature = "flutter")]
mod tests {
use super::*;
use crate::flutter::FlutterHandler;
/// A round's `Remote` over a loopback pair, before any login: what it sends to `peer`
/// arrives at `far`.
async fn remote_and_peer() -> (Remote<FlutterHandler>, Stream, Stream) {
let listener = hbb_common::tcp::new_listener("127.0.0.1:0", false)
.await
.unwrap();
let addr = listener.local_addr().unwrap();
let (peer, accepted) = tokio::join!(
hbb_common::socket_client::connect_tcp(addr.to_string(), 3000),
listener.accept()
);
let (accepted, far_addr) = accepted.unwrap();
let far = Stream::Tcp(hbb_common::tcp::FramedStream::from(accepted, far_addr));
let (sender, receiver) = mpsc::unbounded_channel::<Data>();
let remote = Remote::new(Session::<FlutterHandler>::default(), receiver, sender);
(remote, peer.unwrap(), far)
}
async fn arrives(far: &mut Stream) -> bool {
matches!(hbb_common::timeout(300, far.next()).await, Ok(Some(Ok(_))))
}
fn clipboard() -> Data {
let mut msg = Message::new();
msg.set_clipboard(Clipboard {
content: b"copied while this login was pending".to_vec().into(),
..Default::default()
});
Data::Message(msg)
}
fn auth_2fa() -> Data {
let mut msg = Message::new();
msg.set_auth_2fa(Auth2FA {
code: "123456".to_owned(),
..Default::default()
});
Data::Message(msg)
}
// A clipboard queued before this round's login stays here; what the login itself sends
// through the same queue does not.
#[tokio::test]
async fn a_clipboard_queued_before_this_rounds_login_is_dropped() {
let (mut remote, mut peer, mut far) = remote_and_peer().await;
assert!(!remote.is_connected);
assert!(remote.handle_msg_from_ui(clipboard(), &mut peer).await);
assert!(
!arrives(&mut far).await,
"a clipboard went out before the login"
);
assert!(remote.handle_msg_from_ui(auth_2fa(), &mut peer).await);
assert!(arrives(&mut far).await, "the 2FA code was held back");
remote.is_connected = true;
assert!(remote.handle_msg_from_ui(clipboard(), &mut peer).await);
assert!(
arrives(&mut far).await,
"a clipboard after the login was held back"
);
}
#[tokio::test]
async fn manual_fps_update_allows_the_same_automatic_limit_again() {
let (mut remote, mut peer, mut far) = remote_and_peer().await;
let (video_sender, _video_receiver) = std::sync::mpsc::channel();
remote.video_threads.insert(
0,
VideoThread {
video_queue: Arc::new(RwLock::new(ArrayQueue::new(client::VIDEO_QUEUE_SIZE))),
video_sender,
decode_fps: Arc::new(RwLock::new(Some(23))),
frame_count: Default::default(),
discard_queue: Default::default(),
fps_control: Default::default(),
},
);
for (manual_fps, expected) in [(None, 20), (Some(90), 90), (None, 20)] {
if let Some(fps) = manual_fps {
let lc = &remote.handler.lc;
let msg = lc.write().unwrap().set_custom_fps(fps, false);
remote
.handle_msg_from_ui(Data::Message(msg), &mut peer)
.await;
} else {
remote
.fps_control(true, HashMap::from([(0, 20)]), &mut peer)
.await;
}
let bytes = timeout(1000, far.next())
.await
.expect("FPS update was not sent")
.unwrap()
.unwrap();
let msg = Message::parse_from_bytes(&bytes).unwrap();
assert_eq!(msg.misc().option().custom_fps, expected);
}
}
}
#[cfg(test)]
mod cursor_dedupe_tests {
use super::*;
fn shape(id: u64, hotx: i32, colors: &[u8]) -> CursorData {
CursorData {
id,
hotx,
width: 2,
height: 2,
colors: colors.to_vec().into(),
..Default::default()
}
}
#[test]
fn a_shape_sent_under_a_new_handle_keeps_its_id() {
let mut dedupe = CursorDedupe::default();
let first = CursorDedupe::name(&shape(1, 0, b"arrow"));
assert!(!dedupe.has_decoded(first));
dedupe.record(1, first);
let again = CursorDedupe::name(&shape(2, 0, b"arrow"));
assert_eq!(again, first);
assert!(dedupe.has_decoded(again));
dedupe.record(2, again);
assert_eq!(dedupe.id(2), first);
assert_eq!(dedupe.id(1), first);
}
}
#[cfg(test)]
mod kept_cursor_tests {
use super::*;
fn compressed(width: i32, height: i32) -> CursorData {
let rgba = vec![7u8; (width * height * 4) as usize];
CursorData {
id: 1,
width,
height,
colors: hbb_common::compress::compress(&rgba).into(),
..Default::default()
}
}
#[test]
fn a_kept_shape_is_checked_again_when_it_is_read() {
assert_eq!(kept_cursor_rgba(compressed(4, 4)).unwrap().colors.len(), 64);
assert!(
kept_cursor_rgba(compressed(513, 1)).is_err(),
"over the size cap"
);
let mut bomb = compressed(4, 4);
bomb.colors = hbb_common::compress::compress(&vec![0u8; 1 << 20]).into();
assert!(kept_cursor_rgba(bomb).is_err(), "more pixels than its size");
}
}
Reference in new issue View Git Blame Copy Permalink