From 979ac1a39be002436f5877757319c0a17dfb1c2f Mon Sep 17 00:00:00 2001 From: forkwright Date: Thu, 2 Jul 2026 09:06:53 -0500 Subject: [PATCH] fix(zetesis): harden indexer search (SSRF guard, key redaction, body cap, tx) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - #429: a cancelled HTTP fetch returns a distinct Cancelled error instead of ParseResponse, no longer degrading every in-flight indexer. - #430: fetch_xml enforces a configurable max_response_body_bytes cap (default 16 MiB) on attacker-controlled bodies. - #431: RateLimited honors Retry-After so back-off engages. - #432: upsert_indexer_categories runs its DELETE + INSERTs in one transaction. - #433: indexer API keys are redacted from logged URLs. - #434: torrent download URLs are SSRF-validated — http(s) only, host DNS-resolved and every resolved address rejected if private/loopback/ link-local/CGNAT; fail-closed on resolution failure. - #459: NewznabClient/TorznabClient HTTP status + auth-failure paths tested. - #460: zetesis repo.rs gains an in-memory-sqlite test module. - #467: archon wires the ByparrProxy Cloudflare bypass from config instead of hardwiring NoProxy. Also resolves two quick-xml advisories published 2026-07-02 (RUSTSEC-2026-0194 quadratic-attribute DoS, RUSTSEC-2026-0195 NsReader namespace DoS): bump the direct indexer-XML parser to quick-xml 0.41 and ignore the transitively-pinned 0.37 (feed-rs, librqbit-upnp) with review-by, derived across deny/audit/osv. Closes #429 Closes #430 Closes #431 Closes #432 Closes #433 Closes #434 Closes #459 Closes #460 Closes #467 Gate-Passed: kanon 0.1.5 +stages:fmt,check,clippy,nextest,lint sha:fb78c8d064530f3d0387a4a12fc6cc9cfd1a89ba --- .cargo/audit.toml | 54 ++-- Cargo.lock | 7 +- Cargo.toml | 2 +- crates/archon/src/serve.rs | 136 +++++++++- crates/horismos/src/lib.rs | 37 +++ crates/horismos/src/subsystems.rs | 6 + crates/horismos/src/validation.rs | 20 ++ crates/zetesis/Cargo.toml | 6 +- crates/zetesis/src/cf_bypass/byparr.rs | 22 +- crates/zetesis/src/cf_bypass/noop.rs | 3 +- crates/zetesis/src/client/mod.rs | 318 ++++++++++++++++++++++- crates/zetesis/src/client/newznab.rs | 236 ++++++++++++++++- crates/zetesis/src/client/torznab.rs | 282 ++++++++++++++++++++- crates/zetesis/src/error.rs | 24 ++ crates/zetesis/src/lib.rs | 1 + crates/zetesis/src/repo.rs | 337 ++++++++++++++++++++++++- crates/zetesis/src/search.rs | 312 +++++++++++++++++++++-- crates/zetesis/src/test_support.rs | 80 ++++++ deny.toml | 12 +- docs/architecture/configuration.md | 3 + osv-scanner.toml | 31 ++- 21 files changed, 1800 insertions(+), 129 deletions(-) create mode 100644 crates/zetesis/src/test_support.rs diff --git a/.cargo/audit.toml b/.cargo/audit.toml index d371ee5c..49ca2537 100644 --- a/.cargo/audit.toml +++ b/.cargo/audit.toml @@ -1,49 +1,33 @@ -# cargo-audit configuration -# https://github.com/rustsec/rustsec/blob/main/cargo-audit/README.md +# cargo-audit configuration. # -# WHY: Mirrors `deny.toml` `[advisories].ignore` so that `cargo audit` -# (run with `--deny unmaintained --deny unsound --deny yanked`) and -# `cargo deny check advisories` agree on which RUSTSEC IDs are -# pre-acknowledged. Each suppression must cite the dep chain and the -# reason no safer alternative is available; revisit at the next MSRV -# bump or when upstream publishes a fix. +# WHY: This file is DERIVED from deny.toml [[advisories.ignore]]. +# Do not edit the ignore list here; edit deny.toml and run +# `kanon audit derive-ignores --apply`. See standards/SUPPLY-CHAIN.md. [advisories] ignore = [ - # WHY: rsa timing side-channel via jsonwebtoken/exousia — no safe - # upgrade, local-only use. Pre-existing harmonia ignore. + # rsa timing side-channel via jsonwebtoken/exousia — no safe upgrade, local-only use "RUSTSEC-2023-0071", - - # WHY: backoff unmaintained — transitive dep via librqbit. + # backoff unmaintained — transitive dep via librqbit "RUSTSEC-2025-0012", - - # WHY: bincode unmaintained — transitive dep via librqbit-peer-protocol. + # bincode unmaintained — transitive dep via librqbit-peer-protocol "RUSTSEC-2025-0141", - - # WHY: crypto-hash unmaintained — transitive dep via - # librqbit-sha1-wrapper. + # crypto-hash unmaintained — transitive dep via librqbit-sha1-wrapper "RUSTSEC-2025-0060", - - # WHY: instant unmaintained — transitive dep via backoff/librqbit, - # no maintained alternative on the librqbit chain. + # instant unmaintained — transitive dep via backoff/librqbit, no alternative "RUSTSEC-2024-0384", - - # WHY: paste unmaintained (dtolnay archived) — transitive dep via - # lofty/taxis. + # paste unmaintained (dtolnay archived) — transitive dep via lofty/taxis "RUSTSEC-2024-0436", - - # WHY: time RFC2822 stack exhaustion (RUSTSEC-2026-0009) fix - # requires Rust 1.88; harmonia CI is pinned to Rust 1.85 until MSRV - # policy changes. + # time RFC2822 stack exhaustion fix requires Rust 1.88; Harmonia CI is pinned to Rust 1.85 until MSRV policy changes "RUSTSEC-2026-0009", - - # WHY: audiopus_sys unmaintained via opus/akouo-core; no safe - # upgrade available. + # audiopus_sys unmaintained via opus/akouo-core; no safe upgrade available "RUSTSEC-2026-0150", - - # WHY: rand unsound with custom logger using `rand::rng()` - # (RUSTSEC-2026-0097) — surfaces via sqlx/axum/quinn/reqwest; - # harmonia does not install a custom logger that re-enters - # `rand::rng()`, so the unsound path is unreachable. + # rand unsound with custom logger using `rand::rng()` — surfaces via sqlx/axum/quinn/reqwest; harmonia installs no custom logger that re-enters rand::rng(), so the unsound path is unreachable "RUSTSEC-2026-0097", + # quick-xml < 0.41 quadratic-attribute DoS. The direct indexer-XML parser (zetesis) is on 0.41; the remaining 0.37.5 is transitively pinned by feed-rs (RSS/podcast) and librqbit-upnp (LAN UPnP), neither of which yet targets 0.41. No safe upgrade until upstream bumps. + # review-by: 2026-10-01 + "RUSTSEC-2026-0194", + # quick-xml < 0.41 NsReader namespace-allocation DoS. Same transitive pin as RUSTSEC-2026-0194: the direct zetesis parser is on 0.41; feed-rs and librqbit-upnp still require 0.37. No safe upgrade until upstream bumps. + # review-by: 2026-10-01 + "RUSTSEC-2026-0195", ] diff --git a/Cargo.lock b/Cargo.lock index e107dd95..2a9cc233 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3931,9 +3931,9 @@ dependencies = [ [[package]] name = "quick-xml" -version = "0.40.1" +version = "0.41.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2474bd2e5029e7ccb6abb2ba48cf2383a333851dedf495901544281590c7da7f" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" dependencies = [ "memchr", "serde", @@ -7275,7 +7275,7 @@ dependencies = [ "futures", "horismos", "jiff", - "quick-xml 0.40.1", + "quick-xml 0.41.0", "reqwest", "rstest", "serde", @@ -7286,6 +7286,7 @@ dependencies = [ "tokio", "tokio-util", "tracing", + "url", "uuid", ] diff --git a/Cargo.toml b/Cargo.toml index 74097a9d..5b344cc4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -180,7 +180,7 @@ clap = { version = "4", features = ["derive"] } # ── XML ─────────────────────────────────────────────────────────────────────── -quick-xml = { version = "0.40", features = ["serialize"] } +quick-xml = { version = "0.41", features = ["serialize"] } # ── Concurrency ────────────────────────────────────────────────────────────── dashmap = "6" diff --git a/crates/archon/src/serve.rs b/crates/archon/src/serve.rs index 7393142d..00bbb177 100644 --- a/crates/archon/src/serve.rs +++ b/crates/archon/src/serve.rs @@ -30,6 +30,8 @@ use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tracing::{Instrument, info}; use zetesis::SearchIndexerService; +use zetesis::cf_bypass::CloudflareProxy; +use zetesis::cf_bypass::byparr::ByparrProxy; use zetesis::cf_bypass::noop::NoProxy; use crate::cli::ServeArgs; @@ -656,14 +658,18 @@ pub async fn run_serve(args: ServeArgs, out: &mut impl Write) -> Result<(), Host let shutdown_token = CancellationToken::new(); // Layer 0: Zetesis (indexer protocol) + let cf_proxy = build_cf_proxy(&config.zetesis)?; let zetesis = Arc::new(SearchIndexerService::new( db.read.clone(), db.write.clone(), - Arc::new(NoProxy), + cf_proxy, config.zetesis.clone(), event_tx.clone(), )); - info!("zetesis (indexer search) initialized"); + info!( + cloudflare_bypass = config.zetesis.cloudflare_bypass_enabled, + "zetesis (indexer search) initialized" + ); // Layer 1: Ergasia (download execution) let ergasia_session = Arc::new( @@ -872,6 +878,36 @@ fn resolve_listen_addr(listen_addr: &str, port: u16) -> Result Result, HostError> { + if !config.cloudflare_bypass_enabled { + return Ok(Arc::new(NoProxy)); + } + let url = config + .cf_proxy_url + .as_deref() + .map(str::trim) + .filter(|url| !url.is_empty()) + .ok_or_else(|| HostError::Config { + source: horismos::HorismosError::Validation { + message: "zetesis.cloudflare_bypass_enabled requires zetesis.cf_proxy_url" + .to_string(), + location: snafu::location!(), + }, + location: snafu::location!(), + })?; + Ok(Arc::new(ByparrProxy::new( + url.to_string(), + std::time::Duration::from_secs(config.cf_proxy_timeout_seconds), + ))) +} + fn validate_download_dir(config: &horismos::Config) -> Result<(), HostError> { let dir = &config.ergasia.download_dir; if !dir.exists() { @@ -1061,6 +1097,102 @@ mod tests { assert_eq!(result, Err("import pipeline not wired".to_string())); } + // ── Cloudflare-bypass proxy wiring ────────────────────────────────────── + + #[tokio::test] + async fn build_cf_proxy_disabled_returns_no_proxy() { + let config = horismos::SearchSubsystemConfig::default(); + assert!(!config.cloudflare_bypass_enabled); + + let proxy = build_cf_proxy(&config).expect("disabled bypass builds NoProxy"); + let err = proxy + .get("https://example.com", CancellationToken::new()) + .await + .expect_err("NoProxy always errors"); + assert!(matches!( + err, + zetesis::SearchIndexerError::NoCfBypass { .. } + )); + } + + #[tokio::test] + async fn build_cf_proxy_enabled_without_url_errors() { + let config = horismos::SearchSubsystemConfig { + cloudflare_bypass_enabled: true, + cf_proxy_url: None, + ..horismos::SearchSubsystemConfig::default() + }; + + let Err(err) = build_cf_proxy(&config) else { + panic!("enabled bypass without URL must fail loud"); + }; + assert!(matches!(err, HostError::Config { .. })); + assert!(err.to_string().contains("cf_proxy_url")); + } + + #[tokio::test] + async fn build_cf_proxy_enabled_with_blank_url_errors() { + let config = horismos::SearchSubsystemConfig { + cloudflare_bypass_enabled: true, + cf_proxy_url: Some(" ".to_string()), + ..horismos::SearchSubsystemConfig::default() + }; + + let Err(err) = build_cf_proxy(&config) else { + panic!("blank URL must fail loud"); + }; + assert!(err.to_string().contains("cf_proxy_url")); + } + + #[tokio::test] + async fn build_cf_proxy_enabled_posts_to_byparr_endpoint() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + // One-shot Byparr stub: answers a single POST /v1 with a solved page. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind stub listener"); + let endpoint = format!("http://{}", listener.local_addr().expect("local addr")); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept"); + let mut buf = vec![0u8; 8192]; + let n = stream.read(&mut buf).await.expect("read request head"); + let request = String::from_utf8_lossy(&buf[..n]).into_owned(); + let body = r#"{"status":"ok","message":"solved","solution":{"url":"https://example.com","status":200,"response":"ok","cookies":[],"userAgent":"UA"}}"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + stream + .write_all(response.as_bytes()) + .await + .expect("write response"); + request + }); + + let config = horismos::SearchSubsystemConfig { + cloudflare_bypass_enabled: true, + cf_proxy_url: Some(endpoint), + cf_proxy_timeout_seconds: 5, + ..horismos::SearchSubsystemConfig::default() + }; + + let proxy = build_cf_proxy(&config).expect("enabled bypass builds ByparrProxy"); + let response = proxy + .get("https://cf-protected.example.com", CancellationToken::new()) + .await + .expect("Byparr stub answers"); + + assert_eq!(response.status, 200); + assert_eq!(response.body, "ok"); + let request = server.await.expect("stub request captured"); + assert!( + request.starts_with("POST /v1"), + "expected POST /v1, got: {}", + request.lines().next().unwrap_or_default() + ); + } + #[tokio::test] async fn subtitle_adapter_calls_live_prostheke_for_movie_path() { let pool = SqlitePool::connect("sqlite::memory:") diff --git a/crates/horismos/src/lib.rs b/crates/horismos/src/lib.rs index 92035e77..96fd3ff8 100644 --- a/crates/horismos/src/lib.rs +++ b/crates/horismos/src/lib.rs @@ -84,6 +84,7 @@ mod tests { assert_eq!(config.aggelia.buffer_size, 1024); assert_eq!(config.aggelia.download_queue_size, 512); assert_eq!(config.zetesis.request_timeout_secs, 30); + assert_eq!(config.zetesis.max_response_body_bytes, 16 * 1024 * 1024); assert_eq!(config.epignosis.cache_ttl_secs, 86400); assert_eq!(config.kritike.scan_interval_hours, 24); } @@ -274,6 +275,42 @@ mod tests { assert!(err.to_string().contains("port")); } + // ── Zetesis limits validation ───────────────────────────────────────────── + + #[test] + fn validation_rejects_zero_max_response_body_bytes() { + let mut config = config_with_jwt(valid_jwt_secret()); + config.zetesis.max_response_body_bytes = 0; + let err = validate_config(&config).unwrap_err(); + assert!(err.to_string().contains("max_response_body_bytes")); + } + + #[test] + fn validation_rejects_cf_bypass_without_proxy_url() { + let mut config = config_with_jwt(valid_jwt_secret()); + config.zetesis.cloudflare_bypass_enabled = true; + config.zetesis.cf_proxy_url = None; + let err = validate_config(&config).unwrap_err(); + assert!(err.to_string().contains("cf_proxy_url")); + } + + #[test] + fn validation_rejects_cf_bypass_with_blank_proxy_url() { + let mut config = config_with_jwt(valid_jwt_secret()); + config.zetesis.cloudflare_bypass_enabled = true; + config.zetesis.cf_proxy_url = Some(" ".to_string()); + let err = validate_config(&config).unwrap_err(); + assert!(err.to_string().contains("cf_proxy_url")); + } + + #[test] + fn validation_accepts_cf_bypass_with_proxy_url() { + let mut config = config_with_jwt(valid_jwt_secret()); + config.zetesis.cloudflare_bypass_enabled = true; + config.zetesis.cf_proxy_url = Some("http://byparr:8191".to_string()); + assert!(validate_config(&config).is_ok()); + } + // ── Serialize/Deserialize roundtrip ─────────────────────────────────────── #[test] diff --git a/crates/horismos/src/subsystems.rs b/crates/horismos/src/subsystems.rs index af5a0b5c..e3a52cbf 100644 --- a/crates/horismos/src/subsystems.rs +++ b/crates/horismos/src/subsystems.rs @@ -221,6 +221,11 @@ impl Default for AggeliaConfig { pub struct SearchSubsystemConfig { pub request_timeout_secs: u64, pub max_results_per_indexer: usize, + /// Byte cap on any single indexer HTTP response body. + /// + /// WHY: indexer responses are third-party data — an unbounded read lets a + /// hostile or broken indexer exhaust memory with one response. + pub max_response_body_bytes: u64, pub cloudflare_bypass_enabled: bool, pub max_concurrent_searches: usize, pub per_indexer_rate_limit_requests: u32, @@ -239,6 +244,7 @@ impl Default for SearchSubsystemConfig { Self { request_timeout_secs: 30, max_results_per_indexer: 100, + max_response_body_bytes: 16 * 1024 * 1024, cloudflare_bypass_enabled: false, max_concurrent_searches: 10, per_indexer_rate_limit_requests: 5, diff --git a/crates/horismos/src/validation.rs b/crates/horismos/src/validation.rs index e566ddf5..8c32a99f 100644 --- a/crates/horismos/src/validation.rs +++ b/crates/horismos/src/validation.rs @@ -35,6 +35,26 @@ fn validate_limits(config: &Config) -> Result<(), HorismosError> { } .fail(); } + if config.zetesis.max_response_body_bytes == 0 { + return ValidationSnafu { + message: "zetesis.max_response_body_bytes must be greater than 0".to_string(), + } + .fail(); + } + if config.zetesis.cloudflare_bypass_enabled + && config + .zetesis + .cf_proxy_url + .as_deref() + .is_none_or(|url| url.trim().is_empty()) + { + return ValidationSnafu { + message: + "zetesis.cf_proxy_url must be set when zetesis.cloudflare_bypass_enabled is true" + .to_string(), + } + .fail(); + } Ok(()) } diff --git a/crates/zetesis/Cargo.toml b/crates/zetesis/Cargo.toml index 66eb2726..8ba38021 100644 --- a/crates/zetesis/Cargo.toml +++ b/crates/zetesis/Cargo.toml @@ -20,13 +20,17 @@ dashmap.workspace = true tokio-util.workspace = true futures.workspace = true bytes.workspace = true +url.workspace = true uuid.workspace = true jiff.workspace = true sqlx.workspace = true [dev-dependencies] rstest.workspace = true -tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } + +[dev-dependencies.tokio] +workspace = true +features = ["rt-multi-thread", "macros", "test-util"] [package.metadata.kanon] maturity = "alpha" diff --git a/crates/zetesis/src/cf_bypass/byparr.rs b/crates/zetesis/src/cf_bypass/byparr.rs index fdad6497..664d46d1 100644 --- a/crates/zetesis/src/cf_bypass/byparr.rs +++ b/crates/zetesis/src/cf_bypass/byparr.rs @@ -109,27 +109,29 @@ impl ByparrProxy { let endpoint_url = format!("{}/v1", self.endpoint.trim_end_matches('/')); + // WHY: errors embed the redacted form so the API key never reaches a + // log line; the request itself still carries the real URL. + let redacted = crate::client::redact_api_key(url); let response = tokio::select! { result = self.client.post(&endpoint_url).json(&req).send() => { - result.context(error::HttpRequestSnafu { url })? + result.context(error::HttpRequestSnafu { url: redacted.clone() })? } () = ct.cancelled() => { - return Err(SearchIndexerError::CfProxyTimeout { - url: url.to_string(), - timeout: self.timeout.as_secs() as u32, + return Err(SearchIndexerError::Cancelled { + url: redacted, location: snafu::Location::new(file!(), line!(), column!()), }); } }; - let byparr_resp: ByparrResponse = response - .json() - .await - .context(error::HttpRequestSnafu { url })?; + let byparr_resp: ByparrResponse = + response.json().await.context(error::HttpRequestSnafu { + url: redacted.clone(), + })?; if byparr_resp.status != "ok" { return Err(SearchIndexerError::CfProxyError { - url: url.to_string(), + url: redacted.clone(), status: byparr_resp.status, message: byparr_resp.message, location: snafu::Location::new(file!(), line!(), column!()), @@ -139,7 +141,7 @@ impl ByparrProxy { let solution = byparr_resp .solution .ok_or_else(|| SearchIndexerError::CfProxyError { - url: url.to_string(), + url: redacted, status: "ok".to_string(), message: "no solution in response".to_string(), location: snafu::Location::new(file!(), line!(), column!()), diff --git a/crates/zetesis/src/cf_bypass/noop.rs b/crates/zetesis/src/cf_bypass/noop.rs index 4774197d..2e56d2ac 100644 --- a/crates/zetesis/src/cf_bypass/noop.rs +++ b/crates/zetesis/src/cf_bypass/noop.rs @@ -14,7 +14,8 @@ impl CloudflareProxy for NoProxy { url: &str, _ct: CancellationToken, ) -> Pin> + Send + '_>> { - let url = url.to_string(); + // WHY: redact at construction so the API key never reaches a log line. + let url = crate::client::redact_api_key(url); Box::pin(async move { Err(SearchIndexerError::NoCfBypass { url, diff --git a/crates/zetesis/src/client/mod.rs b/crates/zetesis/src/client/mod.rs index fc98a263..2411c137 100644 --- a/crates/zetesis/src/client/mod.rs +++ b/crates/zetesis/src/client/mod.rs @@ -2,9 +2,14 @@ pub mod newznab; pub mod torznab; pub mod xml; +use std::net::IpAddr; + +use futures::StreamExt; +use snafu::ResultExt; use tokio_util::sync::CancellationToken; +use url::{Host, Url}; -use crate::error::SearchIndexerError; +use crate::error::{self, SearchIndexerError}; use crate::types::{DownloadResponse, IndexerCaps, IndexerStatus, SearchQuery, SearchResult}; pub trait IndexerClient: Send + Sync { @@ -146,6 +151,158 @@ pub(crate) fn build_caps_url(config: &IndexerConfig) -> String { url } +/// Replaces every `apikey=` query value in `url` with `[REDACTED]`. +/// +/// WHY: indexer URLs built by [`build_search_url`]/[`build_caps_url`] embed +/// the API key as a query parameter; redacting at error-construction time +/// keeps the key out of every downstream Display/log path. +pub(crate) fn redact_api_key(url: &str) -> String { + let mut out = String::with_capacity(url.len()); + let mut rest = url; + while let Some((before, after)) = rest.split_once("apikey=") { + out.push_str(before); + out.push_str("apikey=[REDACTED]"); + match after.split_once('&') { + Some((_value, tail)) => { + out.push('&'); + rest = tail; + } + None => rest = "", + } + } + out.push_str(rest); + out +} + +/// Reads a response body into a UTF-8 string, enforcing `max_bytes`. +/// +/// Rejects on a declared `Content-Length` above the cap before reading any +/// body bytes, then enforces the cap again while streaming. +/// +/// WHY: `Content-Length` is attacker-controlled and may be absent or wrong; +/// only a running counter over the actual stream bounds allocation. +pub(crate) async fn read_body_bounded( + response: reqwest::Response, + url: &str, + max_bytes: u64, +) -> Result { + if let Some(declared) = response.content_length() + && declared > max_bytes + { + return Err(SearchIndexerError::ResponseTooLarge { + url: redact_api_key(url), + size: declared, + limit: max_bytes, + location: snafu::Location::new(file!(), line!(), column!()), + }); + } + + let mut body: Vec = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.context(error::HttpRequestSnafu { + url: redact_api_key(url), + })?; + let received = body.len() as u64 + chunk.len() as u64; + if received > max_bytes { + return Err(SearchIndexerError::ResponseTooLarge { + url: redact_api_key(url), + size: received, + limit: max_bytes, + location: snafu::Location::new(file!(), line!(), column!()), + }); + } + body.extend_from_slice(&chunk); + } + + String::from_utf8(body).map_err(|e| SearchIndexerError::ParseResponse { + url: redact_api_key(url), + error: e.to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + }) +} + +/// SSRF guard for attacker-supplied download URLs. +/// +/// Accepts only http(s), requires a host, and rejects any target whose IP — +/// literal or DNS-resolved — is loopback, private, link-local, CGNAT, +/// unspecified, broadcast, or a ULA/mapped equivalent. +/// +/// WHY: download URLs come from indexer response XML (third-party data), so a +/// public-looking hostname can point at internal infrastructure. Every +/// resolved address is checked; resolution failure rejects (fail-closed). +pub(crate) async fn validate_fetch_url(url: &str) -> Result<(), SearchIndexerError> { + let reject = |reason: &str| SearchIndexerError::UnsafeUrl { + url: redact_api_key(url), + reason: reason.to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + }; + + let parsed = Url::parse(url).map_err(|_| reject("not a valid URL"))?; + match parsed.scheme() { + "http" | "https" => {} + _ => return Err(reject("scheme must be http or https")), + } + let host = parsed + .host() + .ok_or_else(|| reject("URL must have a host"))?; + + match host { + Host::Ipv4(ip) if ip_is_disallowed(IpAddr::V4(ip)) => { + Err(reject("host is a private or local address")) + } + Host::Ipv6(ip) if ip_is_disallowed(IpAddr::V6(ip)) => { + Err(reject("host is a private or local address")) + } + Host::Ipv4(_) | Host::Ipv6(_) => Ok(()), + Host::Domain(domain) => { + // WHY: port only satisfies lookup_host's addr format; http/https + // always have a known default so the fallback is unreachable. + let port = parsed.port_or_known_default().unwrap_or(443); + let addrs = tokio::net::lookup_host((domain, port)) + .await + .map_err(|_| reject("host did not resolve"))?; + let mut resolved_any = false; + for addr in addrs { + resolved_any = true; + if ip_is_disallowed(addr.ip()) { + return Err(reject("host resolves to a private or local address")); + } + } + if !resolved_any { + return Err(reject("host did not resolve")); + } + Ok(()) + } + } +} + +fn ip_is_disallowed(ip: IpAddr) -> bool { + match ip { + IpAddr::V4(v4) => { + v4.is_loopback() + || v4.is_private() + || v4.is_link_local() + || v4.is_unspecified() + || v4.is_broadcast() + // NOTE: shared address space (CGNAT), RFC 6598: 100.64.0.0/10 + || (u32::from(v4) & 0xFFC0_0000) == 0x6440_0000 + } + IpAddr::V6(v6) => { + // WHY: an IPv4-mapped IPv6 literal (::ffff:127.0.0.1) must be + // judged by its embedded IPv4 address or it bypasses every v4 + // range check. + if let Some(mapped) = v6.to_ipv4_mapped() { + return ip_is_disallowed(IpAddr::V4(mapped)); + } + v6.is_loopback() + || v6.is_unspecified() + || v6.is_unique_local() + || v6.is_unicast_link_local() + } + } +} + fn urlencoding(s: &str) -> String { let mut result = String::with_capacity(s.len()); for b in s.bytes() { @@ -237,4 +394,163 @@ mod tests { assert_eq!(urlencoding("test&value"), "test%26value"); assert_eq!(urlencoding("normal"), "normal"); } + + // ── redact_api_key ──────────────────────────────────────────────────────── + + #[test] + fn redact_api_key_strips_value() { + let redacted = redact_api_key("https://x/api?t=caps&apikey=secret123"); + assert!(!redacted.contains("secret123"), "leaked: {redacted}"); + assert_eq!(redacted, "https://x/api?t=caps&apikey=[REDACTED]"); + } + + #[test] + fn redact_api_key_preserves_trailing_params() { + let redacted = redact_api_key("https://x/api?apikey=secret123&t=search&q=abc"); + assert!(!redacted.contains("secret123"), "leaked: {redacted}"); + assert_eq!(redacted, "https://x/api?apikey=[REDACTED]&t=search&q=abc"); + } + + #[test] + fn redact_api_key_handles_multiple_occurrences() { + let redacted = redact_api_key("https://x/?apikey=one&t=search&apikey=two"); + assert!(!redacted.contains("one"), "leaked: {redacted}"); + assert!(!redacted.contains("two"), "leaked: {redacted}"); + assert_eq!( + redacted, + "https://x/?apikey=[REDACTED]&t=search&apikey=[REDACTED]" + ); + } + + #[test] + fn redact_api_key_no_key_present_noop() { + assert_eq!( + redact_api_key("https://x/api?t=caps"), + "https://x/api?t=caps" + ); + assert_eq!(redact_api_key(""), ""); + } + + #[test] + fn redact_api_key_value_at_end_without_ampersand() { + let redacted = redact_api_key("https://x/api?apikey=secret123"); + assert!(!redacted.contains("secret123"), "leaked: {redacted}"); + assert_eq!(redacted, "https://x/api?apikey=[REDACTED]"); + } + + // ── validate_fetch_url ──────────────────────────────────────────────────── + + #[tokio::test] + async fn validate_fetch_url_rejects_unparseable() { + assert!(validate_fetch_url("not a url").await.is_err()); + assert!(validate_fetch_url("").await.is_err()); + } + + #[tokio::test] + async fn validate_fetch_url_rejects_non_http_scheme() { + for url in [ + "file:///etc/passwd", + "ftp://example.com/file.torrent", + "gopher://example.com/", + "javascript:alert(1)", + ] { + let err = validate_fetch_url(url).await.expect_err(url); + assert!(matches!(err, SearchIndexerError::UnsafeUrl { .. })); + } + } + + #[tokio::test] + async fn validate_fetch_url_rejects_loopback() { + for url in ["http://127.0.0.1/x", "http://127.8.9.10:8080/x"] { + let err = validate_fetch_url(url).await.expect_err(url); + assert!(matches!(err, SearchIndexerError::UnsafeUrl { .. })); + } + } + + #[tokio::test] + async fn validate_fetch_url_rejects_localhost_hostname() { + // WHY: exercises the DNS-resolution branch — "localhost" resolves via + // the system resolver to a loopback address, which must be rejected. + let err = validate_fetch_url("http://localhost:8080/x") + .await + .expect_err("localhost must be rejected"); + assert!(matches!(err, SearchIndexerError::UnsafeUrl { .. })); + } + + #[tokio::test] + async fn validate_fetch_url_rejects_link_local() { + let err = validate_fetch_url("http://169.254.169.254/latest/meta-data/") + .await + .expect_err("metadata endpoint must be rejected"); + assert!(matches!(err, SearchIndexerError::UnsafeUrl { .. })); + } + + #[tokio::test] + async fn validate_fetch_url_rejects_private_ranges() { + for url in [ + "http://10.0.0.1/x", + "http://172.16.5.5/x", + "http://192.168.1.1/x", + "http://100.64.0.1/x", + "http://0.0.0.0/x", + ] { + let err = validate_fetch_url(url).await.expect_err(url); + assert!(matches!(err, SearchIndexerError::UnsafeUrl { .. })); + } + } + + #[tokio::test] + async fn validate_fetch_url_rejects_ipv6_local_and_mapped() { + for url in [ + "http://[::1]/x", + "http://[fc00::1]/x", + "http://[fe80::1]/x", + "http://[::ffff:127.0.0.1]/x", + "http://[::ffff:192.168.1.1]/x", + ] { + let err = validate_fetch_url(url).await.expect_err(url); + assert!(matches!(err, SearchIndexerError::UnsafeUrl { .. })); + } + } + + #[tokio::test] + async fn validate_fetch_url_allows_public_ip_literal() { + // NOTE: TEST-NET-3 documentation range — public per the enforced + // ranges, never actually fetched by this validation. + assert!( + validate_fetch_url("http://203.0.113.10/file.torrent") + .await + .is_ok() + ); + assert!( + validate_fetch_url("https://203.0.113.10:8443/file.torrent") + .await + .is_ok() + ); + } + + #[tokio::test] + async fn validate_fetch_url_error_redacts_api_key() { + let err = validate_fetch_url("http://127.0.0.1/get?apikey=supersecret") + .await + .expect_err("loopback must be rejected"); + let display = err.to_string(); + assert!(!display.contains("supersecret"), "leaked: {display}"); + } + + #[test] + fn ip_range_classification() { + use std::net::{Ipv4Addr, Ipv6Addr}; + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(10, 1, 2, 3)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(172, 31, 0, 1)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(192, 168, 0, 1)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(169, 254, 1, 1)))); + assert!(ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(100, 127, 0, 1)))); + assert!(ip_is_disallowed(IpAddr::V6(Ipv6Addr::LOCALHOST))); + assert!(!ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 1)))); + assert!(!ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)))); + assert!(!ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(100, 63, 0, 1)))); + assert!(!ip_is_disallowed(IpAddr::V4(Ipv4Addr::new(172, 32, 0, 1)))); + } } diff --git a/crates/zetesis/src/client/newznab.rs b/crates/zetesis/src/client/newznab.rs index e04710f6..1a4d5236 100644 --- a/crates/zetesis/src/client/newznab.rs +++ b/crates/zetesis/src/client/newznab.rs @@ -9,7 +9,10 @@ use tracing::instrument; use crate::cf_bypass::CloudflareProxy; use crate::client::xml::{get_attr_f64, get_attr_u32, parse_caps_xml, parse_feed_xml}; -use crate::client::{IndexerClient, IndexerConfig, build_caps_url, build_search_url}; +use crate::client::{ + IndexerClient, IndexerConfig, build_caps_url, build_search_url, read_body_bounded, + redact_api_key, validate_fetch_url, +}; use crate::error::{self, SearchIndexerError}; use crate::types::{ DownloadResponse, IndexerCaps, IndexerStatus, ReleaseProtocol, SearchQuery, SearchResult, @@ -20,6 +23,7 @@ pub struct NewznabClient { http: reqwest::Client, cf_proxy: Arc, timeout: Duration, + max_body_bytes: u64, } impl NewznabClient { @@ -28,12 +32,14 @@ impl NewznabClient { http: reqwest::Client, cf_proxy: Arc, timeout: Duration, + max_body_bytes: u64, ) -> Self { Self { config, http, cf_proxy, timeout, + max_body_bytes, } } @@ -49,11 +55,10 @@ impl NewznabClient { let fut = self.http.get(url).timeout(self.timeout).send(); let response = tokio::select! { - result = fut => result.context(error::HttpRequestSnafu { url })?, + result = fut => result.context(error::HttpRequestSnafu { url: redact_api_key(url) })?, () = ct.cancelled() => { - return Err(SearchIndexerError::ParseResponse { - url: url.to_string(), - error: "request cancelled".to_string(), + return Err(SearchIndexerError::Cancelled { + url: redact_api_key(url), location: snafu::Location::new(file!(), line!(), column!()), }); } @@ -79,11 +84,7 @@ impl NewznabClient { }); } - let body = response - .text() - .await - .context(error::HttpRequestSnafu { url })?; - Ok(body) + read_body_bounded(response, url, self.max_body_bytes).await } } @@ -97,7 +98,7 @@ impl IndexerClient for NewznabClient { let url = build_search_url(&self.config, query); let xml = self.fetch_xml(&url, ct).await?; let feed = parse_feed_xml(&xml).map_err(|e| SearchIndexerError::ParseResponse { - url: url.clone(), + url: redact_api_key(&url), error: e.to_string(), location: snafu::Location::new(file!(), line!(), column!()), })?; @@ -151,7 +152,7 @@ impl IndexerClient for NewznabClient { let url = build_caps_url(&self.config); let xml = self.fetch_xml(&url, ct).await?; parse_caps_xml(&xml).map_err(|e| SearchIndexerError::ParseResponse { - url, + url: redact_api_key(&url), error: e.to_string(), location: snafu::Location::new(file!(), line!(), column!()), }) @@ -179,7 +180,218 @@ impl IndexerClient for NewznabClient { url: &str, ct: CancellationToken, ) -> Result { + // SAFETY: download URLs originate in indexer response XML (third-party + // data) — validate scheme + resolved addresses before any fetch. + validate_fetch_url(url).await?; let body = self.fetch_xml(url, ct).await?; Ok(DownloadResponse::NzbFile(Bytes::from(body))) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::cf_bypass::noop::NoProxy; + use crate::test_support::{spawn_hang_http, spawn_one_shot_http, spawn_raw_http}; + + const FEED_XML: &str = r#" + + + Usenet Indexer + + Test.Release.2024.NZB + nzb-guid-456 + 524288000 + https://example.com/getnzb/nzb-guid-456 + + + +"#; + + fn client(url: String, api_key: Option<&str>, max_body_bytes: u64) -> NewznabClient { + NewznabClient::new( + IndexerConfig { + id: 2, + name: "Test".to_string(), + url, + api_key: api_key.map(str::to_string), + cf_bypass: false, + }, + reqwest::Client::new(), + Arc::new(NoProxy), + Duration::from_secs(5), + max_body_bytes, + ) + } + + fn query() -> SearchQuery { + SearchQuery { + query_text: Some("test".to_string()), + limit: 100, + ..Default::default() + } + } + + #[tokio::test] + async fn fetch_200_returns_parsed_results() { + let (url, _server) = spawn_one_shot_http(200, "OK", &[], FEED_XML).await; + let results = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].title, "Test.Release.2024.NZB"); + assert_eq!(results[0].protocol, ReleaseProtocol::Nzb); + } + + #[tokio::test] + async fn fetch_401_maps_to_auth_failed() { + let (url, _server) = spawn_one_shot_http(401, "Unauthorized", &[], "").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!( + err, + SearchIndexerError::AuthFailed { indexer_id: 2, .. } + )); + } + + #[tokio::test] + async fn fetch_403_maps_to_auth_failed() { + let (url, _server) = spawn_one_shot_http(403, "Forbidden", &[], "").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!(err, SearchIndexerError::AuthFailed { .. })); + } + + #[tokio::test] + async fn fetch_429_with_retry_after_maps_to_rate_limited() { + let (url, _server) = + spawn_one_shot_http(429, "Too Many Requests", &[("retry-after", "60")], "").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!( + err, + SearchIndexerError::RateLimited { + retry_after_seconds: Some(60), + .. + } + )); + } + + #[tokio::test] + async fn fetch_429_without_retry_after_maps_to_rate_limited_none() { + let (url, _server) = spawn_one_shot_http(429, "Too Many Requests", &[], "").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!( + err, + SearchIndexerError::RateLimited { + retry_after_seconds: None, + .. + } + )); + } + + #[tokio::test] + async fn fetch_500_falls_through_to_parse_error() { + // NOTE: documents current behavior — a 5xx body is read and fails XML + // parsing rather than mapping to a dedicated status error. + let (url, _server) = + spawn_one_shot_http(500, "Internal Server Error", &[], "not xml").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!(err, SearchIndexerError::ParseResponse { .. })); + } + + #[tokio::test] + async fn fetch_cancelled_returns_cancelled_variant() { + let (url, server) = spawn_hang_http().await; + let ct = CancellationToken::new(); + ct.cancel(); + let err = client(url, None, 1 << 20) + .search(&query(), ct) + .await + .unwrap_err(); + assert!( + matches!(err, SearchIndexerError::Cancelled { .. }), + "expected Cancelled, got {err:?}" + ); + server.abort(); + } + + #[tokio::test] + async fn fetch_rejects_oversized_content_length_before_body() { + let raw = + b"HTTP/1.1 200 OK\r\ncontent-length: 10000000\r\nconnection: close\r\n\r\n".to_vec(); + let (url, _server) = spawn_raw_http(raw).await; + let err = client(url, None, 64) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!( + matches!(err, SearchIndexerError::ResponseTooLarge { limit: 64, .. }), + "expected ResponseTooLarge, got {err:?}" + ); + } + + #[tokio::test] + async fn fetch_rejects_oversized_stream_without_content_length() { + let mut raw = b"HTTP/1.1 200 OK\r\nconnection: close\r\n\r\n".to_vec(); + raw.extend_from_slice(&[b'x'; 4096]); + let (url, _server) = spawn_raw_http(raw).await; + let err = client(url, None, 64) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!( + matches!(err, SearchIndexerError::ResponseTooLarge { limit: 64, .. }), + "expected ResponseTooLarge, got {err:?}" + ); + } + + #[tokio::test] + async fn error_display_does_not_leak_api_key() { + let err = client( + "http://127.0.0.1:9/api".to_string(), + Some("supersecret"), + 1 << 20, + ) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + let display = err.to_string(); + assert!(!display.contains("supersecret"), "leaked: {display}"); + assert!( + matches!(err, SearchIndexerError::HttpRequest { ref url, .. } if url.contains("[REDACTED]")) + ); + } + + #[tokio::test] + async fn download_rejects_ssrf_url() { + for target in [ + "http://127.0.0.1:8080/steal", + "http://169.254.169.254/latest/meta-data/", + "http://10.0.0.1/internal", + "ftp://example.com/file.nzb", + ] { + let err = client("http://127.0.0.1:9/api".to_string(), None, 1 << 20) + .download(target, CancellationToken::new()) + .await + .unwrap_err(); + assert!( + matches!(err, SearchIndexerError::UnsafeUrl { .. }), + "expected UnsafeUrl for {target}, got {err:?}" + ); + } + } +} diff --git a/crates/zetesis/src/client/torznab.rs b/crates/zetesis/src/client/torznab.rs index 67c9f7a5..7caa6b15 100644 --- a/crates/zetesis/src/client/torznab.rs +++ b/crates/zetesis/src/client/torznab.rs @@ -9,7 +9,10 @@ use tracing::instrument; use crate::cf_bypass::CloudflareProxy; use crate::client::xml::{get_attr, get_attr_f64, get_attr_u32, parse_caps_xml, parse_feed_xml}; -use crate::client::{IndexerClient, IndexerConfig, build_caps_url, build_search_url}; +use crate::client::{ + IndexerClient, IndexerConfig, build_caps_url, build_search_url, read_body_bounded, + redact_api_key, validate_fetch_url, +}; use crate::error::{self, SearchIndexerError}; use crate::types::{ DownloadResponse, IndexerCaps, IndexerStatus, ReleaseProtocol, SearchQuery, SearchResult, @@ -20,6 +23,7 @@ pub struct TorznabClient { http: reqwest::Client, cf_proxy: Arc, timeout: Duration, + max_body_bytes: u64, } impl TorznabClient { @@ -28,12 +32,14 @@ impl TorznabClient { http: reqwest::Client, cf_proxy: Arc, timeout: Duration, + max_body_bytes: u64, ) -> Self { Self { config, http, cf_proxy, timeout, + max_body_bytes, } } @@ -49,11 +55,10 @@ impl TorznabClient { let fut = self.http.get(url).timeout(self.timeout).send(); let response = tokio::select! { - result = fut => result.context(error::HttpRequestSnafu { url })?, + result = fut => result.context(error::HttpRequestSnafu { url: redact_api_key(url) })?, () = ct.cancelled() => { - return Err(SearchIndexerError::ParseResponse { - url: url.to_string(), - error: "request cancelled".to_string(), + return Err(SearchIndexerError::Cancelled { + url: redact_api_key(url), location: snafu::Location::new(file!(), line!(), column!()), }); } @@ -79,11 +84,7 @@ impl TorznabClient { }); } - let body = response - .text() - .await - .context(error::HttpRequestSnafu { url })?; - Ok(body) + read_body_bounded(response, url, self.max_body_bytes).await } } @@ -97,7 +98,7 @@ impl IndexerClient for TorznabClient { let url = build_search_url(&self.config, query); let xml = self.fetch_xml(&url, ct).await?; let feed = parse_feed_xml(&xml).map_err(|e| SearchIndexerError::ParseResponse { - url: url.clone(), + url: redact_api_key(&url), error: e.to_string(), location: snafu::Location::new(file!(), line!(), column!()), })?; @@ -160,7 +161,7 @@ impl IndexerClient for TorznabClient { let url = build_caps_url(&self.config); let xml = self.fetch_xml(&url, ct).await?; parse_caps_xml(&xml).map_err(|e| SearchIndexerError::ParseResponse { - url, + url: redact_api_key(&url), error: e.to_string(), location: snafu::Location::new(file!(), line!(), column!()), }) @@ -192,7 +193,264 @@ impl IndexerClient for TorznabClient { return Ok(DownloadResponse::MagnetUri(url.to_string())); } + // SAFETY: download URLs originate in indexer response XML (third-party + // data) — validate scheme + resolved addresses before any fetch. + validate_fetch_url(url).await?; let body = self.fetch_xml(url, ct).await?; Ok(DownloadResponse::TorrentFile(Bytes::from(body))) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::cf_bypass::noop::NoProxy; + use crate::test_support::{spawn_hang_http, spawn_one_shot_http, spawn_raw_http}; + + const FEED_XML: &str = r#" + + + Test Indexer + + Test.Release.2024.FLAC + abc123 + 734003200 + https://example.com/download/abc123 + + + + +"#; + + fn client(url: String, api_key: Option<&str>, max_body_bytes: u64) -> TorznabClient { + TorznabClient::new( + IndexerConfig { + id: 1, + name: "Test".to_string(), + url, + api_key: api_key.map(str::to_string), + cf_bypass: false, + }, + reqwest::Client::new(), + Arc::new(NoProxy), + Duration::from_secs(5), + max_body_bytes, + ) + } + + fn query() -> SearchQuery { + SearchQuery { + query_text: Some("test".to_string()), + limit: 100, + ..Default::default() + } + } + + #[tokio::test] + async fn fetch_200_returns_parsed_results() { + let (url, _server) = spawn_one_shot_http(200, "OK", &[], FEED_XML).await; + let results = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].title, "Test.Release.2024.FLAC"); + assert_eq!(results[0].seeders, Some(42)); + assert_eq!(results[0].protocol, ReleaseProtocol::Torrent); + } + + #[tokio::test] + async fn fetch_401_maps_to_auth_failed() { + let (url, _server) = spawn_one_shot_http(401, "Unauthorized", &[], "").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!( + err, + SearchIndexerError::AuthFailed { indexer_id: 1, .. } + )); + } + + #[tokio::test] + async fn fetch_403_maps_to_auth_failed() { + let (url, _server) = spawn_one_shot_http(403, "Forbidden", &[], "").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!(err, SearchIndexerError::AuthFailed { .. })); + } + + #[tokio::test] + async fn fetch_429_with_retry_after_maps_to_rate_limited() { + let (url, _server) = + spawn_one_shot_http(429, "Too Many Requests", &[("retry-after", "120")], "").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!( + err, + SearchIndexerError::RateLimited { + retry_after_seconds: Some(120), + .. + } + )); + } + + #[tokio::test] + async fn fetch_429_without_retry_after_maps_to_rate_limited_none() { + let (url, _server) = spawn_one_shot_http(429, "Too Many Requests", &[], "").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!( + err, + SearchIndexerError::RateLimited { + retry_after_seconds: None, + .. + } + )); + } + + #[tokio::test] + async fn fetch_500_falls_through_to_parse_error() { + // NOTE: documents current behavior — a 5xx body is read and fails XML + // parsing rather than mapping to a dedicated status error. + let (url, _server) = + spawn_one_shot_http(500, "Internal Server Error", &[], "not xml").await; + let err = client(url, None, 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!(matches!(err, SearchIndexerError::ParseResponse { .. })); + } + + #[tokio::test] + async fn fetch_cancelled_returns_cancelled_variant() { + let (url, server) = spawn_hang_http().await; + let ct = CancellationToken::new(); + ct.cancel(); + let err = client(url, None, 1 << 20) + .search(&query(), ct) + .await + .unwrap_err(); + assert!( + matches!(err, SearchIndexerError::Cancelled { .. }), + "expected Cancelled, got {err:?}" + ); + server.abort(); + } + + #[tokio::test] + async fn fetch_rejects_oversized_content_length_before_body() { + let raw = + b"HTTP/1.1 200 OK\r\ncontent-length: 10000000\r\nconnection: close\r\n\r\n".to_vec(); + let (url, _server) = spawn_raw_http(raw).await; + let err = client(url, None, 64) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!( + matches!( + err, + SearchIndexerError::ResponseTooLarge { + size: 10_000_000, + limit: 64, + .. + } + ), + "expected ResponseTooLarge, got {err:?}" + ); + } + + #[tokio::test] + async fn fetch_rejects_oversized_stream_without_content_length() { + // WHY: no content-length header — the streamed running counter is the + // only enforcement (declared length can be absent or forged). + let mut raw = b"HTTP/1.1 200 OK\r\nconnection: close\r\n\r\n".to_vec(); + raw.extend_from_slice(&[b'x'; 4096]); + let (url, _server) = spawn_raw_http(raw).await; + let err = client(url, None, 64) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + assert!( + matches!(err, SearchIndexerError::ResponseTooLarge { limit: 64, .. }), + "expected ResponseTooLarge, got {err:?}" + ); + } + + #[tokio::test] + async fn fetch_allows_body_under_limit() { + let (url, _server) = spawn_one_shot_http(200, "OK", &[], FEED_XML).await; + let results = client(url, None, FEED_XML.len() as u64 + 1) + .search(&query(), CancellationToken::new()) + .await + .unwrap(); + assert_eq!(results.len(), 1); + } + + #[tokio::test] + async fn error_display_does_not_leak_api_key() { + // WHY: port 9 (discard) on loopback refuses connections — forces an + // HttpRequest error whose url field must carry the redacted form. + let err = client( + "http://127.0.0.1:9/api".to_string(), + Some("supersecret"), + 1 << 20, + ) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + let display = err.to_string(); + assert!(!display.contains("supersecret"), "leaked: {display}"); + assert!( + matches!(err, SearchIndexerError::HttpRequest { ref url, .. } if url.contains("[REDACTED]")) + ); + } + + #[tokio::test] + async fn parse_error_display_does_not_leak_api_key() { + let (url, _server) = spawn_one_shot_http(200, "OK", &[], "not xml").await; + let err = client(url, Some("supersecret"), 1 << 20) + .search(&query(), CancellationToken::new()) + .await + .unwrap_err(); + let display = err.to_string(); + assert!(!display.contains("supersecret"), "leaked: {display}"); + } + + #[tokio::test] + async fn download_magnet_uri_short_circuits_without_network() { + // WHY: config URL points at a refusing port — any network attempt + // would error, so Ok(MagnetUri) proves no fetch happened. + let magnet = "magnet:?xt=urn:btih:abc123"; + let response = client("http://127.0.0.1:9/api".to_string(), None, 1 << 20) + .download(magnet, CancellationToken::new()) + .await + .unwrap(); + assert!(matches!(response, DownloadResponse::MagnetUri(uri) if uri == magnet)); + } + + #[tokio::test] + async fn download_rejects_ssrf_url() { + for target in [ + "http://127.0.0.1:8080/steal", + "http://169.254.169.254/latest/meta-data/", + "http://192.168.1.1/admin", + "file:///etc/passwd", + ] { + let err = client("http://127.0.0.1:9/api".to_string(), None, 1 << 20) + .download(target, CancellationToken::new()) + .await + .unwrap_err(); + assert!( + matches!(err, SearchIndexerError::UnsafeUrl { .. }), + "expected UnsafeUrl for {target}, got {err:?}" + ); + } + } +} diff --git a/crates/zetesis/src/error.rs b/crates/zetesis/src/error.rs index 280bab99..58fa3f37 100644 --- a/crates/zetesis/src/error.rs +++ b/crates/zetesis/src/error.rs @@ -21,6 +21,30 @@ pub enum SearchIndexerError { location: snafu::Location, }, + #[snafu(display("request to {url} was cancelled"))] + Cancelled { + url: String, + #[snafu(implicit)] + location: snafu::Location, + }, + + #[snafu(display("response FROM {url} exceeds {limit} byte cap (got at least {size} bytes)"))] + ResponseTooLarge { + url: String, + size: u64, + limit: u64, + #[snafu(implicit)] + location: snafu::Location, + }, + + #[snafu(display("refusing to fetch {url}: {reason}"))] + UnsafeUrl { + url: String, + reason: String, + #[snafu(implicit)] + location: snafu::Location, + }, + #[snafu(display("indexer {indexer_id} returned auth failure (bad API key)"))] AuthFailed { indexer_id: i64, diff --git a/crates/zetesis/src/lib.rs b/crates/zetesis/src/lib.rs index ea9e34c0..8ae1e635 100644 --- a/crates/zetesis/src/lib.rs +++ b/crates/zetesis/src/lib.rs @@ -4,6 +4,7 @@ pub mod error; pub mod rate_limit; pub mod repo; pub mod search; +pub(crate) mod test_support; pub mod types; use std::sync::Arc; diff --git a/crates/zetesis/src/repo.rs b/crates/zetesis/src/repo.rs index 70975506..08ddb6a9 100644 --- a/crates/zetesis/src/repo.rs +++ b/crates/zetesis/src/repo.rs @@ -1,7 +1,7 @@ use apotheke::DbError; use apotheke::error::QuerySnafu; use snafu::ResultExt; -use sqlx::SqlitePool; +use sqlx::{Sqlite, SqliteConnection, SqlitePool}; use crate::types::{IndexerCaps, IndexerCategory}; @@ -116,32 +116,34 @@ pub async fn get_eligible_indexers(pool: &SqlitePool) -> Result, Ok(rows) } -pub async fn update_indexer_status( - pool: &SqlitePool, - id: i64, - status: &str, -) -> Result<(), DbError> { +pub async fn update_indexer_status<'e, E>(executor: E, id: i64, status: &str) -> Result<(), DbError> +where + E: sqlx::Executor<'e, Database = Sqlite>, +{ sqlx::query("UPDATE indexers SET status = ? WHERE id = ?") .bind(status) .bind(id) - .execute(pool) + .execute(executor) .await .context(QuerySnafu { table: "indexers" })?; Ok(()) } -pub async fn update_indexer_caps( - pool: &SqlitePool, +pub async fn update_indexer_caps<'e, E>( + executor: E, id: i64, caps_json: &str, last_tested: &str, -) -> Result<(), DbError> { +) -> Result<(), DbError> +where + E: sqlx::Executor<'e, Database = Sqlite>, +{ sqlx::query("UPDATE indexers SET caps_json = ?, last_tested = ? WHERE id = ?") .bind(caps_json) .bind(last_tested) .bind(id) - .execute(pool) + .execute(executor) .await .context(QuerySnafu { table: "indexers" })?; @@ -158,14 +160,16 @@ pub async fn delete_indexer(pool: &SqlitePool, id: i64) -> Result<(), DbError> { Ok(()) } +// INVARIANT: the caller supplies a connection inside an open transaction — +// the DELETE + INSERT sequence is only atomic under that transaction. pub async fn upsert_indexer_categories( - pool: &SqlitePool, + conn: &mut SqliteConnection, indexer_id: i64, caps: &IndexerCaps, ) -> Result<(), DbError> { sqlx::query("DELETE FROM indexer_categories WHERE indexer_id = ?") .bind(indexer_id) - .execute(pool) + .execute(&mut *conn) .await .context(QuerySnafu { table: "indexer_categories", @@ -189,7 +193,7 @@ pub async fn upsert_indexer_categories( .bind(indexer_id) .bind(i64::from(cat_id)) .bind(&name) - .execute(pool) + .execute(&mut *conn) .await .context(QuerySnafu { table: "indexer_categories", @@ -209,3 +213,308 @@ pub async fn restore_degraded_cf_indexers(pool: &SqlitePool) -> Result SqlitePool { + let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); + MIGRATOR.run(&pool).await.unwrap(); + pool + } + + fn params<'a>(name: &'a str, cf_bypass: bool) -> InsertIndexerParams<'a> { + InsertIndexerParams { + name, + url: "https://example.com/api", + protocol: "torznab", + api_key: Some("key"), + cf_bypass, + priority: 50, + } + } + + fn caps_with(categories: &[(u32, &str)]) -> IndexerCaps { + IndexerCaps { + server: ServerInfo { + title: None, + version: None, + }, + limits: SearchLimits::default(), + search_functions: vec![], + categories: categories + .iter() + .map(|(id, name)| IndexerCategory { + id: *id, + name: (*name).to_string(), + subcategories: vec![], + }) + .collect(), + } + } + + async fn category_rows(pool: &SqlitePool, indexer_id: i64) -> Vec<(i64, String)> { + sqlx::query_as::<_, (i64, String)>( + "SELECT category_id, name FROM indexer_categories + WHERE indexer_id = ? ORDER BY category_id", + ) + .bind(indexer_id) + .fetch_all(pool) + .await + .unwrap() + } + + #[tokio::test] + async fn insert_and_get_indexer_roundtrip() { + let pool = setup().await; + let id = insert_indexer(&pool, params("Roundtrip", false)) + .await + .unwrap(); + + let row = get_indexer(&pool, id).await.unwrap().unwrap(); + assert_eq!(row.id, id); + assert_eq!(row.name, "Roundtrip"); + assert_eq!(row.url, "https://example.com/api"); + assert_eq!(row.protocol, "torznab"); + assert_eq!(row.api_key.as_deref(), Some("key")); + assert!(row.enabled); + assert!(!row.cf_bypass); + assert_eq!(row.status, "active"); + assert_eq!(row.priority, 50); + } + + #[tokio::test] + async fn get_indexer_returns_none_for_unknown_id() { + let pool = setup().await; + assert!(get_indexer(&pool, 9999).await.unwrap().is_none()); + } + + #[tokio::test] + async fn list_indexers_orders_by_priority() { + let pool = setup().await; + let low = insert_indexer( + &pool, + InsertIndexerParams { + priority: 90, + ..params("Low", false) + }, + ) + .await + .unwrap(); + let high = insert_indexer( + &pool, + InsertIndexerParams { + priority: 10, + ..params("High", false) + }, + ) + .await + .unwrap(); + + let rows = list_indexers(&pool).await.unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].id, high); + assert_eq!(rows[1].id, low); + } + + #[tokio::test] + async fn get_eligible_indexers_excludes_disabled() { + let pool = setup().await; + let id = insert_indexer(&pool, params("Disabled", false)) + .await + .unwrap(); + sqlx::query("UPDATE indexers SET enabled = FALSE WHERE id = ?") + .bind(id) + .execute(&pool) + .await + .unwrap(); + + assert!(get_eligible_indexers(&pool).await.unwrap().is_empty()); + } + + #[tokio::test] + async fn get_eligible_indexers_excludes_failed() { + let pool = setup().await; + let id = insert_indexer(&pool, params("Failed", false)) + .await + .unwrap(); + update_indexer_status(&pool, id, "failed").await.unwrap(); + + assert!(get_eligible_indexers(&pool).await.unwrap().is_empty()); + } + + #[tokio::test] + async fn get_eligible_indexers_includes_active_and_degraded() { + let pool = setup().await; + insert_indexer(&pool, params("Active", false)) + .await + .unwrap(); + let degraded = insert_indexer(&pool, params("Degraded", false)) + .await + .unwrap(); + update_indexer_status(&pool, degraded, "degraded") + .await + .unwrap(); + + let eligible = get_eligible_indexers(&pool).await.unwrap(); + assert_eq!(eligible.len(), 2); + } + + #[tokio::test] + async fn update_indexer_status_persists() { + let pool = setup().await; + let id = insert_indexer(&pool, params("Status", false)) + .await + .unwrap(); + + update_indexer_status(&pool, id, "degraded").await.unwrap(); + + let row = get_indexer(&pool, id).await.unwrap().unwrap(); + assert_eq!(row.status, "degraded"); + } + + #[tokio::test] + async fn update_indexer_caps_persists_json_and_timestamp() { + let pool = setup().await; + let id = insert_indexer(&pool, params("Caps", false)).await.unwrap(); + + update_indexer_caps(&pool, id, r#"{"caps":true}"#, "2026-01-01T00:00:00Z") + .await + .unwrap(); + + let row = get_indexer(&pool, id).await.unwrap().unwrap(); + assert_eq!(row.caps_json.as_deref(), Some(r#"{"caps":true}"#)); + assert_eq!(row.last_tested.as_deref(), Some("2026-01-01T00:00:00Z")); + } + + #[tokio::test] + async fn delete_indexer_removes_row() { + let pool = setup().await; + let id = insert_indexer(&pool, params("Doomed", false)) + .await + .unwrap(); + + delete_indexer(&pool, id).await.unwrap(); + + assert!(get_indexer(&pool, id).await.unwrap().is_none()); + } + + #[tokio::test] + async fn upsert_indexer_categories_replaces_previous_set() { + let pool = setup().await; + let id = insert_indexer(&pool, params("Cats", false)).await.unwrap(); + + let mut tx = pool.begin().await.unwrap(); + upsert_indexer_categories( + &mut tx, + id, + &caps_with(&[(1000, "Console"), (2000, "Movies")]), + ) + .await + .unwrap(); + tx.commit().await.unwrap(); + + let mut tx = pool.begin().await.unwrap(); + upsert_indexer_categories(&mut tx, id, &caps_with(&[(3000, "Audio")])) + .await + .unwrap(); + tx.commit().await.unwrap(); + + let rows = category_rows(&pool, id).await; + assert_eq!(rows, vec![(3000, "Audio".to_string())]); + } + + #[tokio::test] + async fn upsert_indexer_categories_flattens_subcategories() { + let pool = setup().await; + let id = insert_indexer(&pool, params("Nested", false)) + .await + .unwrap(); + + let caps = IndexerCaps { + categories: vec![IndexerCategory { + id: 3000, + name: "Audio".to_string(), + subcategories: vec![IndexerCategory { + id: 3010, + name: "Audio/MP3".to_string(), + subcategories: vec![], + }], + }], + ..caps_with(&[]) + }; + + let mut tx = pool.begin().await.unwrap(); + upsert_indexer_categories(&mut tx, id, &caps).await.unwrap(); + tx.commit().await.unwrap(); + + let rows = category_rows(&pool, id).await; + assert_eq!( + rows, + vec![(3000, "Audio".to_string()), (3010, "Audio/MP3".to_string())] + ); + } + + #[tokio::test] + async fn upsert_indexer_categories_rolls_back_with_caller_transaction() { + // WHY: the atomicity contract — a dropped (uncommitted) transaction + // must leave the previous category set fully intact, DELETE included. + let pool = setup().await; + let id = insert_indexer(&pool, params("Atomic", false)) + .await + .unwrap(); + + let mut tx = pool.begin().await.unwrap(); + upsert_indexer_categories(&mut tx, id, &caps_with(&[(1000, "Console")])) + .await + .unwrap(); + tx.commit().await.unwrap(); + + let mut tx = pool.begin().await.unwrap(); + upsert_indexer_categories(&mut tx, id, &caps_with(&[(2000, "Movies")])) + .await + .unwrap(); + drop(tx); + + let rows = category_rows(&pool, id).await; + assert_eq!(rows, vec![(1000, "Console".to_string())]); + } + + #[tokio::test] + async fn restore_degraded_cf_indexers_only_affects_cf_bypass_degraded() { + let pool = setup().await; + let cf_degraded = insert_indexer(&pool, params("CfDegraded", true)) + .await + .unwrap(); + let plain_degraded = insert_indexer(&pool, params("PlainDegraded", false)) + .await + .unwrap(); + let cf_failed = insert_indexer(&pool, params("CfFailed", true)) + .await + .unwrap(); + update_indexer_status(&pool, cf_degraded, "degraded") + .await + .unwrap(); + update_indexer_status(&pool, plain_degraded, "degraded") + .await + .unwrap(); + update_indexer_status(&pool, cf_failed, "failed") + .await + .unwrap(); + + let affected = restore_degraded_cf_indexers(&pool).await.unwrap(); + + assert_eq!(affected, 1); + let restored = get_indexer(&pool, cf_degraded).await.unwrap().unwrap(); + assert_eq!(restored.status, "active"); + let untouched = get_indexer(&pool, plain_degraded).await.unwrap().unwrap(); + assert_eq!(untouched.status, "degraded"); + let still_failed = get_indexer(&pool, cf_failed).await.unwrap().unwrap(); + assert_eq!(still_failed.status, "failed"); + } +} diff --git a/crates/zetesis/src/search.rs b/crates/zetesis/src/search.rs index 2c7498c7..89353605 100644 --- a/crates/zetesis/src/search.rs +++ b/crates/zetesis/src/search.rs @@ -2,8 +2,10 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; +use apotheke::error::TransactionSnafu; use futures::stream::{self, StreamExt}; use horismos::SearchSubsystemConfig; +use snafu::ResultExt; use sqlx::SqlitePool; use themelion::{EventSender, HarmoniaEvent, QueryId}; use tokio_util::sync::CancellationToken; @@ -13,11 +15,17 @@ use crate::cf_bypass::CloudflareProxy; use crate::client::newznab::NewznabClient; use crate::client::torznab::TorznabClient; use crate::client::{DynIndexerClient, IndexerConfig}; -use crate::error::SearchIndexerError; +use crate::error::{self, SearchIndexerError}; use crate::rate_limit::RateLimiter; use crate::repo::{self, IndexerRow}; use crate::types::{IndexerCaps, IndexerStatus, SearchMediaType, SearchQuery, SearchResult}; +/// Fallback back-off when a 429 carries no Retry-After header. +const DEFAULT_RETRY_AFTER_SECS: u64 = 60; +/// Upper bound on honored Retry-After — a hostile header must not park an +/// indexer indefinitely. +const MAX_RETRY_AFTER_SECS: u64 = 3600; + pub struct SearchIndexerService { read_pool: SqlitePool, write_pool: SqlitePool, @@ -86,6 +94,7 @@ impl SearchIndexerService { let cf_proxy = Arc::clone(&self.cf_proxy); let http = self.http.clone(); let timeout = Duration::from_secs(self.config.search_timeout_seconds); + let max_body_bytes = self.config.max_response_body_bytes; let rate_limiter = &self.rate_limiter; let results: Vec = stream::iter(eligible) @@ -96,9 +105,20 @@ impl SearchIndexerService { let q = query.clone(); async move { rate_limiter.acquire(indexer.id).await; - let client = make_client(&indexer, h, cf, timeout); + let client = make_client(&indexer, h, cf, timeout, max_body_bytes); match client.search_boxed(&q, ct).await { Ok(results) => results, + Err(e @ SearchIndexerError::Cancelled { .. }) => { + // WHY: cancellation is caller intent, not indexer + // failure — no warn, no status change. + info!( + indexer_id = indexer.id, + indexer_name = %indexer.name, + error = %e, + "search cancelled for indexer" + ); + Vec::new() + } Err(e) => { warn!( indexer_id = indexer.id, @@ -121,10 +141,12 @@ impl SearchIndexerService { let deduped = deduplicate(results); // Step 5: Emit event - let _ = self.event_tx.send(HarmoniaEvent::SearchCompleted { - query_id, - result_count: deduped.len(), - }); + self.event_tx + .send(HarmoniaEvent::SearchCompleted { + query_id, + result_count: deduped.len(), + }) + .ok(); info!( query_id = %query_id, @@ -146,6 +168,7 @@ impl SearchIndexerService { self.http.clone(), Arc::clone(&self.cf_proxy), Duration::from_secs(self.config.request_timeout_secs), + self.config.max_response_body_bytes, ); let status = client.test_boxed(ct).await?; let db_status = if status.healthy { "active" } else { "degraded" }; @@ -169,6 +192,7 @@ impl SearchIndexerService { self.http.clone(), Arc::clone(&self.cf_proxy), Duration::from_secs(self.config.request_timeout_secs), + self.config.max_response_body_bytes, ); let caps = client @@ -186,24 +210,28 @@ impl SearchIndexerService { location: snafu::Location::new(file!(), line!(), column!()), })?; let now = jiff::Timestamp::now().to_string(); - repo::update_indexer_caps(&self.write_pool, indexer_id, &caps_json, &now) + + // WHY: caps, categories, and status describe one observation of the + // indexer — a single transaction keeps them consistent under failure. + let mut tx = self + .write_pool + .begin() .await - .map_err(|e| SearchIndexerError::Database { - source: e, - location: snafu::Location::new(file!(), line!(), column!()), - })?; - repo::upsert_indexer_categories(&self.write_pool, indexer_id, &caps) + .context(TransactionSnafu) + .context(error::DatabaseSnafu)?; + repo::update_indexer_caps(&mut *tx, indexer_id, &caps_json, &now) .await - .map_err(|e| SearchIndexerError::Database { - source: e, - location: snafu::Location::new(file!(), line!(), column!()), - })?; - repo::update_indexer_status(&self.write_pool, indexer_id, "active") + .context(error::DatabaseSnafu)?; + repo::upsert_indexer_categories(&mut tx, indexer_id, &caps) .await - .map_err(|e| SearchIndexerError::Database { - source: e, - location: snafu::Location::new(file!(), line!(), column!()), - })?; + .context(error::DatabaseSnafu)?; + repo::update_indexer_status(&mut *tx, indexer_id, "active") + .await + .context(error::DatabaseSnafu)?; + tx.commit() + .await + .context(TransactionSnafu) + .context(error::DatabaseSnafu)?; Ok(caps) } @@ -221,13 +249,27 @@ impl SearchIndexerService { } async fn handle_search_error(&self, indexer: &IndexerRow, error: &SearchIndexerError) { + if let SearchIndexerError::RateLimited { + retry_after_seconds, + .. + } = error + { + let secs = retry_after_seconds + .unwrap_or(DEFAULT_RETRY_AFTER_SECS) + .min(MAX_RETRY_AFTER_SECS); + self.rate_limiter + .set_retry_after(indexer.id, Duration::from_secs(secs)) + .await; + } + let new_status = match error { SearchIndexerError::AuthFailed { .. } => Some("failed"), SearchIndexerError::NoCfBypass { .. } => Some("degraded"), SearchIndexerError::CfProxyTimeout { .. } | SearchIndexerError::CfProxyError { .. } => { Some("degraded") } - SearchIndexerError::ParseResponse { .. } => Some("degraded"), + SearchIndexerError::ParseResponse { .. } + | SearchIndexerError::ResponseTooLarge { .. } => Some("degraded"), SearchIndexerError::HttpRequest { .. } => { if indexer.status == "degraded" { Some("failed") @@ -235,6 +277,9 @@ impl SearchIndexerService { Some("degraded") } } + // WHY: cancellation is caller intent and a 429 is back-pressure — + // neither says the indexer itself is unhealthy. + SearchIndexerError::Cancelled { .. } | SearchIndexerError::RateLimited { .. } => None, _ => None, }; @@ -304,6 +349,7 @@ fn make_client( http: reqwest::Client, cf_proxy: Arc, timeout: Duration, + max_body_bytes: u64, ) -> Box { let config = IndexerConfig { id: indexer.id, @@ -314,8 +360,20 @@ fn make_client( }; match indexer.protocol.as_str() { - "newznab" => Box::new(NewznabClient::new(config, http, cf_proxy, timeout)), - _ => Box::new(TorznabClient::new(config, http, cf_proxy, timeout)), + "newznab" => Box::new(NewznabClient::new( + config, + http, + cf_proxy, + timeout, + max_body_bytes, + )), + _ => Box::new(TorznabClient::new( + config, + http, + cf_proxy, + timeout, + max_body_bytes, + )), } } @@ -500,4 +558,210 @@ mod tests { let eligible = filter_by_capability(&indexers, &query); assert_eq!(eligible.len(), 1); } + + // ── handle_search_error / refresh_caps (live service over in-memory db) ── + + use apotheke::migrate::MIGRATOR; + use themelion::create_event_bus; + + use crate::cf_bypass::noop::NoProxy; + use crate::repo::InsertIndexerParams; + use crate::test_support::spawn_one_shot_http; + + async fn make_service() -> (SearchIndexerService, SqlitePool) { + let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); + MIGRATOR.run(&pool).await.unwrap(); + let (event_tx, _) = create_event_bus(16); + let service = SearchIndexerService::new( + pool.clone(), + pool.clone(), + Arc::new(NoProxy), + SearchSubsystemConfig::default(), + event_tx, + ); + (service, pool) + } + + async fn seed_indexer(pool: &SqlitePool, url: &str) -> IndexerRow { + let id = repo::insert_indexer( + pool, + InsertIndexerParams { + name: "Seeded", + url, + protocol: "torznab", + api_key: Some("key"), + cf_bypass: false, + priority: 50, + }, + ) + .await + .unwrap(); + repo::get_indexer(pool, id).await.unwrap().unwrap() + } + + fn cancelled_error() -> SearchIndexerError { + SearchIndexerError::Cancelled { + url: "https://example.com/api".to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + } + } + + fn rate_limited_error(retry_after_seconds: Option) -> SearchIndexerError { + SearchIndexerError::RateLimited { + indexer_id: 1, + retry_after_seconds, + location: snafu::Location::new(file!(), line!(), column!()), + } + } + + #[tokio::test] + async fn handle_search_error_cancelled_leaves_status_unchanged() { + let (service, pool) = make_service().await; + let indexer = seed_indexer(&pool, "https://example.com/api").await; + assert_eq!(indexer.status, "active"); + + service + .handle_search_error(&indexer, &cancelled_error()) + .await; + + let row = repo::get_indexer(&pool, indexer.id).await.unwrap().unwrap(); + assert_eq!(row.status, "active"); + } + + #[tokio::test] + async fn handle_search_error_auth_failed_marks_failed() { + let (service, pool) = make_service().await; + let indexer = seed_indexer(&pool, "https://example.com/api").await; + + let error = SearchIndexerError::AuthFailed { + indexer_id: indexer.id, + location: snafu::Location::new(file!(), line!(), column!()), + }; + service.handle_search_error(&indexer, &error).await; + + let row = repo::get_indexer(&pool, indexer.id).await.unwrap().unwrap(); + assert_eq!(row.status, "failed"); + } + + // WHY: the clock is paused only around the limiter interaction — sqlx's + // sqlite worker runs on a real thread, and a paused clock during pool + // setup auto-advances straight into PoolTimedOut. + #[tokio::test] + async fn handle_search_error_rate_limited_engages_retry_after() { + let (service, pool) = make_service().await; + let indexer = seed_indexer(&pool, "https://example.com/api").await; + + tokio::time::pause(); + service + .handle_search_error(&indexer, &rate_limited_error(Some(120))) + .await; + + let before = tokio::time::Instant::now(); + service.rate_limiter.acquire(indexer.id).await; + let elapsed = before.elapsed(); + tokio::time::resume(); + assert!( + elapsed >= Duration::from_secs(120), + "expected >=120s back-off, got {elapsed:?}" + ); + + // 429 is back-pressure, not indexer failure — status must not change. + let row = repo::get_indexer(&pool, indexer.id).await.unwrap().unwrap(); + assert_eq!(row.status, "active"); + } + + #[tokio::test] + async fn handle_search_error_rate_limited_without_header_uses_default() { + let (service, pool) = make_service().await; + let indexer = seed_indexer(&pool, "https://example.com/api").await; + + tokio::time::pause(); + service + .handle_search_error(&indexer, &rate_limited_error(None)) + .await; + + let before = tokio::time::Instant::now(); + service.rate_limiter.acquire(indexer.id).await; + let elapsed = before.elapsed(); + tokio::time::resume(); + assert!( + elapsed >= Duration::from_secs(DEFAULT_RETRY_AFTER_SECS), + "expected >={DEFAULT_RETRY_AFTER_SECS}s default back-off, got {elapsed:?}" + ); + } + + #[tokio::test] + async fn handle_search_error_rate_limited_clamps_hostile_retry_after() { + let (service, pool) = make_service().await; + let indexer = seed_indexer(&pool, "https://example.com/api").await; + + tokio::time::pause(); + service + .handle_search_error(&indexer, &rate_limited_error(Some(999_999))) + .await; + + let before = tokio::time::Instant::now(); + service.rate_limiter.acquire(indexer.id).await; + let elapsed = before.elapsed(); + tokio::time::resume(); + assert!( + elapsed >= Duration::from_secs(MAX_RETRY_AFTER_SECS), + "expected the cap to engage, got {elapsed:?}" + ); + assert!( + elapsed < Duration::from_secs(MAX_RETRY_AFTER_SECS + 60), + "expected clamp at {MAX_RETRY_AFTER_SECS}s, got {elapsed:?}" + ); + } + + const CAPS_XML: &str = r#" + + + + + + + + + + + +"#; + + #[tokio::test] + async fn refresh_caps_commits_caps_categories_and_status_together() { + let (service, pool) = make_service().await; + let (url, _server) = spawn_one_shot_http(200, "OK", &[], CAPS_XML).await; + let indexer = seed_indexer(&pool, &url).await; + repo::update_indexer_status(&pool, indexer.id, "degraded") + .await + .unwrap(); + + let caps = service + .refresh_caps(indexer.id, CancellationToken::new()) + .await + .unwrap(); + assert_eq!(caps.categories.len(), 1); + + let row = repo::get_indexer(&pool, indexer.id).await.unwrap().unwrap(); + assert_eq!(row.status, "active"); + assert!(row.caps_json.is_some()); + assert!(row.last_tested.is_some()); + + let categories = sqlx::query_as::<_, (i64, String)>( + "SELECT category_id, name FROM indexer_categories + WHERE indexer_id = ? ORDER BY category_id", + ) + .bind(indexer.id) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!( + categories, + vec![ + (2000, "Movies".to_string()), + (2010, "Movies/Foreign".to_string()) + ] + ); + } } diff --git a/crates/zetesis/src/test_support.rs b/crates/zetesis/src/test_support.rs new file mode 100644 index 00000000..c394d832 --- /dev/null +++ b/crates/zetesis/src/test_support.rs @@ -0,0 +1,80 @@ +//! Shared test fixtures — one-shot HTTP servers for indexer client tests. +#![cfg(test)] + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio::task::JoinHandle; + +/// Spawns a TCP server that answers exactly one HTTP request with the given +/// raw response bytes, then resolves to the raw request head it received. +pub(crate) async fn spawn_raw_http(raw_response: Vec) -> (String, JoinHandle) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + + let handle = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + + loop { + let n = stream.read(&mut chunk).await.unwrap(); + assert!(n > 0, "client closed before sending full headers"); + buf.extend_from_slice(&chunk[..n]); + if find_subslice(&buf, b"\r\n\r\n").is_some() { + break; + } + } + + stream.write_all(&raw_response).await.unwrap(); + stream.flush().await.unwrap(); + stream.shutdown().await.unwrap(); + + String::from_utf8_lossy(&buf).into_owned() + }); + + (base_url, handle) +} + +/// Spawns a TCP server that answers exactly one HTTP request with `status`, +/// the given extra headers, and `body` (with a correct `Content-Length`). +pub(crate) async fn spawn_one_shot_http( + status: u16, + reason: &str, + extra_headers: &[(&str, &str)], + body: &str, +) -> (String, JoinHandle) { + let mut response = format!("HTTP/1.1 {status} {reason}\r\n"); + for (name, value) in extra_headers { + response.push_str(&format!("{name}: {value}\r\n")); + } + response.push_str(&format!( + "content-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + )); + spawn_raw_http(response.into_bytes()).await +} + +/// Spawns a TCP server that accepts one connection and never responds. +pub(crate) async fn spawn_hang_http() -> (String, JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + + let handle = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut chunk = [0u8; 4096]; + // WHY: keep reading so the client's request write succeeds, but never + // write a response — the connection hangs until the task is dropped. + loop { + let n = stream.read(&mut chunk).await.unwrap_or(0); + if n == 0 { + std::future::pending::<()>().await; + } + } + }); + + (base_url, handle) +} + +fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option { + haystack.windows(needle.len()).position(|w| w == needle) +} diff --git a/deny.toml b/deny.toml index b1879000..bfa61d43 100644 --- a/deny.toml +++ b/deny.toml @@ -48,6 +48,14 @@ reason = "audiopus_sys unmaintained via opus/akouo-core; no safe upgrade availab id = "RUSTSEC-2026-0097" reason = "rand unsound with custom logger using `rand::rng()` — surfaces via sqlx/axum/quinn/reqwest; harmonia installs no custom logger that re-enters rand::rng(), so the unsound path is unreachable" +[[advisories.ignore]] +id = "RUSTSEC-2026-0194" +reason = "quick-xml < 0.41 quadratic-attribute DoS. The direct indexer-XML parser (zetesis) is on 0.41; the remaining 0.37.5 is transitively pinned by feed-rs (RSS/podcast) and librqbit-upnp (LAN UPnP), neither of which yet targets 0.41. No safe upgrade until upstream bumps. review-by: 2026-10-01" + +[[advisories.ignore]] +id = "RUSTSEC-2026-0195" +reason = "quick-xml < 0.41 NsReader namespace-allocation DoS. Same transitive pin as RUSTSEC-2026-0194: the direct zetesis parser is on 0.41; feed-rs and librqbit-upnp still require 0.37. No safe upgrade until upstream bumps. review-by: 2026-10-01" + [licenses] allow = [ "MIT", @@ -85,9 +93,9 @@ skip = [ { name = "num-complex", version = "0.2" }, { name = "num-complex", version = "0.4" }, - # quick-xml: feed-rs and librqbit-upnp require 0.37; ecosystem uses 0.39 + # quick-xml: feed-rs and librqbit-upnp require 0.37; zetesis indexer uses 0.41 { name = "quick-xml", version = "0.37" }, - { name = "quick-xml", version = "0.39" }, + { name = "quick-xml", version = "0.41" }, # block-buffer: digest 0.10 (transitive) uses 0.10.x; digest 0.11 (sha2 0.11) uses 0.12.x { name = "block-buffer", version = "0.10" }, diff --git a/docs/architecture/configuration.md b/docs/architecture/configuration.md index f4ede63a..84dcea93 100644 --- a/docs/architecture/configuration.md +++ b/docs/architecture/configuration.md @@ -21,6 +21,7 @@ Horismos is the single source of truth for all system configuration. No other su # Indexer search request_timeout_secs = 30 max_results_per_indexer = 100 +max_response_body_bytes = 16777216 # 16 MiB cap on any single indexer response cloudflare_bypass_enabled = false [exousia] @@ -256,6 +257,7 @@ impl Default for ExousiaConfig { pub struct ZetesisConfig { pub request_timeout_secs: u64, // default: 30 pub max_results_per_indexer: usize, // default: 100 + pub max_response_body_bytes: u64, // default: 16 MiB pub cloudflare_bypass_enabled: bool, // default: false } @@ -264,6 +266,7 @@ impl Default for ZetesisConfig { Self { request_timeout_secs: 30, max_results_per_indexer: 100, + max_response_body_bytes: 16 * 1024 * 1024, cloudflare_bypass_enabled: false, } } diff --git a/osv-scanner.toml b/osv-scanner.toml index a0f5b08e..781650e4 100644 --- a/osv-scanner.toml +++ b/osv-scanner.toml @@ -1,38 +1,47 @@ -# WHY: OSV-Scanner does not read cargo-audit's .cargo/audit.toml or -# cargo-deny's deny.toml. Keep this ignore list in sync with both files. +# WHY: This file is DERIVED from deny.toml [[advisories.ignore]]. +# Do not edit the ignore list here; edit deny.toml and run +# `kanon audit derive-ignores --apply`. See standards/SUPPLY-CHAIN.md. [[IgnoredVulns]] id = "RUSTSEC-2023-0071" -reason = "rsa timing side-channel via jsonwebtoken/exousia — no safe upgrade, local-only use." +reason = "rsa timing side-channel via jsonwebtoken/exousia — no safe upgrade, local-only use" [[IgnoredVulns]] id = "RUSTSEC-2025-0012" -reason = "backoff unmaintained — transitive dep via librqbit." +reason = "backoff unmaintained — transitive dep via librqbit" [[IgnoredVulns]] id = "RUSTSEC-2025-0141" -reason = "bincode unmaintained — transitive dep via librqbit-peer-protocol. No safe upgrade: bincode 2.x is a breaking-change rewrite and upstream has not migrated." +reason = "bincode unmaintained — transitive dep via librqbit-peer-protocol" [[IgnoredVulns]] id = "RUSTSEC-2025-0060" -reason = "crypto-hash unmaintained — transitive dep via librqbit-sha1-wrapper." +reason = "crypto-hash unmaintained — transitive dep via librqbit-sha1-wrapper" [[IgnoredVulns]] id = "RUSTSEC-2024-0384" -reason = "instant unmaintained — transitive dep via backoff/librqbit, no maintained alternative on the librqbit chain." +reason = "instant unmaintained — transitive dep via backoff/librqbit, no alternative" [[IgnoredVulns]] id = "RUSTSEC-2024-0436" -reason = "paste unmaintained (dtolnay archived) — transitive dep via lofty/taxis." +reason = "paste unmaintained (dtolnay archived) — transitive dep via lofty/taxis" [[IgnoredVulns]] id = "RUSTSEC-2026-0009" -reason = "time RFC2822 stack exhaustion fix requires Rust 1.88; harmonia CI is pinned to Rust 1.85 until MSRV policy changes." +reason = "time RFC2822 stack exhaustion fix requires Rust 1.88; Harmonia CI is pinned to Rust 1.85 until MSRV policy changes" [[IgnoredVulns]] id = "RUSTSEC-2026-0150" -reason = "audiopus_sys unmaintained via opus/akouo-core; no safe upgrade available." +reason = "audiopus_sys unmaintained via opus/akouo-core; no safe upgrade available" [[IgnoredVulns]] id = "RUSTSEC-2026-0097" -reason = "rand unsound with custom logger using rand::rng() — surfaces via sqlx/axum/quinn/reqwest; harmonia installs no custom logger that re-enters rand::rng(), so the unsound path is unreachable." +reason = "rand unsound with custom logger using `rand::rng()` — surfaces via sqlx/axum/quinn/reqwest; harmonia installs no custom logger that re-enters rand::rng(), so the unsound path is unreachable" + +[[IgnoredVulns]] +id = "RUSTSEC-2026-0194" +reason = "quick-xml < 0.41 quadratic-attribute DoS. The direct indexer-XML parser (zetesis) is on 0.41; the remaining 0.37.5 is transitively pinned by feed-rs (RSS/podcast) and librqbit-upnp (LAN UPnP), neither of which yet targets 0.41. No safe upgrade until upstream bumps. review-by: 2026-10-01" + +[[IgnoredVulns]] +id = "RUSTSEC-2026-0195" +reason = "quick-xml < 0.41 NsReader namespace-allocation DoS. Same transitive pin as RUSTSEC-2026-0194: the direct zetesis parser is on 0.41; feed-rs and librqbit-upnp still require 0.37. No safe upgrade until upstream bumps. review-by: 2026-10-01"