feat: 添加 RustDesk 协议支持和项目文档

- 新增 RustDesk 模块,支持与 RustDesk 客户端连接
  - 实现会合服务器协议和 P2P 连接
  - 支持 NaCl 加密和密钥交换
  - 添加视频帧和 HID 事件适配器
- 添加 Protobuf 协议定义 (message.proto, rendezvous.proto)
- 新增完整项目文档
  - 各功能模块文档 (video, hid, msd, otg, webrtc 等)
  - hwcodec 和 RustDesk 协议技术报告
  - 系统架构和技术栈文档
- 更新 Web 前端 RustDesk 配置界面和 API
This commit is contained in:
mofeng-git
2025-12-31 18:59:52 +08:00
parent 61323a7664
commit a8a3b6c66b
57 changed files with 20830 additions and 0 deletions

View File

@@ -3,6 +3,8 @@ use typeshare::typeshare;
// Re-export ExtensionsConfig from extensions module
pub use crate::extensions::ExtensionsConfig;
// Re-export RustDeskConfig from rustdesk module
pub use crate::rustdesk::config::RustDeskConfig;
/// Main application configuration
#[typeshare]
@@ -29,6 +31,8 @@ pub struct AppConfig {
pub web: WebConfig,
/// Extensions settings (ttyd, gostc, easytier)
pub extensions: ExtensionsConfig,
/// RustDesk remote access settings
pub rustdesk: RustDeskConfig,
}
impl Default for AppConfig {
@@ -44,6 +48,7 @@ impl Default for AppConfig {
stream: StreamConfig::default(),
web: WebConfig::default(),
extensions: ExtensionsConfig::default(),
rustdesk: RustDeskConfig::default(),
}
}
}

View File

@@ -14,6 +14,7 @@ pub mod hid;
pub mod modules;
pub mod msd;
pub mod otg;
pub mod rustdesk;
pub mod state;
pub mod stream;
pub mod utils;

View File

