174a3f8ac0
Session::send_text 收 &[u16] 而不是 &str：整条链（Java 字符串 → JNI → 线格式 → Windows 注入）的原生表示都是 UTF-16，任何一次 UTF-8 往返都有损，故中间层不做转换。 空输入与超限在本层拦掉，超限走 clamp_text_units 并打日志。 nativeSendText 的参数用 char[]（JObject 收再转 JCharArray），经 GetCharArrayRegion 按 u16 直接拷贝。不用 JString + get_string：jni 0.21 那条路走 Modified-UTF-8， IME 提交边界的落单代理项会在往返里丢。JNI 取数失败必须 eprintln，不能静默—— 用户对着远端打字最坏的情况是「看起来发送了」。
455 lines
17 KiB
Rust
455 lines
17 KiB
Rust
//! viewer-core：安卓主控的连接 / 协议 / 收帧层（纯 Rust，平台无关）。
|
||
//!
|
||
//! 设计边界（对齐《多平台架构设计》与 2026-10-29 方案裁决）：
|
||
//! - **只做「怎么连 + 收发什么」**：QUIC 会话、认证握手、视频帧流、输入上行、
|
||
//! Ping/Pong、KeyFrameReq。解码（MediaCodec）、渲染（SurfaceView）、触摸映射
|
||
//! 全在 Kotlin 侧，本 crate 不碰像素，也不引入 openh264。
|
||
//! - **连接层是唯一的可替换点**：`PeerTarget::Direct` 现在只支持 `IP:端口` 局域网
|
||
//! 直连；将来设备码 + 信令 + 打洞只换 `connect()` 内部。**不定义 Relay 变体**
|
||
//! （中继已被团队裁决排除，见《多平台架构设计》§6.1）。
|
||
//! - 帧经 mpsc 通道交付，**不丢帧**：消费方（Kotlin）按 `keyframe` 标志自行决定
|
||
//! 跳帧策略，且跳帧后必须 `request_keyframe()` 并等 IDR（协议 v2 提供该字段）。
|
||
//! 丢帧决策放在这里会让 Rust 层承担它看不到的解码器状态。
|
||
//!
|
||
//! 安全现状：`insecure_skip_verify` 默认 true（与桌面 viewer 的 SkipVerify 一致，
|
||
//! 仅限局域网测试）。产品化前必须换成信令下发的证书指纹钉扎（`signaling.rs`
|
||
//! 已有 `CertFingerprint`），该项登记在方案阶段 4。
|
||
|
||
use anyhow::{bail, Context, Result};
|
||
use common::{
|
||
read_control, read_video_frame, write_control, ControlMsg, InputEvent, VideoFrameHeader,
|
||
PROTO_VERSION,
|
||
};
|
||
use rustls::client::danger::{
|
||
HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier,
|
||
};
|
||
use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
|
||
use rustls::{DigitallySignedStruct, SignatureScheme};
|
||
use std::net::ToSocketAddrs;
|
||
use std::sync::Arc;
|
||
use std::time::{Duration, Instant};
|
||
use tokio::sync::mpsc;
|
||
|
||
/// 连接目标。当前只有局域网直连；设备码 + 打洞在后续阶段替换 `connect()` 内部实现。
|
||
#[derive(Debug, Clone)]
|
||
pub enum PeerTarget {
|
||
/// 局域网 / 可直连地址
|
||
Direct { host: String, port: u16 },
|
||
}
|
||
|
||
/// 一次会话的参数
|
||
#[derive(Debug, Clone)]
|
||
pub struct ConnectConfig {
|
||
pub target: PeerTarget,
|
||
/// 被控端验证码
|
||
pub code: String,
|
||
/// 仅测试用：跳过服务端证书校验（与桌面 viewer 行为一致）。
|
||
/// 公网/产品形态必须为 false 并启用证书钉扎。
|
||
pub insecure_skip_verify: bool,
|
||
}
|
||
|
||
impl ConnectConfig {
|
||
/// 局域网测试的便捷构造
|
||
pub fn direct(host: impl Into<String>, port: u16, code: impl Into<String>) -> Self {
|
||
Self {
|
||
target: PeerTarget::Direct {
|
||
host: host.into(),
|
||
port,
|
||
},
|
||
code: code.into(),
|
||
insecure_skip_verify: true,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 交付给消费方的一帧视频。字段语义与 `common::VideoFrameHeader` 一致，
|
||
/// payload 为标准 Annex-B H.264（可直接喂 MediaCodec）。
|
||
#[derive(Debug, Clone)]
|
||
pub struct Frame {
|
||
pub width: u32,
|
||
pub height: u32,
|
||
pub ts_ms: u64,
|
||
pub seq: u64,
|
||
pub keyframe: bool,
|
||
pub payload: Vec<u8>,
|
||
}
|
||
|
||
impl From<(VideoFrameHeader, Vec<u8>)> for Frame {
|
||
fn from((h, payload): (VideoFrameHeader, Vec<u8>)) -> Self {
|
||
Self {
|
||
width: h.width,
|
||
height: h.height,
|
||
ts_ms: h.ts_ms,
|
||
seq: h.seq,
|
||
keyframe: h.keyframe,
|
||
payload,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 会话状态快照（供 UI 显示，不做 EMA——那属于 Kotlin 的 HUD 层）
|
||
#[derive(Debug, Clone, Copy, Default)]
|
||
pub struct SessionStats {
|
||
/// 最近一次 Ping 往返（ms），0 = 还没测出
|
||
pub rtt_ms: u64,
|
||
/// 收到的帧数
|
||
pub frames_in: u64,
|
||
/// 收到的字节数（视频载荷累计）
|
||
pub bytes_in: u64,
|
||
}
|
||
|
||
/// MVP：跳过服务端证书校验（自签证书，仅局域网）。
|
||
#[derive(Debug)]
|
||
struct SkipVerify;
|
||
impl ServerCertVerifier for SkipVerify {
|
||
fn verify_server_cert(
|
||
&self,
|
||
_end_entity: &CertificateDer<'_>,
|
||
_intermediates: &[CertificateDer<'_>],
|
||
_server_name: &ServerName<'_>,
|
||
_ocsp_response: &[u8],
|
||
_now: UnixTime,
|
||
) -> std::result::Result<ServerCertVerified, rustls::Error> {
|
||
Ok(ServerCertVerified::assertion())
|
||
}
|
||
fn verify_tls12_signature(
|
||
&self,
|
||
_message: &[u8],
|
||
_certificate: &CertificateDer<'_>,
|
||
_dss: &DigitallySignedStruct,
|
||
) -> std::result::Result<HandshakeSignatureValid, rustls::Error> {
|
||
Ok(HandshakeSignatureValid::assertion())
|
||
}
|
||
fn verify_tls13_signature(
|
||
&self,
|
||
_message: &[u8],
|
||
_certificate: &CertificateDer<'_>,
|
||
_dss: &DigitallySignedStruct,
|
||
) -> std::result::Result<HandshakeSignatureValid, rustls::Error> {
|
||
Ok(HandshakeSignatureValid::assertion())
|
||
}
|
||
fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
|
||
vec![
|
||
SignatureScheme::RSA_PKCS1_SHA256,
|
||
SignatureScheme::ECDSA_NISTP256_SHA256,
|
||
SignatureScheme::ED25519,
|
||
SignatureScheme::RSA_PSS_SHA256,
|
||
]
|
||
}
|
||
}
|
||
|
||
/// 输入/控制命令（写入流 2）
|
||
enum Command {
|
||
Input(InputEvent),
|
||
KeyFrameReq,
|
||
Ping(u64),
|
||
}
|
||
|
||
/// 已建立的会话。Drop 时后台任务随连接关闭自然退出；显式 `close()` 更干净。
|
||
pub struct Session {
|
||
tx: mpsc::UnboundedSender<Command>,
|
||
/// 帧接收端。通道关闭 = 会话结束（被控端断开或流错误）。
|
||
/// Mutex 包一层是为了让 `next_frame` 取 `&self`：JNI 层里解码线程独占 recv，
|
||
/// 输入线程并发调 `send_input`（走无锁的 unbounded tx），两者互不阻塞。
|
||
frames: Arc<tokio::sync::Mutex<mpsc::Receiver<Frame>>>,
|
||
stats: Arc<std::sync::Mutex<SessionStats>>,
|
||
/// 被控端桌面尺寸（AuthOk 携带）。主控端据此在解码前定好视图宽高比——
|
||
/// 若等首帧再改尺寸会触发 SurfaceView 重新测量，把刚创建的解码器 surface 打掉。
|
||
desktop: (u32, u32),
|
||
/// 保持 Endpoint 存活（drop 会销毁连接）
|
||
_endpoint: quinn::Endpoint,
|
||
conn: quinn::Connection,
|
||
}
|
||
|
||
impl Session {
|
||
/// 建立连接并完成认证握手 + 首帧关键帧请求。
|
||
pub async fn connect(cfg: ConnectConfig) -> Result<Self> {
|
||
rustls::crypto::ring::default_provider()
|
||
.install_default()
|
||
.ok();
|
||
|
||
let tls = if cfg.insecure_skip_verify {
|
||
rustls::ClientConfig::builder()
|
||
.dangerous()
|
||
.with_custom_certificate_verifier(Arc::new(SkipVerify))
|
||
.with_no_client_auth()
|
||
} else {
|
||
// 证书钉扎尚未接入（方案阶段 4）。这里明确报错而不是静默放行。
|
||
bail!("insecure_skip_verify=false 需要证书钉扎，尚未实现（见方案阶段 4）");
|
||
};
|
||
let qc: quinn::crypto::rustls::QuicClientConfig = Arc::new(tls)
|
||
.try_into()
|
||
.context("build QUIC client config")?;
|
||
|
||
// 接收窗口放大：视频下行大流量，quinn 默认 512KB 流窗口会卡吞吐
|
||
// （与桌面 viewer / agent 的实测调优一致）
|
||
let mut transport = quinn::TransportConfig::default();
|
||
transport.receive_window(quinn::VarInt::from_u32(16 * 1024 * 1024));
|
||
transport.stream_receive_window(quinn::VarInt::from_u32(8 * 1024 * 1024));
|
||
let mut client_config = quinn::ClientConfig::new(Arc::new(qc));
|
||
client_config.transport_config(Arc::new(transport));
|
||
|
||
let mut endpoint = quinn::Endpoint::client("0.0.0.0:0".parse()?)?;
|
||
endpoint.set_default_client_config(client_config);
|
||
|
||
let PeerTarget::Direct { host, port } = &cfg.target;
|
||
let addr = (host.as_str(), *port)
|
||
.to_socket_addrs()?
|
||
.next()
|
||
.with_context(|| format!("resolve {host}:{port}"))?;
|
||
let conn = endpoint
|
||
.connect(addr, "localhost")?
|
||
.await
|
||
.with_context(|| format!("QUIC connect {addr}"))?;
|
||
|
||
// ---- 流 1：认证 + 视频下行 ----
|
||
let (mut v_send, mut v_recv) = conn.open_bi().await?;
|
||
write_control(
|
||
&mut v_send,
|
||
&ControlMsg::Auth {
|
||
code: cfg.code.clone(),
|
||
proto_version: PROTO_VERSION,
|
||
},
|
||
)
|
||
.await
|
||
.context("send Auth")?;
|
||
let desktop: (u32, u32);
|
||
match read_control(&mut v_recv).await.context("read AuthOk")? {
|
||
ControlMsg::AuthOk {
|
||
width,
|
||
height,
|
||
proto_version,
|
||
} => {
|
||
if proto_version != PROTO_VERSION {
|
||
bail!(
|
||
"protocol mismatch: agent={proto_version}, viewer-core={PROTO_VERSION}"
|
||
);
|
||
}
|
||
desktop = (width, height);
|
||
log(&format!("auth ok, remote desktop {width}x{height}"));
|
||
}
|
||
ControlMsg::AuthFail(e) => bail!("auth failed: {e}"),
|
||
m => bail!("unexpected first message: {m:?}"),
|
||
}
|
||
|
||
// ---- 流 2：输入 + 控制 + Ping/Pong ----
|
||
let (mut ctl_send, mut ctl_recv) = conn.open_bi().await?;
|
||
// 入会即请求 IDR：从 GOP 中间接入时裸 P 帧解不出来（桌面端同款教训）
|
||
write_control(&mut ctl_send, &ControlMsg::KeyFrameReq).await?;
|
||
|
||
let stats = Arc::new(std::sync::Mutex::new(SessionStats::default()));
|
||
|
||
// 帧通道：容量 2 足够——消费方（Kotlin）每帧都要喂解码器，
|
||
// 积压超过 2 帧说明该跳帧的是消费方的决策，不是这里的。
|
||
let (frame_tx, frames) = mpsc::channel::<Frame>(2);
|
||
{
|
||
let stats = stats.clone();
|
||
tokio::spawn(async move {
|
||
loop {
|
||
match read_video_frame(&mut v_recv).await {
|
||
Ok((h, payload)) => {
|
||
{
|
||
let mut s = stats.lock().unwrap();
|
||
s.frames_in += 1;
|
||
s.bytes_in += payload.len() as u64;
|
||
}
|
||
let frame: Frame = (h, payload).into();
|
||
if frame_tx.send(frame).await.is_err() {
|
||
break; // 消费方已退出
|
||
}
|
||
}
|
||
Err(e) => {
|
||
log(&format!("video stream ended: {e}"));
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
// 命令通道 → 流 2 写任务
|
||
let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<Command>();
|
||
tokio::spawn(async move {
|
||
loop {
|
||
let Some(cmd) = cmd_rx.recv().await else {
|
||
break;
|
||
};
|
||
let msg = match cmd {
|
||
Command::Input(ev) => ControlMsg::Input(ev),
|
||
Command::KeyFrameReq => ControlMsg::KeyFrameReq,
|
||
Command::Ping(t) => ControlMsg::Ping(t),
|
||
};
|
||
if write_control(&mut ctl_send, &msg).await.is_err() {
|
||
break;
|
||
}
|
||
}
|
||
});
|
||
|
||
// 流 2 读任务：Pong → RTT
|
||
{
|
||
let stats = stats.clone();
|
||
tokio::spawn(async move {
|
||
loop {
|
||
match read_control(&mut ctl_recv).await {
|
||
Ok(ControlMsg::Pong { t, .. }) => {
|
||
let rtt = monotonic_ms().saturating_sub(t);
|
||
stats.lock().unwrap().rtt_ms = rtt;
|
||
}
|
||
Ok(_) => {}
|
||
Err(e) => {
|
||
log(&format!("control stream ended: {e}"));
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
// Ping 任务：500ms 一次（与桌面 viewer 一致）
|
||
{
|
||
let cmd_tx_ping = cmd_tx.clone();
|
||
tokio::spawn(async move {
|
||
loop {
|
||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||
if cmd_tx_ping.send(Command::Ping(monotonic_ms())).is_err() {
|
||
break;
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
Ok(Self {
|
||
tx: cmd_tx,
|
||
frames: Arc::new(tokio::sync::Mutex::new(frames)),
|
||
stats,
|
||
desktop,
|
||
_endpoint: endpoint,
|
||
conn,
|
||
})
|
||
}
|
||
|
||
/// 被控端桌面尺寸 `(width, height)`，来自 AuthOk，解码前即可用。
|
||
pub fn desktop_size(&self) -> (u32, u32) {
|
||
self.desktop
|
||
}
|
||
|
||
/// 取下一帧。`None` = 会话结束。
|
||
/// `&self`：JNI 层里解码线程独占 recv，输入线程可并发 `send_input`。
|
||
pub async fn next_frame(&self) -> Option<Frame> {
|
||
self.frames.lock().await.recv().await
|
||
}
|
||
|
||
/// 发送一个输入事件（非阻塞入队，由写任务串行发出）。
|
||
pub fn send_input(&self, ev: InputEvent) {
|
||
let _ = self.tx.send(Command::Input(ev));
|
||
}
|
||
|
||
/// 协议 v3：发送一段文本（UTF-16 code unit）到被控端注入。
|
||
///
|
||
/// 为什么参数是 `&[u16]` 而不是 `&str`：被控端的注入通道（Windows
|
||
/// `KEYEVENTF_UNICODE`）与主控端来源（Java 字符串）都是 UTF-16 code unit，
|
||
/// 中间任何一次 UTF-8 往返都会把 IME 提交边界的落单代理项变成替换字符。
|
||
/// 见 `common::InputEvent::Text` 的取舍说明。
|
||
///
|
||
/// 超限处理：截断到 [`common::MAX_TEXT_UNITS`] 并打日志——静默丢弃会让用户
|
||
/// 以为长文本已送达。日志只报长度，不报内容。
|
||
pub fn send_text(&self, units: &[u16]) {
|
||
if units.is_empty() {
|
||
return; // 空提交没有注入语义，也不该占用一条消息
|
||
}
|
||
let (units, truncated) = common::clamp_text_units(units.to_vec());
|
||
if truncated {
|
||
log(&format!(
|
||
"文本超过上限，已截断到 {} 个 code unit",
|
||
common::MAX_TEXT_UNITS
|
||
));
|
||
}
|
||
self.send_input(InputEvent::Text { units });
|
||
}
|
||
|
||
/// 请求被控端强制关键帧（解码器错误/跳帧后的恢复路径）。
|
||
pub fn request_keyframe(&self) {
|
||
let _ = self.tx.send(Command::KeyFrameReq);
|
||
}
|
||
|
||
/// 当前会话统计快照。
|
||
pub fn stats(&self) -> SessionStats {
|
||
*self.stats.lock().unwrap()
|
||
}
|
||
|
||
/// 被控端地址（日志/UI 用）
|
||
pub fn peer_addr(&self) -> String {
|
||
self.conn.remote_address().to_string()
|
||
}
|
||
|
||
/// 主动结束会话。
|
||
pub async fn close(self) {
|
||
let _ = self.conn.close(0u32.into(), b"viewer closing");
|
||
}
|
||
}
|
||
|
||
/// 单调毫秒时钟（进程启动起算）。Ping/Pong 两端共用同一口径即可测 RTT。
|
||
fn monotonic_ms() -> u64 {
|
||
use std::sync::OnceLock;
|
||
static T0: OnceLock<Instant> = OnceLock::new();
|
||
T0.get_or_init(Instant::now).elapsed().as_millis() as u64
|
||
}
|
||
|
||
/// 轻量日志出口。阶段 2 接 UniFFI 时替换为 Kotlin 侧回调。
|
||
fn log(msg: &str) {
|
||
eprintln!("[viewer-core] {msg}");
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn frame_from_header_preserves_contract_fields() {
|
||
let h = VideoFrameHeader {
|
||
width: 1920,
|
||
height: 1080,
|
||
ts_ms: 42,
|
||
seq: 7,
|
||
keyframe: true,
|
||
};
|
||
let payload = vec![0u8; 16];
|
||
let f: Frame = (h.clone(), payload.clone()).into();
|
||
assert_eq!((f.width, f.height), (1920, 1080));
|
||
assert_eq!(f.seq, 7);
|
||
assert!(f.keyframe);
|
||
assert_eq!(f.payload, payload);
|
||
}
|
||
|
||
#[test]
|
||
fn direct_config_defaults_to_insecure_lan_mode() {
|
||
let c = ConnectConfig::direct("192.168.1.5", 47800, "abc");
|
||
assert!(c.insecure_skip_verify);
|
||
match c.target {
|
||
PeerTarget::Direct { host, port } => {
|
||
assert_eq!(host, "192.168.1.5");
|
||
assert_eq!(port, 47800);
|
||
}
|
||
}
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn strict_verify_mode_is_explicit_unimplemented_error() {
|
||
// 安全红线不能静默：显式要求校验时必须是明确错误，而不是悄悄放行。
|
||
// 不用 unwrap_err()：它要求 Ok 类型（Session）实现 Debug，而连接对象
|
||
// 的 Debug 输出没有验证价值。
|
||
let cfg = ConnectConfig {
|
||
target: PeerTarget::Direct {
|
||
host: "127.0.0.1".into(),
|
||
port: 1,
|
||
},
|
||
code: "x".into(),
|
||
insecure_skip_verify: false,
|
||
};
|
||
match Session::connect(cfg).await {
|
||
Err(e) => assert!(e.to_string().contains("证书钉扎"), "实际错误: {e}"),
|
||
Ok(_) => panic!("insecure_skip_verify=false 必须明确报错，不得静默放行"),
|
||
}
|
||
}
|
||
}
|