[源码分析]- rustdesk

rustdesk 是一款非常优秀的开源控制软件, 它的社区活跃度很高(35k+✨), 基于Rust+Flutter/Sciter 等技术, 支持多桌面端(Windows, Mac, Linux)互控, 也支持通过手机连接到桌面端, 文件传输, 录屏等功能.
本文基于2022.12.30日, github 主线分支来分析 rustdesk 的一些核心功能和架构.

环境

如果要参与开发, 有些功能需要各种环境测试一遍还是挺蛋疼的…, 需要装很多虚拟机.
本地测试也需要两台可编译(一台控制端, 一台受控端)

Build :: Documentation for RustDesk

流程可以参照官方文档

• git
• Rust
• Flutter(如果是flutter版), 需要在flutter文件夹下运行 flutter pub get, 否则ffi会生成失败
• sciter(下载sdk即可)
• vcpkg: 安装音视频库依赖

一切就绪后, 启动cargo build, cargo run [--features flutter]就可以启动软件的主页面

代码分析

首先我们来看代码主要结构

flutter 目录下为 flutter 界面代码
libs 下为第三方库, 或某些额外feature的代码,
其中 libs/hbb_common 存放了一些公用代码和 pb 消息定义
src 就是软件主要代码目录,

• client 为控制端
• server 为受控端(注意与连接服务器做区分)
• ui_xx_interface 为rust代码与前端(flutter)通讯的接口, flutter 版本使用 dart:ffi + flutter_rust_bridge_codegen (基于现有库子维护的子分支, 可以看到rustdesk维护了一堆相关的依赖库, 可以说对rust桌面开发填了许多坑)
• rendezvous 是与连接服务器通讯相关 ([服务器代码](rustdesk/rustdesk-server: RustDesk Server Program (github.com)), 也是rust开发并开源的, 可以支持自定义服务器)
• lang 为简单的国际化方案, 将所有翻译组成hashmap, 根据用户本地环境和配置(libs/hbb_commom/src/configs 中)选择翻译
• platform 为平台相关代码, 这部分通常是与平台特性,限制相关的代码
• ui : sciter 版前端代码
• core_main 为入口函数, 通过不同的参数来选择各种行为, 如windows下的安装, 更新, 及通用的后台启动, 默认为启动server.
• ipc : rustdesk会为很多耗时/后台任务启动单独线程(如server), 因此封装了ipc模块来做线程间通信, 它基于[parity-tokio-ipc]([open-trade/parity-tokio-ipc: Parity tokio-ipc (github.com), 通过创建临时文件+socket的方式, 一端创建信道, 一端连接, 来进行双向通信, 并支持前缀的方式区分不同通信渠道, 所有数据都封装在ipc:Date枚举中

远程控制作为软件的核心功能, 它的大致流程可分为

注册到服务器:

每一个rustdesk软件启动后, 都会开启受控端server, 在server:start_server函数中, server启动时代码

std::thread::spawn(move || {
if let Err(err) = crate::ipc::start("") {
log::error!("Failed to start ipc: {}", err);
std::process::exit(-1);
}
});
crate::RendezvousMediator::start_all().await;

start_all代码会做下面几件事

1. check_zombie 回收子进程
2. new_server 创建一个ServerPtr, 这是一个受控端server的instance, 在后面会用到
3. test_nat_type 检查nat类型
4. direct_server 本地开启tcp监听直连请求
5. start_listening 本地监听lan局域网peer发现请求
6. 主循环, 该循环通常不会退出, 每次循环体将重设延迟数据, 之后对配置中的所有连接服务器做通信
   与连接服务器的通信循环在start函数中, 它的主要结构是

loop {
select! {
_ = xx.poll() => ...,
_ = yy.poll() => ...,
}
}

通过tokio::select! + async 函数实现类似传统网络编程中的epoll循环, 这样类似的循环代码在rustdesk代码中还可以看到很多次

具体的, 在于中继服务器通信循环中, 有两个事件,
timer.tick => 每隔1s向连接服务器发送RegisterPeer消息(register_peer), 如果通信超时则更新延迟信息, 通过全局变量SHOULD_EXIT控制退出
socket.next => socket是与服务器建立的udp连接, 消息类型是RendezvousMessage, 首先会处理到RegisterPeerResponse, 如果服务器没有找到本地的公钥, 则会要求本地通过RegisterPk发送, 随后响应RegisterPKResponse, 剩余的消息类型如果后续又遇到再关注.

现在我们已经完成流程图中第一步: 与连接服务器建立连接, 此时我们的前端也应该展示出了页面, 由此可以进入开启远程连接的步骤

连接流程

前端发起

我们以flutter桌面版代码中发起连接为起点:
common.dart: connect -> connectMainDesktop -> rustDeskWinManager.newRemoteDesktop,
使用window_manager 来支持flutter桌面版多窗口和窗口之间的通信.
multiwindow会重新跑一遍flutter的main函数, 并加上multi_window参数
main.dart: main-> runMultiWindow -> DesktopRemoteScreen(开启keyboard监听) -> ConnectiomTabPage (注册multiwindow通信) -> DesktopTab -> RemotePage

这里的代码猛一看不是很清晰(可能是前端代码的通病:), 我们重点关注initState函数, 它创建了一个FFI对象, 调用了它的 start(id) 函数, 这个id就是我们从connect中一路传过来的, 通知rust开始连接peer, 并调用 ffiModel.updateEventListener 来注册消息处理, 并且我们可以从_ImagePaintState 中看到窗口大小, 与视频图像也是从 ffi的ImageModel和CanvasModel中获取的

ffi.start代码在连接模式(还有文件传输/portForward/rdp模式)下如下:

/// Start with the given [id]. Only transfer file if [isFileTransfer], only port forward if [isPortForward].

void start(String id,
  {bool isFileTransfer = false, bool isPortForward = false}) {
assert(!(isFileTransfer && isPortForward), 'more than one connect type');

chatModel.resetClientMode();
canvasModel.id = id;
imageModel.id = id;
cursorModel.id = id;

// ignore: unused_local_variable
final addRes = bind.sessionAddSync(
  id: id, isFileTransfer: isFileTransfer, isPortForward: isPortForward);
final stream = bind.sessionStart(id: id);
final cb = ffiModel.startEventListener(id);
() async {
await for (final message in stream) {
if (message is Event) {
try {
Map<String, dynamic> event = json.decode(message.field0);
await cb(event);
} catch (e) {
debugPrint('json.decode fail1(): $e, ${message.field0}');
}
} else if (message is Rgba) {
imageModel.onRgba(message.field0);
}
}
}();
// every instance will bind a stream
this.id = id;

}

bind就是真正的rust桥接代码, 通过 generated_bridge.dart <=> flutter_ffi,
这段代码先调用sessionAddSync和sessionStart开启对端连接, 并获得一个异步流, 并通过 ffiModel.startEventListener 处理异步流中的事件消息(同ffiModel.updateEventListener), 对于 Rgba 直接更新到 imageModel.
session添加和启动的代码在flutter.rs中, session_add会创建 Session<FlutterHandler> 对象, 并添加到全局变量中, session_start_则启动线程跑session的io_loop.
这里的session在ui_session_interface中, 如我们前文所述, ui_xx_interface会负责rust<->flutter通信, 在这个文件中, 是通过 InvokeUISession : 调用ui (有flutter实现和sciter实现), 与 Interface : 向外提供的接口, 两个trait实现的.
在session的io_loop中, 使用start_video_audio_threads创建音视频处理线程, 之后创建client/io_loop:Remote对象, 开启client的io_loop

总结一下: 在用户点击连接按钮后, 新建一个对应的flutter连接窗口, 并创建对应ui页面, 随后调用rust 创建ui_session_interface对象, 绑定好两边的通信关系后, session开启io_loop, 并真正发起连接.

网络协商

在client/io_loop:io_loop中, 首先会调用Client::start, 与连接服务器通信, 从而与server(受控端)建立socket连接.

_start() {
// 1. 判断id是否是ip, 如果是ip则直接直连tcp返回
// 2. 选择一个 rendezvous server(连接server)
// 3. 重试3次, 向连接server 发送 PunchHoleRequest
//    并根据server响应 PunchHoleResponse/RelayResponse
//    进行nat直连/中继连接
// 4. 直连: connect => (如果直连失败) request_relay => secure_connect
// 5. 中继连接: create_relay => secure_connect 
}

// 该函数负责ssl握手
// 基于 sodiumoxide 
// 这里结合server中的create_tcp_connection代码来看
secure_connect() {
// 1. server -> client 
//    server 生成临时 keypair, 
//    用持久化秘钥签名 id 和 临时keypair 的公钥
//    发送 SignedId 消息
// 2. client
//    获取参数中的公钥/默认公钥
//    接收 SigendId 消息
//    利用公钥解出 SignedId 中的 id 和 server 临时公钥
//    生成本地临时 keypair, 使用 secretbox::gen_key 随机生成本地加密会话的对称key
//    seal => keypair秘钥 + server keypair 公钥 加密 对称key
//    发送 PublicKey 消息, 包含本地keypair公钥和加密后的 对称key
//    设置会话秘钥
// 3. server
//    接收 PublicKey 消息, 
//    利用临时私钥和 PublicKey中的 client 公钥, 解密 对称key
//    设置会话秘钥
}

客户端的连接已经完成, 我们再向前回退一点, 来到服务端的 RendezvousMediator loop, 当收到 PunchHole 消息进行 handle_punch_hole 打洞连接, FetchLocalAddr 消息进行 handle_intranet 内网直连, 在这两个函数中受控端会向连接服务器返回id, 网络地址, 版本, 中继server等信息. 并调用 accept_connection 函数准备接收控制连接. 在接收到连接socket后, 在create_tcp_connection中协商秘钥, 最终在server/connection.rs:Connection的start 开启server loop.

受控端初始化

在server loop之前, server还需要完成受控端的初始化,
start 中调用start_ipc中开启connection manager 进程(与该进程通信同样使用ipc机制), rustdesk中多缩写为cm, 该进程以 “–cm” 参数区分, 然后在core_main函数中转发给start_listen_ipc_thread(flutter), 创建ConnectionManager(ui_cm_interface), 开启 cm ipc监听

这里有一个小细节, 1. start_ipc 创建进程, 2. cm 进程监听ipc, 3. start_ipc 连接ipc, 这样直接可以通过连接ipc的结果判断进程是否创建成功.

初看这里启动进程其实挺奇怪的, 因为之前客户端连接时直接利用multiwindow开一个新窗口就可以. 这里蕴含了一点core_main的设计哲学, 将一些比较独立的操作, 如受控, windows中的更新等都扔到新的进程交给core_main重新转发. 这样的好处我认为可以做到解耦和错误隔离

随后交给flutter代码, flutter main中判断”–cm”flag, 转发给runConnectionManagerScreen创建DesktopServerPage, 之后开启handle_input线程处理输入控制信息

连接完成后, 我们终于获得了tcp流. 分别在 server/connection, client/io_loop 中开启受控端和控制端的io_loop. 但rustdesk逻辑上的连接还没有完成, 在真正可以传输数据之前, 我们还需要经历一段登录验证逻辑, 客户端输入密码, 服务端验证, 这里的逻辑并不是我们的重点.
直接进入验证成功后, 服务端的send_logon_response函数, 这里会创建PeerInfo作为登录成功的响应, 随后尝试激活屏幕(后台运行时, 服务端可能已经锁屏), 获取显示器信息(多显示器). 随后为ServerPtr(服务器启动时创建的)添加一个新的connection. 此时我们回头研究一下server的代码

pub struct Server {
connections: ConnMap,
services: HashMap<&'static str, Box<dyn Service>>,
id_count: i32,
}

pub type ServerPtr = Arc<RwLock<Server>>;
pub type ServerPtrWeak = Weak<RwLock<Server>>;


pub fn new() -> ServerPtr {
let mut server = Server {
connections: HashMap::new(),
services: HashMap::new(),
id_count: 0,
};

server.add_service(Box::new(audio_service::new()));
server.add_service(Box::new(video_service::new()));
#[cfg(not(any(target_os = "android", target_os = "ios")))]
{
server.add_service(Box::new(clipboard_service::new()));
if !video_service::capture_cursor_embeded() {
server.add_service(Box::new(input_service::new_cursor()));
server.add_service(Box::new(input_service::new_pos()));
}
}
Arc::new(RwLock::new(server))
}

Server只保留了id信息, connectionMap(存放connInner, 作为与connection传输数据的渠道)和servicesMap, 通过不同service来实现屏幕抓取(video_service), 音频抓取(audio_service)等受控端功能. Service的实现是订阅者模式, 当Service产生数据后发送给自己的订阅者(connInner), 从而解耦解决Rust单一所有权问题.

pub trait Service: Send + Sync {
fn name(&self) -> &'static str;
fn on_subscribe(&self, sub: ConnInner);
fn on_unsubscribe(&self, id: i32);
fn is_subed(&self, id: i32) -> bool;
fn join(&self);
}

pub trait Subscriber: Default + Send + Sync + 'static {
fn id(&self) -> i32;
fn send(&mut self, msg: Arc<Message>);
}

// 默认实现, 
pub struct ServiceInner<T: Subscriber + From<ConnInner>>;
  
// 二次封装, 提供 repeat: 给定间隔内重复执行回调参数, run: 不sleep, 循环执行回调参数 
pub struct ServiceTmpl<T: Subscriber + From<ConnInner>>(Arc<RwLock<ServiceInner<T>>>);

这里与前文的loop+select结合就是rustdesk中的一个代码范式, 每个功能都开启单独线程+tokio异步运行时, 在一个无限循环中, 异步处理事件, 事件的来源可以来自
1. socket: 来自网络, 对端或连接服务器的消息
2. ipc: 不同线程, 进程之间的通信, 消息量较少
3. timer: 定时/重复任务
4. channel: 这里的订阅者, flutter call rust, 等功能.
相信理解了这个范式就可以更好的理解rustdesk的代码了.

再来总结一下受控端的架构

流传输

视频流传输需要经过

1. 视频采集
2. 图像编码(optional: 录屏)
3. 图像传输(VideoFrame消息)
   其中前两项步骤通过libs下源码依赖的scrap

控制流可以也有多种

1. 配置 (这个流可以是多向的, 我们暂时忽略)
2. 键盘: KeyEvent
3. 鼠标: MouseEvent 鼠标点击事件, CursorData 自定义鼠标 CursorPosition 鼠标位置, cursor_id 多鼠标

键盘和鼠标的控制信号模拟都是使用 libs/enigo, 同样是自己魔改的enigo-rs, 这是一个跨平台的输入模拟库, 魔改后加入了 rdev
鼠标需要双端同步位置, 控制端在前端测采集, 调用send_mouse, 受控端在input_service中处理, 直接调用系统api监听,
键盘事件在handle_input->handle_mouse中处理, 值得注意的是键盘映射和控制key的存储.

以上就是我的rustdesk项目的简单梳理.