use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use bytes::Bytes;
use reqwest::Method;
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
use serde::Serialize;
use serde::de::DeserializeOwned;
#[derive(Debug, thiserror::Error)]
pub enum ClientError {
#[error("outbound HTTP request failed: {0}")]
Request(#[from] reqwest::Error),
#[error("JSON error: {0}")]
Json(#[from] serde_json::Error),
#[error("no mock registered for {0} {1}")]
NoMock(String, String),
#[error("outbound circuit breaker is open")]
CircuitBreakerOpen,
#[error("SSRF policy blocked address: {0}")]
SsrfBlocked(String),
#[error("too many redirects (max {0})")]
TooManyRedirects(usize),
#[error("redirect rejected: {0}")]
RedirectRejected(String),
#[error("invalid or unresolvable URL: {0}")]
InvalidUrl(String),
#[error("{0}")]
IncompatiblePinRedirect(&'static str),
#[error("{0}")]
PinRequiresDomainHost(&'static str),
#[error("{0}")]
PinNotAllowedWithSsrfSafe(&'static str),
}
#[derive(Debug)]
pub struct Response {
status: reqwest::StatusCode,
headers: HeaderMap,
body: Bytes,
url: Option<reqwest::Url>,
}
impl Response {
pub const fn status(&self) -> reqwest::StatusCode {
self.status
}
pub const fn headers(&self) -> &HeaderMap {
&self.headers
}
pub fn is_success(&self) -> bool {
self.status.is_success()
}
pub const fn url(&self) -> Option<&reqwest::Url> {
self.url.as_ref()
}
pub fn json<T: DeserializeOwned>(self) -> Result<T, ClientError> {
serde_json::from_slice(&self.body).map_err(ClientError::Json)
}
pub fn text(self) -> String {
String::from_utf8_lossy(&self.body).into_owned()
}
pub fn bytes(self) -> Bytes {
self.body
}
}
#[must_use]
pub fn is_blocked_ip(ip: IpAddr) -> bool {
match ip {
IpAddr::V4(v4) => is_blocked_ipv4(v4),
IpAddr::V6(v6) => is_blocked_ipv6(v6),
}
}
#[must_use]
pub fn is_public_ip(ip: IpAddr) -> bool {
!is_blocked_ip(ip)
}
fn is_blocked_ipv4(ip: Ipv4Addr) -> bool {
let [a, b, c, _d] = ip.octets();
if a == 0 {
return true;
}
if a == 10 {
return true;
}
if a == 100 && (64..=127).contains(&b) {
return true;
}
if a == 127 {
return true;
}
if a == 169 && b == 254 {
return true;
}
if a == 172 && (16..=31).contains(&b) {
return true;
}
if a == 192 && b == 0 && c == 0 {
return true;
}
if a == 192 && b == 0 && c == 2 {
return true;
}
if a == 192 && b == 88 && c == 99 {
return true;
}
if a == 192 && b == 168 {
return true;
}
if a == 198 && (18..=19).contains(&b) {
return true;
}
if a == 198 && b == 51 && c == 100 {
return true;
}
if a == 203 && b == 0 && c == 113 {
return true;
}
if a >= 224 {
return true;
}
false
}
fn embedded_ipv4(ip: Ipv6Addr) -> Option<Ipv4Addr> {
if let Some(v4) = ip.to_ipv4_mapped() {
return Some(v4);
}
let segs = ip.segments();
if segs[0] == 0
&& segs[1] == 0
&& segs[2] == 0
&& segs[3] == 0
&& segs[4] == 0xffff
&& segs[5] == 0
{
let o = ip.octets();
return Some(Ipv4Addr::new(o[12], o[13], o[14], o[15]));
}
let o = ip.octets();
if o[0] == 0x00
&& o[1] == 0x64
&& o[2] == 0xff
&& o[3] == 0x9b
&& o[4..12].iter().all(|&b| b == 0)
{
return Some(Ipv4Addr::new(o[12], o[13], o[14], o[15]));
}
if o[0] == 0x20 && o[1] == 0x02 {
return Some(Ipv4Addr::new(o[2], o[3], o[4], o[5]));
}
ip.to_ipv4()
}
fn is_blocked_ipv6(ip: Ipv6Addr) -> bool {
if let Some(v4) = embedded_ipv4(ip)
&& is_blocked_ipv4(v4)
{
return true;
}
let segs = ip.segments();
if segs[0] == 0x0064 && segs[1] == 0xff9b && segs[2] == 0x0001 {
return true;
}
if ip == Ipv6Addr::UNSPECIFIED {
return true;
}
if ip == Ipv6Addr::LOCALHOST {
return true;
}
if segs[0] & 0xfe00 == 0xfc00 {
return true;
}
if segs[0] & 0xffc0 == 0xfe80 {
return true;
}
if segs[0] & 0xffc0 == 0xfec0 {
return true;
}
if segs[0] & 0xff00 == 0xff00 {
return true;
}
if segs[0] == 0x2001 && segs[1] == 0x0db8 {
return true;
}
let s = segs;
if s[0] == 0x0100 && s[1] == 0 && s[2] == 0 && s[3] == 0 {
return true;
}
if s[0] == 0x2001 && s[1] == 0x0002 && s[2] == 0x0000 {
return true;
}
if s[0] == 0x2001 && (s[1] & 0xFFF0) == 0x0010 {
return true;
}
if s[0] == 0x2001 && (s[1] & 0xFFF0) == 0x0020 {
return true;
}
if s[0] == 0x3fff && (s[1] & 0xF000) == 0x0000 {
return true;
}
if s[0] == 0x5f00 {
return true;
}
if s[0] == 0x2620 && s[1] == 0x004f && s[2] == 0x8000 {
return true;
}
false
}
#[derive(Clone, Debug)]
pub struct RetryPolicy {
pub max_retries: u32,
pub retry_idempotent_only: bool,
pub max_retry_after: Duration,
pub request_timeout: Option<Duration>,
}
impl Default for RetryPolicy {
fn default() -> Self {
Self {
max_retries: 3,
retry_idempotent_only: true,
max_retry_after: Duration::from_secs(10),
request_timeout: Some(Duration::from_secs(30)),
}
}
}
pub(crate) struct MockEntry {
pub(crate) method: Option<Method>,
pub(crate) path: String,
pub(crate) alias: Option<String>,
pub(crate) status: u16,
pub(crate) body: Option<serde_json::Value>,
pub(crate) call_count: Arc<AtomicUsize>,
}
pub(crate) struct MockResponse {
pub(crate) status: u16,
pub(crate) body: Option<serde_json::Value>,
}
pub struct MockRegistry {
entries: Mutex<Vec<MockEntry>>,
}
impl MockRegistry {
#[must_use]
pub const fn new() -> Self {
Self {
entries: Mutex::new(Vec::new()),
}
}
pub(crate) fn register(&self, entry: MockEntry) {
self.entries
.lock()
.expect("mock registry lock poisoned")
.push(entry);
}
pub(crate) fn find_match(
&self,
method: &Method,
url: &str,
alias: Option<&str>,
) -> Option<MockResponse> {
let url_path_owned: String = reqwest::Url::parse(url).map_or_else(
|_| {
let s = url.split_once('?').map_or(url, |(p, _)| p);
s.split_once('#').map_or(s, |(p, _)| p).to_owned()
},
|parsed| parsed.path().to_owned(),
);
let url_path = url_path_owned.as_str();
let found = {
let entries = self.entries.lock().expect("mock registry lock poisoned");
entries.iter().find_map(|entry| {
let method_ok = entry.method.as_ref().is_none_or(|m| m == method);
let path_ok = url_path == entry.path.as_str()
|| url_path
.strip_suffix(entry.path.as_str())
.is_some_and(|prefix| {
prefix.is_empty()
|| prefix.ends_with('/')
|| entry.path.starts_with('/')
});
let alias_ok = entry
.alias
.as_deref()
.is_none_or(|a| alias.is_some_and(|b| a == b));
if method_ok && path_ok && alias_ok {
Some((entry.call_count.clone(), entry.status, entry.body.clone()))
} else {
None
}
})
};
found.map(|(call_count, status, body)| {
call_count.fetch_add(1, Ordering::SeqCst);
MockResponse { status, body }
})
}
}
impl Default for MockRegistry {
fn default() -> Self {
Self::new()
}
}
pub struct HttpMockRegistryExt(pub Arc<MockRegistry>);
#[derive(Clone)]
pub(crate) struct SharedReqwestClient {
pub(crate) client: reqwest::Client,
pub(crate) timeout_secs: u64,
}
pub struct MockHandle {
alias: String,
method: String,
path: String,
call_count: Arc<AtomicUsize>,
}
impl MockHandle {
pub fn expect_called(&self, expected: usize) {
let actual = self.call_count.load(Ordering::SeqCst);
assert_eq!(
actual, expected,
"http mock for {} {} {} expected {} call(s) but got {}",
self.alias, self.method, self.path, expected, actual,
);
}
#[must_use]
pub fn call_count(&self) -> usize {
self.call_count.load(Ordering::SeqCst)
}
}
pub struct MockSetupBuilder {
pub(crate) registry: Arc<MockRegistry>,
pub(crate) alias: String,
pub(crate) method: Option<Method>,
pub(crate) path: Option<String>,
}
impl MockSetupBuilder {
#[must_use]
pub fn get(mut self, path: &str) -> Self {
self.method = Some(Method::GET);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn post(mut self, path: &str) -> Self {
self.method = Some(Method::POST);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn put(mut self, path: &str) -> Self {
self.method = Some(Method::PUT);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn patch(mut self, path: &str) -> Self {
self.method = Some(Method::PATCH);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn delete(mut self, path: &str) -> Self {
self.method = Some(Method::DELETE);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn head(mut self, path: &str) -> Self {
self.method = Some(Method::HEAD);
self.path = Some(path.to_owned());
self
}
#[must_use]
pub fn respond_with(self, status: u16, body: serde_json::Value) -> MockHandle {
let path = self.path.clone().unwrap_or_default();
let method_str = self
.method
.as_ref()
.map_or_else(|| "*".to_owned(), ToString::to_string);
let call_count = Arc::new(AtomicUsize::new(0));
self.registry.register(MockEntry {
method: self.method,
path: path.clone(),
alias: Some(self.alias.clone()),
status,
body: Some(body),
call_count: call_count.clone(),
});
MockHandle {
alias: self.alias,
method: method_str,
path,
call_count,
}
}
#[must_use]
pub fn respond_with_status(self, status: u16) -> MockHandle {
let path = self.path.clone().unwrap_or_default();
let method_str = self
.method
.as_ref()
.map_or_else(|| "*".to_owned(), ToString::to_string);
let call_count = Arc::new(AtomicUsize::new(0));
self.registry.register(MockEntry {
method: self.method,
path: path.clone(),
alias: Some(self.alias.clone()),
status,
body: None,
call_count: call_count.clone(),
});
MockHandle {
alias: self.alias,
method: method_str,
path,
call_count,
}
}
}
#[derive(Clone)]
pub struct Client {
inner: reqwest::Client,
alias: Option<String>,
base_url: Option<String>,
base_urls: HashMap<String, String>,
retry_policy: RetryPolicy,
mock: Option<Arc<MockRegistry>>,
resilience_config: Option<Arc<crate::config::ResilienceConfig>>,
}
impl Client {
#[must_use]
pub fn new() -> Self {
Self::with_timeout(Duration::from_secs(30))
}
#[must_use]
pub fn with_timeout(timeout: Duration) -> Self {
let inner = reqwest::ClientBuilder::new()
.timeout(timeout)
.build()
.expect("failed to build reqwest client");
Self {
inner,
alias: None,
base_url: None,
base_urls: HashMap::new(),
retry_policy: RetryPolicy {
max_retries: 3,
retry_idempotent_only: true,
max_retry_after: Duration::from_secs(10),
request_timeout: Some(timeout),
},
mock: None,
resilience_config: None,
}
}
pub(crate) fn build_inner(config: &crate::config::HttpClientConfig) -> reqwest::Client {
reqwest::ClientBuilder::new()
.timeout(Duration::from_secs(config.timeout_secs))
.build()
.expect("failed to build reqwest client")
}
fn from_config_with_inner(
inner: reqwest::Client,
config: &crate::config::HttpClientConfig,
) -> Self {
let timeout = Duration::from_secs(config.timeout_secs);
Self {
inner,
alias: None,
base_url: None,
base_urls: config.base_urls.clone(),
retry_policy: RetryPolicy {
max_retries: config.max_retries,
retry_idempotent_only: true,
max_retry_after: Duration::from_secs(config.max_retry_after_secs),
request_timeout: Some(timeout),
},
mock: None,
resilience_config: None,
}
}
fn with_inner(inner: reqwest::Client) -> Self {
Self {
inner,
alias: None,
base_url: None,
base_urls: HashMap::new(),
retry_policy: RetryPolicy::default(),
mock: None,
resilience_config: None,
}
}
#[must_use]
pub fn from_config(config: &crate::config::HttpClientConfig) -> Self {
Self::from_config_with_inner(Self::build_inner(config), config)
}
pub(crate) fn with_mock(mut self, registry: Arc<MockRegistry>) -> Self {
self.mock = Some(registry);
self
}
#[must_use]
pub fn from_state(state: &crate::AppState) -> Self {
let autumn_config = state.extension::<crate::config::AutumnConfig>();
let config = state
.extension::<crate::config::HttpConfig>()
.or_else(|| autumn_config.as_ref().map(|c| Arc::new(c.http.clone())));
let effective_timeout_secs = config.as_ref().map_or_else(
|| crate::config::HttpClientConfig::default().timeout_secs,
|c| c.client.timeout_secs,
);
let shared = state.extension::<SharedReqwestClient>().and_then(|s| {
if s.timeout_secs == effective_timeout_secs {
Some(s.client.clone())
} else {
None
}
});
let mut client = match (config, shared) {
(Some(cfg), Some(inner)) => Self::from_config_with_inner(inner, &cfg.client),
(Some(cfg), None) => Self::from_config(&cfg.client),
(None, Some(inner)) => Self::with_inner(inner),
(None, None) => Self::new(),
};
client.resilience_config = autumn_config.map(|c| Arc::new(c.resilience.clone()));
if let Some(ext) = state.extension::<HttpMockRegistryExt>() {
client = client.with_mock(ext.0.clone());
}
client
}
#[must_use]
pub fn named(&self, alias: &str) -> Self {
let base_url = self
.base_urls
.get(alias)
.cloned()
.or_else(|| self.base_url.clone());
Self {
inner: self.inner.clone(),
alias: Some(alias.to_owned()),
base_url,
base_urls: self.base_urls.clone(),
retry_policy: self.retry_policy.clone(),
mock: self.mock.clone(),
resilience_config: self.resilience_config.clone(),
}
}
#[must_use]
pub fn with_base_url(&self, base_url: impl Into<String>) -> Self {
Self {
inner: self.inner.clone(),
alias: self.alias.clone(),
base_url: Some(base_url.into()),
base_urls: self.base_urls.clone(),
retry_policy: self.retry_policy.clone(),
mock: self.mock.clone(),
resilience_config: self.resilience_config.clone(),
}
}
fn build_request(&self, method: Method, url: impl AsRef<str>) -> RequestBuilder {
let url_str = url.as_ref();
let full_url = if url_str.starts_with("http://") || url_str.starts_with("https://") {
url_str.to_owned()
} else if let Some(base) = &self.base_url {
format!(
"{}/{}",
base.trim_end_matches('/'),
url_str.trim_start_matches('/')
)
} else {
url_str.to_owned()
};
RequestBuilder {
client: self.inner.clone(),
method,
url: full_url,
extra_headers: HeaderMap::new(),
body: None,
retry_policy: self.retry_policy.clone(),
mock: self.mock.clone(),
alias: self.alias.clone(),
pending_error: None,
resilience_config: self.resilience_config.clone(),
redirect_mode: RedirectMode::Default,
pin_addr: None,
ssrf_safe: false,
}
}
#[must_use]
pub fn get(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::GET, url)
}
#[must_use]
pub fn post(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::POST, url)
}
#[must_use]
pub fn put(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::PUT, url)
}
#[must_use]
pub fn patch(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::PATCH, url)
}
#[must_use]
pub fn delete(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::DELETE, url)
}
#[must_use]
pub fn head(&self, url: impl AsRef<str>) -> RequestBuilder {
self.build_request(Method::HEAD, url)
}
#[must_use]
pub fn get_ssrf_safe(&self, url: impl Into<String>) -> RequestBuilder {
let mut builder = self.build_request(Method::GET, url.into());
builder.ssrf_safe = true;
builder
}
}
impl Default for Client {
fn default() -> Self {
Self::new()
}
}
impl axum::extract::FromRequestParts<crate::AppState> for Client {
type Rejection = std::convert::Infallible;
async fn from_request_parts(
_parts: &mut http::request::Parts,
state: &crate::AppState,
) -> Result<Self, std::convert::Infallible> {
Ok(Self::from_state(state))
}
}
type RedirectValidator = Arc<dyn Fn(&str) -> bool + Send + Sync>;
enum RedirectMode {
Default,
None,
Follow {
max: usize,
validator: RedirectValidator,
},
}
const SSRF_SAFE_MAX_REDIRECTS: usize = 5;
pub struct RequestBuilder {
client: reqwest::Client,
method: Method,
url: String,
extra_headers: HeaderMap,
body: Option<Bytes>,
retry_policy: RetryPolicy,
mock: Option<Arc<MockRegistry>>,
alias: Option<String>,
pending_error: Option<ClientError>,
resilience_config: Option<Arc<crate::config::ResilienceConfig>>,
redirect_mode: RedirectMode,
pin_addr: Option<SocketAddr>,
ssrf_safe: bool,
}
impl RequestBuilder {
#[must_use]
pub fn header(mut self, name: impl AsRef<str>, value: impl AsRef<str>) -> Self {
let name_str = name.as_ref();
let value_str = value.as_ref();
match (
HeaderName::from_bytes(name_str.as_bytes()),
HeaderValue::from_str(value_str),
) {
(Ok(n), Ok(v)) => {
self.extra_headers.insert(n, v);
}
(Err(e), _) => {
tracing::warn!(header.name = name_str, error = %e, "invalid header name — header skipped");
}
(_, Err(e)) => {
tracing::warn!(header.name = name_str, error = %e, "invalid header value — header skipped");
}
}
self
}
#[must_use]
pub fn json<T: Serialize>(mut self, body: &T) -> Self {
match serde_json::to_vec(body) {
Ok(bytes) => {
self.body = Some(Bytes::from(bytes));
self = self.header("content-type", "application/json");
}
Err(e) => {
self.pending_error = Some(ClientError::Json(e));
}
}
self
}
#[must_use]
pub fn text_body(mut self, body: impl Into<String>) -> Self {
self.body = Some(Bytes::from(body.into().into_bytes()));
self
}
#[must_use]
pub const fn retries(mut self, max: u32) -> Self {
self.retry_policy.max_retries = max;
self.retry_policy.retry_idempotent_only = false;
self
}
#[must_use]
pub const fn max_retry_after(mut self, max: Duration) -> Self {
self.retry_policy.max_retry_after = max;
self
}
#[must_use]
pub const fn no_retry(mut self) -> Self {
self.retry_policy.max_retries = 0;
self
}
#[must_use]
pub fn no_redirect(mut self) -> Self {
self.redirect_mode = RedirectMode::None;
self
}
#[must_use]
pub fn follow_redirects<F>(mut self, max: usize, validator: F) -> Self
where
F: Fn(&str) -> bool + Send + Sync + 'static,
{
self.redirect_mode = RedirectMode::Follow {
max,
validator: Arc::new(validator),
};
self
}
#[must_use]
pub const fn pin_to(mut self, addr: SocketAddr) -> Self {
self.pin_addr = Some(addr);
self
}
pub async fn send(self) -> Result<Response, ClientError> {
if let Some(err) = self.pending_error {
return Err(err);
}
if self.mock.is_some() {
return self.send_inner(false).await;
}
if self.needs_custom_path() {
return self.send_custom().await;
}
let host = url::Url::parse(&self.url).ok().map_or_else(
|| "unknown".to_owned(),
|u| {
let h = u.host_str().unwrap_or("unknown");
u.port()
.map_or_else(|| h.to_owned(), |port| format!("{h}:{port}"))
},
);
let breaker = self.resilience_config.as_ref().map_or_else(
|| {
crate::circuit_breaker::global_registry().get_or_create(
&host,
crate::circuit_breaker::CircuitBreakerPolicy::default(),
)
},
|rc| {
let policy = crate::circuit_breaker::CircuitBreakerPolicy::from_config(rc, &host);
crate::circuit_breaker::global_registry().get_or_create_with_config(&host, policy)
},
);
if breaker.before_call().is_err() {
return Err(ClientError::CircuitBreakerOpen);
}
let guard = crate::circuit_breaker::CircuitBreakerGuard::new(breaker.clone());
let is_half_open = breaker.state() == crate::circuit_breaker::CircuitState::HalfOpen;
let res = self.send_inner(is_half_open).await;
match &res {
Ok(resp) => {
let success = resp.status().as_u16() < 500;
if success {
guard.success();
} else {
guard.failure();
}
}
Err(_) => {
guard.failure();
}
}
res
}
async fn send_inner(self, suppress_retries: bool) -> Result<Response, ClientError> {
if let Some(ref mock) = self.mock {
match mock.find_match(&self.method, &self.url, self.alias.as_deref()) {
Some(mock_resp) => {
let status = reqwest::StatusCode::from_u16(mock_resp.status)
.unwrap_or(reqwest::StatusCode::OK);
let body_bytes = mock_resp
.body
.as_ref()
.map(|v| serde_json::to_vec(v).unwrap_or_default())
.unwrap_or_default();
tracing::info!(
http.method = %self.method,
http.url = %self.url,
http.status = mock_resp.status,
"[mock] outbound request intercepted"
);
return Ok(Response {
status,
headers: HeaderMap::new(),
body: Bytes::from(body_bytes),
url: None,
});
}
None => {
return Err(ClientError::NoMock(
self.method.to_string(),
self.url.clone(),
));
}
}
}
let start = Instant::now();
let max_attempts = if suppress_retries {
1
} else if is_idempotent_method(&self.method) || !self.retry_policy.retry_idempotent_only {
self.retry_policy.max_retries.saturating_add(1)
} else {
1
};
for attempt in 0..max_attempts {
if attempt > 0 {
let exp = (attempt - 1).min(10);
let delay = Duration::from_millis(100 * (1_u64 << exp));
tokio::time::sleep(delay).await;
}
let mut req = self.client.request(self.method.clone(), &self.url);
req = inject_trace_context(req);
for (name, value) in &self.extra_headers {
req = req.header(name.clone(), value.clone());
}
if let Some(body) = &self.body {
req = req.body(body.clone());
}
match req.send().await {
Ok(resp) => {
let status = resp.status();
let headers = resp.headers().clone();
let url_used = resp.url().clone();
if status.as_u16() == 429 && attempt + 1 < max_attempts {
let mut sleep_delay =
parse_retry_after(&headers).unwrap_or(Duration::from_secs(1));
sleep_delay = sleep_delay.min(self.retry_policy.max_retry_after);
if let Some(req_timeout) = self.retry_policy.request_timeout {
sleep_delay = sleep_delay.min(req_timeout);
}
tokio::time::sleep(sleep_delay).await;
continue;
}
if is_retryable_status(status.as_u16()) && attempt + 1 < max_attempts {
continue;
}
let body = resp
.bytes()
.await
.map_err(|e| ClientError::Request(e.without_url()))?;
let elapsed = start.elapsed();
log_request(
self.method.as_str(),
&url_used,
status.as_u16(),
elapsed,
&self.extra_headers,
);
return Ok(Response {
status,
headers,
body,
url: Some(url_used),
});
}
Err(e) if (e.is_connect() || e.is_timeout()) && attempt + 1 < max_attempts => {}
Err(e) => return Err(ClientError::Request(e.without_url())),
}
}
unreachable!("retry loop exited without returning a result — this is a bug")
}
const fn needs_custom_path(&self) -> bool {
self.ssrf_safe
|| self.pin_addr.is_some()
|| !matches!(self.redirect_mode, RedirectMode::Default)
}
async fn send_custom(self) -> Result<Response, ClientError> {
if self.ssrf_safe && self.pin_addr.is_some() {
return Err(ClientError::PinNotAllowedWithSsrfSafe(
"get_ssrf_safe cannot be combined with pin_to: the SSRF-safe path \
performs its own per-hop resolve/validate/pin and never reads the \
pin_to address, so an explicit pin would be silently ignored. Use \
pin_to alone for a caller-chosen address, or get_ssrf_safe alone \
for guarded automatic per-hop pinning.",
));
}
if self.pin_addr.is_some() && matches!(self.redirect_mode, RedirectMode::Follow { .. }) {
return Err(ClientError::IncompatiblePinRedirect(
"pin_to cannot be combined with follow_redirects: the pin only \
covers the first hop and later redirect hops re-resolve via DNS, \
escaping the pin. Use get_ssrf_safe for pinned, per-hop-revalidated \
redirect following; pin_to alone (which returns the 3xx unfollowed); \
or follow_redirects without pin_to.",
));
}
if self.pin_addr.is_some() && url_host_is_ip_literal(&self.url)? {
return Err(ClientError::PinRequiresDomainHost(
"pin_to cannot be honored for an IP-literal URL host because the \
HTTP stack connects to the literal directly and skips the pinned \
address; put the desired IP directly in the URL, or use a domain host.",
));
}
let timeout = self
.retry_policy
.request_timeout
.unwrap_or_else(|| Duration::from_secs(30));
if self.ssrf_safe {
return self.send_ssrf_safe(timeout).await;
}
let follow = match &self.redirect_mode {
RedirectMode::Follow { max, validator } => Some((*max, validator.clone())),
RedirectMode::None | RedirectMode::Default => None,
};
if let Some((max, validator)) = follow {
return self.follow_loop(max, validator, timeout).await;
}
let policy = reqwest::redirect::Policy::none();
let resolve = self.pin_resolve()?;
let client = build_oneshot_client(resolve, policy, timeout)?;
send_one(
&client,
&self.method,
&self.url,
&self.extra_headers,
self.body.as_ref(),
&self.retry_policy,
)
.await
}
fn pin_resolve(&self) -> Result<Option<(String, Vec<SocketAddr>)>, ClientError> {
match self.pin_addr {
Some(addr) => Ok(Some((host_of(&self.url)?, vec![addr]))),
None => Ok(None),
}
}
async fn follow_loop(
self,
max: usize,
validator: RedirectValidator,
timeout: Duration,
) -> Result<Response, ClientError> {
let original =
url::Url::parse(&self.url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
let mut current = self.url.clone();
let mut method = self.method.clone();
let mut headers = self.extra_headers.clone();
let mut body = self.body.clone();
for hop in 0.. {
let resolve = if hop == 0 {
match self.pin_addr {
Some(addr) => Some((host_of(¤t)?, vec![addr])),
None => None,
}
} else {
None
};
if hop > 0 {
strip_sensitive_headers_if_cross_origin(&mut headers, &original, ¤t)?;
}
let client = build_oneshot_client(resolve, reqwest::redirect::Policy::none(), timeout)?;
let resp = send_one(
&client,
&method,
¤t,
&headers,
body.as_ref(),
&self.retry_policy,
)
.await?;
let Some(next) = redirect_target(&resp, ¤t)? else {
return Ok(resp);
};
if hop >= max {
return Err(ClientError::TooManyRedirects(max));
}
if !validator(&next) {
return Err(ClientError::RedirectRejected(next));
}
rewrite_after_redirect(resp.status(), &mut method, &mut body, &mut headers);
current = next;
}
unreachable!("redirect loop is bounded by `max` and always returns")
}
const fn ssrf_redirect_plan(&self) -> (bool, usize) {
match &self.redirect_mode {
RedirectMode::Default => (true, SSRF_SAFE_MAX_REDIRECTS),
RedirectMode::None => (false, 0),
RedirectMode::Follow { max, .. } => (true, *max),
}
}
async fn send_ssrf_safe(self, timeout: Duration) -> Result<Response, ClientError> {
let (follow, max) = self.ssrf_redirect_plan();
let original =
url::Url::parse(&self.url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
let mut current = self.url.clone();
let mut method = self.method.clone();
let mut headers = self.extra_headers.clone();
let mut body = self.body.clone();
for hop in 0.. {
let addrs = resolve_and_validate(¤t).await?;
let host = host_of(¤t)?;
let client = build_oneshot_client(
Some((host, addrs)),
reqwest::redirect::Policy::none(),
timeout,
)?;
if hop > 0 {
strip_sensitive_headers_if_cross_origin(&mut headers, &original, ¤t)?;
}
let resp = send_one(
&client,
&method,
¤t,
&headers,
body.as_ref(),
&self.retry_policy,
)
.await?;
if !follow {
return Ok(resp);
}
let Some(next) = redirect_target(&resp, ¤t)? else {
return Ok(resp);
};
if hop >= max {
return Err(ClientError::TooManyRedirects(max));
}
if scheme_is_https(¤t)? && !scheme_is_https(&next)? {
return Err(ClientError::RedirectRejected(format!(
"https→http scheme downgrade on redirect to {next}"
)));
}
if let RedirectMode::Follow { validator, .. } = &self.redirect_mode
&& !validator(&next)
{
return Err(ClientError::RedirectRejected(next));
}
rewrite_after_redirect(resp.status(), &mut method, &mut body, &mut headers);
current = next;
}
unreachable!("redirect loop is bounded by the SSRF-safe redirect plan")
}
}
const fn is_idempotent_method(method: &Method) -> bool {
matches!(
*method,
Method::GET | Method::HEAD | Method::PUT | Method::DELETE | Method::OPTIONS | Method::TRACE
)
}
const fn is_retryable_status(status: u16) -> bool {
matches!(status, 502..=504)
}
fn build_oneshot_client(
resolve: Option<(String, Vec<SocketAddr>)>,
policy: reqwest::redirect::Policy,
timeout: Duration,
) -> Result<reqwest::Client, ClientError> {
let mut builder = reqwest::ClientBuilder::new()
.timeout(timeout)
.redirect(policy);
if let Some((host, addrs)) = resolve
&& !addrs.is_empty()
{
builder = builder.no_proxy().resolve_to_addrs(&host, &addrs);
}
builder.build().map_err(ClientError::Request)
}
async fn send_one(
client: &reqwest::Client,
method: &Method,
url: &str,
extra_headers: &HeaderMap,
body: Option<&Bytes>,
retry_policy: &RetryPolicy,
) -> Result<Response, ClientError> {
let start = Instant::now();
let max_attempts = if is_idempotent_method(method) || !retry_policy.retry_idempotent_only {
retry_policy.max_retries.saturating_add(1)
} else {
1
};
for attempt in 0..max_attempts {
if attempt > 0 {
let exp = (attempt - 1).min(10);
let delay = Duration::from_millis(100 * (1_u64 << exp));
tokio::time::sleep(delay).await;
}
let mut req = client.request(method.clone(), url);
req = inject_trace_context(req);
for (name, value) in extra_headers {
req = req.header(name.clone(), value.clone());
}
if let Some(body) = body {
req = req.body(body.clone());
}
match req.send().await {
Ok(resp) => {
let status = resp.status();
let headers = resp.headers().clone();
let url_used = resp.url().clone();
if status.as_u16() == 429 && attempt + 1 < max_attempts {
let mut sleep_delay =
parse_retry_after(&headers).unwrap_or(Duration::from_secs(1));
sleep_delay = sleep_delay.min(retry_policy.max_retry_after);
if let Some(req_timeout) = retry_policy.request_timeout {
sleep_delay = sleep_delay.min(req_timeout);
}
tokio::time::sleep(sleep_delay).await;
continue;
}
if is_retryable_status(status.as_u16()) && attempt + 1 < max_attempts {
continue;
}
let body = resp
.bytes()
.await
.map_err(|e| ClientError::Request(e.without_url()))?;
log_request(
method.as_str(),
&url_used,
status.as_u16(),
start.elapsed(),
extra_headers,
);
return Ok(Response {
status,
headers,
body,
url: Some(url_used),
});
}
Err(e) if (e.is_connect() || e.is_timeout()) && attempt + 1 < max_attempts => {}
Err(e) => return Err(ClientError::Request(e.without_url())),
}
}
unreachable!("retry loop exited without returning a result — this is a bug")
}
fn host_of(url: &str) -> Result<String, ClientError> {
let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
parsed
.host_str()
.map(str::to_owned)
.ok_or_else(|| ClientError::InvalidUrl(format!("URL has no host: {url}")))
}
fn url_host_is_ip_literal(url: &str) -> Result<bool, ClientError> {
let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
match parsed.host() {
Some(url::Host::Ipv4(_) | url::Host::Ipv6(_)) => Ok(true),
Some(url::Host::Domain(_)) => Ok(false),
None => Err(ClientError::InvalidUrl(format!("URL has no host: {url}"))),
}
}
fn scheme_is_https(url: &str) -> Result<bool, ClientError> {
let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
Ok(parsed.scheme().eq_ignore_ascii_case("https"))
}
fn redirect_target(resp: &Response, base: &str) -> Result<Option<String>, ClientError> {
match resp.status() {
reqwest::StatusCode::MOVED_PERMANENTLY
| reqwest::StatusCode::FOUND
| reqwest::StatusCode::SEE_OTHER
| reqwest::StatusCode::TEMPORARY_REDIRECT
| reqwest::StatusCode::PERMANENT_REDIRECT => {}
_ => return Ok(None),
}
let Some(location) = resp
.headers()
.get(reqwest::header::LOCATION)
.and_then(|v| v.to_str().ok())
else {
return Ok(None);
};
let base_url = url::Url::parse(base).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
let joined = base_url
.join(location)
.map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
Ok(Some(joined.to_string()))
}
fn strip_sensitive_headers_if_cross_origin(
headers: &mut HeaderMap,
original: &url::Url,
current: &str,
) -> Result<(), ClientError> {
let current_url =
url::Url::parse(current).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
if current_url.origin() != original.origin() {
headers.remove(reqwest::header::AUTHORIZATION);
headers.remove(reqwest::header::COOKIE);
headers.remove(reqwest::header::PROXY_AUTHORIZATION);
}
Ok(())
}
fn rewrite_after_redirect(
status: reqwest::StatusCode,
method: &mut Method,
body: &mut Option<Bytes>,
headers: &mut HeaderMap,
) {
match status.as_u16() {
303 => {
if *method != Method::HEAD {
*method = Method::GET;
}
*body = None;
strip_payload_headers(headers);
}
301 | 302 if *method == Method::POST => {
*method = Method::GET;
*body = None;
strip_payload_headers(headers);
}
_ => {}
}
}
fn strip_payload_headers(headers: &mut HeaderMap) {
use reqwest::header::{
CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH, CONTENT_TYPE, TRANSFER_ENCODING,
};
headers.remove(CONTENT_TYPE);
headers.remove(CONTENT_LENGTH);
headers.remove(TRANSFER_ENCODING);
headers.remove(CONTENT_ENCODING);
headers.remove(CONTENT_LANGUAGE);
}
fn validate_resolved_addrs(addrs: Vec<SocketAddr>) -> Result<Vec<SocketAddr>, ClientError> {
for addr in &addrs {
if is_blocked_ip(addr.ip()) {
return Err(ClientError::SsrfBlocked(addr.ip().to_string()));
}
}
Ok(addrs)
}
async fn resolve_and_validate(url: &str) -> Result<Vec<SocketAddr>, ClientError> {
let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
let scheme = parsed.scheme();
if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
return Err(ClientError::InvalidUrl(format!(
"unsupported URL scheme `{scheme}` (only http/https are allowed): {url}"
)));
}
let port = parsed.port_or_known_default().ok_or_else(|| {
ClientError::InvalidUrl(format!("URL has no port and unknown scheme: {url}"))
})?;
let host = parsed
.host()
.ok_or_else(|| ClientError::InvalidUrl(format!("URL has no host: {url}")))?;
match host {
url::Host::Ipv4(v4) => validate_resolved_addrs(vec![SocketAddr::new(IpAddr::V4(v4), port)]),
url::Host::Ipv6(v6) => validate_resolved_addrs(vec![SocketAddr::new(IpAddr::V6(v6), port)]),
url::Host::Domain(name) => {
let addrs: Vec<SocketAddr> = tokio::net::lookup_host((name, port))
.await
.map_err(|e| ClientError::InvalidUrl(format!("DNS lookup failed for {name}: {e}")))?
.collect();
if addrs.is_empty() {
return Err(ClientError::InvalidUrl(format!(
"DNS lookup for {name} returned no addresses"
)));
}
validate_resolved_addrs(addrs)
}
}
}
fn parse_retry_after(headers: &HeaderMap) -> Option<Duration> {
let value = headers.get("retry-after")?.to_str().ok()?;
if let Ok(secs) = value.parse::<u64>() {
return Some(Duration::from_secs(secs));
}
let dt = chrono::DateTime::parse_from_rfc2822(value).ok()?;
let now = chrono::Utc::now();
let future = dt.with_timezone(&chrono::Utc);
let secs = u64::try_from((future - now).num_seconds().max(0)).unwrap_or(0);
Some(Duration::from_secs(secs))
}
const REDACTED_HEADERS: &[&str] = &["authorization", "cookie", "set-cookie"];
fn is_sensitive_header(name: &str) -> bool {
REDACTED_HEADERS
.iter()
.any(|h| h.eq_ignore_ascii_case(name))
}
fn log_request(
method: &str,
url: &reqwest::Url,
status: u16,
elapsed: Duration,
headers: &HeaderMap,
) {
let host = url.host_str().unwrap_or("unknown");
let path = url.path();
let sent_headers: Vec<&str> = headers
.keys()
.map(HeaderName::as_str)
.filter(|k| !is_sensitive_header(k))
.collect();
tracing::info!(
http.method = method,
http.host = host,
http.path = path,
http.status = status,
http.elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
http.sent_headers = ?sent_headers,
"outbound request"
);
}
#[allow(clippy::missing_const_for_fn)]
fn inject_trace_context(builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
#[cfg(not(feature = "telemetry-otlp"))]
{
builder
}
#[cfg(feature = "telemetry-otlp")]
{
use std::collections::HashMap;
use tracing_opentelemetry::OpenTelemetrySpanExt as _;
let cx = tracing::Span::current().context();
let mut map = HashMap::<String, String>::new();
opentelemetry::global::get_text_map_propagator(|propagator| {
propagator.inject_context(&cx, &mut TraceHeaderInjector(&mut map));
});
let mut builder = builder;
for (k, v) in map {
if let Ok(value) = HeaderValue::from_str(&v) {
builder = builder.header(k, value);
}
}
builder
}
}
#[cfg(feature = "telemetry-otlp")]
struct TraceHeaderInjector<'a>(&'a mut std::collections::HashMap<String, String>);
#[cfg(feature = "telemetry-otlp")]
impl opentelemetry::propagation::Injector for TraceHeaderInjector<'_> {
fn set(&mut self, key: &str, value: String) {
self.0.insert(key.to_owned(), value);
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::HttpClientConfig;
#[test]
fn client_constructs_with_defaults() {
let client = Client::new();
assert!(client.alias.is_none());
assert!(client.base_url.is_none());
assert_eq!(client.retry_policy.max_retries, 3);
}
#[test]
fn request_builder_fluent_api_compiles() {
let client = Client::new();
let _builder = client
.post("https://example.com/api")
.header("x-api-key", "secret")
.json(&serde_json::json!({"key": "value"}))
.retries(2);
}
#[test]
fn response_accessors_work() {
let payload = serde_json::json!({"id": 42, "name": "Alice"});
let body = serde_json::to_vec(&payload).unwrap();
let resp = Response {
status: reqwest::StatusCode::OK,
headers: HeaderMap::new(),
body: Bytes::from(body),
url: None,
};
assert_eq!(resp.status().as_u16(), 200);
assert!(resp.is_success());
}
#[test]
fn response_json_deserialises() {
#[derive(serde::Deserialize, PartialEq, Debug)]
struct User {
id: i32,
name: String,
}
let payload = serde_json::json!({"id": 1, "name": "Bob"});
let resp = Response {
status: reqwest::StatusCode::OK,
headers: HeaderMap::new(),
body: Bytes::from(serde_json::to_vec(&payload).unwrap()),
url: None,
};
let user: User = resp.json().unwrap();
assert_eq!(user.id, 1);
assert_eq!(user.name, "Bob");
}
#[test]
fn response_text_returns_string() {
let resp = Response {
status: reqwest::StatusCode::OK,
headers: HeaderMap::new(),
body: Bytes::from_static(b"hello world"),
url: None,
};
assert_eq!(resp.text(), "hello world");
}
#[test]
fn response_bytes_returns_raw() {
let resp = Response {
status: reqwest::StatusCode::CREATED,
headers: HeaderMap::new(),
body: Bytes::from_static(b"\x00\x01\x02"),
url: None,
};
assert_eq!(resp.bytes(), Bytes::from_static(b"\x00\x01\x02"));
}
#[test]
fn config_deserialises_from_toml() {
let toml = r#"
[client]
timeout_secs = 60
max_retries = 5
[client.base_urls]
stripe = "https://api.stripe.com"
sendgrid = "https://api.sendgrid.com"
"#;
let http_cfg: crate::config::HttpConfig = toml::from_str(toml).unwrap();
let config = &http_cfg.client;
assert_eq!(config.timeout_secs, 60);
assert_eq!(config.max_retries, 5);
assert_eq!(
config.base_urls.get("stripe").map(String::as_str),
Some("https://api.stripe.com")
);
assert_eq!(
config.base_urls.get("sendgrid").map(String::as_str),
Some("https://api.sendgrid.com")
);
}
#[test]
fn config_has_correct_defaults() {
let config = HttpClientConfig::default();
assert_eq!(config.timeout_secs, 30);
assert_eq!(config.max_retries, 3);
assert!(config.base_urls.is_empty());
}
#[test]
fn idempotent_method_classification() {
assert!(is_idempotent_method(&Method::GET));
assert!(is_idempotent_method(&Method::HEAD));
assert!(is_idempotent_method(&Method::PUT));
assert!(is_idempotent_method(&Method::DELETE));
assert!(is_idempotent_method(&Method::OPTIONS));
assert!(is_idempotent_method(&Method::TRACE));
assert!(!is_idempotent_method(&Method::POST));
assert!(!is_idempotent_method(&Method::PATCH));
}
#[test]
fn retryable_status_classification() {
assert!(is_retryable_status(502));
assert!(is_retryable_status(503));
assert!(is_retryable_status(504));
assert!(!is_retryable_status(200));
assert!(!is_retryable_status(400));
assert!(!is_retryable_status(404));
assert!(!is_retryable_status(500));
assert!(!is_retryable_status(429));
}
#[test]
fn retry_after_header_parsing() {
let mut headers = HeaderMap::new();
headers.insert(
reqwest::header::HeaderName::from_static("retry-after"),
HeaderValue::from_static("5"),
);
assert_eq!(parse_retry_after(&headers), Some(Duration::from_secs(5)));
let empty = HeaderMap::new();
assert_eq!(parse_retry_after(&empty), None);
}
#[test]
fn sensitive_header_detection() {
assert!(is_sensitive_header("authorization"));
assert!(is_sensitive_header("Authorization"));
assert!(is_sensitive_header("AUTHORIZATION"));
assert!(is_sensitive_header("cookie"));
assert!(is_sensitive_header("set-cookie"));
assert!(!is_sensitive_header("content-type"));
assert!(!is_sensitive_header("x-api-key"));
}
#[tokio::test]
async fn mock_registry_captures_calls() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::POST),
path: "/charges".to_owned(),
alias: Some("stripe".to_owned()),
status: 200,
body: Some(serde_json::json!({"id": "ch_123"})),
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry).named("stripe");
let resp = client
.post("https://api.stripe.com/charges")
.json(&serde_json::json!({"amount": 1000}))
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let body: serde_json::Value = resp.json().unwrap();
assert_eq!(body["id"], "ch_123");
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn mock_handle_expect_called_passes() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::GET),
path: "/users/1".to_owned(),
alias: None,
status: 200,
body: Some(serde_json::json!({"name": "Alice"})),
call_count: call_count.clone(),
});
let handle = MockHandle {
alias: "test".to_owned(),
method: "GET".to_owned(),
path: "/users/1".to_owned(),
call_count: call_count.clone(),
};
let client = Client::new().with_mock(registry);
client
.get("https://api.example.com/users/1")
.send()
.await
.unwrap();
handle.expect_called(1);
assert_eq!(handle.call_count(), 1);
}
#[tokio::test]
async fn mock_matches_by_path_suffix() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::POST),
path: "/v1/charges".to_owned(),
alias: None,
status: 201,
body: Some(serde_json::json!({"created": true})),
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry);
let resp = client
.post("https://api.stripe.com/v1/charges")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 201);
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn no_mock_error_when_unmatched() {
let registry = Arc::new(MockRegistry::new());
let client = Client::new().with_mock(registry);
let result = client.post("https://api.example.com/unknown").send().await;
assert!(matches!(result, Err(ClientError::NoMock(_, _))));
}
#[tokio::test]
async fn mock_setup_builder_registers_entry() {
let registry = Arc::new(MockRegistry::new());
let builder = MockSetupBuilder {
registry: registry.clone(),
alias: "myservice".to_owned(),
method: None,
path: None,
};
let handle = builder
.post("/api/resource")
.respond_with(201, serde_json::json!({"ok": true}));
let client = Client::new().with_mock(registry).named("myservice");
client
.post("https://myservice.example.com/api/resource")
.send()
.await
.unwrap();
handle.expect_called(1);
}
#[test]
fn client_from_config() {
let config = HttpClientConfig {
timeout_secs: 10,
max_retries: 1,
max_retry_after_secs: 10,
base_urls: std::collections::HashMap::new(),
};
let client = Client::from_config(&config);
assert_eq!(client.retry_policy.max_retries, 1);
}
#[test]
fn named_client_preserves_mock_registry() {
let registry = Arc::new(MockRegistry::new());
let client = Client::new().with_mock(registry);
let named = client.named("stripe");
assert!(named.mock.is_some());
assert_eq!(named.alias.as_deref(), Some("stripe"));
}
#[test]
fn base_url_prepended_to_relative_path() {
let client = Client::new();
let client = client.with_base_url("https://api.stripe.com");
let builder = client.post("/v1/charges");
assert_eq!(builder.url, "https://api.stripe.com/v1/charges");
}
#[test]
fn absolute_url_bypasses_base_url() {
let client = Client::new().with_base_url("https://ignored.example.com");
let builder = client.get("https://actual.example.com/path");
assert_eq!(builder.url, "https://actual.example.com/path");
}
#[test]
fn retry_override_per_request() {
let client = Client::new(); let builder = client.get("https://example.com").retries(0);
assert_eq!(builder.retry_policy.max_retries, 0);
let no_retry = client.get("https://example.com").no_retry();
assert_eq!(no_retry.retry_policy.max_retries, 0);
}
#[tokio::test]
async fn client_extracts_from_state() {
use axum::extract::FromRequestParts;
let state = crate::AppState::for_test();
let mut parts = axum::http::Request::new(axum::body::Body::empty())
.into_parts()
.0;
let client = Client::from_request_parts(&mut parts, &state)
.await
.unwrap();
assert!(client.mock.is_none());
assert!(client.alias.is_none());
}
#[test]
fn mock_registry_ext_round_trips_through_state() {
let registry = Arc::new(MockRegistry::new());
let ext = HttpMockRegistryExt(registry);
let state = crate::AppState::for_test();
state.insert_extension(ext);
let retrieved = state.extension::<HttpMockRegistryExt>();
assert!(retrieved.is_some());
}
#[test]
fn named_client_resolves_base_url_from_config() {
let mut base_urls = std::collections::HashMap::new();
base_urls.insert("stripe".to_owned(), "https://api.stripe.com".to_owned());
let config = HttpClientConfig {
timeout_secs: 30,
max_retries: 3,
max_retry_after_secs: 10,
base_urls,
};
let client = Client::from_config(&config);
let stripe = client.named("stripe");
assert_eq!(stripe.base_url.as_deref(), Some("https://api.stripe.com"));
assert_eq!(stripe.alias.as_deref(), Some("stripe"));
let other = client.named("sendgrid");
assert!(other.base_url.is_none());
}
#[tokio::test]
async fn client_extracts_from_autumn_config_in_state() {
use axum::extract::FromRequestParts;
let mut cfg = crate::config::AutumnConfig::default();
cfg.http.client.max_retries = 7;
let state = crate::AppState::for_test();
state.insert_extension(cfg);
let mut parts = axum::http::Request::new(axum::body::Body::empty())
.into_parts()
.0;
let client = Client::from_request_parts(&mut parts, &state)
.await
.unwrap();
assert_eq!(client.retry_policy.max_retries, 7);
}
#[tokio::test]
async fn respond_with_status_produces_empty_body() {
let registry = Arc::new(MockRegistry::new());
let builder = MockSetupBuilder {
registry: registry.clone(),
alias: "svc".to_owned(),
method: None,
path: None,
};
let _handle = builder.delete("/items/1").respond_with_status(204);
let client = Client::new().with_mock(registry).named("svc");
let resp = client
.delete("https://svc.example.com/items/1")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 204);
assert_eq!(
resp.bytes(),
bytes::Bytes::new(),
"body must be empty, not \"null\""
);
}
#[test]
fn retry_after_http_date_parsing() {
let mut headers = HeaderMap::new();
headers.insert(
reqwest::header::HeaderName::from_static("retry-after"),
HeaderValue::from_static("Tue, 01 Jan 2030 00:00:00 GMT"),
);
let duration = parse_retry_after(&headers);
assert!(duration.is_some(), "should parse HTTP-date Retry-After");
assert!(
duration.unwrap().as_secs() > 0,
"future date should yield positive delay"
);
}
#[tokio::test]
async fn non_idempotent_post_no_retry() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::POST),
path: "/endpoint".to_owned(),
alias: None,
status: 503,
body: None,
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry);
let resp = client
.post("https://example.com/endpoint")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 503);
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn mock_strips_query_from_url_before_matching() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::GET),
path: "/v1/charges".to_owned(),
alias: None,
status: 200,
body: Some(serde_json::json!({"ok": true})),
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry);
let resp = client
.get("https://api.stripe.com/v1/charges?expand[]=balance_transaction")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn mock_suffix_match_with_leading_slash_path() {
let registry = Arc::new(MockRegistry::new());
let call_count = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(Method::POST),
path: "/charges".to_owned(),
alias: None,
status: 201,
body: Some(serde_json::json!({"matched": true})),
call_count: call_count.clone(),
});
let client = Client::new().with_mock(registry);
let resp = client
.post("https://api.stripe.com/v1/charges")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 201);
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[test]
fn retries_clears_idempotent_only_flag() {
let client = Client::new();
let builder = client.post("https://example.com").retries(2);
assert_eq!(builder.retry_policy.max_retries, 2);
assert!(
!builder.retry_policy.retry_idempotent_only,
"explicit retries() call must allow non-idempotent methods to retry"
);
}
#[test]
fn log_request_completes_with_sensitive_headers() {
let url = reqwest::Url::parse("https://api.example.com/v1/resource?q=1").unwrap();
let mut headers = HeaderMap::new();
headers.insert(
HeaderName::from_static("content-type"),
HeaderValue::from_static("application/json"),
);
headers.insert(
HeaderName::from_static("authorization"),
HeaderValue::from_static("Bearer sk_test_xxx"),
);
log_request("POST", &url, 201, Duration::from_millis(12), &headers);
}
#[test]
fn inject_trace_context_passthrough_without_telemetry() {
let inner = reqwest::Client::new();
let builder = inner.get("https://example.com");
let _b = inject_trace_context(builder);
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn real_get_request_covers_network_path() {
use axum::{Router, routing::get};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let app = Router::new().route("/ping", get(|| async { "pong" }));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = Client::new();
let resp = client
.get(format!("http://127.0.0.1:{}/ping", addr.port()))
.header("x-request-id", "test-35")
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
assert!(resp.url().is_some());
assert_eq!(resp.text(), "pong");
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn real_post_with_json_body_covers_body_path() {
use axum::{Json, Router, routing::post};
use serde_json::Value;
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let app = Router::new().route(
"/echo",
post(|Json(body): Json<Value>| async move { Json(body) }),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = Client::new();
let resp = client
.post(format!("http://127.0.0.1:{}/echo", addr.port()))
.json(&serde_json::json!({"hello": "world"}))
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let body: Value = resp.json().unwrap();
assert_eq!(body["hello"], "world");
crate::circuit_breaker::global_registry().clear();
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn real_get_retries_on_503_then_succeeds() {
use axum::{Router, routing::get};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering as SeqOrdering};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let hit = Arc::new(AtomicU32::new(0));
let hit2 = hit.clone();
let app = Router::new().route(
"/flaky",
get(move || {
let c = hit2.clone();
async move {
if c.fetch_add(1, SeqOrdering::SeqCst) == 0 {
axum::http::StatusCode::SERVICE_UNAVAILABLE
} else {
axum::http::StatusCode::OK
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/flaky", addr.port()))
.retries(1)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
assert_eq!(hit.load(SeqOrdering::SeqCst), 2);
crate::circuit_breaker::global_registry().clear();
}
#[test]
fn text_body_sets_body() {
let client = Client::new();
let builder = client.post("https://example.com").text_body("hello");
assert_eq!(builder.body, Some(bytes::Bytes::from_static(b"hello")));
}
#[test]
fn client_error_display() {
let err = ClientError::NoMock("GET".to_owned(), "/path".to_owned());
assert!(err.to_string().contains("GET"));
assert!(err.to_string().contains("/path"));
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn test_http_client_circuit_breaker_integration() {
use axum::{Router, routing::get};
use std::sync::atomic::{AtomicU32, Ordering as SeqOrdering};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let hit = Arc::new(AtomicU32::new(0));
let hit2 = hit.clone();
let app = Router::new().route(
"/flaky",
get(move || {
let c = hit2.clone();
async move {
c.fetch_add(1, SeqOrdering::SeqCst);
axum::http::StatusCode::INTERNAL_SERVER_ERROR
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let mut rc = crate::config::ResilienceConfig::default();
rc.circuit_breaker.defaults.failure_ratio_threshold = Some(0.5);
rc.circuit_breaker.defaults.minimum_sample_count = Some(3);
rc.circuit_breaker.defaults.open_duration_secs = Some(10);
let client = Client::new();
let client = Client {
resilience_config: Some(Arc::new(rc)),
..client
};
let url = format!("http://127.0.0.1:{}/flaky", addr.port());
for _ in 0..3 {
let res = client.get(&url).send().await;
let res = res.unwrap();
assert_eq!(res.status().as_u16(), 500);
}
let res = client.get(&url).send().await;
assert!(matches!(res, Err(ClientError::CircuitBreakerOpen)));
assert_eq!(hit.load(SeqOrdering::SeqCst), 3);
crate::circuit_breaker::global_registry().clear();
}
#[test]
fn shared_reqwest_client_ext_round_trips() {
let ext = SharedReqwestClient {
client: reqwest::Client::new(),
timeout_secs: 30,
};
let state = crate::AppState::for_test();
state.insert_extension(ext);
let retrieved = state.extension::<SharedReqwestClient>();
assert!(retrieved.is_some());
}
#[test]
fn client_head_method_builds_request_builder() {
let client = Client::new();
let _builder = client.head("https://example.com/resource");
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn from_state_reuses_shared_client() {
use axum::{Router, routing::get};
let _lock = crate::circuit_breaker::TEST_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
crate::circuit_breaker::global_registry().clear();
let app = Router::new().route(
"/ua",
get(|req: axum::http::Request<axum::body::Body>| async move {
req.headers()
.get("user-agent")
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_owned()
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let distinctive_inner = reqwest::ClientBuilder::new()
.user_agent("autumn-shared-pool-test")
.build()
.expect("failed to build inner client");
let state = crate::AppState::for_test();
state.insert_extension(SharedReqwestClient {
client: distinctive_inner,
timeout_secs: 30,
});
let client = Client::from_state(&state);
let resp = client
.get(format!("http://127.0.0.1:{}/ua", addr.port()))
.send()
.await
.expect("request should succeed");
assert_eq!(resp.text(), "autumn-shared-pool-test");
crate::circuit_breaker::global_registry().clear();
}
#[cfg(feature = "http-client")]
#[test]
fn from_state_falls_back_when_timeout_mismatches_shared_client() {
use crate::config::{AutumnConfig, HttpClientConfig};
use std::sync::Arc;
let mut config = AutumnConfig::default();
config.http.client = HttpClientConfig {
timeout_secs: 10,
..Default::default()
};
let state = crate::AppState::for_test();
state.insert_extension(SharedReqwestClient {
client: reqwest::Client::new(),
timeout_secs: 5, });
state.insert_extension(Arc::new(config));
let _client = Client::from_state(&state);
}
#[cfg(feature = "http-client")]
#[test]
fn from_state_reuses_shared_client_when_no_config() {
let state = crate::AppState::for_test();
let default_timeout = crate::config::HttpClientConfig::default().timeout_secs;
state.insert_extension(SharedReqwestClient {
client: reqwest::Client::new(),
timeout_secs: default_timeout,
});
let _client = Client::from_state(&state);
}
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
#[test]
fn ssrf_policy_blocks_private_and_reserved_ipv4() {
let blocked = [
"0.0.0.0",
"10.1.2.3",
"100.64.0.1", "127.0.0.1", "169.254.169.254", "172.16.5.4", "192.0.0.1", "192.0.2.5", "192.88.99.1", "192.168.1.1", "198.18.0.1", "198.51.100.7", "203.0.113.9", "224.0.0.1", "240.0.0.1", "255.255.255.255", ];
for s in blocked {
let ip: IpAddr = s.parse().unwrap();
assert!(is_blocked_ip(ip), "{s} should be blocked");
assert!(!is_public_ip(ip), "{s} should not be public");
}
}
#[test]
fn ssrf_policy_allows_public_ipv4() {
for s in ["1.1.1.1", "8.8.8.8", "93.184.216.34"] {
let ip: IpAddr = s.parse().unwrap();
assert!(is_public_ip(ip), "{s} should be public");
assert!(!is_blocked_ip(ip), "{s} should not be blocked");
}
}
#[test]
fn ssrf_policy_ipv6_and_mapped_forms() {
let blocked = [
"::", "::1", "fe80::1", "fc00::1", "ff02::1", "2001:db8::1", "fec0::1", "::ffff:169.254.169.254", "::ffff:127.0.0.1", "100::1", "100::dead:beef", "2001:2::1", "2001:10::1", "2001:20::1", "2001:20:abcd::1", "2001:2f::1", "3fff::1", "3fff:0fff::1", "5f00::1", "5f00:1234::1", "2620:4f:8000::1", ];
for s in blocked {
let ip: IpAddr = s.parse().unwrap();
assert!(is_blocked_ip(ip), "{s} should be blocked");
}
let compat = IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0x7f00, 0x0001));
assert!(
is_blocked_ip(compat),
"::7f00:1 (127.0.0.1) should be blocked"
);
let public: IpAddr = "2606:4700:4700::1111".parse().unwrap();
assert!(
is_public_ip(public),
"2606:4700:4700::1111 should be public"
);
let public_addrs = [
"2001:4860:4860::8888", "2606:4700:4700::1111", "2400:cb00:2048::1", "2620:0:2d0:200::7", "2001:2:1::1", "3fff:abcd::1", "4000::1", "5e00::1", "6000::1", ];
for s in public_addrs {
let ip: IpAddr = s.parse().unwrap();
assert!(is_public_ip(ip), "{s} should be public");
assert!(!is_blocked_ip(ip), "{s} should not be blocked");
}
}
#[test]
fn ssrf_policy_blocks_tunnelled_ipv4() {
let blocked = [
"64:ff9b::a9fe:a9fe", "64:ff9b::7f00:1", "64:ff9b:1::a9fe:a9fe", "64:ff9b:1::7f00:1", "64:ff9b:1::808:808", "2002:a9fe:a9fe::", "2002:7f00:1::", "2002:0a00:0001::", "::ffff:0:169.254.169.254", "::ffff:0:127.0.0.1", ];
for s in blocked {
let ip: IpAddr = s.parse().unwrap();
assert!(is_blocked_ip(ip), "{s} should be blocked");
assert!(!is_public_ip(ip), "{s} should not be public");
}
let anycast: IpAddr = "192.88.99.1".parse().unwrap();
assert!(is_blocked_ip(anycast), "192.88.99.1 should be blocked");
for s in ["2002:0808:0808::", "2606:4700::1111", "::ffff:0:8.8.8.8"] {
let ip: IpAddr = s.parse().unwrap();
assert!(is_public_ip(ip), "{s} should be public");
assert!(!is_blocked_ip(ip), "{s} should not be blocked");
}
}
#[test]
fn ssrf_policy_decimal_encoded_host_is_blocked() {
for raw in [
"http://2130706433/",
"http://0x7f000001/",
"http://127.0.0.1/",
] {
let parsed = url::Url::parse(raw).unwrap();
match parsed.host() {
Some(url::Host::Ipv4(v4)) => {
assert_eq!(
v4,
Ipv4Addr::LOCALHOST,
"{raw} should normalise to 127.0.0.1"
);
assert!(
is_blocked_ip(IpAddr::V4(v4)),
"{raw} host should be blocked"
);
}
other => panic!("{raw} did not parse to an Ipv4 host: {other:?}"),
}
}
}
async fn spawn(app: axum::Router) -> std::net::SocketAddr {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
addr
}
fn redirect_302(location: String) -> axum::response::Response {
axum::response::Response::builder()
.status(302)
.header("location", location)
.body(axum::body::Body::empty())
.unwrap()
}
fn redirect_307(location: String) -> axum::response::Response {
axum::response::Response::builder()
.status(307)
.header("location", location)
.body(axum::body::Body::empty())
.unwrap()
}
fn response_with_location(status: u16, location: String) -> axum::response::Response {
axum::response::Response::builder()
.status(status)
.header("location", location)
.body(axum::body::Body::empty())
.unwrap()
}
#[tokio::test]
async fn no_redirect_returns_3xx_unfollowed() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route(
"/start",
get(|| async { redirect_302("http://127.0.0.1:1/never".to_owned()) }),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", addr.port()))
.no_redirect()
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 302);
assert_eq!(
resp.headers().get("location").and_then(|v| v.to_str().ok()),
Some("http://127.0.0.1:1/never")
);
}
#[tokio::test]
async fn no_redirect_returns_3xx_with_malformed_location() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route(
"/start",
get(|| async { redirect_302("ht!tp://\\bad".to_owned()) }),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", addr.port()))
.no_redirect()
.send()
.await
.expect("no_redirect() must return the 3xx even with a malformed Location");
assert_eq!(resp.status().as_u16(), 302);
assert_eq!(
resp.headers().get("location").and_then(|v| v.to_str().ok()),
Some("ht!tp://\\bad")
);
}
#[tokio::test]
async fn follow_redirects_valid_chain_calls_validator() {
use axum::{Router, routing::get};
let b_addr = spawn(Router::new().route("/final", get(|| async { "final-body" }))).await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/start",
get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/final")) }),
))
.await;
let calls = Arc::new(AtomicUsize::new(0));
let calls2 = calls.clone();
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", a_addr.port()))
.follow_redirects(5, move |_loc| {
calls2.fetch_add(1, Ordering::SeqCst);
true
})
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
assert_eq!(resp.text(), "final-body");
assert_eq!(
calls.load(Ordering::SeqCst),
1,
"validator called once per hop"
);
}
#[tokio::test]
async fn follow_redirects_rejects_private_target() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let priv_addr = spawn(Router::new().route(
"/secret",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"SECRET"
}
}),
))
.await;
let priv_port = priv_addr.port();
let a_addr = spawn(Router::new().route(
"/start",
get(
move || async move { redirect_302(format!("http://127.0.0.1:{priv_port}/secret")) },
),
))
.await;
let validator = |u: &str| -> bool {
let Ok(p) = url::Url::parse(u) else {
return false;
};
match p.host() {
Some(url::Host::Ipv4(v4)) => is_public_ip(IpAddr::V4(v4)),
Some(url::Host::Ipv6(v6)) => is_public_ip(IpAddr::V6(v6)),
_ => true,
}
};
let result = Client::new()
.get(format!("http://127.0.0.1:{}/start", a_addr.port()))
.follow_redirects(5, validator)
.send()
.await;
assert!(
matches!(result, Err(ClientError::RedirectRejected(_))),
"expected RedirectRejected, got {result:?}"
);
assert!(
!touched.load(Ordering::SeqCst),
"the private target must never be connected to"
);
}
#[tokio::test]
async fn follow_redirects_cap_exceeded() {
use axum::{Router, routing::get};
let addr =
spawn(Router::new().route("/loop", get(|| async { redirect_302("/loop".to_owned()) })))
.await;
let result = Client::new()
.get(format!("http://127.0.0.1:{}/loop", addr.port()))
.follow_redirects(2, |_| true)
.send()
.await;
assert!(
matches!(result, Err(ClientError::TooManyRedirects(2))),
"expected TooManyRedirects(2), got {result:?}"
);
}
#[tokio::test]
async fn follow_redirects_zero_max_errors_on_first_3xx() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route(
"/start",
get(|| async { redirect_302("http://127.0.0.1:1/x".to_owned()) }),
))
.await;
let result = Client::new()
.get(format!("http://127.0.0.1:{}/start", addr.port()))
.follow_redirects(0, |_| true)
.send()
.await;
assert!(matches!(result, Err(ClientError::TooManyRedirects(0))));
}
#[tokio::test]
async fn pin_to_bypasses_dns_and_uses_url_port() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route("/ping", get(|| async { "pong" }))).await;
let listener_port = addr.port();
let resp = Client::new()
.get(format!("http://pinned.invalid:{listener_port}/ping"))
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1))
.send()
.await
.expect("pinned request should reach the loopback listener");
assert_eq!(resp.status().as_u16(), 200);
assert_eq!(resp.text(), "pong");
}
#[tokio::test]
async fn get_ssrf_safe_rejects_loopback_host_before_connecting() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let addr = spawn(Router::new().route(
"/x",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"reached"
}
}),
))
.await;
let result = Client::new()
.get_ssrf_safe(format!("http://localhost:{}/x", addr.port()))
.send()
.await;
assert!(
matches!(result, Err(ClientError::SsrfBlocked(_))),
"expected SsrfBlocked, got {result:?}"
);
assert!(
!touched.load(Ordering::SeqCst),
"SSRF guard must reject before any connection"
);
}
#[tokio::test]
async fn get_ssrf_safe_rejects_decimal_encoded_loopback() {
let result = Client::new()
.get_ssrf_safe("http://2130706433/")
.send()
.await;
assert!(
matches!(result, Err(ClientError::SsrfBlocked(_))),
"expected SsrfBlocked, got {result:?}"
);
}
#[tokio::test]
async fn resolve_and_validate_accepts_public_rejects_blocked() {
let ok = resolve_and_validate("http://8.8.8.8:8080/path")
.await
.unwrap();
assert_eq!(
ok,
vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)), 8080)]
);
let ok_https = resolve_and_validate("https://1.1.1.1/").await.unwrap();
assert_eq!(ok_https.len(), 1);
assert_eq!(ok_https[0].port(), 443);
for raw in [
"http://127.0.0.1/",
"http://169.254.169.254/latest/meta-data/",
"http://10.0.0.1/",
"http://2130706433/",
] {
let err = resolve_and_validate(raw).await;
assert!(
matches!(err, Err(ClientError::SsrfBlocked(_))),
"{raw} should be SsrfBlocked, got {err:?}"
);
}
}
#[test]
fn validate_resolved_addrs_returns_all_public_rejects_any_blocked() {
let v4 = |a, b, c, d, p| SocketAddr::new(IpAddr::V4(Ipv4Addr::new(a, b, c, d)), p);
let two = vec![v4(1, 1, 1, 1, 443), v4(8, 8, 8, 8, 443)];
assert_eq!(validate_resolved_addrs(two.clone()).unwrap(), two);
let mixed = vec![v4(1, 1, 1, 1, 443), v4(10, 0, 0, 1, 443)];
assert!(
matches!(
validate_resolved_addrs(mixed),
Err(ClientError::SsrfBlocked(_))
),
"a set containing a blocked address must be rejected"
);
let v6 = SocketAddr::new(
IpAddr::V6(Ipv6Addr::new(0x2606, 0x4700, 0x4700, 0, 0, 0, 0, 0x1111)),
443,
);
let mix = vec![v6, v4(8, 8, 8, 8, 443)];
assert_eq!(validate_resolved_addrs(mix.clone()).unwrap(), mix);
}
#[tokio::test]
async fn pin_to_does_not_follow_redirect_unpinned() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let onward = spawn(Router::new().route(
"/onward",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"REACHED"
}
}),
))
.await;
let onward_port = onward.port();
let start =
spawn(Router::new().route(
"/start",
get(move || async move {
redirect_302(format!("http://127.0.0.1:{onward_port}/onward"))
}),
))
.await;
let start_port = start.port();
let resp = Client::new()
.get(format!("http://pinned.invalid:{start_port}/start"))
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), start_port))
.send()
.await
.expect("pinned request should return the 302 unfollowed");
assert_eq!(
resp.status().as_u16(),
302,
"pin_to must return the redirect unfollowed"
);
assert!(
!touched.load(Ordering::SeqCst),
"pin_to must not silently follow the redirect onward"
);
}
#[tokio::test]
async fn get_ssrf_safe_rejects_non_http_scheme() {
for raw in ["ftp://public.example/resource", "gopher://public.example/"] {
let result = Client::new().get_ssrf_safe(raw).send().await;
assert!(
matches!(result, Err(ClientError::InvalidUrl(_))),
"{raw} should be rejected with InvalidUrl, got {result:?}"
);
}
}
#[tokio::test]
async fn follow_redirects_strips_sensitive_headers_cross_origin() {
use axum::{Router, routing::get};
let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let b_addr = spawn(Router::new().route(
"/dst",
get(move |headers: HeaderMap| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some(headers);
"ok"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/",
get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/", a_addr.port()))
.header("authorization", "secret")
.header("cookie", "session=abc")
.header("proxy-authorization", "Basic zzz")
.follow_redirects(3, |_| true)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let headers = seen
.lock()
.unwrap()
.clone()
.expect("listener B must have been reached");
assert!(
headers.get("authorization").is_none(),
"authorization must be stripped on a cross-origin redirect"
);
assert!(
headers.get("cookie").is_none(),
"cookie must be stripped on a cross-origin redirect"
);
assert!(
headers.get("proxy-authorization").is_none(),
"proxy-authorization must be stripped on a cross-origin redirect"
);
}
#[tokio::test]
async fn follow_redirects_keeps_sensitive_headers_same_origin() {
use axum::{Router, routing::get};
let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let addr = spawn(
Router::new()
.route("/", get(|| async { redirect_302("/next".to_owned()) }))
.route(
"/next",
get(move |headers: HeaderMap| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some(headers);
"ok"
}
}),
),
)
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/", addr.port()))
.header("authorization", "secret")
.follow_redirects(3, |_| true)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let headers = seen
.lock()
.unwrap()
.clone()
.expect("/next must have been reached");
assert_eq!(
headers.get("authorization").and_then(|v| v.to_str().ok()),
Some("secret"),
"authorization must be preserved on a same-origin redirect"
);
}
#[tokio::test]
async fn follow_redirects_302_post_becomes_get() {
use axum::{
Router,
routing::{any, post},
};
let seen: Arc<Mutex<Option<(String, HeaderMap, Bytes)>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let b_addr = spawn(Router::new().route(
"/dst",
any(move |method: Method, headers: HeaderMap, body: Bytes| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some((method.to_string(), headers, body));
"ok"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/",
post(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
))
.await;
let resp = Client::new()
.post(format!("http://127.0.0.1:{}/", a_addr.port()))
.json(&serde_json::json!({"payload": true}))
.follow_redirects(3, |_| true)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let (method, headers, body) = seen
.lock()
.unwrap()
.clone()
.expect("listener B must have been reached");
assert_eq!(method, "GET", "302 must rewrite POST → GET");
assert!(body.is_empty(), "302 POST→GET must drop the request body");
assert!(
!headers.contains_key(reqwest::header::CONTENT_TYPE),
"302 POST→GET must drop the Content-Type payload header"
);
assert!(
!headers.contains_key(reqwest::header::CONTENT_LENGTH),
"302 POST→GET must drop the Content-Length payload header"
);
}
#[tokio::test]
async fn follow_redirects_307_preserves_method_and_body() {
use axum::{
Router,
routing::{any, post},
};
let seen: Arc<Mutex<Option<(String, Bytes)>>> = Arc::new(Mutex::new(None));
let seen2 = seen.clone();
let b_addr = spawn(Router::new().route(
"/dst",
any(move |method: Method, body: Bytes| {
let slot = seen2.clone();
async move {
*slot.lock().unwrap() = Some((method.to_string(), body));
"ok"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/",
post(move || async move { redirect_307(format!("http://127.0.0.1:{b_port}/dst")) }),
))
.await;
let resp = Client::new()
.post(format!("http://127.0.0.1:{}/", a_addr.port()))
.text_body("payload")
.follow_redirects(3, |_| true)
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200);
let (method, body) = seen
.lock()
.unwrap()
.clone()
.expect("listener B must have been reached");
assert_eq!(method, "POST", "307 must preserve the POST method");
assert_eq!(
&body[..],
b"payload",
"307 must preserve the request body verbatim"
);
}
#[test]
fn ssrf_redirect_plan_honours_chained_override() {
let client = Client::new();
let default = client.get_ssrf_safe("https://example.com/");
assert_eq!(
default.ssrf_redirect_plan(),
(true, SSRF_SAFE_MAX_REDIRECTS),
"default SSRF-safe path follows up to SSRF_SAFE_MAX_REDIRECTS"
);
let none = client.get_ssrf_safe("https://example.com/").no_redirect();
let (follow, _max) = none.ssrf_redirect_plan();
assert!(
!follow,
"no_redirect() must disable following on the safe path"
);
let follow3 = client
.get_ssrf_safe("https://example.com/")
.follow_redirects(3, |_| true);
assert_eq!(
follow3.ssrf_redirect_plan(),
(true, 3),
"follow_redirects(3, ..) must cap the safe path at 3 hops"
);
}
#[tokio::test]
async fn pin_then_follow_redirects_is_rejected() {
let result = Client::new()
.get("http://example.com/")
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.follow_redirects(2, |_| true)
.send()
.await;
assert!(
matches!(result, Err(ClientError::IncompatiblePinRedirect(_))),
"expected IncompatiblePinRedirect, got {result:?}"
);
}
#[tokio::test]
async fn follow_redirects_then_pin_is_rejected() {
let result = Client::new()
.get("http://example.com/")
.follow_redirects(2, |_| true)
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.send()
.await;
assert!(
matches!(result, Err(ClientError::IncompatiblePinRedirect(_))),
"expected IncompatiblePinRedirect, got {result:?}"
);
}
#[tokio::test]
async fn pin_with_no_redirect_is_allowed() {
use axum::{Router, routing::get};
let addr = spawn(Router::new().route(
"/start",
get(|| async { redirect_302("http://127.0.0.1:1/onward".to_owned()) }),
))
.await;
let port = addr.port();
let resp = Client::new()
.get(format!("http://pinned.invalid:{port}/start"))
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port))
.no_redirect()
.send()
.await
.expect("pin_to + no_redirect must return the 3xx unfollowed, not error");
assert_eq!(
resp.status().as_u16(),
302,
"pin_to + no_redirect returns the redirect verbatim"
);
}
#[tokio::test]
async fn pin_to_ipv4_literal_host_is_rejected() {
let result = Client::new()
.get("http://198.51.100.1:8080/")
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.send()
.await;
assert!(
matches!(result, Err(ClientError::PinRequiresDomainHost(_))),
"expected PinRequiresDomainHost, got {result:?}"
);
}
#[tokio::test]
async fn pin_to_ipv6_literal_host_is_rejected() {
let result = Client::new()
.get("http://[2606:4700::1111]/")
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.send()
.await;
assert!(
matches!(result, Err(ClientError::PinRequiresDomainHost(_))),
"expected PinRequiresDomainHost, got {result:?}"
);
}
#[tokio::test]
async fn get_ssrf_safe_with_pin_to_is_rejected() {
let result = Client::new()
.get_ssrf_safe("http://example.com/")
.pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
.send()
.await;
assert!(
matches!(result, Err(ClientError::PinNotAllowedWithSsrfSafe(_))),
"expected PinNotAllowedWithSsrfSafe, got {result:?}"
);
}
#[tokio::test]
async fn follow_redirects_does_not_follow_304_with_location() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let b_addr = spawn(Router::new().route(
"/dst",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"SHOULD-NOT-BE-HIT"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/start",
get(move || async move {
response_with_location(304, format!("http://127.0.0.1:{b_port}/dst"))
}),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", a_addr.port()))
.follow_redirects(5, |_| true)
.send()
.await
.unwrap();
assert_eq!(
resp.status().as_u16(),
304,
"a 304 with a Location header must be returned verbatim, not followed"
);
assert!(
!touched.load(Ordering::SeqCst),
"the 304 Location target must never be requested"
);
}
#[tokio::test]
async fn follow_redirects_does_not_follow_300_with_location() {
use axum::{Router, routing::get};
use std::sync::atomic::AtomicBool;
let touched = Arc::new(AtomicBool::new(false));
let touched2 = touched.clone();
let b_addr = spawn(Router::new().route(
"/dst",
get(move || {
let t = touched2.clone();
async move {
t.store(true, Ordering::SeqCst);
"SHOULD-NOT-BE-HIT"
}
}),
))
.await;
let b_port = b_addr.port();
let a_addr = spawn(Router::new().route(
"/start",
get(move || async move {
response_with_location(300, format!("http://127.0.0.1:{b_port}/dst"))
}),
))
.await;
let resp = Client::new()
.get(format!("http://127.0.0.1:{}/start", a_addr.port()))
.follow_redirects(5, |_| true)
.send()
.await
.unwrap();
assert_eq!(
resp.status().as_u16(),
300,
"a 300 with a Location header must be returned verbatim, not followed"
);
assert!(
!touched.load(Ordering::SeqCst),
"the 300 Location target must never be requested"
);
}
}