1use std::error::Error as _;
31use std::fmt::Write as _;
32use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
33use std::sync::Arc;
34
35use async_trait::async_trait;
36use bytes::Bytes;
37use cratefield_core::{
38 HttpClient, HttpError, HttpPolicy, MAX_CONCURRENT_REQUESTS, declared_content_length,
39 scrub_request_url,
40};
41use http::header::{
42 AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE, COOKIE, LOCATION, PROXY_AUTHORIZATION,
43};
44use reqwest::Url;
45use tokio::sync::Semaphore;
46
47const MAX_REDIRECTS: usize = 10;
50
51const GUARD_TAG: &str = "blocked-by-outbound-policy";
54
55const BLOCKED_HOST_SUFFIXES: [&str; 3] = [".internal", ".internal.", ".local"];
58
59const BLOCKED_HOSTS: [&str; 3] = ["metadata", "metadata.google", "metadata.google.internal"];
63
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
67pub struct OutboundOptions {
68 pub allow_loopback: bool,
74 pub max_concurrent: usize,
78}
79
80impl Default for OutboundOptions {
81 fn default() -> Self {
82 Self {
83 allow_loopback: false,
84 max_concurrent: MAX_CONCURRENT_REQUESTS,
85 }
86 }
87}
88
89pub struct ReqwestClient {
92 client: reqwest::Client,
93 budget: Arc<Semaphore>,
94 allow_loopback: bool,
95}
96
97impl Default for ReqwestClient {
98 fn default() -> Self {
99 Self::new()
100 }
101}
102
103impl ReqwestClient {
104 #[must_use]
113 pub fn new() -> Self {
114 Self::with_options(OutboundOptions::default())
115 }
116
117 #[must_use]
125 pub fn with_options(options: OutboundOptions) -> Self {
126 Self {
127 client: reqwest::Client::builder()
128 .redirect(reqwest::redirect::Policy::none())
129 .dns_resolver(Arc::new(GuardResolver {
134 allow_loopback: options.allow_loopback,
135 }))
136 .build()
137 .expect("reqwest initializes: rustls with webpki roots is static"),
138 budget: Arc::new(Semaphore::new(options.max_concurrent.max(1))),
139 allow_loopback: options.allow_loopback,
140 }
141 }
142}
143
144#[async_trait]
145impl HttpClient for ReqwestClient {
146 async fn send(
147 &self,
148 request: http::Request<Bytes>,
149 ) -> Result<http::Response<Bytes>, HttpError> {
150 let policy = HttpPolicy::of_request(&request);
151 let permit = Arc::clone(&self.budget).try_acquire_owned().map_err(|_| {
152 HttpError::Transport("outbound concurrency budget exhausted".to_owned())
153 })?;
154 let start = tokio::time::Instant::now();
155
156 let (parts, mut body) = request.into_parts();
157 let mut method = parts.method;
158 let mut headers = parts.headers;
159 let mut url = Url::parse(&parts.uri.to_string())
160 .map_err(|err| HttpError::Transport(format!("invalid request uri: {err}")))?;
161 vet_static(&url, self.allow_loopback)?;
162 vet_resolved_host(&url, self.allow_loopback).await?;
163
164 let response = self
165 .exchange_until_final(
166 policy,
167 &mut method,
168 &mut headers,
169 &mut body,
170 &mut url,
171 start,
172 )
173 .await?;
174
175 if let Some(declared) = declared_content_length(response.headers())
176 && declared > policy.max_response_bytes
177 {
178 return Err(HttpError::ResponseTooLarge {
179 limit: policy.max_response_bytes,
180 });
181 }
182 let status = response.status();
183 let headers_out = response.headers().clone();
184 let bytes = read_capped(response, policy.max_response_bytes).await?;
185 drop(permit);
188 let mut builder = http::Response::builder().status(status);
189 if let Some(target) = builder.headers_mut() {
190 *target = headers_out;
191 }
192 builder
193 .body(bytes)
194 .map_err(|err| HttpError::Transport(err.to_string()))
195 }
196}
197
198impl ReqwestClient {
199 async fn exchange_until_final(
202 &self,
203 policy: HttpPolicy,
204 method: &mut http::Method,
205 headers: &mut http::HeaderMap,
206 body: &mut Bytes,
207 url: &mut Url,
208 start: tokio::time::Instant,
209 ) -> Result<reqwest::Response, HttpError> {
210 for hop in 0.. {
211 let remaining = policy.timeout.saturating_sub(start.elapsed());
212 if remaining.is_zero() {
213 return Err(HttpError::DeadlineExceeded {
214 after: policy.timeout,
215 });
216 }
217 let mut builder = self
218 .client
219 .request(method.clone(), url.clone())
220 .timeout(remaining);
221 for (name, value) in headers.iter() {
222 builder = builder.header(name.clone(), value.clone());
223 }
224 if !body.is_empty() {
225 builder = builder.body(body.clone());
226 }
227 let response = match builder.send().await {
228 Ok(response) => response,
229 Err(err) if err.is_timeout() => {
230 return Err(HttpError::DeadlineExceeded {
231 after: policy.timeout,
232 });
233 }
234 Err(err) => {
235 let message = safe_message(&err);
239 return Err(if message.contains(GUARD_TAG) {
244 HttpError::BlockedDestination(message)
245 } else {
246 HttpError::Transport(message)
247 });
248 }
249 };
250 let status = response.status();
251 if !is_redirect(status) {
252 return Ok(response);
253 }
254 let Some(next) = response
255 .headers()
256 .get(LOCATION)
257 .and_then(|value| value.to_str().ok())
258 .and_then(|value| url.join(value).ok())
259 else {
260 return Ok(response);
262 };
263 if hop == MAX_REDIRECTS {
264 return Err(HttpError::Transport(format!(
265 "redirect limit ({MAX_REDIRECTS}) exceeded"
266 )));
267 }
268 vet_static(&next, self.allow_loopback)?;
269 vet_resolved_host(&next, self.allow_loopback).await?;
270 if next.authority() != url.authority() {
271 headers.remove(AUTHORIZATION);
274 headers.remove(COOKIE);
275 headers.remove(PROXY_AUTHORIZATION);
276 }
277 if redirect_changes_method(status, method) {
278 *method = http::Method::GET;
279 *body = Bytes::new();
280 headers.remove(CONTENT_TYPE);
281 headers.remove(CONTENT_LENGTH);
282 }
283 *url = next;
284 }
285 unreachable!("bounded by the MAX_REDIRECTS return")
286 }
287}
288
289fn is_redirect(status: http::StatusCode) -> bool {
290 matches!(status.as_u16(), 301 | 302 | 303 | 307 | 308)
291}
292
293fn redirect_changes_method(status: http::StatusCode, method: &http::Method) -> bool {
294 match status.as_u16() {
295 303 => *method != http::Method::GET && *method != http::Method::HEAD,
296 301 | 302 => *method == http::Method::POST,
297 _ => false,
298 }
299}
300
301async fn read_capped(mut response: reqwest::Response, limit: usize) -> Result<Bytes, HttpError> {
305 let mut buffer: Vec<u8> = Vec::new();
306 while let Some(chunk) = response
307 .chunk()
308 .await
309 .map_err(|err| HttpError::Transport(safe_message(&err)))?
310 {
311 if buffer.len() + chunk.len() > limit {
312 return Err(HttpError::ResponseTooLarge { limit });
313 }
314 buffer.extend_from_slice(&chunk);
315 }
316 Ok(Bytes::from(buffer))
317}
318
319fn safe_message(err: &reqwest::Error) -> String {
336 let mut message = err.to_string();
337 let mut cause = err.source();
338 while let Some(current) = cause {
339 let _ = write!(message, ": {current}");
340 cause = current.source();
341 }
342 scrub_url(message, err.url())
343}
344
345fn scrub_url(message: String, url: Option<&Url>) -> String {
355 match url {
356 Some(url) => scrub_request_url(&message, url.as_str()),
357 None => message,
358 }
359}
360
361fn vet_static(url: &Url, allow_loopback: bool) -> Result<(), HttpError> {
366 if url.scheme() != "http" && url.scheme() != "https" {
367 return Err(HttpError::BlockedDestination(format!(
368 "scheme `{}` is not allowed",
369 url.scheme()
370 )));
371 }
372 if !url.username().is_empty() || url.password().is_some() {
373 return Err(HttpError::BlockedDestination(
374 "userinfo in a request uri is not allowed".to_owned(),
375 ));
376 }
377 let Some(raw) = url.host_str() else {
378 return Err(HttpError::BlockedDestination("url has no host".to_owned()));
379 };
380 if let Some(v6) = raw
381 .strip_prefix('[')
382 .and_then(|inner| inner.strip_suffix(']'))
383 .and_then(|inner| inner.parse::<Ipv6Addr>().ok())
384 {
385 return vet_ip_with(IpAddr::V6(v6), allow_loopback);
386 }
387 if let Ok(v4) = raw.parse::<Ipv4Addr>() {
388 return vet_ip_with(IpAddr::V4(v4), allow_loopback);
389 }
390 vet_domain(raw)
391}
392
393fn vet_domain(name: &str) -> Result<(), HttpError> {
397 let lower = name.to_ascii_lowercase();
398 let trimmed = lower.strip_suffix('.').unwrap_or(&lower);
399 let refused = |what: &str| Err(HttpError::BlockedDestination(format!("`{name}` {what}")));
400 if trimmed == "localhost" || trimmed.ends_with(".localhost") {
401 return refused("is a loopback name");
404 }
405 if BLOCKED_HOSTS.contains(&trimmed)
406 || BLOCKED_HOST_SUFFIXES
407 .iter()
408 .any(|suffix| trimmed.ends_with(suffix))
409 {
410 return refused("names a private or metadata service");
411 }
412 Ok(())
413}
414
415async fn vet_resolved_host(url: &Url, allow_loopback: bool) -> Result<(), HttpError> {
420 let Some(raw) = url.host_str() else {
421 return Err(HttpError::BlockedDestination("url has no host".to_owned()));
422 };
423 if raw.parse::<Ipv4Addr>().is_ok()
424 || raw
425 .strip_prefix('[')
426 .and_then(|inner| inner.strip_suffix(']'))
427 .is_some_and(|inner| inner.parse::<Ipv6Addr>().is_ok())
428 {
429 return Ok(());
430 }
431 let addrs = resolve(raw).await.map_err(|err| {
432 HttpError::BlockedDestination(format!(
433 "`{raw}` cannot be resolved ({err}); destinations are vetted fail-closed"
434 ))
435 })?;
436 if addrs.is_empty() {
437 return Err(HttpError::BlockedDestination(format!(
438 "`{raw}` resolves to no address"
439 )));
440 }
441 for ip in addrs {
442 vet_ip_with(ip, allow_loopback)?;
443 }
444 Ok(())
445}
446
447async fn resolve(name: &str) -> Result<Vec<IpAddr>, std::io::Error> {
448 let name = name.to_owned();
449 tokio::task::spawn_blocking(move || {
450 use std::net::ToSocketAddrs;
451 (name.as_str(), 0u16)
452 .to_socket_addrs()
453 .map(|addrs| addrs.map(|addr| addr.ip()).collect::<Vec<_>>())
454 })
455 .await
456 .map_err(std::io::Error::other)?
457}
458
459#[must_use]
468pub fn vet_ip_ok(ip: IpAddr, allow_loopback: bool) -> bool {
469 vet_ip_with(ip, allow_loopback).is_ok()
470}
471
472fn vet_ip_with(ip: IpAddr, allow_loopback: bool) -> Result<(), HttpError> {
473 let blocked = Err(HttpError::BlockedDestination(format!(
474 "{ip} is not a publicly-routable destination"
475 )));
476 match ip {
477 IpAddr::V4(v4) => {
478 if v4.is_loopback() {
479 return if allow_loopback { Ok(()) } else { blocked };
480 }
481 let octets = v4.octets();
482 let cgnat = octets[0] == 100 && octets[1] & 0b1100_0000 == 0b0100_0000;
483 let benchmarking = octets[0] == 198 && octets[1] & 0b1111_1110 == 18;
486 let reserved = octets[0] >= 240;
487 if v4.is_private()
488 || v4.is_link_local()
489 || v4.is_unspecified()
490 || v4.is_multicast()
491 || v4.is_broadcast()
492 || v4.is_documentation()
493 || benchmarking
494 || reserved
495 || octets[0] == 0
496 || cgnat
498 {
499 blocked
500 } else {
501 Ok(())
502 }
503 }
504 IpAddr::V6(v6) => {
505 if v6.is_loopback() {
506 return if allow_loopback { Ok(()) } else { blocked };
507 }
508 let link_local = v6.segments()[0] & 0xffc0 == 0xfe80;
511 let unique_local = v6.segments()[0] & 0xfe00 == 0xfc00;
512 let documentation = v6.segments()[0] == 0x2001 && v6.segments()[1] == 0x0db8;
513 if v6.is_unspecified()
514 || v6.is_multicast()
515 || link_local
516 || unique_local
517 || documentation
518 {
519 return blocked;
520 }
521 if v6.segments()[0] == 0x2001 && v6.segments()[1] == 0 {
525 return blocked;
526 }
527 if let Some(mapped) = v6.to_ipv4_mapped() {
528 return vet_ip_with(IpAddr::V4(mapped), allow_loopback);
529 }
530 let octets = v6.octets();
531 let embedded_v4 = || Ipv4Addr::new(octets[12], octets[13], octets[14], octets[15]);
532 if v6.segments()[..6] == [0, 0, 0, 0, 0, 0] {
538 return vet_ip_with(IpAddr::V4(embedded_v4()), allow_loopback);
539 }
540 if v6.segments()[0] == 0x0064 && v6.segments()[1] == 0xff9b {
544 return vet_ip_with(IpAddr::V4(embedded_v4()), allow_loopback);
545 }
546 if v6.segments()[0] == 0x2002 {
548 let octets = v6.octets();
549 let embedded = Ipv4Addr::new(octets[2], octets[3], octets[4], octets[5]);
550 return vet_ip_with(IpAddr::V4(embedded), allow_loopback);
551 }
552 Ok(())
553 }
554 }
555}
556
557struct GuardResolver {
561 allow_loopback: bool,
562}
563
564impl reqwest::dns::Resolve for GuardResolver {
565 fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving {
566 let allow_loopback = self.allow_loopback;
567 let name = name.as_str().to_owned();
568 Box::pin(async move {
569 let refusal = |why: String| -> Box<dyn std::error::Error + Send + Sync> {
570 Box::new(std::io::Error::other(format!("{GUARD_TAG}: {why}")))
571 };
572 let addrs = resolve(&name)
573 .await
574 .map_err(|err| refusal(format!("`{name}` cannot be resolved ({err})")))?;
575 if addrs.is_empty() {
576 return Err(refusal(format!("`{name}` resolves to no address")));
577 }
578 for ip in &addrs {
579 if let Err(err) = vet_ip_with(*ip, allow_loopback) {
580 return Err(refusal(err.to_string()));
581 }
582 }
583 Ok(Box::new(
584 addrs
585 .into_iter()
586 .map(|ip| std::net::SocketAddr::new(ip, 0))
587 .collect::<Vec<_>>()
588 .into_iter(),
589 ) as reqwest::dns::Addrs)
590 })
591 }
592}
593
594#[cfg(test)]
595mod tests {
596 use super::*;
597
598 fn url(s: &str) -> Url {
599 Url::parse(s).expect("test url")
600 }
601
602 fn refused(s: &str) -> bool {
603 match vet_static(&url(s), false) {
604 Ok(()) => false,
605 Err(HttpError::BlockedDestination(_)) => true,
606 Err(other) => panic!("expected a policy refusal for {s}, got {other}"),
607 }
608 }
609
610 #[test]
611 fn loopback_is_blocked_everywhere_it_can_be_written() {
612 for s in [
615 "http://127.0.0.1/",
616 "http://127.1/",
617 "http://2130706433/",
618 "http://0x7f000001/",
619 "http://0177.0.0.1/",
620 "http://[::1]/",
621 "http://[::ffff:7f00:1]/",
622 "http://[0:0:0:0:0:ffff:127.0.0.1]/",
623 ] {
624 assert!(refused(s), "{s} must be refused");
625 }
626 }
627
628 #[test]
629 fn private_and_special_ranges_are_blocked() {
630 for s in [
631 "http://10.0.0.1/",
632 "http://192.168.70.70/",
633 "http://172.16.0.1/",
634 "http://0.0.0.0/",
635 "http://[fe80::1]/",
636 "http://[::]/",
637 "http://[ff02::1]/",
638 "http://[2001:db8::1]/",
639 "http://100.100.100.200/",
640 "http://255.255.255.255/",
641 ] {
642 assert!(refused(s), "{s} must be refused");
643 }
644 }
645
646 #[test]
647 fn cloud_metadata_endpoints_are_blocked() {
648 for s in [
650 "http://169.254.169.254/latest/meta-data/",
651 "http://[fd00:ec2::254]/",
652 "http://metadata.google.internal/",
653 "http://metadata/",
654 "http://foo.internal/",
655 "http://printer.local/",
656 ] {
657 assert!(refused(s), "{s} must be refused");
658 }
659 }
660
661 #[test]
662 fn non_http_schemes_and_userinfo_are_blocked() {
663 for s in [
664 "file:///etc/passwd",
665 "gopher://127.0.0.1:6379/",
666 "ftp://example.com/",
667 "http://admin:pw@example.com/",
668 "https://tok@169.254.169.254/",
669 ] {
670 assert!(refused(s), "{s} must be refused");
671 }
672 }
673
674 #[test]
675 fn ipv6_tunnel_forms_cannot_smuggle_an_ipv4_destination() {
676 assert!(refused("http://[2002:7f00:1::1]/"));
678 assert!(refused("http://[2001:0:4136:8000:0:63bf:3fff:fdd2]/"));
679 assert!(refused("http://[::127.0.0.1]/"));
682 assert!(refused("http://[::7f00:1]/"));
683 assert!(refused("http://[::10.0.0.1]/"));
684 assert!(refused("http://[64:ff9b::169.254.169.254]/"));
686 assert!(!refused("http://[::93.184.216.34]/"));
689 assert!(!refused("http://[64:ff9b::93.184.216.34]/"));
690 }
691
692 #[test]
693 fn legitimate_public_destinations_still_pass() {
694 for s in [
695 "https://api.resend.com/emails",
696 "https://93.184.216.34/",
697 "http://172.15.255.255/",
699 "http://172.32.0.1/",
700 "http://100.63.255.255/",
701 "http://100.128.0.1/",
702 ] {
703 assert!(!refused(s), "{s} must be allowed");
704 }
705 }
706
707 #[test]
708 fn loopback_opt_in_admits_only_loopback() {
709 assert!(!vet_ip_ok(IpAddr::V4(Ipv4Addr::LOCALHOST), false));
710 assert!(vet_ip_ok(IpAddr::V4(Ipv4Addr::LOCALHOST), true));
711 assert!(
712 !vet_ip_ok(IpAddr::V4(Ipv4Addr::new(169, 254, 169, 254)), true),
713 "metadata must stay blocked with loopback on"
714 );
715 assert!(
716 !vet_ip_ok(IpAddr::V4(Ipv4Addr::new(10, 1, 2, 3)), true),
717 "private must stay blocked with loopback on"
718 );
719 }
720
721 #[test]
722 fn names_are_refused_before_any_lookup() {
723 assert!(refused("http://localhost:8080/x"));
725 assert!(refused("http://LOCALHOST./x"));
726 assert!(refused("http://sub.localhost/x"));
727 assert!(!refused("http://api.example.com/x"));
728 }
729
730 #[tokio::test]
741 async fn a_transport_failure_names_the_origin_and_never_the_request_path() {
742 const TOKEN: &str = "0a1b2c3d4e5f60718293a4b5c6d7e8f900112233445566778899aabbccddeeff";
743
744 let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("a free port");
747 let addr = listener.local_addr().expect("its address");
748 drop(listener);
749
750 let client = ReqwestClient::with_options(OutboundOptions {
754 allow_loopback: true,
755 ..OutboundOptions::default()
756 });
757 let request = http::Request::builder()
758 .method(http::Method::POST)
759 .uri(format!("http://{addr}/3/device/{TOKEN}"))
760 .body(Bytes::new())
761 .expect("test request");
762
763 let err = client
764 .send(request)
765 .await
766 .expect_err("nothing is listening");
767 let HttpError::Transport(message) = &err else {
768 panic!("a refused connect is a transport failure, got {err}");
769 };
770 assert!(
771 !message.contains(TOKEN),
772 "the device token reached the error message: {message}"
773 );
774 assert!(
775 !message.contains("/3/device"),
776 "the request path reached the error message: {message}"
777 );
778 assert!(
781 message.contains(&addr.to_string()),
782 "the destination is named: {message}"
783 );
784 assert!(
785 message.to_ascii_lowercase().contains("refused")
786 || message.to_ascii_lowercase().contains("connect"),
787 "the cause survives the redaction: {message}"
788 );
789 }
790
791 #[test]
792 fn method_change_follows_the_http_rules() {
793 let post = http::Method::POST;
794 let get = http::Method::GET;
795 let code = |n: u16| http::StatusCode::from_u16(n).expect("test status");
796 assert!(redirect_changes_method(code(303), &post));
797 assert!(redirect_changes_method(code(302), &post));
798 assert!(!redirect_changes_method(code(307), &post));
799 assert!(!redirect_changes_method(code(301), &get));
800 assert!(!redirect_changes_method(code(303), &get));
801 }
802}