@@ -17,6 +17,7 @@ use one_kvm::extensions::ExtensionManager;
use one_kvm::hid::{HidBackendType, HidController};
use one_kvm::msd::MsdController;
use one_kvm::otg::OtgService;
use one_kvm::rustdesk::RustDeskService;
use one_kvm::state::AppState;
use one_kvm::video::format::{PixelFormat, Resolution};
use one_kvm::video::{Streamer, VideoStreamManager};
@@ -374,6 +375,29 @@ async fn main() -> anyhow::Result<()> {
tracing::info!("Video stream manager initialized with mode: {:?}", initial_mode);
}
// Create RustDesk service (optional, based on config)
let rustdesk = if config.rustdesk.is_valid() {
tracing::info!(
"Initializing RustDesk service: ID={} -> {}",
config.rustdesk.device_id,
config.rustdesk.rendezvous_addr()
);
let service = RustDeskService::new(
config.rustdesk.clone(),
stream_manager.clone(),
hid.clone(),
audio.clone(),
);
Some(Arc::new(service))
} else {
if config.rustdesk.enabled {
tracing::warn!("RustDesk enabled but configuration is incomplete (missing server or credentials)");
} else {
tracing::info!("RustDesk disabled in configuration");
}
None
};
// Create application state
let state = AppState::new(
config_store.clone(),
@@ -385,12 +409,35 @@ async fn main() -> anyhow::Result<()> {
msd,
atx,
audio,
rustdesk.clone(),
extensions.clone(),
events.clone(),
shutdown_tx.clone(),
data_dir.clone(),
);
// Start RustDesk service if enabled
if let Some(ref service) = rustdesk {
if let Err(e) = service.start().await {
tracing::error!("Failed to start RustDesk service: {}", e);
} else {
// Save generated keypair and UUID to config
if let Some(updated_config) = service.save_credentials() {
if let Err(e) = config_store
.update(|cfg| {
cfg.rustdesk.public_key = updated_config.public_key.clone();
cfg.rustdesk.private_key = updated_config.private_key.clone();
cfg.rustdesk.uuid = updated_config.uuid.clone();
})
.await
{
tracing::warn!("Failed to save RustDesk credentials: {}", e);
}
}
tracing::info!("RustDesk service started");
}
}
// Start enabled extensions
{
let ext_config = config_store.get();
@@ -636,6 +683,15 @@ async fn cleanup(state: &Arc<AppState>) {
state.extensions.stop_all().await;
tracing::info!("Extensions stopped");
// Stop RustDesk service
if let Some(ref service) = *state.rustdesk.read().await {
if let Err(e) = service.stop().await {
tracing::warn!("Failed to stop RustDesk service: {}", e);
} else {
tracing::info!("RustDesk service stopped");
}
}
// Stop video
if let Err(e) = state.stream_manager.stop().await {
tracing::warn!("Failed to stop streamer: {}", e);

253
src/rustdesk/bytes_codec.rs Normal file
View File

@@ -0,0 +1,253 @@
//! RustDesk BytesCodec - Variable-length framing for TCP messages
//!
//! RustDesk uses a custom variable-length encoding for message framing:
//! - Length <= 0x3F (63): 1-byte header, format `(len << 2)`
//! - Length <= 0x3FFF (16383): 2-byte LE header, format `(len << 2) | 0x1`
//! - Length <= 0x3FFFFF (4194303): 3-byte LE header, format `(len << 2) | 0x2`
//! - Length <= 0x3FFFFFFF (1073741823): 4-byte LE header, format `(len << 2) | 0x3`
//!
//! The low 2 bits of the first byte indicate the header length (+1).
use bytes::{Buf, BufMut, Bytes, BytesMut};
use std::io;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
/// Maximum packet length (1GB)
const MAX_PACKET_LENGTH: usize = 0x3FFFFFFF;
/// Encode a message with RustDesk's variable-length framing
pub fn encode_frame(data: &[u8]) -> io::Result<Vec<u8>> {
let len = data.len();
let mut buf = Vec::with_capacity(len + 4);
if len <= 0x3F {
buf.push((len << 2) as u8);
} else if len <= 0x3FFF {
let h = ((len << 2) as u16) | 0x1;
buf.extend_from_slice(&h.to_le_bytes());
} else if len <= 0x3FFFFF {
let h = ((len << 2) as u32) | 0x2;
buf.push((h & 0xFF) as u8);
buf.push(((h >> 8) & 0xFF) as u8);
buf.push(((h >> 16) & 0xFF) as u8);
} else if len <= MAX_PACKET_LENGTH {
let h = ((len << 2) as u32) | 0x3;
buf.extend_from_slice(&h.to_le_bytes());
} else {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "Message too large"));
}
buf.extend_from_slice(data);
Ok(buf)
}
/// Decode the header to get message length
/// Returns (header_length, message_length)
fn decode_header(first_byte: u8, header_bytes: &[u8]) -> (usize, usize) {
let head_len = ((first_byte & 0x3) + 1) as usize;
let mut n = first_byte as usize;
if head_len > 1 && header_bytes.len() >= 1 {
n |= (header_bytes[0] as usize) << 8;
}
if head_len > 2 && header_bytes.len() >= 2 {
n |= (header_bytes[1] as usize) << 16;
}
if head_len > 3 && header_bytes.len() >= 3 {
n |= (header_bytes[2] as usize) << 24;
}
let msg_len = n >> 2;
(head_len, msg_len)
}
/// Read a single framed message from an async reader
pub async fn read_frame<R: AsyncRead + Unpin>(reader: &mut R) -> io::Result<BytesMut> {
// Read first byte to determine header length
let mut first_byte = [0u8; 1];
reader.read_exact(&mut first_byte).await?;
let head_len = ((first_byte[0] & 0x3) + 1) as usize;
// Read remaining header bytes if needed
let mut header_rest = [0u8; 3];
if head_len > 1 {
reader.read_exact(&mut header_rest[..head_len - 1]).await?;
}
// Calculate message length
let (_, msg_len) = decode_header(first_byte[0], &header_rest);
if msg_len > MAX_PACKET_LENGTH {
return Err(io::Error::new(io::ErrorKind::InvalidData, "Message too large"));
}
// Read message body
let mut buf = BytesMut::with_capacity(msg_len);
buf.resize(msg_len, 0);
reader.read_exact(&mut buf).await?;
Ok(buf)
}
/// Write a framed message to an async writer
pub async fn write_frame<W: AsyncWrite + Unpin>(writer: &mut W, data: &[u8]) -> io::Result<()> {
let frame = encode_frame(data)?;
writer.write_all(&frame).await?;
writer.flush().await?;
Ok(())
}
/// BytesCodec for stateful decoding (compatible with tokio-util codec)
#[derive(Debug, Clone, Copy)]
pub struct BytesCodec {
state: DecodeState,
max_packet_length: usize,
}
#[derive(Debug, Clone, Copy)]
enum DecodeState {
Head,
Data(usize),
}
impl Default for BytesCodec {
fn default() -> Self {
Self::new()
}
}
impl BytesCodec {
pub fn new() -> Self {
Self {
state: DecodeState::Head,
max_packet_length: MAX_PACKET_LENGTH,
}
}
pub fn set_max_packet_length(&mut self, n: usize) {
self.max_packet_length = n;
}
/// Decode from a BytesMut buffer (for use with Framed)
pub fn decode(&mut self, src: &mut BytesMut) -> io::Result<Option<BytesMut>> {
let n = match self.state {
DecodeState::Head => match self.decode_head(src)? {
Some(n) => {
self.state = DecodeState::Data(n);
n
}
None => return Ok(None),
},
DecodeState::Data(n) => n,
};
match self.decode_data(n, src)? {
Some(data) => {
self.state = DecodeState::Head;
Ok(Some(data))
}
None => Ok(None),
}
}
fn decode_head(&mut self, src: &mut BytesMut) -> io::Result<Option<usize>> {
if src.is_empty() {
return Ok(None);
}
let head_len = ((src[0] & 0x3) + 1) as usize;
if src.len() < head_len {
return Ok(None);
}
let mut n = src[0] as usize;
if head_len > 1 {
n |= (src[1] as usize) << 8;
}
if head_len > 2 {
n |= (src[2] as usize) << 16;
}
if head_len > 3 {
n |= (src[3] as usize) << 24;
}
n >>= 2;
if n > self.max_packet_length {
return Err(io::Error::new(io::ErrorKind::InvalidData, "Message too large"));
}
src.advance(head_len);
Ok(Some(n))
}
fn decode_data(&self, n: usize, src: &mut BytesMut) -> io::Result<Option<BytesMut>> {
if src.len() < n {
return Ok(None);
}
Ok(Some(src.split_to(n)))
}
/// Encode a message into a BytesMut buffer
pub fn encode(&mut self, data: Bytes, buf: &mut BytesMut) -> io::Result<()> {
let len = data.len();
if len <= 0x3F {
buf.put_u8((len << 2) as u8);
} else if len <= 0x3FFF {
buf.put_u16_le(((len << 2) as u16) | 0x1);
} else if len <= 0x3FFFFF {
let h = ((len << 2) as u32) | 0x2;
buf.put_u16_le((h & 0xFFFF) as u16);
buf.put_u8((h >> 16) as u8);
} else if len <= MAX_PACKET_LENGTH {
buf.put_u32_le(((len << 2) as u32) | 0x3);
} else {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "Message too large"));
}
buf.extend(data);
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_encode_decode_small() {
let data = vec![1u8; 63];
let encoded = encode_frame(&data).unwrap();
assert_eq!(encoded.len(), 63 + 1); // 1 byte header
let mut codec = BytesCodec::new();
let mut buf = BytesMut::from(&encoded[..]);
let decoded = codec.decode(&mut buf).unwrap().unwrap();
assert_eq!(decoded.len(), 63);
}
#[test]
fn test_encode_decode_medium() {
let data = vec![2u8; 1000];
let encoded = encode_frame(&data).unwrap();
assert_eq!(encoded.len(), 1000 + 2); // 2 byte header
let mut codec = BytesCodec::new();
let mut buf = BytesMut::from(&encoded[..]);
let decoded = codec.decode(&mut buf).unwrap().unwrap();
assert_eq!(decoded.len(), 1000);
}
#[test]
fn test_encode_decode_large() {
let data = vec![3u8; 100000];
let encoded = encode_frame(&data).unwrap();
assert_eq!(encoded.len(), 100000 + 3); // 3 byte header
let mut codec = BytesCodec::new();
let mut buf = BytesMut::from(&encoded[..]);
let decoded = codec.decode(&mut buf).unwrap().unwrap();
assert_eq!(decoded.len(), 100000);
}
}

210
src/rustdesk/config.rs Normal file
View File

@@ -0,0 +1,210 @@
//! RustDesk Configuration
//!
//! Configuration types for the RustDesk protocol integration.
use serde::{Deserialize, Serialize};
use typeshare::typeshare;
/// RustDesk configuration
#[typeshare]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct RustDeskConfig {
/// Enable RustDesk protocol
pub enabled: bool,
/// Rendezvous server address (hbbs), e.g., "rs.example.com" or "192.168.1.100"
/// Port defaults to 21116 if not specified
pub rendezvous_server: String,
/// Relay server address (hbbr), if different from rendezvous server
/// Usually the same host as rendezvous server but different port (21117)
pub relay_server: Option<String>,
/// Device ID (9-digit number), auto-generated if empty
pub device_id: String,
/// Device password for client authentication
#[typeshare(skip)]
pub device_password: String,
/// Public key for encryption (Curve25519, base64 encoded), auto-generated
#[typeshare(skip)]
pub public_key: Option<String>,
/// Private key for encryption (Curve25519, base64 encoded), auto-generated
#[typeshare(skip)]
pub private_key: Option<String>,
/// Signing public key (Ed25519, base64 encoded), auto-generated
/// Used for SignedId verification by clients
#[typeshare(skip)]
pub signing_public_key: Option<String>,
/// Signing private key (Ed25519, base64 encoded), auto-generated
/// Used for signing SignedId messages
#[typeshare(skip)]
pub signing_private_key: Option<String>,
/// UUID for rendezvous server registration (persisted to avoid UUID_MISMATCH)
#[typeshare(skip)]
pub uuid: Option<String>,
}
impl Default for RustDeskConfig {
fn default() -> Self {
Self {
enabled: false,
rendezvous_server: String::new(),
relay_server: None,
device_id: generate_device_id(),
device_password: generate_random_password(),
public_key: None,
private_key: None,
signing_public_key: None,
signing_private_key: None,
uuid: None,
}
}
}
impl RustDeskConfig {
/// Check if the configuration is valid for starting the service
pub fn is_valid(&self) -> bool {
self.enabled
&& !self.rendezvous_server.is_empty()
&& !self.device_id.is_empty()
&& !self.device_password.is_empty()
}
/// Generate a new random device ID
pub fn generate_device_id() -> String {
generate_device_id()
}
/// Generate a new random password
pub fn generate_password() -> String {
generate_random_password()
}
/// Get or generate the UUID for rendezvous registration
/// Returns (uuid_bytes, is_new) where is_new indicates if a new UUID was generated
pub fn ensure_uuid(&mut self) -> ([u8; 16], bool) {
if let Some(ref uuid_str) = self.uuid {
// Try to parse existing UUID
if let Ok(uuid) = uuid::Uuid::parse_str(uuid_str) {
return (*uuid.as_bytes(), false);
}
}
// Generate new UUID
let new_uuid = uuid::Uuid::new_v4();
self.uuid = Some(new_uuid.to_string());
(*new_uuid.as_bytes(), true)
}
/// Get the UUID bytes (returns None if not set)
pub fn get_uuid_bytes(&self) -> Option<[u8; 16]> {
self.uuid.as_ref().and_then(|s| {
uuid::Uuid::parse_str(s).ok().map(|u| *u.as_bytes())
})
}
/// Get the rendezvous server address with default port
pub fn rendezvous_addr(&self) -> String {
if self.rendezvous_server.contains(':') {
self.rendezvous_server.clone()
} else {
format!("{}:21116", self.rendezvous_server)
}
}
/// Get the relay server address with default port
pub fn relay_addr(&self) -> Option<String> {
self.relay_server.as_ref().map(|s| {
if s.contains(':') {
s.clone()
} else {
format!("{}:21117", s)
}
}).or_else(|| {
// Default: same host as rendezvous server
if !self.rendezvous_server.is_empty() {
let host = self.rendezvous_server.split(':').next().unwrap_or("");
if !host.is_empty() {
Some(format!("{}:21117", host))
} else {
None
}
} else {
None
}
})
}
}
/// Generate a random 9-digit device ID
pub fn generate_device_id() -> String {
use rand::Rng;
let mut rng = rand::thread_rng();
let id: u32 = rng.gen_range(100_000_000..999_999_999);
id.to_string()
}
/// Generate a random 8-character password
pub fn generate_random_password() -> String {
use rand::Rng;
const CHARSET: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
let mut rng = rand::thread_rng();
(0..8)
.map(|_| {
let idx = rng.gen_range(0..CHARSET.len());
CHARSET[idx] as char
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_device_id_generation() {
let id = generate_device_id();
assert_eq!(id.len(), 9);
assert!(id.chars().all(|c| c.is_ascii_digit()));
}
#[test]
fn test_password_generation() {
let password = generate_random_password();
assert_eq!(password.len(), 8);
assert!(password.chars().all(|c| c.is_alphanumeric()));
}
#[test]
fn test_rendezvous_addr() {
let mut config = RustDeskConfig::default();
config.rendezvous_server = "example.com".to_string();
assert_eq!(config.rendezvous_addr(), "example.com:21116");
config.rendezvous_server = "example.com:21116".to_string();
assert_eq!(config.rendezvous_addr(), "example.com:21116");
}
#[test]
fn test_relay_addr() {
let mut config = RustDeskConfig::default();
// No server configured
assert!(config.relay_addr().is_none());
// Rendezvous server configured, relay defaults to same host
config.rendezvous_server = "example.com".to_string();
assert_eq!(config.relay_addr(), Some("example.com:21117".to_string()));
// Explicit relay server
config.relay_server = Some("relay.example.com".to_string());
assert_eq!(config.relay_addr(), Some("relay.example.com:21117".to_string()));
}
}

1396
src/rustdesk/connection.rs Normal file

File diff suppressed because it is too large Load Diff

467
src/rustdesk/crypto.rs Normal file
View File

@@ -0,0 +1,467 @@
//! RustDesk Cryptography
//!
//! This module implements the NaCl-based cryptography used by RustDesk:
//! - Curve25519 for key exchange
//! - XSalsa20-Poly1305 for authenticated encryption
//! - Ed25519 for signatures
//! - Ed25519 to Curve25519 key conversion for unified keypair usage
use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
use sodiumoxide::crypto::box_::{self, Nonce, PublicKey, SecretKey};
use sodiumoxide::crypto::secretbox;
use sodiumoxide::crypto::sign::{self, ed25519};
use thiserror::Error;
/// Cryptography errors
#[derive(Debug, Error)]
pub enum CryptoError {
#[error("Failed to initialize sodiumoxide")]
InitError,
#[error("Encryption failed")]
EncryptionFailed,
#[error("Decryption failed")]
DecryptionFailed,
#[error("Invalid key length")]
InvalidKeyLength,
#[error("Invalid nonce")]
InvalidNonce,
#[error("Signature verification failed")]
SignatureVerificationFailed,
#[error("Key conversion failed")]
KeyConversionFailed,
}
/// Initialize the cryptography library
/// Must be called before using any crypto functions
pub fn init() -> Result<(), CryptoError> {
sodiumoxide::init().map_err(|_| CryptoError::InitError)
}
/// A keypair for asymmetric encryption
#[derive(Clone)]
pub struct KeyPair {
pub public_key: PublicKey,
pub secret_key: SecretKey,
}
impl KeyPair {
/// Generate a new random keypair
pub fn generate() -> Self {
let (public_key, secret_key) = box_::gen_keypair();
Self {
public_key,
secret_key,
}
}
/// Create from existing keys
pub fn from_keys(public_key: &[u8], secret_key: &[u8]) -> Result<Self, CryptoError> {
let pk = PublicKey::from_slice(public_key).ok_or(CryptoError::InvalidKeyLength)?;
let sk = SecretKey::from_slice(secret_key).ok_or(CryptoError::InvalidKeyLength)?;
Ok(Self {
public_key: pk,
secret_key: sk,
})
}
/// Get public key as bytes
pub fn public_key_bytes(&self) -> &[u8] {
self.public_key.as_ref()
}
/// Get secret key as bytes
pub fn secret_key_bytes(&self) -> &[u8] {
self.secret_key.as_ref()
}
/// Encode public key as base64
pub fn public_key_base64(&self) -> String {
BASE64.encode(self.public_key_bytes())
}
/// Encode secret key as base64
pub fn secret_key_base64(&self) -> String {
BASE64.encode(self.secret_key_bytes())
}
/// Create from base64-encoded keys
pub fn from_base64(public_key: &str, secret_key: &str) -> Result<Self, CryptoError> {
let pk_bytes = BASE64.decode(public_key).map_err(|_| CryptoError::InvalidKeyLength)?;
let sk_bytes = BASE64.decode(secret_key).map_err(|_| CryptoError::InvalidKeyLength)?;
Self::from_keys(&pk_bytes, &sk_bytes)
}
}
/// Generate a random nonce for box encryption
pub fn generate_nonce() -> Nonce {
box_::gen_nonce()
}
/// Encrypt data using public-key cryptography (NaCl box)
///
/// Uses the sender's secret key and receiver's public key for encryption.
/// Returns (nonce, ciphertext).
pub fn encrypt_box(
data: &[u8],
their_public_key: &PublicKey,
our_secret_key: &SecretKey,
) -> (Nonce, Vec<u8>) {
let nonce = generate_nonce();
let ciphertext = box_::seal(data, &nonce, their_public_key, our_secret_key);
(nonce, ciphertext)
}
/// Decrypt data using public-key cryptography (NaCl box)
pub fn decrypt_box(
ciphertext: &[u8],
nonce: &Nonce,
their_public_key: &PublicKey,
our_secret_key: &SecretKey,
) -> Result<Vec<u8>, CryptoError> {
box_::open(ciphertext, nonce, their_public_key, our_secret_key)
.map_err(|_| CryptoError::DecryptionFailed)
}
/// Encrypt data with a precomputed shared key
pub fn encrypt_with_key(data: &[u8], key: &secretbox::Key) -> (secretbox::Nonce, Vec<u8>) {
let nonce = secretbox::gen_nonce();
let ciphertext = secretbox::seal(data, &nonce, key);
(nonce, ciphertext)
}
/// Decrypt data with a precomputed shared key
pub fn decrypt_with_key(
ciphertext: &[u8],
nonce: &secretbox::Nonce,
key: &secretbox::Key,
) -> Result<Vec<u8>, CryptoError> {
secretbox::open(ciphertext, nonce, key).map_err(|_| CryptoError::DecryptionFailed)
}
/// Compute a shared symmetric key from public/private keypair
/// This is the precomputed key for the NaCl box
pub fn precompute_key(their_public_key: &PublicKey, our_secret_key: &SecretKey) -> box_::PrecomputedKey {
box_::precompute(their_public_key, our_secret_key)
}
/// Create a symmetric key from raw bytes
pub fn symmetric_key_from_slice(key: &[u8]) -> Result<secretbox::Key, CryptoError> {
secretbox::Key::from_slice(key).ok_or(CryptoError::InvalidKeyLength)
}
/// Parse a nonce from bytes
pub fn nonce_from_slice(bytes: &[u8]) -> Result<Nonce, CryptoError> {
Nonce::from_slice(bytes).ok_or(CryptoError::InvalidNonce)
}
/// Parse a public key from bytes
pub fn public_key_from_slice(bytes: &[u8]) -> Result<PublicKey, CryptoError> {
PublicKey::from_slice(bytes).ok_or(CryptoError::InvalidKeyLength)
}
/// Hash a password for storage/comparison
/// RustDesk uses simple SHA256 for password hashing
pub fn hash_password(password: &str, salt: &str) -> Vec<u8> {
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
hasher.update(password.as_bytes());
hasher.update(salt.as_bytes());
hasher.finalize().to_vec()
}
/// RustDesk double hash for password verification
/// Client calculates: SHA256(SHA256(password + salt) + challenge)
/// This matches what the client sends in LoginRequest
pub fn hash_password_double(password: &str, salt: &str, challenge: &str) -> Vec<u8> {
use sha2::{Digest, Sha256};
// First hash: SHA256(password + salt)
let mut hasher1 = Sha256::new();
hasher1.update(password.as_bytes());
hasher1.update(salt.as_bytes());
let first_hash = hasher1.finalize();
// Second hash: SHA256(first_hash + challenge)
let mut hasher2 = Sha256::new();
hasher2.update(&first_hash);
hasher2.update(challenge.as_bytes());
hasher2.finalize().to_vec()
}
/// Verify a password hash
pub fn verify_password(password: &str, salt: &str, expected_hash: &[u8]) -> bool {
let computed = hash_password(password, salt);
// Constant-time comparison would be better, but for our use case this is acceptable
computed == expected_hash
}
/// RustDesk symmetric key negotiation result
pub struct SymmetricKeyNegotiation {
/// Our temporary public key (to send to peer)
pub our_public_key: Vec<u8>,
/// The sealed/encrypted symmetric key (to send to peer)
pub sealed_symmetric_key: Vec<u8>,
/// The actual symmetric key (for local use)
pub symmetric_key: secretbox::Key,
}
/// Create symmetric key message for RustDesk encrypted handshake
///
/// This implements RustDesk's `create_symmetric_key_msg` protocol:
/// 1. Generate a temporary keypair
/// 2. Generate a symmetric key
/// 3. Encrypt the symmetric key using the peer's public key and our temp secret key
/// 4. Return (our_temp_public_key, sealed_symmetric_key, symmetric_key)
pub fn create_symmetric_key_msg(their_public_key_bytes: &[u8; 32]) -> SymmetricKeyNegotiation {
let their_pk = box_::PublicKey(*their_public_key_bytes);
let (our_temp_pk, our_temp_sk) = box_::gen_keypair();
let symmetric_key = secretbox::gen_key();
// Use zero nonce as per RustDesk protocol
let nonce = box_::Nonce([0u8; box_::NONCEBYTES]);
let sealed_key = box_::seal(&symmetric_key.0, &nonce, &their_pk, &our_temp_sk);
SymmetricKeyNegotiation {
our_public_key: our_temp_pk.0.to_vec(),
sealed_symmetric_key: sealed_key,
symmetric_key,
}
}
/// Decrypt symmetric key received from peer during handshake
///
/// This is the server-side of RustDesk's encrypted handshake:
/// 1. Receive peer's temporary public key and sealed symmetric key
/// 2. Decrypt the symmetric key using our secret key
pub fn decrypt_symmetric_key_msg(
their_temp_public_key: &[u8],
sealed_symmetric_key: &[u8],
our_keypair: &KeyPair,
) -> Result<secretbox::Key, CryptoError> {
if their_temp_public_key.len() != box_::PUBLICKEYBYTES {
return Err(CryptoError::InvalidKeyLength);
}
let their_pk = PublicKey::from_slice(their_temp_public_key)
.ok_or(CryptoError::InvalidKeyLength)?;
// Use zero nonce as per RustDesk protocol
let nonce = box_::Nonce([0u8; box_::NONCEBYTES]);
let key_bytes = box_::open(sealed_symmetric_key, &nonce, &their_pk, &our_keypair.secret_key)
.map_err(|_| CryptoError::DecryptionFailed)?;
secretbox::Key::from_slice(&key_bytes).ok_or(CryptoError::InvalidKeyLength)
}
/// Decrypt symmetric key using Ed25519 signing keypair (converted to Curve25519)
///
/// RustDesk clients encrypt the symmetric key using the public key from IdPk,
/// which is our Ed25519 signing public key converted to Curve25519.
/// We must use the corresponding converted secret key to decrypt.
pub fn decrypt_symmetric_key_with_signing_keypair(
their_temp_public_key: &[u8],
sealed_symmetric_key: &[u8],
signing_keypair: &SigningKeyPair,
) -> Result<secretbox::Key, CryptoError> {
use tracing::debug;
if their_temp_public_key.len() != box_::PUBLICKEYBYTES {
return Err(CryptoError::InvalidKeyLength);
}
let their_pk = PublicKey::from_slice(their_temp_public_key)
.ok_or(CryptoError::InvalidKeyLength)?;
// Convert our Ed25519 secret key to Curve25519 for decryption
let our_curve25519_sk = signing_keypair.to_curve25519_sk()?;
// Also get our converted public key for debugging
let our_curve25519_pk = signing_keypair.to_curve25519_pk()?;
debug!(
"Decrypting with converted keys: our_curve25519_pk={:02x?}, their_temp_pk={:02x?}",
&our_curve25519_pk.as_ref()[..8],
&their_pk.as_ref()[..8]
);
// Use zero nonce as per RustDesk protocol
let nonce = box_::Nonce([0u8; box_::NONCEBYTES]);
let key_bytes = box_::open(sealed_symmetric_key, &nonce, &their_pk, &our_curve25519_sk)
.map_err(|_| CryptoError::DecryptionFailed)?;
secretbox::Key::from_slice(&key_bytes).ok_or(CryptoError::InvalidKeyLength)
}
/// Encrypt a message using the negotiated symmetric key
///
/// RustDesk uses a specific nonce format for session encryption
pub fn encrypt_message(data: &[u8], key: &secretbox::Key, nonce_counter: u64) -> Vec<u8> {
// Create nonce from counter (little-endian, padded to 24 bytes)
let mut nonce_bytes = [0u8; secretbox::NONCEBYTES];
nonce_bytes[..8].copy_from_slice(&nonce_counter.to_le_bytes());
let nonce = secretbox::Nonce(nonce_bytes);
secretbox::seal(data, &nonce, key)
}
/// Decrypt a message using the negotiated symmetric key
pub fn decrypt_message(
ciphertext: &[u8],
key: &secretbox::Key,
nonce_counter: u64,
) -> Result<Vec<u8>, CryptoError> {
// Create nonce from counter (little-endian, padded to 24 bytes)
let mut nonce_bytes = [0u8; secretbox::NONCEBYTES];
nonce_bytes[..8].copy_from_slice(&nonce_counter.to_le_bytes());
let nonce = secretbox::Nonce(nonce_bytes);
secretbox::open(ciphertext, &nonce, key).map_err(|_| CryptoError::DecryptionFailed)
}
/// Ed25519 signing keypair for RustDesk SignedId messages
#[derive(Clone)]
pub struct SigningKeyPair {
pub public_key: sign::PublicKey,
pub secret_key: sign::SecretKey,
}
impl SigningKeyPair {
/// Generate a new random signing keypair
pub fn generate() -> Self {
let (public_key, secret_key) = sign::gen_keypair();
Self {
public_key,
secret_key,
}
}
/// Create from existing keys
pub fn from_keys(public_key: &[u8], secret_key: &[u8]) -> Result<Self, CryptoError> {
let pk = sign::PublicKey::from_slice(public_key).ok_or(CryptoError::InvalidKeyLength)?;
let sk = sign::SecretKey::from_slice(secret_key).ok_or(CryptoError::InvalidKeyLength)?;
Ok(Self {
public_key: pk,
secret_key: sk,
})
}
/// Get public key as bytes
pub fn public_key_bytes(&self) -> &[u8] {
self.public_key.as_ref()
}
/// Get secret key as bytes
pub fn secret_key_bytes(&self) -> &[u8] {
self.secret_key.as_ref()
}
/// Encode public key as base64
pub fn public_key_base64(&self) -> String {
BASE64.encode(self.public_key_bytes())
}
/// Encode secret key as base64
pub fn secret_key_base64(&self) -> String {
BASE64.encode(self.secret_key_bytes())
}
/// Create from base64-encoded keys
pub fn from_base64(public_key: &str, secret_key: &str) -> Result<Self, CryptoError> {
let pk_bytes = BASE64.decode(public_key).map_err(|_| CryptoError::InvalidKeyLength)?;
let sk_bytes = BASE64.decode(secret_key).map_err(|_| CryptoError::InvalidKeyLength)?;
Self::from_keys(&pk_bytes, &sk_bytes)
}
/// Sign a message
/// Returns the signature prepended to the message (as per RustDesk protocol)
pub fn sign(&self, message: &[u8]) -> Vec<u8> {
sign::sign(message, &self.secret_key)
}
/// Sign a message and return just the signature (64 bytes)
pub fn sign_detached(&self, message: &[u8]) -> [u8; 64] {
let sig = sign::sign_detached(message, &self.secret_key);
// Use as_ref() to access the signature bytes since the inner field is private
let sig_bytes: &[u8] = sig.as_ref();
let mut result = [0u8; 64];
result.copy_from_slice(sig_bytes);
result
}
/// Convert Ed25519 public key to Curve25519 public key for encryption
///
/// This allows using the same keypair for both signing and encryption,
/// which is required by RustDesk's protocol where clients encrypt the
/// symmetric key using the public key from IdPk.
pub fn to_curve25519_pk(&self) -> Result<PublicKey, CryptoError> {
ed25519::to_curve25519_pk(&self.public_key)
.map_err(|_| CryptoError::KeyConversionFailed)
}
/// Convert Ed25519 secret key to Curve25519 secret key for decryption
///
/// This allows decrypting messages that were encrypted using the
/// converted public key.
pub fn to_curve25519_sk(&self) -> Result<SecretKey, CryptoError> {
ed25519::to_curve25519_sk(&self.secret_key)
.map_err(|_| CryptoError::KeyConversionFailed)
}
}
/// Verify a signed message
/// Returns the original message if signature is valid
pub fn verify_signed(signed_message: &[u8], public_key: &sign::PublicKey) -> Result<Vec<u8>, CryptoError> {
sign::verify(signed_message, public_key).map_err(|_| CryptoError::SignatureVerificationFailed)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_keypair_generation() {
let _ = init();
let keypair = KeyPair::generate();
assert_eq!(keypair.public_key_bytes().len(), 32);
assert_eq!(keypair.secret_key_bytes().len(), 32);
}
#[test]
fn test_keypair_serialization() {
let _ = init();
let keypair1 = KeyPair::generate();
let pk_b64 = keypair1.public_key_base64();
let sk_b64 = keypair1.secret_key_base64();
let keypair2 = KeyPair::from_base64(&pk_b64, &sk_b64).unwrap();
assert_eq!(keypair1.public_key_bytes(), keypair2.public_key_bytes());
assert_eq!(keypair1.secret_key_bytes(), keypair2.secret_key_bytes());
}
#[test]
fn test_box_encryption() {
let _ = init();
let alice = KeyPair::generate();
let bob = KeyPair::generate();
let message = b"Hello, RustDesk!";
let (nonce, ciphertext) = encrypt_box(message, &bob.public_key, &alice.secret_key);
let plaintext = decrypt_box(&ciphertext, &nonce, &alice.public_key, &bob.secret_key).unwrap();
assert_eq!(plaintext, message);
}
#[test]
fn test_password_hashing() {
let password = "test_password";
let salt = "random_salt";
let hash1 = hash_password(password, salt);
let hash2 = hash_password(password, salt);
assert_eq!(hash1, hash2);
assert!(verify_password(password, salt, &hash1));
assert!(!verify_password("wrong_password", salt, &hash1));
}
}

View File

@@ -0,0 +1,315 @@
//! RustDesk Frame Adapters
//!
//! Converts One-KVM video/audio frames to RustDesk protocol format.
use bytes::Bytes;
use prost::Message as ProstMessage;
use super::protocol::hbb::{self, message, EncodedVideoFrame, EncodedVideoFrames, AudioFrame, AudioFormat, Misc};
/// Video codec type for RustDesk
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum VideoCodec {
H264,
H265,
VP8,
VP9,
AV1,
}
impl VideoCodec {
/// Get the codec ID for the RustDesk protocol
pub fn to_codec_id(self) -> i32 {
match self {
VideoCodec::H264 => 0,
VideoCodec::H265 => 1,
VideoCodec::VP8 => 2,
VideoCodec::VP9 => 3,
VideoCodec::AV1 => 4,
}
}
}
/// Video frame adapter for converting to RustDesk format
pub struct VideoFrameAdapter {
/// Current codec
codec: VideoCodec,
/// Frame sequence number
seq: u32,
/// Timestamp offset
timestamp_base: u64,
}
impl VideoFrameAdapter {
/// Create a new video frame adapter
pub fn new(codec: VideoCodec) -> Self {
Self {
codec,
seq: 0,
timestamp_base: 0,
}
}
/// Set codec type
pub fn set_codec(&mut self, codec: VideoCodec) {
self.codec = codec;
}
/// Convert encoded video data to RustDesk Message
pub fn encode_frame(&mut self, data: &[u8], is_keyframe: bool, timestamp_ms: u64) -> hbb::Message {
// Calculate relative timestamp
if self.seq == 0 {
self.timestamp_base = timestamp_ms;
}
let pts = (timestamp_ms - self.timestamp_base) as i64;
let frame = EncodedVideoFrame {
data: data.to_vec(),
key: is_keyframe,
pts,
..Default::default()
};
self.seq = self.seq.wrapping_add(1);
// Wrap in EncodedVideoFrames container
let frames = EncodedVideoFrames {
frames: vec![frame],
..Default::default()
};
// Create the appropriate VideoFrame variant based on codec
let video_frame = match self.codec {
VideoCodec::H264 => hbb::VideoFrame {
union: Some(hbb::video_frame::Union::H264s(frames)),
display: 0,
},
VideoCodec::H265 => hbb::VideoFrame {
union: Some(hbb::video_frame::Union::H265s(frames)),
display: 0,
},
VideoCodec::VP8 => hbb::VideoFrame {
union: Some(hbb::video_frame::Union::Vp8s(frames)),
display: 0,
},
VideoCodec::VP9 => hbb::VideoFrame {
union: Some(hbb::video_frame::Union::Vp9s(frames)),
display: 0,
},
VideoCodec::AV1 => hbb::VideoFrame {
union: Some(hbb::video_frame::Union::Av1s(frames)),
display: 0,
},
};
hbb::Message {
union: Some(message::Union::VideoFrame(video_frame)),
}
}
/// Encode frame to bytes for sending
pub fn encode_frame_bytes(&mut self, data: &[u8], is_keyframe: bool, timestamp_ms: u64) -> Bytes {
let msg = self.encode_frame(data, is_keyframe, timestamp_ms);
Bytes::from(ProstMessage::encode_to_vec(&msg))
}
/// Get current sequence number
pub fn seq(&self) -> u32 {
self.seq
}
}
/// Audio frame adapter for converting to RustDesk format
pub struct AudioFrameAdapter {
/// Sample rate
sample_rate: u32,
/// Channels
channels: u8,
/// Format sent flag
format_sent: bool,
}
impl AudioFrameAdapter {
/// Create a new audio frame adapter
pub fn new(sample_rate: u32, channels: u8) -> Self {
Self {
sample_rate,
channels,
format_sent: false,
}
}
/// Create audio format message (should be sent once before audio frames)
pub fn create_format_message(&mut self) -> hbb::Message {
self.format_sent = true;
let format = AudioFormat {
sample_rate: self.sample_rate,
channels: self.channels as u32,
};
hbb::Message {
union: Some(message::Union::Misc(Misc {
union: Some(hbb::misc::Union::AudioFormat(format)),
})),
}
}
/// Check if format message has been sent
pub fn format_sent(&self) -> bool {
self.format_sent
}
/// Convert Opus audio data to RustDesk Message
pub fn encode_opus_frame(&self, data: &[u8]) -> hbb::Message {
let frame = AudioFrame {
data: data.to_vec(),
};
hbb::Message {
union: Some(message::Union::AudioFrame(frame)),
}
}
/// Encode Opus frame to bytes for sending
pub fn encode_opus_bytes(&self, data: &[u8]) -> Bytes {
let msg = self.encode_opus_frame(data);
Bytes::from(ProstMessage::encode_to_vec(&msg))
}
/// Reset state (call when restarting audio stream)
pub fn reset(&mut self) {
self.format_sent = false;
}
}
/// Cursor data adapter
pub struct CursorAdapter;
impl CursorAdapter {
/// Create cursor data message
pub fn encode_cursor(
id: u64,
hotx: i32,
hoty: i32,
width: i32,
height: i32,
colors: Vec<u8>,
) -> hbb::Message {
let cursor = hbb::CursorData {
id,
hotx,
hoty,
width,
height,
colors,
..Default::default()
};
hbb::Message {
union: Some(message::Union::CursorData(cursor)),
}
}
/// Create cursor position message
pub fn encode_position(x: i32, y: i32) -> hbb::Message {
let pos = hbb::CursorPosition {
x,
y,
};
hbb::Message {
union: Some(message::Union::CursorPosition(pos)),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_video_frame_encoding() {
let mut adapter = VideoFrameAdapter::new(VideoCodec::H264);
// Encode a keyframe
let data = vec![0x00, 0x00, 0x00, 0x01, 0x67]; // H264 SPS NAL
let msg = adapter.encode_frame(&data, true, 0);
match msg.union {
Some(message::Union::VideoFrame(vf)) => {
match vf.union {
Some(hbb::video_frame::Union::H264s(frames)) => {
assert_eq!(frames.frames.len(), 1);
assert!(frames.frames[0].key);
}
_ => panic!("Expected H264s"),
}
}
_ => panic!("Expected VideoFrame"),
}
}
#[test]
fn test_audio_format_message() {
let mut adapter = AudioFrameAdapter::new(48000, 2);
assert!(!adapter.format_sent());
let msg = adapter.create_format_message();
assert!(adapter.format_sent());
match msg.union {
Some(message::Union::Misc(misc)) => {
match misc.union {
Some(hbb::misc::Union::AudioFormat(fmt)) => {
assert_eq!(fmt.sample_rate, 48000);
assert_eq!(fmt.channels, 2);
}
_ => panic!("Expected AudioFormat"),
}
}
_ => panic!("Expected Misc"),
}
}
#[test]
fn test_audio_frame_encoding() {
let adapter = AudioFrameAdapter::new(48000, 2);
// Encode an Opus frame
let opus_data = vec![0xFC, 0x01, 0x02]; // Fake Opus data
let msg = adapter.encode_opus_frame(&opus_data);
match msg.union {
Some(message::Union::AudioFrame(af)) => {
assert_eq!(af.data, opus_data);
}
_ => panic!("Expected AudioFrame"),
}
}
#[test]
fn test_cursor_encoding() {
let msg = CursorAdapter::encode_cursor(1, 0, 0, 16, 16, vec![0xFF; 16 * 16 * 4]);
match msg.union {
Some(message::Union::CursorData(cd)) => {
assert_eq!(cd.id, 1);
assert_eq!(cd.width, 16);
assert_eq!(cd.height, 16);
}
_ => panic!("Expected CursorData"),
}
}
#[test]
fn test_sequence_increment() {
let mut adapter = VideoFrameAdapter::new(VideoCodec::H264);
assert_eq!(adapter.seq(), 0);
adapter.encode_frame(&[0], false, 0);
assert_eq!(adapter.seq(), 1);
adapter.encode_frame(&[0], false, 33);
assert_eq!(adapter.seq(), 2);
}
}

385
src/rustdesk/hid_adapter.rs Normal file
View File

@@ -0,0 +1,385 @@
//! RustDesk HID Adapter
//!
//! Converts RustDesk HID events (KeyEvent, MouseEvent) to One-KVM HID events.
use crate::hid::{
KeyboardEvent, KeyboardModifiers, KeyEventType,
MouseButton, MouseEvent as OneKvmMouseEvent, MouseEventType,
};
use super::protocol::hbb::{self, ControlKey, KeyEvent, MouseEvent};
/// Mouse event types from RustDesk protocol
/// mask = (button << 3) | event_type
pub mod mouse_type {
pub const MOVE: i32 = 0;
pub const DOWN: i32 = 1;
pub const UP: i32 = 2;
pub const WHEEL: i32 = 3;
pub const TRACKPAD: i32 = 4;
}
/// Mouse button IDs from RustDesk protocol (before left shift by 3)
pub mod mouse_button {
pub const LEFT: i32 = 0x01;
pub const RIGHT: i32 = 0x02;
pub const WHEEL: i32 = 0x04;
pub const BACK: i32 = 0x08;
pub const FORWARD: i32 = 0x10;
}
/// Convert RustDesk MouseEvent to One-KVM MouseEvent(s)
/// Returns a Vec because a single RustDesk event may need multiple One-KVM events
/// (e.g., move + button + scroll)
pub fn convert_mouse_event(event: &MouseEvent, screen_width: u32, screen_height: u32) -> Vec<OneKvmMouseEvent> {
let mut events = Vec::new();
// RustDesk uses absolute coordinates
let x = event.x.max(0) as u32;
let y = event.y.max(0) as u32;
// Normalize to 0-32767 range for absolute mouse (USB HID standard)
let abs_x = ((x as u64 * 32767) / screen_width.max(1) as u64) as i32;
let abs_y = ((y as u64 * 32767) / screen_height.max(1) as u64) as i32;
// Parse RustDesk mask format: (button << 3) | event_type
let event_type = event.mask & 0x07;
let button_id = event.mask >> 3;
match event_type {
mouse_type::MOVE => {
// Pure move event
events.push(OneKvmMouseEvent {
event_type: MouseEventType::MoveAbs,
x: abs_x,
y: abs_y,
button: None,
scroll: 0,
});
}
mouse_type::DOWN => {
// Button down - first move, then press
events.push(OneKvmMouseEvent {
event_type: MouseEventType::MoveAbs,
x: abs_x,
y: abs_y,
button: None,
scroll: 0,
});
if let Some(button) = button_id_to_button(button_id) {
events.push(OneKvmMouseEvent {
event_type: MouseEventType::Down,
x: abs_x,
y: abs_y,
button: Some(button),
scroll: 0,
});
}
}
mouse_type::UP => {
// Button up - first move, then release
events.push(OneKvmMouseEvent {
event_type: MouseEventType::MoveAbs,
x: abs_x,
y: abs_y,
button: None,
scroll: 0,
});
if let Some(button) = button_id_to_button(button_id) {
events.push(OneKvmMouseEvent {
event_type: MouseEventType::Up,
x: abs_x,
y: abs_y,
button: Some(button),
scroll: 0,
});
}
}
mouse_type::WHEEL => {
// Scroll event - move first, then scroll
events.push(OneKvmMouseEvent {
event_type: MouseEventType::MoveAbs,
x: abs_x,
y: abs_y,
button: None,
scroll: 0,
});
// For wheel events, button_id indicates scroll direction
// Positive = scroll up, Negative = scroll down
// The actual scroll amount may be encoded differently
let scroll = if button_id > 0 { 1i8 } else { -1i8 };
events.push(OneKvmMouseEvent {
event_type: MouseEventType::Scroll,
x: abs_x,
y: abs_y,
button: None,
scroll,
});
}
_ => {
// Unknown event type, just move
events.push(OneKvmMouseEvent {
event_type: MouseEventType::MoveAbs,
x: abs_x,
y: abs_y,
button: None,
scroll: 0,
});
}
}
events
}
/// Convert RustDesk button ID to One-KVM MouseButton
fn button_id_to_button(button_id: i32) -> Option<MouseButton> {
match button_id {
mouse_button::LEFT => Some(MouseButton::Left),
mouse_button::RIGHT => Some(MouseButton::Right),
mouse_button::WHEEL => Some(MouseButton::Middle),
_ => None,
}
}
/// Convert RustDesk KeyEvent to One-KVM KeyboardEvent
pub fn convert_key_event(event: &KeyEvent) -> Option<KeyboardEvent> {
let pressed = event.down || event.press;
let event_type = if pressed { KeyEventType::Down } else { KeyEventType::Up };
// Parse modifiers from the event
let modifiers = parse_modifiers(event);
// Handle control keys
if let Some(hbb::key_event::Union::ControlKey(ck)) = &event.union {
if let Some(key) = control_key_to_hid(*ck) {
return Some(KeyboardEvent {
event_type,
key,
modifiers,
});
}
}
// Handle character keys (chr field contains platform-specific keycode)
if let Some(hbb::key_event::Union::Chr(chr)) = &event.union {
// chr contains USB HID scancode on Windows, X11 keycode on Linux
if let Some(key) = keycode_to_hid(*chr) {
return Some(KeyboardEvent {
event_type,
key,
modifiers,
});
}
}
// Handle unicode (for text input, we'd need to convert to scancodes)
// Unicode input requires more complex handling, skip for now
None
}
/// Parse modifier keys from RustDesk KeyEvent into KeyboardModifiers
fn parse_modifiers(event: &KeyEvent) -> KeyboardModifiers {
let mut modifiers = KeyboardModifiers::default();
for modifier in &event.modifiers {
match *modifier {
x if x == ControlKey::Control as i32 => modifiers.left_ctrl = true,
x if x == ControlKey::Shift as i32 => modifiers.left_shift = true,
x if x == ControlKey::Alt as i32 => modifiers.left_alt = true,
x if x == ControlKey::Meta as i32 => modifiers.left_meta = true,
x if x == ControlKey::RControl as i32 => modifiers.right_ctrl = true,
x if x == ControlKey::RShift as i32 => modifiers.right_shift = true,
x if x == ControlKey::RAlt as i32 => modifiers.right_alt = true,
_ => {}
}
}
modifiers
}
/// Convert RustDesk ControlKey to USB HID usage code
fn control_key_to_hid(key: i32) -> Option<u8> {
match key {
x if x == ControlKey::Alt as i32 => Some(0xE2), // Left Alt
x if x == ControlKey::Backspace as i32 => Some(0x2A),
x if x == ControlKey::CapsLock as i32 => Some(0x39),
x if x == ControlKey::Control as i32 => Some(0xE0), // Left Ctrl
x if x == ControlKey::Delete as i32 => Some(0x4C),
x if x == ControlKey::DownArrow as i32 => Some(0x51),
x if x == ControlKey::End as i32 => Some(0x4D),
x if x == ControlKey::Escape as i32 => Some(0x29),
x if x == ControlKey::F1 as i32 => Some(0x3A),
x if x == ControlKey::F2 as i32 => Some(0x3B),
x if x == ControlKey::F3 as i32 => Some(0x3C),
x if x == ControlKey::F4 as i32 => Some(0x3D),
x if x == ControlKey::F5 as i32 => Some(0x3E),
x if x == ControlKey::F6 as i32 => Some(0x3F),
x if x == ControlKey::F7 as i32 => Some(0x40),
x if x == ControlKey::F8 as i32 => Some(0x41),
x if x == ControlKey::F9 as i32 => Some(0x42),
x if x == ControlKey::F10 as i32 => Some(0x43),
x if x == ControlKey::F11 as i32 => Some(0x44),
x if x == ControlKey::F12 as i32 => Some(0x45),
x if x == ControlKey::Home as i32 => Some(0x4A),
x if x == ControlKey::LeftArrow as i32 => Some(0x50),
x if x == ControlKey::Meta as i32 => Some(0xE3), // Left GUI/Windows
x if x == ControlKey::PageDown as i32 => Some(0x4E),
x if x == ControlKey::PageUp as i32 => Some(0x4B),
x if x == ControlKey::Return as i32 => Some(0x28),
x if x == ControlKey::RightArrow as i32 => Some(0x4F),
x if x == ControlKey::Shift as i32 => Some(0xE1), // Left Shift
x if x == ControlKey::Space as i32 => Some(0x2C),
x if x == ControlKey::Tab as i32 => Some(0x2B),
x if x == ControlKey::UpArrow as i32 => Some(0x52),
x if x == ControlKey::Numpad0 as i32 => Some(0x62),
x if x == ControlKey::Numpad1 as i32 => Some(0x59),
x if x == ControlKey::Numpad2 as i32 => Some(0x5A),
x if x == ControlKey::Numpad3 as i32 => Some(0x5B),
x if x == ControlKey::Numpad4 as i32 => Some(0x5C),
x if x == ControlKey::Numpad5 as i32 => Some(0x5D),
x if x == ControlKey::Numpad6 as i32 => Some(0x5E),
x if x == ControlKey::Numpad7 as i32 => Some(0x5F),
x if x == ControlKey::Numpad8 as i32 => Some(0x60),
x if x == ControlKey::Numpad9 as i32 => Some(0x61),
x if x == ControlKey::Insert as i32 => Some(0x49),
x if x == ControlKey::Pause as i32 => Some(0x48),
x if x == ControlKey::Scroll as i32 => Some(0x47),
x if x == ControlKey::NumLock as i32 => Some(0x53),
x if x == ControlKey::RShift as i32 => Some(0xE5),
x if x == ControlKey::RControl as i32 => Some(0xE4),
x if x == ControlKey::RAlt as i32 => Some(0xE6),
x if x == ControlKey::Multiply as i32 => Some(0x55),
x if x == ControlKey::Add as i32 => Some(0x57),
x if x == ControlKey::Subtract as i32 => Some(0x56),
x if x == ControlKey::Decimal as i32 => Some(0x63),
x if x == ControlKey::Divide as i32 => Some(0x54),
x if x == ControlKey::NumpadEnter as i32 => Some(0x58),
_ => None,
}
}
/// Convert platform keycode to USB HID usage code
/// This is a simplified mapping for X11 keycodes (Linux)
fn keycode_to_hid(keycode: u32) -> Option<u8> {
match keycode {
// Numbers 1-9 then 0 (X11 keycodes 10-19)
10 => Some(0x27), // 0
11..=19 => Some((keycode - 11 + 0x1E) as u8), // 1-9
// Punctuation before letters block
20 => Some(0x2D), // -
21 => Some(0x2E), // =
34 => Some(0x2F), // [
35 => Some(0x30), // ]
// Letters A-Z (X11 keycodes 38-63 map to various letters, not strictly A-Z)
// Note: X11 keycodes are row-based, not alphabetical
// Row 1: q(24) w(25) e(26) r(27) t(28) y(29) u(30) i(31) o(32) p(33)
// Row 2: a(38) s(39) d(40) f(41) g(42) h(43) j(44) k(45) l(46)
// Row 3: z(52) x(53) c(54) v(55) b(56) n(57) m(58)
24 => Some(0x14), // q
25 => Some(0x1A), // w
26 => Some(0x08), // e
27 => Some(0x15), // r
28 => Some(0x17), // t
29 => Some(0x1C), // y
30 => Some(0x18), // u
31 => Some(0x0C), // i
32 => Some(0x12), // o
33 => Some(0x13), // p
38 => Some(0x04), // a
39 => Some(0x16), // s
40 => Some(0x07), // d
41 => Some(0x09), // f
42 => Some(0x0A), // g
43 => Some(0x0B), // h
44 => Some(0x0D), // j
45 => Some(0x0E), // k
46 => Some(0x0F), // l
47 => Some(0x33), // ; (semicolon)
48 => Some(0x34), // ' (apostrophe)
49 => Some(0x35), // ` (grave)
51 => Some(0x31), // \ (backslash)
52 => Some(0x1D), // z
53 => Some(0x1B), // x
54 => Some(0x06), // c
55 => Some(0x19), // v
56 => Some(0x05), // b
57 => Some(0x11), // n
58 => Some(0x10), // m
59 => Some(0x36), // , (comma)
60 => Some(0x37), // . (period)
61 => Some(0x38), // / (slash)
// Space
65 => Some(0x2C),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_parse_mouse_buttons() {
let buttons = parse_mouse_buttons(mouse_mask::LEFT | mouse_mask::RIGHT);
assert!(buttons.contains(&MouseButton::Left));
assert!(buttons.contains(&MouseButton::Right));
assert!(!buttons.contains(&MouseButton::Middle));
}
#[test]
fn test_parse_scroll() {
assert_eq!(parse_scroll(mouse_mask::SCROLL_UP), 1);
assert_eq!(parse_scroll(mouse_mask::SCROLL_DOWN), -1);
assert_eq!(parse_scroll(0), 0);
}
#[test]
fn test_control_key_mapping() {
assert_eq!(control_key_to_hid(ControlKey::Escape as i32), Some(0x29));
assert_eq!(control_key_to_hid(ControlKey::Return as i32), Some(0x28));
assert_eq!(control_key_to_hid(ControlKey::Space as i32), Some(0x2C));
}
#[test]
fn test_convert_mouse_event() {
let rustdesk_event = MouseEvent {
x: 500,
y: 300,
mask: mouse_mask::LEFT,
..Default::default()
};
let events = convert_mouse_event(&rustdesk_event, 1920, 1080);
assert!(!events.is_empty());
// First event should be MoveAbs
assert_eq!(events[0].event_type, MouseEventType::MoveAbs);
// Should have a button down event
assert!(events.iter().any(|e| e.event_type == MouseEventType::Down && e.button == Some(MouseButton::Left)));
}
#[test]
fn test_convert_key_event() {
let key_event = KeyEvent {
down: true,
press: false,
union: Some(hbb::key_event::Union::ControlKey(ControlKey::Return as i32)),
..Default::default()
};
let result = convert_key_event(&key_event);
assert!(result.is_some());
let kb_event = result.unwrap();
assert_eq!(kb_event.event_type, KeyEventType::Down);
assert_eq!(kb_event.key, 0x28); // Return key USB HID code
}
}

587
src/rustdesk/mod.rs Normal file
View File

@@ -0,0 +1,587 @@
//! RustDesk Protocol Integration Module
//!
//! This module implements the RustDesk client protocol, enabling One-KVM devices
//! to be accessed via standard RustDesk clients through existing hbbs/hbbr servers.
//!
//! ## Architecture
//!
//! - `config`: Configuration types for RustDesk settings
//! - `protocol`: Protobuf message wrappers and serialization
//! - `crypto`: NaCl cryptography (key generation, encryption, signatures)
//! - `rendezvous`: Communication with hbbs rendezvous server
//! - `connection`: Client session handling
//! - `frame_adapters`: Video/audio frame conversion to RustDesk format
//! - `hid_adapter`: RustDesk HID events to One-KVM conversion
pub mod bytes_codec;
pub mod config;
pub mod connection;
pub mod crypto;
pub mod frame_adapters;
pub mod hid_adapter;
pub mod protocol;
pub mod rendezvous;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use parking_lot::RwLock;
use prost::Message;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};
use crate::audio::AudioController;
use crate::hid::HidController;
use crate::video::stream_manager::VideoStreamManager;
use self::config::RustDeskConfig;
use self::connection::ConnectionManager;
use self::protocol::hbb::rendezvous_message;
use self::protocol::{make_local_addr, make_relay_response, RendezvousMessage};
use self::rendezvous::{AddrMangle, RendezvousMediator, RendezvousStatus};
/// Relay connection timeout
const RELAY_CONNECT_TIMEOUT_MS: u64 = 10_000;
/// RustDesk service status
#[derive(Debug, Clone, PartialEq)]
pub enum ServiceStatus {
/// Service is stopped
Stopped,
/// Service is starting
Starting,
/// Service is running and registered with rendezvous server
Running,
/// Service encountered an error
Error(String),
}
impl std::fmt::Display for ServiceStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Stopped => write!(f, "stopped"),
Self::Starting => write!(f, "starting"),
Self::Running => write!(f, "running"),
Self::Error(e) => write!(f, "error: {}", e),
}
}
}
/// Default port for direct TCP connections (same as RustDesk)
const DIRECT_LISTEN_PORT: u16 = 21118;
/// RustDesk Service
///
/// Manages the RustDesk protocol integration, including:
/// - Registration with hbbs rendezvous server
/// - Accepting connections from RustDesk clients
/// - Streaming video/audio and receiving HID input
pub struct RustDeskService {
config: Arc<RwLock<RustDeskConfig>>,
status: Arc<RwLock<ServiceStatus>>,
rendezvous: Arc<RwLock<Option<Arc<RendezvousMediator>>>>,
rendezvous_handle: Arc<RwLock<Option<JoinHandle<()>>>>,
tcp_listener_handle: Arc<RwLock<Option<JoinHandle<()>>>>,
listen_port: Arc<RwLock<u16>>,
connection_manager: Arc<ConnectionManager>,
video_manager: Arc<VideoStreamManager>,
hid: Arc<HidController>,
audio: Arc<AudioController>,
shutdown_tx: broadcast::Sender<()>,
}
impl RustDeskService {
/// Create a new RustDesk service instance
pub fn new(
config: RustDeskConfig,
video_manager: Arc<VideoStreamManager>,
hid: Arc<HidController>,
audio: Arc<AudioController>,
) -> Self {
let (shutdown_tx, _) = broadcast::channel(1);
let connection_manager = Arc::new(ConnectionManager::new(config.clone()));
Self {
config: Arc::new(RwLock::new(config)),
status: Arc::new(RwLock::new(ServiceStatus::Stopped)),
rendezvous: Arc::new(RwLock::new(None)),
rendezvous_handle: Arc::new(RwLock::new(None)),
tcp_listener_handle: Arc::new(RwLock::new(None)),
listen_port: Arc::new(RwLock::new(DIRECT_LISTEN_PORT)),
connection_manager,
video_manager,
hid,
audio,
shutdown_tx,
}
}
/// Get the port for direct TCP connections
pub fn listen_port(&self) -> u16 {
*self.listen_port.read()
}
/// Get current service status
pub fn status(&self) -> ServiceStatus {
self.status.read().clone()
}
/// Get current configuration
pub fn config(&self) -> RustDeskConfig {
self.config.read().clone()
}
/// Update configuration
pub fn update_config(&self, config: RustDeskConfig) {
*self.config.write() = config;
}
/// Get rendezvous status
pub fn rendezvous_status(&self) -> Option<RendezvousStatus> {
self.rendezvous.read().as_ref().map(|r| r.status())
}
/// Get device ID
pub fn device_id(&self) -> String {
self.config.read().device_id.clone()
}
/// Get connection count
pub fn connection_count(&self) -> usize {
self.connection_manager.connection_count()
}
/// Start the RustDesk service
pub async fn start(&self) -> anyhow::Result<()> {
let config = self.config.read().clone();
if !config.enabled {
info!("RustDesk service is disabled");
return Ok(());
}
if !config.is_valid() {
warn!("RustDesk configuration is incomplete");
return Ok(());
}
if self.status() == ServiceStatus::Running {
warn!("RustDesk service is already running");
return Ok(());
}
*self.status.write() = ServiceStatus::Starting;
info!(
"Starting RustDesk service with ID: {} -> {}",
config.device_id,
config.rendezvous_addr()
);
// Initialize crypto
if let Err(e) = crypto::init() {
error!("Failed to initialize crypto: {}", e);
*self.status.write() = ServiceStatus::Error(e.to_string());
return Err(e.into());
}
// Create and start rendezvous mediator with relay callback
let mediator = Arc::new(RendezvousMediator::new(config.clone()));
// Set the keypair on connection manager (Curve25519 for encryption)
let keypair = mediator.ensure_keypair();
self.connection_manager.set_keypair(keypair);
// Set the signing keypair on connection manager (Ed25519 for SignedId)
let signing_keypair = mediator.ensure_signing_keypair();
self.connection_manager.set_signing_keypair(signing_keypair);
// Set the HID controller on connection manager
self.connection_manager.set_hid(self.hid.clone());
// Set the video manager on connection manager for video streaming
self.connection_manager.set_video_manager(self.video_manager.clone());
*self.rendezvous.write() = Some(mediator.clone());
// Start TCP listener BEFORE the rendezvous mediator to ensure port is set correctly
// This prevents race condition where mediator starts registration with wrong port
let (tcp_handle, listen_port) = self.start_tcp_listener_with_port().await?;
*self.tcp_listener_handle.write() = Some(tcp_handle);
// Set the listen port on mediator before starting the registration loop
mediator.set_listen_port(listen_port);
// Create relay request handler
let connection_manager = self.connection_manager.clone();
let video_manager = self.video_manager.clone();
let hid = self.hid.clone();
let audio = self.audio.clone();
let service_config = self.config.clone();
// Set the relay callback on the mediator
mediator.set_relay_callback(Arc::new(move |relay_server, uuid, peer_pk| {
let conn_mgr = connection_manager.clone();
let video = video_manager.clone();
let hid = hid.clone();
let audio = audio.clone();
let config = service_config.clone();
tokio::spawn(async move {
if let Err(e) = handle_relay_request(
&relay_server,
&uuid,
&peer_pk,
conn_mgr,
video,
hid,
audio,
config,
).await {
error!("Failed to handle relay request: {}", e);
}
});
}));
// Set the intranet callback on the mediator for same-LAN connections
let connection_manager2 = self.connection_manager.clone();
mediator.set_intranet_callback(Arc::new(move |rendezvous_addr, peer_socket_addr, local_addr, relay_server, device_id| {
let conn_mgr = connection_manager2.clone();
tokio::spawn(async move {
if let Err(e) = handle_intranet_request(
&rendezvous_addr,
&peer_socket_addr,
local_addr,
&relay_server,
&device_id,
conn_mgr,
).await {
error!("Failed to handle intranet request: {}", e);
}
});
}));
// Spawn rendezvous task
let status = self.status.clone();
let handle = tokio::spawn(async move {
loop {
match mediator.start().await {
Ok(_) => {
info!("Rendezvous mediator stopped normally");
break;
}
Err(e) => {
error!("Rendezvous mediator error: {}", e);
*status.write() = ServiceStatus::Error(e.to_string());
// Wait before retry
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
*status.write() = ServiceStatus::Starting;
}
}
}
});
*self.rendezvous_handle.write() = Some(handle);
*self.status.write() = ServiceStatus::Running;
Ok(())
}
/// Start TCP listener for direct peer connections
/// Returns the join handle and the port that was bound
async fn start_tcp_listener_with_port(&self) -> anyhow::Result<(JoinHandle<()>, u16)> {
// Try to bind to the default port, or find an available port
let listener = match TcpListener::bind(format!("0.0.0.0:{}", DIRECT_LISTEN_PORT)).await {
Ok(l) => l,
Err(_) => {
// Try binding to port 0 to get an available port
TcpListener::bind("0.0.0.0:0").await?
}
};
let local_addr = listener.local_addr()?;
let listen_port = local_addr.port();
*self.listen_port.write() = listen_port;
info!("RustDesk TCP listener started on {}", local_addr);
let connection_manager = self.connection_manager.clone();
let mut shutdown_rx = self.shutdown_tx.subscribe();
let handle = tokio::spawn(async move {
loop {
tokio::select! {
result = listener.accept() => {
match result {
Ok((stream, peer_addr)) => {
info!("Accepted direct connection from {}", peer_addr);
let conn_mgr = connection_manager.clone();
tokio::spawn(async move {
if let Err(e) = conn_mgr.accept_connection(stream, peer_addr).await {
error!("Failed to handle direct connection from {}: {}", peer_addr, e);
}
});
}
Err(e) => {
error!("TCP accept error: {}", e);
}
}
}
_ = shutdown_rx.recv() => {
info!("TCP listener shutting down");
break;
}
}
}
});
Ok((handle, listen_port))
}
/// Stop the RustDesk service
pub async fn stop(&self) -> anyhow::Result<()> {
if self.status() == ServiceStatus::Stopped {
return Ok(());
}
info!("Stopping RustDesk service");
// Send shutdown signal (this will stop the TCP listener)
let _ = self.shutdown_tx.send(());
// Close all connections
self.connection_manager.close_all();
// Stop rendezvous mediator
if let Some(mediator) = self.rendezvous.read().as_ref() {
mediator.stop();
}
// Wait for rendezvous task to finish
if let Some(handle) = self.rendezvous_handle.write().take() {
handle.abort();
}
// Wait for TCP listener task to finish
if let Some(handle) = self.tcp_listener_handle.write().take() {
handle.abort();
}
*self.rendezvous.write() = None;
*self.status.write() = ServiceStatus::Stopped;
Ok(())
}
/// Restart the service with new configuration
pub async fn restart(&self, config: RustDeskConfig) -> anyhow::Result<()> {
self.stop().await?;
self.update_config(config);
self.start().await
}
/// Get a shutdown receiver for graceful shutdown handling
#[allow(dead_code)]
pub fn shutdown_rx(&self) -> broadcast::Receiver<()> {
self.shutdown_tx.subscribe()
}
/// Save keypair and UUID to config
/// Returns the updated config if changes were made
pub fn save_credentials(&self) -> Option<RustDeskConfig> {
if let Some(mediator) = self.rendezvous.read().as_ref() {
let kp = mediator.ensure_keypair();
let skp = mediator.ensure_signing_keypair();
let mut config = self.config.write();
let mut changed = false;
// Save encryption keypair (Curve25519)
let pk = kp.public_key_base64();
let sk = kp.secret_key_base64();
if config.public_key.as_ref() != Some(&pk) || config.private_key.as_ref() != Some(&sk) {
config.public_key = Some(pk);
config.private_key = Some(sk);
changed = true;
}
// Save signing keypair (Ed25519)
let signing_pk = skp.public_key_base64();
let signing_sk = skp.secret_key_base64();
if config.signing_public_key.as_ref() != Some(&signing_pk) || config.signing_private_key.as_ref() != Some(&signing_sk) {
config.signing_public_key = Some(signing_pk);
config.signing_private_key = Some(signing_sk);
changed = true;
}
// Save UUID if it was newly generated
if mediator.uuid_needs_save() {
let mediator_config = mediator.config();
if let Some(uuid) = mediator_config.uuid {
if config.uuid.as_ref() != Some(&uuid) {
config.uuid = Some(uuid);
changed = true;
}
}
mediator.mark_uuid_saved();
}
if changed {
return Some(config.clone());
}
}
None
}
/// Save keypair to config (deprecated, use save_credentials instead)
#[deprecated(note = "Use save_credentials instead")]
pub fn save_keypair(&self) {
let _ = self.save_credentials();
}
}
/// Handle relay request from rendezvous server
async fn handle_relay_request(
relay_server: &str,
uuid: &str,
_peer_pk: &[u8],
connection_manager: Arc<ConnectionManager>,
_video_manager: Arc<VideoStreamManager>,
_hid: Arc<HidController>,
_audio: Arc<AudioController>,
_config: Arc<RwLock<RustDeskConfig>>,
) -> anyhow::Result<()> {
info!("Handling relay request: server={}, uuid={}", relay_server, uuid);
// Parse relay server address
let relay_addr: SocketAddr = tokio::net::lookup_host(relay_server)
.await?
.next()
.ok_or_else(|| anyhow::anyhow!("Failed to resolve relay server: {}", relay_server))?;
// Connect to relay server with timeout
let mut stream = tokio::time::timeout(
Duration::from_millis(RELAY_CONNECT_TIMEOUT_MS),
TcpStream::connect(relay_addr),
)
.await
.map_err(|_| anyhow::anyhow!("Relay connection timeout"))??;
info!("Connected to relay server at {}", relay_addr);
// Send relay response to establish the connection
let relay_response = make_relay_response(uuid, None);
let bytes = relay_response.encode_to_vec();
// Send using RustDesk's variable-length framing (NOT big-endian length prefix)
bytes_codec::write_frame(&mut stream, &bytes).await?;
debug!("Sent relay response for uuid={}", uuid);
// Read response from relay using variable-length framing
let msg_buf = bytes_codec::read_frame(&mut stream).await?;
// Parse relay response
if let Ok(msg) = RendezvousMessage::decode(&msg_buf[..]) {
match msg.union {
Some(rendezvous_message::Union::RelayResponse(rr)) => {
debug!("Received relay response: uuid={}, socket_addr_len={}", rr.uuid, rr.socket_addr.len());
// Try to decode peer address from the relay response
// The socket_addr field contains the actual peer's address (mangled)
let peer_addr = if !rr.socket_addr.is_empty() {
rendezvous::AddrMangle::decode(&rr.socket_addr)
.unwrap_or(relay_addr)
} else {
// If no socket_addr in response, use a placeholder
// Note: This is not ideal, but allows the connection to proceed
warn!("No peer socket_addr in relay response, using relay server address");
relay_addr
};
debug!("Peer address from relay: {}", peer_addr);
// At this point, the relay has connected us to the peer
// The stream is now a direct connection to the client
// Accept the connection through connection manager
connection_manager.accept_connection(stream, peer_addr).await?;
info!("Relay connection established for uuid={}, peer={}", uuid, peer_addr);
}
_ => {
warn!("Unexpected message from relay server");
}
}
}
Ok(())
}
/// Handle intranet/same-LAN connection request
///
/// When the server determines that the client and peer are on the same intranet
/// (same public IP or both on LAN), it sends FetchLocalAddr to the peer.
/// The peer must:
/// 1. Open a TCP connection to the rendezvous server
/// 2. Send LocalAddr with our local address
/// 3. Accept the peer connection over that same TCP stream
async fn handle_intranet_request(
rendezvous_addr: &str,
peer_socket_addr: &[u8],
local_addr: SocketAddr,
relay_server: &str,
device_id: &str,
connection_manager: Arc<ConnectionManager>,
) -> anyhow::Result<()> {
info!(
"Handling intranet request: rendezvous={}, local_addr={}, device_id={}",
rendezvous_addr, local_addr, device_id
);
// Decode peer address for logging
let peer_addr = AddrMangle::decode(peer_socket_addr);
debug!("Peer address from FetchLocalAddr: {:?}", peer_addr);
// Connect to rendezvous server via TCP with timeout
let mut stream = tokio::time::timeout(
Duration::from_secs(5),
TcpStream::connect(rendezvous_addr),
)
.await
.map_err(|_| anyhow::anyhow!("Timeout connecting to rendezvous server"))??;
info!("Connected to rendezvous server for intranet: {}", rendezvous_addr);
// Build LocalAddr message with our local address (mangled)
let local_addr_bytes = AddrMangle::encode(local_addr);
let msg = make_local_addr(
peer_socket_addr,
&local_addr_bytes,
relay_server,
device_id,
env!("CARGO_PKG_VERSION"),
);
let bytes = msg.encode_to_vec();
// Send LocalAddr using RustDesk's variable-length framing
bytes_codec::write_frame(&mut stream, &bytes).await?;
info!("Sent LocalAddr to rendezvous server, waiting for peer connection");
// Now the rendezvous server will forward this to the client,
// and the client will connect to us through this same TCP stream.
// The server proxies the connection between client and peer.
// Get peer address for logging/connection tracking
let effective_peer_addr = peer_addr.unwrap_or_else(|| {
// If we can't decode the peer address, use the rendezvous server address
rendezvous_addr.parse().unwrap_or_else(|_| "0.0.0.0:0".parse().unwrap())
});
// Accept the connection - the stream is now a proxied connection to the client
connection_manager.accept_connection(stream, effective_peer_addr).await?;
info!("Intranet connection established via rendezvous server proxy");
Ok(())
}

169
src/rustdesk/protocol.rs Normal file
View File

@@ -0,0 +1,169 @@
//! RustDesk Protocol Messages
//!
//! This module provides the compiled protobuf messages for the RustDesk protocol.
//! Messages are generated from rendezvous.proto and message.proto at build time.
use prost::Message;
// Include the generated protobuf code
pub mod hbb {
include!(concat!(env!("OUT_DIR"), "/hbb.rs"));
}
// Re-export commonly used types (except Message which conflicts with prost::Message)
pub use hbb::{
ConnType, ConfigUpdate, FetchLocalAddr, HealthCheck, KeyExchange, LocalAddr, NatType,
OnlineRequest, OnlineResponse, PeerDiscovery, PunchHole, PunchHoleRequest, PunchHoleResponse,
PunchHoleSent, RegisterPeer, RegisterPeerResponse, RegisterPk, RegisterPkResponse,
RelayResponse, RendezvousMessage, RequestRelay, SoftwareUpdate, TestNatRequest,
TestNatResponse,
};
// Re-export message.proto types
pub use hbb::{
AudioFormat, AudioFrame, Auth2Fa, Clipboard, CursorData, CursorPosition, EncodedVideoFrame,
EncodedVideoFrames, Hash, IdPk, KeyEvent, LoginRequest, LoginResponse, MouseEvent, Misc,
OptionMessage, PeerInfo, PublicKey, SignedId, SupportedDecoding, VideoFrame,
};
/// Trait for encoding/decoding protobuf messages
pub trait ProtobufMessage: Message + Default {
/// Encode the message to bytes
fn encode_to_vec(&self) -> Vec<u8> {
let mut buf = Vec::with_capacity(self.encoded_len());
self.encode(&mut buf).expect("Failed to encode message");
buf
}
/// Decode from bytes
fn decode_from_slice(buf: &[u8]) -> Result<Self, prost::DecodeError> {
Self::decode(buf)
}
}
// Implement for all generated message types
impl<T: Message + Default> ProtobufMessage for T {}
/// Helper to create a RendezvousMessage with RegisterPeer
pub fn make_register_peer(id: &str, serial: i32) -> RendezvousMessage {
RendezvousMessage {
union: Some(hbb::rendezvous_message::Union::RegisterPeer(RegisterPeer {
id: id.to_string(),
serial,
})),
}
}
/// Helper to create a RendezvousMessage with RegisterPk
pub fn make_register_pk(id: &str, uuid: &[u8], pk: &[u8], old_id: &str) -> RendezvousMessage {
RendezvousMessage {
union: Some(hbb::rendezvous_message::Union::RegisterPk(RegisterPk {
id: id.to_string(),
uuid: uuid.to_vec(),
pk: pk.to_vec(),
old_id: old_id.to_string(),
})),
}
}
/// Helper to create a PunchHoleSent message
pub fn make_punch_hole_sent(
socket_addr: &[u8],
id: &str,
relay_server: &str,
nat_type: NatType,
version: &str,
) -> RendezvousMessage {
RendezvousMessage {
union: Some(hbb::rendezvous_message::Union::PunchHoleSent(PunchHoleSent {
socket_addr: socket_addr.to_vec(),
id: id.to_string(),
relay_server: relay_server.to_string(),
nat_type: nat_type.into(),
version: version.to_string(),
})),
}
}
/// Helper to create a RelayResponse message (sent to relay server)
pub fn make_relay_response(uuid: &str, _pk: Option<&[u8]>) -> RendezvousMessage {
RendezvousMessage {
union: Some(hbb::rendezvous_message::Union::RelayResponse(RelayResponse {
socket_addr: vec![],
uuid: uuid.to_string(),
relay_server: String::new(),
..Default::default()
})),
}
}
/// Helper to create a LocalAddr response message
/// This is sent in response to FetchLocalAddr when a peer on the same LAN wants to connect
pub fn make_local_addr(
socket_addr: &[u8],
local_addr: &[u8],
relay_server: &str,
id: &str,
version: &str,
) -> RendezvousMessage {
RendezvousMessage {
union: Some(hbb::rendezvous_message::Union::LocalAddr(LocalAddr {
socket_addr: socket_addr.to_vec(),
local_addr: local_addr.to_vec(),
relay_server: relay_server.to_string(),
id: id.to_string(),
version: version.to_string(),
})),
}
}
/// Decode a RendezvousMessage from bytes
pub fn decode_rendezvous_message(buf: &[u8]) -> Result<RendezvousMessage, prost::DecodeError> {
RendezvousMessage::decode(buf)
}
/// Decode a Message (session message) from bytes
pub fn decode_message(buf: &[u8]) -> Result<hbb::Message, prost::DecodeError> {
hbb::Message::decode(buf)
}
#[cfg(test)]
mod tests {
use super::*;
use prost::Message as ProstMessage;
#[test]
fn test_register_peer_encoding() {
let msg = make_register_peer("123456789", 1);
let encoded = ProstMessage::encode_to_vec(&msg);
assert!(!encoded.is_empty());
let decoded = decode_rendezvous_message(&encoded).unwrap();
match decoded.union {
Some(hbb::rendezvous_message::Union::RegisterPeer(rp)) => {
assert_eq!(rp.id, "123456789");
assert_eq!(rp.serial, 1);
}
_ => panic!("Expected RegisterPeer message"),
}
}
#[test]
fn test_register_pk_encoding() {
let uuid = [1u8; 16];
let pk = [2u8; 32];
let msg = make_register_pk("123456789", &uuid, &pk, "");
let encoded = ProstMessage::encode_to_vec(&msg);
assert!(!encoded.is_empty());
let decoded = decode_rendezvous_message(&encoded).unwrap();
match decoded.union {
Some(hbb::rendezvous_message::Union::RegisterPk(rpk)) => {
assert_eq!(rpk.id, "123456789");
assert_eq!(rpk.uuid.len(), 16);
assert_eq!(rpk.pk.len(), 32);
}
_ => panic!("Expected RegisterPk message"),
}
}
}

828
src/rustdesk/rendezvous.rs Normal file
View File

@@ -0,0 +1,828 @@
//! RustDesk Rendezvous Mediator
//!
//! This module handles communication with the hbbs rendezvous server.
//! It registers the device ID and public key, handles punch hole requests,
//! and relay requests.
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::{Duration, Instant};
use parking_lot::RwLock;
use prost::Message;
use tokio::net::UdpSocket;
use tokio::sync::broadcast;
use tokio::time::interval;
use tracing::{debug, error, info, warn};
use super::config::RustDeskConfig;
use super::crypto::{KeyPair, SigningKeyPair};
use super::protocol::{
hbb::rendezvous_message, make_punch_hole_sent, make_register_peer,
make_register_pk, NatType, RendezvousMessage,
};
/// Registration interval in milliseconds
const REG_INTERVAL_MS: u64 = 12_000;
/// Minimum registration timeout
const MIN_REG_TIMEOUT_MS: u64 = 3_000;
/// Maximum registration timeout
const MAX_REG_TIMEOUT_MS: u64 = 30_000;
/// Connection timeout
#[allow(dead_code)]
const CONNECT_TIMEOUT_MS: u64 = 18_000;
/// Timer interval for checking registration status
const TIMER_INTERVAL_MS: u64 = 300;
/// Rendezvous mediator status
#[derive(Debug, Clone, PartialEq)]
pub enum RendezvousStatus {
Disconnected,
Connecting,
Connected,
Registered,
Error(String),
}
impl std::fmt::Display for RendezvousStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Disconnected => write!(f, "disconnected"),
Self::Connecting => write!(f, "connecting"),
Self::Connected => write!(f, "connected"),
Self::Registered => write!(f, "registered"),
Self::Error(e) => write!(f, "error: {}", e),
}
}
}
/// Callback for handling incoming connection requests
pub type ConnectionCallback = Arc<dyn Fn(ConnectionRequest) + Send + Sync>;
/// Incoming connection request from a RustDesk client
#[derive(Debug, Clone)]
pub struct ConnectionRequest {
/// Peer socket address (encoded)
pub socket_addr: Vec<u8>,
/// Relay server to use
pub relay_server: String,
/// NAT type
pub nat_type: NatType,
/// Connection UUID
pub uuid: String,
/// Whether to use secure connection
pub secure: bool,
}
/// Callback type for relay requests
/// Parameters: relay_server, uuid, peer_public_key
pub type RelayCallback = Arc<dyn Fn(String, String, Vec<u8>) + Send + Sync>;
/// Callback type for intranet/local address connections
/// Parameters: rendezvous_addr, peer_socket_addr (mangled), local_addr, relay_server, device_id
pub type IntranetCallback = Arc<dyn Fn(String, Vec<u8>, SocketAddr, String, String) + Send + Sync>;
/// Rendezvous Mediator
///
/// Handles communication with hbbs rendezvous server:
/// - Registers device ID and public key
/// - Maintains keep-alive with server
/// - Handles punch hole and relay requests
pub struct RendezvousMediator {
config: Arc<RwLock<RustDeskConfig>>,
keypair: Arc<RwLock<Option<KeyPair>>>,
signing_keypair: Arc<RwLock<Option<SigningKeyPair>>>,
status: Arc<RwLock<RendezvousStatus>>,
uuid: Arc<RwLock<[u8; 16]>>,
uuid_needs_save: Arc<RwLock<bool>>,
serial: Arc<RwLock<i32>>,
key_confirmed: Arc<RwLock<bool>>,
keep_alive_ms: Arc<RwLock<i32>>,
relay_callback: Arc<RwLock<Option<RelayCallback>>>,
intranet_callback: Arc<RwLock<Option<IntranetCallback>>>,
listen_port: Arc<RwLock<u16>>,
shutdown_tx: broadcast::Sender<()>,
}
impl RendezvousMediator {
/// Create a new rendezvous mediator
pub fn new(mut config: RustDeskConfig) -> Self {
let (shutdown_tx, _) = broadcast::channel(1);
// Get or generate UUID from config (persisted)
let (uuid, uuid_needs_save) = config.ensure_uuid();
Self {
config: Arc::new(RwLock::new(config)),
keypair: Arc::new(RwLock::new(None)),
signing_keypair: Arc::new(RwLock::new(None)),
status: Arc::new(RwLock::new(RendezvousStatus::Disconnected)),
uuid: Arc::new(RwLock::new(uuid)),
uuid_needs_save: Arc::new(RwLock::new(uuid_needs_save)),
serial: Arc::new(RwLock::new(0)),
key_confirmed: Arc::new(RwLock::new(false)),
keep_alive_ms: Arc::new(RwLock::new(30_000)),
relay_callback: Arc::new(RwLock::new(None)),
intranet_callback: Arc::new(RwLock::new(None)),
listen_port: Arc::new(RwLock::new(21118)),
shutdown_tx,
}
}
/// Set the TCP listen port for direct connections
pub fn set_listen_port(&self, port: u16) {
let old_port = *self.listen_port.read();
if old_port != port {
*self.listen_port.write() = port;
// Port changed, increment serial to notify server
self.increment_serial();
}
}
/// Get the TCP listen port
pub fn listen_port(&self) -> u16 {
*self.listen_port.read()
}
/// Increment the serial number to indicate local state change
pub fn increment_serial(&self) {
let mut serial = self.serial.write();
*serial = serial.wrapping_add(1);
debug!("Serial incremented to {}", *serial);
}
/// Get current serial number
pub fn serial(&self) -> i32 {
*self.serial.read()
}
/// Check if UUID needs to be saved to persistent storage
pub fn uuid_needs_save(&self) -> bool {
*self.uuid_needs_save.read()
}
/// Get the current config (with UUID set)
pub fn config(&self) -> RustDeskConfig {
self.config.read().clone()
}
/// Mark UUID as saved
pub fn mark_uuid_saved(&self) {
*self.uuid_needs_save.write() = false;
}
/// Set the callback for relay requests
pub fn set_relay_callback(&self, callback: RelayCallback) {
*self.relay_callback.write() = Some(callback);
}
/// Set the callback for intranet/local address connections
pub fn set_intranet_callback(&self, callback: IntranetCallback) {
*self.intranet_callback.write() = Some(callback);
}
/// Get current status
pub fn status(&self) -> RendezvousStatus {
self.status.read().clone()
}
/// Update configuration
pub fn update_config(&self, config: RustDeskConfig) {
*self.config.write() = config;
// Config changed, increment serial to notify server
self.increment_serial();
}
/// Initialize or get keypair (Curve25519 for encryption)
pub fn ensure_keypair(&self) -> KeyPair {
let mut keypair_guard = self.keypair.write();
if keypair_guard.is_none() {
let config = self.config.read();
// Try to load from config first
if let (Some(pk), Some(sk)) = (&config.public_key, &config.private_key) {
if let Ok(kp) = KeyPair::from_base64(pk, sk) {
*keypair_guard = Some(kp.clone());
return kp;
}
}
// Generate new keypair
let kp = KeyPair::generate();
*keypair_guard = Some(kp.clone());
kp
} else {
keypair_guard.as_ref().unwrap().clone()
}
}
/// Initialize or get signing keypair (Ed25519 for SignedId)
pub fn ensure_signing_keypair(&self) -> SigningKeyPair {
let mut signing_guard = self.signing_keypair.write();
if signing_guard.is_none() {
let config = self.config.read();
// Try to load from config first
if let (Some(pk), Some(sk)) = (&config.signing_public_key, &config.signing_private_key) {
if let Ok(skp) = SigningKeyPair::from_base64(pk, sk) {
*signing_guard = Some(skp.clone());
return skp;
}
}
// Generate new signing keypair
let skp = SigningKeyPair::generate();
*signing_guard = Some(skp.clone());
skp
} else {
signing_guard.as_ref().unwrap().clone()
}
}
/// Get the device ID
pub fn device_id(&self) -> String {
self.config.read().device_id.clone()
}
/// Start the rendezvous mediator
pub async fn start(&self) -> anyhow::Result<()> {
let config = self.config.read().clone();
if !config.enabled || config.rendezvous_server.is_empty() {
return Ok(());
}
*self.status.write() = RendezvousStatus::Connecting;
let addr = config.rendezvous_addr();
info!("Starting rendezvous mediator for {} to {}", config.device_id, addr);
// Resolve server address
let server_addr: SocketAddr = tokio::net::lookup_host(&addr)
.await?
.next()
.ok_or_else(|| anyhow::anyhow!("Failed to resolve {}", addr))?;
// Create UDP socket
let socket = UdpSocket::bind("0.0.0.0:0").await?;
socket.connect(server_addr).await?;
info!("Connected to rendezvous server at {}", server_addr);
*self.status.write() = RendezvousStatus::Connected;
// Start registration loop
self.registration_loop(socket).await
}
/// Main registration loop
async fn registration_loop(&self, socket: UdpSocket) -> anyhow::Result<()> {
let mut timer = interval(Duration::from_millis(TIMER_INTERVAL_MS));
let mut recv_buf = vec![0u8; 65535];
let mut last_register_sent: Option<Instant> = None;
let mut last_register_resp: Option<Instant> = None;
let mut reg_timeout = MIN_REG_TIMEOUT_MS;
let mut fails = 0;
let mut shutdown_rx = self.shutdown_tx.subscribe();
loop {
tokio::select! {
// Handle incoming messages
result = socket.recv(&mut recv_buf) => {
match result {
Ok(len) => {
if let Ok(msg) = RendezvousMessage::decode(&recv_buf[..len]) {
self.handle_response(&socket, msg, &mut last_register_resp, &mut fails, &mut reg_timeout).await?;
} else {
debug!("Failed to decode rendezvous message");
}
}
Err(e) => {
error!("Failed to receive from socket: {}", e);
*self.status.write() = RendezvousStatus::Error(e.to_string());
break;
}
}
}
// Periodic registration
_ = timer.tick() => {
let now = Instant::now();
let expired = last_register_resp
.map(|x| x.elapsed().as_millis() as u64 >= REG_INTERVAL_MS)
.unwrap_or(true);
let timeout = last_register_sent
.map(|x| x.elapsed().as_millis() as u64 >= reg_timeout)
.unwrap_or(false);
if timeout && reg_timeout < MAX_REG_TIMEOUT_MS {
reg_timeout += MIN_REG_TIMEOUT_MS;
fails += 1;
if fails >= 4 {
warn!("Registration timeout, {} consecutive failures", fails);
}
}
if timeout || (last_register_sent.is_none() && expired) {
self.send_register(&socket).await?;
last_register_sent = Some(now);
}
}
// Shutdown signal
_ = shutdown_rx.recv() => {
info!("Rendezvous mediator shutting down");
break;
}
}
}
*self.status.write() = RendezvousStatus::Disconnected;
Ok(())
}
/// Send registration message
async fn send_register(&self, socket: &UdpSocket) -> anyhow::Result<()> {
let key_confirmed = *self.key_confirmed.read();
if !key_confirmed {
// Send RegisterPk with public key
self.send_register_pk(socket).await
} else {
// Send RegisterPeer heartbeat
self.send_register_peer(socket).await
}
}
/// Send RegisterPeer message
async fn send_register_peer(&self, socket: &UdpSocket) -> anyhow::Result<()> {
let id = self.device_id();
let serial = *self.serial.read();
debug!("Sending RegisterPeer: id={}, serial={}", id, serial);
let msg = make_register_peer(&id, serial);
let bytes = msg.encode_to_vec();
socket.send(&bytes).await?;
Ok(())
}
/// Send RegisterPk message
/// Uses the Ed25519 signing public key for registration
async fn send_register_pk(&self, socket: &UdpSocket) -> anyhow::Result<()> {
let id = self.device_id();
// Use signing public key (Ed25519) for RegisterPk
// This is what clients will use to verify our SignedId signature
let signing_keypair = self.ensure_signing_keypair();
let pk = signing_keypair.public_key_bytes();
let uuid = *self.uuid.read();
debug!("Sending RegisterPk: id={}, signing_pk_len={}", id, pk.len());
let msg = make_register_pk(&id, &uuid, pk, "");
let bytes = msg.encode_to_vec();
socket.send(&bytes).await?;
Ok(())
}
/// Handle FetchLocalAddr - send to callback for proper TCP handling
///
/// The intranet callback will:
/// 1. Open a TCP connection to the rendezvous server
/// 2. Send LocalAddr message
/// 3. Accept the peer connection over that same TCP stream
async fn send_local_addr(
&self,
_udp_socket: &UdpSocket,
peer_socket_addr: &[u8],
relay_server: &str,
) -> anyhow::Result<()> {
let id = self.device_id();
// Get our actual local IP addresses for same-LAN connection
let local_addrs = get_local_addresses();
if local_addrs.is_empty() {
debug!("No local addresses available for LocalAddr response");
return Ok(());
}
// Get the rendezvous server address for TCP connection
let config = self.config.read().clone();
let rendezvous_addr = config.rendezvous_addr();
// Use TCP listen port for direct connections
let listen_port = self.listen_port();
// Use the first local IP
let local_ip = local_addrs[0];
let local_sock_addr = SocketAddr::new(local_ip, listen_port);
info!(
"FetchLocalAddr: calling intranet callback with local_addr={}, rendezvous={}",
local_sock_addr, rendezvous_addr
);
// Call the intranet callback if set
if let Some(callback) = self.intranet_callback.read().as_ref() {
callback(
rendezvous_addr,
peer_socket_addr.to_vec(),
local_sock_addr,
relay_server.to_string(),
id,
);
} else {
warn!("No intranet callback set, cannot handle FetchLocalAddr properly");
}
Ok(())
}
/// Handle response from rendezvous server
async fn handle_response(
&self,
socket: &UdpSocket,
msg: RendezvousMessage,
last_resp: &mut Option<Instant>,
fails: &mut i32,
reg_timeout: &mut u64,
) -> anyhow::Result<()> {
*last_resp = Some(Instant::now());
*fails = 0;
*reg_timeout = MIN_REG_TIMEOUT_MS;
match msg.union {
Some(rendezvous_message::Union::RegisterPeerResponse(rpr)) => {
debug!("Received RegisterPeerResponse, request_pk={}", rpr.request_pk);
if rpr.request_pk {
// Server wants us to register our public key
info!("Server requested public key registration");
*self.key_confirmed.write() = false;
self.send_register_pk(socket).await?;
}
*self.status.write() = RendezvousStatus::Registered;
}
Some(rendezvous_message::Union::RegisterPkResponse(rpr)) => {
debug!("Received RegisterPkResponse: result={}", rpr.result);
match rpr.result {
0 => {
// OK
info!("Public key registered successfully");
*self.key_confirmed.write() = true;
// Increment serial after successful registration
self.increment_serial();
*self.status.write() = RendezvousStatus::Registered;
}
2 => {
// UUID_MISMATCH
warn!("UUID mismatch, need to re-register");
*self.key_confirmed.write() = false;
}
3 => {
// ID_EXISTS
error!("Device ID already exists on server");
*self.status.write() =
RendezvousStatus::Error("Device ID already exists".to_string());
}
4 => {
// TOO_FREQUENT
warn!("Registration too frequent");
}
5 => {
// INVALID_ID_FORMAT
error!("Invalid device ID format");
*self.status.write() =
RendezvousStatus::Error("Invalid ID format".to_string());
}
_ => {
error!("Unknown RegisterPkResponse result: {}", rpr.result);
}
}
if rpr.keep_alive > 0 {
*self.keep_alive_ms.write() = rpr.keep_alive * 1000;
debug!("Keep alive set to {}ms", rpr.keep_alive * 1000);
}
}
Some(rendezvous_message::Union::PunchHole(ph)) => {
// Decode the peer's socket address
let peer_addr = if !ph.socket_addr.is_empty() {
AddrMangle::decode(&ph.socket_addr)
} else {
None
};
info!(
"Received PunchHole request: peer_addr={:?}, socket_addr_len={}, relay_server={}, nat_type={:?}",
peer_addr, ph.socket_addr.len(), ph.relay_server, ph.nat_type
);
// Send PunchHoleSent to acknowledge and provide our address
// Use the TCP listen port address, not the UDP socket's address
let listen_port = self.listen_port();
// Get our public-facing address from the UDP socket
if let Ok(local_addr) = socket.local_addr() {
// Use the same IP as UDP socket but with TCP listen port
let tcp_addr = SocketAddr::new(local_addr.ip(), listen_port);
let our_socket_addr = AddrMangle::encode(tcp_addr);
let id = self.device_id();
info!(
"Sending PunchHoleSent: id={}, socket_addr={}, relay_server={}",
id, tcp_addr, ph.relay_server
);
let msg = make_punch_hole_sent(
&our_socket_addr,
&id,
&ph.relay_server,
NatType::try_from(ph.nat_type).unwrap_or(NatType::UnknownNat),
env!("CARGO_PKG_VERSION"),
);
let bytes = msg.encode_to_vec();
if let Err(e) = socket.send(&bytes).await {
warn!("Failed to send PunchHoleSent: {}", e);
} else {
info!("Sent PunchHoleSent response successfully");
}
}
// For now, we fall back to relay since true UDP hole punching is complex
// and may not work through all NAT types
if !ph.relay_server.is_empty() {
if let Some(callback) = self.relay_callback.read().as_ref() {
let relay_server = if ph.relay_server.contains(':') {
ph.relay_server.clone()
} else {
format!("{}:21117", ph.relay_server)
};
// Use peer's socket_addr to generate a deterministic UUID
// This ensures both sides use the same UUID for the relay
let uuid = if !ph.socket_addr.is_empty() {
use std::hash::{Hash, Hasher};
let mut hasher = std::collections::hash_map::DefaultHasher::new();
ph.socket_addr.hash(&mut hasher);
format!("{:016x}", hasher.finish())
} else {
uuid::Uuid::new_v4().to_string()
};
callback(relay_server, uuid, vec![]);
}
}
}
Some(rendezvous_message::Union::RequestRelay(rr)) => {
info!(
"Received RequestRelay, relay_server={}, uuid={}",
rr.relay_server, rr.uuid
);
// Call the relay callback to handle the connection
if let Some(callback) = self.relay_callback.read().as_ref() {
let relay_server = if rr.relay_server.contains(':') {
rr.relay_server.clone()
} else {
format!("{}:21117", rr.relay_server)
};
callback(relay_server, rr.uuid.clone(), vec![]);
}
}
Some(rendezvous_message::Union::FetchLocalAddr(fla)) => {
// Decode the peer address for logging
let peer_addr = AddrMangle::decode(&fla.socket_addr);
info!(
"Received FetchLocalAddr request: peer_addr={:?}, socket_addr_len={}, relay_server={}",
peer_addr, fla.socket_addr.len(), fla.relay_server
);
// Respond with our local address for same-LAN direct connection
self.send_local_addr(socket, &fla.socket_addr, &fla.relay_server).await?;
}
Some(rendezvous_message::Union::ConfigureUpdate(cu)) => {
info!("Received ConfigureUpdate, serial={}", cu.serial);
*self.serial.write() = cu.serial;
}
Some(other) => {
// Log the actual message type for debugging
let type_name = match other {
rendezvous_message::Union::PunchHoleRequest(_) => "PunchHoleRequest",
rendezvous_message::Union::PunchHoleResponse(_) => "PunchHoleResponse",
rendezvous_message::Union::SoftwareUpdate(_) => "SoftwareUpdate",
rendezvous_message::Union::TestNatRequest(_) => "TestNatRequest",
rendezvous_message::Union::TestNatResponse(_) => "TestNatResponse",
rendezvous_message::Union::PeerDiscovery(_) => "PeerDiscovery",
rendezvous_message::Union::OnlineRequest(_) => "OnlineRequest",
rendezvous_message::Union::OnlineResponse(_) => "OnlineResponse",
rendezvous_message::Union::KeyExchange(_) => "KeyExchange",
rendezvous_message::Union::Hc(_) => "HealthCheck",
rendezvous_message::Union::RelayResponse(_) => "RelayResponse",
_ => "Other",
};
info!("Received unhandled rendezvous message type: {}", type_name);
}
None => {
debug!("Received empty rendezvous message");
}
}
Ok(())
}
/// Stop the rendezvous mediator
pub fn stop(&self) {
info!("Stopping rendezvous mediator");
let _ = self.shutdown_tx.send(());
*self.status.write() = RendezvousStatus::Disconnected;
}
/// Get a shutdown receiver
pub fn shutdown_rx(&self) -> broadcast::Receiver<()> {
self.shutdown_tx.subscribe()
}
}
/// AddrMangle - RustDesk's address encoding scheme
///
/// Certain routers and firewalls scan packets and modify IP addresses.
/// This encoding mangles the address to avoid detection.
pub struct AddrMangle;
impl AddrMangle {
/// Encode a SocketAddr to bytes using RustDesk's mangle algorithm
pub fn encode(addr: SocketAddr) -> Vec<u8> {
// Try to convert IPv6-mapped IPv4 to plain IPv4
let addr = try_into_v4(addr);
match addr {
SocketAddr::V4(addr_v4) => {
use std::time::{SystemTime, UNIX_EPOCH};
let tm = (SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or(std::time::Duration::ZERO)
.as_micros() as u32) as u128;
let ip = u32::from_le_bytes(addr_v4.ip().octets()) as u128;
let port = addr.port() as u128;
let v = ((ip + tm) << 49) | (tm << 17) | (port + (tm & 0xFFFF));
let bytes = v.to_le_bytes();
// Remove trailing zeros
let mut n_padding = 0;
for i in bytes.iter().rev() {
if *i == 0u8 {
n_padding += 1;
} else {
break;
}
}
bytes[..(16 - n_padding)].to_vec()
}
SocketAddr::V6(addr_v6) => {
let mut x = addr_v6.ip().octets().to_vec();
let port: [u8; 2] = addr_v6.port().to_le_bytes();
x.push(port[0]);
x.push(port[1]);
x
}
}
}
/// Decode bytes to SocketAddr using RustDesk's mangle algorithm
#[allow(dead_code)]
pub fn decode(bytes: &[u8]) -> Option<SocketAddr> {
use std::convert::TryInto;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4};
if bytes.len() > 16 {
// IPv6 format: 16 bytes IP + 2 bytes port
if bytes.len() != 18 {
return None;
}
let tmp: [u8; 2] = bytes[16..].try_into().ok()?;
let port = u16::from_le_bytes(tmp);
let tmp: [u8; 16] = bytes[..16].try_into().ok()?;
let ip = Ipv6Addr::from(tmp);
return Some(SocketAddr::new(std::net::IpAddr::V6(ip), port));
}
// IPv4 mangled format
let mut padded = [0u8; 16];
padded[..bytes.len()].copy_from_slice(bytes);
let number = u128::from_le_bytes(padded);
let tm = (number >> 17) & (u32::MAX as u128);
let ip = (((number >> 49).wrapping_sub(tm)) as u32).to_le_bytes();
let port = ((number & 0xFFFFFF).wrapping_sub(tm & 0xFFFF)) as u16;
Some(SocketAddr::V4(SocketAddrV4::new(
Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]),
port,
)))
}
}
/// Try to convert IPv6-mapped IPv4 address to plain IPv4
fn try_into_v4(addr: SocketAddr) -> SocketAddr {
match addr {
SocketAddr::V6(v6) if !addr.ip().is_loopback() => {
if let Some(ipv4) = v6.ip().to_ipv4_mapped() {
return SocketAddr::new(std::net::IpAddr::V4(ipv4), v6.port());
}
}
_ => {}
}
addr
}
/// Check if an interface name belongs to Docker or other virtual networks
fn is_virtual_interface(name: &str) -> bool {
// Docker interfaces
name.starts_with("docker")
|| name.starts_with("br-")
|| name.starts_with("veth")
// Kubernetes/container interfaces
|| name.starts_with("cni")
|| name.starts_with("flannel")
|| name.starts_with("calico")
|| name.starts_with("weave")
// Virtual bridge interfaces
|| name.starts_with("virbr")
|| name.starts_with("lxcbr")
|| name.starts_with("lxdbr")
// VPN interfaces (usually not useful for LAN discovery)
|| name.starts_with("tun")
|| name.starts_with("tap")
}
/// Check if an IP address is in a Docker/container private range
fn is_docker_ip(ip: &std::net::IpAddr) -> bool {
if let std::net::IpAddr::V4(ipv4) = ip {
let octets = ipv4.octets();
// Docker default bridge: 172.17.0.0/16
if octets[0] == 172 && octets[1] == 17 {
return true;
}
// Docker user-defined networks: 172.18-31.0.0/16
if octets[0] == 172 && (18..=31).contains(&octets[1]) {
return true;
}
// Docker overlay networks: 10.0.0.0/8 (common range)
// Note: 10.x.x.x is also used for corporate LANs, so we only filter
// specific Docker-like patterns (10.0.x.x with small third octet)
if octets[0] == 10 && octets[1] == 0 && octets[2] < 10 {
return true;
}
}
false
}
/// Get local IP addresses (non-loopback, non-Docker)
fn get_local_addresses() -> Vec<std::net::IpAddr> {
let mut addrs = Vec::new();
// Use pnet or network-interface crate if available, otherwise use simple method
#[cfg(target_os = "linux")]
{
if let Ok(interfaces) = std::fs::read_dir("/sys/class/net") {
for entry in interfaces.flatten() {
let iface_name = entry.file_name().to_string_lossy().to_string();
// Skip loopback and virtual interfaces
if iface_name == "lo" || is_virtual_interface(&iface_name) {
continue;
}
// Try to get IP via command (simple approach)
if let Ok(output) = std::process::Command::new("ip")
.args(["-4", "addr", "show", &iface_name])
.output()
{
let stdout = String::from_utf8_lossy(&output.stdout);
for line in stdout.lines() {
if let Some(inet_pos) = line.find("inet ") {
let ip_part = &line[inet_pos + 5..];
if let Some(slash_pos) = ip_part.find('/') {
if let Ok(ip) = ip_part[..slash_pos].parse::<std::net::IpAddr>() {
// Skip loopback and Docker IPs
if !ip.is_loopback() && !is_docker_ip(&ip) {
addrs.push(ip);
}
}
}
}
}
}
}
}
}
// Fallback: try to get default route interface IP
if addrs.is_empty() {
// Try using DNS lookup to get local IP (connects to external server)
if let Ok(socket) = std::net::UdpSocket::bind("0.0.0.0:0") {
// Connect to a public DNS server (doesn't actually send data)
if socket.connect("8.8.8.8:53").is_ok() {
if let Ok(local_addr) = socket.local_addr() {
let ip = local_addr.ip();
// Skip loopback and Docker IPs
if !ip.is_loopback() && !is_docker_ip(&ip) {
addrs.push(ip);
}
}
}
}
}
addrs
}

