use std::future::Future;
use std::io;
use std::net::{IpAddr, SocketAddr};
use std::pin::Pin;
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};
use bytes::Bytes;
use http::header::{self, HeaderMap, HeaderName, HeaderValue};
use http::{Method, StatusCode};
use tokio::net::TcpStream;
use url::Url;
use rustls_platform_verifier::BuilderVerifierExt;
use crate::policy::Sources;
use crate::{Error, Policy};
pub trait Resolve: Send + Sync + 'static {
fn resolve<'a>(
&'a self,
host: &'a str,
port: u16,
) -> Pin<Box<dyn Future<Output = io::Result<Vec<SocketAddr>>> + Send + 'a>>;
}
struct System;
impl Resolve for System {
fn resolve<'a>(
&'a self,
host: &'a str,
port: u16,
) -> Pin<Box<dyn Future<Output = io::Result<Vec<SocketAddr>>> + Send + 'a>> {
Box::pin(async move { Ok(tokio::net::lookup_host((host, port)).await?.collect()) })
}
}
struct Unpinned;
impl reqwest::dns::Resolve for Unpinned {
fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving {
let name = name.as_str().to_string();
Box::pin(async move { Err(format!("{name} was not checked against the policy").into()) })
}
}
const REFUSED_HEADERS: &[&str] = &[
"connection",
"content-length",
"expect",
"host",
"keep-alive",
"proxy-authorization",
"proxy-connection",
"te",
"trailer",
"transfer-encoding",
"upgrade",
];
#[derive(Clone)]
#[must_use]
pub struct Request {
method: Method,
url: Url,
headers: HeaderMap,
body: Option<Vec<u8>>,
refused: Option<HeaderName>,
}
impl std::fmt::Debug for Request {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Request")
.field("method", &self.method)
.field("origin", &self.url.origin().ascii_serialization())
.field("headers", &self.headers.keys().collect::<Vec<_>>())
.field("body_len", &self.body.as_ref().map(Vec::len))
.finish()
}
}
impl Request {
pub fn new(method: Method, url: Url) -> Request {
Request {
method,
url,
headers: HeaderMap::new(),
body: None,
refused: None,
}
}
pub fn header(mut self, name: HeaderName, mut value: HeaderValue) -> Request {
if REFUSED_HEADERS.contains(&name.as_str()) {
self.refused.get_or_insert(name);
return self;
}
if name == header::AUTHORIZATION || name == header::COOKIE {
value.set_sensitive(true);
}
self.headers.append(name, value);
self
}
pub fn body(mut self, body: impl Into<Vec<u8>>) -> Request {
self.body = Some(body.into());
self
}
}
pub struct Response {
inner: reqwest::Response,
url: Url,
timeout: Duration,
max_body: usize,
read: usize,
}
impl std::fmt::Debug for Response {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Response")
.field("status", &self.status())
.field("origin", &self.url.origin().ascii_serialization())
.field("headers", &self.headers().keys().collect::<Vec<_>>())
.field("remote_addr", &self.remote_addr())
.field("read", &self.read)
.finish()
}
}
impl Response {
pub fn status(&self) -> StatusCode {
self.inner.status()
}
pub fn headers(&self) -> &HeaderMap {
self.inner.headers()
}
pub fn url(&self) -> &Url {
&self.url
}
pub fn remote_addr(&self) -> Option<SocketAddr> {
self.inner.remote_addr()
}
pub async fn chunk(&mut self) -> Result<Option<Bytes>, Error> {
let Some(chunk) = self.next_chunk().await? else {
return Ok(None);
};
if self.read + chunk.len() > self.max_body {
return Err(Error::BodyTooLarge(self.max_body));
}
self.read += chunk.len();
Ok(Some(chunk))
}
pub async fn bytes(mut self) -> Result<Vec<u8>, Error> {
let mut body = Vec::new();
while let Some(chunk) = self.chunk().await? {
body.extend_from_slice(&chunk);
}
Ok(body)
}
pub async fn bytes_truncated(mut self) -> Result<(Vec<u8>, bool), Error> {
let mut body = Vec::new();
while let Some(chunk) = self.next_chunk().await? {
let room = self.max_body - body.len();
if chunk.len() > room {
body.extend_from_slice(&chunk[..room]);
return Ok((body, true));
}
body.extend_from_slice(&chunk);
}
Ok((body, false))
}
pub async fn text(self) -> Result<String, Error> {
let body = self.bytes().await?;
Ok(String::from_utf8(body)
.unwrap_or_else(|e| String::from_utf8_lossy(e.as_bytes()).into_owned()))
}
async fn next_chunk(&mut self) -> Result<Option<Bytes>, Error> {
self.inner
.chunk()
.await
.map_err(|e| http_error(e, self.timeout))
}
}
#[derive(Clone)]
#[must_use]
pub struct Fetcher {
policy: Arc<Policy>,
resolver: Arc<dyn Resolve>,
user_agent: Option<HeaderValue>,
}
impl std::fmt::Debug for Fetcher {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Fetcher")
.field("policy", &self.policy)
.field("user_agent", &self.user_agent)
.finish()
}
}
impl Default for Fetcher {
fn default() -> Fetcher {
Fetcher::new(Policy::default())
}
}
const MIN_ATTEMPT: Duration = Duration::from_secs(1);
impl Fetcher {
pub fn new(policy: Policy) -> Fetcher {
Fetcher {
policy: Arc::new(policy),
resolver: Arc::new(System),
user_agent: None,
}
}
pub fn with_resolver(mut self, resolver: impl Resolve) -> Fetcher {
self.resolver = Arc::new(resolver);
self
}
pub fn user_agent(mut self, value: HeaderValue) -> Fetcher {
self.user_agent = Some(value);
self
}
pub fn policy(&self) -> &Policy {
&self.policy
}
pub async fn get(&self, url: &str) -> Result<Response, Error> {
let url = Url::parse(url).map_err(|e| Error::InvalidUrl(format!("not a URL: {e}")))?;
self.send(Request::new(Method::GET, url)).await
}
pub async fn send(&self, request: Request) -> Result<Response, Error> {
if let Some(name) = &request.refused {
return Err(Error::RefusedHeader(name.to_string()));
}
let started = Instant::now();
let budget = self.policy.timeout;
match tokio::time::timeout(budget, self.follow(request, started)).await {
Ok(result) => result,
Err(_) => Err(Error::Timeout(budget)),
}
}
async fn follow(&self, request: Request, started: Instant) -> Result<Response, Error> {
let Request {
mut method,
mut url,
mut headers,
body,
refused: _,
} = request;
let budget = self.policy.timeout;
let max = self.policy.max_redirects;
let (mut host, mut port) = target(&url)?;
let mut send_body = body.is_some();
let mut trusted = true;
for hop in 0..=max {
let addrs = self.resolve(&host, port).await?;
let remaining = budget.saturating_sub(started.elapsed());
let client = self.client(&host, addrs, remaining)?;
let mut req = client.request(method.clone(), url.clone());
if trusted {
req = req.headers(headers.clone());
}
if send_body {
req = req.body(body.clone().unwrap_or_default());
}
let resp = req.send().await.map_err(|e| http_error(e, budget))?;
let status = resp.status().as_u16();
let follow = matches!(status, 301 | 302 | 303 | 307 | 308) && max > 0;
let location = resp.headers().get(header::LOCATION).filter(|_| follow);
let Some(location) = location else {
return Ok(Response {
inner: resp,
url,
timeout: budget,
max_body: self.policy.max_body,
read: 0,
});
};
if hop == max {
return Err(Error::TooManyRedirects(max));
}
let location = std::str::from_utf8(location.as_bytes()).map_err(|_| {
Error::BadRedirect(format!("HTTP {status} with a Location that isn't UTF-8"))
})?;
let next = url.join(location).map_err(|_| {
Error::BadRedirect(format!("HTTP {status} to an unreadable Location"))
})?;
(host, port) = target(&next).map_err(|e| Error::BadRedirect(e.to_string()))?;
trusted = trusted && same_origin(&url, &next);
let (next_method, keep_body) = redirect_method(&method, status);
let keep_body = send_body && keep_body && trusted;
if send_body && !keep_body {
for name in [
header::CONTENT_TYPE,
header::CONTENT_ENCODING,
header::CONTENT_LANGUAGE,
header::CONTENT_LOCATION,
] {
headers.remove(name);
}
}
send_body = keep_body;
method = next_method;
url = next;
}
Err(Error::TooManyRedirects(max))
}
fn client(
&self,
host: &str,
addrs: Vec<SocketAddr>,
remaining: Duration,
) -> Result<reqwest::Client, Error> {
let from = self.policy.sources.for_peer(addrs[0].ip()).ok().flatten();
let pinned: Vec<SocketAddr> = match from {
Some(from) => addrs
.into_iter()
.filter(|a| a.is_ipv4() == from.is_ipv4())
.collect(),
None => addrs,
};
let builder = || {
let mut b = reqwest::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.no_proxy()
.dns_resolver(Unpinned)
.connect_timeout(remaining)
.timeout(remaining)
.local_address(from);
if let Some(ua) = &self.user_agent {
b = b.user_agent(ua.clone());
}
if host.parse::<IpAddr>().is_err() {
b = b.resolve_to_addrs(host, &pinned);
}
b
};
builder()
.tls_backend_preconfigured(tls_config()?)
.build()
.map_err(|e| Error::Tls(e.without_url().to_string()))
}
pub async fn resolve(&self, host: &str, port: u16) -> Result<Vec<SocketAddr>, Error> {
let bare = host
.strip_prefix('[')
.and_then(|h| h.strip_suffix(']'))
.unwrap_or(host);
let mut addrs = match bare.parse::<IpAddr>() {
Ok(ip) => vec![SocketAddr::new(ip, port)],
Err(_) => self
.resolver
.resolve(bare, port)
.await
.map_err(|source| Error::Resolve {
host: host.to_string(),
source,
})?,
};
if addrs.is_empty() {
let source = io::Error::new(io::ErrorKind::NotFound, "no addresses");
return Err(Error::Resolve {
host: host.to_string(),
source,
});
}
for a in &mut addrs {
a.set_port(port);
}
if let Some(bad) = addrs.iter().find(|a| !self.policy.allows(a.ip())) {
return Err(Error::Denied {
host: host.to_string(),
addr: bad.ip(),
});
}
let reachable = self.policy.sources.reachable(addrs);
if reachable.is_empty() {
return Err(Error::NoSource {
host: host.to_string(),
});
}
Ok(reachable)
}
pub async fn connect(&self, host: &str, port: u16) -> Result<TcpStream, Error> {
let budget = self.policy.timeout;
let started = Instant::now();
let work = async {
let addrs = self.resolve(host, port).await?;
let count = addrs.len();
let mut last = None;
for (i, addr) in addrs.into_iter().enumerate() {
let left = budget.saturating_sub(started.elapsed());
let share = (left / (count - i) as u32).max(MIN_ATTEMPT.min(left));
match tokio::time::timeout(share, connect_with(self.policy.sources, addr)).await {
Ok(Ok(stream)) => return Ok(stream),
Ok(Err(e)) => last = Some(e),
Err(_) => {
let msg = format!("{addr} didn't answer within {share:?}");
last = Some(io::Error::new(io::ErrorKind::TimedOut, msg));
}
}
}
let source = last.unwrap_or_else(|| io::Error::other("no address to try"));
Err(Error::Connect {
host: host.to_string(),
source,
})
};
tokio::time::timeout(budget, work)
.await
.unwrap_or(Err(Error::Timeout(budget)))
}
}
async fn connect_with(sources: Sources, to: SocketAddr) -> io::Result<TcpStream> {
let from = sources
.for_peer(to.ip())
.map_err(|_| io::Error::other(format!("no source address in the family of {}", to.ip())))?;
let Some(from) = from else {
return TcpStream::connect(to).await;
};
let socket = match to {
SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4()?,
SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6()?,
};
socket.bind(SocketAddr::new(from, 0))?;
socket.connect(to).await
}
fn target(url: &Url) -> Result<(String, u16), Error> {
if !url.username().is_empty() || url.password().is_some() {
return Err(Error::InvalidUrl(
"the URL has a user name or password in it".to_string(),
));
}
if !matches!(url.scheme(), "http" | "https") {
return Err(Error::InvalidUrl(format!(
"{}: URLs can't be fetched, only http and https",
url.scheme()
)));
}
let host = match url.host_str() {
Some(h) if !h.is_empty() => h.trim_start_matches('[').trim_end_matches(']').to_string(),
_ => return Err(Error::InvalidUrl("the URL has no host".to_string())),
};
let port = url.port_or_known_default().unwrap_or(80);
Ok((host, port))
}
fn http_error(e: reqwest::Error, budget: Duration) -> Error {
if e.is_timeout() {
Error::Timeout(budget)
} else {
Error::Http(Box::new(e.without_url()))
}
}
fn redirect_method(method: &Method, status: u16) -> (Method, bool) {
let get = (Method::GET, false);
match status {
303 if *method != Method::HEAD => get,
301 | 302 if *method == Method::POST => get,
_ => (method.clone(), true),
}
}
fn same_origin(a: &Url, b: &Url) -> bool {
a.scheme() == b.scheme()
&& a.host_str() == b.host_str()
&& a.port_or_known_default() == b.port_or_known_default()
}
fn crypto_provider() -> Arc<rustls::crypto::CryptoProvider> {
match rustls::crypto::CryptoProvider::get_default() {
Some(p) => p.clone(),
None => Arc::new(rustls::crypto::aws_lc_rs::default_provider()),
}
}
fn tls_config() -> Result<rustls::ClientConfig, Error> {
static TLS: OnceLock<rustls::ClientConfig> = OnceLock::new();
if let Some(tls) = TLS.get() {
return Ok(tls.clone());
}
let mut tls = rustls::ClientConfig::builder_with_provider(crypto_provider())
.with_safe_default_protocol_versions()
.map_err(|e| Error::Tls(e.to_string()))?
.with_platform_verifier()
.map_err(|e| Error::Tls(format!("no trust store to verify against: {e}")))?
.with_no_client_auth();
tls.alpn_protocols = vec![b"http/1.1".to_vec()];
tls.resumption = rustls::client::Resumption::disabled();
Ok(TLS.get_or_init(|| tls).clone())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn reqwest_takes_the_shared_tls_config() {
reqwest::Client::builder()
.tls_backend_preconfigured(tls_config().unwrap())
.build()
.unwrap();
}
#[test]
fn request_debug_hides_secrets() {
let url = Url::parse("https://example.com/hooks/s-789?token=q-012").unwrap();
let req = Request::new(Method::POST, url)
.header(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer t-456"),
)
.header(
HeaderName::from_static("x-api-key"),
HeaderValue::from_static("k-123"),
)
.body("hunter2");
let shown = format!("{req:?}");
assert!(shown.contains("x-api-key"), "{shown}");
assert!(shown.contains("https://example.com"), "{shown}");
for secret in ["t-456", "k-123", "hunter2", "s-789", "q-012"] {
assert!(!shown.contains(secret), "{shown}");
}
assert!(req.headers[header::AUTHORIZATION].is_sensitive());
}
#[test]
fn redirect_methods() {
assert_eq!(redirect_method(&Method::POST, 303), (Method::GET, false));
assert_eq!(redirect_method(&Method::PUT, 303), (Method::GET, false));
assert_eq!(redirect_method(&Method::HEAD, 303), (Method::HEAD, true));
assert_eq!(redirect_method(&Method::POST, 301), (Method::GET, false));
assert_eq!(redirect_method(&Method::POST, 302), (Method::GET, false));
assert_eq!(redirect_method(&Method::PUT, 302), (Method::PUT, true));
assert_eq!(redirect_method(&Method::POST, 307), (Method::POST, true));
assert_eq!(redirect_method(&Method::PATCH, 308), (Method::PATCH, true));
assert_eq!(redirect_method(&Method::GET, 301), (Method::GET, true));
}
#[test]
fn origins() {
let u = |s: &str| Url::parse(s).unwrap();
assert!(same_origin(
&u("https://example.com/a"),
&u("https://example.com:443/b?c")
));
assert!(!same_origin(
&u("https://example.com/a"),
&u("https://evil.example/a")
));
assert!(
!same_origin(&u("https://example.com/a"), &u("http://example.com/a")),
"a downgrade is another origin"
);
assert!(!same_origin(
&u("https://example.com/a"),
&u("https://example.com:8443/a")
));
assert!(!same_origin(
&u("https://example.com/a"),
&u("https://www.example.com/a")
));
assert!(
same_origin(&u("https://example.com/a"), &u("https://EXAMPLE.com/b")),
"host case is not an origin"
);
}
#[test]
fn targets() {
let t = |s: &str| target(&Url::parse(s).unwrap());
assert_eq!(
t("https://example.com/x").unwrap(),
("example.com".into(), 443)
);
assert_eq!(
t("http://[2001:db8::1]:8080/").unwrap(),
("2001:db8::1".into(), 8080)
);
assert!(t("ftp://example.com/").is_err());
assert!(t("https://user:pw@example.com/").is_err());
assert!(t("https://user@example.com/").is_err());
assert!(t("file:///etc/passwd").is_err());
}
#[test]
fn url_errors_dont_repeat_the_url() {
let t = |s: &str| target(&Url::parse(s).unwrap()).unwrap_err().to_string();
for (url, secret) in [
("https://user:pw-345@example.com/", "pw-345"),
("ftp://user:pw-345@example.com/", "pw-345"),
("ftp://example.com/?token=q-012", "q-012"),
] {
let shown = t(url);
assert!(!shown.contains(secret), "{shown}");
}
}
}