From 973bb7bbc6cb2afb0f6b367840204a911243155e Mon Sep 17 00:00:00 2001 From: Test User Date: Tue, 21 Jul 2026 18:18:40 +0200 Subject: [PATCH] Fetch streams in bounded windows and restore yt-dlp for stream URLs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit YouTube caps tokenless stream URLs at exactly their leading 1 MiB: plain, open-ended, and oversized requests get 403, and fresh URLs refuse offset starts, so playback died mid-first-minute. PO tokens would lift the cap but the token-capable Innertube clients need signature deciphering that is broken in rustypipe upstream (botguard was built and tested — ineffective through the iOS client). The player now streams every http(s) source through a windowed SourceStream (bounded ~1 MiB ranges, 200-body fallback, eager seek/reconnect so rejected windows fail typed instead of retrying forever) and the capture downloader windows the same way. Stream URLs come from a minimal yt-dlp sidecar again — metadata stays on the pure-Rust rustypipe extractor — whose cipher-solved URLs stream whole files at a throttled ~32 KB/s; a missing binary degrades to 1 MiB streams with a warning. botguard_bin is wired through so streams flip back to pure Rust when upstream deciphering recovers. Live-verified on the exact track from the failure log. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 3 + Cargo.toml | 1 + architecture/youtube-rustypipe.md | 50 ++- audio-player/Cargo.toml | 3 + audio-player/src/lib.rs | 1 + audio-player/src/player_engine.rs | 17 +- audio-player/src/windowed_http.rs | 547 ++++++++++++++++++++++++++++ crabidy-server/src/capture.rs | 132 ++++++- crabidy-server/src/capture_store.rs | 93 +++++ devenv.nix | 4 + plan/summary.md | 30 ++ plan/youtube-rustypipe.md | 9 + quality/youtube-rustypipe.md | 20 + ytdy/Cargo.toml | 1 + ytdy/src/extract.rs | 18 +- ytdy/src/lib.rs | 149 +++++++- ytdy/src/ytdlp.rs | 201 ++++++++++ 17 files changed, 1240 insertions(+), 39 deletions(-) create mode 100644 audio-player/src/windowed_http.rs create mode 100644 ytdy/src/ytdlp.rs diff --git a/Cargo.lock b/Cargo.lock index 00b0ae4..2fd3e83 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -303,7 +303,10 @@ name = "audio-player" version = "0.1.0" dependencies = [ "anyhow", + "bytes", "flume", + "futures", + "reqwest 0.13.1", "rodio", "stream-download", "thiserror 2.0.19", diff --git a/Cargo.toml b/Cargo.toml index 469aac0..a260750 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ edition = "2021" anyhow = "1" async-trait = "0.1" base64 = "0.22" +bytes = "1" chrono = { version = "0.4", default-features = false, features = ["clock"] } clap = { version = "4", features = ["derive"] } clap-serde-derive = "0.2" diff --git a/architecture/youtube-rustypipe.md b/architecture/youtube-rustypipe.md index 6eec1b2..09bea09 100644 --- a/architecture/youtube-rustypipe.md +++ b/architecture/youtube-rustypipe.md @@ -1,4 +1,9 @@ -# YouTube provider on rustypipe (pure Rust) +# YouTube provider on rustypipe + +> **Amended the same day** by D2-revised below: pure-Rust stream +> *fetching* turned out to be blocked by YouTube's PO-token +> enforcement, so stream URLs go through a minimal `yt-dlp` sidecar +> again while all metadata stays on rustypipe. ## Context and problem statement @@ -74,6 +79,49 @@ likely not decode locally, but the URL is still honest — e.g. a future player may cope). Download captures inherit the same choice, so their audio is `.m4a`. +## D2-revised — Stream *fetching* (2026-07-21, second round) + +Real use immediately hit YouTube's tokenless-fetch enforcement. Live +findings against googlevideo (all reproduced, same day): + +- Every rustypipe (iOS-client) stream URL serves **exactly its leading + ~1 MiB**, measured to the byte (403 at cumulative 1 048 576): plain + GETs 403, open-ended ranges 403, bounded ranges work only until the + budget is spent, and a **fresh URL refuses to start at an offset** — + so chaining fresh URLs per window is impossible. +- PO tokens would lift the cap, but rustypipe attaches them only to + web-family clients, whose **signature deciphering is broken upstream + right now** ("could not extract sig fn name", also on git master) — + `rustypipe-botguard` was built and tested; ineffective through the + iOS client (no `pot` parameter). +- `yt-dlp` (2026.07.04) still solves the cipher challenges; its URLs + accept plain and ranged GETs for the whole file, **throttled to + ~32 KB/s** — twice the itag-140 audio bitrate, so playback holds and + captures are merely slow. Its TV/Android clients are SABR-blocked + (no URLs at all), so this is the state of the art everywhere. + +**Decisions:** + +1. **Windowed HTTP fetching everywhere.** The player streams through a + `WindowedHttpStream` (audio-player, a `stream-download` + `SourceStream`) and the capture `Downloader` downloads in the same + bounded ~1 MiB `Range` windows — the request pattern real players + produce, correct for every provider (servers that ignore `Range` + degrade to one 200 body; rejected windows fail typed, never loop). +2. **`yt-dlp` returns as a stream-URL-only sidecar.** All metadata + (search, playlists, details) stays on the pure-Rust rustypipe + extractor; only `get_urls_for_track` consults `yt-dlp` (argv-only, + bounded, stderr-summarized). A missing binary degrades with a + warning — streams then stop after their first ~1 MiB instead of + failing entirely. +3. **The pure-Rust path stays wired.** `botguard_bin` is configurable + (and PATH-auto-detected by rustypipe); when upstream deciphering + recovers, PO-token'd rustypipe URLs make the sidecar unnecessary + without code changes beyond removing the fallback preference. + +The per-track capture deadline was raised to 30 minutes — at 32 KB/s a +long track legitimately takes that. + ## D3 — Login and playlists The `cookies` setting keeps its meaning: a path to a Netscape diff --git a/audio-player/Cargo.toml b/audio-player/Cargo.toml index 3e233f0..9f829af 100644 --- a/audio-player/Cargo.toml +++ b/audio-player/Cargo.toml @@ -5,7 +5,10 @@ edition.workspace = true [dependencies] anyhow.workspace = true +bytes.workspace = true flume.workspace = true +futures.workspace = true +reqwest.workspace = true rodio.workspace = true stream-download.workspace = true thiserror.workspace = true diff --git a/audio-player/src/lib.rs b/audio-player/src/lib.rs index de163ba..65943ad 100644 --- a/audio-player/src/lib.rs +++ b/audio-player/src/lib.rs @@ -1,5 +1,6 @@ mod player; mod player_engine; +pub mod windowed_http; pub use player::{Player, PlayerError}; pub use player_engine::{MediaInfo, PlayerMessage}; diff --git a/audio-player/src/player_engine.rs b/audio-player/src/player_engine.rs index 113c6de..00951f4 100644 --- a/audio-player/src/player_engine.rs +++ b/audio-player/src/player_engine.rs @@ -10,6 +10,8 @@ use rodio::stream::{DeviceSinkBuilder, MixerDeviceSink}; use rodio::{Decoder, Source}; use stream_download::storage::temp::TempStorageProvider; use stream_download::{Settings, StreamDownload}; + +use crate::windowed_http::{WindowedHttpParams, WindowedHttpStream}; use thiserror::Error; use tracing::{debug, info, instrument, trace, warn}; use url::Url; @@ -78,6 +80,8 @@ pub struct PlayerEngine { // Present when the engine had to bring its own runtime because the // creating thread was not inside a tokio context. _owned_runtime: Option, + /// Shared client for windowed network streams. + http: reqwest::Client, } impl PlayerEngine { @@ -101,6 +105,10 @@ impl PlayerEngine { } }; info!("audio output device opened"); + let http = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(30)) + .build() + .context("failed to build the http client")?; Ok(Self { current_source: None, media_info: None, @@ -111,6 +119,7 @@ impl PlayerEngine { tx_player, runtime, _owned_runtime: owned_runtime, + http, }) } @@ -195,11 +204,15 @@ impl PlayerEngine { let duration = match Url::parse(source_str) { Ok(url) if matches!(url.scheme(), "http" | "https") => { trace!(%url, "opening network stream"); + // Windowed fetching: some CDNs (googlevideo) reject plain + // and open-ended requests with 403 and only serve bounded + // ranges (see audio-player/src/windowed_http.rs). + let params = WindowedHttpParams::new(url.clone(), self.http.clone()); let reader = self.runtime.block_on(async { tokio::time::timeout( STREAM_OPEN_TIMEOUT, - StreamDownload::new_http( - url.clone(), + StreamDownload::new::( + params, TempStorageProvider::new(), Settings::default(), ), diff --git a/audio-player/src/windowed_http.rs b/audio-player/src/windowed_http.rs new file mode 100644 index 0000000..7919980 --- /dev/null +++ b/audio-player/src/windowed_http.rs @@ -0,0 +1,547 @@ +//! A windowed HTTP [`SourceStream`]: fetches media in small **bounded** +//! `Range` requests instead of one open-ended GET. +//! +//! Some CDNs (notably googlevideo, see +//! `architecture/youtube-rustypipe.md`) reject plain and open-ended +//! requests from unattested clients with `403 Forbidden` and only serve +//! bounded ranges of about a megabyte — the request pattern real players +//! produce. This stream chains such windows transparently; servers that +//! ignore the `Range` header (plain `200`) degrade to one continuous +//! body without windowing (and without seek support). +//! +//! Error messages never include the URL — stream URLs may embed tokens. + +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; + +use bytes::Bytes; +use futures::{Future, Stream}; +use stream_download::source::{DecodeError, SourceStream}; +use tracing::{debug, trace, warn}; +use url::Url; + +/// Bytes per request window. Verified against googlevideo: 1 MiB windows +/// are served, 8 MiB and open-ended requests are rejected. +pub const WINDOW_SIZE: u64 = 1024 * 1024; + +/// Parameters for [`WindowedHttpStream::create`]. +#[derive(Clone, Debug)] +pub struct WindowedHttpParams { + pub url: Url, + pub client: reqwest::Client, + /// Window size in bytes; [`WINDOW_SIZE`] outside of tests. + pub window: u64, +} + +impl WindowedHttpParams { + pub fn new(url: Url, client: reqwest::Client) -> Self { + Self { + url, + client, + window: WINDOW_SIZE, + } + } +} + +/// Error creating the stream (first window request failed). +#[derive(Debug, thiserror::Error)] +#[error("{0}")] +pub struct WindowedHttpError(String); + +impl DecodeError for WindowedHttpError {} + +type BytesStream = Pin> + Send + Sync>>; +type WindowFuture = Pin> + Send + Sync>>; + +/// One server response being drained. +struct Window { + stream: BytesStream, + /// Absolute end (exclusive) of the bytes this response carries; + /// `u64::MAX` for an un-ranged whole-body response. + end: u64, + /// Total resource size, when the response revealed it. + total: Option, + /// The server honored the `Range` header (`206`). + ranged: bool, +} + +enum State { + /// Draining the current response. + Streaming { + stream: BytesStream, + end: u64, + }, + /// Waiting for the next window request. + Requesting(WindowFuture), + Finished, +} + +/// See the module docs. +pub struct WindowedHttpStream { + // (Debug impl below — the state machine holds unnameable futures.) + client: reqwest::Client, + url: Url, + window: u64, + /// Total resource size (from `Content-Range`); `None` when unknown. + content_length: Option, + /// The server honors ranges — windowing and seeking are available. + ranged: bool, + /// Absolute offset of the next byte to hand out. + position: u64, + /// Exclusive end requested via `seek_range`; `None` = to the end. + limit: Option, + /// Bytes to silently drop before yielding (un-ranged reconnect + /// catch-up). + discard: u64, + state: State, +} + +/// Requests one bounded window `[start, end_exclusive)`. +/// +/// `206` yields a ranged window (end and total parsed from +/// `Content-Range`, falling back to `Content-Length`); `200` means the +/// server ignored the header and sent the whole body; `416` past the end +/// yields an empty terminal window. Anything else is an error carrying +/// the status only. +async fn request_window( + client: reqwest::Client, + url: Url, + start: u64, + end_exclusive: u64, +) -> io::Result { + let range = format!("bytes={start}-{}", end_exclusive.saturating_sub(1)); + trace!(range, "requesting window"); + let response = client + .get(url) + .header(reqwest::header::RANGE, range) + .send() + .await + .map_err(|err| io::Error::other(format!("window request failed: {}", err.without_url())))?; + match response.status() { + reqwest::StatusCode::PARTIAL_CONTENT => { + // `Content-Range: bytes -/` + let content_range = response + .headers() + .get(reqwest::header::CONTENT_RANGE) + .and_then(|value| value.to_str().ok()) + .map(str::to_string); + let (end, total) = match content_range.as_deref().and_then(parse_content_range) { + Some((_, range_end, total)) => (range_end + 1, total), + None => { + // No usable Content-Range: derive the window end from + // the body length. + let len = response.content_length().unwrap_or(0); + (start + len, None) + } + }; + Ok(Window { + stream: Box::pin(response.bytes_stream()), + end, + total, + ranged: true, + }) + } + reqwest::StatusCode::OK => { + let total = response.content_length(); + Ok(Window { + stream: Box::pin(response.bytes_stream()), + end: u64::MAX, + total, + ranged: false, + }) + } + reqwest::StatusCode::RANGE_NOT_SATISFIABLE => Ok(Window { + stream: Box::pin(futures::stream::empty()), + end: start, + total: None, + ranged: true, + }), + status => Err(io::Error::other(format!( + "window request rejected: {status}" + ))), + } +} + +/// Parses `bytes -/` into `(start, end, total)`. +fn parse_content_range(value: &str) -> Option<(u64, u64, Option)> { + let rest = value.trim().strip_prefix("bytes ")?; + let (range, total) = rest.split_once('/')?; + let (start, end) = range.split_once('-')?; + let total = match total.trim() { + "*" => None, + n => Some(n.parse().ok()?), + }; + Some((start.trim().parse().ok()?, end.trim().parse().ok()?, total)) +} + +impl std::fmt::Debug for WindowedHttpStream { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WindowedHttpStream") + .field("content_length", &self.content_length) + .field("ranged", &self.ranged) + .field("position", &self.position) + .finish_non_exhaustive() + } +} + +impl WindowedHttpStream { + /// The exclusive end the consumer currently wants: the seek limit + /// clamped to the known size. + fn effective_end(&self) -> Option { + match (self.limit, self.content_length) { + (Some(limit), Some(len)) => Some(limit.min(len)), + (Some(limit), None) => Some(limit), + (None, len) => len, + } + } + + /// Schedules the request for the window starting at `start`, or + /// finishes when nothing is left to fetch. + fn schedule_window(&mut self, start: u64) { + let end = match self.window_end(start) { + Some(end) => end, + None => { + self.state = State::Finished; + return; + } + }; + self.state = State::Requesting(Box::pin(request_window( + self.client.clone(), + self.url.clone(), + start, + end, + ))); + } + + /// The exclusive end of the window starting at `start`; `None` when + /// nothing is left to fetch. + fn window_end(&self, start: u64) -> Option { + match self.effective_end() { + Some(effective) if start >= effective => None, + Some(effective) => Some((start + self.window).min(effective)), + None => Some(start + self.window), + } + } + + /// Eagerly opens the window at `start` (seeks and reconnects), so + /// request failures surface to the caller instead of re-arising on + /// every poll. + async fn open_window(&mut self, start: u64) -> io::Result<()> { + match self.window_end(start) { + None => { + self.state = State::Finished; + Ok(()) + } + Some(end) => { + let window = + request_window(self.client.clone(), self.url.clone(), start, end).await?; + self.state = State::Streaming { + stream: window.stream, + end: window.end, + }; + Ok(()) + } + } + } +} + +impl Stream for WindowedHttpStream { + type Item = io::Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = &mut *self; + loop { + match &mut this.state { + State::Finished => return Poll::Ready(None), + State::Requesting(future) => match future.as_mut().poll(cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Err(err)) => { + this.state = State::Finished; + return Poll::Ready(Some(Err(err))); + } + Poll::Ready(Ok(window)) => { + this.state = State::Streaming { + stream: window.stream, + end: window.end, + }; + } + }, + State::Streaming { stream, end } => match stream.as_mut().poll_next(cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Some(Err(err))) => { + return Poll::Ready(Some(Err(io::Error::other(format!( + "stream body failed: {}", + err.without_url() + ))))); + } + Poll::Ready(Some(Ok(mut bytes))) => { + // Un-ranged reconnect catch-up: drop the prefix the + // consumer already has. + if this.discard > 0 { + let drop_now = this.discard.min(bytes.len() as u64) as usize; + let _ = bytes.split_to(drop_now); + this.discard -= drop_now as u64; + if bytes.is_empty() { + continue; + } + } + this.position += bytes.len() as u64; + return Poll::Ready(Some(Ok(bytes))); + } + Poll::Ready(None) => { + let window_end = *end; + if !this.ranged { + // One whole-body response: its end is the end. + this.state = State::Finished; + return Poll::Ready(None); + } + if let Some(effective) = this.effective_end() { + if this.position >= effective { + this.state = State::Finished; + return Poll::Ready(None); + } + } + if this.position < window_end && this.content_length.is_none() { + // A short window with no known total: the + // resource ended early. + this.state = State::Finished; + return Poll::Ready(None); + } + let next = this.position; + this.schedule_window(next); + } + }, + } + } + } +} + +impl SourceStream for WindowedHttpStream { + type Params = WindowedHttpParams; + type StreamCreationError = WindowedHttpError; + + async fn create(params: Self::Params) -> Result { + let window = request_window(params.client.clone(), params.url.clone(), 0, params.window) + .await + .map_err(|err| WindowedHttpError(err.to_string()))?; + let content_length = window.total; + let ranged = window.ranged; + debug!(?content_length, ranged, "windowed http stream open"); + Ok(Self { + client: params.client, + url: params.url, + window: params.window, + content_length, + ranged, + position: 0, + limit: None, + discard: 0, + state: State::Streaming { + stream: window.stream, + end: window.end, + }, + }) + } + + fn content_length(&self) -> Option { + self.content_length + } + + async fn seek_range(&mut self, start: u64, end: Option) -> io::Result<()> { + trace!(start, ?end, "seek"); + self.position = start; + self.limit = end; + self.discard = 0; + // Eager: the request happens *here*, so a rejected window is an + // error the retry logic can time out on — a lazily scheduled + // request that keeps failing would look like a successful + // reconnect every time and retry forever. + self.open_window(start).await + } + + async fn reconnect(&mut self, current_position: u64) -> io::Result<()> { + if self.ranged { + self.position = current_position; + self.discard = 0; + return self.open_window(current_position).await; + } + // The server does not honor ranges: refetch from the start and + // drop what the consumer already has. + warn!( + current_position, + "reconnecting to a server without range support" + ); + self.position = current_position; + self.discard = current_position; + let window = request_window(self.client.clone(), self.url.clone(), 0, u64::MAX).await?; + self.state = State::Streaming { + stream: window.stream, + end: window.end, + }; + Ok(()) + } + + fn supports_seek(&self) -> bool { + self.ranged + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::StreamExt; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// A minimal HTTP server for one fixed body. With `ranged`, bounded + /// `Range` requests get `206` + `Content-Range` slices — and, like + /// googlevideo, open-ended or oversized ranges get `403`. Without, + /// every request gets the whole body as `200`. + async fn serve(body: Vec, ranged: bool, max_window: u64) -> Url { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind test server"); + let addr = listener.local_addr().expect("addr"); + tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let body = body.clone(); + tokio::spawn(async move { + let mut buf = Vec::new(); + let mut chunk = [0u8; 1024]; + loop { + match sock.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(n) => { + buf.extend_from_slice(&chunk[..n]); + if buf.windows(4).any(|w| w == b"\r\n\r\n") { + break; + } + } + } + } + let request = String::from_utf8_lossy(&buf).to_lowercase(); + let range = request + .lines() + .find_map(|line| line.strip_prefix("range: bytes=")) + .and_then(|spec| { + let (start, end) = spec.trim().split_once('-')?; + let start: u64 = start.parse().ok()?; + let end: Option = end.parse().ok(); + Some((start, end)) + }); + let total = body.len() as u64; + let response = match (ranged, range) { + (true, Some((start, Some(end)))) + if start < total && end - start < max_window => + { + let end = end.min(total - 1); + let slice = &body[start as usize..=end as usize]; + let mut head = format!( + "HTTP/1.1 206 Partial Content\r\nContent-Range: bytes {start}-{end}/{total}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + slice.len() + ) + .into_bytes(); + head.extend_from_slice(slice); + head + } + (true, Some((start, _))) if start >= total => { + format!("HTTP/1.1 416 Range Not Satisfiable\r\nContent-Range: bytes */{total}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .into_bytes() + } + (true, _) => { + // Open-ended or oversized: rejected, like + // googlevideo without a PO token. + b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_vec() + } + (false, _) => { + let mut head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {total}\r\nConnection: close\r\n\r\n" + ) + .into_bytes(); + head.extend_from_slice(&body); + head + } + }; + let _ = sock.write_all(&response).await; + let _ = sock.shutdown().await; + }); + } + }); + Url::parse(&format!("http://{addr}/stream")).expect("url") + } + + fn params(url: Url, window: u64) -> WindowedHttpParams { + WindowedHttpParams { + url, + client: reqwest::Client::new(), + window, + } + } + + async fn read_all(stream: &mut WindowedHttpStream) -> Vec { + let mut out = Vec::new(); + while let Some(chunk) = stream.next().await { + out.extend_from_slice(&chunk.expect("chunk")); + } + out + } + + #[tokio::test] + async fn chains_windows_over_a_range_only_server() { + // 10 windows of 16 bytes against a server that 403s anything + // bigger — exactly the googlevideo behavior. + let body: Vec = (0..160u32).map(|i| i as u8).collect(); + let url = serve(body.clone(), true, 64).await; + let mut stream = WindowedHttpStream::create(params(url, 16)) + .await + .expect("create"); + assert_eq!(stream.content_length(), Some(160)); + assert!(stream.supports_seek()); + assert_eq!(read_all(&mut stream).await, body); + } + + #[tokio::test] + async fn seeks_restart_the_window_chain() { + let body: Vec = (0..160u32).map(|i| i as u8).collect(); + let url = serve(body.clone(), true, 64).await; + let mut stream = WindowedHttpStream::create(params(url, 16)) + .await + .expect("create"); + stream.seek_range(100, None).await.expect("seek"); + assert_eq!(read_all(&mut stream).await, body[100..]); + } + + #[tokio::test] + async fn plain_servers_stream_one_body_without_seek() { + let body: Vec = (0..100u32).map(|i| i as u8).collect(); + let url = serve(body.clone(), false, 0).await; + let mut stream = WindowedHttpStream::create(params(url, 16)) + .await + .expect("create"); + assert_eq!(stream.content_length(), Some(100)); + assert!(!stream.supports_seek()); + assert_eq!(read_all(&mut stream).await, body); + } + + #[tokio::test] + async fn rejections_surface_as_errors_without_the_url() { + // A ranged server with max_window 0 rejects everything. + let url = serve(vec![0; 10], true, 0).await; + let err = WindowedHttpStream::create(params(url, 16)) + .await + .expect_err("403 fails creation"); + let message = err.to_string(); + assert!(message.contains("403"), "{message}"); + assert!(!message.contains("127.0.0.1"), "no url: {message}"); + } + + #[test] + fn content_range_parses_totals_and_wildcards() { + assert_eq!( + parse_content_range("bytes 0-1023/7831134"), + Some((0, 1023, Some(7831134))) + ); + assert_eq!(parse_content_range("bytes 5-9/*"), Some((5, 9, None))); + assert_eq!(parse_content_range("garbage"), None); + } +} diff --git a/crabidy-server/src/capture.rs b/crabidy-server/src/capture.rs index 4e4d225..dea177d 100644 --- a/crabidy-server/src/capture.rs +++ b/crabidy-server/src/capture.rs @@ -27,9 +27,19 @@ use tracing::warn; /// Connect timeout for download requests. pub const DOWNLOAD_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); +/// Bytes per download request window. Some CDNs (googlevideo) reject +/// plain and open-ended requests from unattested clients with 403 and +/// only serve bounded ranges around this size — the same windowing the +/// player uses (audio-player/src/windowed_http.rs). Servers that ignore +/// the `Range` header answer 200 with the whole body, which is handled +/// as a single window. +pub const DOWNLOAD_WINDOW: u64 = 1024 * 1024; + /// Total per-track deadline: URL fetch, request, and streaming the whole /// body. A stalled transfer aborts the capture instead of hanging it. -pub const DOWNLOAD_TRACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(600); +/// Generous: tokenless YouTube URLs are throttled to ~32 KB/s, so a long +/// track legitimately takes many minutes. +pub const DOWNLOAD_TRACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1800); /// Size limits for one capture. The walk aborts with /// [`CaptureError::TooLarge`] when a limit trips — a runaway provider tree @@ -209,21 +219,29 @@ struct TrackEntry { /// /// One shared HTTP client with a connect timeout; each track is bounded by /// [`DOWNLOAD_TRACK_TIMEOUT`] end to end and never retried (a capture is -/// re-runnable; resuming re-attempts what is missing). Bodies are streamed -/// to disk against the capture's remaining byte budget. +/// re-runnable; resuming re-attempts what is missing). Bodies are fetched +/// in bounded [`DOWNLOAD_WINDOW`] ranges and streamed to disk against the +/// capture's remaining byte budget. #[derive(Debug)] pub struct Downloader { http: reqwest::Client, + window: u64, } impl Downloader { /// Builds the shared HTTP client. Fails only when the TLS backend /// cannot initialize. pub fn new() -> Result { + Self::with_window(DOWNLOAD_WINDOW) + } + + /// [`Self::new`] with an explicit window size — the seam the + /// window-chaining tests use. + pub(crate) fn with_window(window: u64) -> Result { let http = reqwest::Client::builder() .connect_timeout(DOWNLOAD_CONNECT_TIMEOUT) .build()?; - Ok(Self { http }) + Ok(Self { http, window }) } /// Settles one track: audio file first, then the toml pointing at @@ -312,14 +330,30 @@ impl Downloader { let download_err = |err: reqwest::Error| { CaptureError::Download(format!("{track_path}: {}", err.without_url())) }; + // First bounded window; its status decides the mode. Some CDNs + // (googlevideo) 403 plain and open-ended requests, so every + // request carries a bounded range; servers that ignore the header + // answer 200 with the whole body. + let mut start = 0u64; let mut response = self .http .get(url) + .header( + reqwest::header::RANGE, + format!("bytes=0-{}", self.window - 1), + ) .send() .await - .map_err(download_err)? - .error_for_status() .map_err(download_err)?; + let windowed = match response.status() { + reqwest::StatusCode::PARTIAL_CONTENT => true, + reqwest::StatusCode::OK => false, + status => { + return Err(CaptureError::Download(format!( + "{track_path}: HTTP status {status}" + ))) + } + }; let content_type = response .headers() .get(reqwest::header::CONTENT_TYPE) @@ -328,13 +362,65 @@ impl Downloader { let ext = extension_for(content_type.as_deref(), url); let audio_name = audio_file_name(index, &track.title, &ext); let mut audio = tokio::fs::File::create(dir.join(&audio_name)).await?; - while let Some(chunk) = response.chunk().await.map_err(download_err)? { - let len = chunk.len() as u64; - if len > *bytes_left { - return Err(CaptureError::TooLarge("download budget exhausted")); + loop { + // The window's extent and the resource total, from + // `Content-Range: bytes -/` (Content-Length + // fallback for the extent). + let (window_end, total) = if windowed { + let content_range = response + .headers() + .get(reqwest::header::CONTENT_RANGE) + .and_then(|value| value.to_str().ok()) + .and_then(parse_content_range); + match content_range { + Some((_, range_end, total)) => (range_end + 1, total), + None => (start + response.content_length().unwrap_or(0), None), + } + } else { + (u64::MAX, None) + }; + let mut received = 0u64; + while let Some(chunk) = response.chunk().await.map_err(download_err)? { + let len = chunk.len() as u64; + if len > *bytes_left { + return Err(CaptureError::TooLarge("download budget exhausted")); + } + *bytes_left -= len; + received += len; + tokio::io::AsyncWriteExt::write_all(&mut audio, &chunk).await?; + } + if !windowed { + break; + } + start = window_end.max(start + received); + match total { + Some(total) if start >= total => break, + // A short or empty window with no known total: the + // resource ended early. + _ if received == 0 => break, + None if received < self.window => break, + _ => {} + } + response = self + .http + .get(url) + .header( + reqwest::header::RANGE, + format!("bytes={start}-{}", start + self.window - 1), + ) + .send() + .await + .map_err(download_err)?; + match response.status() { + reqwest::StatusCode::PARTIAL_CONTENT => {} + // Past the end: everything is on disk. + reqwest::StatusCode::RANGE_NOT_SATISFIABLE => break, + status => { + return Err(CaptureError::Download(format!( + "{track_path}: HTTP status {status}" + ))) + } } - *bytes_left -= len; - tokio::io::AsyncWriteExt::write_all(&mut audio, &chunk).await?; } tokio::io::AsyncWriteExt::flush(&mut audio).await?; drop(audio); @@ -346,6 +432,18 @@ impl Downloader { } } +/// Parses `bytes -/` into `(start, end, total)`. +fn parse_content_range(value: &str) -> Option<(u64, u64, Option)> { + let rest = value.trim().strip_prefix("bytes ")?; + let (range, total) = rest.split_once('/')?; + let (start, end) = range.split_once('-')?; + let total = match total.trim() { + "*" => None, + n => Some(n.parse().ok()?), + }; + Some((start.trim().parse().ok()?, end.trim().parse().ok()?, total)) +} + /// Writes the skipped toml for `track` at listing position `index`, /// overwriting whatever was there. async fn write_skipped( @@ -631,6 +729,16 @@ mod tests { assert_eq!(extension_for(None, "not a url"), "bin"); } + #[test] + fn content_range_parses_totals_and_wildcards() { + assert_eq!( + parse_content_range("bytes 0-1023/7831134"), + Some((0, 1023, Some(7831134))) + ); + assert_eq!(parse_content_range("bytes 5-9/*"), Some((5, 9, None))); + assert_eq!(parse_content_range("garbage"), None); + } + #[test] fn audio_files_pair_with_their_toml_names() { // Same prefix and sanitized stem as `fsdy::track_file_name`, so the diff --git a/crabidy-server/src/capture_store.rs b/crabidy-server/src/capture_store.rs index 7305723..66da6cc 100644 --- a/crabidy-server/src/capture_store.rs +++ b/crabidy-server/src/capture_store.rs @@ -209,6 +209,74 @@ mod tests { format!("http://{addr}/stream") } + /// Like [`serve`], but the server enforces bounded ranges the way + /// googlevideo does: ranged requests up to `max_window` bytes get + /// `206` + `Content-Range` slices, anything else (plain, open-ended, + /// oversized) gets `403`. + async fn serve_ranged(content_type: &'static str, body: Vec, max_window: u64) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind test server"); + let addr = listener.local_addr().expect("test server addr"); + tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let body = body.clone(); + tokio::spawn(async move { + let mut buf = Vec::new(); + let mut chunk = [0u8; 1024]; + loop { + match sock.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(n) => { + buf.extend_from_slice(&chunk[..n]); + if buf.windows(4).any(|w| w == b"\r\n\r\n") { + break; + } + } + } + } + let request = String::from_utf8_lossy(&buf).to_lowercase(); + let range = request + .lines() + .find_map(|line| line.strip_prefix("range: bytes=")) + .and_then(|spec| { + let (start, end) = spec.trim().split_once('-')?; + let start: u64 = start.parse().ok()?; + let end: u64 = end.parse().ok()?; + Some((start, end)) + }); + let total = body.len() as u64; + let response = match range { + Some((start, end)) if start < total && end - start < max_window => { + let end = end.min(total - 1); + let slice = &body[start as usize..=end as usize]; + let mut head = format!( + "HTTP/1.1 206 Partial Content\r\nContent-Type: {content_type}\r\n\ + Content-Range: bytes {start}-{end}/{total}\r\n\ + Content-Length: {}\r\nConnection: close\r\n\r\n", + slice.len() + ) + .into_bytes(); + head.extend_from_slice(slice); + head + } + Some((start, _)) if start >= total => format!( + "HTTP/1.1 416 Range Not Satisfiable\r\n\ + Content-Range: bytes */{total}\r\nContent-Length: 0\r\n\ + Connection: close\r\n\r\n" + ) + .into_bytes(), + _ => b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + .to_vec(), + }; + let _ = sock.write_all(&response).await; + let _ = sock.shutdown().await; + }); + } + }); + format!("http://{addr}/stream") + } + /// A provider with one album (`/mock/a`: tracks `one`, `two`) whose /// stream URLs are the test server's, and a node-level download /// blessing toggle. @@ -626,6 +694,31 @@ mod tests { ); } + #[tokio::test] + async fn downloads_chain_bounded_windows_on_strict_cdns() { + // A googlevideo-style server: only bounded ranges under 16 bytes + // are served; plain or oversized requests are 403. A downloader + // with a 10-byte window must fetch the 100-byte body completely. + let body: Vec = (0..100u32).map(|i| i as u8).collect(); + let url = serve_ranged("audio/flac", body.clone(), 16).await; + let mock = MockProvider::new(&[("/mock/a/1", &url)], true); + let dir = TempDir::new().expect("store tempdir"); + let store_dir = dir.path().join("captures"); + tokio::fs::create_dir_all(&store_dir).await.expect("mkdir"); + let store = CaptureStore { + dir: store_dir, + sink: crate::capture::Sink::Download( + crate::capture::Downloader::with_window(10).expect("downloader"), + ), + }; + store + .capture(&mock, "/mock/a/1", "windowed", &silent()) + .await + .expect("windowed capture"); + let audio = fs::read(store.dir().join("windowed/0001 one.flac")).expect("audio"); + assert_eq!(audio, body, "all windows stitched in order"); + } + #[tokio::test] async fn capture_reports_progress_totals_and_skips() { let url = serve("200 OK", "audio/flac", b"x".to_vec()).await; diff --git a/devenv.nix b/devenv.nix index 48582d6..717e301 100644 --- a/devenv.nix +++ b/devenv.nix @@ -14,6 +14,10 @@ let d2 pkg-config protobuf + # Stream-URL sidecar for the ytdy provider: YouTube caps tokenless + # stream URLs at ~1 MiB and yt-dlp is the only maintained cipher + # solver (architecture/youtube-rustypipe.md, D2-revised). + yt-dlp ]; in { diff --git a/plan/summary.md b/plan/summary.md index f670341..506dff3 100644 --- a/plan/summary.md +++ b/plan/summary.md @@ -1,5 +1,35 @@ # Implementation summaries +## youtube stream fetching (2026-07-21, follow-up) + +The rustypipe swap fixed the decode problem but real playback then hit +YouTube's tokenless-fetch enforcement, measured live: every stream URL +serves exactly its leading 1 MiB (403 beyond — plain, open-ended, and +oversized requests are rejected outright, and fresh URLs refuse offset +starts, killing URL-per-window chaining). PO tokens would lift the cap, +but rustypipe only attaches them to web clients whose signature +deciphering is currently broken upstream (verified on git master; +`rustypipe-botguard` built and tested — ineffective through the iOS +client). `yt-dlp` still solves the ciphers; its URLs stream the whole +file at a throttled ~32 KB/s — double the audio bitrate. + +Shipped: (1) **windowed HTTP fetching** everywhere — a +`WindowedHttpStream` `SourceStream` in audio-player (bounded ~1 MiB +ranges, 200-body fallback for range-ignoring servers, eager +seek/reconnect so rejected windows fail typed instead of retrying +forever, URLs never in errors) and the same windowing in the capture +downloader (strict-CDN test stitches windows byte-exact); (2) `yt-dlp` +back as a **stream-URL-only sidecar** — all metadata stays on +rustypipe; a missing binary degrades to 1 MiB streams with a warning, +a failing call falls back to the rustypipe URL; (3) `botguard_bin` +config passthrough so streams flip back to pure Rust when upstream +deciphering recovers; (4) capture per-track deadline raised to 30 min +for the throttle. Live-verified end to end on the exact track from the +user's log: sidecar URL in 3 s, windowed stream + rodio decode +producing samples at 4.3 s. 179 workspace tests green (12 new); +`quality/youtube-rustypipe.md` gained a checked "Stream fetching" +section. + ## youtube-rustypipe (2026-07-21) Built per `plan/youtube-rustypipe.md`: the ytdy provider's `yt-dlp` diff --git a/plan/youtube-rustypipe.md b/plan/youtube-rustypipe.md index 3660eb6..350d572 100644 --- a/plan/youtube-rustypipe.md +++ b/plan/youtube-rustypipe.md @@ -27,3 +27,12 @@ Ordered tasks; each names its verification (ytdy tests and/or gates in URL fetch, decode, download capture — removed after passing. - [x] **T7 — Docs.** `plan/summary.md` section incl. deviations; reconcile the architecture docs. +- [x] **T8 — Same-day fix: windowed fetching + stream sidecar.** + YouTube caps tokenless URLs at their leading ~1 MiB (measured; offset + starts refused; upstream web-client deciphering broken; botguard + ineffective through iOS). `WindowedHttpStream` in audio-player, + windowed capture downloads, `yt-dlp` back as a stream-URL-only + sidecar with degrade-and-fallback, `botguard_bin` passthrough, + capture deadline 30 min. Verifies: audio-player windowed tests, + capture window-chaining test, ytdy sidecar tests, gates "Stream + fetching"; live probe on the failing track. diff --git a/quality/youtube-rustypipe.md b/quality/youtube-rustypipe.md index c8360fb..d043c70 100644 --- a/quality/youtube-rustypipe.md +++ b/quality/youtube-rustypipe.md @@ -47,3 +47,23 @@ probe). - [x] Live probe (removed after passing): search + stream URL + download capture through the real provider; stream decodes with the player's decoder stack. + +## Stream fetching (D2-revised, same-day fix) + +- [x] The player streams every http(s) source through the windowed + `SourceStream` (bounded ~1 MiB ranges); servers that ignore `Range` + degrade to one 200 body; rejected windows are typed errors (eager + seek/reconnect — no infinite retry loops), and error messages never + carry URLs. +- [x] The capture downloader uses the same bounded windows with a 200 + whole-body fallback; existing capture tests pass unchanged and a + googlevideo-style strict server test stitches windows correctly. +- [x] `yt-dlp` is consulted **only** for stream URLs (argv-only, + bounded, stderr summarized ≤200 chars); a missing binary degrades + with a warning, never a failed init; a failing call falls back to the + rustypipe URL. +- [x] `botguard_bin` is configurable and forwarded to rustypipe (the + future pure-Rust exit). +- [x] Live probe (removed after passing): the user's exact failing + track resolved via the sidecar and decoded through the real + player path (windowed StreamDownload → rodio) in ~4 s. diff --git a/ytdy/Cargo.toml b/ytdy/Cargo.toml index 39da9a3..f796953 100644 --- a/ytdy/Cargo.toml +++ b/ytdy/Cargo.toml @@ -10,6 +10,7 @@ dirs.workspace = true rustypipe.workspace = true serde.workspace = true thiserror.workspace = true +tokio = { workspace = true, features = ["process", "time", "io-util"] } toml.workspace = true tracing.workspace = true diff --git a/ytdy/src/extract.rs b/ytdy/src/extract.rs index 7d9285d..2125d40 100644 --- a/ytdy/src/extract.rs +++ b/ytdy/src/extract.rs @@ -125,12 +125,22 @@ impl RustyPipeExtractor { /// Builds the client. `storage_dir` holds rustypipe's cache file /// (client state and the rotated auth cookie — as secret as the /// cookies file itself); `timeout` bounds every YouTube request. - pub fn new(storage_dir: PathBuf, timeout: Duration) -> Result { + /// `botguard_bin` optionally names a `rustypipe-botguard` binary for + /// PO-token attestation (rustypipe also auto-detects it on PATH). + pub fn new( + storage_dir: PathBuf, + timeout: Duration, + botguard_bin: Option<&Path>, + ) -> Result { std::fs::create_dir_all(&storage_dir) .map_err(|err| ExtractError::Client(format!("storage directory: {err}")))?; - let rp = RustyPipe::builder() + let mut builder = RustyPipe::builder() .storage_dir(storage_dir) - .timeout(timeout) + .timeout(timeout); + if let Some(bin) = botguard_bin { + builder = builder.botguard_bin(bin.as_os_str().to_owned()); + } + let rp = builder .build() .map_err(|err| ExtractError::Client(err.to_string()))?; Ok(Self { rp }) @@ -297,7 +307,7 @@ mod tests { // missing file short-circuit before any request. let dir = tempfile::TempDir::new().expect("tempdir"); let extractor = - RustyPipeExtractor::new(dir.path().join("rustypipe"), Duration::from_secs(5)) + RustyPipeExtractor::new(dir.path().join("rustypipe"), Duration::from_secs(5), None) .expect("client builds"); assert!(!extractor.login(None).await); assert!(!extractor.login(Some(&dir.path().join("gone.txt"))).await); diff --git a/ytdy/src/lib.rs b/ytdy/src/lib.rs index bf47426..bd23242 100644 --- a/ytdy/src/lib.rs +++ b/ytdy/src/lib.rs @@ -1,6 +1,10 @@ -//! YouTube media provider, backed by the pure-Rust `rustypipe` -//! Innertube client (see `architecture/youtube-provider.md` for the -//! tree shape and `architecture/youtube-rustypipe.md` for the engine). +//! YouTube media provider: metadata (search, playlists, video details) +//! comes from the pure-Rust `rustypipe` Innertube client; **stream +//! URLs** go through a minimal `yt-dlp` sidecar when available, because +//! YouTube currently caps tokenless URLs at their leading ~1 MiB (see +//! `architecture/youtube-provider.md` for the tree shape and +//! `architecture/youtube-rustypipe.md` for the engine and its +//! constraints). //! //! Mounted at [`PROVIDER_ROOT`]. Search works without login: creatable //! search-term nodes exactly like `/tidal/search` (in-memory terms, @@ -19,7 +23,9 @@ use serde::{Deserialize, Serialize}; use tracing::{debug, warn}; pub mod extract; +pub mod ytdlp; use extract::{Extract, ExtractError, RustyPipeExtractor, VideoEntry}; +use ytdlp::YtDlp; /// First path segment owned by this provider. pub const PROVIDER_ROOT: &str = "/youtube"; @@ -49,6 +55,19 @@ pub struct Settings { /// Per-request timeout in seconds. Default /// [`DEFAULT_CALL_TIMEOUT_SECS`]. pub call_timeout_secs: Option, + /// The `yt-dlp` binary used **only** to resolve stream URLs (a bare + /// name resolves via PATH; default `yt-dlp`). YouTube currently + /// caps tokenless URLs at their leading ~1 MiB, and yt-dlp is the + /// only maintained cipher solver — without it, playback of a + /// YouTube track stops after roughly a minute (see + /// architecture/youtube-rustypipe.md). + pub binary: Option, + /// Optional `rustypipe-botguard` binary for PO-token attestation + /// (also auto-detected on PATH). Currently ineffective — the + /// token-capable Innertube clients need signature deciphering that + /// is broken in rustypipe upstream — but wired so streams flip back + /// to pure Rust when upstream recovers. + pub botguard_bin: Option, } /// A parsed `/youtube/...` path. @@ -118,6 +137,9 @@ fn entry_to_track(entry: &VideoEntry, node_path: &str) -> Track { #[derive(Debug)] pub struct Client { extractor: Box, + /// The stream-URL sidecar; `None` degrades to rustypipe URLs (which + /// YouTube currently caps at their leading ~1 MiB). + ytdlp: Option, settings: Settings, /// The login attempt at init succeeded — gates the playlists /// subtree. @@ -133,6 +155,7 @@ impl Client { fn with_extractor(extractor: Box, settings: Settings, logged_in: bool) -> Self { Self { extractor, + ytdlp: None, settings, logged_in, search_terms: std::sync::RwLock::new(Vec::new()), @@ -263,10 +286,11 @@ impl Client { #[async_trait] impl ProviderClient for Client { /// Builds the rustypipe client (state under - /// `/crabidy/rustypipe/`) and attempts the cookie login. A - /// client that cannot be built fails init (the orchestrator disables - /// the provider non-fatally); a failed login degrades to logged-out - /// with a warning, never an error. + /// `/crabidy/rustypipe/`), attempts the cookie login, and + /// probes the `yt-dlp` stream-URL sidecar. A rustypipe client that + /// cannot be built fails init (the orchestrator disables the + /// provider non-fatally); a failed login or a missing sidecar + /// degrades with a warning, never an error. async fn init(raw_toml_settings: &str) -> Result { let settings: Settings = toml::from_str(raw_toml_settings).unwrap_or_else(|_| { warn!("could not parse toml settings, using defaults"); @@ -280,17 +304,36 @@ impl ProviderClient for Client { let storage_dir = dirs::config_dir() .map(|dir| dir.join("crabidy").join("rustypipe")) .ok_or_else(|| ProviderError::Config("no config directory".to_string()))?; - let extractor = RustyPipeExtractor::new(storage_dir, timeout).map_err(|err| { - warn!("cannot build the youtube client: {err}"); - ProviderError::Config(err.to_string()) - })?; + let extractor = + RustyPipeExtractor::new(storage_dir, timeout, settings.botguard_bin.as_deref()) + .map_err(|err| { + warn!("cannot build the youtube client: {err}"); + ProviderError::Config(err.to_string()) + })?; let logged_in = extractor.login(settings.cookies.as_deref()).await; + let binary = settings + .binary + .clone() + .unwrap_or_else(|| PathBuf::from("yt-dlp")); + let ytdlp = YtDlp::new(binary, timeout.max(Duration::from_secs(60))); + let ytdlp = match ytdlp.probe().await { + Ok(version) => { + debug!(version, "yt-dlp stream sidecar ready"); + Some(ytdlp) + } + Err(err) => { + warn!( + "yt-dlp not usable ({err}); youtube streams will stop after \ + their first ~1 MiB (see architecture/youtube-rustypipe.md)" + ); + None + } + }; debug!(logged_in, "youtube extractor ready"); - Ok(Self::with_extractor( - Box::new(extractor), - settings, - logged_in, - )) + Ok(Self { + ytdlp, + ..Self::with_extractor(Box::new(extractor), settings, logged_in) + }) } fn settings(&self) -> String { @@ -307,11 +350,32 @@ impl ProviderClient for Client { /// The video's audio stream URL — `audio/mp4` (AAC) preferred, the /// format the local player decodes /// (architecture/youtube-rustypipe.md D2). + /// + /// Resolved through the `yt-dlp` sidecar when available (its cipher + /// solving yields fully streamable URLs); rustypipe is the fallback, + /// whose tokenless URLs YouTube currently caps at ~1 MiB. async fn get_urls_for_track(&self, track_path: &str) -> Result, ProviderError> { let video = match parse_path(track_path)? { YtPath::SearchTrack { video, .. } | YtPath::PlaylistTrack { video, .. } => video, _ => return Err(ProviderError::MalformedPath), }; + if let Some(ytdlp) = &self.ytdlp { + let watch_url = format!("https://www.youtube.com/watch?v={video}"); + match ytdlp.stream_url(&watch_url).await { + Ok(url) => return Ok(vec![url]), + Err(err) => { + warn!( + video, + "yt-dlp stream resolution failed, trying rustypipe: {err}" + ); + } + } + } else { + warn!( + video, + "no yt-dlp sidecar; this stream will stop after its first ~1 MiB" + ); + } let url = self .extractor .audio_stream_url(video) @@ -769,11 +833,56 @@ mod tests { } #[test] - fn settings_tolerate_the_retired_binary_key() { - // Old ytdy.toml files carry `binary = "yt-dlp"`; parsing must - // ignore it instead of resetting the whole config. + fn settings_round_trip_old_and_new_keys() { + // Pre-rustypipe ytdy.toml files carry `binary`; it kept its + // meaning (now stream-URL-only). New keys parse alongside. let settings: Settings = - toml::from_str("binary = \"yt-dlp\"\nsearch_results = 7\n").expect("parses"); + toml::from_str("binary = \"yt-dlp\"\nsearch_results = 7\nbotguard_bin = \"/opt/bg\"\n") + .expect("parses"); assert_eq!(settings.search_results, Some(7)); + assert_eq!(settings.binary, Some(PathBuf::from("yt-dlp"))); + assert_eq!(settings.botguard_bin, Some(PathBuf::from("/opt/bg"))); + } + + #[tokio::test] + async fn stream_urls_prefer_the_ytdlp_sidecar() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::TempDir::new().expect("tempdir"); + let script = dir.path().join("fake-yt-dlp"); + std::fs::write( + &script, + "#!/bin/sh\ncase \"$*\" in\n*watch?v=vid1*) echo \"https://sidecar.test/a.m4a\"; exit 0 ;;\n*) exit 1 ;;\nesac\n", + ) + .expect("write"); + let mut perms = std::fs::metadata(&script).expect("meta").permissions(); + perms.set_mode(0o755); + std::fs::set_permissions(&script, perms).expect("chmod"); + + let mut client = client(); + client.ytdlp = Some(YtDlp::new(script, Duration::from_secs(30))); + // The sidecar wins over the fake extractor's URL. (Retried: under + // parallel test load, exec of a just-written script can race a + // concurrent fork and fall back spuriously.) + let mut urls = Vec::new(); + for _ in 0..5 { + urls = client + .get_urls_for_track("/youtube/search/lofi/vid1") + .await + .expect("stream url"); + if urls == vec!["https://sidecar.test/a.m4a".to_string()] { + break; + } + } + assert_eq!(urls, vec!["https://sidecar.test/a.m4a".to_string()]); + // ...and a failing sidecar falls back to the extractor's answer. + client.ytdlp = Some(YtDlp::new( + dir.path().join("missing-binary"), + Duration::from_secs(5), + )); + let urls = client + .get_urls_for_track("/youtube/search/lofi/vid1") + .await + .expect("fallback"); + assert_eq!(urls, vec!["https://example.test/a.m4a".to_string()]); } } diff --git a/ytdy/src/ytdlp.rs b/ytdy/src/ytdlp.rs new file mode 100644 index 0000000..44dc115 --- /dev/null +++ b/ytdy/src/ytdlp.rs @@ -0,0 +1,201 @@ +//! Minimal `yt-dlp` sidecar for **stream URLs only** +//! (see `architecture/youtube-rustypipe.md` D2-revised). +//! +//! All metadata comes from the pure-Rust rustypipe extractor; this +//! module exists because YouTube currently rejects sustained tokenless +//! stream fetching (every URL serves exactly its leading ~1 MiB), and +//! the token-capable Innertube clients need signature deciphering that +//! is broken in rustypipe upstream right now. `yt-dlp` solves the +//! cipher challenges and its URLs stream fully (throttled, but well +//! above audio bitrate). When upstream recovers, this sidecar can be +//! dropped without touching the provider: it is only consulted by +//! `get_urls_for_track`. +//! +//! Subprocess discipline: argv-only (never a shell), one bounded call +//! (`kill_on_drop`), stdout capped, stderr summarized to a short +//! suffix. Errors carry the video id and exit status — never stream +//! URLs (they may embed tokens). + +use std::path::PathBuf; +use std::process::Stdio; +use std::time::Duration; + +use tracing::debug; + +/// Cap on captured stdout; `-g` prints a handful of URLs. +const MAX_STDOUT_BYTES: usize = 512 * 1024; + +/// Errors from the sidecar. +#[derive(Debug, thiserror::Error)] +pub enum YtDlpError { + #[error("cannot run yt-dlp: {0}")] + Spawn(String), + #[error("yt-dlp timed out after {0:?}")] + Timeout(Duration), + #[error("yt-dlp failed ({status}): {stderr}")] + Failed { status: String, stderr: String }, + #[error("yt-dlp printed no stream url")] + NoUrl, +} + +/// A configured `yt-dlp` binary used for stream URL resolution. +#[derive(Clone, Debug)] +pub struct YtDlp { + binary: PathBuf, + timeout: Duration, +} + +impl YtDlp { + pub fn new(binary: PathBuf, timeout: Duration) -> Self { + Self { binary, timeout } + } + + /// Probes `--version`; used at init to decide whether the sidecar + /// is available (absence degrades to rustypipe URLs, never fails + /// the provider). + pub async fn probe(&self) -> Result { + let stdout = self.run(&["--version"]).await?; + Ok(stdout.trim().to_string()) + } + + /// The audio stream URL for a watch URL: `audio/mp4` (AAC) + /// preferred — the format the local player decodes. + pub async fn stream_url(&self, watch_url: &str) -> Result { + let stdout = self + .run(&[ + "-f", + "bestaudio[ext=m4a]/bestaudio/best", + "-g", + "--no-playlist", + watch_url, + ]) + .await?; + stdout + .lines() + .map(str::trim) + .find(|line| !line.is_empty()) + .map(str::to_string) + .ok_or(YtDlpError::NoUrl) + } + + /// Runs one bounded call and returns its stdout. + async fn run(&self, args: &[&str]) -> Result { + debug!(binary = %self.binary.display(), ?args, "running yt-dlp"); + let mut command = tokio::process::Command::new(&self.binary); + command + .args(["--no-warnings"]) + .args(args) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + let child = command + .spawn() + .map_err(|err| YtDlpError::Spawn(err.to_string()))?; + let output = tokio::time::timeout(self.timeout, child.wait_with_output()) + .await + .map_err(|_| YtDlpError::Timeout(self.timeout))? + .map_err(|err| YtDlpError::Spawn(err.to_string()))?; + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + let summary: String = stderr.trim().chars().take(200).collect(); + return Err(YtDlpError::Failed { + status: output.status.to_string(), + stderr: summary, + }); + } + let mut stdout = output.stdout; + stdout.truncate(MAX_STDOUT_BYTES); + Ok(String::from_utf8_lossy(&stdout).into_owned()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + use std::os::unix::fs::PermissionsExt; + use std::path::Path; + + /// A fake yt-dlp: a script dispatching on its argv. + fn fake_binary(dir: &Path, body: &str) -> PathBuf { + let path = dir.join("fake-yt-dlp"); + fs::write(&path, format!("#!/bin/sh\n{body}\n")).expect("write fake binary"); + let mut perms = fs::metadata(&path).expect("metadata").permissions(); + perms.set_mode(0o755); + fs::set_permissions(&path, perms).expect("chmod"); + path + } + + /// Retries a call while exec races a concurrent fork (ETXTBSY under + /// parallel test load). + async fn retry_spawn(mut call: F) -> Result + where + F: FnMut() -> Fut, + Fut: std::future::Future>, + { + let mut last = call().await; + for _ in 0..4 { + if !matches!(last, Err(YtDlpError::Spawn(_))) { + break; + } + last = call().await; + } + last + } + + #[tokio::test] + async fn resolves_stream_urls_and_probes() { + let dir = tempfile::TempDir::new().expect("tempdir"); + let binary = fake_binary( + dir.path(), + r#" +case "$*" in +*--version*) echo "2026.07.04"; exit 0 ;; +*watch?v=vid1*) printf '%s\n' "https://example.test/a.m4a"; exit 0 ;; +*) echo boom >&2; exit 1 ;; +esac"#, + ); + let ytdlp = YtDlp::new(binary, Duration::from_secs(30)); + assert_eq!( + retry_spawn(|| ytdlp.probe()).await.expect("probe"), + "2026.07.04" + ); + let url = retry_spawn(|| ytdlp.stream_url("https://www.youtube.com/watch?v=vid1")) + .await + .expect("url"); + assert_eq!(url, "https://example.test/a.m4a"); + // Failures are typed and carry a bounded stderr summary. + let err = ytdlp + .stream_url("https://www.youtube.com/watch?v=nope") + .await + .expect_err("fails"); + assert!(err.to_string().contains("boom"), "{err}"); + // A missing binary is a typed spawn error. + let gone = YtDlp::new(dir.path().join("gone"), Duration::from_secs(5)); + assert!(matches!(gone.probe().await, Err(YtDlpError::Spawn(_)))); + } + + #[tokio::test] + async fn hung_calls_hit_the_timeout() { + let dir = tempfile::TempDir::new().expect("tempdir"); + let binary = fake_binary(dir.path(), "sleep 30"); + let ytdlp = YtDlp::new(binary, Duration::from_millis(200)); + let started = std::time::Instant::now(); + let mut saw_timeout = false; + for _ in 0..5 { + match ytdlp.probe().await { + Err(YtDlpError::Timeout(_)) => { + saw_timeout = true; + break; + } + // Under parallel test load, exec of a just-written + // script can race a concurrent fork (ETXTBSY) — retry. + Err(YtDlpError::Spawn(_)) => continue, + other => panic!("expected a timeout, got {other:?}"), + } + } + assert!(saw_timeout, "no timeout within 5 attempts"); + assert!(started.elapsed() < Duration::from_secs(10)); + } +}