diff --git a/src/providers/rule34video.rs b/src/providers/rule34video.rs index 8febb8f..7f2a7c7 100644 --- a/src/providers/rule34video.rs +++ b/src/providers/rule34video.rs @@ -20,9 +20,6 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode}; use std::vec; -/// Hands out `/proxy/rule34video/` URLs instead of the page URLs. Switched off; -/// the proxy and resolver stay in place (`proxies/rule34video.rs`). -const USE_MEDIA_PROXY: bool = false; /// The signed `v-acctoken` format URLs last ~12h; keep details well inside that. const DETAIL_TTL: Duration = Duration::from_secs(2 * 60 * 60); const DETAIL_CACHE_SIZE: usize = 5000; @@ -339,7 +336,6 @@ impl Rule34videoProvider { }; let video_items = self.get_video_items_from_html(text); let video_items = self.enrich_items(video_items, requester.clone()).await; - let video_items = Self::proxy_items(video_items, &options); if video_items.is_empty() { return Ok(stale()); } @@ -421,7 +417,6 @@ impl Rule34videoProvider { }); let video_items = self.get_video_items_from_html(text); let video_items = self.enrich_items(video_items, requester.clone()).await; - let video_items = Self::proxy_items(video_items, &options); if !video_items.is_empty() { cache.insert(index, video_items.clone()); @@ -495,7 +490,6 @@ impl Rule34videoProvider { }); let video_items = self.get_video_items_from_html(text); let video_items = self.enrich_items(video_items, requester.clone()).await; - let video_items = Self::proxy_items(video_items, &options); if !video_items.is_empty() { cache.insert(index, video_items.clone()); @@ -691,15 +685,28 @@ impl Rule34videoProvider { items } - /// With `USE_MEDIA_PROXY`, playback resolves the video page on demand through - /// `/proxy/rule34video/` instead of using the enriched formats. - fn proxy_items(mut items: Vec, options: &ServerOptions) -> Vec { - if !USE_MEDIA_PROXY { - return items; - } + /// Points each format at `/proxy/rule34video/{id}/{slug}/{height}.mp4`. The + /// signed links are bound to the IP that loaded the page (here: a pool + /// proxy), so the server re-resolves the page itself and streams the file. + /// The detail cache keeps the raw links; this runs on every serve. + fn stream_formats(mut items: Vec, options: &ServerOptions) -> Vec { for item in &mut items { - if let Some(path) = item.url.split("/video/").nth(1).filter(|p| !p.is_empty()) { - item.url = crate::providers::build_proxy_url(options, "rule34video", path); + let Some(video) = item + .url + .split("/video/") + .nth(1) + .map(|path| path.trim_matches('/').to_string()) + .filter(|path| !path.is_empty()) + else { + continue; + }; + for format in item.formats.iter_mut().flatten() { + let height = format.pixel_height().unwrap_or(0); + format.url = crate::providers::build_proxy_url( + options, + "rule34video", + &format!("{video}/{height}.mp4"), + ); } } items @@ -816,6 +823,7 @@ impl Provider for Rule34videoProvider { let page_num = page.parse::().unwrap_or(1); let target = self.resolve_target(query.as_deref(), &options).await; + let serve_options = options.clone(); let result = match &target { Target::Latest => self.get(cache, page_num, &sort, options).await, Target::Search(q) => self.query(cache, page_num, q, &sort, options).await, @@ -824,7 +832,7 @@ impl Provider for Rule34videoProvider { match result { // Cached listings pick up details that arrived after their deadline. - Ok(v) => self.with_details(v), + Ok(v) => Self::stream_formats(self.with_details(v), &serve_options), Err(e) => { eprintln!("Error fetching videos: {}", e); vec![] @@ -945,6 +953,39 @@ mod tests { assert!(items[0].formats.is_none()); } + #[test] + fn served_formats_stream_through_the_server_by_height() { + // The signed links only play from the IP that loaded the page (a pool + // proxy), so clients get the server's streaming endpoint instead. + let mut item = VideoItem::new( + "4682247".to_string(), + "t".to_string(), + "https://rule34video.com/video/4682247/wednesday-x/".to_string(), + "Rule34video".to_string(), + String::new(), + 62, + ); + item.formats = Some(vec![ + VideoFormat::new("https://rule34video.com/get_file/a_1080p.mp4/?v-acctoken=x".into(), "1080p".into(), "mp4".into()) + .height(1080), + VideoFormat::new("https://rule34video.com/get_file/a_720p.mp4/?v-acctoken=y".into(), "720p".into(), "mp4".into()) + .height(720), + ]); + let mut options: ServerOptions = serde_json::from_value(serde_json::json!({})).unwrap(); + options.public_url_base = Some("https://hottub.example".to_string()); + + let served = Rule34videoProvider::stream_formats(vec![item], &options); + let urls: Vec<&str> = served[0].formats.as_ref().unwrap().iter().map(|f| f.url.as_str()).collect(); + assert_eq!( + urls, + vec![ + "https://hottub.example/proxy/rule34video/4682247/wednesday-x/1080.mp4", + "https://hottub.example/proxy/rule34video/4682247/wednesday-x/720.mp4", + ] + ); + assert_eq!(served[0].url, "https://rule34video.com/video/4682247/wednesday-x/"); + } + #[test] fn parses_prefixed_queries() { assert_eq!( diff --git a/src/proxies/eporner.rs b/src/proxies/eporner.rs index 8b05119..89c2b2e 100644 --- a/src/proxies/eporner.rs +++ b/src/proxies/eporner.rs @@ -1,30 +1,25 @@ -use std::collections::HashMap; -use std::rc::Rc; -use std::sync::{Mutex, OnceLock}; -use std::time::{Duration, Instant}; +use std::sync::OnceLock; +use std::time::Duration; -use futures::StreamExt; -use http_body_util::BodyExt; use ntex::http::Method; -use ntex::http::body::{Body, SizedStream}; -use ntex::http::header::{ACCEPT_RANGES, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE, RANGE}; -use ntex::util::Bytes; +use ntex::http::header::RANGE; use ntex::web::{self, HttpRequest}; use wreq::Client; use wreq::redirect::Policy; use wreq_util::Emulation; use crate::util::eporner_resolve::resolve_ranked; +use crate::util::media_stream::LinkCache; use crate::util::requester::Requester; /// The signed CDN links expire two hours after resolving; renew well before. const LINK_TTL: Duration = Duration::from_secs(90 * 60); const LINK_CACHE_SIZE: usize = 2000; -/// `{alnum_id}/{height}` -> (resolved at, signed CDN URL). -fn links() -> &'static Mutex> { - static LINKS: OnceLock>> = OnceLock::new(); - LINKS.get_or_init(|| Mutex::new(HashMap::new())) +/// `{alnum_id}/{height}` -> signed CDN URL. +fn links() -> &'static LinkCache { + static LINKS: OnceLock = OnceLock::new(); + LINKS.get_or_init(|| LinkCache::new(LINK_TTL, LINK_CACHE_SIZE)) } /// Redirects are not followed so the CDN's `na.mp4` decoy shows up as a 302. @@ -73,24 +68,12 @@ async fn cached_link( ) -> Option { let key = format!("{alnum_id}/{}", height.unwrap_or(0)); if !fresh { - if let Ok(links) = links().lock() { - if let Some((time, url)) = links.get(&key) { - if time.elapsed() < LINK_TTL { - return Some(url.clone()); - } - } + if let Some(url) = links().get(&key) { + return Some(url); } } let url = resolve_link(requester, alnum_id, height).await?; - if let Ok(mut links) = links().lock() { - if links.len() >= LINK_CACHE_SIZE { - links.retain(|_, (time, _)| time.elapsed() < LINK_TTL); - if links.len() >= LINK_CACHE_SIZE { - links.clear(); - } - } - links.insert(key, (Instant::now(), url.clone())); - } + links().insert(key, url.clone()); Some(url) } @@ -141,40 +124,7 @@ pub async fn serve_media( return Ok(web::HttpResponse::BadGateway().finish()); }; - let status = upstream.status(); - let headers = upstream.headers().clone(); - let mut builder = web::HttpResponse::build(status); - builder.header( - CONTENT_TYPE, - headers - .get(CONTENT_TYPE) - .and_then(|value| value.to_str().ok()) - .unwrap_or("video/mp4"), - ); - builder.header(ACCEPT_RANGES, "bytes"); - if let Some(value) = headers.get(CONTENT_RANGE.as_str()).and_then(|v| v.to_str().ok()) { - builder.header(CONTENT_RANGE, value); - } - if head { - // `finish()` would send `Content-Length: 0` over the upstream length. - return Ok(builder.body(Body::None)); - } - let length = headers - .get(CONTENT_LENGTH.as_str()) - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()); - let stream = Box::pin(upstream.into_data_stream().map(|chunk| { - chunk - .map(|bytes| Bytes::copy_from_slice(bytes.as_ref())) - .map_err(|e| std::io::Error::other(e.to_string())) - })); - let Some(length) = length else { - return Ok(builder.streaming(stream)); - }; - // A sized body keeps `Content-Length` on the (206) responses; a chunked - // one is what AVPlayer refuses to seek in. - let stream = stream.map(|chunk| chunk.map_err(|e| Rc::new(e) as Rc)); - Ok(builder.body(Body::from_message(SizedStream::new(length, stream)))) + Ok(crate::util::media_stream::stream_response(upstream, head)) } #[cfg(test)] diff --git a/src/proxies/rule34video.rs b/src/proxies/rule34video.rs index e005cbe..ad4585c 100644 --- a/src/proxies/rule34video.rs +++ b/src/proxies/rule34video.rs @@ -1,28 +1,102 @@ +use std::sync::OnceLock; +use std::time::Duration; + use ntex::http::Method; -use ntex::http::header::LOCATION; +use ntex::http::header::RANGE; use ntex::web::{self, HttpRequest}; +use wreq::Client; +use wreq::redirect::Policy; +use wreq_util::Emulation; +use crate::util::media_stream::{LinkCache, stream_response}; use crate::util::requester::Requester; -use crate::util::rule34video_resolve::{page_url, resolve_media_url}; +use crate::util::rule34video_resolve::{parse_endpoint, resolve_media_url}; -/// The channel lists `/proxy/rule34video/{id}/{slug}` instead of resolving every -/// video page while listing (which ran into the site's 429 limit): the page is -/// read when the video is played, and the player is sent to its signed mp4. -pub async fn redirect_to_media( +/// The signed links last ~12h; renew well before. +const LINK_TTL: Duration = Duration::from_secs(60 * 60); +const LINK_CACHE_SIZE: usize = 2000; + +/// `{page_url}{height}` -> signed mp4 URL resolved from this server's IP. +fn links() -> &'static LinkCache { + static LINKS: OnceLock = OnceLock::new(); + LINKS.get_or_init(|| LinkCache::new(LINK_TTL, LINK_CACHE_SIZE)) +} + +/// `get_file` answers with a redirect to the CDN host, which is followed. +fn media_client() -> &'static Client { + static CLIENT: OnceLock = OnceLock::new(); + CLIENT.get_or_init(|| { + Client::builder() + .emulation(Emulation::Firefox151) + .redirect(Policy::default()) + .build() + .expect("rule34video media client") + }) +} + +async fn cached_link( + requester: &web::types::State, + page_url: &str, + height: Option, + fresh: bool, +) -> Option { + let key = format!("{page_url}{}", height.unwrap_or(0)); + if !fresh { + if let Some(url) = links().get(&key) { + return Some(url); + } + } + let mut requester = requester.get_ref().clone(); + let url = resolve_media_url(&mut requester, page_url, height).await?; + links().insert(key, url.clone()); + Some(url) +} + +/// Streams `/proxy/rule34video/{id}/{slug}/{height}.mp4` through this server. +/// The signed `get_file` links only play from the IP that loaded the video +/// page (a different IP gets 403), and listings are enriched through pool +/// proxies, so the page is loaded again here, from this server, and the file +/// relayed to the player. +pub async fn serve_media( req: HttpRequest, requester: web::types::State, -) -> impl web::Responder { +) -> Result { let endpoint = req.match_info().query("endpoint").to_string(); - let Some(page_url) = page_url(&endpoint) else { - return web::HttpResponse::BadRequest().finish(); + let Some((page_url, height)) = parse_endpoint(&endpoint) else { + return Ok(web::HttpResponse::BadRequest().finish()); }; - if req.method() == Method::HEAD { - return web::HttpResponse::Ok().finish(); - } + let range = req + .headers() + .get(RANGE) + .and_then(|value| value.to_str().ok()) + .map(str::to_string); + let head = req.method() == Method::HEAD; - let mut requester = requester.get_ref().clone(); - match resolve_media_url(&mut requester, &page_url, None).await { - Some(url) => web::HttpResponse::Found().header(LOCATION, url).finish(), - None => web::HttpResponse::BadGateway().finish(), + let mut upstream = None; + for fresh in [false, true] { + let Some(url) = cached_link(&requester, &page_url, height, fresh).await else { + break; + }; + let mut request = if head { + media_client().head(&url) + } else { + media_client().get(&url) + }; + request = request.header("Referer", "https://rule34video.com/"); + if let Some(range) = &range { + request = request.header("Range", range); + } + match request.send().await { + // A 403 is an expired link, or one signed for another IP. + Ok(response) if response.status().is_success() => { + upstream = Some(response); + break; + } + _ => continue, + } } + let Some(upstream) = upstream else { + return Ok(web::HttpResponse::BadGateway().finish()); + }; + Ok(stream_response(upstream, head)) } diff --git a/src/proxy.rs b/src/proxy.rs index 04b5670..1f9537a 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -159,8 +159,8 @@ pub fn config(cfg: &mut web::ServiceConfig) { ) .service( web::resource("/rule34video/{endpoint}*") - .route(web::get().to(crate::proxies::rule34video::redirect_to_media)) - .route(web::head().to(crate::proxies::rule34video::redirect_to_media)), + .route(web::get().to(crate::proxies::rule34video::serve_media)) + .route(web::head().to(crate::proxies::rule34video::serve_media)), ) .service( web::resource("/hanime/{endpoint}*") diff --git a/src/util/media_stream.rs b/src/util/media_stream.rs new file mode 100644 index 0000000..9d3719d --- /dev/null +++ b/src/util/media_stream.rs @@ -0,0 +1,89 @@ +//! Streaming a remote media file through this server, for CDNs whose signed +//! links only play from the IP that resolved them. + +use std::collections::HashMap; +use std::rc::Rc; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use futures::StreamExt; +use http_body_util::BodyExt; +use ntex::http::body::{Body, SizedStream}; +use ntex::http::header::{ACCEPT_RANGES, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE}; +use ntex::util::Bytes; +use ntex::web; + +/// Signed media links by key (`{video}/{height}`), reused until `ttl`. +pub struct LinkCache { + ttl: Duration, + capacity: usize, + links: Mutex>, +} + +impl LinkCache { + pub fn new(ttl: Duration, capacity: usize) -> Self { + LinkCache { + ttl, + capacity, + links: Mutex::new(HashMap::new()), + } + } + + pub fn get(&self, key: &str) -> Option { + let links = self.links.lock().ok()?; + let (time, url) = links.get(key)?; + (time.elapsed() < self.ttl).then(|| url.clone()) + } + + pub fn insert(&self, key: String, url: String) { + if let Ok(mut links) = self.links.lock() { + if links.len() >= self.capacity { + links.retain(|_, (time, _)| time.elapsed() < self.ttl); + if links.len() >= self.capacity { + links.clear(); + } + } + links.insert(key, (Instant::now(), url)); + } + } +} + +/// Relays an upstream media response (status, type, range headers, body). +/// Pass `head` for HEAD requests so no body is sent. +pub fn stream_response(upstream: wreq::Response, head: bool) -> web::HttpResponse { + let status = upstream.status(); + let headers = upstream.headers().clone(); + let mut builder = web::HttpResponse::build(status); + builder.header( + CONTENT_TYPE, + headers + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .unwrap_or("video/mp4"), + ); + builder.header(ACCEPT_RANGES, "bytes"); + if let Some(value) = headers.get(CONTENT_RANGE.as_str()).and_then(|v| v.to_str().ok()) { + builder.header(CONTENT_RANGE, value); + } + if head { + // `finish()` would send `Content-Length: 0` over the upstream length. + return builder.body(Body::None); + } + let length = headers + .get(CONTENT_LENGTH.as_str()) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()); + let stream = Box::pin(upstream.into_data_stream().map(|chunk| { + chunk + .map(|bytes| Bytes::copy_from_slice(bytes.as_ref())) + .map_err(|e| std::io::Error::other(e.to_string())) + })); + let Some(length) = length else { + return builder.streaming(stream); + }; + // A sized body keeps `Content-Length` on the (206) responses; a chunked + // one is what AVPlayer refuses to seek in. + let stream = stream.map(|chunk| chunk.map_err(|e| Rc::new(e) as Rc)); + builder.body(Body::from_message(SizedStream::new(length, stream))) +} + diff --git a/src/util/mod.rs b/src/util/mod.rs index d2eb499..218cffd 100644 --- a/src/util/mod.rs +++ b/src/util/mod.rs @@ -8,6 +8,7 @@ pub mod flaresolverr; pub mod flow_debug; pub mod hanime_resolve; pub mod hoster_proxy; +pub mod media_stream; pub mod proxy; pub mod proxy_pool; pub mod requester; diff --git a/src/util/rule34video_resolve.rs b/src/util/rule34video_resolve.rs index 8a05e18..6a1fb32 100644 --- a/src/util/rule34video_resolve.rs +++ b/src/util/rule34video_resolve.rs @@ -89,6 +89,20 @@ pub fn page_url(endpoint: &str) -> Option { } } +/// `{id}/{slug}[/{height}.mp4]` -> (video page URL, requested height). +pub fn parse_endpoint(endpoint: &str) -> Option<(String, Option)> { + let parts: Vec<&str> = endpoint.trim_matches('/').split('/').collect(); + let (video, height) = match parts.as_slice() { + [id, slug] => (format!("{id}/{slug}"), None), + [id, slug, file] => { + let height = file.strip_suffix(".mp4")?.parse::().ok()?; + (format!("{id}/{slug}"), Some(height)) + } + _ => return None, + }; + Some((page_url(&video)?, height)) +} + /// The best signed mp4 URL (or the one of `height`) for a video page. pub async fn resolve_media_url( requester: &mut Requester, @@ -134,6 +148,20 @@ mod tests { assert_eq!(page_url("123/slug/extra"), None); } + #[test] + fn parses_the_streaming_endpoint_with_its_height() { + assert_eq!( + parse_endpoint("4682247/wednesday-x/720.mp4"), + Some(("https://rule34video.com/video/4682247/wednesday-x/".to_string(), Some(720))) + ); + assert_eq!( + parse_endpoint("4682247/wednesday-x"), + Some(("https://rule34video.com/video/4682247/wednesday-x/".to_string(), None)) + ); + assert_eq!(parse_endpoint("4682247/wednesday-x/../../etc"), None); + assert_eq!(parse_endpoint("4682247/../720.mp4"), None); + } + fn env_usize(name: &str, default: usize) -> usize { std::env::var(name) .ok() diff --git a/src/videos.rs b/src/videos.rs index d1dc969..35adeb3 100644 --- a/src/videos.rs +++ b/src/videos.rs @@ -402,6 +402,10 @@ impl VideoFormat { self.height = Some(height); self } + /// The height set with [`Self::height`], if any. + pub fn pixel_height(&self) -> Option { + self.height + } pub fn tbr(mut self, tbr: u32) -> Self { self.tbr = Some(tbr); self