1use std::collections::HashMap;
49use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
50use std::sync::atomic::{AtomicUsize, Ordering};
51use std::sync::{Arc, Mutex};
52use std::time::{Duration, Instant};
53
54use bytes::Bytes;
55use reqwest::Method;
56use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
57use serde::Serialize;
58use serde::de::DeserializeOwned;
59
60#[derive(Debug, thiserror::Error)]
64pub enum ClientError {
65 #[error("outbound HTTP request failed: {0}")]
67 Request(#[from] reqwest::Error),
68 #[error("JSON error: {0}")]
70 Json(#[from] serde_json::Error),
71 #[error("no mock registered for {0} {1}")]
73 NoMock(String, String),
74 #[error("outbound circuit breaker is open")]
76 CircuitBreakerOpen,
77 #[error("SSRF policy blocked address: {0}")]
81 SsrfBlocked(String),
82 #[error("too many redirects (max {0})")]
84 TooManyRedirects(usize),
85 #[error("redirect rejected: {0}")]
88 RedirectRejected(String),
89 #[error("invalid or unresolvable URL: {0}")]
92 InvalidUrl(String),
93 #[error("{0}")]
102 IncompatiblePinRedirect(&'static str),
103 #[error("{0}")]
112 PinRequiresDomainHost(&'static str),
113 #[error("{0}")]
120 PinNotAllowedWithSsrfSafe(&'static str),
121}
122
123#[derive(Debug)]
130pub struct Response {
131 status: reqwest::StatusCode,
132 headers: HeaderMap,
133 body: Bytes,
134 url: Option<reqwest::Url>,
135}
136
137impl Response {
138 pub const fn status(&self) -> reqwest::StatusCode {
140 self.status
141 }
142
143 pub const fn headers(&self) -> &HeaderMap {
145 &self.headers
146 }
147
148 pub fn is_success(&self) -> bool {
150 self.status.is_success()
151 }
152
153 pub const fn url(&self) -> Option<&reqwest::Url> {
155 self.url.as_ref()
156 }
157
158 pub fn json<T: DeserializeOwned>(self) -> Result<T, ClientError> {
163 serde_json::from_slice(&self.body).map_err(ClientError::Json)
164 }
165
166 pub fn text(self) -> String {
168 String::from_utf8_lossy(&self.body).into_owned()
169 }
170
171 pub fn bytes(self) -> Bytes {
173 self.body
174 }
175}
176
177#[must_use]
194pub fn is_blocked_ip(ip: IpAddr) -> bool {
195 match ip {
196 IpAddr::V4(v4) => is_blocked_ipv4(v4),
197 IpAddr::V6(v6) => is_blocked_ipv6(v6),
198 }
199}
200
201#[must_use]
204pub fn is_public_ip(ip: IpAddr) -> bool {
205 !is_blocked_ip(ip)
206}
207
208fn is_blocked_ipv4(ip: Ipv4Addr) -> bool {
209 let [a, b, c, _d] = ip.octets();
210 if a == 0 {
212 return true;
213 }
214 if a == 10 {
216 return true;
217 }
218 if a == 100 && (64..=127).contains(&b) {
220 return true;
221 }
222 if a == 127 {
224 return true;
225 }
226 if a == 169 && b == 254 {
228 return true;
229 }
230 if a == 172 && (16..=31).contains(&b) {
232 return true;
233 }
234 if a == 192 && b == 0 && c == 0 {
236 return true;
237 }
238 if a == 192 && b == 0 && c == 2 {
240 return true;
241 }
242 if a == 192 && b == 88 && c == 99 {
244 return true;
245 }
246 if a == 192 && b == 168 {
248 return true;
249 }
250 if a == 198 && (18..=19).contains(&b) {
252 return true;
253 }
254 if a == 198 && b == 51 && c == 100 {
256 return true;
257 }
258 if a == 203 && b == 0 && c == 113 {
260 return true;
261 }
262 if a >= 224 {
264 return true;
265 }
266 false
267}
268
269fn embedded_ipv4(ip: Ipv6Addr) -> Option<Ipv4Addr> {
287 if let Some(v4) = ip.to_ipv4_mapped() {
288 return Some(v4);
289 }
290 let segs = ip.segments();
291 if segs[0] == 0
298 && segs[1] == 0
299 && segs[2] == 0
300 && segs[3] == 0
301 && segs[4] == 0xffff
302 && segs[5] == 0
303 {
304 let o = ip.octets();
305 return Some(Ipv4Addr::new(o[12], o[13], o[14], o[15]));
306 }
307 let o = ip.octets();
308 if o[0] == 0x00
311 && o[1] == 0x64
312 && o[2] == 0xff
313 && o[3] == 0x9b
314 && o[4..12].iter().all(|&b| b == 0)
315 {
316 return Some(Ipv4Addr::new(o[12], o[13], o[14], o[15]));
317 }
318 if o[0] == 0x20 && o[1] == 0x02 {
320 return Some(Ipv4Addr::new(o[2], o[3], o[4], o[5]));
321 }
322 ip.to_ipv4()
324}
325
326fn is_blocked_ipv6(ip: Ipv6Addr) -> bool {
327 if let Some(v4) = embedded_ipv4(ip)
334 && is_blocked_ipv4(v4)
335 {
336 return true;
337 }
338
339 let segs = ip.segments();
340 if segs[0] == 0x0064 && segs[1] == 0xff9b && segs[2] == 0x0001 {
347 return true;
348 }
349 if ip == Ipv6Addr::UNSPECIFIED {
351 return true;
352 }
353 if ip == Ipv6Addr::LOCALHOST {
355 return true;
356 }
357 if segs[0] & 0xfe00 == 0xfc00 {
359 return true;
360 }
361 if segs[0] & 0xffc0 == 0xfe80 {
363 return true;
364 }
365 if segs[0] & 0xffc0 == 0xfec0 {
367 return true;
368 }
369 if segs[0] & 0xff00 == 0xff00 {
371 return true;
372 }
373 if segs[0] == 0x2001 && segs[1] == 0x0db8 {
375 return true;
376 }
377
378 let s = segs;
394 if s[0] == 0x0100 && s[1] == 0 && s[2] == 0 && s[3] == 0 {
396 return true;
397 }
398 if s[0] == 0x2001 && s[1] == 0x0002 && s[2] == 0x0000 {
400 return true;
401 }
402 if s[0] == 0x2001 && (s[1] & 0xFFF0) == 0x0010 {
404 return true;
405 }
406 if s[0] == 0x2001 && (s[1] & 0xFFF0) == 0x0020 {
408 return true;
409 }
410 if s[0] == 0x3fff && (s[1] & 0xF000) == 0x0000 {
412 return true;
413 }
414 if s[0] == 0x5f00 {
416 return true;
417 }
418 if s[0] == 0x2620 && s[1] == 0x004f && s[2] == 0x8000 {
420 return true;
421 }
422
423 false
424}
425
426#[derive(Clone, Debug)]
430pub struct RetryPolicy {
431 pub max_retries: u32,
434 pub retry_idempotent_only: bool,
437 pub max_retry_after: Duration,
439 pub request_timeout: Option<Duration>,
441}
442
443impl Default for RetryPolicy {
444 fn default() -> Self {
445 Self {
446 max_retries: 3,
447 retry_idempotent_only: true,
448 max_retry_after: Duration::from_secs(10),
449 request_timeout: Some(Duration::from_secs(30)),
450 }
451 }
452}
453
454pub(crate) struct MockEntry {
458 pub(crate) method: Option<Method>,
459 pub(crate) path: String,
461 pub(crate) alias: Option<String>,
463 pub(crate) status: u16,
464 pub(crate) body: Option<serde_json::Value>,
465 pub(crate) call_count: Arc<AtomicUsize>,
466}
467
468pub(crate) struct MockResponse {
470 pub(crate) status: u16,
471 pub(crate) body: Option<serde_json::Value>,
472}
473
474pub struct MockRegistry {
480 entries: Mutex<Vec<MockEntry>>,
481}
482
483impl MockRegistry {
484 #[must_use]
486 pub const fn new() -> Self {
487 Self {
488 entries: Mutex::new(Vec::new()),
489 }
490 }
491
492 pub(crate) fn register(&self, entry: MockEntry) {
494 self.entries
495 .lock()
496 .expect("mock registry lock poisoned")
497 .push(entry);
498 }
499
500 pub(crate) fn find_match(
503 &self,
504 method: &Method,
505 url: &str,
506 alias: Option<&str>,
507 ) -> Option<MockResponse> {
508 let url_path_owned: String = reqwest::Url::parse(url).map_or_else(
514 |_| {
515 let s = url.split_once('?').map_or(url, |(p, _)| p);
516 s.split_once('#').map_or(s, |(p, _)| p).to_owned()
517 },
518 |parsed| parsed.path().to_owned(),
519 );
520 let url_path = url_path_owned.as_str();
521
522 let found = {
524 let entries = self.entries.lock().expect("mock registry lock poisoned");
525 entries.iter().find_map(|entry| {
526 let method_ok = entry.method.as_ref().is_none_or(|m| m == method);
527 let path_ok = url_path == entry.path.as_str()
532 || url_path
533 .strip_suffix(entry.path.as_str())
534 .is_some_and(|prefix| {
535 prefix.is_empty()
536 || prefix.ends_with('/')
537 || entry.path.starts_with('/')
538 });
539 let alias_ok = entry
540 .alias
541 .as_deref()
542 .is_none_or(|a| alias.is_some_and(|b| a == b));
543 if method_ok && path_ok && alias_ok {
544 Some((entry.call_count.clone(), entry.status, entry.body.clone()))
545 } else {
546 None
547 }
548 })
549 };
550
551 found.map(|(call_count, status, body)| {
552 call_count.fetch_add(1, Ordering::SeqCst);
553 MockResponse { status, body }
554 })
555 }
556}
557
558impl Default for MockRegistry {
559 fn default() -> Self {
560 Self::new()
561 }
562}
563
564pub struct HttpMockRegistryExt(pub Arc<MockRegistry>);
567
568#[derive(Clone)]
578pub(crate) struct SharedReqwestClient {
579 pub(crate) client: reqwest::Client,
580 pub(crate) timeout_secs: u64,
581}
582
583pub struct MockHandle {
586 alias: String,
587 method: String,
588 path: String,
589 call_count: Arc<AtomicUsize>,
590}
591
592impl MockHandle {
593 pub fn expect_called(&self, expected: usize) {
599 let actual = self.call_count.load(Ordering::SeqCst);
600 assert_eq!(
601 actual, expected,
602 "http mock for {} {} {} expected {} call(s) but got {}",
603 self.alias, self.method, self.path, expected, actual,
604 );
605 }
606
607 #[must_use]
609 pub fn call_count(&self) -> usize {
610 self.call_count.load(Ordering::SeqCst)
611 }
612}
613
614pub struct MockSetupBuilder {
620 pub(crate) registry: Arc<MockRegistry>,
621 pub(crate) alias: String,
622 pub(crate) method: Option<Method>,
623 pub(crate) path: Option<String>,
624}
625
626impl MockSetupBuilder {
627 #[must_use]
629 pub fn get(mut self, path: &str) -> Self {
630 self.method = Some(Method::GET);
631 self.path = Some(path.to_owned());
632 self
633 }
634 #[must_use]
636 pub fn post(mut self, path: &str) -> Self {
637 self.method = Some(Method::POST);
638 self.path = Some(path.to_owned());
639 self
640 }
641 #[must_use]
643 pub fn put(mut self, path: &str) -> Self {
644 self.method = Some(Method::PUT);
645 self.path = Some(path.to_owned());
646 self
647 }
648 #[must_use]
650 pub fn patch(mut self, path: &str) -> Self {
651 self.method = Some(Method::PATCH);
652 self.path = Some(path.to_owned());
653 self
654 }
655 #[must_use]
657 pub fn delete(mut self, path: &str) -> Self {
658 self.method = Some(Method::DELETE);
659 self.path = Some(path.to_owned());
660 self
661 }
662
663 #[must_use]
665 pub fn head(mut self, path: &str) -> Self {
666 self.method = Some(Method::HEAD);
667 self.path = Some(path.to_owned());
668 self
669 }
670
671 #[must_use]
676 pub fn respond_with(self, status: u16, body: serde_json::Value) -> MockHandle {
677 let path = self.path.clone().unwrap_or_default();
678 let method_str = self
679 .method
680 .as_ref()
681 .map_or_else(|| "*".to_owned(), ToString::to_string);
682 let call_count = Arc::new(AtomicUsize::new(0));
683
684 self.registry.register(MockEntry {
685 method: self.method,
686 path: path.clone(),
687 alias: Some(self.alias.clone()),
688 status,
689 body: Some(body),
690 call_count: call_count.clone(),
691 });
692
693 MockHandle {
694 alias: self.alias,
695 method: method_str,
696 path,
697 call_count,
698 }
699 }
700
701 #[must_use]
706 pub fn respond_with_status(self, status: u16) -> MockHandle {
707 let path = self.path.clone().unwrap_or_default();
708 let method_str = self
709 .method
710 .as_ref()
711 .map_or_else(|| "*".to_owned(), ToString::to_string);
712 let call_count = Arc::new(AtomicUsize::new(0));
713
714 self.registry.register(MockEntry {
715 method: self.method,
716 path: path.clone(),
717 alias: Some(self.alias.clone()),
718 status,
719 body: None,
720 call_count: call_count.clone(),
721 });
722
723 MockHandle {
724 alias: self.alias,
725 method: method_str,
726 path,
727 call_count,
728 }
729 }
730}
731
732#[derive(Clone)]
760pub struct Client {
761 inner: reqwest::Client,
762 alias: Option<String>,
764 base_url: Option<String>,
766 base_urls: HashMap<String, String>,
768 retry_policy: RetryPolicy,
769 mock: Option<Arc<MockRegistry>>,
771 resilience_config: Option<Arc<crate::config::ResilienceConfig>>,
773}
774
775impl Client {
776 #[must_use]
779 pub fn new() -> Self {
780 Self::with_timeout(Duration::from_secs(30))
781 }
782
783 #[must_use]
790 pub fn with_timeout(timeout: Duration) -> Self {
791 let inner = reqwest::ClientBuilder::new()
792 .timeout(timeout)
793 .build()
794 .expect("failed to build reqwest client");
795 Self {
796 inner,
797 alias: None,
798 base_url: None,
799 base_urls: HashMap::new(),
800 retry_policy: RetryPolicy {
801 max_retries: 3,
802 retry_idempotent_only: true,
803 max_retry_after: Duration::from_secs(10),
804 request_timeout: Some(timeout),
805 },
806 mock: None,
807 resilience_config: None,
808 }
809 }
810
811 pub(crate) fn build_inner(config: &crate::config::HttpClientConfig) -> reqwest::Client {
821 reqwest::ClientBuilder::new()
822 .timeout(Duration::from_secs(config.timeout_secs))
823 .build()
824 .expect("failed to build reqwest client")
825 }
826
827 fn from_config_with_inner(
831 inner: reqwest::Client,
832 config: &crate::config::HttpClientConfig,
833 ) -> Self {
834 let timeout = Duration::from_secs(config.timeout_secs);
835 Self {
836 inner,
837 alias: None,
838 base_url: None,
839 base_urls: config.base_urls.clone(),
840 retry_policy: RetryPolicy {
841 max_retries: config.max_retries,
842 retry_idempotent_only: true,
843 max_retry_after: Duration::from_secs(config.max_retry_after_secs),
844 request_timeout: Some(timeout),
845 },
846 mock: None,
847 resilience_config: None,
848 }
849 }
850
851 fn with_inner(inner: reqwest::Client) -> Self {
855 Self {
856 inner,
857 alias: None,
858 base_url: None,
859 base_urls: HashMap::new(),
860 retry_policy: RetryPolicy::default(),
861 mock: None,
862 resilience_config: None,
863 }
864 }
865
866 #[must_use]
873 pub fn from_config(config: &crate::config::HttpClientConfig) -> Self {
874 Self::from_config_with_inner(Self::build_inner(config), config)
875 }
876
877 pub(crate) fn with_mock(mut self, registry: Arc<MockRegistry>) -> Self {
879 self.mock = Some(registry);
880 self
881 }
882
883 #[must_use]
894 pub fn from_state(state: &crate::AppState) -> Self {
895 let autumn_config = state.extension::<crate::config::AutumnConfig>();
896 let config = state
897 .extension::<crate::config::HttpConfig>()
898 .or_else(|| autumn_config.as_ref().map(|c| Arc::new(c.http.clone())));
899
900 let effective_timeout_secs = config.as_ref().map_or_else(
906 || crate::config::HttpClientConfig::default().timeout_secs,
907 |c| c.client.timeout_secs,
908 );
909 let shared = state.extension::<SharedReqwestClient>().and_then(|s| {
910 if s.timeout_secs == effective_timeout_secs {
911 Some(s.client.clone())
912 } else {
913 None
914 }
915 });
916
917 let mut client = match (config, shared) {
918 (Some(cfg), Some(inner)) => Self::from_config_with_inner(inner, &cfg.client),
919 (Some(cfg), None) => Self::from_config(&cfg.client),
920 (None, Some(inner)) => Self::with_inner(inner),
921 (None, None) => Self::new(),
922 };
923
924 client.resilience_config = autumn_config.map(|c| Arc::new(c.resilience.clone()));
925
926 if let Some(ext) = state.extension::<HttpMockRegistryExt>() {
927 client = client.with_mock(ext.0.clone());
928 }
929
930 client
931 }
932
933 #[must_use]
940 pub fn named(&self, alias: &str) -> Self {
941 let base_url = self
942 .base_urls
943 .get(alias)
944 .cloned()
945 .or_else(|| self.base_url.clone());
946 Self {
947 inner: self.inner.clone(),
948 alias: Some(alias.to_owned()),
949 base_url,
950 base_urls: self.base_urls.clone(),
951 retry_policy: self.retry_policy.clone(),
952 mock: self.mock.clone(),
953 resilience_config: self.resilience_config.clone(),
954 }
955 }
956
957 #[must_use]
959 pub fn with_base_url(&self, base_url: impl Into<String>) -> Self {
960 Self {
961 inner: self.inner.clone(),
962 alias: self.alias.clone(),
963 base_url: Some(base_url.into()),
964 base_urls: self.base_urls.clone(),
965 retry_policy: self.retry_policy.clone(),
966 mock: self.mock.clone(),
967 resilience_config: self.resilience_config.clone(),
968 }
969 }
970
971 fn build_request(&self, method: Method, url: impl AsRef<str>) -> RequestBuilder {
972 let url_str = url.as_ref();
973 let full_url = if url_str.starts_with("http://") || url_str.starts_with("https://") {
974 url_str.to_owned()
975 } else if let Some(base) = &self.base_url {
976 format!(
977 "{}/{}",
978 base.trim_end_matches('/'),
979 url_str.trim_start_matches('/')
980 )
981 } else {
982 url_str.to_owned()
983 };
984
985 RequestBuilder {
986 client: self.inner.clone(),
987 method,
988 url: full_url,
989 extra_headers: HeaderMap::new(),
990 body: None,
991 retry_policy: self.retry_policy.clone(),
992 mock: self.mock.clone(),
993 alias: self.alias.clone(),
994 pending_error: None,
995 resilience_config: self.resilience_config.clone(),
996 redirect_mode: RedirectMode::Default,
997 pin_addr: None,
998 ssrf_safe: false,
999 }
1000 }
1001
1002 #[must_use]
1004 pub fn get(&self, url: impl AsRef<str>) -> RequestBuilder {
1005 self.build_request(Method::GET, url)
1006 }
1007 #[must_use]
1009 pub fn post(&self, url: impl AsRef<str>) -> RequestBuilder {
1010 self.build_request(Method::POST, url)
1011 }
1012 #[must_use]
1014 pub fn put(&self, url: impl AsRef<str>) -> RequestBuilder {
1015 self.build_request(Method::PUT, url)
1016 }
1017 #[must_use]
1019 pub fn patch(&self, url: impl AsRef<str>) -> RequestBuilder {
1020 self.build_request(Method::PATCH, url)
1021 }
1022 #[must_use]
1024 pub fn delete(&self, url: impl AsRef<str>) -> RequestBuilder {
1025 self.build_request(Method::DELETE, url)
1026 }
1027
1028 #[must_use]
1030 pub fn head(&self, url: impl AsRef<str>) -> RequestBuilder {
1031 self.build_request(Method::HEAD, url)
1032 }
1033
1034 #[must_use]
1080 pub fn get_ssrf_safe(&self, url: impl Into<String>) -> RequestBuilder {
1081 let mut builder = self.build_request(Method::GET, url.into());
1082 builder.ssrf_safe = true;
1083 builder
1084 }
1085}
1086
1087impl Default for Client {
1088 fn default() -> Self {
1089 Self::new()
1090 }
1091}
1092
1093impl axum::extract::FromRequestParts<crate::AppState> for Client {
1094 type Rejection = std::convert::Infallible;
1095
1096 async fn from_request_parts(
1097 _parts: &mut http::request::Parts,
1098 state: &crate::AppState,
1099 ) -> Result<Self, std::convert::Infallible> {
1100 Ok(Self::from_state(state))
1101 }
1102}
1103
1104type RedirectValidator = Arc<dyn Fn(&str) -> bool + Send + Sync>;
1108
1109enum RedirectMode {
1115 Default,
1117 None,
1119 Follow {
1122 max: usize,
1123 validator: RedirectValidator,
1124 },
1125}
1126
1127const SSRF_SAFE_MAX_REDIRECTS: usize = 5;
1129
1130pub struct RequestBuilder {
1132 client: reqwest::Client,
1133 method: Method,
1134 url: String,
1135 extra_headers: HeaderMap,
1136 body: Option<Bytes>,
1138 retry_policy: RetryPolicy,
1139 mock: Option<Arc<MockRegistry>>,
1140 alias: Option<String>,
1141 pending_error: Option<ClientError>,
1143 resilience_config: Option<Arc<crate::config::ResilienceConfig>>,
1145 redirect_mode: RedirectMode,
1147 pin_addr: Option<SocketAddr>,
1150 ssrf_safe: bool,
1153}
1154
1155impl RequestBuilder {
1156 #[must_use]
1162 pub fn header(mut self, name: impl AsRef<str>, value: impl AsRef<str>) -> Self {
1163 let name_str = name.as_ref();
1164 let value_str = value.as_ref();
1165 match (
1166 HeaderName::from_bytes(name_str.as_bytes()),
1167 HeaderValue::from_str(value_str),
1168 ) {
1169 (Ok(n), Ok(v)) => {
1170 self.extra_headers.insert(n, v);
1171 }
1172 (Err(e), _) => {
1173 tracing::warn!(header.name = name_str, error = %e, "invalid header name — header skipped");
1174 }
1175 (_, Err(e)) => {
1176 tracing::warn!(header.name = name_str, error = %e, "invalid header value — header skipped");
1177 }
1178 }
1179 self
1180 }
1181
1182 #[must_use]
1187 pub fn json<T: Serialize>(mut self, body: &T) -> Self {
1188 match serde_json::to_vec(body) {
1189 Ok(bytes) => {
1190 self.body = Some(Bytes::from(bytes));
1191 self = self.header("content-type", "application/json");
1192 }
1193 Err(e) => {
1194 self.pending_error = Some(ClientError::Json(e));
1195 }
1196 }
1197 self
1198 }
1199
1200 #[must_use]
1202 pub fn text_body(mut self, body: impl Into<String>) -> Self {
1203 self.body = Some(Bytes::from(body.into().into_bytes()));
1204 self
1205 }
1206
1207 #[must_use]
1212 pub const fn retries(mut self, max: u32) -> Self {
1213 self.retry_policy.max_retries = max;
1214 self.retry_policy.retry_idempotent_only = false;
1215 self
1216 }
1217
1218 #[must_use]
1220 pub const fn max_retry_after(mut self, max: Duration) -> Self {
1221 self.retry_policy.max_retry_after = max;
1222 self
1223 }
1224
1225 #[must_use]
1227 pub const fn no_retry(mut self) -> Self {
1228 self.retry_policy.max_retries = 0;
1229 self
1230 }
1231
1232 #[must_use]
1239 pub fn no_redirect(mut self) -> Self {
1240 self.redirect_mode = RedirectMode::None;
1241 self
1242 }
1243
1244 #[must_use]
1264 pub fn follow_redirects<F>(mut self, max: usize, validator: F) -> Self
1265 where
1266 F: Fn(&str) -> bool + Send + Sync + 'static,
1267 {
1268 self.redirect_mode = RedirectMode::Follow {
1269 max,
1270 validator: Arc::new(validator),
1271 };
1272 self
1273 }
1274
1275 #[must_use]
1322 pub const fn pin_to(mut self, addr: SocketAddr) -> Self {
1323 self.pin_addr = Some(addr);
1324 self
1325 }
1326
1327 pub async fn send(self) -> Result<Response, ClientError> {
1341 if let Some(err) = self.pending_error {
1343 return Err(err);
1344 }
1345
1346 if self.mock.is_some() {
1348 return self.send_inner(false).await;
1349 }
1350
1351 if self.needs_custom_path() {
1358 return self.send_custom().await;
1359 }
1360
1361 let host = url::Url::parse(&self.url).ok().map_or_else(
1363 || "unknown".to_owned(),
1364 |u| {
1365 let h = u.host_str().unwrap_or("unknown");
1366 u.port()
1367 .map_or_else(|| h.to_owned(), |port| format!("{h}:{port}"))
1368 },
1369 );
1370
1371 let breaker = self.resilience_config.as_ref().map_or_else(
1372 || {
1373 crate::circuit_breaker::global_registry().get_or_create(
1374 &host,
1375 crate::circuit_breaker::CircuitBreakerPolicy::default(),
1376 )
1377 },
1378 |rc| {
1379 let policy = crate::circuit_breaker::CircuitBreakerPolicy::from_config(rc, &host);
1380 crate::circuit_breaker::global_registry().get_or_create_with_config(&host, policy)
1381 },
1382 );
1383
1384 if breaker.before_call().is_err() {
1386 return Err(ClientError::CircuitBreakerOpen);
1387 }
1388 let guard = crate::circuit_breaker::CircuitBreakerGuard::new(breaker.clone());
1389
1390 let is_half_open = breaker.state() == crate::circuit_breaker::CircuitState::HalfOpen;
1391 let res = self.send_inner(is_half_open).await;
1392 match &res {
1393 Ok(resp) => {
1394 let success = resp.status().as_u16() < 500;
1395 if success {
1396 guard.success();
1397 } else {
1398 guard.failure();
1399 }
1400 }
1401 Err(_) => {
1402 guard.failure();
1403 }
1404 }
1405 res
1406 }
1407
1408 async fn send_inner(self, suppress_retries: bool) -> Result<Response, ClientError> {
1409 if let Some(ref mock) = self.mock {
1411 match mock.find_match(&self.method, &self.url, self.alias.as_deref()) {
1412 Some(mock_resp) => {
1413 let status = reqwest::StatusCode::from_u16(mock_resp.status)
1414 .unwrap_or(reqwest::StatusCode::OK);
1415 let body_bytes = mock_resp
1416 .body
1417 .as_ref()
1418 .map(|v| serde_json::to_vec(v).unwrap_or_default())
1419 .unwrap_or_default();
1420
1421 tracing::info!(
1422 http.method = %self.method,
1423 http.url = %self.url,
1424 http.status = mock_resp.status,
1425 "[mock] outbound request intercepted"
1426 );
1427
1428 return Ok(Response {
1429 status,
1430 headers: HeaderMap::new(),
1431 body: Bytes::from(body_bytes),
1432 url: None,
1433 });
1434 }
1435 None => {
1436 return Err(ClientError::NoMock(
1439 self.method.to_string(),
1440 self.url.clone(),
1441 ));
1442 }
1443 }
1444 }
1445
1446 let start = Instant::now();
1448 let max_attempts = if suppress_retries {
1449 1
1450 } else if is_idempotent_method(&self.method) || !self.retry_policy.retry_idempotent_only {
1451 self.retry_policy.max_retries.saturating_add(1)
1452 } else {
1453 1
1454 };
1455
1456 for attempt in 0..max_attempts {
1457 if attempt > 0 {
1458 let exp = (attempt - 1).min(10);
1460 let delay = Duration::from_millis(100 * (1_u64 << exp));
1461 tokio::time::sleep(delay).await;
1462 }
1463
1464 let mut req = self.client.request(self.method.clone(), &self.url);
1465
1466 req = inject_trace_context(req);
1468
1469 for (name, value) in &self.extra_headers {
1471 req = req.header(name.clone(), value.clone());
1472 }
1473
1474 if let Some(body) = &self.body {
1475 req = req.body(body.clone());
1476 }
1477
1478 match req.send().await {
1479 Ok(resp) => {
1480 let status = resp.status();
1481 let headers = resp.headers().clone();
1482 let url_used = resp.url().clone();
1483
1484 if status.as_u16() == 429 && attempt + 1 < max_attempts {
1486 let mut sleep_delay =
1487 parse_retry_after(&headers).unwrap_or(Duration::from_secs(1));
1488 sleep_delay = sleep_delay.min(self.retry_policy.max_retry_after);
1489 if let Some(req_timeout) = self.retry_policy.request_timeout {
1490 sleep_delay = sleep_delay.min(req_timeout);
1491 }
1492 tokio::time::sleep(sleep_delay).await;
1493 continue;
1494 }
1495
1496 if is_retryable_status(status.as_u16()) && attempt + 1 < max_attempts {
1498 continue;
1499 }
1500
1501 let body = resp
1502 .bytes()
1503 .await
1504 .map_err(|e| ClientError::Request(e.without_url()))?;
1505 let elapsed = start.elapsed();
1506 log_request(
1507 self.method.as_str(),
1508 &url_used,
1509 status.as_u16(),
1510 elapsed,
1511 &self.extra_headers,
1512 );
1513
1514 return Ok(Response {
1515 status,
1516 headers,
1517 body,
1518 url: Some(url_used),
1519 });
1520 }
1521 Err(e) if (e.is_connect() || e.is_timeout()) && attempt + 1 < max_attempts => {}
1524 Err(e) => return Err(ClientError::Request(e.without_url())),
1525 }
1526 }
1527
1528 unreachable!("retry loop exited without returning a result — this is a bug")
1530 }
1531
1532 const fn needs_custom_path(&self) -> bool {
1534 self.ssrf_safe
1535 || self.pin_addr.is_some()
1536 || !matches!(self.redirect_mode, RedirectMode::Default)
1537 }
1538
1539 async fn send_custom(self) -> Result<Response, ClientError> {
1541 if self.ssrf_safe && self.pin_addr.is_some() {
1550 return Err(ClientError::PinNotAllowedWithSsrfSafe(
1551 "get_ssrf_safe cannot be combined with pin_to: the SSRF-safe path \
1552 performs its own per-hop resolve/validate/pin and never reads the \
1553 pin_to address, so an explicit pin would be silently ignored. Use \
1554 pin_to alone for a caller-chosen address, or get_ssrf_safe alone \
1555 for guarded automatic per-hop pinning.",
1556 ));
1557 }
1558
1559 if self.pin_addr.is_some() && matches!(self.redirect_mode, RedirectMode::Follow { .. }) {
1566 return Err(ClientError::IncompatiblePinRedirect(
1567 "pin_to cannot be combined with follow_redirects: the pin only \
1568 covers the first hop and later redirect hops re-resolve via DNS, \
1569 escaping the pin. Use get_ssrf_safe for pinned, per-hop-revalidated \
1570 redirect following; pin_to alone (which returns the 3xx unfollowed); \
1571 or follow_redirects without pin_to.",
1572 ));
1573 }
1574
1575 if self.pin_addr.is_some() && url_host_is_ip_literal(&self.url)? {
1585 return Err(ClientError::PinRequiresDomainHost(
1586 "pin_to cannot be honored for an IP-literal URL host because the \
1587 HTTP stack connects to the literal directly and skips the pinned \
1588 address; put the desired IP directly in the URL, or use a domain host.",
1589 ));
1590 }
1591
1592 let timeout = self
1593 .retry_policy
1594 .request_timeout
1595 .unwrap_or_else(|| Duration::from_secs(30));
1596
1597 if self.ssrf_safe {
1598 return self.send_ssrf_safe(timeout).await;
1599 }
1600
1601 let follow = match &self.redirect_mode {
1603 RedirectMode::Follow { max, validator } => Some((*max, validator.clone())),
1604 RedirectMode::None | RedirectMode::Default => None,
1605 };
1606 if let Some((max, validator)) = follow {
1607 return self.follow_loop(max, validator, timeout).await;
1608 }
1609
1610 let policy = reqwest::redirect::Policy::none();
1619 let resolve = self.pin_resolve()?;
1620 let client = build_oneshot_client(resolve, policy, timeout)?;
1621 send_one(
1622 &client,
1623 &self.method,
1624 &self.url,
1625 &self.extra_headers,
1626 self.body.as_ref(),
1627 &self.retry_policy,
1628 )
1629 .await
1630 }
1631
1632 fn pin_resolve(&self) -> Result<Option<(String, Vec<SocketAddr>)>, ClientError> {
1634 match self.pin_addr {
1635 Some(addr) => Ok(Some((host_of(&self.url)?, vec![addr]))),
1638 None => Ok(None),
1639 }
1640 }
1641
1642 async fn follow_loop(
1644 self,
1645 max: usize,
1646 validator: RedirectValidator,
1647 timeout: Duration,
1648 ) -> Result<Response, ClientError> {
1649 let original =
1650 url::Url::parse(&self.url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
1651 let mut current = self.url.clone();
1652 let mut method = self.method.clone();
1655 let mut headers = self.extra_headers.clone();
1656 let mut body = self.body.clone();
1657 for hop in 0.. {
1658 let resolve = if hop == 0 {
1660 match self.pin_addr {
1661 Some(addr) => Some((host_of(¤t)?, vec![addr])),
1662 None => None,
1663 }
1664 } else {
1665 None
1666 };
1667 if hop > 0 {
1670 strip_sensitive_headers_if_cross_origin(&mut headers, &original, ¤t)?;
1671 }
1672 let client = build_oneshot_client(resolve, reqwest::redirect::Policy::none(), timeout)?;
1673 let resp = send_one(
1674 &client,
1675 &method,
1676 ¤t,
1677 &headers,
1678 body.as_ref(),
1679 &self.retry_policy,
1680 )
1681 .await?;
1682
1683 let Some(next) = redirect_target(&resp, ¤t)? else {
1684 return Ok(resp);
1685 };
1686 if hop >= max {
1687 return Err(ClientError::TooManyRedirects(max));
1688 }
1689 if !validator(&next) {
1690 return Err(ClientError::RedirectRejected(next));
1691 }
1692 rewrite_after_redirect(resp.status(), &mut method, &mut body, &mut headers);
1694 current = next;
1695 }
1696 unreachable!("redirect loop is bounded by `max` and always returns")
1697 }
1698
1699 const fn ssrf_redirect_plan(&self) -> (bool, usize) {
1710 match &self.redirect_mode {
1711 RedirectMode::Default => (true, SSRF_SAFE_MAX_REDIRECTS),
1712 RedirectMode::None => (false, 0),
1713 RedirectMode::Follow { max, .. } => (true, *max),
1714 }
1715 }
1716
1717 async fn send_ssrf_safe(self, timeout: Duration) -> Result<Response, ClientError> {
1726 let (follow, max) = self.ssrf_redirect_plan();
1727 let original =
1728 url::Url::parse(&self.url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
1729 let mut current = self.url.clone();
1730 let mut method = self.method.clone();
1733 let mut headers = self.extra_headers.clone();
1734 let mut body = self.body.clone();
1735 for hop in 0.. {
1736 let addrs = resolve_and_validate(¤t).await?;
1740 let host = host_of(¤t)?;
1741 let client = build_oneshot_client(
1742 Some((host, addrs)),
1743 reqwest::redirect::Policy::none(),
1744 timeout,
1745 )?;
1746 if hop > 0 {
1749 strip_sensitive_headers_if_cross_origin(&mut headers, &original, ¤t)?;
1750 }
1751 let resp = send_one(
1752 &client,
1753 &method,
1754 ¤t,
1755 &headers,
1756 body.as_ref(),
1757 &self.retry_policy,
1758 )
1759 .await?;
1760
1761 if !follow {
1769 return Ok(resp);
1770 }
1771 let Some(next) = redirect_target(&resp, ¤t)? else {
1772 return Ok(resp);
1773 };
1774 if hop >= max {
1775 return Err(ClientError::TooManyRedirects(max));
1776 }
1777 if scheme_is_https(¤t)? && !scheme_is_https(&next)? {
1779 return Err(ClientError::RedirectRejected(format!(
1780 "https→http scheme downgrade on redirect to {next}"
1781 )));
1782 }
1783 if let RedirectMode::Follow { validator, .. } = &self.redirect_mode
1787 && !validator(&next)
1788 {
1789 return Err(ClientError::RedirectRejected(next));
1790 }
1791 rewrite_after_redirect(resp.status(), &mut method, &mut body, &mut headers);
1793 current = next;
1794 }
1798 unreachable!("redirect loop is bounded by the SSRF-safe redirect plan")
1799 }
1800}
1801
1802const fn is_idempotent_method(method: &Method) -> bool {
1805 matches!(
1806 *method,
1807 Method::GET | Method::HEAD | Method::PUT | Method::DELETE | Method::OPTIONS | Method::TRACE
1808 )
1809}
1810
1811const fn is_retryable_status(status: u16) -> bool {
1812 matches!(status, 502..=504)
1813}
1814
1815fn build_oneshot_client(
1829 resolve: Option<(String, Vec<SocketAddr>)>,
1830 policy: reqwest::redirect::Policy,
1831 timeout: Duration,
1832) -> Result<reqwest::Client, ClientError> {
1833 let mut builder = reqwest::ClientBuilder::new()
1834 .timeout(timeout)
1835 .redirect(policy);
1836 if let Some((host, addrs)) = resolve
1837 && !addrs.is_empty()
1838 {
1839 builder = builder.no_proxy().resolve_to_addrs(&host, &addrs);
1845 }
1846 builder.build().map_err(ClientError::Request)
1847}
1848
1849async fn send_one(
1853 client: &reqwest::Client,
1854 method: &Method,
1855 url: &str,
1856 extra_headers: &HeaderMap,
1857 body: Option<&Bytes>,
1858 retry_policy: &RetryPolicy,
1859) -> Result<Response, ClientError> {
1860 let start = Instant::now();
1861 let max_attempts = if is_idempotent_method(method) || !retry_policy.retry_idempotent_only {
1862 retry_policy.max_retries.saturating_add(1)
1863 } else {
1864 1
1865 };
1866
1867 for attempt in 0..max_attempts {
1868 if attempt > 0 {
1869 let exp = (attempt - 1).min(10);
1870 let delay = Duration::from_millis(100 * (1_u64 << exp));
1871 tokio::time::sleep(delay).await;
1872 }
1873
1874 let mut req = client.request(method.clone(), url);
1875 req = inject_trace_context(req);
1876 for (name, value) in extra_headers {
1877 req = req.header(name.clone(), value.clone());
1878 }
1879 if let Some(body) = body {
1880 req = req.body(body.clone());
1881 }
1882
1883 match req.send().await {
1884 Ok(resp) => {
1885 let status = resp.status();
1886 let headers = resp.headers().clone();
1887 let url_used = resp.url().clone();
1888
1889 if status.as_u16() == 429 && attempt + 1 < max_attempts {
1890 let mut sleep_delay =
1891 parse_retry_after(&headers).unwrap_or(Duration::from_secs(1));
1892 sleep_delay = sleep_delay.min(retry_policy.max_retry_after);
1893 if let Some(req_timeout) = retry_policy.request_timeout {
1894 sleep_delay = sleep_delay.min(req_timeout);
1895 }
1896 tokio::time::sleep(sleep_delay).await;
1897 continue;
1898 }
1899 if is_retryable_status(status.as_u16()) && attempt + 1 < max_attempts {
1900 continue;
1901 }
1902
1903 let body = resp
1904 .bytes()
1905 .await
1906 .map_err(|e| ClientError::Request(e.without_url()))?;
1907 log_request(
1908 method.as_str(),
1909 &url_used,
1910 status.as_u16(),
1911 start.elapsed(),
1912 extra_headers,
1913 );
1914 return Ok(Response {
1915 status,
1916 headers,
1917 body,
1918 url: Some(url_used),
1919 });
1920 }
1921 Err(e) if (e.is_connect() || e.is_timeout()) && attempt + 1 < max_attempts => {}
1922 Err(e) => return Err(ClientError::Request(e.without_url())),
1923 }
1924 }
1925
1926 unreachable!("retry loop exited without returning a result — this is a bug")
1927}
1928
1929fn host_of(url: &str) -> Result<String, ClientError> {
1931 let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
1932 parsed
1933 .host_str()
1934 .map(str::to_owned)
1935 .ok_or_else(|| ClientError::InvalidUrl(format!("URL has no host: {url}")))
1936}
1937
1938fn url_host_is_ip_literal(url: &str) -> Result<bool, ClientError> {
1942 let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
1943 match parsed.host() {
1944 Some(url::Host::Ipv4(_) | url::Host::Ipv6(_)) => Ok(true),
1945 Some(url::Host::Domain(_)) => Ok(false),
1946 None => Err(ClientError::InvalidUrl(format!("URL has no host: {url}"))),
1947 }
1948}
1949
1950fn scheme_is_https(url: &str) -> Result<bool, ClientError> {
1952 let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
1953 Ok(parsed.scheme().eq_ignore_ascii_case("https"))
1954}
1955
1956fn redirect_target(resp: &Response, base: &str) -> Result<Option<String>, ClientError> {
1963 match resp.status() {
1969 reqwest::StatusCode::MOVED_PERMANENTLY
1970 | reqwest::StatusCode::FOUND
1971 | reqwest::StatusCode::SEE_OTHER
1972 | reqwest::StatusCode::TEMPORARY_REDIRECT
1973 | reqwest::StatusCode::PERMANENT_REDIRECT => {}
1974 _ => return Ok(None),
1975 }
1976 let Some(location) = resp
1977 .headers()
1978 .get(reqwest::header::LOCATION)
1979 .and_then(|v| v.to_str().ok())
1980 else {
1981 return Ok(None);
1982 };
1983 let base_url = url::Url::parse(base).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
1984 let joined = base_url
1985 .join(location)
1986 .map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
1987 Ok(Some(joined.to_string()))
1988}
1989
1990fn strip_sensitive_headers_if_cross_origin(
2000 headers: &mut HeaderMap,
2001 original: &url::Url,
2002 current: &str,
2003) -> Result<(), ClientError> {
2004 let current_url =
2005 url::Url::parse(current).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
2006 if current_url.origin() != original.origin() {
2007 headers.remove(reqwest::header::AUTHORIZATION);
2008 headers.remove(reqwest::header::COOKIE);
2009 headers.remove(reqwest::header::PROXY_AUTHORIZATION);
2010 }
2011 Ok(())
2012}
2013
2014fn rewrite_after_redirect(
2034 status: reqwest::StatusCode,
2035 method: &mut Method,
2036 body: &mut Option<Bytes>,
2037 headers: &mut HeaderMap,
2038) {
2039 match status.as_u16() {
2040 303 => {
2041 if *method != Method::HEAD {
2042 *method = Method::GET;
2043 }
2044 *body = None;
2045 strip_payload_headers(headers);
2046 }
2047 301 | 302 if *method == Method::POST => {
2048 *method = Method::GET;
2049 *body = None;
2050 strip_payload_headers(headers);
2051 }
2052 _ => {}
2053 }
2054}
2055
2056fn strip_payload_headers(headers: &mut HeaderMap) {
2060 use reqwest::header::{
2061 CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH, CONTENT_TYPE, TRANSFER_ENCODING,
2062 };
2063 headers.remove(CONTENT_TYPE);
2064 headers.remove(CONTENT_LENGTH);
2065 headers.remove(TRANSFER_ENCODING);
2066 headers.remove(CONTENT_ENCODING);
2067 headers.remove(CONTENT_LANGUAGE);
2068}
2069
2070fn validate_resolved_addrs(addrs: Vec<SocketAddr>) -> Result<Vec<SocketAddr>, ClientError> {
2079 for addr in &addrs {
2080 if is_blocked_ip(addr.ip()) {
2081 return Err(ClientError::SsrfBlocked(addr.ip().to_string()));
2082 }
2083 }
2084 Ok(addrs)
2085}
2086
2087async fn resolve_and_validate(url: &str) -> Result<Vec<SocketAddr>, ClientError> {
2099 let parsed = url::Url::parse(url).map_err(|e| ClientError::InvalidUrl(e.to_string()))?;
2100 let scheme = parsed.scheme();
2105 if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
2106 return Err(ClientError::InvalidUrl(format!(
2107 "unsupported URL scheme `{scheme}` (only http/https are allowed): {url}"
2108 )));
2109 }
2110 let port = parsed.port_or_known_default().ok_or_else(|| {
2111 ClientError::InvalidUrl(format!("URL has no port and unknown scheme: {url}"))
2112 })?;
2113 let host = parsed
2114 .host()
2115 .ok_or_else(|| ClientError::InvalidUrl(format!("URL has no host: {url}")))?;
2116
2117 match host {
2118 url::Host::Ipv4(v4) => validate_resolved_addrs(vec![SocketAddr::new(IpAddr::V4(v4), port)]),
2119 url::Host::Ipv6(v6) => validate_resolved_addrs(vec![SocketAddr::new(IpAddr::V6(v6), port)]),
2120 url::Host::Domain(name) => {
2121 let addrs: Vec<SocketAddr> = tokio::net::lookup_host((name, port))
2122 .await
2123 .map_err(|e| ClientError::InvalidUrl(format!("DNS lookup failed for {name}: {e}")))?
2124 .collect();
2125 if addrs.is_empty() {
2126 return Err(ClientError::InvalidUrl(format!(
2127 "DNS lookup for {name} returned no addresses"
2128 )));
2129 }
2130 validate_resolved_addrs(addrs)
2133 }
2134 }
2135}
2136
2137fn parse_retry_after(headers: &HeaderMap) -> Option<Duration> {
2138 let value = headers.get("retry-after")?.to_str().ok()?;
2139 if let Ok(secs) = value.parse::<u64>() {
2141 return Some(Duration::from_secs(secs));
2142 }
2143 let dt = chrono::DateTime::parse_from_rfc2822(value).ok()?;
2145 let now = chrono::Utc::now();
2146 let future = dt.with_timezone(&chrono::Utc);
2147 let secs = u64::try_from((future - now).num_seconds().max(0)).unwrap_or(0);
2148 Some(Duration::from_secs(secs))
2149}
2150
2151const REDACTED_HEADERS: &[&str] = &["authorization", "cookie", "set-cookie"];
2152
2153fn is_sensitive_header(name: &str) -> bool {
2154 REDACTED_HEADERS
2155 .iter()
2156 .any(|h| h.eq_ignore_ascii_case(name))
2157}
2158
2159fn log_request(
2160 method: &str,
2161 url: &reqwest::Url,
2162 status: u16,
2163 elapsed: Duration,
2164 headers: &HeaderMap,
2165) {
2166 let host = url.host_str().unwrap_or("unknown");
2167 let path = url.path();
2168
2169 let sent_headers: Vec<&str> = headers
2171 .keys()
2172 .map(HeaderName::as_str)
2173 .filter(|k| !is_sensitive_header(k))
2174 .collect();
2175
2176 tracing::info!(
2177 http.method = method,
2178 http.host = host,
2179 http.path = path,
2180 http.status = status,
2181 http.elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
2182 http.sent_headers = ?sent_headers,
2183 "outbound request"
2184 );
2185}
2186
2187#[allow(clippy::missing_const_for_fn)]
2191fn inject_trace_context(builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
2192 #[cfg(not(feature = "telemetry-otlp"))]
2193 {
2194 builder
2195 }
2196 #[cfg(feature = "telemetry-otlp")]
2197 {
2198 use std::collections::HashMap;
2199 use tracing_opentelemetry::OpenTelemetrySpanExt as _;
2200 let cx = tracing::Span::current().context();
2201 let mut map = HashMap::<String, String>::new();
2202 opentelemetry::global::get_text_map_propagator(|propagator| {
2203 propagator.inject_context(&cx, &mut TraceHeaderInjector(&mut map));
2204 });
2205 let mut builder = builder;
2206 for (k, v) in map {
2207 if let Ok(value) = HeaderValue::from_str(&v) {
2208 builder = builder.header(k, value);
2209 }
2210 }
2211 builder
2212 }
2213}
2214
2215#[cfg(feature = "telemetry-otlp")]
2216struct TraceHeaderInjector<'a>(&'a mut std::collections::HashMap<String, String>);
2217
2218#[cfg(feature = "telemetry-otlp")]
2219impl opentelemetry::propagation::Injector for TraceHeaderInjector<'_> {
2220 fn set(&mut self, key: &str, value: String) {
2221 self.0.insert(key.to_owned(), value);
2222 }
2223}
2224
2225#[cfg(test)]
2228mod tests {
2229 use super::*;
2230 use crate::config::HttpClientConfig;
2231
2232 #[test]
2234 fn client_constructs_with_defaults() {
2235 let client = Client::new();
2236 assert!(client.alias.is_none());
2237 assert!(client.base_url.is_none());
2238 assert_eq!(client.retry_policy.max_retries, 3);
2239 }
2240
2241 #[test]
2243 fn request_builder_fluent_api_compiles() {
2244 let client = Client::new();
2245 let _builder = client
2246 .post("https://example.com/api")
2247 .header("x-api-key", "secret")
2248 .json(&serde_json::json!({"key": "value"}))
2249 .retries(2);
2250 }
2251
2252 #[test]
2254 fn response_accessors_work() {
2255 let payload = serde_json::json!({"id": 42, "name": "Alice"});
2256 let body = serde_json::to_vec(&payload).unwrap();
2257 let resp = Response {
2258 status: reqwest::StatusCode::OK,
2259 headers: HeaderMap::new(),
2260 body: Bytes::from(body),
2261 url: None,
2262 };
2263 assert_eq!(resp.status().as_u16(), 200);
2264 assert!(resp.is_success());
2265 }
2266
2267 #[test]
2269 fn response_json_deserialises() {
2270 #[derive(serde::Deserialize, PartialEq, Debug)]
2271 struct User {
2272 id: i32,
2273 name: String,
2274 }
2275 let payload = serde_json::json!({"id": 1, "name": "Bob"});
2276 let resp = Response {
2277 status: reqwest::StatusCode::OK,
2278 headers: HeaderMap::new(),
2279 body: Bytes::from(serde_json::to_vec(&payload).unwrap()),
2280 url: None,
2281 };
2282 let user: User = resp.json().unwrap();
2283 assert_eq!(user.id, 1);
2284 assert_eq!(user.name, "Bob");
2285 }
2286
2287 #[test]
2289 fn response_text_returns_string() {
2290 let resp = Response {
2291 status: reqwest::StatusCode::OK,
2292 headers: HeaderMap::new(),
2293 body: Bytes::from_static(b"hello world"),
2294 url: None,
2295 };
2296 assert_eq!(resp.text(), "hello world");
2297 }
2298
2299 #[test]
2301 fn response_bytes_returns_raw() {
2302 let resp = Response {
2303 status: reqwest::StatusCode::CREATED,
2304 headers: HeaderMap::new(),
2305 body: Bytes::from_static(b"\x00\x01\x02"),
2306 url: None,
2307 };
2308 assert_eq!(resp.bytes(), Bytes::from_static(b"\x00\x01\x02"));
2309 }
2310
2311 #[test]
2313 fn config_deserialises_from_toml() {
2314 let toml = r#"
2316 [client]
2317 timeout_secs = 60
2318 max_retries = 5
2319 [client.base_urls]
2320 stripe = "https://api.stripe.com"
2321 sendgrid = "https://api.sendgrid.com"
2322 "#;
2323 let http_cfg: crate::config::HttpConfig = toml::from_str(toml).unwrap();
2324 let config = &http_cfg.client;
2325 assert_eq!(config.timeout_secs, 60);
2326 assert_eq!(config.max_retries, 5);
2327 assert_eq!(
2328 config.base_urls.get("stripe").map(String::as_str),
2329 Some("https://api.stripe.com")
2330 );
2331 assert_eq!(
2332 config.base_urls.get("sendgrid").map(String::as_str),
2333 Some("https://api.sendgrid.com")
2334 );
2335 }
2336
2337 #[test]
2339 fn config_has_correct_defaults() {
2340 let config = HttpClientConfig::default();
2341 assert_eq!(config.timeout_secs, 30);
2342 assert_eq!(config.max_retries, 3);
2343 assert!(config.base_urls.is_empty());
2344 }
2345
2346 #[test]
2348 fn idempotent_method_classification() {
2349 assert!(is_idempotent_method(&Method::GET));
2350 assert!(is_idempotent_method(&Method::HEAD));
2351 assert!(is_idempotent_method(&Method::PUT));
2352 assert!(is_idempotent_method(&Method::DELETE));
2353 assert!(is_idempotent_method(&Method::OPTIONS));
2354 assert!(is_idempotent_method(&Method::TRACE));
2355 assert!(!is_idempotent_method(&Method::POST));
2356 assert!(!is_idempotent_method(&Method::PATCH));
2357 }
2358
2359 #[test]
2361 fn retryable_status_classification() {
2362 assert!(is_retryable_status(502));
2363 assert!(is_retryable_status(503));
2364 assert!(is_retryable_status(504));
2365 assert!(!is_retryable_status(200));
2366 assert!(!is_retryable_status(400));
2367 assert!(!is_retryable_status(404));
2368 assert!(!is_retryable_status(500));
2369 assert!(!is_retryable_status(429));
2370 }
2371
2372 #[test]
2374 fn retry_after_header_parsing() {
2375 let mut headers = HeaderMap::new();
2376 headers.insert(
2377 reqwest::header::HeaderName::from_static("retry-after"),
2378 HeaderValue::from_static("5"),
2379 );
2380 assert_eq!(parse_retry_after(&headers), Some(Duration::from_secs(5)));
2381
2382 let empty = HeaderMap::new();
2383 assert_eq!(parse_retry_after(&empty), None);
2384 }
2385
2386 #[test]
2388 fn sensitive_header_detection() {
2389 assert!(is_sensitive_header("authorization"));
2390 assert!(is_sensitive_header("Authorization"));
2391 assert!(is_sensitive_header("AUTHORIZATION"));
2392 assert!(is_sensitive_header("cookie"));
2393 assert!(is_sensitive_header("set-cookie"));
2394 assert!(!is_sensitive_header("content-type"));
2395 assert!(!is_sensitive_header("x-api-key"));
2396 }
2397
2398 #[tokio::test]
2400 async fn mock_registry_captures_calls() {
2401 let registry = Arc::new(MockRegistry::new());
2402 let call_count = Arc::new(AtomicUsize::new(0));
2403
2404 registry.register(MockEntry {
2405 method: Some(Method::POST),
2406 path: "/charges".to_owned(),
2407 alias: Some("stripe".to_owned()),
2408 status: 200,
2409 body: Some(serde_json::json!({"id": "ch_123"})),
2410 call_count: call_count.clone(),
2411 });
2412
2413 let client = Client::new().with_mock(registry).named("stripe");
2414
2415 let resp = client
2416 .post("https://api.stripe.com/charges")
2417 .json(&serde_json::json!({"amount": 1000}))
2418 .send()
2419 .await
2420 .unwrap();
2421
2422 assert_eq!(resp.status().as_u16(), 200);
2423 let body: serde_json::Value = resp.json().unwrap();
2424 assert_eq!(body["id"], "ch_123");
2425 assert_eq!(call_count.load(Ordering::SeqCst), 1);
2426 }
2427
2428 #[tokio::test]
2430 async fn mock_handle_expect_called_passes() {
2431 let registry = Arc::new(MockRegistry::new());
2432 let call_count = Arc::new(AtomicUsize::new(0));
2433
2434 registry.register(MockEntry {
2435 method: Some(Method::GET),
2436 path: "/users/1".to_owned(),
2437 alias: None,
2438 status: 200,
2439 body: Some(serde_json::json!({"name": "Alice"})),
2440 call_count: call_count.clone(),
2441 });
2442
2443 let handle = MockHandle {
2444 alias: "test".to_owned(),
2445 method: "GET".to_owned(),
2446 path: "/users/1".to_owned(),
2447 call_count: call_count.clone(),
2448 };
2449
2450 let client = Client::new().with_mock(registry);
2451 client
2452 .get("https://api.example.com/users/1")
2453 .send()
2454 .await
2455 .unwrap();
2456
2457 handle.expect_called(1);
2458 assert_eq!(handle.call_count(), 1);
2459 }
2460
2461 #[tokio::test]
2463 async fn mock_matches_by_path_suffix() {
2464 let registry = Arc::new(MockRegistry::new());
2465 let call_count = Arc::new(AtomicUsize::new(0));
2466
2467 registry.register(MockEntry {
2468 method: Some(Method::POST),
2469 path: "/v1/charges".to_owned(),
2470 alias: None,
2471 status: 201,
2472 body: Some(serde_json::json!({"created": true})),
2473 call_count: call_count.clone(),
2474 });
2475
2476 let client = Client::new().with_mock(registry);
2477 let resp = client
2478 .post("https://api.stripe.com/v1/charges")
2479 .send()
2480 .await
2481 .unwrap();
2482
2483 assert_eq!(resp.status().as_u16(), 201);
2484 assert_eq!(call_count.load(Ordering::SeqCst), 1);
2485 }
2486
2487 #[tokio::test]
2489 async fn no_mock_error_when_unmatched() {
2490 let registry = Arc::new(MockRegistry::new());
2491 let client = Client::new().with_mock(registry);
2492 let result = client.post("https://api.example.com/unknown").send().await;
2493 assert!(matches!(result, Err(ClientError::NoMock(_, _))));
2494 }
2495
2496 #[tokio::test]
2498 async fn mock_setup_builder_registers_entry() {
2499 let registry = Arc::new(MockRegistry::new());
2500 let builder = MockSetupBuilder {
2501 registry: registry.clone(),
2502 alias: "myservice".to_owned(),
2503 method: None,
2504 path: None,
2505 };
2506
2507 let handle = builder
2508 .post("/api/resource")
2509 .respond_with(201, serde_json::json!({"ok": true}));
2510
2511 let client = Client::new().with_mock(registry).named("myservice");
2512 client
2513 .post("https://myservice.example.com/api/resource")
2514 .send()
2515 .await
2516 .unwrap();
2517
2518 handle.expect_called(1);
2519 }
2520
2521 #[test]
2523 fn client_from_config() {
2524 let config = HttpClientConfig {
2525 timeout_secs: 10,
2526 max_retries: 1,
2527 max_retry_after_secs: 10,
2528 base_urls: std::collections::HashMap::new(),
2529 };
2530 let client = Client::from_config(&config);
2531 assert_eq!(client.retry_policy.max_retries, 1);
2532 }
2533
2534 #[test]
2536 fn named_client_preserves_mock_registry() {
2537 let registry = Arc::new(MockRegistry::new());
2538 let client = Client::new().with_mock(registry);
2539 let named = client.named("stripe");
2540 assert!(named.mock.is_some());
2541 assert_eq!(named.alias.as_deref(), Some("stripe"));
2542 }
2543
2544 #[test]
2546 fn base_url_prepended_to_relative_path() {
2547 let client = Client::new();
2548 let client = client.with_base_url("https://api.stripe.com");
2549 let builder = client.post("/v1/charges");
2550 assert_eq!(builder.url, "https://api.stripe.com/v1/charges");
2551 }
2552
2553 #[test]
2555 fn absolute_url_bypasses_base_url() {
2556 let client = Client::new().with_base_url("https://ignored.example.com");
2557 let builder = client.get("https://actual.example.com/path");
2558 assert_eq!(builder.url, "https://actual.example.com/path");
2559 }
2560
2561 #[test]
2563 fn retry_override_per_request() {
2564 let client = Client::new(); let builder = client.get("https://example.com").retries(0);
2566 assert_eq!(builder.retry_policy.max_retries, 0);
2567
2568 let no_retry = client.get("https://example.com").no_retry();
2569 assert_eq!(no_retry.retry_policy.max_retries, 0);
2570 }
2571
2572 #[tokio::test]
2574 async fn client_extracts_from_state() {
2575 use axum::extract::FromRequestParts;
2576 let state = crate::AppState::for_test();
2577 let mut parts = axum::http::Request::new(axum::body::Body::empty())
2578 .into_parts()
2579 .0;
2580 let client = Client::from_request_parts(&mut parts, &state)
2581 .await
2582 .unwrap();
2583 assert!(client.mock.is_none());
2585 assert!(client.alias.is_none());
2586 }
2587
2588 #[test]
2590 fn mock_registry_ext_round_trips_through_state() {
2591 let registry = Arc::new(MockRegistry::new());
2592 let ext = HttpMockRegistryExt(registry);
2593 let state = crate::AppState::for_test();
2594 state.insert_extension(ext);
2595 let retrieved = state.extension::<HttpMockRegistryExt>();
2596 assert!(retrieved.is_some());
2597 }
2598
2599 #[test]
2601 fn named_client_resolves_base_url_from_config() {
2602 let mut base_urls = std::collections::HashMap::new();
2603 base_urls.insert("stripe".to_owned(), "https://api.stripe.com".to_owned());
2604 let config = HttpClientConfig {
2605 timeout_secs: 30,
2606 max_retries: 3,
2607 max_retry_after_secs: 10,
2608 base_urls,
2609 };
2610 let client = Client::from_config(&config);
2611 let stripe = client.named("stripe");
2612 assert_eq!(stripe.base_url.as_deref(), Some("https://api.stripe.com"));
2613 assert_eq!(stripe.alias.as_deref(), Some("stripe"));
2614
2615 let other = client.named("sendgrid");
2617 assert!(other.base_url.is_none());
2618 }
2619
2620 #[tokio::test]
2622 async fn client_extracts_from_autumn_config_in_state() {
2623 use axum::extract::FromRequestParts;
2624 let mut cfg = crate::config::AutumnConfig::default();
2625 cfg.http.client.max_retries = 7;
2626 let state = crate::AppState::for_test();
2627 state.insert_extension(cfg);
2628
2629 let mut parts = axum::http::Request::new(axum::body::Body::empty())
2630 .into_parts()
2631 .0;
2632 let client = Client::from_request_parts(&mut parts, &state)
2633 .await
2634 .unwrap();
2635 assert_eq!(client.retry_policy.max_retries, 7);
2636 }
2637
2638 #[tokio::test]
2640 async fn respond_with_status_produces_empty_body() {
2641 let registry = Arc::new(MockRegistry::new());
2642 let builder = MockSetupBuilder {
2643 registry: registry.clone(),
2644 alias: "svc".to_owned(),
2645 method: None,
2646 path: None,
2647 };
2648 let _handle = builder.delete("/items/1").respond_with_status(204);
2649
2650 let client = Client::new().with_mock(registry).named("svc");
2651 let resp = client
2652 .delete("https://svc.example.com/items/1")
2653 .send()
2654 .await
2655 .unwrap();
2656
2657 assert_eq!(resp.status().as_u16(), 204);
2658 assert_eq!(
2659 resp.bytes(),
2660 bytes::Bytes::new(),
2661 "body must be empty, not \"null\""
2662 );
2663 }
2664
2665 #[test]
2667 fn retry_after_http_date_parsing() {
2668 let mut headers = HeaderMap::new();
2669 headers.insert(
2671 reqwest::header::HeaderName::from_static("retry-after"),
2672 HeaderValue::from_static("Tue, 01 Jan 2030 00:00:00 GMT"),
2673 );
2674 let duration = parse_retry_after(&headers);
2675 assert!(duration.is_some(), "should parse HTTP-date Retry-After");
2676 assert!(
2677 duration.unwrap().as_secs() > 0,
2678 "future date should yield positive delay"
2679 );
2680 }
2681
2682 #[tokio::test]
2684 async fn non_idempotent_post_no_retry() {
2685 let registry = Arc::new(MockRegistry::new());
2686 let call_count = Arc::new(AtomicUsize::new(0));
2687 registry.register(MockEntry {
2688 method: Some(Method::POST),
2689 path: "/endpoint".to_owned(),
2690 alias: None,
2691 status: 503,
2692 body: None,
2693 call_count: call_count.clone(),
2694 });
2695
2696 let client = Client::new().with_mock(registry);
2698 let resp = client
2699 .post("https://example.com/endpoint")
2700 .send()
2701 .await
2702 .unwrap();
2703
2704 assert_eq!(resp.status().as_u16(), 503);
2705 assert_eq!(call_count.load(Ordering::SeqCst), 1);
2707 }
2708
2709 #[tokio::test]
2711 async fn mock_strips_query_from_url_before_matching() {
2712 let registry = Arc::new(MockRegistry::new());
2713 let call_count = Arc::new(AtomicUsize::new(0));
2714 registry.register(MockEntry {
2715 method: Some(Method::GET),
2716 path: "/v1/charges".to_owned(),
2717 alias: None,
2718 status: 200,
2719 body: Some(serde_json::json!({"ok": true})),
2720 call_count: call_count.clone(),
2721 });
2722
2723 let client = Client::new().with_mock(registry);
2725 let resp = client
2726 .get("https://api.stripe.com/v1/charges?expand[]=balance_transaction")
2727 .send()
2728 .await
2729 .unwrap();
2730
2731 assert_eq!(resp.status().as_u16(), 200);
2732 assert_eq!(call_count.load(Ordering::SeqCst), 1);
2733 }
2734
2735 #[tokio::test]
2737 async fn mock_suffix_match_with_leading_slash_path() {
2738 let registry = Arc::new(MockRegistry::new());
2739 let call_count = Arc::new(AtomicUsize::new(0));
2740 registry.register(MockEntry {
2742 method: Some(Method::POST),
2743 path: "/charges".to_owned(),
2744 alias: None,
2745 status: 201,
2746 body: Some(serde_json::json!({"matched": true})),
2747 call_count: call_count.clone(),
2748 });
2749
2750 let client = Client::new().with_mock(registry);
2751 let resp = client
2753 .post("https://api.stripe.com/v1/charges")
2754 .send()
2755 .await
2756 .unwrap();
2757
2758 assert_eq!(resp.status().as_u16(), 201);
2759 assert_eq!(call_count.load(Ordering::SeqCst), 1);
2760 }
2761
2762 #[test]
2764 fn retries_clears_idempotent_only_flag() {
2765 let client = Client::new();
2766 let builder = client.post("https://example.com").retries(2);
2767 assert_eq!(builder.retry_policy.max_retries, 2);
2768 assert!(
2769 !builder.retry_policy.retry_idempotent_only,
2770 "explicit retries() call must allow non-idempotent methods to retry"
2771 );
2772 }
2773
2774 #[test]
2776 fn log_request_completes_with_sensitive_headers() {
2777 let url = reqwest::Url::parse("https://api.example.com/v1/resource?q=1").unwrap();
2778 let mut headers = HeaderMap::new();
2779 headers.insert(
2780 HeaderName::from_static("content-type"),
2781 HeaderValue::from_static("application/json"),
2782 );
2783 headers.insert(
2784 HeaderName::from_static("authorization"),
2785 HeaderValue::from_static("Bearer sk_test_xxx"),
2786 );
2787 log_request("POST", &url, 201, Duration::from_millis(12), &headers);
2789 }
2790
2791 #[test]
2793 fn inject_trace_context_passthrough_without_telemetry() {
2794 let inner = reqwest::Client::new();
2795 let builder = inner.get("https://example.com");
2796 let _b = inject_trace_context(builder);
2798 }
2799
2800 #[tokio::test]
2803 #[allow(clippy::await_holding_lock)]
2804 async fn real_get_request_covers_network_path() {
2805 use axum::{Router, routing::get};
2806
2807 let _lock = crate::circuit_breaker::TEST_LOCK
2808 .lock()
2809 .unwrap_or_else(std::sync::PoisonError::into_inner);
2810 crate::circuit_breaker::global_registry().clear();
2811
2812 let app = Router::new().route("/ping", get(|| async { "pong" }));
2813 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
2814 let addr = listener.local_addr().unwrap();
2815 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
2816
2817 let client = Client::new();
2818 let resp = client
2819 .get(format!("http://127.0.0.1:{}/ping", addr.port()))
2820 .header("x-request-id", "test-35")
2821 .send()
2822 .await
2823 .unwrap();
2824
2825 assert_eq!(resp.status().as_u16(), 200);
2826 assert!(resp.url().is_some());
2827 assert_eq!(resp.text(), "pong");
2828
2829 crate::circuit_breaker::global_registry().clear();
2830 }
2831
2832 #[tokio::test]
2834 #[allow(clippy::await_holding_lock)]
2835 async fn real_post_with_json_body_covers_body_path() {
2836 use axum::{Json, Router, routing::post};
2837 use serde_json::Value;
2838
2839 let _lock = crate::circuit_breaker::TEST_LOCK
2840 .lock()
2841 .unwrap_or_else(std::sync::PoisonError::into_inner);
2842 crate::circuit_breaker::global_registry().clear();
2843
2844 let app = Router::new().route(
2845 "/echo",
2846 post(|Json(body): Json<Value>| async move { Json(body) }),
2847 );
2848 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
2849 let addr = listener.local_addr().unwrap();
2850 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
2851
2852 let client = Client::new();
2853 let resp = client
2854 .post(format!("http://127.0.0.1:{}/echo", addr.port()))
2855 .json(&serde_json::json!({"hello": "world"}))
2856 .send()
2857 .await
2858 .unwrap();
2859
2860 assert_eq!(resp.status().as_u16(), 200);
2861 let body: Value = resp.json().unwrap();
2862 assert_eq!(body["hello"], "world");
2863
2864 crate::circuit_breaker::global_registry().clear();
2865 }
2866
2867 #[tokio::test]
2869 #[allow(clippy::await_holding_lock)]
2870 async fn real_get_retries_on_503_then_succeeds() {
2871 use axum::{Router, routing::get};
2872 use std::sync::Arc;
2873 use std::sync::atomic::{AtomicU32, Ordering as SeqOrdering};
2874
2875 let _lock = crate::circuit_breaker::TEST_LOCK
2876 .lock()
2877 .unwrap_or_else(std::sync::PoisonError::into_inner);
2878 crate::circuit_breaker::global_registry().clear();
2879
2880 let hit = Arc::new(AtomicU32::new(0));
2881 let hit2 = hit.clone();
2882 let app = Router::new().route(
2883 "/flaky",
2884 get(move || {
2885 let c = hit2.clone();
2886 async move {
2887 if c.fetch_add(1, SeqOrdering::SeqCst) == 0 {
2888 axum::http::StatusCode::SERVICE_UNAVAILABLE
2889 } else {
2890 axum::http::StatusCode::OK
2891 }
2892 }
2893 }),
2894 );
2895 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
2896 let addr = listener.local_addr().unwrap();
2897 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
2898
2899 let resp = Client::new()
2901 .get(format!("http://127.0.0.1:{}/flaky", addr.port()))
2902 .retries(1)
2903 .send()
2904 .await
2905 .unwrap();
2906
2907 assert_eq!(resp.status().as_u16(), 200);
2908 assert_eq!(hit.load(SeqOrdering::SeqCst), 2);
2909
2910 crate::circuit_breaker::global_registry().clear();
2911 }
2912
2913 #[test]
2915 fn text_body_sets_body() {
2916 let client = Client::new();
2917 let builder = client.post("https://example.com").text_body("hello");
2918 assert_eq!(builder.body, Some(bytes::Bytes::from_static(b"hello")));
2919 }
2920
2921 #[test]
2923 fn client_error_display() {
2924 let err = ClientError::NoMock("GET".to_owned(), "/path".to_owned());
2925 assert!(err.to_string().contains("GET"));
2926 assert!(err.to_string().contains("/path"));
2927 }
2928
2929 #[tokio::test]
2931 #[allow(clippy::await_holding_lock)]
2932 async fn test_http_client_circuit_breaker_integration() {
2933 use axum::{Router, routing::get};
2934 use std::sync::atomic::{AtomicU32, Ordering as SeqOrdering};
2935
2936 let _lock = crate::circuit_breaker::TEST_LOCK
2937 .lock()
2938 .unwrap_or_else(std::sync::PoisonError::into_inner);
2939 crate::circuit_breaker::global_registry().clear();
2940
2941 let hit = Arc::new(AtomicU32::new(0));
2942 let hit2 = hit.clone();
2943 let app = Router::new().route(
2944 "/flaky",
2945 get(move || {
2946 let c = hit2.clone();
2947 async move {
2948 c.fetch_add(1, SeqOrdering::SeqCst);
2949 axum::http::StatusCode::INTERNAL_SERVER_ERROR
2950 }
2951 }),
2952 );
2953 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
2954 let addr = listener.local_addr().unwrap();
2955 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
2956
2957 let mut rc = crate::config::ResilienceConfig::default();
2959 rc.circuit_breaker.defaults.failure_ratio_threshold = Some(0.5);
2960 rc.circuit_breaker.defaults.minimum_sample_count = Some(3);
2961 rc.circuit_breaker.defaults.open_duration_secs = Some(10);
2962
2963 let client = Client::new();
2964 let client = Client {
2966 resilience_config: Some(Arc::new(rc)),
2967 ..client
2968 };
2969
2970 let url = format!("http://127.0.0.1:{}/flaky", addr.port());
2971
2972 for _ in 0..3 {
2974 let res = client.get(&url).send().await;
2975 let res = res.unwrap();
2976 assert_eq!(res.status().as_u16(), 500);
2977 }
2978
2979 let res = client.get(&url).send().await;
2981 assert!(matches!(res, Err(ClientError::CircuitBreakerOpen)));
2982
2983 assert_eq!(hit.load(SeqOrdering::SeqCst), 3);
2985 crate::circuit_breaker::global_registry().clear();
2986 }
2987
2988 #[test]
2990 fn shared_reqwest_client_ext_round_trips() {
2991 let ext = SharedReqwestClient {
2992 client: reqwest::Client::new(),
2993 timeout_secs: 30,
2994 };
2995 let state = crate::AppState::for_test();
2996 state.insert_extension(ext);
2997 let retrieved = state.extension::<SharedReqwestClient>();
2998 assert!(retrieved.is_some());
2999 }
3000
3001 #[test]
3003 fn client_head_method_builds_request_builder() {
3004 let client = Client::new();
3005 let _builder = client.head("https://example.com/resource");
3006 }
3007
3008 #[tokio::test]
3014 #[allow(clippy::await_holding_lock)]
3015 async fn from_state_reuses_shared_client() {
3016 use axum::{Router, routing::get};
3017
3018 let _lock = crate::circuit_breaker::TEST_LOCK
3019 .lock()
3020 .unwrap_or_else(std::sync::PoisonError::into_inner);
3021 crate::circuit_breaker::global_registry().clear();
3022
3023 let app = Router::new().route(
3024 "/ua",
3025 get(|req: axum::http::Request<axum::body::Body>| async move {
3026 req.headers()
3027 .get("user-agent")
3028 .and_then(|v| v.to_str().ok())
3029 .unwrap_or("")
3030 .to_owned()
3031 }),
3032 );
3033 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
3034 let addr = listener.local_addr().unwrap();
3035 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
3036
3037 let distinctive_inner = reqwest::ClientBuilder::new()
3038 .user_agent("autumn-shared-pool-test")
3039 .build()
3040 .expect("failed to build inner client");
3041 let state = crate::AppState::for_test();
3042 state.insert_extension(SharedReqwestClient {
3043 client: distinctive_inner,
3044 timeout_secs: 30,
3045 });
3046
3047 let client = Client::from_state(&state);
3048 let resp = client
3049 .get(format!("http://127.0.0.1:{}/ua", addr.port()))
3050 .send()
3051 .await
3052 .expect("request should succeed");
3053
3054 assert_eq!(resp.text(), "autumn-shared-pool-test");
3055 crate::circuit_breaker::global_registry().clear();
3056 }
3057
3058 #[cfg(feature = "http-client")]
3059 #[test]
3060 fn from_state_falls_back_when_timeout_mismatches_shared_client() {
3061 use crate::config::{AutumnConfig, HttpClientConfig};
3065 use std::sync::Arc;
3066
3067 let mut config = AutumnConfig::default();
3068 config.http.client = HttpClientConfig {
3069 timeout_secs: 10,
3070 ..Default::default()
3071 };
3072
3073 let state = crate::AppState::for_test();
3074 state.insert_extension(SharedReqwestClient {
3075 client: reqwest::Client::new(),
3076 timeout_secs: 5, });
3078 state.insert_extension(Arc::new(config));
3079
3080 let _client = Client::from_state(&state);
3082 }
3083
3084 #[cfg(feature = "http-client")]
3085 #[test]
3086 fn from_state_reuses_shared_client_when_no_config() {
3087 let state = crate::AppState::for_test();
3090 let default_timeout = crate::config::HttpClientConfig::default().timeout_secs;
3093 state.insert_extension(SharedReqwestClient {
3094 client: reqwest::Client::new(),
3095 timeout_secs: default_timeout,
3096 });
3097
3098 let _client = Client::from_state(&state);
3099 }
3100
3101 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
3104
3105 #[test]
3107 fn ssrf_policy_blocks_private_and_reserved_ipv4() {
3108 let blocked = [
3109 "0.0.0.0",
3110 "10.1.2.3",
3111 "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", ];
3126 for s in blocked {
3127 let ip: IpAddr = s.parse().unwrap();
3128 assert!(is_blocked_ip(ip), "{s} should be blocked");
3129 assert!(!is_public_ip(ip), "{s} should not be public");
3130 }
3131 }
3132
3133 #[test]
3135 fn ssrf_policy_allows_public_ipv4() {
3136 for s in ["1.1.1.1", "8.8.8.8", "93.184.216.34"] {
3137 let ip: IpAddr = s.parse().unwrap();
3138 assert!(is_public_ip(ip), "{s} should be public");
3139 assert!(!is_blocked_ip(ip), "{s} should not be blocked");
3140 }
3141 }
3142
3143 #[test]
3145 fn ssrf_policy_ipv6_and_mapped_forms() {
3146 let blocked = [
3147 "::", "::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", ];
3170 for s in blocked {
3171 let ip: IpAddr = s.parse().unwrap();
3172 assert!(is_blocked_ip(ip), "{s} should be blocked");
3173 }
3174 let compat = IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0x7f00, 0x0001));
3176 assert!(
3177 is_blocked_ip(compat),
3178 "::7f00:1 (127.0.0.1) should be blocked"
3179 );
3180
3181 let public: IpAddr = "2606:4700:4700::1111".parse().unwrap();
3183 assert!(
3184 is_public_ip(public),
3185 "2606:4700:4700::1111 should be public"
3186 );
3187
3188 let public_addrs = [
3191 "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", ];
3201 for s in public_addrs {
3202 let ip: IpAddr = s.parse().unwrap();
3203 assert!(is_public_ip(ip), "{s} should be public");
3204 assert!(!is_blocked_ip(ip), "{s} should not be blocked");
3205 }
3206 }
3207
3208 #[test]
3212 fn ssrf_policy_blocks_tunnelled_ipv4() {
3213 let blocked = [
3214 "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", ];
3229 for s in blocked {
3230 let ip: IpAddr = s.parse().unwrap();
3231 assert!(is_blocked_ip(ip), "{s} should be blocked");
3232 assert!(!is_public_ip(ip), "{s} should not be public");
3233 }
3234
3235 let anycast: IpAddr = "192.88.99.1".parse().unwrap();
3237 assert!(is_blocked_ip(anycast), "192.88.99.1 should be blocked");
3238
3239 for s in ["2002:0808:0808::", "2606:4700::1111", "::ffff:0:8.8.8.8"] {
3246 let ip: IpAddr = s.parse().unwrap();
3247 assert!(is_public_ip(ip), "{s} should be public");
3248 assert!(!is_blocked_ip(ip), "{s} should not be blocked");
3249 }
3250 }
3251
3252 #[test]
3255 fn ssrf_policy_decimal_encoded_host_is_blocked() {
3256 for raw in [
3257 "http://2130706433/",
3258 "http://0x7f000001/",
3259 "http://127.0.0.1/",
3260 ] {
3261 let parsed = url::Url::parse(raw).unwrap();
3262 match parsed.host() {
3263 Some(url::Host::Ipv4(v4)) => {
3264 assert_eq!(
3265 v4,
3266 Ipv4Addr::LOCALHOST,
3267 "{raw} should normalise to 127.0.0.1"
3268 );
3269 assert!(
3270 is_blocked_ip(IpAddr::V4(v4)),
3271 "{raw} host should be blocked"
3272 );
3273 }
3274 other => panic!("{raw} did not parse to an Ipv4 host: {other:?}"),
3275 }
3276 }
3277 }
3278
3279 async fn spawn(app: axum::Router) -> std::net::SocketAddr {
3281 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
3282 let addr = listener.local_addr().unwrap();
3283 tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
3284 addr
3285 }
3286
3287 fn redirect_302(location: String) -> axum::response::Response {
3288 axum::response::Response::builder()
3289 .status(302)
3290 .header("location", location)
3291 .body(axum::body::Body::empty())
3292 .unwrap()
3293 }
3294
3295 fn redirect_307(location: String) -> axum::response::Response {
3296 axum::response::Response::builder()
3297 .status(307)
3298 .header("location", location)
3299 .body(axum::body::Body::empty())
3300 .unwrap()
3301 }
3302
3303 fn response_with_location(status: u16, location: String) -> axum::response::Response {
3307 axum::response::Response::builder()
3308 .status(status)
3309 .header("location", location)
3310 .body(axum::body::Body::empty())
3311 .unwrap()
3312 }
3313
3314 #[tokio::test]
3316 async fn no_redirect_returns_3xx_unfollowed() {
3317 use axum::{Router, routing::get};
3318 let addr = spawn(Router::new().route(
3319 "/start",
3320 get(|| async { redirect_302("http://127.0.0.1:1/never".to_owned()) }),
3321 ))
3322 .await;
3323
3324 let resp = Client::new()
3325 .get(format!("http://127.0.0.1:{}/start", addr.port()))
3326 .no_redirect()
3327 .send()
3328 .await
3329 .unwrap();
3330
3331 assert_eq!(resp.status().as_u16(), 302);
3332 assert_eq!(
3333 resp.headers().get("location").and_then(|v| v.to_str().ok()),
3334 Some("http://127.0.0.1:1/never")
3335 );
3336 }
3337
3338 #[tokio::test]
3352 async fn no_redirect_returns_3xx_with_malformed_location() {
3353 use axum::{Router, routing::get};
3354 let addr = spawn(Router::new().route(
3355 "/start",
3356 get(|| async { redirect_302("ht!tp://\\bad".to_owned()) }),
3359 ))
3360 .await;
3361
3362 let resp = Client::new()
3363 .get(format!("http://127.0.0.1:{}/start", addr.port()))
3364 .no_redirect()
3365 .send()
3366 .await
3367 .expect("no_redirect() must return the 3xx even with a malformed Location");
3368
3369 assert_eq!(resp.status().as_u16(), 302);
3370 assert_eq!(
3371 resp.headers().get("location").and_then(|v| v.to_str().ok()),
3372 Some("ht!tp://\\bad")
3373 );
3374 }
3375
3376 #[tokio::test]
3378 async fn follow_redirects_valid_chain_calls_validator() {
3379 use axum::{Router, routing::get};
3380
3381 let b_addr = spawn(Router::new().route("/final", get(|| async { "final-body" }))).await;
3382 let b_port = b_addr.port();
3383 let a_addr = spawn(Router::new().route(
3384 "/start",
3385 get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/final")) }),
3386 ))
3387 .await;
3388
3389 let calls = Arc::new(AtomicUsize::new(0));
3390 let calls2 = calls.clone();
3391 let resp = Client::new()
3392 .get(format!("http://127.0.0.1:{}/start", a_addr.port()))
3393 .follow_redirects(5, move |_loc| {
3394 calls2.fetch_add(1, Ordering::SeqCst);
3395 true
3396 })
3397 .send()
3398 .await
3399 .unwrap();
3400
3401 assert_eq!(resp.status().as_u16(), 200);
3402 assert_eq!(resp.text(), "final-body");
3403 assert_eq!(
3404 calls.load(Ordering::SeqCst),
3405 1,
3406 "validator called once per hop"
3407 );
3408 }
3409
3410 #[tokio::test]
3414 async fn follow_redirects_rejects_private_target() {
3415 use axum::{Router, routing::get};
3416 use std::sync::atomic::AtomicBool;
3417
3418 let touched = Arc::new(AtomicBool::new(false));
3420 let touched2 = touched.clone();
3421 let priv_addr = spawn(Router::new().route(
3422 "/secret",
3423 get(move || {
3424 let t = touched2.clone();
3425 async move {
3426 t.store(true, Ordering::SeqCst);
3427 "SECRET"
3428 }
3429 }),
3430 ))
3431 .await;
3432 let priv_port = priv_addr.port();
3433
3434 let a_addr = spawn(Router::new().route(
3436 "/start",
3437 get(
3438 move || async move { redirect_302(format!("http://127.0.0.1:{priv_port}/secret")) },
3439 ),
3440 ))
3441 .await;
3442
3443 let validator = |u: &str| -> bool {
3444 let Ok(p) = url::Url::parse(u) else {
3445 return false;
3446 };
3447 match p.host() {
3448 Some(url::Host::Ipv4(v4)) => is_public_ip(IpAddr::V4(v4)),
3449 Some(url::Host::Ipv6(v6)) => is_public_ip(IpAddr::V6(v6)),
3450 _ => true,
3451 }
3452 };
3453
3454 let result = Client::new()
3455 .get(format!("http://127.0.0.1:{}/start", a_addr.port()))
3456 .follow_redirects(5, validator)
3457 .send()
3458 .await;
3459
3460 assert!(
3461 matches!(result, Err(ClientError::RedirectRejected(_))),
3462 "expected RedirectRejected, got {result:?}"
3463 );
3464 assert!(
3465 !touched.load(Ordering::SeqCst),
3466 "the private target must never be connected to"
3467 );
3468 }
3469
3470 #[tokio::test]
3472 async fn follow_redirects_cap_exceeded() {
3473 use axum::{Router, routing::get};
3474
3475 let addr =
3477 spawn(Router::new().route("/loop", get(|| async { redirect_302("/loop".to_owned()) })))
3478 .await;
3479
3480 let result = Client::new()
3481 .get(format!("http://127.0.0.1:{}/loop", addr.port()))
3482 .follow_redirects(2, |_| true)
3483 .send()
3484 .await;
3485
3486 assert!(
3487 matches!(result, Err(ClientError::TooManyRedirects(2))),
3488 "expected TooManyRedirects(2), got {result:?}"
3489 );
3490 }
3491
3492 #[tokio::test]
3494 async fn follow_redirects_zero_max_errors_on_first_3xx() {
3495 use axum::{Router, routing::get};
3496 let addr = spawn(Router::new().route(
3497 "/start",
3498 get(|| async { redirect_302("http://127.0.0.1:1/x".to_owned()) }),
3499 ))
3500 .await;
3501
3502 let result = Client::new()
3503 .get(format!("http://127.0.0.1:{}/start", addr.port()))
3504 .follow_redirects(0, |_| true)
3505 .send()
3506 .await;
3507
3508 assert!(matches!(result, Err(ClientError::TooManyRedirects(0))));
3509 }
3510
3511 #[tokio::test]
3519 async fn pin_to_bypasses_dns_and_uses_url_port() {
3520 use axum::{Router, routing::get};
3521 let addr = spawn(Router::new().route("/ping", get(|| async { "pong" }))).await;
3522 let listener_port = addr.port();
3523
3524 let resp = Client::new()
3525 .get(format!("http://pinned.invalid:{listener_port}/ping"))
3526 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 1))
3528 .send()
3529 .await
3530 .expect("pinned request should reach the loopback listener");
3531
3532 assert_eq!(resp.status().as_u16(), 200);
3533 assert_eq!(resp.text(), "pong");
3534 }
3535
3536 #[tokio::test]
3540 async fn get_ssrf_safe_rejects_loopback_host_before_connecting() {
3541 use axum::{Router, routing::get};
3542 use std::sync::atomic::AtomicBool;
3543
3544 let touched = Arc::new(AtomicBool::new(false));
3545 let touched2 = touched.clone();
3546 let addr = spawn(Router::new().route(
3547 "/x",
3548 get(move || {
3549 let t = touched2.clone();
3550 async move {
3551 t.store(true, Ordering::SeqCst);
3552 "reached"
3553 }
3554 }),
3555 ))
3556 .await;
3557
3558 let result = Client::new()
3559 .get_ssrf_safe(format!("http://localhost:{}/x", addr.port()))
3560 .send()
3561 .await;
3562
3563 assert!(
3564 matches!(result, Err(ClientError::SsrfBlocked(_))),
3565 "expected SsrfBlocked, got {result:?}"
3566 );
3567 assert!(
3568 !touched.load(Ordering::SeqCst),
3569 "SSRF guard must reject before any connection"
3570 );
3571 }
3572
3573 #[tokio::test]
3576 async fn get_ssrf_safe_rejects_decimal_encoded_loopback() {
3577 let result = Client::new()
3578 .get_ssrf_safe("http://2130706433/")
3579 .send()
3580 .await;
3581 assert!(
3582 matches!(result, Err(ClientError::SsrfBlocked(_))),
3583 "expected SsrfBlocked, got {result:?}"
3584 );
3585 }
3586
3587 #[tokio::test]
3594 async fn resolve_and_validate_accepts_public_rejects_blocked() {
3595 let ok = resolve_and_validate("http://8.8.8.8:8080/path")
3598 .await
3599 .unwrap();
3600 assert_eq!(
3601 ok,
3602 vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8)), 8080)]
3603 );
3604
3605 let ok_https = resolve_and_validate("https://1.1.1.1/").await.unwrap();
3607 assert_eq!(ok_https.len(), 1);
3608 assert_eq!(ok_https[0].port(), 443);
3609
3610 for raw in [
3612 "http://127.0.0.1/",
3613 "http://169.254.169.254/latest/meta-data/",
3614 "http://10.0.0.1/",
3615 "http://2130706433/",
3616 ] {
3617 let err = resolve_and_validate(raw).await;
3618 assert!(
3619 matches!(err, Err(ClientError::SsrfBlocked(_))),
3620 "{raw} should be SsrfBlocked, got {err:?}"
3621 );
3622 }
3623 }
3624
3625 #[test]
3632 fn validate_resolved_addrs_returns_all_public_rejects_any_blocked() {
3633 let v4 = |a, b, c, d, p| SocketAddr::new(IpAddr::V4(Ipv4Addr::new(a, b, c, d)), p);
3634
3635 let two = vec![v4(1, 1, 1, 1, 443), v4(8, 8, 8, 8, 443)];
3637 assert_eq!(validate_resolved_addrs(two.clone()).unwrap(), two);
3638
3639 let mixed = vec![v4(1, 1, 1, 1, 443), v4(10, 0, 0, 1, 443)];
3641 assert!(
3642 matches!(
3643 validate_resolved_addrs(mixed),
3644 Err(ClientError::SsrfBlocked(_))
3645 ),
3646 "a set containing a blocked address must be rejected"
3647 );
3648
3649 let v6 = SocketAddr::new(
3652 IpAddr::V6(Ipv6Addr::new(0x2606, 0x4700, 0x4700, 0, 0, 0, 0, 0x1111)),
3653 443,
3654 );
3655 let mix = vec![v6, v4(8, 8, 8, 8, 443)];
3656 assert_eq!(validate_resolved_addrs(mix.clone()).unwrap(), mix);
3657 }
3658
3659 #[tokio::test]
3664 async fn pin_to_does_not_follow_redirect_unpinned() {
3665 use axum::{Router, routing::get};
3666 use std::sync::atomic::AtomicBool;
3667
3668 let touched = Arc::new(AtomicBool::new(false));
3670 let touched2 = touched.clone();
3671 let onward = spawn(Router::new().route(
3672 "/onward",
3673 get(move || {
3674 let t = touched2.clone();
3675 async move {
3676 t.store(true, Ordering::SeqCst);
3677 "REACHED"
3678 }
3679 }),
3680 ))
3681 .await;
3682 let onward_port = onward.port();
3683
3684 let start =
3685 spawn(Router::new().route(
3686 "/start",
3687 get(move || async move {
3688 redirect_302(format!("http://127.0.0.1:{onward_port}/onward"))
3689 }),
3690 ))
3691 .await;
3692 let start_port = start.port();
3693
3694 let resp = Client::new()
3699 .get(format!("http://pinned.invalid:{start_port}/start"))
3700 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), start_port))
3701 .send()
3702 .await
3703 .expect("pinned request should return the 302 unfollowed");
3704
3705 assert_eq!(
3706 resp.status().as_u16(),
3707 302,
3708 "pin_to must return the redirect unfollowed"
3709 );
3710 assert!(
3711 !touched.load(Ordering::SeqCst),
3712 "pin_to must not silently follow the redirect onward"
3713 );
3714 }
3715
3716 #[tokio::test]
3719 async fn get_ssrf_safe_rejects_non_http_scheme() {
3720 for raw in ["ftp://public.example/resource", "gopher://public.example/"] {
3721 let result = Client::new().get_ssrf_safe(raw).send().await;
3722 assert!(
3723 matches!(result, Err(ClientError::InvalidUrl(_))),
3724 "{raw} should be rejected with InvalidUrl, got {result:?}"
3725 );
3726 }
3727 }
3728
3729 #[tokio::test]
3734 async fn follow_redirects_strips_sensitive_headers_cross_origin() {
3735 use axum::{Router, routing::get};
3736
3737 let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
3738 let seen2 = seen.clone();
3739 let b_addr = spawn(Router::new().route(
3742 "/dst",
3743 get(move |headers: HeaderMap| {
3744 let slot = seen2.clone();
3745 async move {
3746 *slot.lock().unwrap() = Some(headers);
3747 "ok"
3748 }
3749 }),
3750 ))
3751 .await;
3752 let b_port = b_addr.port();
3753
3754 let a_addr = spawn(Router::new().route(
3756 "/",
3757 get(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
3758 ))
3759 .await;
3760
3761 let resp = Client::new()
3762 .get(format!("http://127.0.0.1:{}/", a_addr.port()))
3763 .header("authorization", "secret")
3764 .header("cookie", "session=abc")
3765 .header("proxy-authorization", "Basic zzz")
3766 .follow_redirects(3, |_| true)
3767 .send()
3768 .await
3769 .unwrap();
3770
3771 assert_eq!(resp.status().as_u16(), 200);
3772 let headers = seen
3773 .lock()
3774 .unwrap()
3775 .clone()
3776 .expect("listener B must have been reached");
3777 assert!(
3778 headers.get("authorization").is_none(),
3779 "authorization must be stripped on a cross-origin redirect"
3780 );
3781 assert!(
3782 headers.get("cookie").is_none(),
3783 "cookie must be stripped on a cross-origin redirect"
3784 );
3785 assert!(
3786 headers.get("proxy-authorization").is_none(),
3787 "proxy-authorization must be stripped on a cross-origin redirect"
3788 );
3789 }
3790
3791 #[tokio::test]
3794 async fn follow_redirects_keeps_sensitive_headers_same_origin() {
3795 use axum::{Router, routing::get};
3796
3797 let seen: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
3798 let seen2 = seen.clone();
3799 let addr = spawn(
3800 Router::new()
3801 .route("/", get(|| async { redirect_302("/next".to_owned()) }))
3802 .route(
3803 "/next",
3804 get(move |headers: HeaderMap| {
3805 let slot = seen2.clone();
3806 async move {
3807 *slot.lock().unwrap() = Some(headers);
3808 "ok"
3809 }
3810 }),
3811 ),
3812 )
3813 .await;
3814
3815 let resp = Client::new()
3816 .get(format!("http://127.0.0.1:{}/", addr.port()))
3817 .header("authorization", "secret")
3818 .follow_redirects(3, |_| true)
3819 .send()
3820 .await
3821 .unwrap();
3822
3823 assert_eq!(resp.status().as_u16(), 200);
3824 let headers = seen
3825 .lock()
3826 .unwrap()
3827 .clone()
3828 .expect("/next must have been reached");
3829 assert_eq!(
3830 headers.get("authorization").and_then(|v| v.to_str().ok()),
3831 Some("secret"),
3832 "authorization must be preserved on a same-origin redirect"
3833 );
3834 }
3835
3836 #[tokio::test]
3839 async fn follow_redirects_302_post_becomes_get() {
3840 use axum::{
3841 Router,
3842 routing::{any, post},
3843 };
3844
3845 let seen: Arc<Mutex<Option<(String, HeaderMap, Bytes)>>> = Arc::new(Mutex::new(None));
3846 let seen2 = seen.clone();
3847 let b_addr = spawn(Router::new().route(
3849 "/dst",
3850 any(move |method: Method, headers: HeaderMap, body: Bytes| {
3851 let slot = seen2.clone();
3852 async move {
3853 *slot.lock().unwrap() = Some((method.to_string(), headers, body));
3854 "ok"
3855 }
3856 }),
3857 ))
3858 .await;
3859 let b_port = b_addr.port();
3860
3861 let a_addr = spawn(Router::new().route(
3862 "/",
3863 post(move || async move { redirect_302(format!("http://127.0.0.1:{b_port}/dst")) }),
3864 ))
3865 .await;
3866
3867 let resp = Client::new()
3872 .post(format!("http://127.0.0.1:{}/", a_addr.port()))
3873 .json(&serde_json::json!({"payload": true}))
3874 .follow_redirects(3, |_| true)
3875 .send()
3876 .await
3877 .unwrap();
3878
3879 assert_eq!(resp.status().as_u16(), 200);
3880 let (method, headers, body) = seen
3881 .lock()
3882 .unwrap()
3883 .clone()
3884 .expect("listener B must have been reached");
3885 assert_eq!(method, "GET", "302 must rewrite POST → GET");
3886 assert!(body.is_empty(), "302 POST→GET must drop the request body");
3887 assert!(
3888 !headers.contains_key(reqwest::header::CONTENT_TYPE),
3889 "302 POST→GET must drop the Content-Type payload header"
3890 );
3891 assert!(
3892 !headers.contains_key(reqwest::header::CONTENT_LENGTH),
3893 "302 POST→GET must drop the Content-Length payload header"
3894 );
3895 }
3896
3897 #[tokio::test]
3901 async fn follow_redirects_307_preserves_method_and_body() {
3902 use axum::{
3903 Router,
3904 routing::{any, post},
3905 };
3906
3907 let seen: Arc<Mutex<Option<(String, Bytes)>>> = Arc::new(Mutex::new(None));
3908 let seen2 = seen.clone();
3909 let b_addr = spawn(Router::new().route(
3910 "/dst",
3911 any(move |method: Method, body: Bytes| {
3912 let slot = seen2.clone();
3913 async move {
3914 *slot.lock().unwrap() = Some((method.to_string(), body));
3915 "ok"
3916 }
3917 }),
3918 ))
3919 .await;
3920 let b_port = b_addr.port();
3921
3922 let a_addr = spawn(Router::new().route(
3923 "/",
3924 post(move || async move { redirect_307(format!("http://127.0.0.1:{b_port}/dst")) }),
3925 ))
3926 .await;
3927
3928 let resp = Client::new()
3929 .post(format!("http://127.0.0.1:{}/", a_addr.port()))
3930 .text_body("payload")
3931 .follow_redirects(3, |_| true)
3932 .send()
3933 .await
3934 .unwrap();
3935
3936 assert_eq!(resp.status().as_u16(), 200);
3937 let (method, body) = seen
3938 .lock()
3939 .unwrap()
3940 .clone()
3941 .expect("listener B must have been reached");
3942 assert_eq!(method, "POST", "307 must preserve the POST method");
3943 assert_eq!(
3944 &body[..],
3945 b"payload",
3946 "307 must preserve the request body verbatim"
3947 );
3948 }
3949
3950 #[test]
3959 fn ssrf_redirect_plan_honours_chained_override() {
3960 let client = Client::new();
3961
3962 let default = client.get_ssrf_safe("https://example.com/");
3964 assert_eq!(
3965 default.ssrf_redirect_plan(),
3966 (true, SSRF_SAFE_MAX_REDIRECTS),
3967 "default SSRF-safe path follows up to SSRF_SAFE_MAX_REDIRECTS"
3968 );
3969
3970 let none = client.get_ssrf_safe("https://example.com/").no_redirect();
3972 let (follow, _max) = none.ssrf_redirect_plan();
3973 assert!(
3974 !follow,
3975 "no_redirect() must disable following on the safe path"
3976 );
3977
3978 let follow3 = client
3980 .get_ssrf_safe("https://example.com/")
3981 .follow_redirects(3, |_| true);
3982 assert_eq!(
3983 follow3.ssrf_redirect_plan(),
3984 (true, 3),
3985 "follow_redirects(3, ..) must cap the safe path at 3 hops"
3986 );
3987 }
3988
3989 #[tokio::test]
3994 async fn pin_then_follow_redirects_is_rejected() {
3995 let result = Client::new()
3996 .get("http://example.com/")
3997 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
3998 .follow_redirects(2, |_| true)
3999 .send()
4000 .await;
4001
4002 assert!(
4003 matches!(result, Err(ClientError::IncompatiblePinRedirect(_))),
4004 "expected IncompatiblePinRedirect, got {result:?}"
4005 );
4006 }
4007
4008 #[tokio::test]
4011 async fn follow_redirects_then_pin_is_rejected() {
4012 let result = Client::new()
4013 .get("http://example.com/")
4014 .follow_redirects(2, |_| true)
4015 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
4016 .send()
4017 .await;
4018
4019 assert!(
4020 matches!(result, Err(ClientError::IncompatiblePinRedirect(_))),
4021 "expected IncompatiblePinRedirect, got {result:?}"
4022 );
4023 }
4024
4025 #[tokio::test]
4028 async fn pin_with_no_redirect_is_allowed() {
4029 use axum::{Router, routing::get};
4030
4031 let addr = spawn(Router::new().route(
4032 "/start",
4033 get(|| async { redirect_302("http://127.0.0.1:1/onward".to_owned()) }),
4034 ))
4035 .await;
4036 let port = addr.port();
4037
4038 let resp = Client::new()
4041 .get(format!("http://pinned.invalid:{port}/start"))
4042 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port))
4043 .no_redirect()
4044 .send()
4045 .await
4046 .expect("pin_to + no_redirect must return the 3xx unfollowed, not error");
4047
4048 assert_eq!(
4049 resp.status().as_u16(),
4050 302,
4051 "pin_to + no_redirect returns the redirect verbatim"
4052 );
4053 }
4054
4055 #[tokio::test]
4062 async fn pin_to_ipv4_literal_host_is_rejected() {
4063 let result = Client::new()
4064 .get("http://198.51.100.1:8080/")
4065 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
4066 .send()
4067 .await;
4068
4069 assert!(
4070 matches!(result, Err(ClientError::PinRequiresDomainHost(_))),
4071 "expected PinRequiresDomainHost, got {result:?}"
4072 );
4073 }
4074
4075 #[tokio::test]
4077 async fn pin_to_ipv6_literal_host_is_rejected() {
4078 let result = Client::new()
4079 .get("http://[2606:4700::1111]/")
4080 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
4081 .send()
4082 .await;
4083
4084 assert!(
4085 matches!(result, Err(ClientError::PinRequiresDomainHost(_))),
4086 "expected PinRequiresDomainHost, got {result:?}"
4087 );
4088 }
4089
4090 #[tokio::test]
4096 async fn get_ssrf_safe_with_pin_to_is_rejected() {
4097 let result = Client::new()
4098 .get_ssrf_safe("http://example.com/")
4099 .pin_to(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080))
4100 .send()
4101 .await;
4102
4103 assert!(
4104 matches!(result, Err(ClientError::PinNotAllowedWithSsrfSafe(_))),
4105 "expected PinNotAllowedWithSsrfSafe, got {result:?}"
4106 );
4107 }
4108
4109 #[tokio::test]
4114 async fn follow_redirects_does_not_follow_304_with_location() {
4115 use axum::{Router, routing::get};
4116 use std::sync::atomic::AtomicBool;
4117
4118 let touched = Arc::new(AtomicBool::new(false));
4120 let touched2 = touched.clone();
4121 let b_addr = spawn(Router::new().route(
4122 "/dst",
4123 get(move || {
4124 let t = touched2.clone();
4125 async move {
4126 t.store(true, Ordering::SeqCst);
4127 "SHOULD-NOT-BE-HIT"
4128 }
4129 }),
4130 ))
4131 .await;
4132 let b_port = b_addr.port();
4133
4134 let a_addr = spawn(Router::new().route(
4136 "/start",
4137 get(move || async move {
4138 response_with_location(304, format!("http://127.0.0.1:{b_port}/dst"))
4139 }),
4140 ))
4141 .await;
4142
4143 let resp = Client::new()
4144 .get(format!("http://127.0.0.1:{}/start", a_addr.port()))
4145 .follow_redirects(5, |_| true)
4146 .send()
4147 .await
4148 .unwrap();
4149
4150 assert_eq!(
4151 resp.status().as_u16(),
4152 304,
4153 "a 304 with a Location header must be returned verbatim, not followed"
4154 );
4155 assert!(
4156 !touched.load(Ordering::SeqCst),
4157 "the 304 Location target must never be requested"
4158 );
4159 }
4160
4161 #[tokio::test]
4165 async fn follow_redirects_does_not_follow_300_with_location() {
4166 use axum::{Router, routing::get};
4167 use std::sync::atomic::AtomicBool;
4168
4169 let touched = Arc::new(AtomicBool::new(false));
4170 let touched2 = touched.clone();
4171 let b_addr = spawn(Router::new().route(
4172 "/dst",
4173 get(move || {
4174 let t = touched2.clone();
4175 async move {
4176 t.store(true, Ordering::SeqCst);
4177 "SHOULD-NOT-BE-HIT"
4178 }
4179 }),
4180 ))
4181 .await;
4182 let b_port = b_addr.port();
4183
4184 let a_addr = spawn(Router::new().route(
4185 "/start",
4186 get(move || async move {
4187 response_with_location(300, format!("http://127.0.0.1:{b_port}/dst"))
4188 }),
4189 ))
4190 .await;
4191
4192 let resp = Client::new()
4193 .get(format!("http://127.0.0.1:{}/start", a_addr.port()))
4194 .follow_redirects(5, |_| true)
4195 .send()
4196 .await
4197 .unwrap();
4198
4199 assert_eq!(
4200 resp.status().as_u16(),
4201 300,
4202 "a 300 with a Location header must be returned verbatim, not followed"
4203 );
4204 assert!(
4205 !touched.load(Ordering::SeqCst),
4206 "the 300 Location target must never be requested"
4207 );
4208 }
4209}