diff --git a/Cargo.lock b/Cargo.lock index 5c45cc8..9829ec0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/Cargo.toml b/Cargo.toml index d5fcda1..af65a74 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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"] diff --git a/viewer-core/Cargo.toml b/viewer-core/Cargo.toml new file mode 100644 index 0000000..9c559cd --- /dev/null +++ b/viewer-core/Cargo.toml @@ -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"] } diff --git a/viewer-core/examples/headless_check.rs b/viewer-core/examples/headless_check.rs new file mode 100644 index 0000000..9a8adf7 --- /dev/null +++ b/viewer-core/examples/headless_check.rs @@ -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 { + let a: Vec = 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 = 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(()) +} diff --git a/viewer-core/src/lib.rs b/viewer-core/src/lib.rs new file mode 100644 index 0000000..c61140a --- /dev/null +++ b/viewer-core/src/lib.rs @@ -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, port: u16, code: impl Into) -> 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, +} + +impl From<(VideoFrameHeader, Vec)> for Frame { + fn from((h, payload): (VideoFrameHeader, Vec)) -> 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 { + Ok(ServerCertVerified::assertion()) + } + fn verify_tls12_signature( + &self, + _message: &[u8], + _certificate: &CertificateDer<'_>, + _dss: &DigitallySignedStruct, + ) -> std::result::Result { + Ok(HandshakeSignatureValid::assertion()) + } + fn verify_tls13_signature( + &self, + _message: &[u8], + _certificate: &CertificateDer<'_>, + _dss: &DigitallySignedStruct, + ) -> std::result::Result { + Ok(HandshakeSignatureValid::assertion()) + } + fn supported_verify_schemes(&self) -> Vec { + 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, + /// 帧接收端。通道关闭 = 会话结束（被控端断开或流错误）。 + frames: mpsc::Receiver, + stats: Arc>, + /// 保持 Endpoint 存活（drop 会销毁连接） + _endpoint: quinn::Endpoint, + conn: quinn::Connection, +} + +impl Session { + /// 建立连接并完成认证握手 + 首帧关键帧请求。 + pub async fn connect(cfg: ConnectConfig) -> Result { + 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::(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::(); + 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 { + 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 = 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 必须明确报错，不得静默放行"), + } + } +}