diff --git a/Cargo.lock b/Cargo.lock index f8a43799f918f..681143ce0ca04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13268,11 +13268,16 @@ dependencies = [ "http 0.2.12", "http 1.3.1", "http-body 0.4.6", + "http-body 1.0.1", + "http-body-util", "http-serde", "humantime", "hyper 0.14.32", + "hyper 1.7.0", + "hyper-openssl 0.10.2", "hyper-openssl 0.9.2", "hyper-proxy", + "hyper-util", "indexmap 2.12.0", "indoc", "inventory", diff --git a/Cargo.toml b/Cargo.toml index 050d818fe2ae6..1fb7e587abf2a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -416,10 +416,15 @@ http = { version = "0.2.9", default-features = false } http-1 = { package = "http", version = "1.0", default-features = false, features = ["std"] } http-serde = "1.1.3" http-body = { version = "0.4.6", default-features = false } +http-body-1 = { package = "http-body", version = "1", default-features = false } +http-body-util = { version = "0.1", default-features = false } humantime.workspace = true hyper = { version = "0.14.32", default-features = false, features = ["client", "runtime", "http1", "http2", "server", "stream", "backports", "deprecated"] } +hyper-1 = { package = "hyper", version = "1", default-features = false, features = ["client", "http1", "http2"] } hyper-openssl = { version = "0.9.2", default-features = false } +hyper-openssl-1 = { package = "hyper-openssl", version = "0.10", default-features = false, features = ["client-legacy"] } hyper-proxy = { version = "0.9.1", default-features = false, features = ["openssl-tls"] } +hyper-util = { version = "0.1", default-features = false, features = ["client", "client-legacy", "http1", "http2", "tokio"] } indexmap.workspace = true inventory = { version = "0.3.20", default-features = false } ipnet = { version = "2", default-features = false, optional = true, features = ["serde", "std"] } diff --git a/src/http.rs b/src/http.rs index 0407956444732..01d2adc42c712 100644 --- a/src/http.rs +++ b/src/http.rs @@ -813,6 +813,8 @@ impl IntoIterator for QueryParameterValue { pub type QueryParameters = HashMap; +mod client_v1; + #[cfg(test)] mod transport_tests; diff --git a/src/http/client_v1.rs b/src/http/client_v1.rs new file mode 100644 index 0000000000000..00add89712a20 --- /dev/null +++ b/src/http/client_v1.rs @@ -0,0 +1,631 @@ +#![allow(dead_code)] // This client is private and currently exercised only by its transport contract. + +use std::{ + convert::Infallible, + error::Error, + future::Future, + io, + pin::Pin, + sync::Arc, + task::{Context, Poll}, +}; + +use bytes::Bytes; +use http_1::{ + HeaderValue, Method, Request, Response, Uri, + header::{ACCEPT_ENCODING, HOST, PROXY_AUTHORIZATION, USER_AGENT}, +}; +use http_body_util::{BodyExt, Empty, combinators::UnsyncBoxBody}; +use hyper_1::{body::Incoming, rt}; +use hyper_openssl_1::{SslStream, client::legacy::HttpsConnector}; +use hyper_util::{ + client::legacy::{ + Client, + connect::{Connected, Connection, HttpConnector}, + }, + rt::TokioExecutor, +}; +use openssl::ssl::SslConnector; +use percent_encoding::percent_decode_str; +use tower::Service; +use url::{Host, Url}; + +use crate::{ + config::ProxyConfig, + tls::{MaybeTlsSettings, TlsSettings, tls_connector_builder}, +}; + +type BoxError = Box; +pub(crate) type RequestBody = UnsyncBoxBody; +const HTTP1_ALPN: &[u8] = b"\x08http/1.1"; + +pub(crate) fn empty_body() -> RequestBody { + Empty::::new() + .map_err(|never: Infallible| match never {}) + .boxed_unsync() +} + +pub(crate) struct HttpClientV1 { + client: Client, + routes: Arc, + user_agent: HeaderValue, +} + +impl HttpClientV1 { + pub(crate) fn new(tls: MaybeTlsSettings, proxy: &ProxyConfig) -> Result { + let routes = Arc::new(RoutePlanner::new(proxy)?); + let connector = HttpProxyConnectorV1::new(tls, Arc::clone(&routes))?; + let client = Client::builder(TokioExecutor::new()).build(connector); + + Ok(Self { + client, + routes, + user_agent: default_user_agent(), + }) + } + + pub(crate) async fn send( + &self, + mut request: Request, + ) -> Result, BoxError> { + default_request_headers(&mut request, &self.user_agent); + + if let Route::ForwardProxy { + authorization: Some(authorization), + .. + } = self.routes.route(request.uri())? + && !request.headers().contains_key(PROXY_AUTHORIZATION) + { + request + .headers_mut() + .insert(PROXY_AUTHORIZATION, authorization); + } + + self.client + .request(request) + .await + .map_err(|error| Box::new(error) as BoxError) + } +} + +fn default_user_agent() -> HeaderValue { + HeaderValue::from_str(&format!( + "{}/{}", + crate::get_app_name(), + crate::get_version() + )) + .expect("the application name and version must form a valid user agent") +} + +fn default_request_headers(request: &mut Request, user_agent: &HeaderValue) { + if !request.headers().contains_key(USER_AGENT) { + request.headers_mut().insert(USER_AGENT, user_agent.clone()); + } + if !request.headers().contains_key(ACCEPT_ENCODING) { + request + .headers_mut() + .insert(ACCEPT_ENCODING, HeaderValue::from_static("identity")); + } +} + +#[derive(Clone)] +struct ProxyEndpoint { + uri: Uri, + authorization: Option, +} + +impl ProxyEndpoint { + fn parse(value: &str) -> Result { + let url = Url::parse(value)?; + if url.scheme() != "http" && url.scheme() != "https" { + return Err(invalid_input("proxy URI scheme must be http or https")); + } + + let host = match url.host() { + Some(Host::Domain(host)) => host.to_owned(), + Some(Host::Ipv4(host)) => host.to_string(), + Some(Host::Ipv6(host)) => format!("[{host}]"), + None => return Err(invalid_input("proxy URI must contain a host")), + }; + let authority = match url.port() { + Some(port) => format!("{host}:{port}"), + None => host, + }; + let uri = Uri::builder() + .scheme(url.scheme()) + .authority(authority) + .path_and_query("/") + .build()?; + + let authorization = url + .password() + .map(|password| -> Result<_, BoxError> { + let username = percent_decode_str(url.username()).decode_utf8()?; + let password = percent_decode_str(password).decode_utf8()?; + let encoded = + openssl::base64::encode_block(format!("{username}:{password}").as_bytes()); + let mut value = HeaderValue::from_str(&format!("Basic {encoded}"))?; + value.set_sensitive(true); + Ok(value) + }) + .transpose()?; + + Ok(Self { uri, authorization }) + } +} + +#[derive(Clone)] +struct RoutePlanner { + config: ProxyConfig, + http: Option, + https: Option, +} + +impl RoutePlanner { + fn new(config: &ProxyConfig) -> Result { + let (http, https) = if config.enabled { + ( + config + .http + .as_deref() + .map(ProxyEndpoint::parse) + .transpose()?, + config + .https + .as_deref() + .map(ProxyEndpoint::parse) + .transpose()?, + ) + } else { + (None, None) + }; + + Ok(Self { + config: config.clone(), + http, + https, + }) + } + + fn route(&self, destination: &Uri) -> Result { + let scheme = destination + .scheme_str() + .ok_or_else(|| invalid_input("destination URI must contain a scheme"))?; + if scheme != "http" && scheme != "https" { + return Err(invalid_input( + "destination URI scheme must be http or https", + )); + } + + if !self.config.enabled || self.bypasses_proxy(destination) { + return Ok(Route::Direct { + destination: destination.clone(), + }); + } + + let proxy = if scheme == "https" { + self.https.clone() + } else { + self.http.clone() + }; + let Some(proxy) = proxy else { + return Ok(Route::Direct { + destination: destination.clone(), + }); + }; + + if scheme == "https" { + Ok(Route::ConnectProxy { + proxy: proxy.uri, + destination: destination.clone(), + authorization: proxy.authorization, + }) + } else { + Ok(Route::ForwardProxy { + proxy: proxy.uri, + authorization: proxy.authorization, + }) + } + } + + fn bypasses_proxy(&self, destination: &Uri) -> bool { + destination.host().is_some_and(|host| { + self.config.no_proxy.matches(host) + || destination + .port_u16() + .is_some_and(|port| self.config.no_proxy.matches(&format!("{host}:{port}"))) + }) + } +} + +enum Route { + Direct { + destination: Uri, + }, + ForwardProxy { + proxy: Uri, + authorization: Option, + }, + ConnectProxy { + proxy: Uri, + destination: Uri, + authorization: Option, + }, +} + +type BaseConnector = HttpsConnector; + +#[derive(Clone)] +struct HttpProxyConnectorV1 { + direct: BaseConnector, + proxy: BaseConnector, + destination_tls: SslConnector, + tls_settings: Option, + routes: Arc, +} + +impl HttpProxyConnectorV1 { + fn new(tls: MaybeTlsSettings, routes: Arc) -> Result { + let tls_settings = tls.tls().cloned(); + let direct = https_connector(&tls, false)?; + let proxy = https_connector(&tls, true)?; + let destination_tls = tls_connector_builder(&tls)?.build(); + + Ok(Self { + direct, + proxy, + destination_tls, + tls_settings, + routes, + }) + } + + async fn connect_tunnel( + mut proxy: BaseConnector, + destination_tls: SslConnector, + tls_settings: Option, + proxy_uri: Uri, + destination: Uri, + authorization: Option, + ) -> Result { + let proxy_stream = proxy.call(proxy_uri).await?; + let (mut sender, connection) = + hyper_1::client::conn::http1::handshake(proxy_stream).await?; + tokio::spawn(async move { + if let Err(error) = connection.with_upgrades().await { + tracing::debug!(message = "Proxy connection closed.", %error); + } + }); + + let authority = connect_authority(&destination)?; + let mut request = Request::builder() + .method(Method::CONNECT) + .uri(&authority) + .header(HOST, &authority) + .body(Empty::::new())?; + if let Some(authorization) = authorization { + request + .headers_mut() + .insert(PROXY_AUTHORIZATION, authorization); + } + + let mut response = sender.send_request(request).await?; + if !response.status().is_success() { + return Err(invalid_input(format!( + "proxy CONNECT failed with status {}", + response.status() + ))); + } + let upgraded = hyper_1::upgrade::on(&mut response).await?; + + let host = tls_host(&destination)?; + let mut configuration = destination_tls.configure()?; + if let Some(settings) = &tls_settings { + settings.apply_connect_configuration(&mut configuration, false)?; + } + let ssl = configuration.into_ssl(host)?; + let mut stream = SslStream::new(ssl, upgraded)?; + Pin::new(&mut stream).connect().await?; + let negotiated_h2 = stream.ssl().selected_alpn_protocol() == Some(b"h2".as_slice()); + + Ok(BoxedIo::new(TunnelIo { + inner: stream, + negotiated_h2, + })) + } +} + +impl Service for HttpProxyConnectorV1 { + type Response = BoxedIo; + type Error = BoxError; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, destination: Uri) -> Self::Future { + let route = self.routes.route(&destination); + let mut direct = self.direct.clone(); + let mut proxy = self.proxy.clone(); + let destination_tls = self.destination_tls.clone(); + let tls_settings = self.tls_settings.clone(); + + Box::pin(async move { + match route? { + Route::Direct { destination } => direct.call(destination).await.map(BoxedIo::new), + Route::ForwardProxy { + proxy: proxy_uri, .. + } => proxy + .call(proxy_uri) + .await + .map(|stream| BoxedIo::new(ForwardProxyIo(stream))), + Route::ConnectProxy { + proxy: proxy_uri, + destination, + authorization, + } => { + Self::connect_tunnel( + proxy, + destination_tls, + tls_settings, + proxy_uri, + destination, + authorization, + ) + .await + } + } + }) + } +} + +fn https_connector( + tls: &MaybeTlsSettings, + skip_server_name: bool, +) -> Result { + let mut http = HttpConnector::new(); + http.enforce_http(false); + let mut tls_builder = tls_connector_builder(tls)?; + if skip_server_name { + // Proxy connections are driven through HTTP/1.1, including CONNECT. Do not allow a TLS + // proxy to negotiate h2 from the destination's configured ALPN list. + tls_builder.set_alpn_protos(HTTP1_ALPN)?; + } + let mut https = HttpsConnector::with_connector(http, tls_builder)?; + let settings = tls.tls().cloned(); + https.set_callback(move |configuration, _uri| { + if let Some(settings) = &settings { + settings.apply_connect_configuration(configuration, skip_server_name)?; + } + Ok(()) + }); + Ok(https) +} + +fn connect_authority(destination: &Uri) -> Result { + let authority = destination + .authority() + .ok_or_else(|| invalid_input("HTTPS destination must contain an authority"))?; + Ok(if authority.port_u16().is_some() { + authority.as_str().to_owned() + } else { + format!("{authority}:443") + }) +} + +fn tls_host(destination: &Uri) -> Result<&str, BoxError> { + let host = destination + .host() + .ok_or_else(|| invalid_input("HTTPS destination must contain a host"))?; + Ok(host + .strip_prefix('[') + .and_then(|host| host.strip_suffix(']')) + .unwrap_or(host)) +} + +trait ConnectionIo: rt::Read + rt::Write + Connection + Unpin + Send {} + +impl ConnectionIo for T where T: rt::Read + rt::Write + Connection + Unpin + Send {} + +struct BoxedIo(Pin>); + +impl BoxedIo { + fn new(stream: T) -> Self + where + T: ConnectionIo + 'static, + { + Self(Box::pin(stream)) + } +} + +impl rt::Read for BoxedIo { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: rt::ReadBufCursor<'_>, + ) -> Poll> { + self.0.as_mut().poll_read(cx, buffer) + } +} + +impl rt::Write for BoxedIo { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + self.0.as_mut().poll_write(cx, buffer) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.0.as_mut().poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.0.as_mut().poll_shutdown(cx) + } + + fn is_write_vectored(&self) -> bool { + self.0.is_write_vectored() + } + + fn poll_write_vectored( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffers: &[io::IoSlice<'_>], + ) -> Poll> { + self.0.as_mut().poll_write_vectored(cx, buffers) + } +} + +impl Connection for BoxedIo { + fn connected(&self) -> Connected { + self.0.as_ref().get_ref().connected() + } +} + +struct ForwardProxyIo(T); + +impl rt::Read for ForwardProxyIo { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: rt::ReadBufCursor<'_>, + ) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_read(cx, buffer) + } +} + +impl rt::Write for ForwardProxyIo { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_write(cx, buffer) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_shutdown(cx) + } + + fn is_write_vectored(&self) -> bool { + self.0.is_write_vectored() + } + + fn poll_write_vectored( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffers: &[io::IoSlice<'_>], + ) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_write_vectored(cx, buffers) + } +} + +impl Connection for ForwardProxyIo { + fn connected(&self) -> Connected { + self.0.connected().proxy(true) + } +} + +struct TunnelIo { + inner: T, + negotiated_h2: bool, +} + +impl rt::Read for TunnelIo { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: rt::ReadBufCursor<'_>, + ) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_read(cx, buffer) + } +} + +impl rt::Write for TunnelIo { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_write(cx, buffer) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_shutdown(cx) + } + + fn is_write_vectored(&self) -> bool { + self.inner.is_write_vectored() + } + + fn poll_write_vectored( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffers: &[io::IoSlice<'_>], + ) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_write_vectored(cx, buffers) + } +} + +impl Connection for TunnelIo { + fn connected(&self) -> Connected { + let connected = Connected::new(); + if self.negotiated_h2 { + connected.negotiated_h2() + } else { + connected + } + } +} + +fn invalid_input(message: impl Into) -> BoxError { + Box::new(io::Error::new(io::ErrorKind::InvalidInput, message.into())) +} + +#[cfg(test)] +mod tests { + use super::{connect_authority, tls_host}; + + #[test] + fn connect_authority_includes_a_port() { + assert_eq!( + connect_authority(&"https://example.com/path".parse().unwrap()).unwrap(), + "example.com:443" + ); + assert_eq!( + connect_authority(&"https://example.com:8443/path".parse().unwrap()).unwrap(), + "example.com:8443" + ); + assert_eq!( + connect_authority(&"https://[::1]/path".parse().unwrap()).unwrap(), + "[::1]:443" + ); + } + + #[test] + fn tls_host_normalizes_ipv6_literals() { + assert_eq!( + tls_host(&"https://[::1]/path".parse().unwrap()).unwrap(), + "::1" + ); + assert_eq!( + tls_host(&"https://127.0.0.1/path".parse().unwrap()).unwrap(), + "127.0.0.1" + ); + assert_eq!( + tls_host(&"https://example.com/path".parse().unwrap()).unwrap(), + "example.com" + ); + } +} diff --git a/src/http/transport_tests.rs b/src/http/transport_tests.rs index f13731f64f801..7482938c16793 100644 --- a/src/http/transport_tests.rs +++ b/src/http/transport_tests.rs @@ -8,9 +8,13 @@ use std::{ use async_trait::async_trait; use http::{HeaderMap, Method, Request, Response, StatusCode, Uri, header}; use hyper::{Body, body::HttpBody as _, client::conn, server::conn::http1, service::service_fn}; +use rstest::rstest; use tokio::{io::copy_bidirectional, net::TcpStream, task::JoinHandle, time::timeout}; -use super::HttpClient; +use super::{ + HttpClient, + client_v1::{HttpClientV1, empty_body}, +}; use crate::{ config::ProxyConfig, tls::{ @@ -68,49 +72,66 @@ impl Drop for TestServer { #[derive(Debug)] struct TestResponse { - status: StatusCode, + status: u16, body: Vec, } #[async_trait] trait TestClient: Send + Sync { async fn get(&self, uri: &str) -> Result; - async fn get_with_headers(&self, uri: &str, headers: HeaderMap) - -> Result; -} - -trait TestClientFactory: Copy { - type Client: TestClient; - - fn build(self, tls: MaybeTlsSettings, proxy: &ProxyConfig) -> Result; -} - -#[derive(Clone, Copy)] -struct LegacyClientFactory; - -impl TestClientFactory for LegacyClientFactory { - type Client = HttpClient; - - fn build(self, tls: MaybeTlsSettings, proxy: &ProxyConfig) -> Result { - HttpClient::new(tls, proxy).map_err(|error| error.to_string()) + async fn get_with_headers( + &self, + uri: &str, + headers: &[(&str, &str)], + ) -> Result; +} + +#[derive(Clone, Copy, Debug)] +enum ClientVersion { + Legacy, + V1, +} + +impl ClientVersion { + fn build( + self, + tls: MaybeTlsSettings, + proxy: &ProxyConfig, + ) -> Result, String> { + match self { + Self::Legacy => Ok(Box::new( + HttpClient::new(tls, proxy).map_err(|error| error.to_string())?, + )), + Self::V1 => Ok(Box::new( + HttpClientV1::new(tls, proxy).map_err(|error| error.to_string())?, + )), + } } } #[async_trait] impl TestClient for HttpClient { async fn get(&self, uri: &str) -> Result { - self.get_with_headers(uri, HeaderMap::new()).await + self.get_with_headers(uri, &[]).await } async fn get_with_headers( &self, uri: &str, - headers: HeaderMap, + headers: &[(&str, &str)], ) -> Result { let mut request = Request::get(uri) .body(Body::empty()) .map_err(|error| error.to_string())?; - *request.headers_mut() = headers; + for &(name, value) in headers { + request.headers_mut().insert( + http::header::HeaderName::from_bytes(name.as_bytes()) + .map_err(|error| error.to_string())?, + value + .parse() + .map_err(|error: http::header::InvalidHeaderValue| error.to_string())?, + ); + } let response = timeout(REQUEST_TIMEOUT, self.send(request)) .await .map_err(|_| "request timed out".to_owned())? @@ -121,6 +142,49 @@ impl TestClient for HttpClient { .map_err(|_| "response body timed out".to_owned())? .map_err(|error| error.to_string())? .to_bytes(); + Ok(TestResponse { + status: status.as_u16(), + body: body.to_vec(), + }) + } +} + +#[async_trait] +impl TestClient for HttpClientV1 { + async fn get(&self, uri: &str) -> Result { + self.get_with_headers(uri, &[]).await + } + + async fn get_with_headers( + &self, + uri: &str, + headers: &[(&str, &str)], + ) -> Result { + let mut request = http_1::Request::get(uri) + .body(empty_body()) + .map_err(|error| error.to_string())?; + for &(name, value) in headers { + request.headers_mut().insert( + http_1::header::HeaderName::from_bytes(name.as_bytes()) + .map_err(|error| error.to_string())?, + value + .parse() + .map_err(|error: http_1::header::InvalidHeaderValue| error.to_string())?, + ); + } + let response = timeout(REQUEST_TIMEOUT, self.send(request)) + .await + .map_err(|_| "request timed out".to_owned())? + .map_err(|error| error.to_string())?; + let status = response.status().as_u16(); + let body = timeout( + REQUEST_TIMEOUT, + http_body_util::BodyExt::collect(response.into_body()), + ) + .await + .map_err(|_| "response body timed out".to_owned())? + .map_err(|error| error.to_string())? + .to_bytes(); Ok(TestResponse { status, body: body.to_vec(), @@ -144,6 +208,13 @@ fn trusted_client_tls() -> MaybeTlsSettings { } fn server_tls(require_client_certificate: bool) -> MaybeTlsSettings { + server_tls_with_alpn(require_client_certificate, None) +} + +fn server_tls_with_alpn( + require_client_certificate: bool, + alpn_protocols: Option>, +) -> MaybeTlsSettings { MaybeTlsSettings::from_config( Some(&TlsEnableableConfig { enabled: Some(true), @@ -152,6 +223,7 @@ fn server_tls(require_client_certificate: bool) -> MaybeTlsSettings { ca_file: require_client_certificate.then(|| TEST_PEM_CA_PATH.into()), crt_file: Some(TEST_PEM_CRT_PATH.into()), key_file: Some(TEST_PEM_KEY_PATH.into()), + alpn_protocols, ..Default::default() }, }), @@ -200,6 +272,14 @@ async fn spawn_origin(tls: MaybeTlsSettings) -> TestServer { } async fn spawn_proxy(tls: MaybeTlsSettings, require_authentication: bool) -> TestServer { + spawn_proxy_with_connect_status(tls, require_authentication, StatusCode::OK).await +} + +async fn spawn_proxy_with_connect_status( + tls: MaybeTlsSettings, + require_authentication: bool, + connect_status: StatusCode, +) -> TestServer { let addr = "127.0.0.1:0".parse().unwrap(); let mut listener = tls.bind(&addr).await.unwrap(); let addr = listener.local_addr().unwrap(); @@ -217,7 +297,13 @@ async fn spawn_proxy(tls: MaybeTlsSettings, require_authentication: bool) -> Tes let observations = Arc::clone(&observations); async move { Ok::<_, Infallible>( - proxy_request(request, observations, require_authentication).await, + proxy_request( + request, + observations, + require_authentication, + connect_status, + ) + .await, ) } }); @@ -243,6 +329,7 @@ async fn proxy_request( mut request: Request, observations: Arc>>, require_authentication: bool, + connect_status: StatusCode, ) -> Response { observations.lock().unwrap().push(RequestObservation { method: request.method().clone(), @@ -272,7 +359,10 @@ async fn proxy_request( tracing::debug!(message = "Proxy tunnel closed.", %error); } }); - return Response::new(Body::empty()); + return Response::builder() + .status(connect_status) + .body(Body::empty()) + .unwrap(); } forward_http(request).await @@ -338,30 +428,44 @@ fn proxy_config(proxy: &TestServer, tls: bool, auth: bool) -> ProxyConfig { } fn assert_success(response: TestResponse) { - assert_eq!(response.status, StatusCode::OK); + assert_eq!(response.status, StatusCode::OK.as_u16()); assert_eq!(response.body, b"origin response"); } -async fn direct_http(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn direct_http(#[case] client_version: ClientVersion) { let origin = spawn_origin(no_tls()).await; - let client = factory.build(no_tls(), &ProxyConfig::default()).unwrap(); + let client = client_version + .build(no_tls(), &ProxyConfig::default()) + .unwrap(); assert_success(client.get(&origin.http_uri()).await.unwrap()); assert_eq!(origin.observations().len(), 1); } -async fn direct_https_with_trusted_ca(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn direct_https_with_trusted_ca(#[case] client_version: ClientVersion) { let origin = spawn_origin(server_tls(false)).await; - let client = factory + let client = client_version .build(trusted_client_tls(), &ProxyConfig::default()) .unwrap(); assert_success(client.get(&origin.https_uri()).await.unwrap()); } -async fn direct_https_rejects_untrusted_ca(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn direct_https_rejects_untrusted_ca(#[case] client_version: ClientVersion) { let origin = spawn_origin(server_tls(false)).await; - let client = factory + let client = client_version .build( client_tls(TlsConfig { ca_file: Some(INVALID_CA_PATH.into()), @@ -377,9 +481,13 @@ async fn direct_https_rejects_untrusted_ca(factory: F) { .expect_err("an untrusted server certificate must fail"); } -async fn direct_https_honors_server_name(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn direct_https_honors_server_name(#[case] client_version: ClientVersion) { let origin = spawn_origin(server_tls(false)).await; - let without_override = factory + let without_override = client_version .build(trusted_client_tls(), &ProxyConfig::default()) .unwrap(); without_override @@ -387,7 +495,7 @@ async fn direct_https_honors_server_name(factory: F) { .await .expect_err("the server certificate does not cover its IP address"); - let with_override = factory + let with_override = client_version .build( client_tls(TlsConfig { ca_file: Some(TEST_PEM_CA_PATH.into()), @@ -400,9 +508,13 @@ async fn direct_https_honors_server_name(factory: F) { assert_success(with_override.get(&origin.https_ip_uri()).await.unwrap()); } -async fn direct_https_supports_mtls(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn direct_https_supports_mtls(#[case] client_version: ClientVersion) { let origin = spawn_origin(server_tls(true)).await; - let without_identity = factory + let without_identity = client_version .build(trusted_client_tls(), &ProxyConfig::default()) .unwrap(); without_identity @@ -410,7 +522,7 @@ async fn direct_https_supports_mtls(factory: F) { .await .expect_err("the server requires a client certificate"); - let with_identity = factory + let with_identity = client_version .build( client_tls(TlsConfig { ca_file: Some(TEST_PEM_CA_PATH.into()), @@ -424,21 +536,23 @@ async fn direct_https_supports_mtls(factory: F) { assert_success(with_identity.get(&origin.https_uri()).await.unwrap()); } -async fn http_via_authenticated_proxy(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn http_via_authenticated_proxy(#[case] client_version: ClientVersion) { let origin = spawn_origin(no_tls()).await; let proxy = spawn_proxy(no_tls(), true).await; - let client = factory + let client = client_version .build(no_tls(), &proxy_config(&proxy, false, true)) .unwrap(); - let mut destination_headers = HeaderMap::new(); - destination_headers.insert( - header::AUTHORIZATION, - "Bearer destination-token".parse().unwrap(), - ); assert_success( client - .get_with_headers(&origin.http_uri(), destination_headers) + .get_with_headers( + &origin.http_uri(), + &[("authorization", "Bearer destination-token")], + ) .await .unwrap(), ); @@ -458,10 +572,14 @@ async fn http_via_authenticated_proxy(factory: F) { ); } -async fn http_proxy_credentials_do_not_reach_origin(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn http_proxy_credentials_do_not_reach_origin(#[case] client_version: ClientVersion) { let origin = spawn_origin(no_tls()).await; let proxy = spawn_proxy(no_tls(), true).await; - let client = factory + let client = client_version .build(no_tls(), &proxy_config(&proxy, false, true)) .unwrap(); @@ -481,21 +599,23 @@ async fn http_proxy_credentials_do_not_reach_origin(factor assert!(!origin_headers.contains_key(header::AUTHORIZATION)); } -async fn https_via_authenticated_connect_proxy(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn https_via_authenticated_connect_proxy(#[case] client_version: ClientVersion) { let origin = spawn_origin(server_tls(false)).await; let proxy = spawn_proxy(no_tls(), true).await; - let client = factory + let client = client_version .build(trusted_client_tls(), &proxy_config(&proxy, false, true)) .unwrap(); - let mut destination_headers = HeaderMap::new(); - destination_headers.insert( - header::AUTHORIZATION, - "Bearer destination-token".parse().unwrap(), - ); assert_success( client - .get_with_headers(&origin.https_uri(), destination_headers) + .get_with_headers( + &origin.https_uri(), + &[("authorization", "Bearer destination-token")], + ) .await .unwrap(), ); @@ -516,32 +636,77 @@ async fn https_via_authenticated_connect_proxy(factory: F) ); } -async fn no_proxy_bypasses_proxy(factory: F) { +#[tokio::test] +async fn v1_accepts_any_successful_connect_status() { + let origin = spawn_origin(server_tls(false)).await; + let proxy = spawn_proxy_with_connect_status(no_tls(), false, StatusCode::CREATED).await; + let client = ClientVersion::V1 + .build(trusted_client_tls(), &proxy_config(&proxy, false, false)) + .unwrap(); + + assert_success(client.get(&origin.https_uri()).await.unwrap()); +} + +#[tokio::test] +async fn v1_tls_proxy_uses_http1_alpn() { + let origin = spawn_origin(server_tls(false)).await; + let proxy = spawn_proxy( + server_tls_with_alpn(false, Some(vec!["h2".to_owned(), "http/1.1".to_owned()])), + false, + ) + .await; + let client = ClientVersion::V1 + .build( + client_tls(TlsConfig { + ca_file: Some(TEST_PEM_CA_PATH.into()), + alpn_protocols: Some(vec!["h2".to_owned(), "http/1.1".to_owned()]), + ..Default::default() + }), + &proxy_config(&proxy, true, false), + ) + .unwrap(); + + assert_success(client.get(&origin.https_uri()).await.unwrap()); +} + +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn no_proxy_bypasses_proxy(#[case] client_version: ClientVersion) { let origin = spawn_origin(no_tls()).await; let proxy = spawn_proxy(no_tls(), true).await; let mut config = proxy_config(&proxy, false, false); config.no_proxy = "127.0.0.1".into(); - let client = factory.build(no_tls(), &config).unwrap(); + let client = client_version.build(no_tls(), &config).unwrap(); assert_success(client.get(&origin.http_ip_uri()).await.unwrap()); assert!(proxy.observations().is_empty()); } -async fn disabled_proxy_is_bypassed(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn disabled_proxy_is_bypassed(#[case] client_version: ClientVersion) { let origin = spawn_origin(no_tls()).await; let proxy = spawn_proxy(no_tls(), true).await; let mut config = proxy_config(&proxy, false, false); config.enabled = false; - let client = factory.build(no_tls(), &config).unwrap(); + let client = client_version.build(no_tls(), &config).unwrap(); assert_success(client.get(&origin.http_uri()).await.unwrap()); assert!(proxy.observations().is_empty()); } -async fn http_via_tls_proxy(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn http_via_tls_proxy(#[case] client_version: ClientVersion) { let origin = spawn_origin(no_tls()).await; let proxy = spawn_proxy(server_tls(false), false).await; - let client = factory + let client = client_version .build(trusted_client_tls(), &proxy_config(&proxy, true, false)) .unwrap(); @@ -549,10 +714,14 @@ async fn http_via_tls_proxy(factory: F) { assert_eq!(proxy.observations().len(), 1); } -async fn https_via_tls_connect_proxy(factory: F) { +#[rstest] +#[case::legacy(ClientVersion::Legacy)] +#[case::v1(ClientVersion::V1)] +#[tokio::test] +async fn https_via_tls_connect_proxy(#[case] client_version: ClientVersion) { let origin = spawn_origin(server_tls(false)).await; let proxy = spawn_proxy(server_tls(false), false).await; - let client = factory + let client = client_version .build(trusted_client_tls(), &proxy_config(&proxy, true, false)) .unwrap(); @@ -561,63 +730,3 @@ async fn https_via_tls_connect_proxy(factory: F) { assert_eq!(observations.len(), 1); assert_eq!(observations[0].method, Method::CONNECT); } - -#[tokio::test] -async fn legacy_direct_http() { - direct_http(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_direct_https_with_trusted_ca() { - direct_https_with_trusted_ca(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_direct_https_rejects_untrusted_ca() { - direct_https_rejects_untrusted_ca(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_direct_https_honors_server_name() { - direct_https_honors_server_name(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_direct_https_supports_mtls() { - direct_https_supports_mtls(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_http_via_authenticated_proxy() { - http_via_authenticated_proxy(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_http_proxy_credentials_do_not_reach_origin() { - http_proxy_credentials_do_not_reach_origin(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_https_via_authenticated_connect_proxy() { - https_via_authenticated_connect_proxy(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_no_proxy_bypasses_proxy() { - no_proxy_bypasses_proxy(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_disabled_proxy_is_bypassed() { - disabled_proxy_is_bypassed(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_http_via_tls_proxy() { - http_via_tls_proxy(LegacyClientFactory).await; -} - -#[tokio::test] -async fn legacy_https_via_tls_connect_proxy() { - https_via_tls_connect_proxy(LegacyClientFactory).await; -}