diff --git a/Cargo.lock b/Cargo.lock index b9a28507f..c18df7e3c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1327,6 +1327,8 @@ dependencies = [ "gtk4", "hmac", "httpdate", + "hyper", + "hyper-util", "libadwaita", "libc", "librqbit", @@ -1341,6 +1343,7 @@ dependencies = [ "sha2", "thiserror 2.0.21", "tokio", + "tower", "tracing", "tracing-subscriber", "url", diff --git a/Cargo.toml b/Cargo.toml index 7ca134710..9da984b70 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,10 @@ serde_json = "1" httpdate = "1" # Local HTTP server for browser extension integration (replaces native messaging). axum = { version = "0.8", features = ["json"] } +# Configure HTTP/1 header-read deadlines for the loopback extension server. +hyper = { version = "1", features = ["http1", "server"] } +hyper-util = { version = "0.1", default-features = false, features = ["service", "tokio"] } +tower = { version = "0.5", features = ["util"] } # SHA-256 for verifying downloaded tool binaries against upstream release digests. sha2 = "0.11" # HMAC-SHA256 for the browser-extension pairing handshake (extension_server). diff --git a/src/extension_server.rs b/src/extension_server.rs index 7595f24c9..1c8f57f7c 100644 --- a/src/extension_server.rs +++ b/src/extension_server.rs @@ -47,6 +47,11 @@ use axum::{ routing::{get, post}, }; use hmac::{Hmac, KeyInit as _, Mac}; +use hyper::Request as HyperRequest; +use hyper::body::Incoming; +use hyper::server::conn::http1; +use hyper_util::rt::{TokioIo, TokioTimer}; +use hyper_util::service::TowerToHyperService; use serde::Deserialize; use sha2::Sha256; use std::collections::{HashMap, HashSet, VecDeque}; @@ -60,6 +65,7 @@ use tokio::net::TcpListener; use tokio::sync::Semaphore; use tokio::sync::mpsc::{Sender, error::TrySendError}; use tokio::sync::oneshot; +use tower::ServiceExt as _; const PORT: u16 = 9412; /// A URL plus JSON framing; the intake itself caps URLs at `download_intake::MAX_URL_LEN`. @@ -69,6 +75,11 @@ pub(crate) const QUEUE_DEPTH: usize = 16; /// How long `/add?confirm=1` waits for the user to answer the dialog. The /// extension's own timeout must be longer. pub(crate) const CONFIRM_WAIT: Duration = Duration::from_secs(120); +/// Bound slow HTTP/1 request headers on the loopback extension API. +const EXTENSION_HEADER_READ_TIMEOUT: Duration = Duration::from_secs(10); +/// Bound the lifetime of one connection, including stalled request bodies. +/// Longer than CONFIRM_WAIT so a legitimate user confirmation can complete. +const EXTENSION_CONNECTION_TIMEOUT: Duration = Duration::from_secs(180); /// Upper bound on confirmation requests in flight. The handoff queue depth /// alone does not bound dialogs: the GTK thread drains it into one dialog per /// request. Kept well below the queue so a flood cannot bury the user. @@ -419,12 +430,14 @@ struct HandoffCookies { staged_instant: Instant, } -#[derive(Clone, Copy)] +#[derive(Clone)] struct SuppressedHost { + /// Effective cutoff: newest approved suppression or in-flight request. at: Instant, - /// Number of in-flight cookieless handoffs that rely on this tombstone. - /// The entry disappears only when the last one is refused/declined. - pending: usize, + /// Last approved cutoff remains in force while any older jar exists. + approved_at: Option, + /// Unique cutoff for each pending handoff, so declines do not undo others. + pending: Vec<(u64, Instant)>, } /// Cookie files handed off from the extension, keyed by canonical URL. Every @@ -482,6 +495,7 @@ const MAX_SUPPRESSED_HOSTS: usize = MAX_HANDOFF_ENTRIES * 4; /// downloads run immediately; this only affects cards left open for a while. const HANDOFF_FRESH: Duration = Duration::from_secs(30 * 60); static COOKIE_FILE_COUNTER: AtomicU64 = AtomicU64::new(0); +static SUPPRESSION_ID_COUNTER: AtomicU64 = AtomicU64::new(0); /// Canonical map key for a handoff URL: the `url::Url` serialization, so the /// raw extension string and the intake-normalized form hit the same entry @@ -609,7 +623,22 @@ fn write_handoff_cookies_with_count(url: &str, cookies: &[Cookie]) -> Option<(Pa opts.mode(0o600); } let mut file = opts.open(&path).ok()?; - file.write_all(content.as_bytes()).ok()?; + if let Err(error) = file.write_all(content.as_bytes()) { + drop(file); + if let Err(cleanup_error) = std::fs::remove_file(&path) { + tracing::debug!( + path = %path.display(), + error = %cleanup_error, + "could not remove incomplete extension cookie file" + ); + } + tracing::debug!( + path = %path.display(), + %error, + "could not write extension cookie file" + ); + return None; + } Some((path, written)) } @@ -647,6 +676,10 @@ pub(crate) fn store_handoff_cookies(url: &str, path: PathBuf, user_agent: Option let _ = std::fs::remove_file(old.path); } } + let hosts: Vec = store.suppressed_hosts.keys().cloned().collect(); + for host in hosts { + prune_suppression(&mut store, &host); + } } /// `--cookies` argv for a URL's staged handoff cookies, if any. Peeks: the @@ -687,15 +720,13 @@ fn handoff_entry(url: &str) -> Option { }) } -/// A suppression owned by one cookieless handoff. Dropping it (decline, -/// timeout, queue refusal, or handler cancellation) releases exactly this -/// request's reference; approving the handoff calls [`Self::keep`]. +/// A suppression owned by one cookieless handoff. Each request has its own cutoff. struct Suppression { host: String, - keep: bool, - /// The handoff reached the GTK queue. If this handler is cancelled while - /// waiting for the answer, GTK may still show the dialog and route the - /// handoff, so cancellation must keep the tombstone (fail closed). + id: u64, + at: Instant, + finished: bool, + /// A queued handoff may still be processed by GTK after client disconnect. sent: bool, } @@ -704,38 +735,93 @@ impl Suppression { self.sent = true; } - /// Release explicitly after a decline/failure/timeout. + /// Release after a decline, timeout, queue refusal, or other failure. fn release(&mut self) { - if !self.keep { - unsuppress_handoff_host(&self.host); - self.keep = true; + if !self.finished { + finish_suppression(&self.host, self.id, self.at, false); + self.finished = true; } } + /// Keep older staged jars suppressed after an approved handoff. fn keep(mut self) { - self.keep = true; + if !self.finished { + finish_suppression(&self.host, self.id, self.at, true); + self.finished = true; + } } } impl Drop for Suppression { fn drop(&mut self) { - if !self.keep && !self.sent { - unsuppress_handoff_host(&self.host); + if !self.finished { + // Once sent to GTK, cancellation cannot prove the handoff was declined. + // Fail closed until the older jar is removed or replaced. + finish_suppression(&self.host, self.id, self.at, self.sent); + self.finished = true; } } } -/// Suppress older staged jars for `url`'s host until this request is -/// approved. Each in-flight request holds one reference so declining one -/// cannot re-expose a jar another still needs hidden. +/// Suppress older staged jars for a URL's host until the request is approved. +fn host_has_stale_handoff(store: &HandoffStore, host: &str, at: Instant) -> bool { + store.by_key.iter().any(|(key, entry)| { + entry.staged_instant < at + && url::Url::parse(key) + .ok() + .and_then(|parsed| parsed.host_str().map(str::to_ascii_lowercase)) + .is_some_and(|key_host| key_host == host) + }) +} + +/// Remove a tombstone only when no request depends on it and deleting it cannot +/// expose a jar older than the last approved cutoff. +fn prune_suppression(store: &mut HandoffStore, host: &str) { + let Some(suppressed) = store.suppressed_hosts.get(host).cloned() else { + return; + }; + let removable = suppressed.pending.is_empty() + && suppressed + .approved_at + .is_none_or(|at| !host_has_stale_handoff(store, host, at)); + if removable { + store.suppressed_hosts.remove(host); + store.suppressed_order.retain(|candidate| candidate != host); + } +} + +fn finish_suppression(host: &str, id: u64, at: Instant, approved: bool) { + let mut store = lock_handoff_cookies(); + if let Some(suppressed) = store.suppressed_hosts.get_mut(host) { + if let Some(index) = suppressed + .pending + .iter() + .position(|(pending_id, _)| *pending_id == id) + { + suppressed.pending.remove(index); + } + if approved { + suppressed.approved_at = Some(suppressed.approved_at.map_or(at, |old| old.max(at))); + } + if let Some(cutoff) = suppressed + .pending + .iter() + .map(|(_, cutoff)| *cutoff) + .chain(suppressed.approved_at) + .max() + { + suppressed.at = cutoff; + } + } + prune_suppression(&mut store, host); +} + fn suppress_handoff_host(url: &str) -> Option { let host = url::Url::parse(url) .ok() .and_then(|parsed| parsed.host_str().map(str::to_ascii_lowercase))?; let mut store = lock_handoff_cookies(); - // Without an older jar there is nothing to suppress. This also keeps a - // cookieless flood from growing the tombstone map: only hosts that had a - // staged entry can need one. + // Without a staged jar there is nothing to suppress and no tombstone is needed. let has_staged = store.by_key.keys().any(|key| { url::Url::parse(key) .ok() @@ -745,30 +831,36 @@ fn suppress_handoff_host(url: &str) -> Option { if !has_staged { return None; } - // Moving `at` forward is more restrictive for older jars, never less. + // Every current jar predates this cutoff. Decline later restores the prior + // approved cutoff, while overlapping requests retain independent references. + let id = SUPPRESSION_ID_COUNTER.fetch_add(1, Ordering::Relaxed); + let at = Instant::now(); store .suppressed_hosts .entry(host.clone()) .and_modify(|suppressed| { - suppressed.at = Instant::now(); - suppressed.pending += 1; + suppressed.at = suppressed.at.max(at); + suppressed.pending.push((id, at)); }) .or_insert(SuppressedHost { - at: Instant::now(), - pending: 1, + at, + approved_at: None, + pending: vec![(id, at)], }); - store.suppressed_order.retain(|h| h != &host); + store + .suppressed_order + .retain(|candidate| candidate != &host); store.suppressed_order.push_back(host.clone()); while store.suppressed_hosts.len() > MAX_SUPPRESSED_HOSTS { - // Never evict a tombstone an in-flight handoff still relies on; skip - // pending entries and only drop an idle one. - let idle = store.suppressed_order.iter().position(|host| { - store - .suppressed_hosts - .get(host) - .is_some_and(|suppressed| suppressed.pending == 0) + let removable = store.suppressed_order.iter().position(|candidate| { + store.suppressed_hosts.get(candidate).is_some_and(|entry| { + entry.pending.is_empty() + && entry + .approved_at + .is_none_or(|cutoff| !host_has_stale_handoff(&store, candidate, cutoff)) + }) }); - let Some(index) = idle else { + let Some(index) = removable else { break; }; let Some(oldest) = store.suppressed_order.remove(index) else { @@ -778,26 +870,13 @@ fn suppress_handoff_host(url: &str) -> Option { } Some(Suppression { host, - keep: false, + id, + at, + finished: false, sent: false, }) } -fn unsuppress_handoff_host(host: &str) { - let mut store = lock_handoff_cookies(); - let release = store - .suppressed_hosts - .get_mut(host) - .is_some_and(|suppressed| { - suppressed.pending = suppressed.pending.saturating_sub(1); - suppressed.pending == 0 - }); - if release { - store.suppressed_hosts.remove(host); - store.suppressed_order.retain(|h| h != host); - } -} - /// Browser UA staged with a handoff's cookies, if any. Same lookup as the /// cookies (exact URL, then same-host fallback). pub(crate) fn handoff_user_agent(url: &str, browser: &str) -> Option { @@ -915,8 +994,15 @@ pub(crate) fn discard_handoff_cookies_async(url: &str) { fn take_handoff_cookie_path(url: &str) -> Option { handoff_key(url).and_then(|key| { let mut store = lock_handoff_cookies(); - store.order.retain(|k| k != &key); - store.by_key.remove(&key).map(|e| e.path) + store.order.retain(|candidate| candidate != &key); + let path = store.by_key.remove(&key).map(|entry| entry.path); + if let Some(host) = url::Url::parse(&key) + .ok() + .and_then(|parsed| parsed.host_str().map(str::to_ascii_lowercase)) + { + prune_suppression(&mut store, &host); + } + path }) } @@ -1053,11 +1139,32 @@ async fn serve_listener_with_cap( pairing: Option>, confirm_wait: Duration, confirm_cap: usize, +) { + serve_listener_with_cap_and_timeouts( + listener, + tx, + pairing, + confirm_wait, + confirm_cap, + EXTENSION_HEADER_READ_TIMEOUT, + EXTENSION_CONNECTION_TIMEOUT, + ) + .await; +} + +async fn serve_listener_with_cap_and_timeouts( + listener: TcpListener, + tx: Sender, + pairing: Option>, + confirm_wait: Duration, + confirm_cap: usize, + header_read_timeout: Duration, + connection_timeout: Duration, ) { let port = match listener.local_addr() { - Ok(a) => a.port(), - Err(e) => { - tracing::warn!("Extension HTTP server has no local address: {e}"); + Ok(address) => address.port(), + Err(error) => { + tracing::warn!("Extension HTTP server has no local address: {error}"); return; } }; @@ -1067,8 +1174,41 @@ async fn serve_listener_with_cap( confirm_wait, confirm_slots: Arc::new(Semaphore::new(confirm_cap)), }; - if let Err(e) = axum::serve(listener, router(state, port)).await { - tracing::warn!("Extension HTTP server error: {e}"); + let app = router(state, port); + loop { + let (stream, peer) = match listener.accept().await { + Ok(connection) => connection, + Err(error) => { + tracing::warn!(%error, "Extension HTTP server accept failed"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + }; + let app = app.clone(); + tokio::spawn(async move { + let service = app + .map_request(|request: HyperRequest| request.map(axum::body::Body::new)); + let service = TowerToHyperService::new(service); + let mut builder = http1::Builder::new(); + builder + .timer(TokioTimer::default()) + .header_read_timeout(header_read_timeout) + .keep_alive(false); + match tokio::time::timeout( + connection_timeout, + builder.serve_connection(TokioIo::new(stream), service), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(error)) => { + tracing::debug!(%peer, %error, "Extension HTTP connection ended"); + } + Err(_) => { + tracing::debug!(%peer, "Extension HTTP connection timed out"); + } + } + }); } } @@ -1382,16 +1522,37 @@ mod tests { pairing: Option>, wait: Duration, confirm_cap: usize, + ) -> Server { + start_with_cap_and_timeouts( + depth, + pairing, + wait, + confirm_cap, + EXTENSION_HEADER_READ_TIMEOUT, + EXTENSION_CONNECTION_TIMEOUT, + ) + .await + } + + async fn start_with_cap_and_timeouts( + depth: usize, + pairing: Option>, + wait: Duration, + confirm_cap: usize, + header_read_timeout: Duration, + connection_timeout: Duration, ) -> Server { let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); let port = listener.local_addr().unwrap().port(); let (tx, rx) = channel(depth); - tokio::spawn(serve_listener_with_cap( + tokio::spawn(serve_listener_with_cap_and_timeouts( listener, tx, pairing, wait, confirm_cap, + header_read_timeout, + connection_timeout, )); Server { port, rx } } @@ -2235,6 +2396,117 @@ mod tests { clear_handoff_cookies(); } + #[test] + fn approved_suppression_is_reclaimed_when_the_old_jar_is_removed() { + let _g = exclusive(); + clear_handoff_cookies(); + let url = "https://reclaimed.example/file"; + let path = write_handoff_cookies(url, &[cookie("c", "old")]).expect("old jar"); + store_handoff_cookies(url, path, None); + suppress_handoff_host(url).expect("suppression").keep(); + + assert!(handoff_cookie_args(url).is_empty()); + assert!( + lock_handoff_cookies() + .suppressed_hosts + .contains_key("reclaimed.example") + ); + discard_handoff_cookies(url); + assert!( + !lock_handoff_cookies() + .suppressed_hosts + .contains_key("reclaimed.example") + ); + clear_handoff_cookies(); + } + + #[test] + fn approved_suppression_is_reclaimed_when_the_old_jar_is_replaced() { + let _g = exclusive(); + clear_handoff_cookies(); + let url = "https://replaced.example/file"; + let old_path = write_handoff_cookies(url, &[cookie("c", "old")]).expect("old jar"); + store_handoff_cookies(url, old_path.clone(), None); + suppress_handoff_host(url).expect("suppression").keep(); + + let fresh_path = write_handoff_cookies(url, &[cookie("c", "fresh")]).expect("fresh jar"); + store_handoff_cookies(url, fresh_path.clone(), None); + + assert!( + !lock_handoff_cookies() + .suppressed_hosts + .contains_key("replaced.example") + ); + let args = handoff_cookie_args(url); + assert_eq!(args.len(), 2); + assert_eq!(PathBuf::from(&args[1]), fresh_path); + assert_ne!(PathBuf::from(&args[1]), old_path); + clear_handoff_cookies(); + } + + #[test] + fn declined_suppression_restores_approved_cutoff_and_allows_newer_fallback() { + let _g = exclusive(); + clear_handoff_cookies(); + let old_url = "https://cutoff.example/old"; + let old_path = write_handoff_cookies(old_url, &[cookie("c", "old")]).expect("old jar"); + store_handoff_cookies(old_url, old_path.clone(), None); + suppress_handoff_host(old_url) + .expect("first suppression") + .keep(); + + let fresh_url = "https://cutoff.example/fresh"; + let fresh_path = + write_handoff_cookies(fresh_url, &[cookie("c", "fresh")]).expect("fresh jar"); + store_handoff_cookies(fresh_url, fresh_path.clone(), None); + let fresh_args = handoff_cookie_args(fresh_url); + assert_eq!(fresh_args.len(), 2); + assert_eq!(PathBuf::from(&fresh_args[1]), fresh_path); + + let declined = suppress_handoff_host(fresh_url).expect("second suppression"); + assert!(handoff_cookie_args(fresh_url).is_empty()); + drop(declined); + + let fresh_args = handoff_cookie_args(fresh_url); + assert_eq!(fresh_args.len(), 2); + assert_eq!(PathBuf::from(&fresh_args[1]), fresh_path); + + // Declining the newer request rolls back only its pending cutoff. The + // old exact-URL jar stays suppressed, while lookup may use the newer + // same-host jar as fallback instead of reviving the old jar. + let old_args = handoff_cookie_args(old_url); + assert_eq!(old_args.len(), 2); + assert_eq!(PathBuf::from(&old_args[1]), fresh_path); + assert_ne!(PathBuf::from(&old_args[1]), old_path); + clear_handoff_cookies(); + } + + #[test] + fn approved_suppression_entries_stay_bounded_as_old_jars_are_evicted() { + let _g = exclusive(); + clear_handoff_cookies(); + + for index in 0..(MAX_SUPPRESSED_HOSTS * 2) { + let url = format!("https://host-{index}.example.com/file"); + let path = write_handoff_cookies(&url, &[cookie("c", "v")]).expect("jar"); + store_handoff_cookies(&url, path, None); + suppress_handoff_host(&url) + .expect("staged jar should be suppressed") + .keep(); + + let store = lock_handoff_cookies(); + assert!( + store.suppressed_hosts.len() <= MAX_HANDOFF_ENTRIES, + "suppression entries without a stale staged jar should be reclaimed" + ); + assert!( + store.suppressed_hosts.len() <= MAX_SUPPRESSED_HOSTS, + "tombstone map must remain within its documented cap" + ); + } + clear_handoff_cookies(); + } + #[test] fn cookieless_suppression_ignores_hosts_without_a_staged_jar() { let _g = exclusive(); @@ -2949,6 +3221,59 @@ mod tests { assert_eq!(status(&task.await.unwrap()), "409"); } + #[tokio::test] + async fn header_read_timeout_closes_an_incomplete_request() { + let server = start_with_cap_and_timeouts( + 4, + test_pairing(), + Duration::from_secs(5), + MAX_PENDING_CONFIRMATIONS, + Duration::from_millis(50), + Duration::from_secs(5), + ) + .await; + let mut stream = tokio::net::TcpStream::connect((Ipv4Addr::LOCALHOST, server.port)) + .await + .unwrap(); + stream + .write_all(b"GET /hello HTTP/1.1\r\nHost:") + .await + .unwrap(); + let mut byte = [0_u8; 1]; + let result = tokio::time::timeout(Duration::from_secs(2), stream.read(&mut byte)).await; + assert!( + matches!(result, Ok(Ok(0)) | Ok(Err(_))), + "server should close an incomplete request after the header deadline" + ); + } + + #[tokio::test] + async fn connection_timeout_closes_a_stalled_request_body() { + let server = start_with_cap_and_timeouts( + 4, + test_pairing(), + Duration::from_secs(5), + MAX_PENDING_CONFIRMATIONS, + Duration::from_secs(5), + Duration::from_millis(50), + ) + .await; + let host = format!("127.0.0.1:{}", server.port); + let request = format!( + "POST /add HTTP/1.1\r\nHost: {host}\r\nOrigin: chrome-extension://abc123\r\nContent-Type: application/json\r\nContent-Length: 100\r\nConnection: close\r\n\r\n{{\"url\":" + ); + let mut stream = tokio::net::TcpStream::connect((Ipv4Addr::LOCALHOST, server.port)) + .await + .unwrap(); + stream.write_all(request.as_bytes()).await.unwrap(); + let mut byte = [0_u8; 1]; + let result = tokio::time::timeout(Duration::from_secs(2), stream.read(&mut byte)).await; + assert!( + matches!(result, Ok(Ok(0)) | Ok(Err(_))), + "server should close a request whose body stalls past the connection deadline" + ); + } + #[tokio::test] async fn confirm_times_out_when_nobody_answers() { let mut srv = start_with(4, test_pairing(), Duration::from_millis(50)).await; diff --git a/src/video_tools.rs b/src/video_tools.rs index a0f0ba99c..744514986 100644 --- a/src/video_tools.rs +++ b/src/video_tools.rs @@ -830,14 +830,37 @@ async fn fetch_ytdlp_inner( })?; tokio::fs::rename(part, dest) .await - .map_err(|e| VideoError::install(format!("couldn't install {}: {e}", dest.display()))) + .map_err(|e| VideoError::install(format!("couldn't install {}: {e}", dest.display())))?; + sync_install_directory(dest).await; + Ok(()) +} + +/// Best-effort directory sync after atomically installing a managed tool. +/// A failed sync is logged without reporting installation failure because the +/// rename has already succeeded and the executable is in place. +async fn sync_install_directory(path: &Path) { + let Some(parent) = path.parent() else { + tracing::warn!(path = %path.display(), "installed tool has no parent directory to sync"); + return; + }; + let result = match tokio::fs::File::open(parent).await { + Ok(directory) => directory.sync_all().await, + Err(error) => Err(error), + }; + if let Err(error) = result { + tracing::warn!( + path = %parent.display(), + %error, + "installed tool, but could not sync its directory entry" + ); + } } /// Stable Deno Linux archives are published by denoland/deno with per-asset SHA-256 digests. const DENO_MAX_ARCHIVE_BYTES: u64 = 64 * 1024 * 1024; const DENO_MAX_BINARY_BYTES: u64 = 256 * 1024 * 1024; const DENO_LATEST_RELEASE_API: &str = "https://api.github.com/repos/denoland/deno/releases/latest"; -const DENO_RELEASE_METADATA_MAX_BYTES: usize = 1024 * 1024; +const RELEASE_METADATA_MAX_BYTES: usize = 1024 * 1024; const DENO_MIN_VERSION: [u32; 3] = [2, 3, 0]; /// Candidate validated by ensure_deno, kept in sync with yt-dlp's runtime path. @@ -863,7 +886,7 @@ fn selected_deno_path() -> Option { } #[derive(Clone, Debug, PartialEq, Eq)] -struct DenoRelease { +struct ReleaseAsset { tag: String, download_url: String, sha256: Option, @@ -948,21 +971,26 @@ pub(crate) async fn bounded_response_text( String::from_utf8(body).ok() } -fn parse_deno_release_metadata(body: &str, arch: &str) -> Option { +/// Parse release metadata only when its tag and selected asset URL match the +/// expected upstream release shape. An absent or invalid digest is preserved as +/// None, so installation fails closed while release checks can show the tag. +fn parse_pinned_release_asset( + body: &str, + asset_name: &str, + expected_url_for_tag: impl FnOnce(&str) -> Option, +) -> Option { let release: serde_json::Value = serde_json::from_str(body).ok()?; let tag = release.get("tag_name")?.as_str()?.to_owned(); if !valid_release_tag(&tag) { return None; } - let expected_url = deno_download_url_for_arch(&tag, arch)?; - let asset_name = deno_asset_name_for_arch(arch)?; + let expected_url = expected_url_for_tag(&tag)?; let asset = release.get("assets")?.as_array()?.iter().find(|asset| { asset.get("name").and_then(serde_json::Value::as_str) == Some(asset_name) })?; let download_url = asset.get("browser_download_url")?.as_str()?.to_owned(); - // Only accept the expected upstream asset URL; never fetch a URL supplied - // by arbitrary release metadata, even when the digest field is present. + // Never fetch a URL supplied by arbitrary release metadata. if download_url != expected_url { return None; } @@ -972,14 +1000,33 @@ fn parse_deno_release_metadata(body: &str, arch: &str) -> Option { .and_then(|value| value.strip_prefix("sha256:")) .filter(|value| valid_sha256_digest(value)) .map(str::to_owned); - Some(DenoRelease { + Some(ReleaseAsset { tag, download_url, sha256, }) } -async fn latest_deno_release() -> Option { +fn parse_deno_release_metadata(body: &str, arch: &str) -> Option { + let asset_name = deno_asset_name_for_arch(arch)?; + parse_pinned_release_asset(body, asset_name, |tag| { + deno_download_url_for_arch(tag, arch) + }) +} + +fn ytdlp_download_url(tag: &str) -> Option { + valid_release_tag(tag).then(|| { + format!( + "https://github.com/yt-dlp/yt-dlp-nightly-builds/releases/download/{tag}/yt-dlp_linux" + ) + }) +} + +fn parse_ytdlp_release_metadata(body: &str) -> Option { + parse_pinned_release_asset(body, "yt-dlp_linux", ytdlp_download_url) +} + +async fn latest_deno_release() -> Option { let handle = crate::runtime::tokio_rt().spawn(async move { let client = tool_client_builder_proxied() .ok()? @@ -994,7 +1041,7 @@ async fn latest_deno_release() -> Option { .ok()? .error_for_status() .ok()?; - let body = bounded_response_text(response, DENO_RELEASE_METADATA_MAX_BYTES).await?; + let body = bounded_response_text(response, RELEASE_METADATA_MAX_BYTES).await?; parse_deno_release_metadata(&body, std::env::consts::ARCH) }); handle.await.ok().flatten() @@ -1157,24 +1204,8 @@ async fn install_deno_binary(dir: PathBuf) -> Result { part.display() ))); } - match std::fs::symlink_metadata(&dest) { - Ok(metadata) if metadata.file_type().is_symlink() => { - std::fs::remove_file(&part).ok(); - return Err(VideoError::install(format!( - "refusing to replace Deno through symlink {}", - dest.display() - ))); - } - Ok(_) => {} - Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} - Err(error) => { - std::fs::remove_file(&part).ok(); - return Err(VideoError::install(format!( - "couldn't inspect {}: {error}", - dest.display() - ))); - } - } + // Rename replaces the destination directory entry atomically; it does + // not follow a symlink at dest, so a pre-check would be a TOCTOU window. if let Err(error) = tokio::fs::rename(&part, &dest).await { std::fs::remove_file(&part).ok(); return Err(VideoError::install(format!( @@ -1182,6 +1213,7 @@ async fn install_deno_binary(dir: PathBuf) -> Result { dest.display() ))); } + sync_install_directory(&dest).await; Ok(dest) } @@ -2035,39 +2067,43 @@ pub(crate) fn ffmpeg_location_dir(ffmpeg_bin: &Path) -> String { .into_owned() } -/// Latest nightly yt-dlp download URL and its SHA-256 digest. -/// Fail closed: `None` digest means no verified install. -/// The nightly builds publish `yt-dlp_linux` with a `digest: "sha256:…"` field. -pub async fn latest_ytdlp_asset() -> Option<(String, Option)> { +const YTDLP_LATEST_RELEASE_API: &str = + "https://api.github.com/repos/yt-dlp/yt-dlp-nightly-builds/releases/latest"; + +/// Fetch and validate the latest yt-dlp release through the configured app proxy. +/// Metadata is size-bounded like Deno's, and the parser pins the asset URL. +async fn latest_ytdlp_release() -> Option { let handle = crate::runtime::tokio_rt().spawn(async move { - let fetcher = - yt_dlp::client::deps::github::GitHubFetcher::new("yt-dlp", "yt-dlp-nightly-builds"); - let release = fetcher.fetch_latest_release(None).await.ok()?; - let asset = release.assets.iter().find(|a| a.name == "yt-dlp_linux")?; - let digest = asset - .digest - .as_deref() - .and_then(|d| d.strip_prefix("sha256:")) - .map(str::to_owned); - Some((asset.download_url.clone(), digest)) + let client = tool_client_builder_proxied() + .ok()? + .user_agent(concat!("Grab/", env!("CARGO_PKG_VERSION"))) + .build() + .ok()?; + let response = client + .get(YTDLP_LATEST_RELEASE_API) + .header(reqwest::header::ACCEPT, "application/vnd.github+json") + .send() + .await + .ok()? + .error_for_status() + .ok()?; + let body = bounded_response_text(response, RELEASE_METADATA_MAX_BYTES).await?; + parse_ytdlp_release_metadata(&body) }); handle.await.ok().flatten() } -/// Latest nightly yt-dlp tag without downloading anything: one user-initiated -/// GitHub API call. `None` on any network/API failure — the row then reports the -/// check failed instead of prompting. +/// Latest nightly yt-dlp download URL and its SHA-256 digest. +/// An absent or invalid digest means installation fails closed. +pub async fn latest_ytdlp_asset() -> Option<(String, Option)> { + latest_ytdlp_release() + .await + .map(|release| (release.download_url, release.sha256)) +} + +/// Latest nightly yt-dlp tag without downloading the asset. pub async fn latest_ytdlp_tag() -> Option { - let handle = crate::runtime::tokio_rt().spawn(async move { - let fetcher = - yt_dlp::client::deps::github::GitHubFetcher::new("yt-dlp", "yt-dlp-nightly-builds"); - fetcher - .fetch_latest_release(None) - .await - .ok() - .map(|release| release.tag_name) - }); - handle.await.ok().flatten() + latest_ytdlp_release().await.map(|release| release.tag) } /// Whether a failed yt-dlp log shows the extractor breaking, which a newer @@ -2859,7 +2895,7 @@ mod tests { } #[cfg(test)] -mod deno_release_tests { +mod release_metadata_tests { use super::*; fn unique_dir(tag: &str) -> PathBuf { @@ -3002,6 +3038,33 @@ mod deno_release_tests { assert!(parse_deno_release_metadata(&body, "unsupported").is_none()); } + #[test] + fn ytdlp_release_metadata_pins_tag_asset_url_and_digest() { + let tag = "2026.10.28.123456"; + let digest = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + let url = ytdlp_download_url(tag).expect("valid tag"); + let body = format!( + r#"{{"tag_name":"{tag}","assets":[{{"name":"yt-dlp_linux","browser_download_url":"{url}","digest":"sha256:{digest}"}}]}}"# + ); + + let release = parse_ytdlp_release_metadata(&body).expect("valid release metadata"); + assert_eq!(release.tag, tag); + assert_eq!(release.download_url, url); + assert_eq!(release.sha256.as_deref(), Some(digest)); + + let bad_url = body.replace(&url, "https://attacker.example/yt-dlp_linux"); + assert!(parse_ytdlp_release_metadata(&bad_url).is_none()); + let bad_tag = body.replace(tag, "../main"); + assert!(parse_ytdlp_release_metadata(&bad_tag).is_none()); + let bad_digest = body.replace(digest, "abc123"); + assert_eq!( + parse_ytdlp_release_metadata(&bad_digest) + .expect("asset URL remains valid") + .sha256, + None + ); + } + #[test] fn archive_extraction_only_accepts_the_expected_binary() { use std::io::Write as _;