From 60d6e7f04694cde57a6aa533f5cab38af66173df Mon Sep 17 00:00:00 2001 From: dailz Date: Mon, 13 Jul 2026 16:20:56 +0800 Subject: [PATCH] refactor(cap_portal): split 1313-LOC file into 7 submodules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 2b.1: structural split (no function decomposition — that's 2b.2). src/cap_portal.rs (1313 -> 176 LOC) now contains only the CapPortal struct, its constructor (new), accessors (frame_receiver/event_receiver/dropped_count/ capture_queue_depth), and Drop impl. Six new sibling submodules under src/cap_portal/: - types.rs (79 LOC) timeout constants, PortalPhaseTimeout enum, pub types PwDmaBufFrame / PortalFormatInfo / PwCtrlEvent - logging.rs (18 LOC) log_portal_phase_timeout helper - fourcc.rs (73 LOC) spa_to_drm_fourcc + its 2 tests - token_fs.rs (362 LOC) 8 restore-token fs helpers + 11 security tests - setup.rs (192 LOC) impl CapPortal { setup_portal + _setup_portal_inner } (associated fns; no self access — clean extract) - pipewire_thread.rs (446 LOC) PwThreadCtx (now private to this file), pipewire_thread body (verbatim, 18 SAFETY comments preserved), new spawn_pipewire_thread helper that constructs PwThreadCtx internally and returns JoinHandle. CapPortal::new now calls pipewire_thread::spawn_pipewire_thread(...) instead of inlining the PwThreadCtx construction. Oracle audit points honored: - PwThreadCtx moved as a whole; Drop in mod.rs and pipewire_thread in pipewire_thread.rs share zero state through it (PwThreadCtx consumed by-value inside pipewire_thread; spawn helper owns the construction). - All // SAFETY comments travel verbatim with their unsafe blocks. - The 18 SAFETY comments in pipewire_thread are intact; clippy undocumented_unsafe_blocks=deny still passes. API stability: - pub use types::{PwCtrlEvent, PwDmaBufFrame} preserves the existing wl_webrtc::cap_portal::{PwCtrlEvent, PwDmaBufFrame} paths used by both bench binaries (verified by cargo check --bin vaapi_import_bench --bin sw_encode_bench). - PortalFormatInfo was nominally pub in the original file but never referenced outside cap_portal; kept pub in types.rs (for cross- submodule access) but not re-exported from cap_portal.rs, so the accidental over-exposure is now scoped back. Verification (all green): - cargo build / cargo build --release - cargo test (79 lib + 3 integration = 82 pass, 1 ignored — unchanged) - cap_portal test count: 13 (fourcc=2 + token_fs=11) — matches baseline - cargo clippy --all-targets -- -D warnings - cargo fmt --check - cargo check --bin vaapi_import_bench --bin sw_encode_bench --- src/cap_portal.rs | 1181 +---------------------------- src/cap_portal/fourcc.rs | 73 ++ src/cap_portal/logging.rs | 18 + src/cap_portal/pipewire_thread.rs | 446 +++++++++++ src/cap_portal/setup.rs | 192 +++++ src/cap_portal/token_fs.rs | 362 +++++++++ src/cap_portal/types.rs | 79 ++ 7 files changed, 1192 insertions(+), 1159 deletions(-) create mode 100644 src/cap_portal/fourcc.rs create mode 100644 src/cap_portal/logging.rs create mode 100644 src/cap_portal/pipewire_thread.rs create mode 100644 src/cap_portal/setup.rs create mode 100644 src/cap_portal/token_fs.rs create mode 100644 src/cap_portal/types.rs diff --git a/src/cap_portal.rs b/src/cap_portal.rs index 6977b3e..2f297fd 100644 --- a/src/cap_portal.rs +++ b/src/cap_portal.rs @@ -12,118 +12,24 @@ // - crossbeam-channel: 高性能有界通道，用于线程间帧传递 use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; -use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; -use std::thread::{self, JoinHandle}; +use std::thread::JoinHandle; use anyhow::Result; -use crossbeam_channel::{bounded, Receiver, Sender}; +use crossbeam_channel::{bounded, Receiver}; use tokio::runtime::Runtime; use crate::args::Args; -/// Portal phase timeout when no user interaction is expected (proxy/session -/// creation, token-path select/start, PipeWire fd). 5s is generous for -/// healthy xdg-desktop-portal (<500ms typical) but bounded for fast failure. -const PORTAL_SERVICE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); +mod fourcc; +mod logging; +mod pipewire_thread; +mod setup; +mod token_fs; +mod types; -/// Portal phase timeout when user must click "Allow" in desktop dialog -/// (select/start without restore token). 30s gives time to find the dialog. -const PORTAL_USER_DIALOG_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); - -/// Classification of Portal phase timeouts to drive retry behavior. -#[derive(Debug)] -enum PortalPhaseTimeout { - /// Portal service unresponsive; not retried (user should restart service). - Service, - /// Timed out in token-dependent phase; retried once after clearing token. - TokenDependent, -} - -impl std::fmt::Display for PortalPhaseTimeout { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Service => write!(f, "Portal phase timed out (service)"), - Self::TokenDependent => write!(f, "Portal phase timed out (token-dependent)"), - } - } -} - -impl std::error::Error for PortalPhaseTimeout {} - -/// Log an actionable diagnostic when a Portal phase times out. -/// -/// Mirrors the message format from `backend_detect.rs::log_portal_unresponsive` -/// but additionally suggests `--no-persist` when the timeout occurred in a -/// phase that was using a restore token. -fn log_portal_phase_timeout(phase: &str, used_restore_token: bool) { - let persist_hint = if used_restore_token { - " If this recurs, try: wl-webrtc --no-persist" - } else { - "" - }; - tracing::error!( - "Portal service did not respond within timeout while {phase}. \ - This usually means xdg-desktop-portal or xdg-desktop-portal-kde is stuck. \ - Try: systemctl --user restart xdg-desktop-portal xdg-desktop-portal-kde, \ - then re-run wl-webrtc.{persist_hint}" - ); -} - -/// PipeWire DMA-BUF 帧数据 -/// -/// 表示从 PipeWire 流中接收到的一帧视频数据。 -/// 帧的像素数据存储在 DMA-BUF（Linux 的零拷贝 buffer 共享机制）中， -/// 通过文件描述符 (fd) 引用，消费者通过 mmap 或 DRM 导入来访问像素数据。 -pub struct PwDmaBufFrame { - /// DMA-BUF 文件描述符，指向 GPU 显存中的帧缓冲区 - pub fd: OwnedFd, - /// 帧数据在 DMA-BUF 中的字节偏移量 - pub offset: u64, - /// 每行像素的字节跨度（可能大于 width * bpp，因为可能有对齐填充） - pub stride: u32, - /// DRM 格式修饰符，描述 buffer 的内存布局（如线性布局、tiling 等） - pub modifier: u64, - /// 帧宽度（像素） - pub width: u32, - /// 帧高度（像素） - pub height: u32, - /// DRM FourCC 格式标识符（如 BGRA、RGBA 等） - pub format: u32, - /// 显示时间戳 (PTS, Presentation Time Stamp)，单位为纳秒 - pub pts: i64, -} - -/// PipeWire-negotiated video format snapshot, stashed in a `Cell` for cross-callback -/// sharing (format-change callback writes it; process callback reads it). The four -/// fields are the minimal subset of `PwDmaBufFrame`'s metadata that the process -/// callback needs to construct the frame once a buffer arrives. -/// -/// `Copy` is required because we store it inside `Cell>`; -/// `Cell` requires its contents to be `Copy` (no borrowed interior state). -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct PortalFormatInfo { - pub width: u32, - pub height: u32, - /// DRM FourCC format code (e.g. `0x34325258` for XR24 / XRGB8888). - pub drm_format: u32, - /// DRM format modifier describing buffer layout (linear, tiling, etc.). - pub modifier: u64, -} - -/// PipeWire 控制事件枚举 -/// -/// 从 PipeWire 捕获线程发送给消费者的控制事件。 -/// 与帧数据分离，通过独立的 channel 传输，确保控制事件不被帧数据淹没。 -pub enum PwCtrlEvent { - /// 流已结束（PipeWire 流断开连接或进入错误状态） - StreamEnded, - /// Format/dimensions changed mid-stream - FormatChanged { width: u32, height: u32 }, - /// 发生错误，包含错误描述信息 - Error(String), -} +pub use types::{PwCtrlEvent, PwDmaBufFrame}; /// 屏幕捕获门户（Portal）封装 /// @@ -147,19 +53,6 @@ pub struct CapPortal { pw_dropped: Arc, } -/// PipeWire 捕获线程的上下文数据 -/// -/// 从主线程传递给 PipeWire 捕获线程的所有必要资源。 -/// 该结构体在线程创建时一次性 move 到线程中使用。 -struct PwThreadCtx { - frame_tx: Sender, - event_tx: Sender, - dropped: Arc, - shutdown_read: OwnedFd, - pw_fd: OwnedFd, - node_id: u32, -} - impl CapPortal { /// 创建屏幕捕获实例 /// @@ -201,30 +94,23 @@ impl CapPortal { let pw_dropped = Arc::new(AtomicU64::new(0)); - let ctx = PwThreadCtx { + let pw_thread = pipewire_thread::spawn_pipewire_thread( frame_tx, event_tx, - dropped: pw_dropped.clone(), - // SAFETY: `efd` is the freshly-created eventfd (>= 0 checked above) and we - // are its sole owner. OwnedFd::from_raw_fd takes ownership and will close() - // it on Drop. Ownership transfers into PwThreadCtx and then into the - // PipeWire thread via pipewire_thread. - shutdown_read: unsafe { OwnedFd::from_raw_fd(efd) }, + pw_dropped.clone(), + // SAFETY: `efd` is the freshly-created eventfd (>= 0 checked above) and + // we are its sole owner. OwnedFd::from_raw_fd takes ownership and will + // close() it on Drop. Ownership transfers into the spawn helper, which + // moves it into PwThreadCtx and then into the PipeWire thread. + unsafe { OwnedFd::from_raw_fd(efd) }, pw_fd, node_id, - }; - - let pw_thread = thread::Builder::new() - .name("pipewire-capture".into()) - .spawn(move || { - pipewire_thread(ctx); - }) - .map_err(|e| { - // SAFETY: `write_fd` is the open dup'd eventfd we own (>= 0 checked - // above); closing on thread-spawn failure to avoid fd leak. - unsafe { libc::close(write_fd) }; - anyhow::anyhow!("thread spawn failed: {e}") - })?; + ) + .inspect_err(|_| { + // SAFETY: `write_fd` is the open dup'd eventfd we own (>= 0 checked + // above); closing on thread-spawn failure to avoid fd leak. + unsafe { libc::close(write_fd) }; + })?; Ok(Self { // SAFETY: `write_fd` is the freshly-dup'd eventfd (>= 0 checked above) and @@ -256,375 +142,6 @@ impl CapPortal { pub fn capture_queue_depth(&self) -> usize { self.frame_rx.len() } - - /// 通过 XDG Desktop Portal 建立屏幕录制会话 - /// - /// 与桌面环境的 D-Bus 服务交互，请求用户授权屏幕录制。 - /// 流程: - /// 1. 创建 Screencast 代理（D-Bus 代理） - /// 2. 创建 ScreenCast 会话 - /// 3. 配置源选择参数（光标模式、显示器源、不持久化会话） - /// 4. 启动录制，获取流信息（包含 PipeWire node_id） - /// 5. 打开 PipeWire 远程连接，获取文件描述符 - /// - /// 返回 (PipeWire fd, node_id)，供 PipeWire 线程连接使用 - /// - /// Wraps `_setup_portal_inner` with token-aware retry: on a `TokenDependent` - /// timeout (phases 3 or 4 with a restore token in use) AND `no_persist == - /// false`, clears the cached restore token and retries once with - /// `no_persist = true`. - async fn setup_portal(no_persist: bool) -> Result<(OwnedFd, u32)> { - match Self::_setup_portal_inner(no_persist, false).await { - Ok(result) => Ok(result), - Err(e) if e.is::() => { - let inner_err = e.downcast_ref::().unwrap(); - match inner_err { - PortalPhaseTimeout::TokenDependent if !no_persist => { - tracing::warn!( - "Portal timed out during token-using phase. \ - Clearing cached restore token and retrying with fresh authorization." - ); - delete_restore_token(); - Self::_setup_portal_inner(true, true).await - } - _ => Err(e), - } - } - Err(e) => Err(e), - } - } - - /// Inner Portal setup with phased timeouts. See `setup_portal` for the - /// retry wrapper. - /// - /// `is_retry == true` disables further retry attempts (max 1 retry). - async fn _setup_portal_inner(no_persist: bool, is_retry: bool) -> Result<(OwnedFd, u32)> { - use ashpd::desktop::screencast::{ - CursorMode, Screencast, SelectSourcesOptions, SourceType, - }; - use ashpd::desktop::PersistMode; - - // Phase 1: Screencast proxy (no user interaction). - let proxy = match tokio::time::timeout(PORTAL_SERVICE_TIMEOUT, Screencast::new()).await { - Ok(Ok(p)) => p, - Ok(Err(e)) => return Err(anyhow::anyhow!("Failed to create Screencast proxy: {e}")), - Err(_) => { - log_portal_phase_timeout("creating Screencast proxy", false); - return Err(PortalPhaseTimeout::Service.into()); - } - }; - - // Phase 2: create_session (no user interaction). - let session = match tokio::time::timeout( - PORTAL_SERVICE_TIMEOUT, - proxy.create_session(Default::default()), - ) - .await - { - Ok(Ok(s)) => s, - Ok(Err(e)) => return Err(anyhow::anyhow!("Failed to create ScreenCast session: {e}")), - Err(_) => { - log_portal_phase_timeout("creating session", false); - return Err(PortalPhaseTimeout::Service.into()); - } - }; - - let version_supported = proxy.version() >= 4; - - let (persist_mode, saved_token) = if !no_persist && version_supported { - let token = load_restore_token(); - if token.is_some() { - if is_retry { - tracing::info!("Re-attempting portal session after token clear"); - } else { - tracing::info!("Attempting to restore portal session with saved token"); - } - } - (PersistMode::ExplicitlyRevoked, token) - } else { - (PersistMode::DoNot, None) - }; - - let mut options = SelectSourcesOptions::default() - .set_cursor_mode(CursorMode::Embedded) - .set_sources(ashpd::enumflags2::BitFlags::from(SourceType::Monitor)) - .set_multiple(false) - .set_persist_mode(persist_mode); - - if let Some(ref token) = saved_token { - options = options.set_restore_token(token.as_str()); - } - - // Phase 3: select_sources — token path is fast (no dialog); fresh - // authorization may pop a dialog. - let token_in_use = saved_token.is_some(); - let phase3_timeout = if token_in_use { - PORTAL_SERVICE_TIMEOUT - } else { - PORTAL_USER_DIALOG_TIMEOUT - }; - match tokio::time::timeout(phase3_timeout, proxy.select_sources(&session, options)).await { - Ok(Ok(_)) => {} - Ok(Err(e)) => return Err(anyhow::anyhow!("Screen sharing permission denied: {e}")), - Err(_) => { - log_portal_phase_timeout("selecting sources", token_in_use); - return Err(if token_in_use { - PortalPhaseTimeout::TokenDependent - } else { - PortalPhaseTimeout::Service - } - .into()); - } - } - - // Phase 4: start + response — same dialog-vs-token reasoning as phase 3. - let phase4_timeout = if token_in_use { - PORTAL_SERVICE_TIMEOUT - } else { - PORTAL_USER_DIALOG_TIMEOUT - }; - let start_fut = async { - proxy - .start(&session, None, Default::default()) - .await? - .response() - }; - let response = match tokio::time::timeout(phase4_timeout, start_fut).await { - Ok(Ok(r)) => r, - Ok(Err(e)) => return Err(anyhow::anyhow!("ScreenCast start/response error: {e}")), - Err(_) => { - log_portal_phase_timeout("starting session", token_in_use); - return Err(if token_in_use { - PortalPhaseTimeout::TokenDependent - } else { - PortalPhaseTimeout::Service - } - .into()); - } - }; - - if !no_persist && version_supported { - if let Some(new_token) = response.restore_token() { - save_restore_token(new_token); - } - } - - let stream = response - .streams() - .first() - .ok_or_else(|| anyhow::anyhow!("No streams returned from ScreenCast"))?; - - let node_id = stream.pipe_wire_node_id(); - - // Phase 5: open_pipe_wire_remote (no user interaction). - let fd = match tokio::time::timeout( - PORTAL_SERVICE_TIMEOUT, - proxy.open_pipe_wire_remote(&session, Default::default()), - ) - .await - { - Ok(Ok(f)) => f, - Ok(Err(e)) => return Err(anyhow::anyhow!("Failed to open PipeWire remote: {e}")), - Err(_) => { - log_portal_phase_timeout("opening PipeWire remote", false); - return Err(PortalPhaseTimeout::Service.into()); - } - }; - - tracing::info!("Portal session established: node_id={node_id}"); - - Ok((fd, node_id)) - } -} - -fn token_path() -> Option { - dirs::cache_dir().map(|base| base.join("wl-webrtc").join("portal-restore-token")) -} - -/// Verify that `path` is a directory owned by the current user with no group/other permissions. -/// Rejects symlinks at the path itself (but allows the resolved target to be a real dir). -fn verify_secure_dir(path: &std::path::Path) -> bool { - use std::os::unix::fs::{MetadataExt, PermissionsExt}; - - match std::fs::symlink_metadata(path) { - Ok(meta) => { - if meta.file_type().is_symlink() { - tracing::warn!( - "Token parent dir is a symlink, rejecting: {}", - path.display() - ); - return false; - } - // Must be a directory - if !meta.is_dir() { - tracing::warn!("Token parent path is not a directory: {}", path.display()); - return false; - } - // Must be owned by current user - // SAFETY: libc::getuid has no preconditions and cannot fail; it simply - // returns the calling process's real user ID. - // SAFETY: libc::getuid has no preconditions and cannot fail. - if meta.uid() != unsafe { libc::getuid() } { - tracing::warn!( - "Token parent dir not owned by current user: {}", - path.display() - ); - return false; - } - // No group or other permissions (mode must be 0o700 exactly within the 0o777 mask) - let mode = meta.permissions().mode() & 0o777; - if mode != 0o700 { - tracing::warn!( - "Token parent dir has insecure permissions {:o}, expected 0700: {}", - mode, - path.display() - ); - return false; - } - true - } - Err(e) => { - tracing::warn!("Failed to stat token parent dir: {e}"); - false - } - } -} - -/// Ensure the parent directory exists with restrictive permissions (0o700). -/// Returns false if the directory could not be created or is insecure. -fn ensure_secure_parent(parent: &std::path::Path) -> bool { - use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; - - if parent.exists() { - // Directory exists — try to tighten permissions, then verify. - // set_permissions follows symlinks, which is fine here since - // we verify with symlink_metadata in verify_secure_dir. - if let Err(e) = std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700)) { - tracing::warn!("Failed to set directory permissions: {e}"); - return false; - } - return verify_secure_dir(parent); - } - - // Create with restrictive mode — DirBuilderExt::mode bypasses umask. - let mut builder = std::fs::DirBuilder::new(); - builder.recursive(true); - builder.mode(0o700); - if let Err(e) = builder.create(parent) { - tracing::warn!("Failed to create token directory: {e}"); - return false; - } - - // Verify after creation (belt-and-suspenders) - verify_secure_dir(parent) -} - -fn load_restore_token() -> Option { - load_restore_token_from(token_path()?) -} - -fn load_restore_token_from(path: PathBuf) -> Option { - use std::os::unix::fs::{MetadataExt, PermissionsExt}; - - let meta = match std::fs::symlink_metadata(&path) { - Ok(m) => m, - Err(_) => return None, - }; - - if meta.file_type().is_symlink() { - tracing::warn!( - "Token file is a symlink, refusing to read: {}", - path.display() - ); - return None; - } - if !meta.is_file() { - tracing::warn!("Token path is not a regular file: {}", path.display()); - return None; - } - // SAFETY: libc::getuid has no preconditions and cannot fail. - if meta.uid() != unsafe { libc::getuid() } { - tracing::warn!("Token file not owned by current user: {}", path.display()); - return None; - } - let mode = meta.permissions().mode() & 0o777; - if mode & 0o077 != 0 { - tracing::warn!( - "Token file has insecure permissions {:o}, refusing to read: {}", - mode, - path.display() - ); - return None; - } - - let token = std::fs::read_to_string(&path).ok()?; - let trimmed = token.trim().to_string(); - if trimmed.is_empty() { - None - } else { - Some(trimmed) - } -} - -fn save_restore_token(token: &str) { - let Some(path) = token_path() else { - tracing::warn!("No secure cache directory available, skipping token save"); - return; - }; - save_restore_token_to(token, &path); -} - -fn delete_restore_token() { - let Some(path) = token_path() else { - return; - }; - match std::fs::remove_file(&path) { - Ok(()) => tracing::info!("Deleted stale portal restore token at {}", path.display()), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => tracing::warn!( - "Failed to delete stale restore token at {}: {e}", - path.display() - ), - } -} - -fn save_restore_token_to(token: &str, path: &std::path::Path) { - use std::fs::OpenOptions; - use std::io::Write; - use std::os::unix::fs::OpenOptionsExt; - - let Some(parent) = path.parent() else { - tracing::warn!("Token path has no parent directory"); - return; - }; - - if !ensure_secure_parent(parent) { - tracing::warn!("Parent directory is insecure, refusing to save token"); - return; - } - - // Use a unique temp file to prevent symlink attacks. - // create_new(true) guarantees exclusive creation — fails if file already exists, - // and does NOT follow existing symlinks. - let tmp_path = path.with_extension(format!("{}.tmp", std::process::id())); - let result = (|| -> std::io::Result<()> { - let mut f = OpenOptions::new() - .write(true) - .create_new(true) - .mode(0o600) - .open(&tmp_path)?; - f.write_all(token.as_bytes())?; - f.sync_all()?; - std::fs::rename(&tmp_path, path)?; - Ok(()) - })(); - match result { - Ok(()) => tracing::info!("Saved portal restore token"), - Err(e) => { - let _ = std::fs::remove_file(&tmp_path); - tracing::warn!("Failed to save restore token: {e}"); - } - } } impl Drop for CapPortal { @@ -657,657 +174,3 @@ impl Drop for CapPortal { } } } - -/// PipeWire 捕获线程主函数 -/// -/// 在独立线程中运行 PipeWire 事件循环，接收来自 Portal 的屏幕捕获帧。 -/// 整体流程: -/// 1. 初始化 PipeWire 库 (pw::init) -/// 2. 创建 MainLoop（事件循环）、Context、Core（连接） -/// 3. 使用 Portal 提供的 fd 和 node_id 创建并连接视频流 -/// 4. 注册事件监听器（状态变化、格式协商、帧处理） -/// 5. 将 shutdown eventfd 注册到事件循环，实现安全退出 -/// 6. 运行事件循环，直到收到关闭信号 -/// 7. 清理资源，调用 pw::deinit() -/// -/// 注意: 此函数使用 Rc> 而非 Arc>，因为 PipeWire 的回调 -/// 都在同一个线程中执行，无需跨线程同步。 -fn pipewire_thread(ctx: PwThreadCtx) { - use pipewire as pw; - use pw::properties::properties; - use pw::spa::param::video::VideoInfoRaw; - use pw::stream::{StreamBox, StreamFlags}; - use std::cell::Cell; - use std::rc::Rc; - - // 初始化 PipeWire 进程全局库。 - // - // pipewire-rs 内部使用 OnceCell 保护 pw::init()，确保只调用一次。 - // pw::deinit() 是 unsafe 且要求"进程生命周期内仅调用一次，且所有 - // PipeWire 使用已停止"。由于 CapPortal 可被多次创建销毁，此函数 - // 不调用 pw::deinit()——进程退出时全局状态由 OS 回收。 - pw::init(); - - let PwThreadCtx { - frame_tx, - event_tx, - dropped, - shutdown_read, - pw_fd, - node_id, - } = ctx; - - let mainloop = match pw::main_loop::MainLoopBox::new(None) { - Ok(ml) => ml, - Err(e) => { - if let Err(e) = - event_tx.try_send(PwCtrlEvent::Error(format!("MainLoop::new failed: {e}"))) - { - tracing::error!("MainLoop::new failed and error channel also failed: {e}"); - } - return; - } - }; - - let context = match pw::context::ContextBox::new(mainloop.loop_(), None) { - Ok(c) => c, - Err(e) => { - if let Err(e) = - event_tx.try_send(PwCtrlEvent::Error(format!("Context::new failed: {e}"))) - { - tracing::error!("Context::new failed and error channel also failed: {e}"); - } - return; - } - }; - - let core = match context.connect_fd(pw_fd, None) { - Ok(c) => c, - Err(e) => { - if let Err(e) = event_tx.try_send(PwCtrlEvent::Error(format!("connect_fd failed: {e}"))) - { - tracing::error!("connect_fd failed and error channel also failed: {e}"); - } - return; - } - }; - - // 创建 PipeWire 视频流 - // 属性配置: - // - MEDIA_TYPE = "Video": 媒体类型为视频 - // - MEDIA_CATEGORY = "Capture": 类别为捕获（而非回放） - // - MEDIA_ROLE = "Screen": 角色为屏幕（用于策略管理） - let stream = match StreamBox::new( - &core, - "wl-webrtc", - properties! { - *pw::keys::MEDIA_TYPE => "Video", - *pw::keys::MEDIA_CATEGORY => "Capture", - *pw::keys::MEDIA_ROLE => "Screen", - *pw::keys::NODE_FORCE_QUANTUM => "512", - }, - ) { - Ok(s) => s, - Err(e) => { - if let Err(e) = - event_tx.try_send(PwCtrlEvent::Error(format!("Stream::new failed: {e}"))) - { - tracing::error!("Stream::new failed and error channel also failed: {e}"); - } - return; - } - }; - - let format_info: Rc>> = Rc::new(Cell::new(None)); - - let event_tx_state = event_tx.clone(); - let _listener = stream - .add_local_listener::<()>() - .state_changed(move |_, _, old, new| { - tracing::info!("PipeWire stream state: {old:?} -> {new:?}"); - match new { - pw::stream::StreamState::Error(e) => { - tracing::error!("PipeWire stream error: {e}"); - let _ = event_tx_state.try_send(PwCtrlEvent::StreamEnded); - } - pw::stream::StreamState::Unconnected => { - let _ = event_tx_state.try_send(PwCtrlEvent::StreamEnded); - } - pw::stream::StreamState::Paused => { - tracing::warn!("PipeWire stream paused (compositor may be switching content)"); - } - pw::stream::StreamState::Streaming => { - tracing::info!("PipeWire stream (re)started"); - } - pw::stream::StreamState::Connecting => {} - } - }) - // 参数变化回调（格式协商） - // PipeWire 在流格式协商完成后触发此回调 - // id 为参数类型，param 包含具体的格式参数（分辨率、像素格式等） - .param_changed({ - let format_info = format_info.clone(); - let event_tx = event_tx.clone(); - move |_, _, id, param| { - // 仅处理 Format 类型的参数变化 - let Some(param) = param else { return }; - if id != pw::spa::param::ParamType::Format.as_raw() { - return; - } - // 解析视频格式信息（分辨率、像素格式、修饰符等） - let mut info = VideoInfoRaw::new(); - if let Err(e) = info.parse(param) { - tracing::warn!("Failed to parse video format: {e}"); - return; - } - let width = info.size().width; - let height = info.size().height; - // 将 SPA 视频格式转换为 DRM FourCC 格式标识符 - let drm_format = spa_to_drm_fourcc(info.format()); - // 获取 DRM 修饰符，描述 GPU buffer 的内存布局（如 tiling 模式） - let modifier = info.modifier(); - let framerate = info.framerate(); - let max_framerate = info.max_framerate(); - // 保存协商后的格式信息，供 process 回调读取 - let previous_format = format_info.get(); - format_info.set(Some(PortalFormatInfo { - width, - height, - drm_format, - modifier, - })); - if let Some(prev) = previous_format { - if width != prev.width || height != prev.height { - tracing::warn!( - "PipeWire dimensions changed: {}x{} (format renegotiation)", - width, - height - ); - let _ = event_tx.try_send(PwCtrlEvent::FormatChanged { width, height }); - } - } - tracing::info!( - "PipeWire format negotiated: {width}x{height}, \ - drm_format={drm_format:#010x}, modifier={modifier:#x}, \ - framerate={}/{}, max_framerate={}/{}", - framerate.num, - framerate.denom, - max_framerate.num, - max_framerate.denom, - ); - } - }) - // 帧处理回调 —— 这是核心的数据路径 - // 每当 PipeWire 有新的帧数据可用时触发 - // 关键操作: 从 buffer 中提取 DMA-BUF fd，dup 后通过 channel 发送给消费者 - .process({ - let format_info = format_info.clone(); - let frame_tx = frame_tx.clone(); - move |stream, _| { - // SAFETY: raw_buf ownership invariant — PipeWire's process callback - // contract requires that every buffer acquired via `dequeue_raw_buffer` - // is returned to the queue EXACTLY ONCE via `queue_raw_buffer` before - // the callback returns — on every exit path, success or error. Failure - // to requeue leaks the buffer slot and eventually stalls the stream. - // - // Audit map of this closure (verified 2026-06-28): - // - null raw_buf (dequeue returned NULL) → nothing to requeue, return. - // - null spa_buf / no data / bad fd / null chunk / no format_info / - // invalid dims / dup_fd < 0 → all requeue before early-return. - // - success (try_send Ok / Full / Disconnected) → final requeue at end. - // The fd ownership is independent: dup() creates a fresh fd that lives - // inside PwDmaBufFrame; on try_send error the frame Drops and closes it. - let raw_buf = unsafe { stream.dequeue_raw_buffer() }; - if raw_buf.is_null() { - tracing::trace!("process: null raw_buf"); - return; - } - - // 获取 SPA buffer 结构体，包含数据数组、元数据等 - // SAFETY: raw_buf was checked non-null above. `pw_buffer.buffer` is a - // valid raw pointer for the lifetime of raw_buf (PipeWire keeps the - // buffer alive until we queue it back). - let spa_buf = unsafe { (*raw_buf).buffer }; - if spa_buf.is_null() { - tracing::trace!("process: null spa_buf"); - // SAFETY: raw_buf is the non-null buffer we still own; returning it. - unsafe { stream.queue_raw_buffer(raw_buf) }; - return; - } - - // 获取 buffer 中的数据项数量和数据指针 - // 对于 DMA-BUF 帧，通常只有 1 个数据项（包含 fd） - // SAFETY: spa_buf checked non-null above; `n_datas` is a plain u32 field. - let n_datas = unsafe { (*spa_buf).n_datas }; - // SAFETY: same as above; `datas` is a raw pointer field, may be null. - let datas_ptr = unsafe { (*spa_buf).datas }; - if n_datas == 0 || datas_ptr.is_null() { - tracing::trace!("process: no data (n_datas={n_datas})"); - // SAFETY: raw_buf still owned, returning it. - unsafe { stream.queue_raw_buffer(raw_buf) }; - return; - } - - // 从第一个数据项中获取 DMA-BUF 文件描述符 - // 通过 libspa 的 Data 包装类型安全地访问 SPA 数据结构 - // SAFETY: datas_ptr is non-null and n_datas > 0 (checked above). We cast - // to pw::spa::buffer::Data and take a shared borrow; PipeWire does not - // mutate the data array during a process cycle, so a shared reference - // for the duration of this callback is sound. - let data_ref: &pw::spa::buffer::Data = - unsafe { &*(datas_ptr as *const pw::spa::buffer::Data) }; - let fd = data_ref.fd(); - if fd < 0 { - tracing::trace!("process: invalid fd={fd}"); - // SAFETY: raw_buf still owned, returning it. - unsafe { stream.queue_raw_buffer(raw_buf) }; - return; - } - - if data_ref.as_raw().chunk.is_null() { - tracing::trace!("process: null chunk"); - // SAFETY: raw_buf still owned, returning it. - unsafe { stream.queue_raw_buffer(raw_buf) }; - return; - } - let chunk = data_ref.chunk(); - let offset = chunk.offset() as u64; - let stride = chunk.stride() as u32; - - // 从 SPA_META_Header 元数据中提取 PTS (显示时间戳) - // 遍历 buffer 的所有元数据项，查找 Header 类型的元数据 - // PTS 可用于音视频同步和帧率控制 - // SAFETY: spa_buf is non-null. `metas` is checked for null before - // iteration. We iterate `i in 0..n_metas` reading shared POD fields - // (type_, size, data) — PipeWire keeps the meta array immutable during - // a process cycle. The size guard (`meta.size >= size_of::()`) - // and null-data check before reading ensure we never read past the - // meta's actual extent. - let pts: i64 = unsafe { - let mut pts_val: i64 = 0; - let n_metas = (*spa_buf).n_metas; - let metas = (*spa_buf).metas; - if !metas.is_null() { - for i in 0..n_metas { - let meta = &*metas.add(i as usize); - if meta.type_ == libspa::sys::SPA_META_Header - && meta.size as usize - >= std::mem::size_of::() - && !meta.data.is_null() - { - let header = &*(meta.data as *const libspa::sys::spa_meta_header); - pts_val = header.pts; - break; - } - } - } - pts_val - }; - - // 验证格式信息已协商完成，且分辨率和格式有效 - let Some(fmt) = format_info.get() else { - // SAFETY: raw_buf still owned, returning it. - unsafe { stream.queue_raw_buffer(raw_buf) }; - return; - }; - let PortalFormatInfo { - width, - height, - drm_format: format, - modifier, - } = fmt; - if width == 0 || height == 0 || format == 0 { - tracing::trace!("process: invalid dimensions {width}x{height} format={format}"); - // SAFETY: raw_buf still owned, returning it. - unsafe { stream.queue_raw_buffer(raw_buf) }; - return; - } - - // 复制 DMA-BUF 文件描述符 - // 必须 dup，因为原始 fd 由 PipeWire 管理，我们不能持有它 - // dup 后的 fd 由 PwDmaBufFrame 持有，生命周期独立于 PipeWire buffer - // SAFETY: `fd` is the open DMA-BUF fd reported by PipeWire (>= 0 checked - // above). libc::dup is the standard POSIX fd duplication call. The - // original `fd` remains owned by PipeWire (returned with raw_buf later). - let dup_fd = unsafe { libc::dup(fd) }; - if dup_fd < 0 { - // SAFETY: raw_buf still owned, returning it. No fd cleanup needed - // because dup() failed and never returned a new fd. - unsafe { stream.queue_raw_buffer(raw_buf) }; - return; - } - - // 构建帧数据对象，所有必要的帧信息已收集完毕 - // SAFETY: `dup_fd` is a freshly-dup'd open file descriptor (>= 0 checked - // above) and we are its sole owner. OwnedFd::from_raw_fd takes ownership - // and will close() it on Drop. The fd's lifecycle is independent of - // raw_buf: whether try_send succeeds (frame moves into the channel) or - // fails (Full/Disconnected — the error payload owns the frame and drops - // it at the end of the match arm), exactly one close() occurs per dup(). - let frame_fd = unsafe { OwnedFd::from_raw_fd(dup_fd) }; - let frame = PwDmaBufFrame { - fd: frame_fd, - offset, - stride, - modifier, - width, - height, - format, - pts, - }; - - match frame_tx.try_send(frame) { - Ok(()) => {} - Err(crossbeam_channel::TrySendError::Full(_)) => { - dropped.fetch_add(1, Ordering::Relaxed); - } - Err(crossbeam_channel::TrySendError::Disconnected(_)) => {} - } - // SAFETY: final exactly-once requeue of raw_buf. Every path above - // either returned early with its own requeue, or falls through to here. - unsafe { stream.queue_raw_buffer(raw_buf) }; - } - }) - .register(); - - let mut params: [&pw::spa::pod::Pod; 0] = []; - - if let Err(e) = stream.connect( - pw::spa::utils::Direction::Input, - Some(node_id), - StreamFlags::AUTOCONNECT | StreamFlags::MAP_BUFFERS, - &mut params, - ) { - if let Err(e) = event_tx.try_send(PwCtrlEvent::Error(format!("stream.connect failed: {e}"))) - { - tracing::error!("stream.connect failed and error channel also failed: {e}"); - } - return; - } - - let loop_ = mainloop.loop_(); - - // Register the shutdown eventfd on the PipeWire loop. - // - // When CapPortal::drop writes to the eventfd, the loop wakes up and - // dispatches this callback on the loop thread. Because the callback - // only fires while mainloop.run() is blocking this thread, mainloop - // is guaranteed alive — eliminating the UAF that existed with the - // previous detached helper thread approach. - // 保存 mainloop 的原始指针，用于在 shutdown 回调中调用 pw_main_loop_quit - // 这是安全的，因为回调只在 mainloop.run() 阻塞期间执行 - let mainloop_ptr = mainloop.as_raw_ptr(); - - let _shutdown_source = loop_.add_io( - shutdown_read, - libspa::support::system::IoFlags::IN, - move |fd| { - // Drain the eventfd so it doesn't re-trigger - let mut buf: u64 = 0; - // SAFETY: `fd` is the registered eventfd owned by the mainloop source; the - // buffer is a stack u64 of 8 bytes matching the count argument. POSIX - // read(2) is the standard fd-read syscall; eventfd semantics require the - // 8-byte buffer. - let _ = unsafe { - libc::read( - fd.as_raw_fd(), - &mut buf as *mut u64 as *mut _, - std::mem::size_of::(), - ) - }; - // SAFETY: This callback only executes while mainloop.run() is - // blocking this thread, so mainloop is guaranteed alive. - unsafe { pipewire::sys::pw_main_loop_quit(mainloop_ptr) }; - }, - ); - - // 启动 PipeWire 主事件循环 - // 此调用会阻塞当前线程，直到 mainloop.quit() 被调用 - // quit() 由 shutdown eventfd 的 IO 回调触发 - mainloop.run(); - - // run() returned — _shutdown_source drops first (reverse declaration order), - // which unregisters the callback from the loop. Then mainloop drops. - // No dangling raw pointers are possible. - // PipeWire global state is intentionally not deinitialized here — see pw::init() comment above. -} - -/// 将 PipeWire SPA 视频格式转换为 DRM FourCC 格式 -/// -/// PipeWire 使用自己的 VideoFormat 枚举，而 DRM/KMS 使用 FourCC 格式标识。 -/// 此函数建立了两者之间的映射关系。 -/// -/// 支持的格式: -/// 不支持的格式返回 0 -/// DRM 格式名描述像素值位布局（大端序），而非内存字节序。 -/// 例如 DRM_FORMAT_ARGB8888 在小端 x86 上内存为 [B,G,R,A] = PipeWire BGRA。 -fn spa_to_drm_fourcc(format: libspa::param::video::VideoFormat) -> u32 { - use drm_fourcc::DrmFourcc; - use libspa::param::video::VideoFormat; - match format { - VideoFormat::BGRA => DrmFourcc::Argb8888 as u32, - VideoFormat::BGRx => DrmFourcc::Xrgb8888 as u32, - VideoFormat::RGBA => DrmFourcc::Abgr8888 as u32, - VideoFormat::RGBx => DrmFourcc::Xbgr8888 as u32, - VideoFormat::ARGB => DrmFourcc::Bgra8888 as u32, - VideoFormat::xRGB => DrmFourcc::Bgrx8888 as u32, - VideoFormat::ABGR => DrmFourcc::Rgba8888 as u32, - VideoFormat::xBGR => DrmFourcc::Rgbx8888 as u32, - _ => 0, - } -} - -#[cfg(test)] -mod tests { - use super::*; - use drm_fourcc::DrmFourcc; - use std::os::unix::fs::PermissionsExt; - - #[test] - fn spa_to_drm_fourcc_all_32bit() { - use libspa::param::video::VideoFormat; - assert_eq!( - spa_to_drm_fourcc(VideoFormat::BGRA), - DrmFourcc::Argb8888 as u32 - ); - assert_eq!( - spa_to_drm_fourcc(VideoFormat::BGRx), - DrmFourcc::Xrgb8888 as u32 - ); - assert_eq!( - spa_to_drm_fourcc(VideoFormat::RGBA), - DrmFourcc::Abgr8888 as u32 - ); - assert_eq!( - spa_to_drm_fourcc(VideoFormat::RGBx), - DrmFourcc::Xbgr8888 as u32 - ); - assert_eq!( - spa_to_drm_fourcc(VideoFormat::ARGB), - DrmFourcc::Bgra8888 as u32 - ); - assert_eq!( - spa_to_drm_fourcc(VideoFormat::xRGB), - DrmFourcc::Bgrx8888 as u32 - ); - assert_eq!( - spa_to_drm_fourcc(VideoFormat::ABGR), - DrmFourcc::Rgba8888 as u32 - ); - assert_eq!( - spa_to_drm_fourcc(VideoFormat::xBGR), - DrmFourcc::Rgbx8888 as u32 - ); - } - - #[test] - fn spa_to_drm_fourcc_unsupported() { - use libspa::param::video::VideoFormat; - assert_eq!(spa_to_drm_fourcc(VideoFormat::NV12), 0); - } - - #[test] - fn token_path_never_uses_tmp() { - assert!(token_path().is_some(), "token_path should resolve on Linux"); - let path = token_path().unwrap(); - assert!(!path.starts_with("/tmp"), "must not fallback to /tmp"); - } - - #[test] - fn verify_secure_dir_rejects_wrong_permissions() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path(); - - // 0o700 should pass - std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)).unwrap(); - assert!(verify_secure_dir(path)); - - // 0o755 should fail - std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755)).unwrap(); - assert!(!verify_secure_dir(path)); - - // 0o777 should fail - std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o777)).unwrap(); - assert!(!verify_secure_dir(path)); - } - - #[test] - fn verify_secure_dir_rejects_non_directory() { - let dir = tempfile::tempdir().unwrap(); - let file_path = dir.path().join("not-a-dir"); - std::fs::write(&file_path, b"test").unwrap(); - assert!(!verify_secure_dir(&file_path)); - } - - #[test] - fn ensure_secure_parent_creates_with_0700() { - let base = tempfile::tempdir().unwrap(); - let new_dir = base.path().join("wl-test-new-dir"); - assert!(!new_dir.exists()); - - assert!(ensure_secure_parent(&new_dir)); - assert!(new_dir.is_dir()); - - let meta = std::fs::symlink_metadata(&new_dir).unwrap(); - let mode = meta.permissions().mode() & 0o777; - assert_eq!( - mode, 0o700, - "created directory should be 0700, got {mode:o}" - ); - } - - #[test] - fn ensure_secure_parent_tightens_existing_dir() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path(); - - // Simulate an existing directory with loose permissions - std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755)).unwrap(); - - assert!(ensure_secure_parent(path)); - - let meta = std::fs::symlink_metadata(path).unwrap(); - let mode = meta.permissions().mode() & 0o777; - assert_eq!( - mode, 0o700, - "tightened directory should be 0700, got {mode:o}" - ); - } - - #[test] - fn save_creates_file_with_0600() { - let dir = tempfile::tempdir().unwrap(); - let token_path = dir.path().join("portal-restore-token"); - - save_restore_token_to("secret-token-123", &token_path); - - assert!(token_path.exists()); - let meta = std::fs::symlink_metadata(&token_path).unwrap(); - let mode = meta.permissions().mode() & 0o777; - assert_eq!(mode, 0o600, "token file should be 0600, got {mode:o}"); - assert_eq!( - std::fs::read_to_string(&token_path).unwrap(), - "secret-token-123" - ); - } - - #[test] - fn load_reads_secure_file() { - let dir = tempfile::tempdir().unwrap(); - let token_path = dir.path().join("portal-restore-token"); - - // Write a valid 0o600 token file - use std::os::unix::fs::OpenOptionsExt; - let mut f = std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .mode(0o600) - .open(&token_path) - .unwrap(); - std::io::Write::write_all(&mut f, b"my-secret\n").unwrap(); - - let result = load_restore_token_from(token_path); - assert_eq!(result, Some("my-secret".to_string())); - } - - #[test] - fn load_rejects_group_readable_file() { - let dir = tempfile::tempdir().unwrap(); - let token_path = dir.path().join("portal-restore-token"); - - // Write with 0o640 (group readable) — should be rejected - use std::os::unix::fs::OpenOptionsExt; - let mut f = std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .mode(0o640) - .open(&token_path) - .unwrap(); - std::io::Write::write_all(&mut f, b"leaked-token\n").unwrap(); - - let result = load_restore_token_from(token_path); - assert!(result.is_none(), "should reject group-readable token file"); - } - - #[test] - fn load_rejects_world_readable_file() { - let dir = tempfile::tempdir().unwrap(); - let token_path = dir.path().join("portal-restore-token"); - - use std::os::unix::fs::OpenOptionsExt; - let mut f = std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .mode(0o604) - .open(&token_path) - .unwrap(); - std::io::Write::write_all(&mut f, b"leaked-token\n").unwrap(); - - let result = load_restore_token_from(token_path); - assert!(result.is_none(), "should reject world-readable token file"); - } - - #[test] - fn load_rejects_symlink() { - let dir = tempfile::tempdir().unwrap(); - let real_path = dir.path().join("real-file"); - let link_path = dir.path().join("portal-restore-token"); - - std::fs::write(&real_path, b"target-content\n").unwrap(); - std::os::unix::fs::symlink(&real_path, &link_path).unwrap(); - - let result = load_restore_token_from(link_path); - assert!(result.is_none(), "should reject symlinked token file"); - } - - #[test] - fn save_then_load_roundtrip() { - let dir = tempfile::tempdir().unwrap(); - let token_path = dir.path().join("portal-restore-token"); - - save_restore_token_to("roundtrip-token", &token_path); - let loaded = load_restore_token_from(token_path); - - assert_eq!(loaded, Some("roundtrip-token".to_string())); - } -} diff --git a/src/cap_portal/fourcc.rs b/src/cap_portal/fourcc.rs new file mode 100644 index 0000000..6f437b7 --- /dev/null +++ b/src/cap_portal/fourcc.rs @@ -0,0 +1,73 @@ +/// 将 PipeWire SPA 视频格式转换为 DRM FourCC 格式 +/// +/// PipeWire 使用自己的 VideoFormat 枚举，而 DRM/KMS 使用 FourCC 格式标识。 +/// 此函数建立了两者之间的映射关系。 +/// +/// 支持的格式: +/// 不支持的格式返回 0 +/// DRM 格式名描述像素值位布局（大端序），而非内存字节序。 +/// 例如 DRM_FORMAT_ARGB8888 在小端 x86 上内存为 [B,G,R,A] = PipeWire BGRA。 +pub(super) fn spa_to_drm_fourcc(format: libspa::param::video::VideoFormat) -> u32 { + use drm_fourcc::DrmFourcc; + use libspa::param::video::VideoFormat; + match format { + VideoFormat::BGRA => DrmFourcc::Argb8888 as u32, + VideoFormat::BGRx => DrmFourcc::Xrgb8888 as u32, + VideoFormat::RGBA => DrmFourcc::Abgr8888 as u32, + VideoFormat::RGBx => DrmFourcc::Xbgr8888 as u32, + VideoFormat::ARGB => DrmFourcc::Bgra8888 as u32, + VideoFormat::xRGB => DrmFourcc::Bgrx8888 as u32, + VideoFormat::ABGR => DrmFourcc::Rgba8888 as u32, + VideoFormat::xBGR => DrmFourcc::Rgbx8888 as u32, + _ => 0, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use drm_fourcc::DrmFourcc; + + #[test] + fn spa_to_drm_fourcc_all_32bit() { + use libspa::param::video::VideoFormat; + assert_eq!( + spa_to_drm_fourcc(VideoFormat::BGRA), + DrmFourcc::Argb8888 as u32 + ); + assert_eq!( + spa_to_drm_fourcc(VideoFormat::BGRx), + DrmFourcc::Xrgb8888 as u32 + ); + assert_eq!( + spa_to_drm_fourcc(VideoFormat::RGBA), + DrmFourcc::Abgr8888 as u32 + ); + assert_eq!( + spa_to_drm_fourcc(VideoFormat::RGBx), + DrmFourcc::Xbgr8888 as u32 + ); + assert_eq!( + spa_to_drm_fourcc(VideoFormat::ARGB), + DrmFourcc::Bgra8888 as u32 + ); + assert_eq!( + spa_to_drm_fourcc(VideoFormat::xRGB), + DrmFourcc::Bgrx8888 as u32 + ); + assert_eq!( + spa_to_drm_fourcc(VideoFormat::ABGR), + DrmFourcc::Rgba8888 as u32 + ); + assert_eq!( + spa_to_drm_fourcc(VideoFormat::xBGR), + DrmFourcc::Rgbx8888 as u32 + ); + } + + #[test] + fn spa_to_drm_fourcc_unsupported() { + use libspa::param::video::VideoFormat; + assert_eq!(spa_to_drm_fourcc(VideoFormat::NV12), 0); + } +} diff --git a/src/cap_portal/logging.rs b/src/cap_portal/logging.rs new file mode 100644 index 0000000..1c8cad5 --- /dev/null +++ b/src/cap_portal/logging.rs @@ -0,0 +1,18 @@ +/// Log an actionable diagnostic when a Portal phase times out. +/// +/// Mirrors the message format from `backend_detect.rs::log_portal_unresponsive` +/// but additionally suggests `--no-persist` when the timeout occurred in a +/// phase that was using a restore token. +pub(super) fn log_portal_phase_timeout(phase: &str, used_restore_token: bool) { + let persist_hint = if used_restore_token { + " If this recurs, try: wl-webrtc --no-persist" + } else { + "" + }; + tracing::error!( + "Portal service did not respond within timeout while {phase}. \ + This usually means xdg-desktop-portal or xdg-desktop-portal-kde is stuck. \ + Try: systemctl --user restart xdg-desktop-portal xdg-desktop-portal-kde, \ + then re-run wl-webrtc.{persist_hint}" + ); +} diff --git a/src/cap_portal/pipewire_thread.rs b/src/cap_portal/pipewire_thread.rs new file mode 100644 index 0000000..5f5c75f --- /dev/null +++ b/src/cap_portal/pipewire_thread.rs @@ -0,0 +1,446 @@ +use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; +use std::sync::Arc; +use std::thread::{self, JoinHandle}; + +use anyhow::Result; +use crossbeam_channel::Sender; + +use super::fourcc::spa_to_drm_fourcc; +use super::types::{PortalFormatInfo, PwCtrlEvent, PwDmaBufFrame}; + +/// PipeWire 捕获线程的上下文数据 +/// +/// 从主线程传递给 PipeWire 捕获线程的所有必要资源。 +/// 该结构体在线程创建时一次性 move 到线程中使用。 +struct PwThreadCtx { + frame_tx: Sender, + event_tx: Sender, + dropped: Arc, + shutdown_read: OwnedFd, + pw_fd: OwnedFd, + node_id: u32, +} + +fn pipewire_thread(ctx: PwThreadCtx) { + use pipewire as pw; + use pw::properties::properties; + use pw::spa::param::video::VideoInfoRaw; + use pw::stream::{StreamBox, StreamFlags}; + use std::cell::Cell; + use std::rc::Rc; + + // 初始化 PipeWire 进程全局库。 + // + // pipewire-rs 内部使用 OnceCell 保护 pw::init()，确保只调用一次。 + // pw::deinit() 是 unsafe 且要求"进程生命周期内仅调用一次，且所有 + // PipeWire 使用已停止"。由于 CapPortal 可被多次创建销毁，此函数 + // 不调用 pw::deinit()——进程退出时全局状态由 OS 回收。 + pw::init(); + + let PwThreadCtx { + frame_tx, + event_tx, + dropped, + shutdown_read, + pw_fd, + node_id, + } = ctx; + + let mainloop = match pw::main_loop::MainLoopBox::new(None) { + Ok(ml) => ml, + Err(e) => { + if let Err(e) = + event_tx.try_send(PwCtrlEvent::Error(format!("MainLoop::new failed: {e}"))) + { + tracing::error!("MainLoop::new failed and error channel also failed: {e}"); + } + return; + } + }; + + let context = match pw::context::ContextBox::new(mainloop.loop_(), None) { + Ok(c) => c, + Err(e) => { + if let Err(e) = + event_tx.try_send(PwCtrlEvent::Error(format!("Context::new failed: {e}"))) + { + tracing::error!("Context::new failed and error channel also failed: {e}"); + } + return; + } + }; + + let core = match context.connect_fd(pw_fd, None) { + Ok(c) => c, + Err(e) => { + if let Err(e) = event_tx.try_send(PwCtrlEvent::Error(format!("connect_fd failed: {e}"))) + { + tracing::error!("connect_fd failed and error channel also failed: {e}"); + } + return; + } + }; + + // 创建 PipeWire 视频流 + // 属性配置: + // - MEDIA_TYPE = "Video": 媒体类型为视频 + // - MEDIA_CATEGORY = "Capture": 类别为捕获（而非回放） + // - MEDIA_ROLE = "Screen": 角色为屏幕（用于策略管理） + let stream = match StreamBox::new( + &core, + "wl-webrtc", + properties! { + *pw::keys::MEDIA_TYPE => "Video", + *pw::keys::MEDIA_CATEGORY => "Capture", + *pw::keys::MEDIA_ROLE => "Screen", + *pw::keys::NODE_FORCE_QUANTUM => "512", + }, + ) { + Ok(s) => s, + Err(e) => { + if let Err(e) = + event_tx.try_send(PwCtrlEvent::Error(format!("Stream::new failed: {e}"))) + { + tracing::error!("Stream::new failed and error channel also failed: {e}"); + } + return; + } + }; + + let format_info: Rc>> = Rc::new(Cell::new(None)); + + let event_tx_state = event_tx.clone(); + let _listener = stream + .add_local_listener::<()>() + .state_changed(move |_, _, old, new| { + tracing::info!("PipeWire stream state: {old:?} -> {new:?}"); + match new { + pw::stream::StreamState::Error(e) => { + tracing::error!("PipeWire stream error: {e}"); + let _ = event_tx_state.try_send(PwCtrlEvent::StreamEnded); + } + pw::stream::StreamState::Unconnected => { + let _ = event_tx_state.try_send(PwCtrlEvent::StreamEnded); + } + pw::stream::StreamState::Paused => { + tracing::warn!("PipeWire stream paused (compositor may be switching content)"); + } + pw::stream::StreamState::Streaming => { + tracing::info!("PipeWire stream (re)started"); + } + pw::stream::StreamState::Connecting => {} + } + }) + // 参数变化回调（格式协商） + // PipeWire 在流格式协商完成后触发此回调 + // id 为参数类型，param 包含具体的格式参数（分辨率、像素格式等） + .param_changed({ + let format_info = format_info.clone(); + let event_tx = event_tx.clone(); + move |_, _, id, param| { + // 仅处理 Format 类型的参数变化 + let Some(param) = param else { return }; + if id != pw::spa::param::ParamType::Format.as_raw() { + return; + } + // 解析视频格式信息（分辨率、像素格式、修饰符等） + let mut info = VideoInfoRaw::new(); + if let Err(e) = info.parse(param) { + tracing::warn!("Failed to parse video format: {e}"); + return; + } + let width = info.size().width; + let height = info.size().height; + // 将 SPA 视频格式转换为 DRM FourCC 格式标识符 + let drm_format = spa_to_drm_fourcc(info.format()); + // 获取 DRM 修饰符，描述 GPU buffer 的内存布局（如 tiling 模式） + let modifier = info.modifier(); + let framerate = info.framerate(); + let max_framerate = info.max_framerate(); + // 保存协商后的格式信息，供 process 回调读取 + let previous_format = format_info.get(); + format_info.set(Some(PortalFormatInfo { + width, + height, + drm_format, + modifier, + })); + if let Some(prev) = previous_format { + if width != prev.width || height != prev.height { + tracing::warn!( + "PipeWire dimensions changed: {}x{} (format renegotiation)", + width, + height + ); + let _ = event_tx.try_send(PwCtrlEvent::FormatChanged { width, height }); + } + } + tracing::info!( + "PipeWire format negotiated: {width}x{height}, \ + drm_format={drm_format:#010x}, modifier={modifier:#x}, \ + framerate={}/{}, max_framerate={}/{}", + framerate.num, + framerate.denom, + max_framerate.num, + max_framerate.denom, + ); + } + }) + // 帧处理回调 —— 这是核心的数据路径 + // 每当 PipeWire 有新的帧数据可用时触发 + // 关键操作: 从 buffer 中提取 DMA-BUF fd，dup 后通过 channel 发送给消费者 + .process({ + let format_info = format_info.clone(); + let frame_tx = frame_tx.clone(); + move |stream, _| { + // SAFETY: raw_buf ownership invariant — PipeWire's process callback + // contract requires that every buffer acquired via `dequeue_raw_buffer` + // is returned to the queue EXACTLY ONCE via `queue_raw_buffer` before + // the callback returns — on every exit path, success or error. Failure + // to requeue leaks the buffer slot and eventually stalls the stream. + // + // Audit map of this closure (verified 2026-06-28): + // - null raw_buf (dequeue returned NULL) → nothing to requeue, return. + // - null spa_buf / no data / bad fd / null chunk / no format_info / + // invalid dims / dup_fd < 0 → all requeue before early-return. + // - success (try_send Ok / Full / Disconnected) → final requeue at end. + // The fd ownership is independent: dup() creates a fresh fd that lives + // inside PwDmaBufFrame; on try_send error the frame Drops and closes it. + let raw_buf = unsafe { stream.dequeue_raw_buffer() }; + if raw_buf.is_null() { + tracing::trace!("process: null raw_buf"); + return; + } + + // 获取 SPA buffer 结构体，包含数据数组、元数据等 + // SAFETY: raw_buf was checked non-null above. `pw_buffer.buffer` is a + // valid raw pointer for the lifetime of raw_buf (PipeWire keeps the + // buffer alive until we queue it back). + let spa_buf = unsafe { (*raw_buf).buffer }; + if spa_buf.is_null() { + tracing::trace!("process: null spa_buf"); + // SAFETY: raw_buf is the non-null buffer we still own; returning it. + unsafe { stream.queue_raw_buffer(raw_buf) }; + return; + } + + // 获取 buffer 中的数据项数量和数据指针 + // 对于 DMA-BUF 帧，通常只有 1 个数据项（包含 fd） + // SAFETY: spa_buf checked non-null above; `n_datas` is a plain u32 field. + let n_datas = unsafe { (*spa_buf).n_datas }; + // SAFETY: same as above; `datas` is a raw pointer field, may be null. + let datas_ptr = unsafe { (*spa_buf).datas }; + if n_datas == 0 || datas_ptr.is_null() { + tracing::trace!("process: no data (n_datas={n_datas})"); + // SAFETY: raw_buf still owned, returning it. + unsafe { stream.queue_raw_buffer(raw_buf) }; + return; + } + + // 从第一个数据项中获取 DMA-BUF 文件描述符 + // 通过 libspa 的 Data 包装类型安全地访问 SPA 数据结构 + // SAFETY: datas_ptr is non-null and n_datas > 0 (checked above). We cast + // to pw::spa::buffer::Data and take a shared borrow; PipeWire does not + // mutate the data array during a process cycle, so a shared reference + // for the duration of this callback is sound. + let data_ref: &pw::spa::buffer::Data = + unsafe { &*(datas_ptr as *const pw::spa::buffer::Data) }; + let fd = data_ref.fd(); + if fd < 0 { + tracing::trace!("process: invalid fd={fd}"); + // SAFETY: raw_buf still owned, returning it. + unsafe { stream.queue_raw_buffer(raw_buf) }; + return; + } + + if data_ref.as_raw().chunk.is_null() { + tracing::trace!("process: null chunk"); + // SAFETY: raw_buf still owned, returning it. + unsafe { stream.queue_raw_buffer(raw_buf) }; + return; + } + let chunk = data_ref.chunk(); + let offset = chunk.offset() as u64; + let stride = chunk.stride() as u32; + + // 从 SPA_META_Header 元数据中提取 PTS (显示时间戳) + // 遍历 buffer 的所有元数据项，查找 Header 类型的元数据 + // PTS 可用于音视频同步和帧率控制 + // SAFETY: spa_buf is non-null. `metas` is checked for null before + // iteration. We iterate `i in 0..n_metas` reading shared POD fields + // (type_, size, data) — PipeWire keeps the meta array immutable during + // a process cycle. The size guard (`meta.size >= size_of::()`) + // and null-data check before reading ensure we never read past the + // meta's actual extent. + let pts: i64 = unsafe { + let mut pts_val: i64 = 0; + let n_metas = (*spa_buf).n_metas; + let metas = (*spa_buf).metas; + if !metas.is_null() { + for i in 0..n_metas { + let meta = &*metas.add(i as usize); + if meta.type_ == libspa::sys::SPA_META_Header + && meta.size as usize + >= std::mem::size_of::() + && !meta.data.is_null() + { + let header = &*(meta.data as *const libspa::sys::spa_meta_header); + pts_val = header.pts; + break; + } + } + } + pts_val + }; + + // 验证格式信息已协商完成，且分辨率和格式有效 + let Some(fmt) = format_info.get() else { + // SAFETY: raw_buf still owned, returning it. + unsafe { stream.queue_raw_buffer(raw_buf) }; + return; + }; + let PortalFormatInfo { + width, + height, + drm_format: format, + modifier, + } = fmt; + if width == 0 || height == 0 || format == 0 { + tracing::trace!("process: invalid dimensions {width}x{height} format={format}"); + // SAFETY: raw_buf still owned, returning it. + unsafe { stream.queue_raw_buffer(raw_buf) }; + return; + } + + // 复制 DMA-BUF 文件描述符 + // 必须 dup，因为原始 fd 由 PipeWire 管理，我们不能持有它 + // dup 后的 fd 由 PwDmaBufFrame 持有，生命周期独立于 PipeWire buffer + // SAFETY: `fd` is the open DMA-BUF fd reported by PipeWire (>= 0 checked + // above). libc::dup is the standard POSIX fd duplication call. The + // original `fd` remains owned by PipeWire (returned with raw_buf later). + let dup_fd = unsafe { libc::dup(fd) }; + if dup_fd < 0 { + // SAFETY: raw_buf still owned, returning it. No fd cleanup needed + // because dup() failed and never returned a new fd. + unsafe { stream.queue_raw_buffer(raw_buf) }; + return; + } + + // 构建帧数据对象，所有必要的帧信息已收集完毕 + // SAFETY: `dup_fd` is a freshly-dup'd open file descriptor (>= 0 checked + // above) and we are its sole owner. OwnedFd::from_raw_fd takes ownership + // and will close() it on Drop. The fd's lifecycle is independent of + // raw_buf: whether try_send succeeds (frame moves into the channel) or + // fails (Full/Disconnected — the error payload owns the frame and drops + // it at the end of the match arm), exactly one close() occurs per dup(). + let frame_fd = unsafe { OwnedFd::from_raw_fd(dup_fd) }; + let frame = PwDmaBufFrame { + fd: frame_fd, + offset, + stride, + modifier, + width, + height, + format, + pts, + }; + + match frame_tx.try_send(frame) { + Ok(()) => {} + Err(crossbeam_channel::TrySendError::Full(_)) => { + dropped.fetch_add(1, Ordering::Relaxed); + } + Err(crossbeam_channel::TrySendError::Disconnected(_)) => {} + } + // SAFETY: final exactly-once requeue of raw_buf. Every path above + // either returned early with its own requeue, or falls through to here. + unsafe { stream.queue_raw_buffer(raw_buf) }; + } + }) + .register(); + + let mut params: [&pw::spa::pod::Pod; 0] = []; + + if let Err(e) = stream.connect( + pw::spa::utils::Direction::Input, + Some(node_id), + StreamFlags::AUTOCONNECT | StreamFlags::MAP_BUFFERS, + &mut params, + ) { + if let Err(e) = event_tx.try_send(PwCtrlEvent::Error(format!("stream.connect failed: {e}"))) + { + tracing::error!("stream.connect failed and error channel also failed: {e}"); + } + return; + } + + let loop_ = mainloop.loop_(); + + // Register the shutdown eventfd on the PipeWire loop. + // + // When CapPortal::drop writes to the eventfd, the loop wakes up and + // dispatches this callback on the loop thread. Because the callback + // only fires while mainloop.run() is blocking this thread, mainloop + // is guaranteed alive — eliminating the UAF that existed with the + // previous detached helper thread approach. + // 保存 mainloop 的原始指针，用于在 shutdown 回调中调用 pw_main_loop_quit + // 这是安全的，因为回调只在 mainloop.run() 阻塞期间执行 + let mainloop_ptr = mainloop.as_raw_ptr(); + + let _shutdown_source = loop_.add_io( + shutdown_read, + libspa::support::system::IoFlags::IN, + move |fd| { + // Drain the eventfd so it doesn't re-trigger + let mut buf: u64 = 0; + // SAFETY: `fd` is the registered eventfd owned by the mainloop source; the + // buffer is a stack u64 of 8 bytes matching the count argument. POSIX + // read(2) is the standard fd-read syscall; eventfd semantics require the + // 8-byte buffer. + let _ = unsafe { + libc::read( + fd.as_raw_fd(), + &mut buf as *mut u64 as *mut _, + std::mem::size_of::(), + ) + }; + // SAFETY: This callback only executes while mainloop.run() is + // blocking this thread, so mainloop is guaranteed alive. + unsafe { pipewire::sys::pw_main_loop_quit(mainloop_ptr) }; + }, + ); + + // 启动 PipeWire 主事件循环 + // 此调用会阻塞当前线程，直到 mainloop.quit() 被调用 + // quit() 由 shutdown eventfd 的 IO 回调触发 + mainloop.run(); + + // run() returned — _shutdown_source drops first (reverse declaration order), + // which unregisters the callback from the loop. Then mainloop drops. + // No dangling raw pointers are possible. + // PipeWire global state is intentionally not deinitialized here — see pw::init() comment above. +} + +pub(super) fn spawn_pipewire_thread( + frame_tx: Sender, + event_tx: Sender, + dropped: Arc, + shutdown_read: OwnedFd, + pw_fd: OwnedFd, + node_id: u32, +) -> Result> { + let ctx = PwThreadCtx { + frame_tx, + event_tx, + dropped, + shutdown_read, + pw_fd, + node_id, + }; + thread::Builder::new() + .name("pipewire-capture".into()) + .spawn(move || pipewire_thread(ctx)) + .map_err(|e| anyhow::anyhow!("thread spawn failed: {e}")) +} diff --git a/src/cap_portal/setup.rs b/src/cap_portal/setup.rs new file mode 100644 index 0000000..5e40184 --- /dev/null +++ b/src/cap_portal/setup.rs @@ -0,0 +1,192 @@ +use std::os::fd::OwnedFd; + +use anyhow::Result; + +use super::logging::log_portal_phase_timeout; +use super::token_fs::{delete_restore_token, load_restore_token, save_restore_token}; +use super::types::{PortalPhaseTimeout, PORTAL_SERVICE_TIMEOUT, PORTAL_USER_DIALOG_TIMEOUT}; +use super::CapPortal; + +impl CapPortal { + /// 通过 XDG Desktop Portal 建立屏幕录制会话 + /// + /// 与桌面环境的 D-Bus 服务交互，请求用户授权屏幕录制。 + /// 流程: + /// 1. 创建 Screencast 代理（D-Bus 代理） + /// 2. 创建 ScreenCast 会话 + /// 3. 配置源选择参数（光标模式、显示器源、不持久化会话） + /// 4. 启动录制，获取流信息（包含 PipeWire node_id） + /// 5. 打开 PipeWire 远程连接，获取文件描述符 + /// + /// 返回 (PipeWire fd, node_id)，供 PipeWire 线程连接使用 + /// + /// Wraps `_setup_portal_inner` with token-aware retry: on a `TokenDependent` + /// timeout (phases 3 or 4 with a restore token in use) AND `no_persist == + /// false`, clears the cached restore token and retries once with + /// `no_persist = true`. + pub(super) async fn setup_portal(no_persist: bool) -> Result<(OwnedFd, u32)> { + match Self::_setup_portal_inner(no_persist, false).await { + Ok(result) => Ok(result), + Err(e) if e.is::() => { + let inner_err = e.downcast_ref::().unwrap(); + match inner_err { + PortalPhaseTimeout::TokenDependent if !no_persist => { + tracing::warn!( + "Portal timed out during token-using phase. \ + Clearing cached restore token and retrying with fresh authorization." + ); + delete_restore_token(); + Self::_setup_portal_inner(true, true).await + } + _ => Err(e), + } + } + Err(e) => Err(e), + } + } + + /// Inner Portal setup with phased timeouts. See `setup_portal` for the + /// retry wrapper. + /// + /// `is_retry == true` disables further retry attempts (max 1 retry). + pub(super) async fn _setup_portal_inner( + no_persist: bool, + is_retry: bool, + ) -> Result<(OwnedFd, u32)> { + use ashpd::desktop::screencast::{ + CursorMode, Screencast, SelectSourcesOptions, SourceType, + }; + use ashpd::desktop::PersistMode; + + // Phase 1: Screencast proxy (no user interaction). + let proxy = match tokio::time::timeout(PORTAL_SERVICE_TIMEOUT, Screencast::new()).await { + Ok(Ok(p)) => p, + Ok(Err(e)) => return Err(anyhow::anyhow!("Failed to create Screencast proxy: {e}")), + Err(_) => { + log_portal_phase_timeout("creating Screencast proxy", false); + return Err(PortalPhaseTimeout::Service.into()); + } + }; + + // Phase 2: create_session (no user interaction). + let session = match tokio::time::timeout( + PORTAL_SERVICE_TIMEOUT, + proxy.create_session(Default::default()), + ) + .await + { + Ok(Ok(s)) => s, + Ok(Err(e)) => return Err(anyhow::anyhow!("Failed to create ScreenCast session: {e}")), + Err(_) => { + log_portal_phase_timeout("creating session", false); + return Err(PortalPhaseTimeout::Service.into()); + } + }; + + let version_supported = proxy.version() >= 4; + + let (persist_mode, saved_token) = if !no_persist && version_supported { + let token = load_restore_token(); + if token.is_some() { + if is_retry { + tracing::info!("Re-attempting portal session after token clear"); + } else { + tracing::info!("Attempting to restore portal session with saved token"); + } + } + (PersistMode::ExplicitlyRevoked, token) + } else { + (PersistMode::DoNot, None) + }; + + let mut options = SelectSourcesOptions::default() + .set_cursor_mode(CursorMode::Embedded) + .set_sources(ashpd::enumflags2::BitFlags::from(SourceType::Monitor)) + .set_multiple(false) + .set_persist_mode(persist_mode); + + if let Some(ref token) = saved_token { + options = options.set_restore_token(token.as_str()); + } + + // Phase 3: select_sources — token path is fast (no dialog); fresh + // authorization may pop a dialog. + let token_in_use = saved_token.is_some(); + let phase3_timeout = if token_in_use { + PORTAL_SERVICE_TIMEOUT + } else { + PORTAL_USER_DIALOG_TIMEOUT + }; + match tokio::time::timeout(phase3_timeout, proxy.select_sources(&session, options)).await { + Ok(Ok(_)) => {} + Ok(Err(e)) => return Err(anyhow::anyhow!("Screen sharing permission denied: {e}")), + Err(_) => { + log_portal_phase_timeout("selecting sources", token_in_use); + return Err(if token_in_use { + PortalPhaseTimeout::TokenDependent + } else { + PortalPhaseTimeout::Service + } + .into()); + } + } + + // Phase 4: start + response — same dialog-vs-token reasoning as phase 3. + let phase4_timeout = if token_in_use { + PORTAL_SERVICE_TIMEOUT + } else { + PORTAL_USER_DIALOG_TIMEOUT + }; + let start_fut = async { + proxy + .start(&session, None, Default::default()) + .await? + .response() + }; + let response = match tokio::time::timeout(phase4_timeout, start_fut).await { + Ok(Ok(r)) => r, + Ok(Err(e)) => return Err(anyhow::anyhow!("ScreenCast start/response error: {e}")), + Err(_) => { + log_portal_phase_timeout("starting session", token_in_use); + return Err(if token_in_use { + PortalPhaseTimeout::TokenDependent + } else { + PortalPhaseTimeout::Service + } + .into()); + } + }; + + if !no_persist && version_supported { + if let Some(new_token) = response.restore_token() { + save_restore_token(new_token); + } + } + + let stream = response + .streams() + .first() + .ok_or_else(|| anyhow::anyhow!("No streams returned from ScreenCast"))?; + + let node_id = stream.pipe_wire_node_id(); + + // Phase 5: open_pipe_wire_remote (no user interaction). + let fd = match tokio::time::timeout( + PORTAL_SERVICE_TIMEOUT, + proxy.open_pipe_wire_remote(&session, Default::default()), + ) + .await + { + Ok(Ok(f)) => f, + Ok(Err(e)) => return Err(anyhow::anyhow!("Failed to open PipeWire remote: {e}")), + Err(_) => { + log_portal_phase_timeout("opening PipeWire remote", false); + return Err(PortalPhaseTimeout::Service.into()); + } + }; + + tracing::info!("Portal session established: node_id={node_id}"); + + Ok((fd, node_id)) + } +} diff --git a/src/cap_portal/token_fs.rs b/src/cap_portal/token_fs.rs new file mode 100644 index 0000000..d10f76c --- /dev/null +++ b/src/cap_portal/token_fs.rs @@ -0,0 +1,362 @@ +use std::path::PathBuf; + +pub(super) fn token_path() -> Option { + dirs::cache_dir().map(|base| base.join("wl-webrtc").join("portal-restore-token")) +} + +/// Verify that `path` is a directory owned by the current user with no group/other permissions. +/// Rejects symlinks at the path itself (but allows the resolved target to be a real dir). +pub(super) fn verify_secure_dir(path: &std::path::Path) -> bool { + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + + match std::fs::symlink_metadata(path) { + Ok(meta) => { + if meta.file_type().is_symlink() { + tracing::warn!( + "Token parent dir is a symlink, rejecting: {}", + path.display() + ); + return false; + } + // Must be a directory + if !meta.is_dir() { + tracing::warn!("Token parent path is not a directory: {}", path.display()); + return false; + } + // Must be owned by current user + // SAFETY: libc::getuid has no preconditions and cannot fail; it simply + // returns the calling process's real user ID. + // SAFETY: libc::getuid has no preconditions and cannot fail. + if meta.uid() != unsafe { libc::getuid() } { + tracing::warn!( + "Token parent dir not owned by current user: {}", + path.display() + ); + return false; + } + // No group or other permissions (mode must be 0o700 exactly within the 0o777 mask) + let mode = meta.permissions().mode() & 0o777; + if mode != 0o700 { + tracing::warn!( + "Token parent dir has insecure permissions {:o}, expected 0700: {}", + mode, + path.display() + ); + return false; + } + true + } + Err(e) => { + tracing::warn!("Failed to stat token parent dir: {e}"); + false + } + } +} + +/// Ensure the parent directory exists with restrictive permissions (0o700). +/// Returns false if the directory could not be created or is insecure. +pub(super) fn ensure_secure_parent(parent: &std::path::Path) -> bool { + use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; + + if parent.exists() { + // Directory exists — try to tighten permissions, then verify. + // set_permissions follows symlinks, which is fine here since + // we verify with symlink_metadata in verify_secure_dir. + if let Err(e) = std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700)) { + tracing::warn!("Failed to set directory permissions: {e}"); + return false; + } + return verify_secure_dir(parent); + } + + // Create with restrictive mode — DirBuilderExt::mode bypasses umask. + let mut builder = std::fs::DirBuilder::new(); + builder.recursive(true); + builder.mode(0o700); + if let Err(e) = builder.create(parent) { + tracing::warn!("Failed to create token directory: {e}"); + return false; + } + + // Verify after creation (belt-and-suspenders) + verify_secure_dir(parent) +} + +pub(super) fn load_restore_token() -> Option { + load_restore_token_from(token_path()?) +} + +pub(super) fn load_restore_token_from(path: PathBuf) -> Option { + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + + let meta = match std::fs::symlink_metadata(&path) { + Ok(m) => m, + Err(_) => return None, + }; + + if meta.file_type().is_symlink() { + tracing::warn!( + "Token file is a symlink, refusing to read: {}", + path.display() + ); + return None; + } + if !meta.is_file() { + tracing::warn!("Token path is not a regular file: {}", path.display()); + return None; + } + // SAFETY: libc::getuid has no preconditions and cannot fail. + if meta.uid() != unsafe { libc::getuid() } { + tracing::warn!("Token file not owned by current user: {}", path.display()); + return None; + } + let mode = meta.permissions().mode() & 0o777; + if mode & 0o077 != 0 { + tracing::warn!( + "Token file has insecure permissions {:o}, refusing to read: {}", + mode, + path.display() + ); + return None; + } + + let token = std::fs::read_to_string(&path).ok()?; + let trimmed = token.trim().to_string(); + if trimmed.is_empty() { + None + } else { + Some(trimmed) + } +} + +pub(super) fn save_restore_token(token: &str) { + let Some(path) = token_path() else { + tracing::warn!("No secure cache directory available, skipping token save"); + return; + }; + save_restore_token_to(token, &path); +} + +pub(super) fn delete_restore_token() { + let Some(path) = token_path() else { + return; + }; + match std::fs::remove_file(&path) { + Ok(()) => tracing::info!("Deleted stale portal restore token at {}", path.display()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => tracing::warn!( + "Failed to delete stale restore token at {}: {e}", + path.display() + ), + } +} + +pub(super) fn save_restore_token_to(token: &str, path: &std::path::Path) { + use std::fs::OpenOptions; + use std::io::Write; + use std::os::unix::fs::OpenOptionsExt; + + let Some(parent) = path.parent() else { + tracing::warn!("Token path has no parent directory"); + return; + }; + + if !ensure_secure_parent(parent) { + tracing::warn!("Parent directory is insecure, refusing to save token"); + return; + } + + // Use a unique temp file to prevent symlink attacks. + // create_new(true) guarantees exclusive creation — fails if file already exists, + // and does NOT follow existing symlinks. + let tmp_path = path.with_extension(format!("{}.tmp", std::process::id())); + let result = (|| -> std::io::Result<()> { + let mut f = OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(&tmp_path)?; + f.write_all(token.as_bytes())?; + f.sync_all()?; + std::fs::rename(&tmp_path, path)?; + Ok(()) + })(); + match result { + Ok(()) => tracing::info!("Saved portal restore token"), + Err(e) => { + let _ = std::fs::remove_file(&tmp_path); + tracing::warn!("Failed to save restore token: {e}"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::unix::fs::PermissionsExt; + + #[test] + fn token_path_never_uses_tmp() { + assert!(token_path().is_some(), "token_path should resolve on Linux"); + let path = token_path().unwrap(); + assert!(!path.starts_with("/tmp"), "must not fallback to /tmp"); + } + + #[test] + fn verify_secure_dir_rejects_wrong_permissions() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path(); + + // 0o700 should pass + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)).unwrap(); + assert!(verify_secure_dir(path)); + + // 0o755 should fail + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755)).unwrap(); + assert!(!verify_secure_dir(path)); + + // 0o777 should fail + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o777)).unwrap(); + assert!(!verify_secure_dir(path)); + } + + #[test] + fn verify_secure_dir_rejects_non_directory() { + let dir = tempfile::tempdir().unwrap(); + let file_path = dir.path().join("not-a-dir"); + std::fs::write(&file_path, b"test").unwrap(); + assert!(!verify_secure_dir(&file_path)); + } + + #[test] + fn ensure_secure_parent_creates_with_0700() { + let base = tempfile::tempdir().unwrap(); + let new_dir = base.path().join("wl-test-new-dir"); + assert!(!new_dir.exists()); + + assert!(ensure_secure_parent(&new_dir)); + assert!(new_dir.is_dir()); + + let meta = std::fs::symlink_metadata(&new_dir).unwrap(); + let mode = meta.permissions().mode() & 0o777; + assert_eq!( + mode, 0o700, + "created directory should be 0700, got {mode:o}" + ); + } + + #[test] + fn ensure_secure_parent_tightens_existing_dir() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path(); + + // Simulate an existing directory with loose permissions + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755)).unwrap(); + + assert!(ensure_secure_parent(path)); + + let meta = std::fs::symlink_metadata(path).unwrap(); + let mode = meta.permissions().mode() & 0o777; + assert_eq!( + mode, 0o700, + "tightened directory should be 0700, got {mode:o}" + ); + } + + #[test] + fn save_creates_file_with_0600() { + let dir = tempfile::tempdir().unwrap(); + let token_path = dir.path().join("portal-restore-token"); + + save_restore_token_to("secret-token-123", &token_path); + + assert!(token_path.exists()); + let meta = std::fs::symlink_metadata(&token_path).unwrap(); + let mode = meta.permissions().mode() & 0o777; + assert_eq!(mode, 0o600, "token file should be 0600, got {mode:o}"); + assert_eq!( + std::fs::read_to_string(&token_path).unwrap(), + "secret-token-123" + ); + } + + #[test] + fn load_reads_secure_file() { + let dir = tempfile::tempdir().unwrap(); + let token_path = dir.path().join("portal-restore-token"); + + // Write a valid 0o600 token file + use std::os::unix::fs::OpenOptionsExt; + let mut f = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(&token_path) + .unwrap(); + std::io::Write::write_all(&mut f, b"my-secret\n").unwrap(); + + let result = load_restore_token_from(token_path); + assert_eq!(result, Some("my-secret".to_string())); + } + + #[test] + fn load_rejects_group_readable_file() { + let dir = tempfile::tempdir().unwrap(); + let token_path = dir.path().join("portal-restore-token"); + + // Write with 0o640 (group readable) — should be rejected + use std::os::unix::fs::OpenOptionsExt; + let mut f = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o640) + .open(&token_path) + .unwrap(); + std::io::Write::write_all(&mut f, b"leaked-token\n").unwrap(); + + let result = load_restore_token_from(token_path); + assert!(result.is_none(), "should reject group-readable token file"); + } + + #[test] + fn load_rejects_world_readable_file() { + let dir = tempfile::tempdir().unwrap(); + let token_path = dir.path().join("portal-restore-token"); + + use std::os::unix::fs::OpenOptionsExt; + let mut f = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o604) + .open(&token_path) + .unwrap(); + std::io::Write::write_all(&mut f, b"leaked-token\n").unwrap(); + + let result = load_restore_token_from(token_path); + assert!(result.is_none(), "should reject world-readable token file"); + } + + #[test] + fn load_rejects_symlink() { + let dir = tempfile::tempdir().unwrap(); + let real_path = dir.path().join("real-file"); + let link_path = dir.path().join("portal-restore-token"); + + std::fs::write(&real_path, b"target-content\n").unwrap(); + std::os::unix::fs::symlink(&real_path, &link_path).unwrap(); + + let result = load_restore_token_from(link_path); + assert!(result.is_none(), "should reject symlinked token file"); + } + + #[test] + fn save_then_load_roundtrip() { + let dir = tempfile::tempdir().unwrap(); + let token_path = dir.path().join("portal-restore-token"); + + save_restore_token_to("roundtrip-token", &token_path); + let loaded = load_restore_token_from(token_path); + + assert_eq!(loaded, Some("roundtrip-token".to_string())); + } +} diff --git a/src/cap_portal/types.rs b/src/cap_portal/types.rs new file mode 100644 index 0000000..1f12493 --- /dev/null +++ b/src/cap_portal/types.rs @@ -0,0 +1,79 @@ +use std::os::fd::OwnedFd; + +pub(super) const PORTAL_SERVICE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); +pub(super) const PORTAL_USER_DIALOG_TIMEOUT: std::time::Duration = + std::time::Duration::from_secs(30); + +/// Classification of Portal phase timeouts to drive retry behavior. +#[derive(Debug)] +pub(super) enum PortalPhaseTimeout { + /// Portal service unresponsive; not retried (user should restart service). + Service, + /// Timed out in token-dependent phase; retried once after clearing token. + TokenDependent, +} + +impl std::fmt::Display for PortalPhaseTimeout { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Service => write!(f, "Portal phase timed out (service)"), + Self::TokenDependent => write!(f, "Portal phase timed out (token-dependent)"), + } + } +} + +impl std::error::Error for PortalPhaseTimeout {} + +/// PipeWire DMA-BUF 帧数据 +/// +/// 表示从 PipeWire 流中接收到的一帧视频数据。 +/// 帧的像素数据存储在 DMA-BUF（Linux 的零拷贝 buffer 共享机制）中， +/// 通过文件描述符 (fd) 引用，消费者通过 mmap 或 DRM 导入来访问像素数据。 +pub struct PwDmaBufFrame { + /// DMA-BUF 文件描述符，指向 GPU 显存中的帧缓冲区 + pub fd: OwnedFd, + /// 帧数据在 DMA-BUF 中的字节偏移量 + pub offset: u64, + /// 每行像素的字节跨度（可能大于 width * bpp，因为可能有对齐填充） + pub stride: u32, + /// DRM 格式修饰符，描述 buffer 的内存布局（如线性布局、tiling 等） + pub modifier: u64, + /// 帧宽度（像素） + pub width: u32, + /// 帧高度（像素） + pub height: u32, + /// DRM FourCC 格式标识符（如 BGRA、RGBA 等） + pub format: u32, + /// 显示时间戳 (PTS, Presentation Time Stamp)，单位为纳秒 + pub pts: i64, +} + +/// PipeWire-negotiated video format snapshot, stashed in a `Cell` for cross-callback +/// sharing (format-change callback writes it; process callback reads it). The four +/// fields are the minimal subset of `PwDmaBufFrame`'s metadata that the process +/// callback needs to construct the frame once a buffer arrives. +/// +/// `Copy` is required because we store it inside `Cell>`; +/// `Cell` requires its contents to be `Copy` (no borrowed interior state). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PortalFormatInfo { + pub width: u32, + pub height: u32, + /// DRM FourCC format code (e.g. `0x34325258` for XR24 / XRGB8888). + pub drm_format: u32, + /// DRM format modifier describing buffer layout (linear, tiling, etc.). + pub modifier: u64, +} + +/// PipeWire 控制事件枚举 +/// +/// 从 PipeWire 捕获线程发送给消费者的控制事件。 +/// 与帧数据分离，通过独立的 channel 传输，确保控制事件不被帧数据淹没。 +pub enum PwCtrlEvent { + /// 流已结束（PipeWire 流断开连接或进入错误状态） + StreamEnded, + /// Format/dimensions changed mid-stream + FormatChanged { width: u32, height: u32 }, + /// 发生错误，包含错误描述信息 + Error(String), +}