diff --git a/Cargo.toml b/Cargo.toml index 8f55c69..8eed6a7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,7 +20,7 @@ ntex-files = "3" serde = "1.0.228" serde_json = "1.0.145" tokio = { version = "1.49", features = ["full"] } -wreq = { version = "6.0.0-rc.31", features = ["cookies", "multipart", "json", "gzip", "brotli", "deflate", "zstd"] } +wreq = { version = "6.0.0-rc.31", features = ["cookies", "multipart", "json", "gzip", "brotli", "deflate", "zstd", "socks"] } wreq-util = "3.0.0-rc.14" percent-encoding = "2.3.2" capitalize = "0.3.4" diff --git a/SKILL.md b/SKILL.md index 09e279e..f5cadb4 100644 --- a/SKILL.md +++ b/SKILL.md @@ -64,6 +64,9 @@ Runtime env vars: - `PROXY` optional. Any value other than `"0"` enables proxy mode in the shared requester. - `BURP_URL` optional. Outbound HTTP proxy used when `PROXY` is enabled. - `FLARE_URL` optional but strongly recommended for provider work. Used for FlareSolverr fallback and some providers that call it directly. +- `PROXY_POOL` optional. `0` disables the public-proxy pool used by rule34video enrichment (default on). +- `PROXY_POOL_LISTS` optional. Comma-separated proxy list URLs replacing the default TheSpeedX lists. +- `PROXY_POOL_CONCURRENCY` optional. Cap on concurrent proxy checks (default 500). - `DOMAIN` optional. Used for the `/` redirect target. - `DISCORD_WEBHOOK` optional. Enables `/api/test` and provider error reporting to Discord. diff --git a/docs/architecture.md b/docs/architecture.md index 9ce9e31..0bbadd0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -23,9 +23,10 @@ Hottub is a Rust server that exposes Hot Tub compatible endpoints for channel di 1. `main` loads `.env` and ensures `RUST_LOG` is set. 2. It creates the Diesel SQLite pool from `DATABASE_URL`. 3. It creates a shared `Requester`, enables Burp proxying when `PROXY != 0`, and builds the LRU video cache. -4. It configures provider runtime validation in `providers::configure_runtime_validation`. -5. It spawns a background thread that forces provider initialization via `providers::init_providers_now()`. -6. It starts an `ntex` HTTP server on `0.0.0.0:18080`. +4. It starts the public-proxy pool (`util::proxy_pool::start`) unless `PROXY_POOL=0`. +5. It configures provider runtime validation in `providers::configure_runtime_validation`. +6. It spawns a background thread that forces provider initialization via `providers::init_providers_now()`. +7. It starts an `ntex` HTTP server on `0.0.0.0:18080`. ## Runtime Environment @@ -36,6 +37,9 @@ Important environment variables: - `PROXY`: enables Burp proxying when not equal to `0`. - `BURP_URL`: outbound proxy URL used when `PROXY` is enabled. - `FLARE_URL`: FlareSolverr endpoint used as the last HTML-fetch fallback. +- `PROXY_POOL`: `0` disables the public-proxy pool (default on). +- `PROXY_POOL_LISTS`: comma-separated proxy list URLs replacing the default TheSpeedX lists. +- `PROXY_POOL_CONCURRENCY`: cap on concurrent proxy checks (default 500). - `DOMAIN`: used by the `/` redirect target. - `DISCORD_WEBHOOK`: enables `/api/test` and provider error reporting. @@ -120,7 +124,9 @@ Sends a Discord error test if `DISCORD_WEBHOOK` is configured. ### `/api/proxies` -Returns the background-fetched outbound proxy snapshot from `src/util/proxy.rs`. +Status of the public-proxy pool in `src/util/proxy_pool.rs`: the last list sweep, the fastest exit IPs, and per site (for example `rule34video.com`) the proxies ready for it with their request counters (`ok`/`failed`/`throttled`) and any 429 cooldown. Proxy credentials are never shown. + +The pool downloads public proxy lists hourly and keeps only proxies that pass a TLS-verified check against `cdn-cgi/trace` twice with a median under 1.5s, one per exit IP. Providers register a site (`proxy_pool::register_site`) with a cheap check URL; the pool checks its candidates against that URL and rotates requests over the ready ones (`next_proxy` / `report`). rule34video spreads its detail-page enrichment over it to stay under the site's per-IP 429 limit. ## Core Data Structures diff --git a/src/api.rs b/src/api.rs index ca95893..1d5434a 100644 --- a/src/api.rs +++ b/src/api.rs @@ -6,7 +6,6 @@ use crate::providers::{ use crate::uploaders::{UploaderProfile, UploadersRequest}; use crate::util::cache::VideoCache; use crate::util::discord::send_discord_error_report; -use crate::util::proxy::{Proxy, all_proxies_snapshot}; use crate::util::requester::Requester; use crate::{DbPool, db, status::*, videos::*}; use ntex::http::header; @@ -992,26 +991,7 @@ pub async fn test() -> Result { } pub async fn proxies() -> Result { - let proxies = all_proxies_snapshot().await.unwrap_or_default(); - crate::flow_debug!("proxies endpoint snapshot_count={}", proxies.len()); - let mut by_protocol: std::collections::BTreeMap> = - std::collections::BTreeMap::new(); - for proxy in proxies { - by_protocol - .entry(proxy.protocol.clone()) - .or_default() - .push(proxy); - } - for proxies in by_protocol.values_mut() { - proxies.sort_by(|a, b| { - a.host - .cmp(&b.host) - .then(a.port.cmp(&b.port)) - .then(a.username.cmp(&b.username)) - .then(a.password.cmp(&b.password)) - }); - } - Ok(web::HttpResponse::Ok().json(&by_protocol)) + Ok(web::HttpResponse::Ok().json(&crate::util::proxy_pool::status())) } #[cfg(test)] diff --git a/src/main.rs b/src/main.rs index a1d66ac..465951e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -61,6 +61,7 @@ async fn main() -> std::io::Result<()> { let proxy_enabled = env::var("PROXY").unwrap_or("0".to_string()) != "0".to_string(); requester.set_proxy(proxy_enabled); crate::flow_debug!("requester initialized proxy_enabled={}", proxy_enabled); + util::proxy_pool::start(); let cache: util::cache::VideoCache = crate::util::cache::VideoCache::new() .max_size(100_000) diff --git a/src/providers/rule34video.rs b/src/providers/rule34video.rs index be7bcba..8febb8f 100644 --- a/src/providers/rule34video.rs +++ b/src/providers/rule34video.rs @@ -5,8 +5,11 @@ use crate::status::*; use crate::util::cache::VideoCache; use crate::util::discord::send_discord_error_report; use crate::util::parse_abbreviated_number; +use crate::util::proxy_pool::{self, Outcome}; +use crate::util::requester::Requester; +use crate::util::rule34video_resolve; use crate::util::time::parse_time_to_seconds; -use crate::videos::{ServerOptions, VideoItem}; +use crate::videos::{ServerOptions, VideoFormat, VideoItem}; use async_trait::async_trait; use error_chain::error_chain; use htmlentity::entity::{ICodedDataTrait, decode}; @@ -20,6 +23,18 @@ 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; +/// Proxy-pool site key; its proxies are checked against `robots.txt`. +const POOL_HOST: &str = "rule34video.com"; +const POOL_CHECK_URL: &str = "https://rule34video.com/robots.txt"; +const ENRICH_CONCURRENCY: usize = 8; +/// A listing waits this long for its detail pages, then goes out regardless. +const ENRICH_DEADLINE: Duration = Duration::from_secs(4); +const DETAIL_TIMEOUT: Duration = Duration::from_secs(3); +/// A failed detail fetch is retried on another proxy only with this much time left. +const RETRY_MIN_LEFT: Duration = Duration::from_secs(1); const TARGET_TTL: Duration = Duration::from_secs(6 * 60 * 60); const TARGET_CACHE_SIZE: usize = 2000; /// A probed archive smaller than this loses to keyword search: "elf girl" is a @@ -71,14 +86,23 @@ impl Target { #[derive(Clone)] pub struct Rule34videoProvider { url: String, + /// Enriched items (formats, tags, artist) by video URL. Every listing costs + /// ~23 detail pages, again for the prefetched next page. + details: VideoCache, /// Bare query -> resolved target, so later pages skip the archive probes. targets: Arc>>, } impl Rule34videoProvider { pub fn new() -> Self { + let mut details = VideoCache::new(); + details.max_size(DETAIL_CACHE_SIZE); + // Registered at construction so the pool checks proxies against the + // site as soon as it has candidates, before the first listing. + proxy_pool::register_site(POOL_HOST, POOL_CHECK_URL); Rule34videoProvider { url: "https://rule34video.com".to_string(), + details, targets: Arc::new(Mutex::new(HashMap::new())), } } @@ -314,6 +338,7 @@ 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()); @@ -395,6 +420,7 @@ impl Rule34videoProvider { "".to_string() }); 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() { @@ -468,6 +494,7 @@ impl Rule34videoProvider { "".to_string() }); 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() { @@ -481,8 +508,191 @@ impl Rule34videoProvider { } } - /// Playback resolves the video page on demand through `/proxy/rule34video/`, - /// instead of fetching every listed page up front (which ran into 429s). + /// Fills formats, tags and the artist from a video page; `None` when the + /// page carries no `flashvars` (challenge, error page). + fn apply_detail(&self, mut item: VideoItem, html: &str) -> Option { + let formats: Vec = rule34video_resolve::parse_formats(html) + .into_iter() + .map(|(height, url)| { + let quality = match height { + 0 => "auto".to_string(), + h => format!("{h}p"), + }; + let format = VideoFormat::new(url, quality.clone(), "mp4".to_string()) + .format_id(format!("mp4-{quality}")); + match height { + 0 => format, + h => format.height(h), + } + }) + .collect(); + if formats.is_empty() { + return None; + } + item.formats = Some(formats); + + let mut tags = rule34video_resolve::parse_tags(html); + let artists = Self::parse_artists(html); + if let Some((name, slug)) = artists.first() { + item.uploader = Some(name.clone()); + item.uploaderUrl = Some(format!("{}/models/{slug}/", self.url)); + item.uploaderId = Some(format!("rule34video:{slug}")); + } + for (name, _) in &artists { + if !tags.iter().any(|t| t.eq_ignore_ascii_case(name)) { + tags.push(name.clone()); + } + } + if !tags.is_empty() { + item.tags = Some(tags); + } + Some(item) + } + + /// Artists credited on a video page: `(name, slug)` from the "Artist" group. + fn parse_artists(html: &str) -> Vec<(String, String)> { + let document = Html::parse_document(html); + let Ok(group_sel) = Selector::parse("[data-suggest-type=\"model\"] a[href*=\"/models/\"]") + else { + return vec![]; + }; + let name_sel = Selector::parse(".name").ok(); + let mut artists: Vec<(String, String)> = Vec::new(); + for link in document.select(&group_sel) { + let href = link.value().attr("href").unwrap_or_default(); + let slug = href + .split("/models/") + .nth(1) + .unwrap_or_default() + .trim_matches('/') + .to_string(); + let name = name_sel + .as_ref() + .and_then(|sel| link.select(sel).next()) + .map(|n| n.text().collect::()) + .unwrap_or_else(|| link.text().collect::()); + let name = name.split_whitespace().collect::>().join(" "); + if slug.is_empty() || name.is_empty() || artists.iter().any(|(_, s)| *s == slug) { + continue; + } + artists.push((name, slug)); + } + artists + } + + + /// Fetches the detail pages of a listing through the proxy pool, spread over + /// its proxies so the site's per-IP 429 limit is never reached. Returns at + /// the deadline with whatever arrived; the rest keep running in the + /// background and land in the detail cache for the next time the listing + /// is served. Without ready proxies the listing data is returned as is. + async fn enrich_items(&self, items: Vec, requester: Requester) -> Vec { + let mut items = self.with_details(items); + let pending: Vec = items + .iter() + .enumerate() + .filter(|(_, item)| item.formats.is_none() && !item.url.is_empty()) + .map(|(index, _)| index) + .collect(); + if pending.is_empty() || !proxy_pool::is_ready(POOL_HOST) { + return items; + } + + let deadline = tokio::time::Instant::now() + ENRICH_DEADLINE; + let permits = Arc::new(tokio::sync::Semaphore::new(ENRICH_CONCURRENCY)); + let handles: Vec<(usize, tokio::task::JoinHandle>)> = pending + .into_iter() + .map(|index| { + let provider = self.clone(); + let item = items[index].clone(); + let requester = requester.clone(); + let permits = permits.clone(); + let task = tokio::spawn(async move { + let _permit = permits.acquire_owned().await.ok()?; + let html = Self::fetch_detail(&requester, &item.url, deadline).await?; + let enriched = provider.apply_detail(item, &html)?; + provider.store_detail(&enriched); + Some(enriched) + }); + (index, task) + }) + .collect(); + // A timed-out wait drops only the handle; the task itself runs on. + for (index, handle) in handles { + if let Ok(Ok(Some(enriched))) = tokio::time::timeout_at(deadline, handle).await { + items[index] = enriched; + } + } + items + } + + /// One video page through the next pool proxy, retried once on another + /// proxy while at least `RETRY_MIN_LEFT` of the deadline remains. + async fn fetch_detail( + requester: &Requester, + url: &str, + deadline: tokio::time::Instant, + ) -> Option { + for attempt in 0..2 { + let left = deadline.saturating_duration_since(tokio::time::Instant::now()); + if attempt > 0 && left < RETRY_MIN_LEFT { + return None; + } + let proxy = proxy_pool::next_proxy(POOL_HOST)?; + let headers = vec![("Referer".to_string(), "https://rule34video.com/".to_string())]; + let (outcome, page) = match requester.get_via(&proxy, url, headers, DETAIL_TIMEOUT).await { + Err(_) => (Outcome::Failed, None), + Ok(response) => { + let status = response.status().as_u16(); + match response.text().await { + Err(_) => (Outcome::Failed, None), + Ok(body) if status == 429 || Requester::looks_like_cloudflare_challenge(&body) => { + (Outcome::Throttled, None) + } + Ok(body) if body.contains("var flashvars") => (Outcome::Ok, Some(Some(body))), + // The video is gone; the proxy did its job. + Ok(_) if status == 404 || status == 410 => (Outcome::Ok, Some(None)), + Ok(_) => (Outcome::Failed, None), + } + } + }; + proxy_pool::report(POOL_HOST, &proxy, outcome); + if let Some(page) = page { + return page; + } + } + None + } + + fn store_detail(&self, item: &VideoItem) { + self.details.insert(item.url.clone(), vec![item.clone()]); + } + + /// Copies cached details (formats, tags, artist) onto listing items. Runs on + /// every served listing, so details that arrived after a listing's deadline + /// show up the next time it is served from cache. + fn with_details(&self, mut items: Vec) -> Vec { + for item in &mut items { + let Some((time, cached)) = self.details.get(&item.url) else { + continue; + }; + if time.elapsed().unwrap_or(DETAIL_TTL) >= DETAIL_TTL { + continue; + } + let Some(cached) = cached.into_iter().next() else { + continue; + }; + item.formats = cached.formats; + item.tags = cached.tags; + item.uploader = cached.uploader; + item.uploaderUrl = cached.uploaderUrl; + item.uploaderId = cached.uploaderId; + } + 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; @@ -613,7 +823,8 @@ impl Provider for Rule34videoProvider { }; match result { - Ok(v) => v, + // Cached listings pick up details that arrived after their deadline. + Ok(v) => self.with_details(v), Err(e) => { eprintln!("Error fetching videos: {}", e); vec![] @@ -660,6 +871,80 @@ mod tests { assert_eq!(items[0].duration, 88); } + #[test] + fn served_listings_pick_up_details_that_arrived_after_the_deadline() { + let provider = Rule34videoProvider::new(); + let bare = VideoItem::new( + "42".to_string(), + "Some Title".to_string(), + "https://rule34video.com/video/42/some-title/".to_string(), + "Rule34video".to_string(), + String::new(), + 88, + ); + let mut late = bare.clone(); + late.tags = Some(vec!["3D".to_string()]); + late.formats = Some(vec![VideoFormat::new( + "https://rule34video.com/get_file/a_720p.mp4/".to_string(), + "720p".to_string(), + "mp4".to_string(), + )]); + provider.store_detail(&late); + + let served = provider.with_details(vec![bare]); + assert_eq!(served[0].tags, Some(vec!["3D".to_string()])); + assert_eq!(served[0].formats.as_ref().map(Vec::len), Some(1)); + } + + #[test] + fn a_detail_page_fills_formats_tags_and_artist() { + let html = r#"
Artist
+ + JOIGang (VA)
+ "#; + let item = VideoItem::new( + "1".to_string(), + "t".to_string(), + "https://rule34video.com/video/1/t/".to_string(), + "Rule34video".to_string(), + String::new(), + 0, + ); + let item = Rule34videoProvider::new().apply_detail(item, html).expect("flashvars present"); + let formats = item.formats.expect("formats"); + assert_eq!(formats.len(), 2); + assert!(formats[0].url.contains("a_720p"), "best quality first"); + assert_eq!( + item.tags, + Some(vec!["3D".into(), "Cyberpunk 2077".into(), "bondage".into(), "JOIGang (VA)".into()]) + ); + assert_eq!(item.uploader.as_deref(), Some("JOIGang (VA)")); + assert_eq!(item.uploaderUrl.as_deref(), Some("https://rule34video.com/models/joigang-va/")); + assert!(Rule34videoProvider::new().apply_detail(VideoItem::new( + "1".into(), "t".into(), "u".into(), "Rule34video".into(), String::new(), 0, + ), "DDoS-Guard").is_none()); + } + + #[tokio::test] + async fn without_ready_proxies_enrichment_is_skipped_not_done_directly() { + let provider = Rule34videoProvider::new(); + let item = VideoItem::new( + "1".to_string(), + "t".to_string(), + "https://rule34video.com/video/1/t/".to_string(), + "Rule34video".to_string(), + String::new(), + 0, + ); + let started = std::time::Instant::now(); + let items = provider + .enrich_items(vec![item], crate::util::requester::Requester::new()) + .await; + assert!(started.elapsed() < Duration::from_millis(500), "no request was made"); + assert_eq!(items.len(), 1); + assert!(items[0].formats.is_none()); + } + #[test] fn parses_prefixed_queries() { assert_eq!( diff --git a/src/util/mod.rs b/src/util/mod.rs index 8e80313..d2eb499 100644 --- a/src/util/mod.rs +++ b/src/util/mod.rs @@ -9,6 +9,7 @@ pub mod flow_debug; pub mod hanime_resolve; pub mod hoster_proxy; pub mod proxy; +pub mod proxy_pool; pub mod requester; pub mod rule34video_resolve; pub mod time; diff --git a/src/util/proxy.rs b/src/util/proxy.rs index 74c9798..fdf74ef 100644 --- a/src/util/proxy.rs +++ b/src/util/proxy.rs @@ -1,26 +1,9 @@ use std::fmt; use std::str::FromStr; -use std::sync::Arc; use serde::Serialize; -use tokio::sync::{OnceCell, RwLock}; -use tokio::time::{self, Duration}; use url::Url; use wreq::Proxy as WreqProxy; -use wreq::{Client, Version}; - -use crate::util::requester::Requester; - -pub static ALL_PROXIES: OnceCell>>> = OnceCell::const_new(); -static ALL_PROXY_KEYS: OnceCell>>> = - OnceCell::const_new(); - -const PROXY_LIST: [&str; 3] = [ - "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt", - "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt", - "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks4.txt", -]; -const IFCONFIG_URL: &str = "https://ifconfig.co"; #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct Proxy { @@ -44,12 +27,6 @@ impl ProxyParseError { } } -pub async fn all_proxies_snapshot() -> Option> { - let handle = ALL_PROXIES.get()?.clone(); - let proxies = handle.read().await; - Some(proxies.clone()) -} - impl fmt::Display for ProxyParseError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "{}", self.message) @@ -78,6 +55,11 @@ impl fmt::Display for Proxy { } impl Proxy { + /// `protocol://host:port`, without credentials: safe to show or log. + pub fn redacted(&self) -> String { + format!("{}://{}:{}", self.protocol, self.host, self.port) + } + pub fn parse(input: &str) -> Result { input.parse() } @@ -99,147 +81,6 @@ impl Proxy { } } -pub async fn fetch_proxies( - requester: &mut Requester, - proxy_list_url: &str, - proxy_sink: Arc>>, -) -> Result { - let default_protocol = protocol_from_list_url(proxy_list_url)?; - let body = requester - .get(proxy_list_url, None) - .await - .map_err(|e| ProxyParseError::new(format!("failed to fetch proxy list: {e}")))?; - let mut tasks = tokio::task::JoinSet::new(); - for (line_index, line) in body.lines().enumerate() { - let trimmed = line.trim(); - if trimmed.is_empty() { - continue; - } - - let proxy_input = if trimmed.contains("://") { - trimmed.to_string() - } else { - format!("{default_protocol}://{trimmed}") - }; - - let proxy = Proxy::parse(&proxy_input).map_err(|e| { - ProxyParseError::new(format!("invalid proxy on line {}: {e}", line_index + 1)) - })?; - - tasks.spawn(async move { verify_proxy(proxy).await }); - } - - let mut added = 0usize; - while let Some(result) = tasks.join_next().await { - match result { - Ok(Ok(proxy)) => { - if let Some(proxy_keys) = ALL_PROXY_KEYS.get() { - let key = proxy_key(&proxy); - let mut keys = proxy_keys.write().await; - if !keys.insert(key) { - continue; - } - let mut proxies = proxy_sink.write().await; - proxies.push(proxy); - added += 1; - } - } - Ok(Err(_err)) => { - // eprintln!("Proxy verification failed: {err}"); - } - Err(err) => { - eprintln!("Proxy verification task failed: {err}"); - } - } - } - - Ok(added) -} - -fn protocol_from_list_url(list_url: &str) -> Result<&'static str, ProxyParseError> { - if list_url.contains("http.txt") { - Ok("http") - } else if list_url.contains("socks4.txt") { - Ok("socks4") - } else if list_url.contains("socks5.txt") { - Ok("socks5") - } else { - Err(ProxyParseError::new(format!( - "unknown proxy list protocol for url: {list_url}" - ))) - } -} - -async fn verify_proxy(proxy: Proxy) -> Result { - let wreq_proxy = proxy.to_wreq_proxy()?; - let client = Client::builder() - .tls_cert_verification(false) - .build() - .map_err(|e| ProxyParseError::new(format!("failed to build http client: {e}")))?; - - let response = client - .get(IFCONFIG_URL) - .version(Version::HTTP_11) - .proxy(wreq_proxy) - .timeout(Duration::from_secs(5)) - .send() - .await - .map_err(|e| ProxyParseError::new(format!("proxy request failed: {e}")))?; - - if response.status().is_success() { - Ok(proxy) - } else { - Err(ProxyParseError::new(format!( - "proxy returned status {}", - response.status() - ))) - } -} - -pub fn init_all_proxies_background(requester: Requester) { - if ALL_PROXIES.get().is_some() { - return; - } - - if tokio::runtime::Handle::try_current().is_err() { - eprintln!("Skipping proxy list init: no Tokio runtime available"); - return; - } - - let proxy_cache = Arc::new(RwLock::new(Vec::new())); - let proxy_keys = Arc::new(RwLock::new(std::collections::HashSet::new())); - if ALL_PROXIES.set(proxy_cache.clone()).is_err() { - return; - } - let _ = ALL_PROXY_KEYS.set(proxy_keys); - - tokio::spawn(async move { - let mut interval = time::interval(Duration::from_secs(60 * 60)); - loop { - let mut tasks = tokio::task::JoinSet::new(); - for list in PROXY_LIST { - let proxy_cache = proxy_cache.clone(); - let mut requester = requester.clone(); - tasks.spawn(async move { fetch_proxies(&mut requester, list, proxy_cache).await }); - } - - while let Some(result) = tasks.join_next().await { - match result { - Ok(Ok(_added)) => {} - Ok(Err(err)) => { - eprintln!("Failed to fetch proxy list: {err}"); - } - Err(err) => { - eprintln!("Proxy list task failed: {err}"); - } - } - } - - interval.tick().await; - } - }); -} - impl FromStr for Proxy { type Err = ProxyParseError; @@ -261,7 +102,6 @@ impl FromStr for Proxy { || url.query().is_some() || url.fragment().is_some() { - println!("Parsed proxy URL: {:?}", url); return Err(ProxyParseError::new(format!( "proxy url must not include path, query, or fragment: {:?}", input @@ -287,14 +127,3 @@ impl FromStr for Proxy { }) } } - -fn proxy_key(proxy: &Proxy) -> String { - format!( - "{}://{}:{}@{}:{}", - proxy.protocol, - proxy.username.as_deref().unwrap_or(""), - proxy.password.as_deref().unwrap_or(""), - proxy.host, - proxy.port - ) -} diff --git a/src/util/proxy_pool.rs b/src/util/proxy_pool.rs new file mode 100644 index 0000000..3088d08 --- /dev/null +++ b/src/util/proxy_pool.rs @@ -0,0 +1,757 @@ +//! Pool of fast public proxies that rate-limited providers spread requests over. + +use std::collections::{HashMap, VecDeque}; +use std::time::{Duration, Instant}; + +use crate::util::proxy::Proxy; + +/// A proxy only counts once it has answered this many checks: many listed +/// proxies answer once and are gone a minute later. +const MIN_SUCCESSES: u32 = 2; +/// Proxies slower than this (median) are not worth waiting for. +const LATENCY_CEILING_SECS: f64 = 1.5; +/// Latency samples kept per proxy for the median. +const SAMPLE_WINDOW: usize = 5; +/// Consecutive failed checks or requests after which a proxy is out. +const MAX_CONSECUTIVE_FAILURES: u32 = 2; + +struct Entry { + proxy: Proxy, + successes: u32, + samples: VecDeque, + exit_ip: Option, + consecutive_failures: u32, + benched_until: Option, + /// Real requests sent through this proxy, for the status page. + requests: RequestCounts, +} + +#[derive(Default, Clone, Copy, serde::Serialize)] +pub struct RequestCounts { + pub ok: u64, + pub failed: u64, + pub throttled: u64, +} + +impl Entry { + fn median(&self) -> Option { + let mut sorted: Vec = self.samples.iter().copied().collect(); + sorted.sort_by(f64::total_cmp); + sorted.get(sorted.len() / 2).copied() + } +} + +/// Health record of every proxy that has passed at least one check. +#[derive(Default)] +pub(crate) struct Ranking { + entries: HashMap, +} + +impl Ranking { + pub(crate) fn record_success(&mut self, proxy: &Proxy, secs: f64, exit_ip: Option<&str>) { + let entry = self.entries.entry(proxy.to_string()).or_insert_with(|| Entry { + proxy: proxy.clone(), + successes: 0, + samples: VecDeque::new(), + exit_ip: None, + consecutive_failures: 0, + benched_until: None, + requests: RequestCounts::default(), + }); + entry.successes += 1; + entry.consecutive_failures = 0; + if exit_ip.is_some() { + entry.exit_ip = exit_ip.map(str::to_string); + } + if entry.samples.len() == SAMPLE_WINDOW { + entry.samples.pop_front(); + } + entry.samples.push_back(secs); + } + + pub(crate) fn record_failure(&mut self, proxy: &Proxy) { + if let Some(entry) = self.entries.get_mut(&proxy.to_string()) { + entry.consecutive_failures += 1; + } + } + + /// Keeps a working proxy out of rotation until `until` without counting it as broken. + pub(crate) fn bench(&mut self, proxy: &Proxy, until: Instant) { + if let Some(entry) = self.entries.get_mut(&proxy.to_string()) { + entry.benched_until = Some(until); + } + } + + fn count(&mut self, proxy: &Proxy, outcome: Outcome) { + if let Some(entry) = self.entries.get_mut(&proxy.to_string()) { + match outcome { + Outcome::Ok => entry.requests.ok += 1, + Outcome::Failed => entry.requests.failed += 1, + Outcome::Throttled => entry.requests.throttled += 1, + } + } + } + + /// Status rows for the usable proxies, fastest first. Benched proxies are + /// listed too, with the time they still sit out. + pub(crate) fn status(&self, now: Instant, limit: usize) -> Vec { + self.usable(now, limit, true) + .into_iter() + .map(|(median, entry)| ProxyStatus { + proxy: entry.proxy.redacted(), + exit_ip: entry.exit_ip.clone(), + median_ms: (median * 1000.0).round() as u64, + consecutive_failures: entry.consecutive_failures, + requests: entry.requests, + benched_for_secs: entry + .benched_until + .filter(|until| *until > now) + .map(|until| (until - now).as_secs()), + }) + .collect() + } + + /// Usable proxies, fastest first. + pub(crate) fn ranked(&self, now: Instant, limit: usize) -> Vec { + self.usable(now, limit, false) + .into_iter() + .map(|(_, entry)| entry.proxy.clone()) + .collect() + } + + fn usable(&self, now: Instant, limit: usize, include_benched: bool) -> Vec<(f64, &Entry)> { + let mut usable: Vec<(f64, &Entry)> = self + .entries + .values() + .filter(|entry| entry.successes >= MIN_SUCCESSES) + .filter(|entry| entry.consecutive_failures < MAX_CONSECUTIVE_FAILURES) + .filter(|entry| { + include_benched || entry.benched_until.is_none_or(|until| until <= now) + }) + .filter_map(|entry| entry.median().map(|median| (median, entry))) + .filter(|(median, _)| *median < LATENCY_CEILING_SECS) + .collect(); + usable.sort_by(|a, b| a.0.total_cmp(&b.0)); + // Many list entries are ports of one host: spreading requests over them + // would still come from one IP, so keep only the fastest per exit IP. + let mut seen_ips = std::collections::HashSet::new(); + usable + .into_iter() + .filter(|(_, entry)| entry.exit_ip.as_ref().is_none_or(|ip| seen_ips.insert(ip.clone()))) + .take(limit) + .collect() + } + + /// Forgets proxies that are out (repeated failures), so the record stays small. + pub(crate) fn prune(&mut self) { + self.entries + .retain(|_, entry| entry.consecutive_failures < MAX_CONSECUTIVE_FAILURES); + } + + fn exit_ip(&self, proxy: &Proxy) -> Option { + self.entries.get(&proxy.to_string())?.exit_ip.clone() + } +} + +#[derive(serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ProxyStatus { + /// `protocol://host:port`, credentials left out. + pub proxy: String, + pub exit_ip: Option, + pub median_ms: u64, + pub consecutive_failures: u32, + pub requests: RequestCounts, + pub benched_for_secs: Option, +} + +/// Proxies in a list file; bare `host:port` lines take the protocol from the +/// file name (`socks5.txt`, ...), as the TheSpeedX lists do. +pub(crate) fn parse_list(list_url: &str, body: &str) -> Vec { + let name = list_url.rsplit('/').next().unwrap_or_default(); + let protocol = ["socks5", "socks4", "http"] + .into_iter() + .find(|p| name.contains(p)) + .unwrap_or("http"); + body.lines() + .map(str::trim) + .filter(|line| !line.is_empty()) + .filter_map(|line| match line.contains("://") { + true => Proxy::parse(line).ok(), + false => Proxy::parse(&format!("{protocol}://{line}")).ok(), + }) + .collect() +} + +/// The `ip=` line of `cdn-cgi/trace`: the address the target saw. +pub(crate) fn trace_exit_ip(body: &str) -> Option { + body.lines() + .find_map(|line| line.strip_prefix("ip=")) + .map(str::trim) + .filter(|ip| !ip.is_empty()) + .map(str::to_string) +} + +/// What happened to a request sent through a proxy. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + Ok, + /// Connection error, timeout or garbage: counts like a failed check. + Failed, + /// 429 or a challenge page: the proxy works, its IP is throttled there. + Throttled, +} + +/// Proxies a site rotates over. +pub(crate) const SITE_SET_SIZE: usize = 20; +pub(crate) const THROTTLE_COOLDOWN: Duration = Duration::from_secs(10 * 60); + +/// A site's own view of the pool: proxies checked against the site itself. +pub(crate) struct Site { + pub(crate) check_url: String, + pub(crate) ranking: Ranking, + cursor: usize, +} + +impl Site { + pub(crate) fn new(check_url: &str) -> Self { + Site { + check_url: check_url.to_string(), + ranking: Ranking::default(), + cursor: 0, + } + } + + /// The next proxy in round-robin order, so requests spread evenly. + pub(crate) fn next(&mut self, now: Instant) -> Option { + let ready = self.ranking.ranked(now, SITE_SET_SIZE); + if ready.is_empty() { + return None; + } + let proxy = ready[self.cursor % ready.len()].clone(); + self.cursor = self.cursor.wrapping_add(1); + Some(proxy) + } + + pub(crate) fn report(&mut self, proxy: &Proxy, outcome: Outcome, now: Instant) { + self.ranking.count(proxy, outcome); + match outcome { + Outcome::Ok => {} + Outcome::Failed => self.ranking.record_failure(proxy), + Outcome::Throttled => self.ranking.bench(proxy, now + THROTTLE_COOLDOWN), + } + } +} + + +// --------------------------------------------------------------------------- +// The live pool: one process-wide instance, fed by a background thread. +// --------------------------------------------------------------------------- + +const DEFAULT_LISTS: [&str; 3] = [ + "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt", + "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt", + "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks4.txt", +]; +/// Tiny, served from Cloudflare's edge everywhere, and it echoes the exit IP. +const TRACE_URL: &str = "https://www.cloudflare.com/cdn-cgi/trace"; +const CHECK_TIMEOUT: Duration = Duration::from_secs(3); +const LIST_TIMEOUT: Duration = Duration::from_secs(30); +const SWEEP_INTERVAL: Duration = Duration::from_secs(60 * 60); +const RECHECK_INTERVAL: Duration = Duration::from_secs(5 * 60); +const DEFAULT_CONCURRENCY: usize = 500; +/// Fastest exit IPs listed on the status page. All proxies under the latency +/// ceiling stay in the pool and are rechecked; sites draw from all of them. +const FAST_SET_SIZE: usize = 20; + +#[derive(Default)] +struct PoolState { + enabled: bool, + generic: Ranking, + sites: HashMap, + last_sweep: Option, +} + +#[derive(Clone, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct SweepStats { + pub finished_unix: u64, + pub candidates: usize, + pub passed: usize, + pub fast: usize, + pub duration_secs: u64, +} + +#[derive(serde::Serialize)] +pub struct SiteStatus { + pub ready: usize, + pub proxies: Vec, +} + +#[derive(serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct PoolStatus { + pub enabled: bool, + pub last_sweep: Option, + pub fast: Vec, + pub sites: std::collections::BTreeMap, +} + +fn state() -> &'static std::sync::Mutex { + static STATE: std::sync::OnceLock> = std::sync::OnceLock::new(); + STATE.get_or_init(Default::default) +} + +fn with_state(f: impl FnOnce(&mut PoolState) -> T) -> T { + let mut guard = state().lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + f(&mut guard) +} + +/// Lets a provider draw on the pool for `host`. Records the site only (no +/// network): its proxies are checked against `check_url` once the pool has +/// candidates, so registering from a provider's constructor warms it up. +pub fn register_site(host: &str, check_url: &str) { + with_state(|state| { + state + .sites + .entry(host.to_string()) + .or_insert_with(|| Site::new(check_url)); + }); +} + +/// The next proxy for `host` in round-robin order; `None` while the site has none ready. +pub fn next_proxy(host: &str) -> Option { + with_state(|state| state.sites.get_mut(host)?.next(Instant::now())) +} + +/// Whether `host` has at least one proxy to send requests through. +pub fn is_ready(host: &str) -> bool { + with_state(|state| { + state + .sites + .get(host) + .is_some_and(|site| !site.ranking.ranked(Instant::now(), 1).is_empty()) + }) +} + +/// Feeds a real request's outcome back, so a dead or throttled proxy leaves +/// the rotation before its next scheduled check. +pub fn report(host: &str, proxy: &Proxy, outcome: Outcome) { + with_state(|state| { + // Eviction and cooldown are per site; the generic record only counts. + state.generic.count(proxy, outcome); + if let Some(site) = state.sites.get_mut(host) { + site.report(proxy, outcome, Instant::now()); + } + }); +} + +pub fn status() -> PoolStatus { + with_state(|state| { + let now = Instant::now(); + PoolStatus { + enabled: state.enabled, + last_sweep: state.last_sweep.clone(), + fast: state.generic.status(now, FAST_SET_SIZE), + sites: state + .sites + .iter() + .map(|(host, site)| { + let ready = site.ranking.ranked(now, SITE_SET_SIZE).len(); + let proxies = site.ranking.status(now, SITE_SET_SIZE); + (host.clone(), SiteStatus { ready, proxies }) + }) + .collect(), + } + }) +} + +/// Starts the background sweeps unless `PROXY_POOL=0`. Called once from `main`. +pub fn start() { + if std::env::var("PROXY_POOL").is_ok_and(|v| v == "0") { + println!("proxy pool disabled (PROXY_POOL=0)"); + return; + } + with_state(|state| state.enabled = true); + let lists: Vec = match std::env::var("PROXY_POOL_LISTS") { + Ok(value) if !value.trim().is_empty() => value + .split(',') + .map(|url| url.trim().to_string()) + .filter(|url| !url.is_empty()) + .collect(), + _ => DEFAULT_LISTS.iter().map(|url| url.to_string()).collect(), + }; + let concurrency = std::env::var("PROXY_POOL_CONCURRENCY") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|n| *n > 0) + .unwrap_or(DEFAULT_CONCURRENCY); + + std::thread::spawn(move || { + let runtime = match tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + { + Ok(runtime) => runtime, + Err(e) => { + eprintln!("proxy pool: no runtime: {e}"); + return; + } + }; + runtime.block_on(run(lists, concurrency)); + }); +} + +/// Our own public address, as the trace endpoint sees it without a proxy. +async fn own_exit_ip() -> String { + loop { + let direct = check(None, TRACE_URL, CHECK_TIMEOUT).await.ok(); + if let Some(ip) = direct.and_then(|(_, body)| trace_exit_ip(&body)) { + return ip; + } + tokio::time::sleep(Duration::from_secs(30)).await; + } +} + +async fn run(lists: Vec, concurrency: usize) { + loop { + // Re-read every sweep: the server's own address can change. + let own_ip = own_exit_ip().await; + sweep(&lists, concurrency, &own_ip).await; + check_sites(concurrency).await; + let rounds = (SWEEP_INTERVAL.as_secs() / RECHECK_INTERVAL.as_secs()).saturating_sub(1); + for _ in 0..rounds { + tokio::time::sleep(RECHECK_INTERVAL).await; + let fast = with_state(|state| state.generic.ranked(Instant::now(), usize::MAX)); + check_generic(fast, concurrency, &own_ip).await; + with_state(|state| state.generic.prune()); + check_sites(concurrency).await; + } + tokio::time::sleep(RECHECK_INTERVAL).await; + } +} + +/// Downloads the lists and checks every candidate, then the survivors again: +/// a proxy needs two successful checks to count. +async fn sweep(lists: &[String], concurrency: usize, own_ip: &str) { + let started = Instant::now(); + let mut candidates: Vec = Vec::new(); + for list in lists { + match check(None, list, LIST_TIMEOUT).await.ok() { + Some((_, body)) => candidates.extend(parse_list(list, &body)), + None => eprintln!("proxy pool: could not fetch list {list}"), + } + } + let mut seen = std::collections::HashSet::new(); + candidates.retain(|proxy| seen.insert(proxy.to_string())); + let total = candidates.len(); + + let passed = check_generic(candidates, concurrency, own_ip).await; + let passed_count = passed.len(); + check_generic(passed, concurrency, own_ip).await; + let fast = with_state(|state| { + state.generic.prune(); + let fast = state.generic.ranked(Instant::now(), usize::MAX).len(); + state.last_sweep = Some(SweepStats { + finished_unix: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0), + candidates: total, + passed: passed_count, + fast, + duration_secs: started.elapsed().as_secs(), + }); + fast + }); + println!( + "proxy pool: sweep of {total} candidates: {passed_count} passed, {fast} fast exit IPs ({}s)", + started.elapsed().as_secs() + ); +} + +/// Trace-checks `proxies` and records the results; returns the ones that passed. +async fn check_generic(proxies: Vec, concurrency: usize, own_ip: &str) -> Vec { + use futures::stream::{self, StreamExt}; + let results: Vec<(Proxy, Option<(f64, String)>)> = stream::iter(proxies) + .map(|proxy| async move { + let result = check(Some(&proxy), TRACE_URL, CHECK_TIMEOUT) + .await + .ok() + .and_then(|(secs, body)| trace_exit_ip(&body).map(|ip| (secs, ip))) + // A proxy that shows our own address hides nothing. + .filter(|(_, ip)| ip != own_ip); + (proxy, result) + }) + .buffer_unordered(concurrency) + .collect() + .await; + with_state(|state| { + let mut passed = Vec::new(); + for (proxy, result) in results { + match result { + Some((secs, ip)) => { + state.generic.record_success(&proxy, secs, Some(&ip)); + passed.push(proxy); + } + None => state.generic.record_failure(&proxy), + } + } + passed + }) +} + +/// Checks every registered site against all fast exit IPs (not only the top +/// 20) plus its own current set: many proxies that pass the trace check cannot +/// reach a given site (DDoS-Guard drops whole datacenter ranges), so the site's +/// set is filled from the wider list. A proxy evicted from a site is replaced +/// by the next site-checked one at the latest on the following round. +async fn check_sites(concurrency: usize) { + use futures::stream::{self, StreamExt}; + let jobs: Vec<(String, String, Vec)> = with_state(|state| { + let now = Instant::now(); + let fast = state.generic.ranked(now, usize::MAX); + state + .sites + .iter() + .map(|(host, site)| { + let mut proxies = fast.clone(); + for proxy in site.ranking.ranked(now, usize::MAX) { + if !proxies.contains(&proxy) { + proxies.push(proxy); + } + } + (host.clone(), site.check_url.clone(), proxies) + }) + .collect() + }); + for (host, check_url, proxies) in jobs { + // Two rounds: a site proxy also needs two successful checks. + for _ in 0..2 { + let results: Vec<(Proxy, Result)> = stream::iter(proxies.clone()) + .map(|proxy| { + let check_url = check_url.clone(); + async move { + let secs = check(Some(&proxy), &check_url, CHECK_TIMEOUT) + .await + .map(|(secs, _)| secs); + (proxy, secs) + } + }) + .buffer_unordered(concurrency) + .collect() + .await; + with_state(|state| { + let exit_ips: HashMap> = results + .iter() + .map(|(proxy, _)| (proxy.to_string(), state.generic.exit_ip(proxy))) + .collect(); + let Some(site) = state.sites.get_mut(&host) else { + return; + }; + let now = Instant::now(); + for (proxy, result) in &results { + match result { + Ok(secs) => { + let ip = exit_ips.get(&proxy.to_string()).cloned().flatten(); + site.ranking.record_success(proxy, *secs, ip.as_deref()); + } + // The proxy works; its IP is just throttled there. + Err(CheckError::Throttled) => { + site.ranking.bench(proxy, now + THROTTLE_COOLDOWN) + } + Err(CheckError::Failed) => site.ranking.record_failure(proxy), + } + } + site.ranking.prune(); + }); + } + let ready = with_state(|state| { + state + .sites + .get(&host) + .map(|site| site.ranking.ranked(Instant::now(), SITE_SET_SIZE).len()) + .unwrap_or(0) + }); + println!("proxy pool: {host}: {ready} proxies ready"); + } +} + +fn check_client() -> &'static wreq::Client { + static CLIENT: std::sync::OnceLock = std::sync::OnceLock::new(); + // Certificates are verified: public proxies are untrusted, and one that + // tampers with TLS must fail its check rather than feed us altered pages. + CLIENT.get_or_init(|| { + wreq::Client::builder() + .emulation(wreq_util::Emulation::Firefox151) + .build() + .expect("proxy pool check client") + }) +} + +enum CheckError { + /// 429 from the target: the proxy works, its IP is rate-limited there. + Throttled, + Failed, +} + +/// GET `url` (through `proxy`, if any) within `timeout`. +/// Returns the wall time and the body of a 2xx answer. +async fn check( + proxy: Option<&Proxy>, + url: &str, + timeout: Duration, +) -> Result<(f64, String), CheckError> { + let mut request = check_client().get(url).timeout(timeout); + if let Some(proxy) = proxy { + request = request.proxy(proxy.to_wreq_proxy().map_err(|_| CheckError::Failed)?); + } + let started = Instant::now(); + let response = request.send().await.map_err(|_| CheckError::Failed)?; + match response.status().as_u16() { + 429 => return Err(CheckError::Throttled), + status if !(200..300).contains(&status) => return Err(CheckError::Failed), + _ => {} + } + let body = response.text().await.map_err(|_| CheckError::Failed)?; + Ok((started.elapsed().as_secs_f64(), body)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn proxy(s: &str) -> Proxy { + Proxy::parse(s).unwrap() + } + + #[test] + fn a_proxy_needs_two_successful_checks_to_rank() { + let mut ranking = Ranking::default(); + let p = proxy("socks5://10.0.0.1:1080"); + ranking.record_success(&p, 0.4, Some("1.1.1.1")); + assert!(ranking.ranked(Instant::now(), 20).is_empty()); + ranking.record_success(&p, 0.5, Some("1.1.1.1")); + assert_eq!(ranking.ranked(Instant::now(), 20), vec![p]); + } + + #[test] + fn ranks_by_median_latency_and_drops_slow_proxies() { + let mut ranking = Ranking::default(); + let (fast, medium, slow) = ( + proxy("http://10.0.0.1:80"), + proxy("http://10.0.0.2:80"), + proxy("http://10.0.0.3:80"), + ); + // medium has one fast outlier, but its median is 0.9s. + for (p, samples, ip) in [ + (&medium, [0.1, 0.9, 1.0], "2.2.2.2"), + (&slow, [1.4, 1.6, 2.0], "3.3.3.3"), + (&fast, [0.3, 0.4, 0.2], "1.1.1.1"), + ] { + for secs in samples { + ranking.record_success(p, secs, Some(ip)); + } + } + assert_eq!(ranking.ranked(Instant::now(), 20), vec![fast.clone(), medium]); + assert_eq!(ranking.ranked(Instant::now(), 1), vec![fast]); + } + + #[test] + fn keeps_only_the_fastest_proxy_per_exit_ip() { + let mut ranking = Ranking::default(); + let port_a = proxy("http://147.161.246.39:10878"); + let port_b = proxy("http://147.161.246.39:11997"); + let other = proxy("socks5://5.5.5.5:1080"); + for (p, secs, ip) in [ + (&port_a, 0.6, "147.161.247.7"), + (&port_b, 0.4, "147.161.247.7"), + (&other, 0.9, "5.5.5.5"), + ] { + ranking.record_success(p, secs, Some(ip)); + ranking.record_success(p, secs, Some(ip)); + } + assert_eq!(ranking.ranked(Instant::now(), 20), vec![port_b, other]); + } + + #[test] + fn two_failures_in_a_row_evict_until_the_next_success() { + let mut ranking = Ranking::default(); + let p = proxy("http://10.0.0.1:80"); + ranking.record_success(&p, 0.3, Some("1.1.1.1")); + ranking.record_success(&p, 0.3, Some("1.1.1.1")); + ranking.record_failure(&p); + assert_eq!(ranking.ranked(Instant::now(), 20), vec![p.clone()], "one failure is tolerated"); + ranking.record_failure(&p); + assert!(ranking.ranked(Instant::now(), 20).is_empty()); + ranking.record_success(&p, 0.3, Some("1.1.1.1")); + assert_eq!(ranking.ranked(Instant::now(), 20), vec![p]); + } + + fn site_with(proxies: &[&Proxy]) -> Site { + let mut site = Site::new("https://example.com/robots.txt"); + for (i, p) in proxies.iter().enumerate() { + let ip = format!("9.9.9.{i}"); + site.ranking.record_success(p, 0.2 + i as f64 / 10.0, Some(&ip)); + site.ranking.record_success(p, 0.2 + i as f64 / 10.0, Some(&ip)); + } + site + } + + #[test] + fn site_requests_rotate_evenly_over_its_proxies() { + let (a, b, c) = ( + proxy("http://10.0.0.1:80"), + proxy("http://10.0.0.2:80"), + proxy("http://10.0.0.3:80"), + ); + let mut site = site_with(&[&a, &b, &c]); + let now = Instant::now(); + let picks: Vec = (0..6).filter_map(|_| site.next(now)).collect(); + assert_eq!(picks, vec![a.clone(), b.clone(), c.clone(), a, b, c]); + } + + #[test] + fn a_throttled_proxy_sits_out_its_cooldown_for_that_site() { + let (a, b) = (proxy("http://10.0.0.1:80"), proxy("http://10.0.0.2:80")); + let mut site = site_with(&[&a, &b]); + let now = Instant::now(); + site.report(&a, Outcome::Throttled, now); + let picks: Vec = (0..3).filter_map(|_| site.next(now)).collect(); + assert_eq!(picks, vec![b.clone(), b.clone(), b.clone()]); + let later = now + THROTTLE_COOLDOWN + std::time::Duration::from_secs(1); + let picks: Vec = (0..2).filter_map(|_| site.next(later)).collect(); + assert!(picks.contains(&a) && picks.contains(&b)); + } + + #[test] + fn list_lines_take_their_protocol_from_the_list_name_unless_they_carry_one() { + let body = "1.2.3.4:1080\n\n socks4://5.6.7.8:4145 \nnot a proxy line:::\n"; + assert_eq!( + parse_list("https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt", body), + vec![proxy("socks5://1.2.3.4:1080"), proxy("socks4://5.6.7.8:4145")] + ); + } + + #[test] + fn reads_the_exit_ip_from_a_cloudflare_trace() { + let body = "fl=12f1\nh=www.cloudflare.com\nip=147.161.247.7\nts=1730790000.1\n"; + assert_eq!(trace_exit_ip(body).as_deref(), Some("147.161.247.7")); + assert_eq!(trace_exit_ip("hijacked"), None); + } + + #[test] + fn status_rows_never_show_proxy_credentials() { + let mut ranking = Ranking::default(); + let p = proxy("http://user:secret@10.0.0.1:8080"); + ranking.record_success(&p, 0.25, Some("1.1.1.1")); + ranking.record_success(&p, 0.35, Some("1.1.1.1")); + let rows = ranking.status(Instant::now(), 20); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].proxy, "http://10.0.0.1:8080"); + assert_eq!(rows[0].exit_ip.as_deref(), Some("1.1.1.1")); + assert_eq!(rows[0].median_ms, 350); + assert!(!serde_json::to_string(&rows).unwrap().contains("secret")); + } +} diff --git a/src/util/requester.rs b/src/util/requester.rs index e5d015b..e5b491c 100644 --- a/src/util/requester.rs +++ b/src/util/requester.rs @@ -311,18 +311,45 @@ impl Requester { let cookie_jar = Self::shared_cookie_jar(); let client = Self::build_client(cookie_jar.clone(), None); - let requester = Requester { + Requester { client, cookie_jar, debug_trace_id: None, proxy: false, flaresolverr_session: None, user_agent: None, - }; + } + } - proxy::init_all_proxies_background(requester.clone()); - - requester + /// GET through one pool proxy, as part of this requester's session (same + /// cookie jar and user agent). Unlike the direct paths this verifies + /// certificates: a public proxy must not be able to alter what we read. + /// It deliberately builds its own client and skips the Jina/FlareSolverr + /// fallbacks: a pooled request must leave from the proxy's IP or fail. + pub async fn get_via( + &self, + via: &proxy::Proxy, + url: &str, + headers: Vec<(String, String)>, + timeout: Duration, + ) -> Result { + let mut builder = Client::builder() + .emulation(Emulation::Firefox151) + .cookie_provider(self.cookie_jar.clone()) + .redirect(Policy::default()); + if let Some(user_agent) = self.user_agent.as_deref() { + let mut default_headers = HeaderMap::new(); + if let Ok(value) = HeaderValue::from_str(user_agent) { + default_headers.insert(USER_AGENT, value); + builder = builder.default_headers(default_headers); + } + } + let client = builder.build()?; + let mut request = client.get(url).timeout(timeout).proxy(via.to_wreq_proxy()?); + for (key, value) in headers { + request = request.header(key, value); + } + Ok(request.send().await?) } pub fn set_proxy(&mut self, proxy: bool) { diff --git a/src/util/rule34video_resolve.rs b/src/util/rule34video_resolve.rs index 89bc137..8a05e18 100644 --- a/src/util/rule34video_resolve.rs +++ b/src/util/rule34video_resolve.rs @@ -56,6 +56,25 @@ pub fn parse_formats(html: &str) -> Vec<(u32, String)> { formats } +/// Categories then tags from the page's `flashvars`, deduplicated. +pub fn parse_tags(html: &str) -> Vec { + let Some(start) = html.find("var flashvars") else { + return vec![]; + }; + let block = &html[start..]; + let block = &block[..block.len().min(6000)]; + let mut tags: Vec = Vec::new(); + for key in ["video_categories", "video_tags"] { + for tag in flashvar(block, key).unwrap_or_default().split(',') { + let tag = tag.trim(); + if !tag.is_empty() && !tags.iter().any(|t| t.eq_ignore_ascii_case(tag)) { + tags.push(tag.to_string()); + } + } + } + tags +} + /// `{id}/{slug}[/]` -> the video page URL. Only digits, letters, `-` and `_` /// are accepted so the endpoint cannot steer the request elsewhere. pub fn page_url(endpoint: &str) -> Option { @@ -114,4 +133,89 @@ mod tests { assert_eq!(page_url("abc/slug"), None); assert_eq!(page_url("123/slug/extra"), None); } + + fn env_usize(name: &str, default: usize) -> usize { + std::env::var(name) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) + } + + /// Bursts `/embed/{id}` requests to see whether they share the detail pages' + /// rate limit (~500 requests in ~100s from one IP -> 429). Walks down from the + /// newest id in the RSS feed; gaps (deleted videos) still count as requests. + /// + /// `R34_EMBED_REQUESTS` (default 600) and `R34_EMBED_CONCURRENCY` (default 8) + /// tune the burst. Fails on the first 429 and reports when it happened. + /// + /// cargo test embed_burst -- --ignored --nocapture + #[tokio::test] + #[ignore = "live rate-limit probe against rule34video.com"] + async fn embed_burst_does_not_trigger_429() { + use futures::stream::{self, StreamExt}; + use std::collections::BTreeMap; + use std::time::Instant; + + let total = env_usize("R34_EMBED_REQUESTS", 600); + let concurrency = env_usize("R34_EMBED_CONCURRENCY", 8); + let requester = Requester::new(); + + let rss = requester + .clone() + .get_raw(&format!("{BASE_URL}/rss/")) + .await + .expect("rss request") + .text() + .await + .expect("rss body"); + let newest: u64 = rss + .split("/video/") + .nth(1) + .and_then(|rest| rest.split('/').next()) + .and_then(|id| id.parse().ok()) + .expect("newest video id from rss"); + + let started = Instant::now(); + let results: Vec<(usize, u16, f64)> = stream::iter(0..total) + .map(|i| { + let mut requester = requester.clone(); + let url = format!("{BASE_URL}/embed/{}", newest - i as u64); + async move { + let status = match requester + .get_raw_with_headers(&url, vec![("Referer".to_string(), format!("{BASE_URL}/"))]) + .await + { + Ok(response) => response.status().as_u16(), + Err(_) => 0, + }; + (i, status, started.elapsed().as_secs_f64()) + } + }) + .buffer_unordered(concurrency) + .collect() + .await; + let elapsed = started.elapsed().as_secs_f64(); + + let mut by_status: BTreeMap = BTreeMap::new(); + for (_, status, _) in &results { + *by_status.entry(*status).or_default() += 1; + } + let first_429 = results + .iter() + .filter(|(_, status, _)| *status == 429) + .min_by(|a, b| a.2.total_cmp(&b.2)); + let completed_before_429 = first_429.map(|(_, _, at)| { + results.iter().filter(|(_, s, t)| *s != 429 && t < at).count() + }); + + println!( + "{total} embed requests from id {newest} down, concurrency {concurrency}: \ + {elapsed:.1}s ({:.1} req/s); statuses {by_status:?}", + total as f64 / elapsed + ); + if let (Some((i, _, at)), Some(before)) = (first_429, completed_before_429) { + println!("first 429 at {at:.1}s (request #{i}), after {before} successful requests"); + } + assert!(first_429.is_none(), "embed burst was rate-limited: {by_status:?}"); + } }