View File

@@ -10,6 +10,7 @@ use crate::extensions::ExtensionManager;
use crate::hid::HidController;
use crate::msd::MsdController;
use crate::otg::OtgService;
use crate::rustdesk::RustDeskService;
use crate::video::VideoStreamManager;
/// Application-wide state shared across handlers
@@ -44,6 +45,8 @@ pub struct AppState {
pub atx: Arc<RwLock<Option<AtxController>>>,
/// Audio controller
pub audio: Arc<AudioController>,
/// RustDesk remote access service (optional)
pub rustdesk: Arc<RwLock<Option<Arc<RustDeskService>>>>,
/// Extension manager (ttyd, gostc, easytier)
pub extensions: Arc<ExtensionManager>,
/// Event bus for real-time notifications
@@ -66,6 +69,7 @@ impl AppState {
msd: Option<MsdController>,
atx: Option<AtxController>,
audio: Arc<AudioController>,
rustdesk: Option<Arc<RustDeskService>>,
extensions: Arc<ExtensionManager>,
events: Arc<EventBus>,
shutdown_tx: broadcast::Sender<()>,
@@ -81,6 +85,7 @@ impl AppState {
msd: Arc::new(RwLock::new(msd)),
atx: Arc::new(RwLock::new(atx)),
audio,
rustdesk: Arc::new(RwLock::new(rustdesk)),
extensions,
events,
shutdown_tx,

View File

@@ -541,6 +541,78 @@ impl VideoStreamManager {
self.streamer.frame_sender().await
}
/// Subscribe to encoded video frames from the shared video pipeline
///
/// This allows RustDesk (and other consumers) to receive H264/H265/VP8/VP9
/// encoded frames without running a separate encoder. The encoding is shared
/// with WebRTC sessions.
///
/// This method ensures video capture is running before subscribing.
/// Returns None if video capture cannot be started or pipeline creation fails.
pub async fn subscribe_encoded_frames(
&self,
) -> Option<tokio::sync::broadcast::Receiver<crate::video::shared_video_pipeline::EncodedVideoFrame>> {
// 1. Ensure video capture is initialized
if self.streamer.state().await == StreamerState::Uninitialized {
tracing::info!("Initializing video capture for encoded frame subscription");
if let Err(e) = self.streamer.init_auto().await {
tracing::error!("Failed to initialize video capture for encoded frames: {}", e);
return None;
}
}
// 2. Ensure video capture is running (streaming)
if self.streamer.state().await != StreamerState::Streaming {
tracing::info!("Starting video capture for encoded frame subscription");
if let Err(e) = self.streamer.start().await {
tracing::error!("Failed to start video capture for encoded frames: {}", e);
return None;
}
// Wait for capture to stabilize
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
}
// 3. Get frame sender from running capture
let frame_tx = match self.streamer.frame_sender().await {
Some(tx) => tx,
None => {
tracing::warn!("Cannot subscribe to encoded frames: no frame sender available");
return None;
}
};
// 4. Synchronize WebRTC config with actual capture format
let (format, resolution, fps) = self.streamer.current_video_config().await;
tracing::info!(
"Connecting encoded frame subscription: {}x{} {:?} @ {}fps",
resolution.width, resolution.height, format, fps
);
self.webrtc_streamer.update_video_config(resolution, format, fps).await;
// 5. Use WebRtcStreamer to ensure the shared video pipeline is running
// This will create the pipeline if needed
match self.webrtc_streamer.ensure_video_pipeline_for_external(frame_tx).await {
Ok(pipeline) => Some(pipeline.subscribe()),
Err(e) => {
tracing::error!("Failed to start shared video pipeline: {}", e);
None
}
}
}
/// Get the current video encoding configuration from the shared pipeline
pub async fn get_encoding_config(&self) -> Option<crate::video::shared_video_pipeline::SharedVideoPipelineConfig> {
self.webrtc_streamer.get_pipeline_config().await
}
/// Set video codec for the shared video pipeline
///
/// This allows external consumers (like RustDesk) to set the video codec
/// before subscribing to encoded frames.
pub async fn set_video_codec(&self, codec: crate::video::encoder::VideoCodecType) -> crate::error::Result<()> {
self.webrtc_streamer.set_video_codec(codec).await
}
/// Publish event to event bus
async fn publish_event(&self, event: SystemEvent) {
if let Some(ref events) = *self.events.read().await {

View File

@@ -360,3 +360,62 @@ pub async fn apply_audio_config(
Ok(())
}
/// 应用 RustDesk 配置变更
pub async fn apply_rustdesk_config(
state: &Arc<AppState>,
old_config: &crate::rustdesk::config::RustDeskConfig,
new_config: &crate::rustdesk::config::RustDeskConfig,
) -> Result<()> {
tracing::info!("Applying RustDesk config changes...");
let mut rustdesk_guard = state.rustdesk.write().await;
// Check if service needs to be stopped
if old_config.enabled && !new_config.enabled {
// Disable service
if let Some(ref service) = *rustdesk_guard {
if let Err(e) = service.stop().await {
tracing::error!("Failed to stop RustDesk service: {}", e);
}
tracing::info!("RustDesk service stopped");
}
*rustdesk_guard = None;
return Ok(());
}
// Check if service needs to be started or restarted
if new_config.enabled {
let need_restart = old_config.rendezvous_server != new_config.rendezvous_server
|| old_config.device_id != new_config.device_id
|| old_config.device_password != new_config.device_password;
if rustdesk_guard.is_none() {
// Create new service
tracing::info!("Initializing RustDesk service...");
let service = crate::rustdesk::RustDeskService::new(
new_config.clone(),
state.stream_manager.clone(),
state.hid.clone(),
state.audio.clone(),
);
if let Err(e) = service.start().await {
tracing::error!("Failed to start RustDesk service: {}", e);
} else {
tracing::info!("RustDesk service started with ID: {}", new_config.device_id);
}
*rustdesk_guard = Some(std::sync::Arc::new(service));
} else if need_restart {
// Restart existing service with new config
if let Some(ref service) = *rustdesk_guard {
if let Err(e) = service.restart(new_config.clone()).await {
tracing::error!("Failed to restart RustDesk service: {}", e);
} else {
tracing::info!("RustDesk service restarted with ID: {}", new_config.device_id);
}
}
}
}
Ok(())
}

View File

@@ -13,6 +13,8 @@
//! - PATCH /api/config/atx - 更新 ATX 配置
//! - GET /api/config/audio - 获取音频配置
//! - PATCH /api/config/audio - 更新音频配置
//! - GET /api/config/rustdesk - 获取 RustDesk 配置
//! - PATCH /api/config/rustdesk - 更新 RustDesk 配置
mod apply;
mod types;
@@ -23,6 +25,7 @@ mod hid;
mod msd;
mod atx;
mod audio;
mod rustdesk;
// 导出 handler 函数
pub use video::{get_video_config, update_video_config};
@@ -31,6 +34,10 @@ pub use hid::{get_hid_config, update_hid_config};
pub use msd::{get_msd_config, update_msd_config};
pub use atx::{get_atx_config, update_atx_config};
pub use audio::{get_audio_config, update_audio_config};
pub use rustdesk::{
get_rustdesk_config, get_rustdesk_status, update_rustdesk_config,
regenerate_device_id, regenerate_device_password, get_device_password,
};
// 保留全局配置查询(向后兼容)
use axum::{extract::State, Json};

View File

@@ -0,0 +1,142 @@
//! RustDesk 配置 Handler
use axum::{extract::State, Json};
use std::sync::Arc;
use crate::error::Result;
use crate::rustdesk::config::RustDeskConfig;
use crate::state::AppState;
use super::apply::apply_rustdesk_config;
use super::types::RustDeskConfigUpdate;
/// RustDesk 配置响应(隐藏敏感信息)
#[derive(Debug, serde::Serialize)]
pub struct RustDeskConfigResponse {
pub enabled: bool,
pub rendezvous_server: String,
pub relay_server: Option<String>,
pub device_id: String,
/// 是否已设置密码
pub has_password: bool,
/// 是否已设置密钥对
pub has_keypair: bool,
}
impl From<&RustDeskConfig> for RustDeskConfigResponse {
fn from(config: &RustDeskConfig) -> Self {
Self {
enabled: config.enabled,
rendezvous_server: config.rendezvous_server.clone(),
relay_server: config.relay_server.clone(),
device_id: config.device_id.clone(),
has_password: !config.device_password.is_empty(),
has_keypair: config.public_key.is_some() && config.private_key.is_some(),
}
}
}
/// RustDesk 状态响应
#[derive(Debug, serde::Serialize)]
pub struct RustDeskStatusResponse {
pub config: RustDeskConfigResponse,
pub service_status: String,
pub rendezvous_status: Option<String>,
}
/// 获取 RustDesk 配置
pub async fn get_rustdesk_config(State(state): State<Arc<AppState>>) -> Json<RustDeskConfigResponse> {
Json(RustDeskConfigResponse::from(&state.config.get().rustdesk))
}
/// 获取 RustDesk 完整状态(配置 + 服务状态)
pub async fn get_rustdesk_status(State(state): State<Arc<AppState>>) -> Json<RustDeskStatusResponse> {
let config = state.config.get().rustdesk.clone();
// 获取服务状态
let (service_status, rendezvous_status) = {
let guard = state.rustdesk.read().await;
if let Some(ref service) = *guard {
let status = format!("{}", service.status());
let rv_status = service.rendezvous_status().map(|s| format!("{}", s));
(status, rv_status)
} else {
("not_initialized".to_string(), None)
}
};
Json(RustDeskStatusResponse {
config: RustDeskConfigResponse::from(&config),
service_status,
rendezvous_status,
})
}
/// 更新 RustDesk 配置
pub async fn update_rustdesk_config(
State(state): State<Arc<AppState>>,
Json(req): Json<RustDeskConfigUpdate>,
) -> Result<Json<RustDeskConfigResponse>> {
// 1. 验证请求
req.validate()?;
// 2. 获取旧配置
let old_config = state.config.get().rustdesk.clone();
// 3. 应用更新到配置存储
state
.config
.update(|config| {
req.apply_to(&mut config.rustdesk);
})
.await?;
// 4. 获取新配置
let new_config = state.config.get().rustdesk.clone();
// 5. 应用到子系统(热重载)
if let Err(e) = apply_rustdesk_config(&state, &old_config, &new_config).await {
tracing::error!("Failed to apply RustDesk config: {}", e);
}
Ok(Json(RustDeskConfigResponse::from(&new_config)))
}
/// 重新生成设备 ID
pub async fn regenerate_device_id(
State(state): State<Arc<AppState>>,
) -> Result<Json<RustDeskConfigResponse>> {
state
.config
.update(|config| {
config.rustdesk.device_id = RustDeskConfig::generate_device_id();
})
.await?;
let new_config = state.config.get().rustdesk.clone();
Ok(Json(RustDeskConfigResponse::from(&new_config)))
}
/// 重新生成设备密码
pub async fn regenerate_device_password(
State(state): State<Arc<AppState>>,
) -> Result<Json<RustDeskConfigResponse>> {
state
.config
.update(|config| {
config.rustdesk.device_password = RustDeskConfig::generate_password();
})
.await?;
let new_config = state.config.get().rustdesk.clone();
Ok(Json(RustDeskConfigResponse::from(&new_config)))
}
/// 获取设备密码(管理员专用)
pub async fn get_device_password(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
let config = state.config.get().rustdesk.clone();
Json(serde_json::json!({
"device_id": config.device_id,
"device_password": config.device_password
}))
}

View File

@@ -2,6 +2,7 @@ use serde::Deserialize;
use typeshare::typeshare;
use crate::config::*;
use crate::error::AppError;
use crate::rustdesk::config::RustDeskConfig;
// ===== Video Config =====
#[typeshare]
@@ -394,3 +395,60 @@ impl AudioConfigUpdate {
}
}
}
// ===== RustDesk Config =====
#[typeshare]
#[derive(Debug, Deserialize)]
pub struct RustDeskConfigUpdate {
pub enabled: Option<bool>,
pub rendezvous_server: Option<String>,
pub relay_server: Option<String>,
pub device_password: Option<String>,
}
impl RustDeskConfigUpdate {
pub fn validate(&self) -> crate::error::Result<()> {
// Validate rendezvous server format (should be host:port)
if let Some(ref server) = self.rendezvous_server {
if !server.is_empty() && !server.contains(':') {
return Err(AppError::BadRequest(
"Rendezvous server must be in format 'host:port' (e.g., rs.example.com:21116)".into(),
));
}
}
// Validate relay server format if provided
if let Some(ref server) = self.relay_server {
if !server.is_empty() && !server.contains(':') {
return Err(AppError::BadRequest(
"Relay server must be in format 'host:port' (e.g., rs.example.com:21117)".into(),
));
}
}
// Validate password (minimum 6 characters if provided)
if let Some(ref password) = self.device_password {
if !password.is_empty() && password.len() < 6 {
return Err(AppError::BadRequest(
"Device password must be at least 6 characters".into(),
));
}
}
Ok(())
}
pub fn apply_to(&self, config: &mut RustDeskConfig) {
if let Some(enabled) = self.enabled {
config.enabled = enabled;
}
if let Some(ref server) = self.rendezvous_server {
config.rendezvous_server = server.clone();
}
if let Some(ref server) = self.relay_server {
config.relay_server = if server.is_empty() { None } else { Some(server.clone()) };
}
if let Some(ref password) = self.device_password {
if !password.is_empty() {
config.device_password = password.clone();
}
}
}
}

View File

@@ -89,6 +89,13 @@ pub fn create_router(state: Arc<AppState>) -> Router {
.route("/config/atx", patch(handlers::config::update_atx_config))
.route("/config/audio", get(handlers::config::get_audio_config))
.route("/config/audio", patch(handlers::config::update_audio_config))
// RustDesk configuration endpoints
.route("/config/rustdesk", get(handlers::config::get_rustdesk_config))
.route("/config/rustdesk", patch(handlers::config::update_rustdesk_config))
.route("/config/rustdesk/status", get(handlers::config::get_rustdesk_status))
.route("/config/rustdesk/password", get(handlers::config::get_device_password))
.route("/config/rustdesk/regenerate-id", post(handlers::config::regenerate_device_id))
.route("/config/rustdesk/regenerate-password", post(handlers::config::regenerate_device_password))
// MSD (Mass Storage Device) endpoints
.route("/msd/status", get(handlers::msd_status))
.route("/msd/images", get(handlers::msd_images_list))

View File

@@ -293,6 +293,26 @@ impl WebRtcStreamer {
Ok(pipeline)
}
/// Ensure video pipeline is running and return it for external consumers
///
/// This is a public wrapper around ensure_video_pipeline for external
/// components (like RustDesk) that need to share the encoded video stream.
pub async fn ensure_video_pipeline_for_external(
&self,
tx: broadcast::Sender<VideoFrame>,
) -> Result<Arc<SharedVideoPipeline>> {
self.ensure_video_pipeline(tx).await
}
/// Get the current pipeline configuration (if pipeline is running)
pub async fn get_pipeline_config(&self) -> Option<SharedVideoPipelineConfig> {
if let Some(ref pipeline) = *self.video_pipeline.read().await {
Some(pipeline.config().await)
} else {
None
}
}
// === Audio Management ===
/// Check if audio is enabled