rustdesk/rustdesk-server
1
0 Fork 0
Code Releases 1 Activity

Merge remote-tracking branch 'upstream/master' into forapi

Browse Source 
# Conflicts:
#	.github/workflows/build.yaml
#	.github/workflows/ghcr.yml
#	README-DE.md
#	README-NL.md
#	README-TW.md
#	README.md
#	libs/hbb_common
#	src/main.rs
#	src/rendezvous_server.rs
This commit is contained in:
root committed 2026-09-05 23:50:23 +08:00
2 parents fb8b5b9cf4 a7736be5e4
commit 916999abd3
24 files changed
+3331 -455
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/rendezvous_server.rs
+133 -38
View File
@@ -23,7 +23,7 @@ use hbb_common::{
sodiumoxide::hex,
tcp,
tcp::Encrypt,
tcp::{listen_any, FramedStream},
tcp::FramedStream,
timeout,
tokio::{
self,
@@ -56,7 +56,7 @@ enum Data {
RelayServers(RelayServers),
}
const REG_TIMEOUT: i32 = 30_000;
const REG_TIMEOUT: i64 = 30_000;
type TcpStreamSink = SplitSink<Framed<TcpStream, BytesCodec>, Bytes>;
type WsSink = SplitSink<tokio_tungstenite::WebSocketStream<TcpStream>, tungstenite::Message>;
struct SafeWsSink {
@@ -105,6 +105,14 @@ const CHECK_RELAY_TIMEOUT: u64 = 3_000;
static ALWAYS_USE_RELAY: AtomicBool = AtomicBool::new(false);
static MUST_LOGIN: AtomicBool = AtomicBool::new(false);
// Store punch hole requests
use once_cell::sync::Lazy;
use tokio::sync::Mutex as TokioMutex; // differentiate if needed
#[derive(Clone)]
struct PunchReqEntry { tm: Instant, from_ip: String, to_ip: String, to_id: String }
static PUNCH_REQS: Lazy<TokioMutex<Vec<PunchReqEntry>>> = Lazy::new(|| TokioMutex::new(Vec::new()));
const PUNCH_REQ_DEDUPE_SEC: u64 = 60;
#[derive(Clone)]
struct Inner {
serial: i32,
@@ -134,21 +142,29 @@ enum LoopFailure {
Listener3,
Listener2,
Listener,
ConsoleListener,
}
impl RendezvousServer {
pub fn start(port: i32, serial: i32, key: &str, rmem: usize) -> ResultType<()> {
Self::start_with_bind(None, port, serial, key, rmem)
}
#[tokio::main(flavor = "multi_thread")]
pub async fn start(port: i32, serial: i32, key: &str, rmem: usize) -> ResultType<()> {
pub async fn start_with_bind(
bind_addr: Option<IpAddr>,
port: i32,
serial: i32,
key: &str,
rmem: usize,
) -> ResultType<()> {
let (key, sk) = Self::get_server_sk(key);
let nat_port = port - 1;
let ws_port = port + 2;
let pm = PeerMap::new().await?;
log::info!("serial={}", serial);
let rendezvous_servers = get_servers(&get_arg("rendezvous-servers"), "rendezvous-servers");
log::info!("Listening on tcp/udp :{}", port);
log::info!("Listening on tcp :{}, extra port for NAT test", nat_port);
log::info!("Listening on websocket :{}", ws_port);
let mut socket = create_udp_listener(port, rmem).await?;
let mut socket = create_udp_listener(bind_addr, port, rmem).await?;
let (tx, mut rx) = mpsc::unbounded_channel::<Data>();
let software_url = get_arg("software-url");
let version = hbb_common::get_version_from_url(&software_url);
@@ -191,15 +207,18 @@ impl RendezvousServer {
log::info!("local-ip: {:?}", rs.inner.local_ip);
std::env::set_var("PORT_FOR_API", port.to_string());
rs.parse_relay_servers(&get_arg("relay-servers"));
let mut listener = create_tcp_listener(port).await?;
let mut listener2 = create_tcp_listener(nat_port).await?;
let mut listener3 = create_tcp_listener(ws_port).await?;
let test_addr = std::env::var("TEST_HBBS").unwrap_or_default();
if std::env::var("ALWAYS_USE_RELAY")
.unwrap_or_default()
.to_uppercase()
== "Y"
{
let mut listener = create_tcp_listener(bind_addr, port).await?;
let mut listener2 = create_tcp_listener(bind_addr, nat_port).await?;
let mut listener3 = create_tcp_listener(bind_addr, ws_port).await?;
let mut listener_console = listen_console(bind_addr, nat_port as _).await?;
log::info!("Listening on tcp/udp {}", listener.local_addr()?);
log::info!(
"Listening on tcp {}, extra port for NAT test",
listener2.local_addr()?
);
log::info!("Listening on websocket {}", listener3.local_addr()?);
let test_addr = get_arg("TEST_HBBS");
if get_arg("ALWAYS_USE_RELAY").to_uppercase() == "Y" {
ALWAYS_USE_RELAY.store(true, Ordering::SeqCst);
}
log::info!(
@@ -262,6 +281,7 @@ impl RendezvousServer {
&mut listener,
&mut listener2,
&mut listener3,
&mut listener_console,
&mut socket,
&key,
)
@@ -269,19 +289,23 @@ impl RendezvousServer {
{
LoopFailure::UdpSocket => {
drop(socket);
socket = create_udp_listener(port, rmem).await?;
socket = create_udp_listener(bind_addr, port, rmem).await?;
}
LoopFailure::Listener => {
drop(listener);
listener = create_tcp_listener(port).await?;
listener = create_tcp_listener(bind_addr, port).await?;
}
LoopFailure::Listener2 => {
drop(listener2);
listener2 = create_tcp_listener(nat_port).await?;
listener2 = create_tcp_listener(bind_addr, nat_port).await?;
}
LoopFailure::ConsoleListener => {
drop(listener_console.take());
listener_console = listen_console(bind_addr, nat_port as _).await?;
}
LoopFailure::Listener3 => {
drop(listener3);
listener3 = create_tcp_listener(ws_port).await?;
listener3 = create_tcp_listener(bind_addr, ws_port).await?;
}
}
}
@@ -299,6 +323,7 @@ impl RendezvousServer {
listener: &mut TcpListener,
listener2: &mut TcpListener,
listener3: &mut TcpListener,
listener_console: &mut Option<TcpListener>,
socket: &mut FramedSocket,
key: &str,
) -> LoopFailure {
@@ -350,6 +375,18 @@ impl RendezvousServer {
}
}
}
res = accept_or_pending(listener_console.as_ref()) => {
match res {
Ok((stream, addr)) => {
stream.set_nodelay(true).ok();
self.handle_listener2(stream, addr).await;
}
Err(err) => {
log::error!("console listener.accept failed: {}", err);
return LoopFailure::ConsoleListener;
}
}
}
res = listener3.accept() => {
match res {
Ok((stream, addr)) => {
@@ -432,22 +469,14 @@ impl RendezvousServer {
}
}
Some(rendezvous_message::Union::PunchHoleRequest(ph)) => {
if self.pm.is_in_memory(&ph.id).await {
self.handle_udp_punch_hole_request(addr, ph, key).await?;
} else {
// not in memory, fetch from db with spawn in case blocking me
let mut me = self.clone();
let key = key.to_owned();
tokio::spawn(async move {
allow_err!(me.handle_udp_punch_hole_request(addr, ph, &key).await);
});
}
// UDP PunchHoleRequest is intentionally unsupported.
// The supported client path sends PunchHoleRequest over TCP/WS.
}
Some(rendezvous_message::Union::PunchHoleSent(phs)) => {
self.handle_hole_sent(phs, addr, Some(socket)).await?;
// UDP PunchHoleSent is intentionally unsupported to avoid UDP reflection/amplification
}
Some(rendezvous_message::Union::LocalAddr(la)) => {
self.handle_local_addr(la, addr, Some(socket)).await?;
// UDP LocalAddr is intentionally unsupported to avoid UDP reflection/amplification
}
Some(rendezvous_message::Union::ConfigureUpdate(mut cu)) => {
if try_into_v4(addr).ip().is_loopback() && cu.serial > self.inner.serial {
@@ -665,7 +694,7 @@ impl RendezvousServer {
let mut states = BytesMut::zeroed((peers.len() + 7) / 8);
for (i, peer_id) in peers.iter().enumerate() {
if let Some(peer) = self.pm.get_in_memory(peer_id).await {
let elapsed = peer.read().await.last_reg_time.elapsed().as_millis() as i32;
let elapsed = peer.read().await.last_reg_time.elapsed().as_millis() as i64;
// bytes index from left to right
let states_idx = i / 8;
let bit_idx = 7 - i % 8;
@@ -891,6 +920,7 @@ impl RendezvousServer {
) -> ResultType<(RendezvousMessage, Option<SocketAddr>)> {
let mut ph = ph;
if !key.is_empty() && ph.licence_key != key {
log::warn!("Authentication failed from {} for peer {} - invalid key", addr, ph.id);
let mut msg_out = RendezvousMessage::new();
msg_out.set_punch_hole_response(PunchHoleResponse {
failure: punch_hole_response::Failure::LICENSE_MISMATCH.into(),
@@ -930,7 +960,7 @@ impl RendezvousServer {
if let Some(peer) = self.pm.get(&id).await {
let (elapsed, peer_addr) = {
let r = peer.read().await;
(r.last_reg_time.elapsed().as_millis() as i32, r.socket_addr)
(r.last_reg_time.elapsed().as_millis() as i64, r.socket_addr)
};
if elapsed >= REG_TIMEOUT {
let mut msg_out = RendezvousMessage::new();
@@ -940,6 +970,23 @@ impl RendezvousServer {
});
return Ok((msg_out, None));
}
// record punch hole request (from addr -> peer id/peer_addr)
{
let from_ip = try_into_v4(addr).ip().to_string();
let to_ip = try_into_v4(peer_addr).ip().to_string();
let to_id_clone = id.clone();
let mut lock = PUNCH_REQS.lock().await;
let mut dup = false;
for e in lock.iter().rev().take(30) { // only check recent tail subset for speed
if e.from_ip == from_ip && e.to_id == to_id_clone {
if e.tm.elapsed().as_secs() < PUNCH_REQ_DEDUPE_SEC { dup = true; }
break;
}
}
if !dup { lock.push(PunchReqEntry { tm: Instant::now(), from_ip, to_ip, to_id: to_id_clone }); }
}
let mut msg_out = RendezvousMessage::new();
let peer_is_lan = self.is_lan(peer_addr);
let is_lan = self.is_lan(addr);
@@ -1136,11 +1183,12 @@ impl RendezvousServer {
match fds.next() {
Some("h") => {
res = format!(
"{}\n{}\n{}\n{}\n{}\n{}\n{}\n",
"{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n",
"relay-servers(rs) <separated by ,>",
"reload-geo(rg)",
"ip-blocker(ib) [<ip>|<number>] [-]",
"ip-changes(ic) [<id>|<number>] [-]",
"punch-requests(pr) [<number>] [-]",
"always-use-relay(aur) [Y|N]",
"test-geo(tg) <ip1> <ip2>",
"must-login(ml) [Y|N]",
@@ -1241,6 +1289,24 @@ impl RendezvousServer {
}
}
}
Some("punch-requests" | "pr") => {
use std::fmt::Write as _;
let mut lock = PUNCH_REQS.lock().await;
let arg = fds.next();
if let Some("-") = arg { lock.clear(); }
else {
let mut start = arg.and_then(|x| x.parse::<usize>().ok()).unwrap_or(0);
let mut page_size = fds.next().and_then(|x| x.parse::<usize>().ok()).unwrap_or(10);
if page_size == 0 { page_size = 10; }
for (_, e) in lock.iter().enumerate().skip(start).take(page_size) {
let age = e.tm.elapsed();
let event_system = std::time::SystemTime::now() - age;
let event_iso = chrono::DateTime::<chrono::Utc>::from(event_system)
.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
let _ = writeln!(res, "{} {} -> {}@{}", event_iso, e.from_ip, e.to_id, e.to_ip);
}
}
}
Some("always-use-relay" | "aur") => {
if let Some(rs) = fds.next() {
if rs.to_uppercase() == "Y" {
@@ -1348,6 +1414,15 @@ impl RendezvousServer {
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
let callback = |req: &Request, response: Response| {
let headers = req.headers();
// X-Real-IP / X-Forwarded-For are trusted as-is so that the real
// client IP is preserved when the WebSocket port runs behind a
// reverse proxy (WSS). They are NOT validated: anyone who can reach
// this port directly can spoof an arbitrary IP, bypassing IP-based
// rate limiting / blocking and corrupting logged IPs. Do not expose
// the WebSocket port directly to untrusted networks; only the
// reverse proxy, which overwrites these headers, should be able to
// connect to it.
// https://github.com/rustdesk/rustdesk-server/issues/634
let real_ip = headers
.get("X-Real-IP")
.or_else(|| headers.get("X-Forwarded-For"))
@@ -1582,7 +1657,15 @@ async fn send_rk_res(
socket.send(&msg_out, addr).await
}
async fn create_udp_listener(port: i32, rmem: usize) -> ResultType<FramedSocket> {
async fn create_udp_listener(
bind_addr: Option<IpAddr>,
port: i32,
rmem: usize,
) -> ResultType<FramedSocket> {
if let Some(bind_addr) = bind_addr {
let addr = SocketAddr::new(bind_addr, port as _);
return FramedSocket::new_reuse(&addr, true, rmem).await;
}
let addr = SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), port as _);
if let Ok(s) = FramedSocket::new_reuse(&addr, true, rmem).await {
log::debug!("listen on udp {:?}", s.local_addr());
@@ -1595,8 +1678,8 @@ async fn create_udp_listener(port: i32, rmem: usize) -> ResultType<FramedSocket>
}
#[inline]
async fn create_tcp_listener(port: i32) -> ResultType<TcpListener> {
let s = listen_any(port as _).await?;
async fn create_tcp_listener(bind_addr: Option<IpAddr>, port: i32) -> ResultType<TcpListener> {
let s = listen_tcp(bind_addr, port as _).await?;
log::debug!("listen on tcp {:?}", s.local_addr());
Ok(s)
}
@@ -1619,3 +1702,15 @@ fn get_symetric_key_from_msg(
Err(e) => panic!("Error while opening the seal key{:?}", e),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[hbb_common::tokio::test]
async fn udp_listener_uses_bind_address() {
let bind_addr = IpAddr::V4(Ipv4Addr::LOCALHOST);
let socket = create_udp_listener(Some(bind_addr), 0, 0).await.unwrap();
assert_eq!(socket.local_addr().unwrap().ip(), bind_addr);
}
}