feat(viewer-core): 新增安卓主控的连接/协议/收帧层(阶段 1 桌面验证通过)

定位(2026-10-29 方案裁决 A1-A4):
纯 Rust、平台无关,只做「怎么连 + 收发什么」——QUIC 会话、协议 v2 认证、
视频帧流、输入上行、Ping/Pong、KeyFrameReq。解码/渲染/触摸全在 Kotlin 侧,
本 crate 不碰像素、不依赖 openh264(dev-dependency 仅验证工具用)。

设计要点
- PeerTarget::Direct 是唯一的连接形态;不做中继、不定义 Relay 变体
  (《多平台架构设计》§6.1 裁决)。将来设备码+打洞只换 connect() 内部。
- 帧经容量 2 的 mpsc 交付,**不丢帧**:跳帧决策属于看得到解码器状态的消费方
  (Kotlin),Rust 层丢帧会重演 R-3 教训(跳帧断参考链)。
- insecure_skip_verify=false 时明确报错而非静默放行——证书钉扎登记在方案
  阶段 4,安全红线不允许默默降级。
- 双流结构与桌面 viewer 一致(流1 认证+视频,流2 输入+控制),入会即请求 IDR。

验证(headless,桌面→桌面 127.0.0.1)
- 单测 3 passed(帧契约、默认配置、strict-verify 显式报错)
- examples/headless_check:300 帧,seq 违例 0、keyframe 标志不符 0、
  解码成功 300/300、RTT 路径工作(RTT=1ms 出现)
- 全 workspace 回归:common 20 passed,构建全绿
This commit is contained in:
郑博文 2026-10-29 18:14:27 +08:00
parent 1837658c85
commit 34df7b070b
5 changed files with 600 additions and 1 deletions

12
Cargo.lock generated
View File

@ -1362,6 +1362,18 @@ dependencies = [
"tokio",
]
[[package]]
name = "viewer-core"
version = "0.1.0"
dependencies = [
"anyhow",
"common",
"openh264",
"quinn",
"rustls",
"tokio",
]
[[package]]
name = "walkdir"
version = "2.5.0"

View File

@ -2,7 +2,8 @@
resolver = "2"
# agent-core:平台无关会话层 + 软件编码视频管线(当前仅 linux-agent 使用)
# linux-agent:Linux 被控端(X11 + XTest)
members = ["common", "agent-core", "agent", "linux-agent", "viewer"]
# viewer-core:安卓主控的连接/协议/收帧层(桌面 headless 可验证;Android 经 UniFFI 复用)
members = ["common", "agent-core", "agent", "linux-agent", "viewer", "viewer-core"]
# 默认成员保持 Windows 三件套,`cargo build --release` 行为与改动前完全一致;
# Linux 被控端显式构建:`cargo build --release -p linux-agent`
default-members = ["common", "agent", "viewer"]

22
viewer-core/Cargo.toml Normal file
View File

@ -0,0 +1,22 @@
[package]
name = "viewer-core"
version = "0.1.0"
edition = "2021"
# 阶段 1 先做纯 Rust 库并在桌面 headless 验证;
# UniFFI / cdylib 在下一步(Android 接入)再加,隔离未知项。
[lib]
name = "viewer_core"
crate-type = ["lib"]
[dependencies]
common = { path = "../common" }
anyhow.workspace = true
tokio.workspace = true
quinn.workspace = true
rustls.workspace = true
[dev-dependencies]
# headless_check 用软解交叉验证收到的码流合法(库本身刻意不依赖 openh264,
# 解码在 Android 侧由 MediaCodec 承担)
openh264 = { version = "0.9.9", features = ["source"] }

View File

