#![deny(missing_docs)]
use bytes::Bytes;
use flate2::read::{GzDecoder, ZlibDecoder};
use http::header::{
HeaderName, HeaderValue, ACCEPT_ENCODING, AUTHORIZATION, CONTENT_ENCODING, CONTENT_LENGTH,
CONTENT_TYPE, COOKIE, LOCATION, PROXY_AUTHORIZATION, TRANSFER_ENCODING, USER_AGENT,
};
use http::{HeaderMap, Method, Request, StatusCode, Uri};
use http_body_util::{BodyExt, Full, Limited};
use hyper_util::client::legacy::Client as HyperClient;
use hyper_util::rt::TokioExecutor;
use serde::de::DeserializeOwned;
use serde::Serialize;
use std::io::{self, Read, Write};
use std::time::Duration;
const DEFAULT_MAX_RESPONSE_BYTES: usize = 16 * 1024 * 1024;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
const DEFAULT_MAX_REDIRECTS: usize = 10;
#[cfg(feature = "tls")]
type Connector = hyper_rustls::HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>;
#[cfg(not(feature = "tls"))]
type Connector = hyper_util::client::legacy::connect::HttpConnector;
#[derive(Debug)]
pub enum ClientError {
Url(String),
Request(String),
Transport(String),
Timeout(Duration),
BodyTooLarge(usize),
Body(String),
Decode(String),
TooManyRedirects(usize),
}
impl std::fmt::Display for ClientError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Url(why) => write!(f, "invalid url: {why}"),
Self::Request(why) => write!(f, "could not build request: {why}"),
Self::Transport(why) => write!(f, "transport failure: {why}"),
Self::Timeout(after) => write!(f, "request timed out after {after:?}"),
Self::BodyTooLarge(limit) => {
write!(f, "response body exceeded the {limit} byte limit")
}
Self::Body(why) => write!(f, "could not read response body: {why}"),
Self::Decode(why) => write!(f, "could not decode response: {why}"),
Self::TooManyRedirects(limit) => write!(f, "more than {limit} redirects"),
}
}
}
impl std::error::Error for ClientError {}
#[derive(Clone, Debug)]
pub struct Client {
inner: HyperClient<Connector, Full<Bytes>>,
timeout: Duration,
max_response_bytes: usize,
max_redirects: usize,
user_agent: HeaderValue,
default_headers: HeaderMap,
auto_decompress: bool,
}
impl Default for Client {
fn default() -> Self {
Self::new()
}
}
impl Client {
pub fn new() -> Self {
#[cfg(feature = "tls")]
let connector = {
let mut http = hyper_util::client::legacy::connect::HttpConnector::new();
http.enforce_http(false);
hyper_rustls::HttpsConnectorBuilder::new()
.with_webpki_roots()
.https_or_http()
.enable_all_versions()
.wrap_connector(http)
};
#[cfg(not(feature = "tls"))]
let connector = hyper_util::client::legacy::connect::HttpConnector::new();
Self {
inner: HyperClient::builder(TokioExecutor::new()).build(connector),
timeout: DEFAULT_TIMEOUT,
max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
max_redirects: DEFAULT_MAX_REDIRECTS,
user_agent: HeaderValue::from_static(concat!("churust/", env!("CARGO_PKG_VERSION"))),
default_headers: HeaderMap::new(),
auto_decompress: true,
}
}
pub fn timeout(mut self, d: Duration) -> Self {
self.timeout = d;
self
}
pub fn max_response_bytes(mut self, bytes: usize) -> Self {
self.max_response_bytes = bytes;
self
}
pub fn max_redirects(mut self, n: usize) -> Self {
self.max_redirects = n;
self
}
pub fn auto_decompress(mut self, enabled: bool) -> Self {
self.auto_decompress = enabled;
self
}
pub fn user_agent(mut self, value: &str) -> Result<Self, ClientError> {
self.user_agent =
HeaderValue::from_str(value).map_err(|e| ClientError::Request(e.to_string()))?;
Ok(self)
}
pub fn default_header(mut self, name: &str, value: &str) -> Result<Self, ClientError> {
let name = HeaderName::from_bytes(name.as_bytes())
.map_err(|e| ClientError::Request(e.to_string()))?;
let value =
HeaderValue::from_str(value).map_err(|e| ClientError::Request(e.to_string()))?;
self.default_headers.insert(name, value);
Ok(self)
}
pub fn request(&self, method: Method, url: impl Into<String>) -> RequestBuilder {
RequestBuilder {
client: self.clone(),
method,
url: url.into(),
headers: HeaderMap::new(),
body: Bytes::new(),
timeout: None,
error: None,
}
}
pub fn get(&self, url: impl Into<String>) -> RequestBuilder {
self.request(Method::GET, url)
}
pub fn post(&self, url: impl Into<String>) -> RequestBuilder {
self.request(Method::POST, url)
}
pub fn put(&self, url: impl Into<String>) -> RequestBuilder {
self.request(Method::PUT, url)
}
pub fn patch(&self, url: impl Into<String>) -> RequestBuilder {
self.request(Method::PATCH, url)
}
pub fn delete(&self, url: impl Into<String>) -> RequestBuilder {
self.request(Method::DELETE, url)
}
pub fn head(&self, url: impl Into<String>) -> RequestBuilder {
self.request(Method::HEAD, url)
}
}
#[derive(Debug)]
pub struct RequestBuilder {
client: Client,
method: Method,
url: String,
headers: HeaderMap,
body: Bytes,
timeout: Option<Duration>,
error: Option<ClientError>,
}
impl RequestBuilder {
pub fn header(mut self, name: &str, value: &str) -> Self {
match (
HeaderName::from_bytes(name.as_bytes()),
HeaderValue::from_str(value),
) {
(Ok(name), Ok(value)) => {
self.headers.insert(name, value);
}
_ => self.fail(ClientError::Request(format!("invalid header {name}"))),
}
self
}
pub fn bearer(self, token: &str) -> Self {
self.header(AUTHORIZATION.as_str(), &format!("Bearer {token}"))
}
pub fn query<T: Serialize>(mut self, pairs: &T) -> Self {
match serde_html_form::to_string(pairs) {
Ok(encoded) if encoded.is_empty() => {}
Ok(encoded) => {
let separator = if self.url.contains('?') { '&' } else { '?' };
self.url.push(separator);
self.url.push_str(&encoded);
}
Err(e) => self.fail(ClientError::Request(e.to_string())),
}
self
}
pub fn body(mut self, body: impl Into<Bytes>) -> Self {
self.body = body.into();
self
}
pub fn json<T: Serialize>(mut self, value: &T) -> Self {
match serde_json::to_vec(value) {
Ok(encoded) => {
self.body = Bytes::from(encoded);
self.headers
.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
}
Err(e) => self.fail(ClientError::Request(e.to_string())),
}
self
}
pub fn form<T: Serialize>(mut self, value: &T) -> Self {
match serde_html_form::to_string(value) {
Ok(encoded) => {
self.body = Bytes::from(encoded);
self.headers.insert(
CONTENT_TYPE,
HeaderValue::from_static("application/x-www-form-urlencoded"),
);
}
Err(e) => self.fail(ClientError::Request(e.to_string())),
}
self
}
pub fn timeout(mut self, d: Duration) -> Self {
self.timeout = Some(d);
self
}
fn fail(&mut self, error: ClientError) {
if self.error.is_none() {
self.error = Some(error);
}
}
pub async fn send(self) -> Result<Response, ClientError> {
if let Some(error) = self.error {
return Err(error);
}
let deadline = self.timeout.unwrap_or(self.client.timeout);
let send = self.send_following();
match tokio::time::timeout(deadline, send).await {
Ok(result) => result,
Err(_) => Err(ClientError::Timeout(deadline)),
}
}
async fn send_following(self) -> Result<Response, ClientError> {
let RequestBuilder {
client,
method,
url,
headers,
body,
..
} = self;
let mut headers = headers;
let mut uri: Uri = url.parse().map_err(|e| ClientError::Url(format!("{e}")))?;
let mut method = method;
let mut body = body;
let mut hops = 0usize;
let mut stripped: std::collections::HashSet<http::HeaderName> =
std::collections::HashSet::new();
loop {
check_scheme(&uri)?;
let mut request = Request::builder()
.method(method.clone())
.uri(uri.clone())
.body(Full::new(body.clone()))
.map_err(|e| ClientError::Request(e.to_string()))?;
let target = request.headers_mut();
for (name, value) in &client.default_headers {
if stripped.contains(name) {
continue;
}
target.insert(name, value.clone());
}
for (name, value) in &headers {
target.insert(name, value.clone());
}
target
.entry(USER_AGENT)
.or_insert_with(|| client.user_agent.clone());
if client.auto_decompress && !stripped.contains(&ACCEPT_ENCODING) {
target
.entry(ACCEPT_ENCODING)
.or_insert_with(|| HeaderValue::from_static("gzip, deflate"));
}
let response = client
.inner
.request(request)
.await
.map_err(|e| ClientError::Transport(e.to_string()))?;
let status = response.status();
let redirect = matches!(
status,
StatusCode::MOVED_PERMANENTLY
| StatusCode::FOUND
| StatusCode::SEE_OTHER
| StatusCode::TEMPORARY_REDIRECT
| StatusCode::PERMANENT_REDIRECT
);
if redirect && client.max_redirects > 0 {
if hops >= client.max_redirects {
return Err(ClientError::TooManyRedirects(client.max_redirects));
}
if let Some(next) = response
.headers()
.get(LOCATION)
.and_then(|v| v.to_str().ok())
.map(|v| v.to_string())
{
let previous = uri.clone();
uri = resolve(&uri, &next)?;
if !same_origin(&previous, &uri) {
for name in [AUTHORIZATION, COOKIE, PROXY_AUTHORIZATION] {
headers.remove(&name);
stripped.insert(name);
}
}
if previous.scheme_str() == Some("https") && uri.scheme_str() == Some("http") {
return Err(ClientError::Url(
"refusing a redirect from https to http".into(),
));
}
if matches!(
status,
StatusCode::MOVED_PERMANENTLY | StatusCode::FOUND | StatusCode::SEE_OTHER
) && method != Method::HEAD
{
method = Method::GET;
body = Bytes::new();
for name in [
CONTENT_TYPE,
CONTENT_LENGTH,
CONTENT_ENCODING,
TRANSFER_ENCODING,
] {
headers.remove(&name);
stripped.insert(name);
}
}
hops += 1;
continue;
}
}
let (parts, incoming) = response.into_parts();
let collected = Limited::new(incoming, client.max_response_bytes)
.collect()
.await
.map_err(|e| {
if e.downcast_ref::<http_body_util::LengthLimitError>()
.is_some()
{
ClientError::BodyTooLarge(client.max_response_bytes)
} else {
ClientError::Body(e.to_string())
}
})?;
let mut headers = parts.headers;
let body = if client.auto_decompress {
maybe_decompress(
&mut headers,
collected.to_bytes(),
client.max_response_bytes,
)?
} else {
collected.to_bytes()
};
return Ok(Response {
status: parts.status,
headers,
body,
});
}
}
}
fn maybe_decompress(
headers: &mut HeaderMap,
body: Bytes,
max_bytes: usize,
) -> Result<Bytes, ClientError> {
let Some(raw) = headers.get(CONTENT_ENCODING).and_then(|v| v.to_str().ok()) else {
return Ok(body);
};
let coding = raw
.split(',')
.next()
.unwrap_or("")
.trim()
.to_ascii_lowercase();
let decoded = match coding.as_str() {
"gzip" | "x-gzip" => inflate_limited(GzDecoder::new(body.as_ref()), max_bytes)?,
"deflate" => inflate_limited(ZlibDecoder::new(body.as_ref()), max_bytes)?,
"identity" | "" => return Ok(body),
_ => return Ok(body),
};
headers.remove(CONTENT_ENCODING);
headers.remove(CONTENT_LENGTH);
Ok(Bytes::from(decoded))
}
fn inflate_limited<R: Read>(mut r: R, max_bytes: usize) -> Result<Vec<u8>, ClientError> {
let mut out = LimitedWriter {
buf: Vec::new(),
max: max_bytes,
};
match std::io::copy(&mut r, &mut out) {
Ok(_) => Ok(out.buf),
Err(e) if e.kind() == io::ErrorKind::Other && e.to_string().contains("body too large") => {
Err(ClientError::BodyTooLarge(max_bytes))
}
Err(e) => Err(ClientError::Decode(format!("decompress: {e}"))),
}
}
struct LimitedWriter {
buf: Vec<u8>,
max: usize,
}
impl Write for LimitedWriter {
fn write(&mut self, data: &[u8]) -> io::Result<usize> {
if self.buf.len().saturating_add(data.len()) > self.max {
return Err(io::Error::other("body too large"));
}
self.buf.extend_from_slice(data);
Ok(data.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
fn same_origin(a: &Uri, b: &Uri) -> bool {
fn port(u: &Uri) -> Option<u16> {
u.port_u16().or(match u.scheme_str() {
Some("http") => Some(80),
Some("https") => Some(443),
_ => None,
})
}
a.scheme_str() == b.scheme_str() && a.host() == b.host() && port(a) == port(b)
}
fn check_scheme(uri: &Uri) -> Result<(), ClientError> {
match uri.scheme_str() {
Some("http") => Ok(()),
#[cfg(feature = "tls")]
Some("https") => Ok(()),
#[cfg(not(feature = "tls"))]
Some("https") => Err(ClientError::Url(
"https needs the `tls` feature on churust-client".into(),
)),
Some(other) => Err(ClientError::Url(format!("unsupported scheme: {other}"))),
None => Err(ClientError::Url("no scheme in url".into())),
}
}
fn names_its_own_scheme(location: &str) -> bool {
let Some(boundary) = location.find([':', '/', '?', '#']) else {
return false;
};
if location.as_bytes()[boundary] != b':' {
return false;
}
let mut scheme = location[..boundary].chars();
scheme.next().is_some_and(|c| c.is_ascii_alphabetic())
&& scheme.all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '-' | '.'))
}
fn resolve(current: &Uri, location: &str) -> Result<Uri, ClientError> {
if names_its_own_scheme(location) {
return location
.parse()
.map_err(|e| ClientError::Url(format!("bad redirect target: {e}")));
}
let authority = current
.authority()
.ok_or_else(|| ClientError::Url("cannot resolve a relative redirect".into()))?;
let scheme = current.scheme_str().unwrap_or("http");
let joined = if location.starts_with('/') {
format!("{scheme}://{authority}{location}")
} else if location.starts_with('?') {
format!("{scheme}://{authority}{}{location}", current.path())
} else {
let base = current
.path()
.rsplit_once('/')
.map(|(head, _)| head)
.unwrap_or("");
format!("{scheme}://{authority}{base}/{location}")
};
joined
.parse()
.map_err(|e| ClientError::Url(format!("bad redirect target: {e}")))
}
#[derive(Clone, Debug)]
pub struct Response {
status: StatusCode,
headers: HeaderMap,
body: Bytes,
}
impl Response {
pub fn status(&self) -> StatusCode {
self.status
}
pub fn headers(&self) -> &HeaderMap {
&self.headers
}
pub fn header(&self, name: &str) -> Option<&str> {
self.headers.get(name).and_then(|v| v.to_str().ok())
}
pub fn bytes(&self) -> &Bytes {
&self.body
}
pub fn text(&self) -> Result<String, ClientError> {
std::str::from_utf8(&self.body)
.map(str::to_owned)
.map_err(|e| ClientError::Decode(e.to_string()))
}
pub fn json<T: DeserializeOwned>(&self) -> Result<T, ClientError> {
serde_json::from_slice(&self.body).map_err(|e| ClientError::Decode(e.to_string()))
}
pub fn error_for_status(self) -> Result<Self, ClientError> {
if self.status.is_client_error() || self.status.is_server_error() {
let preview: String = self.text().unwrap_or_default().chars().take(200).collect();
return Err(ClientError::Transport(format!(
"{}: {preview}",
self.status
)));
}
Ok(self)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_relative_redirect_resolves_against_the_current_path() {
let current: Uri = "http://example.com/a/b".parse().unwrap();
assert_eq!(
resolve(¤t, "c").unwrap().to_string(),
"http://example.com/a/c"
);
}
#[test]
fn an_absolute_path_redirect_replaces_the_path() {
let current: Uri = "http://example.com/a/b".parse().unwrap();
assert_eq!(
resolve(¤t, "/z").unwrap().to_string(),
"http://example.com/z"
);
}
#[test]
fn an_absolute_redirect_is_taken_whole() {
let current: Uri = "http://example.com/a".parse().unwrap();
assert_eq!(
resolve(¤t, "http://other.test/x")
.unwrap()
.to_string(),
"http://other.test/x"
);
}
#[test]
fn a_relative_redirect_whose_query_contains_a_url_still_resolves() {
let current: Uri = "http://api.example.com/dashboard".parse().unwrap();
assert_eq!(
resolve(¤t, "/login?next=https://api.example.com/dashboard")
.unwrap()
.to_string(),
"http://api.example.com/login?next=https://api.example.com/dashboard"
);
}
#[test]
fn a_redirect_naming_a_scheme_without_a_double_slash_is_not_joined_onto_the_origin() {
let current: Uri = "http://example.com/a".parse().unwrap();
let target = resolve(¤t, "mailto:ops@example.com").unwrap();
assert_eq!(target.to_string(), "mailto:ops@example.com");
assert!(matches!(check_scheme(&target), Err(ClientError::Url(_))));
}
#[test]
fn a_query_only_redirect_keeps_the_path_it_came_from() {
let current: Uri = "http://example.com/a/b".parse().unwrap();
assert_eq!(
resolve(¤t, "?page=2").unwrap().to_string(),
"http://example.com/a/b?page=2"
);
}
#[test]
fn a_scheme_the_client_cannot_speak_is_refused() {
let uri: Uri = "ftp://example.com/x".parse().unwrap();
assert!(matches!(check_scheme(&uri), Err(ClientError::Url(_))));
}
#[tokio::test]
async fn a_file_url_never_reaches_the_connector() {
let err = Client::new()
.get("file:///etc/passwd")
.send()
.await
.expect_err("a file url must not be fetched");
assert!(matches!(err, ClientError::Url(_)), "{err:?}");
}
#[test]
fn plain_http_is_allowed() {
let uri: Uri = "http://example.com/".parse().unwrap();
assert!(check_scheme(&uri).is_ok());
}
#[cfg(not(feature = "tls"))]
#[test]
fn https_without_the_tls_feature_says_so() {
let uri: Uri = "https://example.com/".parse().unwrap();
match check_scheme(&uri) {
Err(ClientError::Url(why)) => assert!(why.contains("tls")),
other => panic!("expected a url error, got {other:?}"),
}
}
#[test]
fn query_pairs_are_appended_to_an_existing_query() {
let client = Client::new();
let req = client
.get("http://example.com/search?a=1")
.query(&[("b", "2")]);
assert_eq!(req.url, "http://example.com/search?a=1&b=2");
}
#[test]
fn query_pairs_open_a_query_when_there_is_none() {
let client = Client::new();
let req = client.get("http://example.com/search").query(&[("b", "2")]);
assert_eq!(req.url, "http://example.com/search?b=2");
}
#[test]
fn an_invalid_header_surfaces_at_send_not_at_the_builder() {
let client = Client::new();
let req = client.get("http://example.com/").header("bad header", "x");
assert!(matches!(req.error, Some(ClientError::Request(_))));
}
#[test]
fn json_sets_the_content_type() {
let client = Client::new();
let req = client
.post("http://example.com/")
.json(&serde_json::json!({"a": 1}));
assert_eq!(req.headers.get(CONTENT_TYPE).unwrap(), "application/json");
assert_eq!(req.body, Bytes::from(r#"{"a":1}"#));
}
#[test]
fn error_for_status_keeps_success() {
let ok = Response {
status: StatusCode::OK,
headers: HeaderMap::new(),
body: Bytes::from("fine"),
};
assert!(ok.error_for_status().is_ok());
}
#[test]
fn error_for_status_reports_the_body_of_a_failure() {
let bad = Response {
status: StatusCode::BAD_REQUEST,
headers: HeaderMap::new(),
body: Bytes::from("missing field: name"),
};
match bad.error_for_status() {
Err(ClientError::Transport(why)) => {
assert!(why.contains("400"));
assert!(why.contains("missing field"));
}
other => panic!("expected a transport error, got {other:?}"),
}
}
}