src/server.rs
Rama master · réplica de lectura del repositorio original.
use std::{
collections::HashMap,
net::SocketAddr,
sync::{Arc, Mutex, RwLock, Weak},
time::Duration,
};
use bytes::Bytes;
pub use connection::*;
#[cfg(not(any(target_os = "android", target_os = "ios")))]
use hbb_common::config::Config2;
use hbb_common::tcp::{self, new_listener};
use hbb_common::{
allow_err,
anyhow::Context,
bail,
config::{Config, CONNECT_TIMEOUT, RELAY_PORT},
log,
protobuf::{Enum, Message as _},
rendezvous_proto::*,
socket_client,
sodiumoxide::crypto::{box_, sign},
timeout, tokio,
websocket::{check_ws, is_ws_endpoint},
ResultType, Stream,
};
use base::message_proto::*;
use scrap::camera;
#[cfg(not(any(target_os = "android", target_os = "ios")))]
use service::ServiceTmpl;
use service::{EmptyExtraFieldService, GenericService, Service, Subscriber};
use video_service::VideoSource;
use crate::ipc::Data;
pub mod audio_service;
#[cfg(target_os = "windows")]
pub mod terminal_helper;
#[cfg(not(any(target_os = "android", target_os = "ios")))]
pub mod terminal_service;
cfg_if::cfg_if! {
if #[cfg(not(target_os = "ios"))] {
mod clipboard_service;
#[cfg(target_os = "android")]
pub use clipboard_service::is_clipboard_service_ok;
#[cfg(target_os = "linux")]
pub(crate) mod wayland;
#[cfg(all(target_os = "linux", feature = "drm"))]
pub(crate) mod drm_capturer;
#[cfg(target_os = "linux")]
pub mod uinput;
#[cfg(target_os = "linux")]
pub mod rdp_input;
#[cfg(target_os = "linux")]
pub mod dbus;
#[cfg(not(target_os = "android"))]
pub mod input_service;
} else {
mod clipboard_service {
pub const NAME: &'static str = "";
}
}
}
#[cfg(any(target_os = "android", target_os = "ios"))]
pub mod input_service {
pub const NAME_CURSOR: &'static str = "";
pub const NAME_POS: &'static str = "";
pub const NAME_WINDOW_FOCUS: &'static str = "";
}
mod connection;
mod login_failure_check;
pub(crate) mod port_forward_mux;
pub mod display_service;
#[cfg(windows)]
pub mod portable_service;
mod service;
mod video_qos;
pub mod video_service;
#[cfg(all(target_os = "windows", feature = "flutter"))]
pub mod printer_service;
pub type Childs = Arc<Mutex<Vec<std::process::Child>>>;
type ConnMap = HashMap<i32, ConnInner>;
#[derive(Clone, Default)]
pub struct ConnectionMeta {
pub control_permissions: Option<ControlPermissions>,
pub controlled_context: Option<ControlledContext>,
}
#[cfg(any(target_os = "macos", target_os = "linux"))]
const CONFIG_SYNC_INTERVAL_SECS: f32 = 0.3;
#[cfg(any(target_os = "macos", target_os = "linux"))]
// 3s is enough for at least one initial sync attempt:
// 0.3s backoff + up to 1s connect timeout + up to 1s response timeout.
const CONFIG_SYNC_INITIAL_WAIT_SECS: u64 = 3;
lazy_static::lazy_static! {
pub static ref CHILD_PROCESS: Childs = Default::default();
// A client server used to provide local services(audio, video, clipboard, etc.)
// for all initiative connections.
//
// [Note]
// ugly
// Now we use this [`CLIENT_SERVER`] to do following operations:
// - record local audio, and send to remote
pub static ref CLIENT_SERVER: ServerPtr = new();
}
pub struct Server {
connections: ConnMap,
services: HashMap<String, Box<dyn Service>>,
id_count: i32,
}
pub type ServerPtr = Arc<RwLock<Server>>;
pub type ServerPtrWeak = Weak<RwLock<Server>>;
#[cfg(test)]
pub fn new_for_test() -> ServerPtr {
Arc::new(RwLock::new(Server {
connections: HashMap::new(),
services: HashMap::new(),
id_count: 1000,
}))
}
pub fn new() -> ServerPtr {
let mut server = Server {
connections: HashMap::new(),
services: HashMap::new(),
id_count: hbb_common::rand::random::<i32>() % 1000 + 1000, // ensure positive
};
server.add_service(Box::new(audio_service::new()));
#[cfg(not(target_os = "ios"))]
{
server.add_service(Box::new(display_service::new()));
server.add_service(Box::new(clipboard_service::new(
clipboard_service::NAME.to_owned(),
)));
#[cfg(feature = "unix-file-copy-paste")]
server.add_service(Box::new(clipboard_service::new(
clipboard_service::FILE_NAME.to_owned(),
)));
}
#[cfg(not(any(target_os = "android", target_os = "ios")))]
{
if !display_service::capture_cursor_embedded() {
server.add_service(Box::new(input_service::new_cursor()));
server.add_service(Box::new(input_service::new_pos()));
#[cfg(target_os = "linux")]
if scrap::is_x11() {
// wayland does not support multiple displays currently
server.add_service(Box::new(input_service::new_window_focus()));
}
#[cfg(not(target_os = "linux"))]
server.add_service(Box::new(input_service::new_window_focus()));
}
}
#[cfg(all(target_os = "windows", feature = "flutter"))]
{
match printer_service::init(&crate::get_app_name()) {
Ok(()) => {
log::info!("printer service initialized");
server.add_service(Box::new(printer_service::new(
printer_service::NAME.to_owned(),
)));
}
Err(e) => {
log::error!("printer service init failed: {}", e);
}
}
}
// Terminal service is created per connection, not globally
Arc::new(RwLock::new(server))
}
async fn accept_connection_(
server: ServerPtr,
socket: Stream,
secure: bool,
meta: ConnectionMeta,
slot: crate::rendezvous_mediator::PunchSlot,
) -> ResultType<()> {
let local_addr = socket.local_addr();
drop(socket);
// even we drop socket, below still may fail if not use reuse_addr,
// there is TIME_WAIT before socket really released, so sometimes we
// see "Only one usage of each socket address is normally permitted" on windows sometimes,
let listener = new_listener(local_addr, true).await?;
log::info!("Server listening on: {}", &listener.local_addr()?);
if let Ok((stream, addr)) = timeout(CONNECT_TIMEOUT, listener.accept()).await? {
// The peer is in: the place goes back before the session runs, as every punch's does.
drop(slot);
stream.set_nodelay(true).ok();
let stream_addr = stream.local_addr()?;
create_tcp_connection(
server,
Stream::from(stream, stream_addr),
addr,
secure,
meta,
)
.await?;
}
Ok(())
}
pub async fn create_tcp_connection(
server: ServerPtr,
stream: Stream,
addr: SocketAddr,
secure: bool,
meta: ConnectionMeta,
) -> ResultType<()> {
let mut stream = stream;
// The address the connection layer keys on, whitelist and admission alike.
let addr = hbb_common::try_into_v4(addr);
let id = server.write().unwrap().get_new_id();
// Admitted before the identity handshake, so a peer that stalls in it, or after it without
// logging in, holds its place the whole time; an address over its share is turned away.
let Some(unauthorized) = admit_unauthorized(id, addr.ip()) else {
bail!("too many unauthenticated connections from {}", addr.ip());
};
// Before the handshake, so its read is bounded too; lifted again at authorization.
stream.set_max_packet_length(MAX_UNAUTHORIZED_MESSAGE);
tokio::select! {
handshake = identity_handshake(&mut stream, secure) => handshake?,
_ = unauthorized.evicted() => {
bail!("evicted to make room for a newer unauthenticated connection");
}
}
#[cfg(target_os = "macos")]
{
use std::process::Command;
if let Ok(task) = Command::new("/usr/bin/caffeinate")
.arg("-u")
.arg("-t 5")
.spawn()
{
super::CHILD_PROCESS.lock().unwrap().push(task);
}
log::info!("wake up macos");
}
Connection::start(
addr,
stream,
id,
Arc::downgrade(&server),
meta,
unauthorized,
)
.await;
Ok(())
}
/// Our signed identity goes out and, when `secure`, the controller's reply keys `stream`.
/// Separate so it can be raced against the connection's eviction.
async fn identity_handshake(stream: &mut Stream, secure: bool) -> ResultType<()> {
let (sk, pk) = Config::get_key_pair();
if secure && pk.len() == sign::PUBLICKEYBYTES && sk.len() == sign::SECRETKEYBYTES {
let mut sk_ = [0u8; sign::SECRETKEYBYTES];
sk_[..].copy_from_slice(&sk);
let sk = sign::SecretKey(sk_);
let mut msg_out = Message::new();
let (our_pk_b, our_sk_b) = box_::gen_keypair();
// On a WebRTC transport, bind our DTLS certificate fingerprint to our signed identity so
// the controller can verify the DTLS channel it negotiated actually terminates at us
// (not a rendezvous/relay that swapped the SDP fingerprint). Empty on other transports.
// Fail immediately on WebRTC if the local fingerprint is unavailable: signing "" would
// only make the client fail-closed after a wasted round-trip.
let dtls_fingerprint = stream.dtls_fingerprint(true).await.unwrap_or_default();
if stream.is_webrtc() && dtls_fingerprint.is_empty() {
bail!("WebRTC local DTLS fingerprint unavailable");
}
// A WebRTC stream is encrypted by DTLS and takes no stream key of its own, split or
// not, so what it advertises is what it runs: 0. Saying 1 there would have a future
// controller that splits keys on WebRTC agree on a version this side never applies.
let advertised = if stream.is_webrtc() {
0
} else {
tcp::KX_VERSION_LATEST
};
msg_out.set_signed_id(SignedId {
id: sign::sign(
&IdPk {
id: Config::get_id(),
pk: Bytes::from(our_pk_b.0.to_vec()),
dtls_fingerprint,
kx_version: advertised,
..Default::default()
}
.write_to_bytes()
.unwrap_or_default(),
&sk,
)
.into(),
..Default::default()
});
timeout(CONNECT_TIMEOUT, stream.send(&msg_out)).await??;
match timeout(CONNECT_TIMEOUT, stream.next()).await? {
Some(res) => {
let bytes = res?;
if let Ok(msg_in) = Message::parse_from_bytes(&bytes) {
if let Some(message::Union::PublicKey(pk)) = msg_in.union {
if pk.asymmetric_value.len() == box_::PUBLICKEYBYTES {
let key = tcp::Encrypt::decode(
&pk.symmetric_value,
&pk.asymmetric_value,
&our_sk_b,
)?;
if pk.kx_version > advertised {
bail!(
"Handshake failed: key exchange version {} not offered, {} was",
pk.kx_version,
advertised
);
}
stream.set_negotiated_key(
key,
false,
&tcp::KxTranscript {
initiator_pk: &pk.asymmetric_value,
responder_pk: &our_pk_b.0,
advertised,
picked: pk.kx_version,
},
)?;
} else if pk.asymmetric_value.is_empty() {
Config::set_key_confirmed(false);
log::info!("Force to update pk");
} else {
bail!("Handshake failed: invalid public sign key length from peer");
}
} else {
log::error!("Handshake failed: invalid message type");
}
} else {
bail!("Handshake failed: invalid message format");
}
}
None => {
bail!("Failed to receive public key");
}
}
}
Ok(())
}
pub(crate) async fn accept_connection(
server: ServerPtr,
socket: Stream,
peer_addr: SocketAddr,
secure: bool,
meta: ConnectionMeta,
slot: crate::rendezvous_mediator::PunchSlot,
) {
if let Err(err) = accept_connection_(server, socket, secure, meta, slot).await {
log::warn!("Failed to accept connection from {}: {}", peer_addr, err);
}
}
pub async fn create_relay_connection(
server: ServerPtr,
relay_server: String,
uuid: String,
peer_addr: SocketAddr,
secure: bool,
ipv4: bool,
meta: ConnectionMeta,
) {
if let Err(err) = create_relay_connection_(
server,
relay_server,
uuid.clone(),
peer_addr,
secure,
ipv4,
meta,
)
.await
{
log::error!(
"Failed to create relay connection for {} with uuid {}: {}",
peer_addr,
uuid,
err
);
}
}
async fn create_relay_connection_(
server: ServerPtr,
relay_server: String,
uuid: String,
peer_addr: SocketAddr,
secure: bool,
ipv4: bool,
meta: ConnectionMeta,
) -> ResultType<()> {
// The proxy's address family does not constrain the relay target.
let relay_server = crate::check_port(relay_server, RELAY_PORT);
let use_ws_proxy = Config::is_proxy() && is_ws_endpoint(&check_ws(&relay_server));
let mut stream = socket_client::connect_tcp(
socket_client::ipv4_to_ipv6(relay_server, ipv4 || use_ws_proxy),
CONNECT_TIMEOUT,
)
.await?;
let mut msg_out = RendezvousMessage::new();
let licence_key = crate::get_key(true).await;
msg_out.set_request_relay(RequestRelay {
licence_key,
uuid,
..Default::default()
});
stream.send(&msg_out).await?;
create_tcp_connection(server, stream, peer_addr, secure, meta).await?;
Ok(())
}
impl Server {
fn is_video_service_name(name: &str) -> bool {
name.starts_with(VideoSource::Monitor.service_name_prefix())
|| name.starts_with(VideoSource::Camera.service_name_prefix())
}
pub fn try_add_primary_camera_service(&mut self) {
if !camera::primary_camera_exists() {
return;
}
let primary_camera_name =
video_service::get_service_name(VideoSource::Camera, camera::PRIMARY_CAMERA_IDX);
if !self.contains(&primary_camera_name) {
self.add_service(Box::new(video_service::new(
VideoSource::Camera,
camera::PRIMARY_CAMERA_IDX,
)));
}
}
pub fn try_add_monitor_service(&mut self, display_idx: usize) {
let monitor_service_name =
video_service::get_service_name(VideoSource::Monitor, display_idx);
if !self.contains(&monitor_service_name) {
self.add_service(Box::new(video_service::new(
VideoSource::Monitor,
display_idx,
)));
}
}
pub fn add_camera_connection(&mut self, conn: ConnInner) {
if camera::primary_camera_exists() {
let primary_camera_name =
video_service::get_service_name(VideoSource::Camera, camera::PRIMARY_CAMERA_IDX);
if let Some(s) = self.services.get(&primary_camera_name) {
s.on_subscribe(conn.clone());
}
}
self.connections.insert(conn.id(), conn);
}
pub fn add_monitor_connection(
&mut self,
conn: ConnInner,
noperms: &Vec<&'static str>,
display_idx: usize,
) {
let monitor_service_name =
video_service::get_service_name(VideoSource::Monitor, display_idx);
for s in self.services.values() {
let name = s.name();
if Self::is_video_service_name(&name) && name != monitor_service_name {
continue;
}
if !noperms.contains(&(&name as _)) {
s.on_subscribe(conn.clone());
}
}
#[cfg(target_os = "macos")]
self.update_enable_retina();
self.connections.insert(conn.id(), conn);
}
pub fn remove_connection(&mut self, conn: &ConnInner) {
for s in self.services.values() {
s.on_unsubscribe(conn.id());
}
self.connections.remove(&conn.id());
#[cfg(target_os = "macos")]
self.update_enable_retina();
}
pub fn close_connections(&mut self) {
let conn_inners: Vec<_> = self.connections.values_mut().collect();
for c in conn_inners {
let mut misc = Misc::new();
misc.set_stop_service(true);
let mut msg = Message::new();
msg.set_misc(misc);
c.send(Arc::new(msg));
}
}
fn add_service(&mut self, service: Box<dyn Service>) {
let name = service.name();
self.services.insert(name, service);
}
pub fn contains(&self, name: &str) -> bool {
self.services.contains_key(name)
}
pub fn subscribe(&mut self, name: &str, conn: ConnInner, sub: bool) {
if let Some(s) = self.services.get(name) {
if s.is_subed(conn.id()) == sub {
return;
}
if sub {
s.on_subscribe(conn.clone());
} else {
s.on_unsubscribe(conn.id());
}
#[cfg(target_os = "macos")]
self.update_enable_retina();
}
}
// get a new unique id
pub fn get_new_id(&mut self) -> i32 {
self.id_count += 1;
self.id_count
}
pub fn set_video_service_opt(
&self,
display: Option<(VideoSource, usize)>,
opt: &str,
value: &str,
) {
for (k, v) in self.services.iter() {
if let Some((source, display)) = display {
if k != &video_service::get_service_name(source, display) {
continue;
}
}
if Self::is_video_service_name(k) {
v.set_option(opt, value);
}
}
}
fn get_subbed_displays_count(&self, conn_id: i32) -> usize {
self.services
.keys()
.filter(|k| {
Self::is_video_service_name(k)
&& self
.services
.get(*k)
.map(|s| s.is_subed(conn_id))
.unwrap_or(false)
})
.count()
}
fn capture_displays(
&mut self,
conn: ConnInner,
source: VideoSource,
displays: &[usize],
include: bool,
exclude: bool,
) {
let displays = displays
.iter()
.map(|d| video_service::get_service_name(source, *d))
.collect::<Vec<_>>();
let keys = self.services.keys().cloned().collect::<Vec<_>>();
for name in keys.iter() {
if Self::is_video_service_name(&name) {
if displays.contains(&name) {
if include {
self.subscribe(&name, conn.clone(), true);
}
} else {
if exclude {
self.subscribe(&name, conn.clone(), false);
}
}
}
}
}
#[cfg(target_os = "macos")]
fn update_enable_retina(&self) {
let mut video_service_count = 0;
for (name, service) in self.services.iter() {
if Self::is_video_service_name(&name) && service.ok() {
video_service_count += 1;
}
}
*scrap::quartz::ENABLE_RETINA.lock().unwrap() = video_service_count < 2;
}
}
impl Drop for Server {
fn drop(&mut self) {
for s in self.services.values() {
s.join();
}
#[cfg(target_os = "linux")]
wayland::clear();
}
}
pub fn check_zombie() {
std::thread::spawn(|| loop {
let mut lock = CHILD_PROCESS.lock().unwrap();
let mut i = 0;
while i != lock.len() {
let c = &mut (*lock)[i];
if let Ok(Some(_)) = c.try_wait() {
lock.remove(i);
} else {
i += 1;
}
}
drop(lock);
std::thread::sleep(Duration::from_millis(100));
});
}
/// Start the host server that allows the remote peer to control the current machine.
///
/// # Arguments
///
/// * `is_server` - Whether the current client is definitely the server.
/// If true, the server will be started.
/// Otherwise, client will check if there's already a server and start one if not.
#[cfg(any(target_os = "android", target_os = "ios"))]
#[tokio::main]
pub async fn start_server(_is_server: bool) {
allow_err!(crate::trusted_devices::initialize());
crate::RendezvousMediator::start_all().await;
}
/// Start the host server that allows the remote peer to control the current machine.
///
/// # Arguments
///
/// * `is_server` - Whether the current client is definitely the server.
/// If true, the server will be started.
/// Otherwise, client will check if there's already a server and start one if not.
/// * `no_server` - If `is_server` is false, whether to start a server if not found.
#[cfg(not(any(target_os = "android", target_os = "ios")))]
#[tokio::main]
pub async fn start_server(is_server: bool, no_server: bool) {
use std::sync::Once;
static ONCE: Once = Once::new();
ONCE.call_once(|| {
#[cfg(target_os = "linux")]
{
log::info!("DISPLAY={:?}", std::env::var("DISPLAY"));
log::info!("XAUTHORITY={:?}", std::env::var("XAUTHORITY"));
}
#[cfg(windows)]
base::platform::windows::start_cpu_performance_monitor();
});
if is_server {
#[cfg(windows)]
if !crate::is_server() {
if let Err(err) = start_gui_ipc().await {
log::error!("Local hosting unavailable: failed to start GUI IPC listener: {err:#}");
return;
}
}
crate::common::set_server_running(true);
if !cfg!(windows) || crate::is_server() {
std::thread::spawn(move || {
if let Err(err) = crate::ipc::start("") {
log::error!("Failed to start ipc: {}", err);
if crate::is_server() {
#[cfg(windows)]
{
const MAIN_IPC_STARTUP_GRACE: Duration = Duration::from_secs(1);
// Give the GUI time to arrange its replacement before forced takeover.
std::thread::sleep(MAIN_IPC_STARTUP_GRACE);
match crate::ipc::start("") {
Ok(()) => return,
Err(err) => log::error!("Failed to retry ipc startup: {err}"),
}
}
log::error!("ipc is occupied by another process, try kill it");
std::thread::spawn(stop_main_window_process).join().ok();
}
std::process::exit(-1);
}
});
}
// Warm the DRM availability cache before any client connects, so the first connection does
// not race a cold `_drm` probe and ship an empty display list ("No displays" + retry).
// X11 is skipped -- probing there makes the root service open DRM readers for a path this
// session can never take -- but that decision belongs to `warm_availability`, which already
// makes it, and NOT to this call site. Deciding it here is the same one-shot-at-startup
// mistake the pre-warm had: `is_x11()` answers "x11" whenever loginctl cannot yet name the
// seat0 session, which during a boot is exactly when this runs, and nothing revisits it --
// so a Wayland host that came up slowly skipped the warm for the life of the process and
// got back the cold-probe "No displays" symptom the warm exists to remove.
#[cfg(all(target_os = "linux", feature = "drm"))]
if let Err(err) = std::thread::Builder::new()
.name("drm-warm".into())
.spawn(drm_capturer::warm_availability)
{
// Same reason as the root service's startup threads: `thread::spawn` panics on EAGAIN
// and that would abort `start_server`. Skipping the warm costs the first session the
// cold probe, which is what happened before the warm existed.
log::warn!("drm: could not spawn the availability warm ({err}); skipping it");
}
input_service::fix_key_down_timeout_loop();
#[cfg(target_os = "linux")]
if input_service::wayland_use_uinput() {
allow_err!(input_service::setup_uinput(0, 1920, 0, 1080).await);
}
#[cfg(any(target_os = "macos", target_os = "linux"))]
wait_initial_config_sync().await;
allow_err!(crate::trusted_devices::initialize());
#[cfg(target_os = "windows")]
crate::platform::try_kill_broker();
#[cfg(feature = "hwcodec")]
scrap::hwcodec::start_check_process();
crate::RendezvousMediator::start_all().await;
} else {
match crate::ipc::connect(1000, "").await {
Ok(mut conn) => {
if conn.send(&Data::SyncConfig(None)).await.is_ok() {
if let Ok(Some(data)) = conn.next_timeout(1000).await {
match data {
Data::SyncConfig(Some(configs)) => {
let (config, config2) = *configs;
#[cfg(target_os = "windows")]
allow_err!(
crate::trusted_devices::config::update(move || {
crate::trusted_devices::config::sync_config(config, config2)
})
.await
);
#[cfg(not(target_os = "windows"))]
{
if Config::set(config) {
log::info!("config synced");
}
if Config2::set(config2) {
log::info!("config2 synced");
}
}
}
_ => {}
}
}
}
#[cfg(feature = "hwcodec")]
#[cfg(any(target_os = "windows", target_os = "linux"))]
crate::ipc::client_get_hwcodec_config_thread(0);
}
Err(err) => {
log::info!("server not started: {err:?}, no_server: {no_server}");
if no_server {
hbb_common::sleep(1.0).await;
std::thread::spawn(|| start_server(false, true));
} else {
log::info!("try start server");
std::thread::spawn(|| start_server(true, false));
}
}
}
}
}
#[cfg(windows)]
async fn start_gui_ipc() -> ResultType<()> {
let incoming = crate::ipc::new_listener("").await?;
tokio::spawn(crate::ipc::start_with_listener("", incoming));
Ok(())
}
#[cfg(target_os = "macos")]
#[tokio::main(flavor = "current_thread")]
pub async fn start_ipc_url_server() {
log::debug!("Start an ipc server for listening to url schemes");
match crate::ipc::new_listener("_url").await {
Ok(mut incoming) => {
while let Some(Ok(conn)) = incoming.next().await {
let mut conn = crate::ipc::Connection::new(conn);
match conn.next_timeout(1000).await {
Ok(Some(data)) => match data {
#[cfg(feature = "flutter")]
Data::UrlLink(url) => {
let mut m = HashMap::new();
m.insert("name", "on_url_scheme_received");
m.insert("url", url.as_str());
let event = serde_json::to_string(&m).unwrap_or("".to_owned());
match crate::flutter::push_global_event(
crate::flutter::APP_TYPE_MAIN,
event,
) {
None => log::warn!("No main window app found!"),
Some(..) => {}
}
}
_ => {
log::warn!("An unexpected data was sent to the ipc url server.")
}
},
Err(err) => {
log::error!("{}", err);
}
_ => {}
}
}
}
Err(err) => {
log::error!("{}", err);
}
}
}
#[cfg(any(target_os = "macos", target_os = "linux"))]
async fn wait_initial_config_sync() {
if crate::platform::is_root() {
return;
}
// Non-server process should not block startup, but still keeps background sync/watch alive.
if !crate::is_server() {
tokio::spawn(async move {
sync_and_watch_config_dir(None).await;
});
return;
}
let (sync_done_tx, mut sync_done_rx) = tokio::sync::oneshot::channel::<()>();
tokio::spawn(async move {
sync_and_watch_config_dir(Some(sync_done_tx)).await;
});
// Server process waits up to N seconds for initial root->local sync to reduce stale-start window.
tokio::select! {
_ = &mut sync_done_rx => {
}
_ = tokio::time::sleep(Duration::from_secs(CONFIG_SYNC_INITIAL_WAIT_SECS)) => {
log::warn!(
"timed out waiting {}s for initial config sync, continue startup and keep syncing in background",
CONFIG_SYNC_INITIAL_WAIT_SECS
);
}
}
}
#[cfg(any(target_os = "macos", target_os = "linux"))]
async fn sync_and_watch_config_dir(sync_done_tx: Option<tokio::sync::oneshot::Sender<()>>) {
let mut cfg0 = (Config::get(), Config2::get());
let mut synced = false;
let mut is_root_config_empty = false;
let mut sync_done_tx = sync_done_tx;
let tries = if crate::is_server() { 30 } else { 3 };
log::debug!("#tries of ipc service connection: {}", tries);
use hbb_common::sleep;
for i in 1..=tries {
sleep(i as f32 * CONFIG_SYNC_INTERVAL_SECS).await;
match crate::ipc::connect_trusted_device_service(1000).await {
Ok(mut conn) => {
if !synced {
if conn.send(&Data::SyncConfig(None)).await.is_ok() {
if let Ok(Some(data)) = conn.next_timeout(1000).await {
match data {
Data::SyncConfig(Some(configs)) => {
let (config, config2) = *configs;
let _chk = crate::ipc::CheckIfRestart::new();
#[cfg(target_os = "macos")]
let _chk_pk = crate::CheckIfResendPk::new();
if !config.is_empty() {
let result =
crate::trusted_devices::config::update(move || {
crate::trusted_devices::config::sync_config(
config.clone(),
config2.clone(),
)?;
Ok((config, config2))
})
.await;
match result {
Ok(configs) => cfg0 = configs,
Err(error) => {
crate::trusted_devices::config::log_sync_error(
error,
);
continue;
}
}
log::info!("sync config from root");
} else {
// only on macos, because this issue was only reproduced on macos
#[cfg(target_os = "macos")]
{
// root config is empty, mark for sync in watch loop
// to prevent root from generating a new config on login screen
is_root_config_empty = true;
}
}
synced = true;
// Notify startup waiter once initial sync phase finishes successfully.
if let Some(tx) = sync_done_tx.take() {
let _ = tx.send(());
}
}
_ => {}
};
};
}
if !synced {
log::warn!(
"initial config sync from root failed, reconnecting to ipc_service"
);
continue;
}
}
loop {
sleep(CONFIG_SYNC_INTERVAL_SECS).await;
let cfg = (Config::get(), Config2::get());
let should_sync = cfg != cfg0 || (is_root_config_empty && !cfg.0.is_empty());
if should_sync {
if is_root_config_empty {
log::info!("root config is empty, sync our config to root");
} else {
log::info!("config updated, sync to root");
}
match conn.send(&Data::SyncConfig(Some(cfg.clone().into()))).await {
Err(e) => {
log::error!("sync config to root failed: {}", e);
match crate::ipc::connect_trusted_device_service(1000).await {
Ok(mut _conn) => {
conn = _conn;
log::info!("reconnected to ipc_service");
}
_ => {}
}
}
_ => {
match conn.next_timeout(1000).await {
Ok(Some(Data::SyncConfig(None))) => {
cfg0 = cfg;
is_root_config_empty = false;
}
Ok(Some(Data::SecurityConfigError(error))) => {
crate::trusted_devices::config::log_sync_error(error)
}
_ => crate::trusted_devices::config::log_sync_error(
"Host did not confirm configuration sync",
),
}
}
}
}
}
}
Err(_) => {
log::info!("#{} try: failed to connect to ipc_service", i);
}
}
}
// Notify startup waiter even when initial sync is skipped/failed, to avoid unnecessary waiting.
if let Some(tx) = sync_done_tx.take() {
let _ = tx.send(());
}
log::warn!("skipped config sync");
}
#[tokio::main(flavor = "current_thread")]
pub async fn stop_main_window_process() {
// this may also kill another --server process,
// but --server usually can be auto restarted by --service, so it is ok
if let Ok(mut conn) = crate::ipc::connect(1000, "").await {
conn.send(&crate::ipc::Data::Close).await.ok();
}
#[cfg(windows)]
{
// in case above failure, e.g. zombie process
if let Err(e) = crate::platform::try_kill_rustdesk_main_window_process() {
log::error!("kill failed: {}", e);
}
}
}
#[cfg(all(test, windows))]
mod windows_gui_startup_tests {
use super::*;
use parity_tokio_ipc::{Endpoint, SecurityAttributes};
use std::{os::windows::io::AsRawHandle, process::Command};
use windows::Win32::{
Foundation::{HANDLE, WAIT_OBJECT_0},
System::Threading::WaitForSingleObject,
};
const CHILD_TEST: &str = "RUSTDESK_TEST_GUI_IPC_CHILD";
const CHILD_TIMEOUT_MS: u32 = 10_000;
// An unmatched test filter exits with zero without running the child assertions.
const CHILD_SUCCESS_EXIT_CODE: i32 = 42;
#[test]
fn windows_gui_survives_ipc_bind_failure() {
if std::env::var_os(CHILD_TEST).is_some() {
*hbb_common::config::APP_NAME.write().unwrap() =
format!("RustDesk-ipc-test-{}", std::process::id());
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let _incoming = runtime.block_on(async {
let mut endpoint = Endpoint::new(Config::ipc_path(""));
endpoint
.set_security_attributes(SecurityAttributes::allow_everyone_create().unwrap());
endpoint.incoming().unwrap()
});
start_server(true, false);
assert!(!crate::common::is_server_running());
std::process::exit(CHILD_SUCCESS_EXIT_CODE);
}
let mut child = Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
"server::windows_gui_startup_tests::windows_gui_survives_ipc_bind_failure",
"--nocapture",
])
.env(CHILD_TEST, "1")
.spawn()
.unwrap();
let result =
unsafe { WaitForSingleObject(HANDLE(child.as_raw_handle()), CHILD_TIMEOUT_MS) };
if result != WAIT_OBJECT_0 {
child.kill().unwrap();
child.wait().unwrap();
panic!("GUI IPC startup test timed out or failed to wait: {result:?}");
}
assert_eq!(
child.wait().unwrap().code(),
Some(CHILD_SUCCESS_EXIT_CODE),
"GUI subprocess did not complete the startup assertions"
);
}
}