@ -0,0 +1,147 @@
//! viewer-core 的桌面 headless 验证工具(阶段 1 验收用,不依赖 Android)。
//!
//! 证明三件事,逐条打印:
//! 1. viewer-core 能完成 QUIC 连接 + 协议 v2 认证握手;
//! 2. 帧流按契约交付(seq 单调递增、keyframe 标志与码流实际 NAL 一致);
//! 3. 收到的码流能被 openh264 解出画面(即:Android 侧 MediaCodec 将拿到同样
//! 合法的 Annex-B——解码器不同,但码流合法性在这里就证明了)。
//!
//! 用法:
//! 终端1: target\release\agent.exe --port 47900 --code t
//! 终端2: cargo run -p viewer-core --bin headless_check -- --port 47900 --code t --frames 60
use anyhow::Result;
use std::time::Duration;
use viewer_core::{ConnectConfig, Session};
fn arg(name: &str) -> Option<String> {
let a: Vec<String> = std::env::args().collect();
a.iter()
.position(|x| x == name)
.and_then(|i| a.get(i + 1))
.cloned()
}
/// 与 dump_stream 同款:扫 Annex-B 起始码拿 NAL 类型,交叉校验 keyframe 标志
fn has_idr(payload: &[u8]) -> bool {
let mut i = 0usize;
while i + 3 < payload.len() {
let sc = if payload[i..].starts_with(&[0, 0, 0, 1]) {
4
} else if payload[i..].starts_with(&[0, 0, 1]) {
3
} else {
i += 1;
continue;
};
match payload.get(i + sc) {
Some(&b) if b & 0x1F == 5 => return true,
Some(_) => i += sc + 1,
None => break,
}
}
false
}
#[tokio::main]
async fn main() -> Result<()> {
let port: u16 = arg("--port")
.map(|s| s.parse())
.transpose()?
.unwrap_or(47800);
let code = arg("--code").unwrap_or_else(|| "123456".into());
let frames_wanted: u64 = arg("--frames")
.map(|s| s.parse())
.transpose()?
.unwrap_or(60);
let mut session = Session::connect(ConnectConfig::direct("127.0.0.1", port, code)).await?;
println!("[check] 连接成功 peer={}", session.peer_addr());
let mut decoder = openh264::decoder::Decoder::new()?;
let mut last_seq: Option<u64> = None;
let mut got = 0u64;
let mut seq_violations = 0u64;
let mut kf_mismatches = 0u64;
let mut decoded = 0u64;
let mut decode_failed_before_idr = 0u64;
let mut awaiting_idr = true; // 与 Android 侧相同的状态机
let mut rtt_shown = false;
let deadline = std::time::Instant::now() + Duration::from_secs(30);
while got < frames_wanted && std::time::Instant::now() < deadline {
let Some(f) = tokio::time::timeout(Duration::from_secs(10), session.next_frame())
.await
.ok()
.flatten()
else {
break;
};
got += 1;
// 契约 2a:seq 单调递增
if let Some(prev) = last_seq {
if f.seq <= prev {
seq_violations += 1;
println!("[check] ⚠️ seq 回退: {prev} -> {}", f.seq);
}
}
last_seq = Some(f.seq);
// 契约 2b:keyframe 标志与码流一致
let real_kf = has_idr(&f.payload);
if f.keyframe != real_kf {
kf_mismatches += 1;
println!(
"[check] ⚠️ 帧头 keyframe={} 但码流实际 IDR={}(seq={})",
f.keyframe, real_kf, f.seq
);
}
// 契约 3:与 Android 相同的状态机下解码必须成功
if awaiting_idr && !f.keyframe {
continue; // 等 IDR,P 帧不喂(这是 R-3 教训的实现)
}
if f.keyframe {
awaiting_idr = false;
}
match decoder.decode(&f.payload) {
Ok(_) => decoded += 1,
Err(e) => {
decode_failed_before_idr += 1;
println!("[check] ❌ seq={} 解码失败: {e}", f.seq);
awaiting_idr = true;
session.request_keyframe();
}
}
// RTT 出现即记录一次
if !rtt_shown {
let s = session.stats();
if s.rtt_ms > 0 {
println!("[check] RTT={}ms frames_in={} bytes_in={}", s.rtt_ms, s.frames_in, s.bytes_in);
rtt_shown = true;
}
}
}
let s = session.stats();
println!("\n========== 阶段 1 headless 验收 ==========");
println!("收到帧: {got}(目标 {frames_wanted})");
println!("seq 违例: {seq_violations}");
println!("keyframe 标志不符: {kf_mismatches}");
println!("解码成功: {decoded} / 解码失败: {decode_failed_before_idr}");
println!("统计: rtt={}ms frames_in={} bytes_in={}", s.rtt_ms, s.frames_in, s.bytes_in);
let ok = got >= frames_wanted.min(30)
&& seq_violations == 0
&& kf_mismatches == 0
&& decoded > 0
&& decode_failed_before_idr == 0;
println!(
"结论: {}",
if ok { "✅ viewer-core 收帧链路通过" } else { "❌ 存在未满足项" }
);
session.close().await;
Ok(())
}

417
viewer-core/src/lib.rs Normal file
View File

@ -0,0 +1,417 @@
//! 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>,
/// 帧接收端。通道关闭 = 会话结束(被控端断开或流错误)。
frames: mpsc::Receiver<Frame>,
stats: Arc<std::sync::Mutex<SessionStats>>,
/// 保持 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")?;
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}"
);
}
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,
stats,
_endpoint: endpoint,
conn,
})
}
/// 取下一帧。`None` = 会话结束。
pub async fn next_frame(&mut self) -> Option<Frame> {
self.frames.recv().await
}
/// 发送一个输入事件(非阻塞入队,由写任务串行发出)。
pub fn send_input(&self, ev: InputEvent) {
let _ = self.tx.send(Command::Input(ev));
}
/// 请求被控端强制关键帧(解码器错误/跳帧后的恢复路径)。
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 必须明确报错,不得静默放行"),
}
}
}