use async_trait::async_trait;
use crw_core::Deadline;
use crw_core::error::{CrwError, CrwResult};
use crw_core::types::FetchResult;
use std::collections::HashMap;
use std::time::Instant;
use crate::traits::PageFetcher;
const MAX_RESPONSE_BYTES: usize = 50 * 1024 * 1024;
const HTTP_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
const HTTP_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
const HTTP_MAX_RETRIES: u32 = 1;
const HTTP_RETRY_BACKOFF: std::time::Duration = std::time::Duration::from_millis(250);
fn is_retriable_error(e: &reqwest::Error) -> bool {
e.is_connect() || e.is_timeout()
}
fn is_retriable_status(status: u16) -> bool {
matches!(status, 502..=504)
}
fn is_ratelimit_status(status: u16) -> bool {
matches!(status, 429)
}
fn should_arm_proxy(status: u16, cf_mitigated: bool) -> bool {
is_ratelimit_status(status) || cf_mitigated
}
fn tls_relaxed_fallback_enabled() -> bool {
std::env::var("CRW_HTTP_TLS_RELAXED_FALLBACK")
.map(|v| {
let v = v.trim().to_ascii_lowercase();
v == "true" || v == "1" || v == "yes"
})
.unwrap_or(false)
}
fn ratelimit_proxy_url() -> Option<String> {
std::env::var("CRW_HTTP_RATELIMIT_PROXY_URL")
.ok()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
}
fn is_cert_error(e: &reqwest::Error) -> bool {
let mut src: Option<&(dyn std::error::Error + 'static)> = Some(e);
while let Some(s) = src {
let m = s.to_string().to_ascii_lowercase();
if m.contains("certificate")
|| m.contains("peerfailedverification")
|| m.contains("sslconnecterror")
|| m.contains("invalid peer cert")
|| m.contains("certusedasend")
|| m.contains("cert verify")
|| m.contains("tls handshake")
|| (m.contains("ssl") && (m.contains("verif") || m.contains("cert")))
{
return true;
}
src = s.source();
}
false
}
const STEALTH_ACCEPT: &str =
"text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8";
const STEALTH_SEC_CH_UA: &str =
r#""Google Chrome";v="150", "Chromium";v="150", "Not_A Brand";v="24""#;
fn build_client(
user_agent: &str,
proxy: Option<&str>,
request_timeout: std::time::Duration,
relaxed_tls: bool,
) -> CrwResult<reqwest::Client> {
let mut builder = reqwest::Client::builder()
.user_agent(user_agent)
.connect_timeout(HTTP_CONNECT_TIMEOUT)
.timeout(request_timeout)
.redirect(crw_core::url_safety::safe_redirect_policy());
if relaxed_tls {
builder = builder
.danger_accept_invalid_certs(true)
.danger_accept_invalid_hostnames(true);
}
if let Some(proxy_url) = proxy {
let p = reqwest::Proxy::all(proxy_url)
.map_err(|e| CrwError::ConfigError(format!("invalid proxy URL '{proxy_url}': {e}")))?;
builder = builder.proxy(p);
}
builder
.build()
.map_err(|e| CrwError::ConfigError(format!("failed to build HTTP client: {e}")))
}
pub struct HttpFetcher {
client: reqwest::Client,
relaxed_client: Option<reqwest::Client>,
ratelimit_proxy_client: Option<reqwest::Client>,
inject_stealth_headers: bool,
}
impl HttpFetcher {
pub fn new(user_agent: &str, proxy: Option<&str>, inject_stealth_headers: bool) -> Self {
Self::with_timeout(
user_agent,
proxy,
inject_stealth_headers,
HTTP_REQUEST_TIMEOUT,
)
}
pub fn with_timeout(
user_agent: &str,
proxy: Option<&str>,
inject_stealth_headers: bool,
request_timeout: std::time::Duration,
) -> Self {
let client = build_client(user_agent, proxy, request_timeout, false).unwrap_or_else(|e| {
tracing::error!("{e}, using default client");
reqwest::Client::new()
});
let relaxed_client = if tls_relaxed_fallback_enabled() {
build_client(user_agent, proxy, request_timeout, true).ok()
} else {
None
};
let ratelimit_proxy_client = ratelimit_proxy_url().and_then(|purl| {
build_client(user_agent, Some(purl.as_str()), request_timeout, false).ok()
});
Self {
client,
relaxed_client,
ratelimit_proxy_client,
inject_stealth_headers,
}
}
pub fn with_proxy(
user_agent: &str,
proxy_url: &str,
inject_stealth_headers: bool,
request_timeout: std::time::Duration,
) -> CrwResult<Self> {
let client = build_client(user_agent, Some(proxy_url), request_timeout, false)?;
let relaxed_client = if tls_relaxed_fallback_enabled() {
build_client(user_agent, Some(proxy_url), request_timeout, true).ok()
} else {
None
};
let ratelimit_proxy_client = ratelimit_proxy_url().and_then(|purl| {
build_client(user_agent, Some(purl.as_str()), request_timeout, false).ok()
});
Ok(Self {
client,
relaxed_client,
ratelimit_proxy_client,
inject_stealth_headers,
})
}
}
#[async_trait]
impl PageFetcher for HttpFetcher {
async fn fetch(
&self,
url: &str,
headers: &HashMap<String, String>,
_wait_for_ms: Option<u64>,
deadline: Deadline,
) -> CrwResult<FetchResult> {
if deadline.expired() {
return Err(CrwError::HttpError(format!(
"deadline expired before HTTP fetch of {url}"
)));
}
let start = Instant::now();
let build_request = |client: &reqwest::Client| {
let mut req = client.get(url);
if self.inject_stealth_headers {
req = req
.header("Accept", STEALTH_ACCEPT)
.header("Accept-Language", "en-US,en;q=0.9")
.header("Sec-Ch-Ua", STEALTH_SEC_CH_UA)
.header("Sec-Ch-Ua-Mobile", "?0")
.header("Sec-Ch-Ua-Platform", "\"Windows\"")
.header("Sec-Fetch-Dest", "document")
.header("Sec-Fetch-Mode", "navigate")
.header("Sec-Fetch-Site", "none")
.header("Sec-Fetch-User", "?1")
.header("Upgrade-Insecure-Requests", "1")
.header("Priority", "u=0, i");
}
for (k, v) in headers {
req = req.header(k.as_str(), v.as_str());
}
req
};
let mut attempt: u32 = 0;
let mut use_relaxed = false;
let mut use_proxy = false;
let resp = loop {
let remaining = deadline.remaining();
if remaining.is_zero() {
return Err(CrwError::Timeout(
(start.elapsed().as_millis().max(1)) as u64,
));
}
let active_client = if use_proxy {
self.ratelimit_proxy_client.as_ref().unwrap_or(&self.client)
} else if use_relaxed {
self.relaxed_client.as_ref().unwrap_or(&self.client)
} else {
&self.client
};
let send_fut = build_request(active_client).send();
let send_result = tokio::time::timeout(remaining, send_fut).await;
match send_result {
Err(_) => {
return Err(CrwError::Timeout(remaining.as_millis() as u64));
}
Ok(Ok(r))
if attempt < HTTP_MAX_RETRIES && is_retriable_status(r.status().as_u16()) =>
{
tracing::debug!(
"HTTP {} from {url}, retrying (attempt {})",
r.status(),
attempt + 1
);
drop(r);
attempt += 1;
let backoff = HTTP_RETRY_BACKOFF.min(deadline.remaining());
if !backoff.is_zero() {
tokio::time::sleep(backoff).await;
}
}
Ok(Ok(r))
if !use_proxy
&& self.ratelimit_proxy_client.is_some()
&& should_arm_proxy(
r.status().as_u16(),
r.headers()
.get("cf-mitigated")
.and_then(|v| v.to_str().ok())
.map(crate::detector::is_cloudflare_mitigated_header)
.unwrap_or(false),
) =>
{
tracing::warn!(
"HTTP {} from {url} (origin rate-limited or cf-mitigated); retrying once via proxy (ratelimit_bypassed)",
r.status()
);
drop(r);
use_proxy = true;
}
Ok(Ok(r)) => break r,
Ok(Err(e))
if !use_relaxed && self.relaxed_client.is_some() && is_cert_error(&e) =>
{
tracing::warn!(
"TLS verification failed for {url} ({e}); retrying once with relaxed TLS (tls_unverified)"
);
use_relaxed = true;
}
Ok(Err(e)) if attempt < HTTP_MAX_RETRIES && is_retriable_error(&e) => {
tracing::debug!(
"transient HTTP error to {url} ({e}), retrying (attempt {})",
attempt + 1
);
attempt += 1;
let backoff = HTTP_RETRY_BACKOFF.min(deadline.remaining());
if !backoff.is_zero() {
tokio::time::sleep(backoff).await;
}
}
Ok(Err(e)) => {
return Err(if e.is_connect() {
CrwError::TargetUnreachable(format!("Could not reach {url}: {e}"))
} else {
CrwError::HttpError(e.to_string())
});
}
}
};
let status = resp.status().as_u16();
if let Some(len) = resp.content_length()
&& len as usize > MAX_RESPONSE_BYTES
{
return Err(CrwError::HttpError(format!(
"Response too large: {len} bytes (max {MAX_RESPONSE_BYTES})"
)));
}
let content_type_header = resp
.headers()
.get("content-type")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
let content_type = content_type_header
.as_deref()
.map(|s| s.split(';').next().unwrap_or(s).trim().to_lowercase());
let header_charset = content_type_header
.as_deref()
.and_then(charset_from_content_type);
let cf_mitigated = resp
.headers()
.get("cf-mitigated")
.and_then(|v| v.to_str().ok())
.map(crate::detector::is_cloudflare_mitigated_header)
.unwrap_or(false);
let is_pdf = content_type.as_deref() == Some("application/pdf");
let final_url_str = resp.url().as_str().to_string();
let bytes = resp
.bytes()
.await
.map_err(|e| CrwError::HttpError(e.to_string()))?;
if bytes.len() > MAX_RESPONSE_BYTES {
return Err(CrwError::HttpError(format!(
"Response too large: {} bytes (max {MAX_RESPONSE_BYTES})",
bytes.len()
)));
}
let (html, raw_bytes) = if is_pdf {
(String::new(), Some(bytes.to_vec()))
} else {
(decode_html_bytes(&bytes, header_charset.as_deref()), None)
};
let final_url = if final_url_str != url {
Some(final_url_str)
} else {
None
};
Ok(FetchResult {
url: url.to_string(),
final_url,
status_code: status,
html,
content_type,
raw_bytes,
rendered_with: if is_pdf {
Some("pdf".to_string())
} else {
Some("http".to_string())
},
elapsed_ms: start.elapsed().as_millis() as u64,
warning: if cf_mitigated {
Some("cloudflare_mitigated".to_string())
} else {
None
},
render_decision: None,
credit_cost: 0,
warnings: if cf_mitigated {
vec!["cf-mitigated header indicates Cloudflare challenge or block".to_string()]
} else {
Vec::new()
},
truncated: false,
deadline_exceeded: false,
captured_responses: Vec::new(),
screenshot: None,
})
}
fn name(&self) -> &str {
"http"
}
fn supports_js(&self) -> bool {
false
}
async fn is_available(&self) -> bool {
true
}
}
fn charset_from_content_type(ct: &str) -> Option<String> {
let lower = ct.to_ascii_lowercase();
let idx = lower.find("charset")?;
let after = ct[idx + "charset".len()..].trim_start();
let after = after.strip_prefix('=')?.trim_start();
let after = after.trim_start_matches(['"', '\'']);
let end = after
.find(|c: char| !(c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == ':'))
.unwrap_or(after.len());
let label = after[..end].trim();
(!label.is_empty()).then(|| label.to_string())
}
fn sniff_meta_charset(bytes: &[u8]) -> Option<String> {
let head = &bytes[..bytes.len().min(2048)];
let text = String::from_utf8_lossy(head).to_ascii_lowercase();
let idx = text.find("charset")?;
let after = text[idx + "charset".len()..].trim_start();
let after = after.strip_prefix('=')?.trim_start();
let after = after.trim_start_matches(['"', '\'']);
let end = after
.find(|c: char| !(c.is_ascii_alphanumeric() || c == '-' || c == '_'))
.unwrap_or(after.len());
let label = &after[..end];
(!label.is_empty()).then(|| label.to_string())
}
fn decode_html_bytes(bytes: &[u8], header_charset: Option<&str>) -> String {
let enc = header_charset
.and_then(|l| encoding_rs::Encoding::for_label(l.as_bytes()))
.or_else(|| {
sniff_meta_charset(bytes).and_then(|l| encoding_rs::Encoding::for_label(l.as_bytes()))
});
match enc {
Some(enc) => enc.decode(bytes).0.into_owned(),
None => String::from_utf8_lossy(bytes).into_owned(),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn should_arm_proxy_truth_table() {
assert!(should_arm_proxy(429, false), "429 arms on its own");
assert!(
!should_arm_proxy(403, false),
"403 without header does not arm"
);
assert!(should_arm_proxy(403, true), "403 + cf-mitigated arms");
assert!(should_arm_proxy(200, true), "challenge served as 200 arms");
assert!(!should_arm_proxy(200, false), "clean 200 does not arm");
}
#[test]
fn charset_from_content_type_parses_label() {
assert_eq!(
charset_from_content_type("text/html; charset=ISO-8859-1").as_deref(),
Some("ISO-8859-1")
);
assert_eq!(
charset_from_content_type("text/html;charset=\"utf-8\"").as_deref(),
Some("utf-8")
);
assert_eq!(charset_from_content_type("text/html").as_deref(), None);
}
#[test]
fn decode_latin1_via_header_no_replacement_char() {
let bytes = b"caf\xE9 \xFBber";
let out = decode_html_bytes(bytes, Some("iso-8859-1"));
assert_eq!(out, "café ûber");
assert!(!out.contains('\u{FFFD}'));
}
#[test]
fn decode_windows1254_via_meta_sniff() {
let bytes = b"<meta charset=windows-1254><p>i\xE7in</p>";
let out = decode_html_bytes(bytes, None);
assert!(out.contains("için"), "got: {out}");
assert!(!out.contains('\u{FFFD}'));
}
#[test]
fn decode_bogus_header_falls_through_to_meta_then_utf8() {
let bytes = b"<meta charset=windows-1254><p>i\xE7in</p>";
let out = decode_html_bytes(bytes, Some("x-bogus-nonsense"));
assert!(out.contains("için"), "got: {out}");
let plain = decode_html_bytes(b"plain ascii", Some("x-bogus"));
assert_eq!(plain, "plain ascii");
}
#[test]
fn decode_utf8_unchanged() {
let bytes = "café İstanbul 東京".as_bytes();
assert_eq!(
decode_html_bytes(bytes, Some("utf-8")),
"café İstanbul 東京"
);
assert_eq!(decode_html_bytes(bytes, None), "café İstanbul 東京");
}
#[test]
fn with_proxy_is_fail_closed_on_bad_url() {
assert!(
HttpFetcher::with_proxy("ua", "", false, std::time::Duration::from_secs(5)).is_err()
);
assert!(
HttpFetcher::with_proxy("ua", "not a url", false, std::time::Duration::from_secs(5))
.is_err()
);
}
#[test]
fn with_proxy_accepts_valid_url() {
assert!(
HttpFetcher::with_proxy(
"ua",
"http://user:pass@host:8080",
false,
std::time::Duration::from_secs(5),
)
.is_ok()
);
}
}