diff --git a/Cargo.lock b/Cargo.lock index 1d4f4f2..c99a434 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2507,6 +2507,7 @@ dependencies = [ "async-trait", "chromiumoxide", "chrono", + "futures-util", "relais-core", "reqwest", "serde", @@ -2546,6 +2547,7 @@ dependencies = [ "bytes", "encoding_rs", "futures-core", + "futures-util", "h2", "http", "http-body", @@ -2567,12 +2569,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-native-tls", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", ] @@ -3595,6 +3599,19 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasmparser" version = "0.244.0" diff --git a/Cargo.toml b/Cargo.toml index 308eca7..4e809e2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,7 +26,7 @@ thiserror = "2" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } async-trait = "0.1" -reqwest = { version = "0.12", features = ["json"] } +reqwest = { version = "0.12", features = ["json", "stream"] } chrono = { version = "0.4", features = ["serde"] } relais-core = { path = "crates/core", version = "0.1.0" } diff --git a/crates/adapters/github/src/lib.rs b/crates/adapters/github/src/lib.rs index 4afb7dd..20f95d3 100644 --- a/crates/adapters/github/src/lib.rs +++ b/crates/adapters/github/src/lib.rs @@ -16,7 +16,7 @@ pub struct GitHubAdapter { impl GitHubAdapter { pub fn new() -> Self { Self { - client: Client::new(), + client: relais_core::http::client(relais_core::http::Profile::Default), } } } diff --git a/crates/adapters/hackernews/src/lib.rs b/crates/adapters/hackernews/src/lib.rs index b3f36d6..c424c5b 100644 --- a/crates/adapters/hackernews/src/lib.rs +++ b/crates/adapters/hackernews/src/lib.rs @@ -17,7 +17,7 @@ pub struct HackerNewsAdapter { impl HackerNewsAdapter { pub fn new() -> Self { Self { - client: Client::new(), + client: relais_core::http::client(relais_core::http::Profile::Default), } } diff --git a/crates/adapters/scs-legacy/src/lib.rs b/crates/adapters/scs-legacy/src/lib.rs index 38fdac1..3fa124e 100644 --- a/crates/adapters/scs-legacy/src/lib.rs +++ b/crates/adapters/scs-legacy/src/lib.rs @@ -76,7 +76,7 @@ impl ScsLegacyAdapter { ); } Self { - client: Client::new(), + client: relais_core::http::client(relais_core::http::Profile::Default), base_url, } } diff --git a/crates/adapters/scs/src/lib.rs b/crates/adapters/scs/src/lib.rs index e127625..b577924 100644 --- a/crates/adapters/scs/src/lib.rs +++ b/crates/adapters/scs/src/lib.rs @@ -37,7 +37,7 @@ impl ScsAdapter { pub fn with_base_url(base_url: impl Into) -> Self { let base_url = base_url.into().trim_end_matches('/').to_string(); Self { - client: Client::new(), + client: relais_core::http::client(relais_core::http::Profile::Default), base_url, } } diff --git a/crates/cli/src/commands/auth.rs b/crates/cli/src/commands/auth.rs index c179101..7d3088b 100644 --- a/crates/cli/src/commands/auth.rs +++ b/crates/cli/src/commands/auth.rs @@ -144,7 +144,7 @@ async fn run_oauth_flow(config: &OAuthConfig, site_id: &str) -> Result<()> { }; // 6. Exchange code for tokens. - let client = reqwest::Client::new(); + let client = relais_core::http::client(relais_core::http::Profile::Default); let token_response = client .post(&config.token_url) .header("Accept", "application/json") diff --git a/crates/core/src/http.rs b/crates/core/src/http.rs new file mode 100644 index 0000000..107ff4e --- /dev/null +++ b/crates/core/src/http.rs @@ -0,0 +1,32 @@ +//! Shared outbound HTTP client builder with sane timeouts (always compiled). +//! +//! Every adapter/provider/token-refresh path should build its `reqwest::Client` +//! here so a stalled upstream fails by timeout instead of pinning a task forever. + +use std::time::Duration; + +/// Timeout profile. Most upstreams are quick API calls; LLM-provider completions are +/// legitimately slow, so they get a longer ceiling. +#[derive(Debug, Clone, Copy)] +pub enum Profile { + /// Adapters, OAuth token exchange, token refresh. + Default, + /// LLM provider completions (slow, but still bounded). + Llm, +} + +/// Build a `reqwest::Client` with connect + total timeouts for the given profile. +pub fn client(profile: Profile) -> reqwest::Client { + let (connect_secs, total_secs) = match profile { + Profile::Default => (5, 30), + Profile::Llm => (10, 180), + }; + reqwest::Client::builder() + .connect_timeout(Duration::from_secs(connect_secs)) + .timeout(Duration::from_secs(total_secs)) + .user_agent("relais/0.1") + .build() + // Building only fails on TLS-backend/resolver init, which is fatal config — + // fail closed at startup rather than silently returning a no-timeout client. + .expect("failed to build the HTTP client (TLS backend init?)") +} diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index a77f1b7..f25ce68 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -2,6 +2,7 @@ pub mod adapter; #[cfg(feature = "audit")] pub mod audit; pub mod error; +pub mod http; pub mod net_guard; pub mod oauth; pub mod redact; diff --git a/crates/core/src/token_refresh.rs b/crates/core/src/token_refresh.rs index f61425c..dd05e1b 100644 --- a/crates/core/src/token_refresh.rs +++ b/crates/core/src/token_refresh.rs @@ -53,7 +53,7 @@ pub async fn maybe_refresh( .ok_or_else(|| RefreshError::NoProviderConfig(site_id.to_string()))?; // Exchange the refresh token for a new access token. - let client = reqwest::Client::new(); + let client = crate::http::client(crate::http::Profile::Default); let response = client .post(&config.token_url) .header("Accept", "application/json") diff --git a/crates/llm-fallback/Cargo.toml b/crates/llm-fallback/Cargo.toml index c93adf2..d506ed7 100644 --- a/crates/llm-fallback/Cargo.toml +++ b/crates/llm-fallback/Cargo.toml @@ -14,6 +14,7 @@ async-trait.workspace = true tokio.workspace = true reqwest.workspace = true chromiumoxide = { version = "0.7", features = ["tokio-runtime"] } +futures-util = "0.3" tracing.workspace = true thiserror.workspace = true anyhow.workspace = true diff --git a/crates/llm-fallback/src/browser.rs b/crates/llm-fallback/src/browser.rs index ee75079..1d61349 100644 --- a/crates/llm-fallback/src/browser.rs +++ b/crates/llm-fallback/src/browser.rs @@ -1,11 +1,15 @@ use std::collections::HashMap; use std::time::Duration; +use futures_util::StreamExt; use relais_core::net_guard; use tracing::debug; use crate::LlmError; +/// Hard cap on a fetched body, to bound memory against a hostile/huge response. +const MAX_BODY_BYTES: usize = 8 * 1024 * 1024; + /// Cookies plus the domain they were captured for, so they are only ever sent to a /// matching host. pub struct CookieScope { @@ -76,7 +80,28 @@ pub async fn fetch_html(url: &str, cookie_scope: Option<&CookieScope>) -> Result return Err(LlmError::Browser(format!("HTTP {}", resp.status()))); } - Ok(resp.text().await?) + // Reject an oversized declared body up front. + if let Some(len) = resp.content_length() { + if len > MAX_BODY_BYTES as u64 { + return Err(LlmError::Browser(format!( + "response too large: {len} bytes (cap {MAX_BODY_BYTES})" + ))); + } + } + + // Stream with a running byte cap so a chunked/undeclared body can't exhaust memory. + let mut buf: Vec = Vec::new(); + let mut stream = resp.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + let remaining = MAX_BODY_BYTES.saturating_sub(buf.len()); + if chunk.len() >= remaining { + buf.extend_from_slice(&chunk[..remaining]); + break; // hit the cap; truncate + } + buf.extend_from_slice(&chunk); + } + Ok(String::from_utf8_lossy(&buf).into_owned()) } /// Whether `host` is on `RELAIS_FALLBACK_ALLOW` (comma-separated). Unset/empty ⇒ diff --git a/crates/llm-fallback/src/extractor.rs b/crates/llm-fallback/src/extractor.rs index ba9bcd6..be33644 100644 --- a/crates/llm-fallback/src/extractor.rs +++ b/crates/llm-fallback/src/extractor.rs @@ -3,6 +3,19 @@ use serde_json::Value; const MAX_HTML_LEN: usize = 50_000; +/// Largest byte index `<= max` that is a UTF-8 char boundary of `s` (so slicing +/// `&s[..n]` never panics by splitting a multibyte char). +fn floor_char_boundary(s: &str, max: usize) -> usize { + if max >= s.len() { + return s.len(); + } + let mut end = max; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + end +} + pub struct Extractor { provider: Box, } @@ -13,7 +26,9 @@ impl Extractor { } pub async fn extract(&self, html: &str, action: &str) -> Result { - let truncated = &html[..html.len().min(MAX_HTML_LEN)]; + // Truncate on a UTF-8 char boundary: byte-indexing a `str` panics if it + // splits a multibyte char. + let truncated = &html[..floor_char_boundary(html, MAX_HTML_LEN)]; let prompt = format!( "You are a web data extraction agent. Given the following HTML content, \ extract the requested information and return it as JSON.\n\n\ @@ -24,9 +39,26 @@ impl Extractor { let response = self.provider.complete(&prompt).await?; - let data: Value = serde_json::from_str(&response) - .map_err(|e| LlmError::ParseError(e.to_string()))?; + let data: Value = + serde_json::from_str(&response).map_err(|e| LlmError::ParseError(e.to_string()))?; Ok(data) } } + +#[cfg(test)] +mod tests { + use super::floor_char_boundary; + + #[test] + fn floor_backs_off_inside_multibyte_char() { + // "a€b" = [0x61, 0xE2,0x82,0xAC, 0x62] — '€' spans bytes 1..4. + let s = "a€b"; + assert_eq!(floor_char_boundary(s, 2), 1); // inside € → back to 1 + assert_eq!(floor_char_boundary(s, 3), 1); + assert_eq!(floor_char_boundary(s, 4), 4); // start of 'b' + assert_eq!(floor_char_boundary(s, 100), s.len()); + // the crux: slicing at the returned index never panics + let _ = &s[..floor_char_boundary(s, 2)]; + } +} diff --git a/crates/llm-fallback/src/providers/anthropic.rs b/crates/llm-fallback/src/providers/anthropic.rs index c131d9c..2293743 100644 --- a/crates/llm-fallback/src/providers/anthropic.rs +++ b/crates/llm-fallback/src/providers/anthropic.rs @@ -14,7 +14,7 @@ impl AnthropicClient { Self { api_key, model, - client: reqwest::Client::new(), + client: relais_core::http::client(relais_core::http::Profile::Llm), } } } diff --git a/crates/llm-fallback/src/providers/ollama.rs b/crates/llm-fallback/src/providers/ollama.rs index 6f9899c..a71bb3c 100644 --- a/crates/llm-fallback/src/providers/ollama.rs +++ b/crates/llm-fallback/src/providers/ollama.rs @@ -22,7 +22,7 @@ impl OllamaClient { Self { base_url, model, - client: reqwest::Client::new(), + client: relais_core::http::client(relais_core::http::Profile::Llm), } } } diff --git a/crates/llm-fallback/src/providers/openai.rs b/crates/llm-fallback/src/providers/openai.rs index 12c1471..dad6a68 100644 --- a/crates/llm-fallback/src/providers/openai.rs +++ b/crates/llm-fallback/src/providers/openai.rs @@ -14,7 +14,7 @@ impl OpenAiClient { Self { api_key, model, - client: reqwest::Client::new(), + client: relais_core::http::client(relais_core::http::Profile::Llm), } } }