rama/http/client/
mod.rs

1//! rama http client support
2//!
3//! Contains re-exports from `rama-http-backend::client`
4//! and adds `EasyHttpWebClient`, an opiniated http web client which
5//! supports most common use cases and provides sensible defaults.
6use std::{fmt, io};
7
8use crate::{
9    Layer, Service,
10    error::BoxError,
11    extensions::ExtensionsRef,
12    http::{Body as ResponseBody, Request, Response, StreamingBody},
13    net::client::EstablishedClientConnection,
14    rt::Executor,
15    service::BoxService,
16    telemetry::tracing,
17};
18
19#[doc(inline)]
20pub use ::rama_http::service::client::blocking::{
21    Body as BlockingBody, Client as BlockingHttpClient, Response as BlockingResponse,
22};
23#[doc(inline)]
24pub use ::rama_http_backend::client::*;
25use rama_core::{
26    error::{ErrorContext, ErrorExt as _, extra::OpaqueError},
27    extensions::Egress,
28    layer::MapErr,
29};
30use rama_http::{
31    layer::{
32        forward_proxy::{HttpForwardProxyLayer, HttpForwardProxyService},
33        map_response_body::MapResponseBody,
34    },
35    proxy::PlaintextHttpProxyMode,
36};
37
38#[cfg(any(feature = "boring", feature = "rustls"))]
39use crate::tls::client::TlsClientConfig;
40
41#[cfg(feature = "boring")]
42use {
43    crate::{quic::tls::BoringTlsProvider, tls::boring::client::TlsClientSessionCache},
44    std::sync::Arc,
45};
46
47pub mod builder;
48#[doc(inline)]
49pub use builder::EasyHttpConnectorBuilder;
50
51#[cfg(feature = "socks5")]
52mod proxy_connector;
53#[cfg(feature = "socks5")]
54#[cfg_attr(docsrs, doc(cfg(feature = "socks5")))]
55#[doc(inline)]
56pub use proxy_connector::{MaybeProxiedConnection, ProxyConnector, ProxyConnectorLayer};
57
58/// An opiniated http client that can be used to serve HTTP requests.
59///
60/// Use [`EasyHttpWebClient::connector_builder()`] to easily create a client with
61/// a common Http connector setup (tcp + proxy + tls + http) or bring your
62/// own http connector.
63///
64/// [`Default`] uses Rama's default multiplexing connection pool. Build the
65/// connector explicitly with
66/// [`EasyHttpConnectorBuilder::without_connection_pool`] when connection reuse
67/// is unwanted.
68///
69/// You can fork this http client in case you have use cases not possible with this service example.
70/// E.g. perhaps you wish to have middleware in into outbound requests, after they
71/// passed through your "connector" setup. All this and more is possible by defining your own
72/// http client. Rama is here to empower you, the building blocks are there, go crazy
73/// with your own service fork and use the full power of Rust at your fingertips ;)
74pub struct EasyHttpWebClient<BodyIn, ConnResponse, L> {
75    connector: BoxService<Request<BodyIn>, ConnResponse, OpaqueError>,
76    forward_proxy_layer: HttpForwardProxyLayer,
77    plaintext_http_proxy_mode: Option<PlaintextHttpProxyMode>,
78    jit_layers: L,
79}
80
81impl<BodyIn, ConnResponse, L> fmt::Debug for EasyHttpWebClient<BodyIn, ConnResponse, L> {
82    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
83        f.debug_struct("EasyHttpWebClient").finish()
84    }
85}
86
87impl<BodyIn, ConnResponse, L: Clone> Clone for EasyHttpWebClient<BodyIn, ConnResponse, L> {
88    fn clone(&self) -> Self {
89        Self {
90            connector: self.connector.clone(),
91            forward_proxy_layer: self.forward_proxy_layer.clone(),
92            plaintext_http_proxy_mode: self.plaintext_http_proxy_mode,
93            jit_layers: self.jit_layers.clone(),
94        }
95    }
96}
97
98impl EasyHttpWebClient<(), (), ()> {
99    /// Create a [`EasyHttpConnectorBuilder`] to easily create a [`EasyHttpWebClient`] with a custom connector
100    #[must_use]
101    pub fn connector_builder() -> EasyHttpConnectorBuilder {
102        EasyHttpConnectorBuilder::new()
103    }
104
105    /// Create a cloneable blocking HTTP(S) client with its own dedicated
106    /// runtime thread and Rama's default web connector stack.
107    ///
108    /// ```no_run
109    /// use rama::http::client::EasyHttpWebClient;
110    ///
111    /// # fn main() -> Result<(), rama::error::BoxError> {
112    /// let client = EasyHttpWebClient::try_blocking()?;
113    /// let client_for_worker = client.clone();
114    ///
115    /// let text = client_for_worker
116    ///     .get("https://example.com/")
117    ///     .send()?
118    ///     .try_into_string()?;
119    /// # _ = text;
120    /// # Ok(())
121    /// # }
122    /// ```
123    pub fn try_blocking() -> io::Result<BlockingHttpWebClient> {
124        BlockingHttpClient::try_new(EasyHttpWebClient::default())
125    }
126}
127
128/// Rama's default asynchronous HTTP(S) client, including its default
129/// multiplexing connection pool.
130///
131/// With BoringSSL or Rustls plus `ring`/`aws-lc`, HTTP/3 is available through
132/// alternative-service discovery or an explicit HTTP/3 request. Its shared UDP
133/// endpoint is bound lazily, so constructing the client requires no runtime.
134///
135/// Like a browser it resumes TLS sessions (with BoringSSL, also over TCP), which
136/// lets servers link the connections of one client. It acts for a single user:
137/// proxies relay traffic instead of sharing one client between their users.
138pub type DefaultHttpWebClient<Body = crate::http::Body> = EasyHttpWebClient<
139    Body,
140    EstablishedClientConnection<
141        rama_http::layer::alt_svc::AltSvc<
142            BindBodyToConn<
143                crate::net::client::pool::MultiplexedConnection<
144                    HttpClientService<Body>,
145                    HttpConnId,
146                >,
147            >,
148        >,
149        Request<Body>,
150    >,
151    (),
152>;
153
154/// A blocking HTTP(S) client using Rama's default pooled web connector stack.
155pub type BlockingHttpWebClient = BlockingHttpClient<DefaultHttpWebClient>;
156
157impl<Body> Default for DefaultHttpWebClient<Body>
158where
159    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
160{
161    #[inline(always)]
162    fn default() -> Self {
163        Self::default_with_executor(Executor::default())
164    }
165}
166
167impl<Body> DefaultHttpWebClient<Body>
168where
169    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
170{
171    core::cfg_select! {
172        feature = "boring" => {
173            pub fn default_with_executor(exec: Executor) -> Self {
174                let tls_config = TlsClientConfig::default_http();
175                #[expect(
176                    clippy::expect_used,
177                    reason = "fixed default H3 limits and the explicit BoringSSL provider are valid"
178                )]
179                let h3 = Http3Connector::builder(exec.clone())
180                    .with_tls_config(tls_config.clone())
181                    .with_tls_provider(Arc::new(BoringTlsProvider))
182                    .build_lazy()
183                    .expect("default BoringSSL HTTP/3 configuration is valid");
184
185                EasyHttpConnectorBuilder::new()
186                    .with_default_transport_connector()
187                    .with_default_dns_connector()
188                    .with_tls_proxy_support_using_boringssl()
189                    .with_proxy_support()
190                    .with_tls_support_using_boringssl(tls_config)
191                    .with_tls_session_store(Arc::new(TlsClientSessionCache::default()))
192                    .with_default_http_connector(exec)
193                    .with_http3_support(h3)
194                    .with_default_connection_pool()
195                    .build_client()
196            }
197        }
198        feature = "rustls" => {
199            pub fn default_with_executor(exec: Executor) -> Self {
200                let tls_config = TlsClientConfig::default_http();
201                #[cfg(any(feature = "ring", feature = "aws-lc"))]
202                #[expect(
203                    clippy::expect_used,
204                    reason = "fixed default H3 limits and a compiled-in Rustls crypto provider are valid"
205                )]
206                let h3 = Http3Connector::builder(exec.clone())
207                    .with_tls_config(tls_config.clone())
208                    .build_lazy()
209                    .expect("default Rustls HTTP/3 configuration is valid");
210
211                let builder = EasyHttpConnectorBuilder::new()
212                    .with_default_transport_connector()
213                    .with_default_dns_connector()
214                    .with_tls_proxy_support_using_rustls()
215                    .with_proxy_support()
216                    .with_tls_support_using_rustls(tls_config)
217                    .with_default_http_connector(exec);
218                #[cfg(any(feature = "ring", feature = "aws-lc"))]
219                let builder = builder.with_http3_support(h3);
220
221                builder
222                    .with_default_connection_pool()
223                    .build_client()
224            }
225        }
226        _ => {
227            pub fn default_with_executor(exec: Executor) -> Self {
228                EasyHttpConnectorBuilder::new()
229                    .with_default_transport_connector()
230                    .with_default_dns_connector()
231                    .without_tls_proxy_support()
232                    .with_proxy_support()
233                    .without_tls_support()
234                    .with_default_http_connector(exec)
235                    .with_default_connection_pool()
236                    .build_client()
237            }
238        }
239    }
240}
241
242impl<BodyIn, ConnResponse> EasyHttpWebClient<BodyIn, ConnResponse, ()>
243where
244    BodyIn: Send + 'static,
245{
246    /// Create a new [`EasyHttpWebClient`] using the provided connector.
247    ///
248    /// Custom proxy connectors must honor [`PlaintextHttpProxyMode`] and publish
249    /// [`EstablishedProxyRoute`](crate::net::client::EstablishedProxyRoute).
250    /// Connection wrappers must preserve that metadata through [`ExtensionsRef`].
251    #[must_use]
252    pub fn new<S>(connector: S) -> Self
253    where
254        S: Service<Request<BodyIn>, Output = ConnResponse, Error: Into<BoxError>>,
255    {
256        Self {
257            connector: MapErr::into_opaque_error(connector).boxed(),
258            forward_proxy_layer: HttpForwardProxyLayer::new(),
259            plaintext_http_proxy_mode: None,
260            jit_layers: (),
261        }
262    }
263}
264
265impl<BodyIn, ConnResponse, L> EasyHttpWebClient<BodyIn, ConnResponse, L> {
266    /// Convert this asynchronous web client into a cloneable blocking client
267    /// with its own dedicated runtime thread.
268    pub fn try_into_blocking(self) -> io::Result<BlockingHttpClient<Self>> {
269        BlockingHttpClient::try_new(self)
270    }
271
272    /// Convert this asynchronous web client into a blocking client using a
273    /// caller-supplied runtime.
274    #[must_use]
275    pub fn into_blocking_with_runtime(
276        self,
277        runtime: &crate::rt::blocking::Runtime,
278    ) -> BlockingHttpClient<Self> {
279        BlockingHttpClient::with_runtime(self, runtime)
280    }
281
282    /// Set the connector that this [`EasyHttpWebClient`] will use.
283    ///
284    /// Custom proxy connectors follow the metadata contract described in [`Self::new`].
285    #[must_use]
286    pub fn with_connector<S, BodyInNew, ConnResponseNew>(
287        self,
288        connector: S,
289    ) -> EasyHttpWebClient<BodyInNew, ConnResponseNew, L>
290    where
291        S: Service<Request<BodyInNew>, Output = ConnResponseNew, Error: Into<BoxError>>,
292        BodyInNew: Send + 'static,
293    {
294        EasyHttpWebClient {
295            connector: MapErr::into_opaque_error(connector).boxed(),
296            forward_proxy_layer: self.forward_proxy_layer,
297            plaintext_http_proxy_mode: self.plaintext_http_proxy_mode,
298            jit_layers: self.jit_layers,
299        }
300    }
301
302    /// [`Layer`] which will be applied just in time (JIT) before the request is sent, but after
303    /// the connection has been established. Rama's built-in forward-proxy
304    /// policy is the innermost JIT service so it can act on the actual
305    /// connection after caller middleware has processed the request, and can
306    /// isolate a proxy challenge before caller middleware sees the response.
307    ///
308    /// Simplified flow of how the [`EasyHttpWebClient`] works:
309    /// 1. External: let response = client.serve(request)
310    /// 2. Internal: let http_connection = self.connector.serve(request)
311    /// 3. Internal: wrap the connection in Rama's forward-proxy policy
312    /// 4. Internal: let response = jit_layers.layer(http_connection).serve(request)
313    pub fn with_jit_layer<T>(self, jit_layers: T) -> EasyHttpWebClient<BodyIn, ConnResponse, T> {
314        EasyHttpWebClient {
315            connector: self.connector,
316            forward_proxy_layer: self.forward_proxy_layer,
317            plaintext_http_proxy_mode: self.plaintext_http_proxy_mode,
318            jit_layers,
319        }
320    }
321
322    crate::utils::macros::generate_set_and_with! {
323        /// Enable or disable automatic Basic or Bearer credentials on requests
324        /// sent directly to an HTTP forward proxy.
325        ///
326        /// This is enabled by default and acts only when the established connection
327        /// is positively identified as an HTTP forward-proxy connection. It never
328        /// adds credentials to direct, SOCKS, or HTTP CONNECT-tunneled requests.
329        ///
330        /// Disabling this only disables insertion. Caller-provided
331        /// `Proxy-Authorization` headers are preserved on established HTTP
332        /// forward routes and always stripped on every other route, including
333        /// connections without established route metadata.
334        pub fn forward_proxy_auth(mut self, enabled: bool) -> Self {
335            self.forward_proxy_layer.set_proxy_auth(enabled);
336            self
337        }
338    }
339
340    /// Disable automatic Basic or Bearer credentials on HTTP forward-proxy
341    /// requests. The credential-stripping policy described by
342    /// [`Self::with_forward_proxy_auth`] still applies.
343    #[must_use]
344    pub fn without_forward_proxy_auth(self) -> Self {
345        self.with_forward_proxy_auth(false)
346    }
347
348    crate::utils::macros::generate_set_and_with! {
349        /// Enable or disable carrying plaintext HTTP through an HTTP(S) proxy with
350        /// CONNECT instead of using ordinary forward-proxy semantics
351        /// (absolute-form on HTTP/1).
352        ///
353        /// If this method is not called, a request-level
354        /// [`PlaintextHttpProxyMode`]
355        /// is honored and
356        /// otherwise forwarding is the connector default. Calling this method
357        /// explicitly selects Tunnel (`true`) or Forward (`false`) for the client.
358        /// Tunneling does not encrypt the origin traffic: a plaintext `http://`
359        /// request remains plaintext inside the proxy tunnel.
360        pub fn tunnel_plaintext_http(mut self, enabled: bool) -> Self {
361            self.plaintext_http_proxy_mode = Some(if enabled {
362                PlaintextHttpProxyMode::Tunnel
363            } else {
364                PlaintextHttpProxyMode::Forward
365            });
366            self
367        }
368    }
369
370    crate::utils::macros::generate_set_and_with! {
371        /// Enable or disable isolation of `407 Proxy Authentication Required`
372        /// responses received from an established HTTP forward proxy.
373        ///
374        /// Ordinary clients expose such responses by default. Intermediaries should
375        /// enable this option so an upstream proxy's challenge, headers, and body
376        /// cannot be forwarded to a different downstream proxy client.
377        pub fn isolate_forward_proxy_auth_error(mut self, enabled: bool) -> Self {
378            self.forward_proxy_layer.set_isolate_auth_error(enabled);
379            self
380        }
381    }
382}
383
384impl<Body, ConnectionBody, Connection, IncomingBody, L> Service<Request<Body>>
385    for EasyHttpWebClient<Body, EstablishedClientConnection<Connection, Request<ConnectionBody>>, L>
386where
387    Body: StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
388    Connection: Service<Request<ConnectionBody>, Output = Response<IncomingBody>, Error = BoxError>
389        + ExtensionsRef,
390    IncomingBody: Send + Sync + 'static,
391    ResponseBody: From<IncomingBody>,
392    // Body type this connection will be able to send, this is not necessarily the same one that
393    // was used in the request that created this connection
394    ConnectionBody:
395        StreamingBody<Data: Send + 'static, Error: Into<BoxError>> + Unpin + Send + 'static,
396    L: Layer<
397            HttpForwardProxyService<MapResponseBody<Connection, fn(IncomingBody) -> ResponseBody>>,
398            Service: Service<Request<ConnectionBody>, Output = Response, Error = BoxError>,
399        > + Send
400        + Sync
401        + 'static,
402{
403    type Output = Response;
404    type Error = OpaqueError;
405
406    async fn serve(&self, req: Request<Body>) -> Result<Self::Output, Self::Error> {
407        let uri = req.uri().clone();
408
409        if let Some(mode) = self.plaintext_http_proxy_mode {
410            req.extensions().insert(mode);
411        }
412
413        let EstablishedClientConnection {
414            input: req,
415            conn: http_connection,
416        } = self.connector.serve(req).await.into_opaque_error()?;
417
418        // Publish connection metadata for JIT middleware. The forward-proxy
419        // layer refreshes it after those layers run; the backend independently
420        // refreshes it for callers that use the backend without this client.
421        req.extensions()
422            .insert(Egress(http_connection.extensions().clone()));
423
424        let http_connection = MapResponseBody::into_boxed_streaming_body(http_connection);
425        let http_connection = self.forward_proxy_layer.layer(http_connection);
426        let http_connection = self.jit_layers.layer(http_connection);
427
428        // NOTE: stack might change request version based on connector data,
429        tracing::trace!(url.full = %uri, "send http req to connector stack");
430
431        let result = http_connection.serve(req).await;
432
433        match result {
434            Ok(resp) => {
435                tracing::trace!(url.full = %uri, "response received from connector stack");
436                Ok(resp)
437            }
438            Err(err) => Err(err
439                .context("http request failure")
440                .context_field("uri", uri)
441                .into_opaque_error()),
442        }
443    }
444}
445
446#[cfg(test)]
447mod tests {
448    use std::{
449        convert::Infallible,
450        num::NonZeroUsize,
451        sync::{
452            Arc,
453            atomic::{AtomicUsize, Ordering},
454        },
455        time::Duration,
456    };
457
458    use rama_core::extensions::Extensions;
459    use rama_core::{error::BoxErrorExt as _, service::service_fn};
460    use rama_http::{Body, BodyExtractExt, Version};
461    use rama_http_backend::server::HttpServer;
462    use rama_net::{
463        address::ProxyAddress,
464        client::{
465            ConnectRequest, ConnectionError, ConnectionErrorDomain, ConnectionErrorKind,
466            ConnectorService, ConnectorTarget, EstablishedProxyRoute, ProxyRoute,
467            ProxyRouteFailureCache, ProxyRouteFailureCacheConfig, ProxyRouteFailureCacheScope,
468            ProxyRoutes,
469        },
470        test_utils::client::{MockConnectorService, MockSocket},
471    };
472    use rama_utils::octets::kib;
473    use serde::{Deserialize, Serialize};
474    use tokio::time::sleep;
475
476    #[cfg(feature = "boring")]
477    use {
478        crate::{
479            net::address::SocketAddress,
480            tcp::{TcpStream, server::TcpListener},
481            tls::{
482                boring::{TlsStream, server::TlsAcceptorLayer},
483                client::{NegotiatedTlsParameters, ServerVerifyMode, TlsServerVerify},
484                server::{GeneratedServerAuthConfig, ServerAuthData, TlsServerConfig},
485            },
486        },
487        rama_core::{extensions::ExtensionsRef as _, layer::MapInputLayer},
488        rama_http::header::CONNECTION,
489    };
490
491    use super::*;
492
493    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
494    struct Output {
495        conn: usize,
496        resp: usize,
497    }
498
499    #[tokio::test]
500    async fn default_client_request_future_stays_within_stack_budget() {
501        let client = DefaultHttpWebClient::<Body>::default();
502        let request = Request::builder()
503            .uri("https://example.com/")
504            .body(Body::empty())
505            .unwrap();
506        let request = client.serve(request);
507        let inner_size = std::mem::size_of_val(&request);
508        let request = tokio::time::timeout(Duration::from_secs(1), request);
509        let size = std::mem::size_of_val(&request);
510        eprintln!("default client request: {inner_size} bytes; with timeout: {size} bytes");
511        assert!(
512            size <= kib(16),
513            "default client request future is {size} bytes"
514        );
515    }
516
517    #[test]
518    fn default_client_can_be_constructed_and_cloned_without_a_runtime() {
519        let client = DefaultHttpWebClient::<Body>::default();
520        drop(client.clone());
521        drop(client);
522    }
523
524    #[derive(Debug, Clone, Default)]
525    struct EmptyHttpConnection {
526        extensions: Extensions,
527    }
528
529    impl ExtensionsRef for EmptyHttpConnection {
530        fn extensions(&self) -> &Extensions {
531            &self.extensions
532        }
533    }
534
535    impl Service<Request> for EmptyHttpConnection {
536        type Output = Response;
537        type Error = BoxError;
538
539        async fn serve(&self, _request: Request) -> Result<Self::Output, Self::Error> {
540            Ok(Response::new(Body::empty()))
541        }
542    }
543
544    #[derive(Debug, Clone)]
545    struct InspectConnectionRouteLayer(EstablishedProxyRoute);
546
547    impl<S: ExtensionsRef> Layer<S> for InspectConnectionRouteLayer {
548        type Service = S;
549
550        fn layer(&self, inner: S) -> Self::Service {
551            assert_eq!(
552                inner.extensions().get_ref::<EstablishedProxyRoute>(),
553                Some(&self.0),
554            );
555            inner
556        }
557    }
558
559    fn dummy_server<Input: Send + 'static>()
560    -> impl Service<
561        Input,
562        Output = EstablishedClientConnection<MockSocket, Input>,
563        Error = Infallible,
564    > + Clone {
565        let created_connections = Arc::new(AtomicUsize::new(0));
566        MockConnectorService::new(move || {
567            let created_connections = created_connections.clone();
568            let conn = created_connections.fetch_add(1, Ordering::Relaxed);
569
570            // count responses created on this specific connection
571            let created_response = Arc::new(AtomicUsize::new(0));
572
573            HttpServer::auto(Executor::default()).service(service_fn(move |_req: Request| {
574                let created_response = created_response.clone();
575                let resp = created_response.fetch_add(1, Ordering::Relaxed);
576                async move {
577                    sleep(Duration::from_millis(5)).await;
578                    let out = Output { conn, resp };
579                    let resp = Response::new(Body::from(serde_json::to_vec(&out).unwrap()));
580                    Ok::<_, Infallible>(resp)
581                }
582            }))
583        })
584    }
585
586    #[tokio::test]
587    async fn custom_connector_receives_plaintext_http_proxy_mode() {
588        let connector = service_fn(|request: Request| async move {
589            assert_eq!(
590                request.extensions().get_ref::<PlaintextHttpProxyMode>(),
591                Some(&PlaintextHttpProxyMode::Tunnel)
592            );
593
594            let conn = EmptyHttpConnection::default();
595            Ok::<_, Infallible>(EstablishedClientConnection {
596                input: request,
597                conn,
598            })
599        });
600        let client = EasyHttpWebClient::new(connector).with_tunnel_plaintext_http(true);
601        let request = Request::builder()
602            .uri("http://example.com/")
603            .body(Body::empty())
604            .unwrap();
605
606        let response = client.serve(request).await.unwrap();
607        assert_eq!(response.status(), crate::http::StatusCode::OK);
608    }
609
610    #[tokio::test]
611    async fn jit_layer_can_read_established_connection_extensions() {
612        let connector = service_fn(|request: Request| async move {
613            let conn = EmptyHttpConnection::default();
614            conn.extensions().insert(EstablishedProxyRoute::Direct);
615            Ok::<_, Infallible>(EstablishedClientConnection {
616                input: request,
617                conn,
618            })
619        });
620        let client = EasyHttpWebClient::new(connector)
621            .with_jit_layer(InspectConnectionRouteLayer(EstablishedProxyRoute::Direct));
622        let request = Request::builder()
623            .uri("http://example.com/")
624            .extension(ProxyRoute::Direct)
625            .body(Body::empty())
626            .unwrap();
627
628        let response = client.serve(request).await.unwrap();
629        assert_eq!(response.status(), crate::http::StatusCode::OK);
630    }
631
632    #[tokio::test]
633    async fn jit_request_metadata_cannot_change_proxy_credentials_target_or_challenge_isolation() {
634        use rama_core::layer::{MapInputLayer, MapOutputLayer};
635        use rama_http::{HeaderValue, StatusCode, header::PROXY_AUTHORIZATION};
636
637        #[derive(Debug, Clone)]
638        struct InspectProxyConnection {
639            extensions: Extensions,
640        }
641
642        impl ExtensionsRef for InspectProxyConnection {
643            fn extensions(&self) -> &Extensions {
644                &self.extensions
645            }
646        }
647
648        impl Service<Request> for InspectProxyConnection {
649            type Output = Response;
650            type Error = BoxError;
651
652            async fn serve(&self, request: Request) -> Result<Response, BoxError> {
653                let route = self.extensions.get_ref::<EstablishedProxyRoute>();
654                let is_forward = route.is_some_and(EstablishedProxyRoute::is_http_forward);
655                assert_eq!(
656                    request
657                        .extensions()
658                        .egress()
659                        .unwrap()
660                        .0
661                        .get_ref::<EstablishedProxyRoute>(),
662                    route,
663                );
664                assert_eq!(
665                    request.extensions().get_ref::<ProxyRoute>(),
666                    Some(&ProxyRoute::Proxy(
667                        "http://wrong:request-secret@requested.example:8080"
668                            .parse()
669                            .unwrap(),
670                    )),
671                    "forward policy must preserve the caller's requested route",
672                );
673                let mut target = Vec::new();
674                rama_http::proto::h1::head::encode_request_target(
675                    request.method(),
676                    request.uri(),
677                    request.extensions(),
678                    &mut target,
679                )
680                .unwrap();
681                if is_forward {
682                    assert_eq!(
683                        request.headers().get(PROXY_AUTHORIZATION).unwrap(),
684                        "Basic dXBzdHJlYW06c2VjcmV0",
685                    );
686                    assert_eq!(&target[..], b"http://origin.example/resource");
687                } else {
688                    assert!(request.headers().get(PROXY_AUTHORIZATION).is_none());
689                    assert_eq!(&target[..], b"/resource");
690                }
691                Ok(Response::builder()
692                    .status(StatusCode::PROXY_AUTHENTICATION_REQUIRED)
693                    .header("proxy-authenticate", "Basic realm=private-upstream")
694                    .body(Body::from("private upstream challenge"))
695                    .unwrap())
696            }
697        }
698
699        let proxy: ProxyAddress = "http://upstream:secret@proxy.example:8080".parse().unwrap();
700        for isolate in [false, true] {
701            for route in [
702                None,
703                Some(EstablishedProxyRoute::Direct),
704                Some(EstablishedProxyRoute::Tunnel(proxy.clone())),
705                Some(EstablishedProxyRoute::Tunnel(
706                    "socks5://proxy.example:1080".parse().unwrap(),
707                )),
708                Some(EstablishedProxyRoute::Forward(proxy.clone())),
709            ] {
710                let is_forward = route
711                    .as_ref()
712                    .is_some_and(EstablishedProxyRoute::is_http_forward);
713                let connector = service_fn(move |request: Request| {
714                    let route = route.clone();
715                    async move {
716                        let extensions = Extensions::new();
717                        if let Some(route) = route {
718                            extensions.insert(route);
719                        }
720                        Ok::<_, Infallible>(EstablishedClientConnection {
721                            input: request,
722                            conn: InspectProxyConnection { extensions },
723                        })
724                    }
725                });
726                let stale_route = if is_forward {
727                    EstablishedProxyRoute::Direct
728                } else {
729                    EstablishedProxyRoute::Forward(proxy.clone())
730                };
731                let observed_responses = Arc::new(AtomicUsize::new(0));
732                let client = EasyHttpWebClient::new(connector)
733                    .with_isolate_forward_proxy_auth_error(isolate)
734                    .with_jit_layer((
735                        MapInputLayer::new(move |mut request: Request| {
736                            request.extensions().insert(stale_route.clone());
737                            let stale_egress = Extensions::new();
738                            stale_egress.insert(stale_route.clone());
739                            request.extensions().insert(Egress(stale_egress));
740                            request.headers_mut().insert(
741                                PROXY_AUTHORIZATION,
742                                HeaderValue::from_static("Basic downstream-secret"),
743                            );
744                            request
745                        }),
746                        MapOutputLayer::new({
747                            let observed_responses = observed_responses.clone();
748                            move |response: Response| {
749                                observed_responses.fetch_add(1, Ordering::Relaxed);
750                                response
751                            }
752                        }),
753                    ));
754                let request = Request::builder()
755                    .uri("http://origin.example/resource")
756                    .body(Body::empty())
757                    .unwrap();
758                request.extensions().insert(ProxyRoute::Proxy(
759                    "http://wrong:request-secret@requested.example:8080"
760                        .parse()
761                        .unwrap(),
762                ));
763                let result = client.serve(request).await;
764                if isolate && is_forward {
765                    assert!(result.is_err());
766                    assert_eq!(observed_responses.load(Ordering::Relaxed), 0);
767                } else {
768                    assert_eq!(
769                        result.unwrap().status(),
770                        StatusCode::PROXY_AUTHENTICATION_REQUIRED
771                    );
772                    assert_eq!(observed_responses.load(Ordering::Relaxed), 1);
773                }
774            }
775        }
776    }
777
778    #[test]
779    fn blocking_client_drives_the_composed_http_stack() {
780        let client = EasyHttpWebClient::connector_builder()
781            .with_custom_transport_connector(dummy_server())
782            .without_dns_connector()
783            .without_tls_proxy_support()
784            .without_proxy_support()
785            .without_tls_support()
786            .with_default_http_connector(Executor::default())
787            .without_connection_pool()
788            .build_client()
789            .try_into_blocking()
790            .unwrap();
791
792        let cloned = client.clone();
793        drop(client);
794        let response = cloned.get("http://example.com").send().unwrap();
795        assert_eq!(
796            response.try_into_json::<Output>().unwrap(),
797            Output { conn: 0, resp: 0 }
798        );
799    }
800
801    #[test]
802    fn default_blocking_http_client_is_cloneable_and_pooled() {
803        fn assert_default_client(_: &DefaultHttpWebClient) {}
804
805        let client = EasyHttpWebClient::try_blocking().unwrap();
806        assert_default_client(client.get_ref());
807        let cloned = client.clone();
808        drop(client);
809        let request = cloned.get("https://example.com").build().unwrap();
810        assert_eq!(request.uri(), &"https://example.com".parse().unwrap());
811    }
812
813    #[cfg(feature = "ws")]
814    #[test]
815    fn default_blocking_http_client_builds_websocket_requests() {
816        use crate::http::ws::handshake::client::BlockingHttpClientWebSocketExt as _;
817
818        let client = EasyHttpWebClient::try_blocking().unwrap();
819        let _from_url = client
820            .websocket("wss://example.com/chat")
821            .with_header("authorization", "Bearer secret");
822
823        let request = Request::builder()
824            .uri("wss://example.com/chat")
825            .body(Body::empty())
826            .unwrap();
827        let _from_request = client.websocket_with_request(request);
828    }
829
830    #[tokio::test]
831    async fn no_pool_tries_proxy_routes_in_order() {
832        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
833        let transport = service_fn({
834            let attempts = attempts.clone();
835            let direct = dummy_server::<ConnectRequest>();
836            move |input: ConnectRequest| {
837                let attempts = attempts.clone();
838                let direct = direct.clone();
839                async move {
840                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
841                    attempts.lock().push(route.clone());
842                    if route.proxy_address().is_some() {
843                        Err(ConnectionError::transport(
844                            BoxError::from_static_str("proxy unavailable"),
845                            ConnectionErrorKind::Unavailable,
846                        ))
847                    } else {
848                        direct.connect(input).await
849                    }
850                }
851            }
852        });
853        let client = EasyHttpWebClient::connector_builder()
854            .with_custom_transport_connector(transport)
855            .without_dns_connector()
856            .without_tls_proxy_support()
857            .with_custom_proxy_connector(())
858            .without_tls_support()
859            .with_default_http_connector(Executor::default())
860            .without_connection_pool()
861            .build_client();
862        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse::<ProxyAddress>().unwrap());
863        let request = || {
864            let request = Request::builder()
865                .uri("http://example.com")
866                .body(Body::empty())
867                .unwrap();
868            request
869                .extensions()
870                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
871            request
872        };
873
874        for _ in 0..2 {
875            client
876                .serve(request())
877                .await
878                .context("serve request through direct fallback")
879                .unwrap();
880        }
881
882        assert_eq!(
883            attempts.lock().as_slice(),
884            [proxy, ProxyRoute::Direct, ProxyRoute::Direct]
885        );
886    }
887
888    #[tokio::test]
889    async fn no_proxy_tls_support_rejects_https_proxy() {
890        let client = EasyHttpWebClient::connector_builder()
891            .with_custom_transport_connector(dummy_server())
892            .without_dns_connector()
893            .without_tls_proxy_support()
894            .with_proxy_support()
895            .without_tls_support()
896            .with_default_http_connector(Executor::default())
897            .without_connection_pool()
898            .build_client();
899        let request = Request::builder()
900            .uri("http://example.com")
901            .body(Body::empty())
902            .unwrap();
903        request.extensions().insert(ProxyRoutes::new([
904            ProxyRoute::Proxy(
905                "https://proxy.example:8443"
906                    .parse::<ProxyAddress>()
907                    .unwrap(),
908            ),
909            ProxyRoute::Direct,
910        ]));
911
912        let error =
913            ConnectionError::from(client.serve(request).await.unwrap_err().into_box_error());
914        assert_eq!(error.domain(), ConnectionErrorDomain::Transport);
915        assert_eq!(error.kind(), ConnectionErrorKind::Protocol);
916    }
917
918    #[tokio::test]
919    async fn easy_client_pools_plaintext_proxy_versions_separately() {
920        let proxy: ProxyAddress = "http://proxy.example:8080".parse().unwrap();
921        let dials = Arc::new(AtomicUsize::new(0));
922        let transport = service_fn({
923            let inner = dummy_server::<ConnectRequest>();
924            let proxy = proxy.clone();
925            let dials = dials.clone();
926            move |input: ConnectRequest| {
927                let inner = inner.clone();
928                let proxy = proxy.clone();
929                let dials = dials.clone();
930                async move {
931                    assert_eq!(
932                        input.extensions.get_ref::<ConnectorTarget>(),
933                        Some(&ConnectorTarget(proxy.address.clone())),
934                    );
935                    assert_eq!(
936                        input
937                            .extensions
938                            .get_ref::<ProxyRoute>()
939                            .and_then(ProxyRoute::proxy_address),
940                        Some(&proxy),
941                    );
942                    dials.fetch_add(1, Ordering::Relaxed);
943                    inner.connect(input).await
944                }
945            }
946        });
947        let client = EasyHttpWebClient::connector_builder()
948            .with_custom_transport_connector(transport)
949            .without_dns_connector()
950            .without_tls_proxy_support()
951            .with_proxy_support()
952            .without_tls_support()
953            .with_default_http_connector(Executor::default())
954            .with_default_connection_pool()
955            .build_client();
956
957        for (version, expected_conn) in [(Version::HTTP_11, 0), (Version::HTTP_2, 1)] {
958            let request = Request::builder()
959                .uri("http://example.com")
960                .version(version)
961                .body(Body::empty())
962                .unwrap();
963            request
964                .extensions()
965                .insert(ProxyRoutes::from(proxy.clone()));
966
967            let response = client.serve(request).await.unwrap();
968            assert_eq!(response.version(), version);
969            assert_eq!(
970                response.try_into_json::<Output>().await.unwrap(),
971                Output {
972                    conn: expected_conn,
973                    resp: 0,
974                },
975            );
976        }
977        assert_eq!(dials.load(Ordering::Relaxed), 2);
978    }
979
980    #[cfg(feature = "socks5")]
981    #[tokio::test]
982    async fn umbrella_proxy_connector_falls_back_across_supported_plan() {
983        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
984        let transport = service_fn({
985            let attempts = attempts.clone();
986            let direct = dummy_server::<ConnectRequest>();
987            move |input: ConnectRequest| {
988                let attempts = attempts.clone();
989                let direct = direct.clone();
990                async move {
991                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap().clone();
992                    attempts.lock().push(route.clone());
993                    if route.proxy_address().is_some() {
994                        Err(ConnectionError::transport(
995                            BoxError::from_static_str("proxy unavailable"),
996                            ConnectionErrorKind::Unavailable,
997                        ))
998                    } else {
999                        direct.connect(input).await
1000                    }
1001                }
1002            }
1003        });
1004        let client = EasyHttpWebClient::connector_builder()
1005            .with_custom_transport_connector(transport)
1006            .without_dns_connector()
1007            .without_tls_proxy_support()
1008            .with_proxy_support()
1009            .without_tls_support()
1010            .with_default_http_connector(Executor::default())
1011            .without_connection_pool()
1012            .build_client();
1013        let request = Request::builder()
1014            .uri("http://example.com")
1015            .body(Body::empty())
1016            .unwrap();
1017        let socks = ProxyRoute::Proxy(
1018            "socks5://socks.example:1080"
1019                .parse::<ProxyAddress>()
1020                .unwrap(),
1021        );
1022        let http = ProxyRoute::Proxy("http://http.example:8080".parse::<ProxyAddress>().unwrap());
1023        request.extensions().insert(ProxyRoutes::new([
1024            socks.clone(),
1025            http.clone(),
1026            ProxyRoute::Direct,
1027        ]));
1028
1029        let response = client.serve(request).await.unwrap();
1030        let output = response.try_into_json::<Output>().await.unwrap();
1031        assert_eq!(output, Output { conn: 0, resp: 0 });
1032        assert_eq!(
1033            attempts.lock().as_slice(),
1034            [socks, http, ProxyRoute::Direct]
1035        );
1036    }
1037
1038    #[tokio::test]
1039    async fn default_pool_caches_failed_route_and_reuses_selected_connection() {
1040        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
1041        let transport = service_fn({
1042            let attempts = attempts.clone();
1043            let direct = dummy_server::<ConnectRequest>();
1044            move |input: ConnectRequest| {
1045                let attempts = attempts.clone();
1046                let direct = direct.clone();
1047                async move {
1048                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
1049                    attempts.lock().push(route.clone());
1050                    if route.proxy_address().is_some() {
1051                        Err(ConnectionError::transport(
1052                            BoxError::from_static_str("proxy unavailable"),
1053                            ConnectionErrorKind::Unavailable,
1054                        ))
1055                    } else {
1056                        direct.connect(input).await
1057                    }
1058                }
1059            }
1060        });
1061        let client = EasyHttpWebClient::connector_builder()
1062            .with_custom_transport_connector(transport)
1063            .without_dns_connector()
1064            .without_tls_proxy_support()
1065            .with_custom_proxy_connector(())
1066            .without_tls_support()
1067            .with_default_http_connector(Executor::default())
1068            .with_default_connection_pool()
1069            .build_client();
1070        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse::<ProxyAddress>().unwrap());
1071        let request = || {
1072            let request = Request::builder()
1073                .uri("http://example.com")
1074                .body(Body::empty())
1075                .unwrap();
1076            request
1077                .extensions()
1078                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
1079            request
1080        };
1081
1082        for expected_response_index in 0..2 {
1083            let response = client.serve(request()).await.unwrap();
1084            let output = response.try_into_json::<Output>().await.unwrap();
1085            assert_eq!(output.conn, 0);
1086            assert_eq!(output.resp, expected_response_index);
1087        }
1088
1089        assert_eq!(attempts.lock().as_slice(), [proxy, ProxyRoute::Direct]);
1090    }
1091
1092    #[tokio::test]
1093    async fn easy_client_can_disable_proxy_route_failure_cache() {
1094        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
1095        let transport = service_fn({
1096            let attempts = attempts.clone();
1097            let direct = dummy_server::<ConnectRequest>();
1098            move |input: ConnectRequest| {
1099                let attempts = attempts.clone();
1100                let direct = direct.clone();
1101                async move {
1102                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
1103                    attempts.lock().push(route.clone());
1104                    if route.proxy_address().is_some() {
1105                        Err(ConnectionError::transport(
1106                            BoxError::from_static_str("proxy unavailable"),
1107                            ConnectionErrorKind::Unavailable,
1108                        ))
1109                    } else {
1110                        direct.connect(input).await
1111                    }
1112                }
1113            }
1114        });
1115        let client = EasyHttpWebClient::connector_builder()
1116            .with_custom_transport_connector(transport)
1117            .without_dns_connector()
1118            .without_tls_proxy_support()
1119            .with_custom_proxy_connector(())
1120            .without_tls_support()
1121            .with_default_http_connector(Executor::default())
1122            .without_proxy_route_failure_cache()
1123            .without_connection_pool()
1124            .build_client();
1125        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse().unwrap());
1126
1127        for _ in 0..2 {
1128            let request = Request::builder()
1129                .uri("http://example.com")
1130                .body(Body::empty())
1131                .unwrap();
1132            request
1133                .extensions()
1134                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
1135            client.serve(request).await.unwrap();
1136        }
1137
1138        assert_eq!(
1139            attempts.lock().as_slice(),
1140            [proxy.clone(), ProxyRoute::Direct, proxy, ProxyRoute::Direct]
1141        );
1142    }
1143
1144    #[tokio::test]
1145    async fn proxy_free_easy_client_omits_proxy_route_failure_cache() {
1146        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
1147        let transport = service_fn({
1148            let attempts = attempts.clone();
1149            let direct = dummy_server::<ConnectRequest>();
1150            move |input: ConnectRequest| {
1151                let attempts = attempts.clone();
1152                let direct = direct.clone();
1153                async move {
1154                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
1155                    attempts.lock().push(route.clone());
1156                    if route.proxy_address().is_some() {
1157                        Err(ConnectionError::transport(
1158                            BoxError::from_static_str("proxy unavailable"),
1159                            ConnectionErrorKind::Unavailable,
1160                        ))
1161                    } else {
1162                        direct.connect(input).await
1163                    }
1164                }
1165            }
1166        });
1167        let client = EasyHttpWebClient::connector_builder()
1168            .with_custom_transport_connector(transport)
1169            .without_dns_connector()
1170            .without_tls_proxy_support()
1171            .without_proxy_support()
1172            .without_tls_support()
1173            .with_default_http_connector(Executor::default())
1174            .without_connection_pool()
1175            .build_client();
1176        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse().unwrap());
1177
1178        for _ in 0..2 {
1179            let request = Request::builder()
1180                .uri("http://example.com")
1181                .body(Body::empty())
1182                .unwrap();
1183            request
1184                .extensions()
1185                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
1186            client.serve(request).await.unwrap();
1187        }
1188
1189        assert_eq!(
1190            attempts.lock().as_slice(),
1191            [proxy.clone(), ProxyRoute::Direct, proxy, ProxyRoute::Direct]
1192        );
1193    }
1194
1195    #[tokio::test]
1196    async fn easy_client_accepts_custom_proxy_route_failure_cache() {
1197        let attempts = Arc::new(parking_lot::Mutex::new(Vec::new()));
1198        let transport = service_fn({
1199            let attempts = attempts.clone();
1200            let direct = dummy_server::<ConnectRequest>();
1201            move |input: ConnectRequest| {
1202                let attempts = attempts.clone();
1203                let direct = direct.clone();
1204                async move {
1205                    let route = input.extensions.get_ref::<ProxyRoute>().unwrap();
1206                    attempts.lock().push(route.clone());
1207                    if route.proxy_address().is_some() {
1208                        Err(ConnectionError::transport(
1209                            BoxError::from_static_str("proxy unavailable"),
1210                            ConnectionErrorKind::Unavailable,
1211                        ))
1212                    } else {
1213                        direct.connect(input).await
1214                    }
1215                }
1216            }
1217        });
1218        let mut failure_cache_config = ProxyRouteFailureCacheConfig::default();
1219        failure_cache_config.scope = ProxyRouteFailureCacheScope::PerProxy;
1220        let failure_cache = ProxyRouteFailureCache::try_new(failure_cache_config).unwrap();
1221        let client = EasyHttpWebClient::connector_builder()
1222            .with_custom_transport_connector(transport)
1223            .without_dns_connector()
1224            .without_tls_proxy_support()
1225            .without_proxy_support()
1226            .without_tls_support()
1227            .with_default_http_connector(Executor::default())
1228            .with_proxy_route_failure_cache(failure_cache)
1229            .without_connection_pool()
1230            .build_client();
1231        let proxy = ProxyRoute::Proxy("http://proxy.example:8080".parse().unwrap());
1232
1233        for destination in ["one.example", "two.example"] {
1234            let request = Request::builder()
1235                .uri(format!("http://{destination}"))
1236                .body(Body::empty())
1237                .unwrap();
1238            request
1239                .extensions()
1240                .insert(ProxyRoutes::new([proxy.clone(), ProxyRoute::Direct]));
1241            client.serve(request).await.unwrap();
1242        }
1243
1244        assert_eq!(
1245            attempts.lock().as_slice(),
1246            [proxy, ProxyRoute::Direct, ProxyRoute::Direct]
1247        );
1248    }
1249
1250    #[cfg(all(feature = "rustls", any(feature = "aws-lc", feature = "ring")))]
1251    #[tokio::test]
1252    async fn rustls_https_proxy_alpn_is_scoped_across_connect() {
1253        use crate::{
1254            extensions::ExtensionsRef as _,
1255            net::{
1256                Protocol,
1257                address::HostWithPort,
1258                client::{EstablishedClientConnection, ProxyRoute},
1259                stream::service::EchoService,
1260            },
1261            tls::{
1262                client::{NegotiatedTlsParameters, TlsClientConfig},
1263                rustls::{client::TlsConnector, server::TlsAcceptorLayer},
1264                server::{GeneratedServerAuthConfig, ServerAuthData, TlsServerConfig},
1265            },
1266        };
1267        use rama_core::ServiceInput;
1268        use rama_crypto::cert::generate_server_auth;
1269        use rama_http::io::upgrade::handle_upgrade;
1270        use rama_http_backend::client::proxy::layer::HttpProxyConnectorLayer;
1271        use rama_net::http::TargetHttpVersion;
1272        use std::sync::Arc;
1273
1274        let (proxy_chain, proxy_key) =
1275            generate_server_auth(GeneratedServerAuthConfig::default()).expect("proxy auth");
1276        let proxy_trust = proxy_chain.last().expect("proxy trust anchor").clone();
1277        let (origin_chain, origin_key) =
1278            generate_server_auth(GeneratedServerAuthConfig::default()).expect("origin auth");
1279        let origin_trust = origin_chain.last().expect("origin trust anchor").clone();
1280
1281        let origin_server =
1282            TlsAcceptorLayer::new(TlsServerConfig::new().with_single_cert(ServerAuthData {
1283                cert_chain: origin_chain,
1284                private_key: origin_key,
1285                ocsp: None,
1286            }))
1287            .into_layer(EchoService::new());
1288        let (origin_done_tx, origin_done_rx) = tokio::sync::oneshot::channel();
1289        let origin_done_tx = Arc::new(parking_lot::Mutex::new(Some(origin_done_tx)));
1290
1291        let connect_version = Arc::new(parking_lot::Mutex::new(None));
1292        let observed_version = connect_version.clone();
1293        let proxy_http =
1294            HttpServer::auto(Executor::default()).service(service_fn(move |req: Request| {
1295                let origin_server = origin_server.clone();
1296                let origin_done_tx = origin_done_tx.clone();
1297                let observed_version = observed_version.clone();
1298                async move {
1299                    assert_eq!(req.method(), rama_http::Method::CONNECT);
1300                    *observed_version.lock() = Some(req.version());
1301                    let upgrade = handle_upgrade(&req);
1302                    tokio::spawn(async move {
1303                        let tunnel = upgrade.await.expect("server CONNECT upgrade");
1304                        // The client deliberately drops immediately after the
1305                        // handshake assertions, so the TLS server may finish
1306                        // with an EOF/close-notify error.
1307                        let _origin_result = origin_server.serve(tunnel).await;
1308                        if let Some(tx) = origin_done_tx.lock().take() {
1309                            tx.send(()).expect("origin completion receiver");
1310                        }
1311                    });
1312                    Ok::<_, Infallible>(Response::new(Body::empty()))
1313                }
1314            }));
1315        let proxy_server = TlsAcceptorLayer::new(
1316            TlsServerConfig::new()
1317                .with_single_cert(ServerAuthData {
1318                    cert_chain: proxy_chain,
1319                    private_key: proxy_key,
1320                    ocsp: None,
1321                })
1322                .with_alpn_http_2(),
1323        )
1324        .into_layer(proxy_http);
1325
1326        let (client_io, server_io) = tokio::io::duplex(64 * 1024);
1327        let client_io = Arc::new(parking_lot::Mutex::new(Some(client_io)));
1328        let transport = service_fn(move |input: ConnectRequest| {
1329            let conn = ServiceInput::new(client_io.lock().take().expect("one proxy connection"));
1330            async move { Ok::<_, ConnectionError>(EstablishedClientConnection { input, conn }) }
1331        });
1332
1333        let proxy_config = TlsClientConfig::new()
1334            .with_alpn_http_2()
1335            .with_server_name(crate::net::address::Host::from_static("localhost"))
1336            .try_with_server_trust_anchors([proxy_trust])
1337            .expect("proxy trust");
1338        let proxy_tls = TlsConnector::tunnel(transport, None).with_base_config(proxy_config);
1339        let proxy = HttpProxyConnectorLayer::default().into_layer(proxy_tls);
1340        let origin_config = TlsClientConfig::new()
1341            .with_alpn(Default::default())
1342            .with_server_name(crate::net::address::Host::from_static("localhost"))
1343            .try_with_server_trust_anchors([origin_trust])
1344            .expect("origin trust");
1345        let connector = TlsConnector::auto(proxy).with_base_config(origin_config);
1346
1347        let input = ConnectRequest::new(HostWithPort::try_from("localhost:443").unwrap())
1348            .with_application_protocol(Protocol::HTTPS);
1349        input
1350            .extensions
1351            .insert(ProxyRoute::Proxy("https://localhost:8443".parse().unwrap()));
1352        let client = async move {
1353            let established = Box::pin(connector.serve(input))
1354                .await
1355                .expect("two TLS handshakes");
1356
1357            assert_eq!(*connect_version.lock(), Some(Version::HTTP_2));
1358            assert_eq!(
1359                established
1360                    .conn
1361                    .extensions()
1362                    .get_ref::<NegotiatedTlsParameters>()
1363                    .expect("origin TLS parameters")
1364                    .application_layer_protocol,
1365                None
1366            );
1367            assert!(
1368                established
1369                    .conn
1370                    .extensions()
1371                    .get_ref::<TargetHttpVersion>()
1372                    .is_none(),
1373                "proxy HTTP/2 must not leak past CONNECT into a no-ALPN origin"
1374            );
1375            drop(established);
1376        };
1377        let (proxy_result, ()) =
1378            Box::pin(tokio::time::timeout(Duration::from_secs(5), async move {
1379                tokio::join!(proxy_server.serve(ServiceInput::new(server_io)), client)
1380            }))
1381            .await
1382            .expect("proxy/origin exchange");
1383        proxy_result.expect("proxy server");
1384        tokio::time::timeout(Duration::from_secs(5), origin_done_rx)
1385            .await
1386            .expect("origin server shutdown")
1387            .expect("origin completion signal");
1388    }
1389
1390    #[cfg(feature = "boring")]
1391    #[tokio::test]
1392    async fn boring_https_proxy_alpn_is_scoped_across_connect() {
1393        use crate::{
1394            extensions::ExtensionsRef as _,
1395            net::{
1396                Protocol,
1397                address::HostWithPort,
1398                client::{EstablishedClientConnection, ProxyRoute},
1399                stream::service::EchoService,
1400            },
1401            tls::{
1402                boring::{client::TlsConnector, server::TlsAcceptorLayer},
1403                client::{NegotiatedTlsParameters, TlsClientConfig},
1404                server::{GeneratedServerAuthConfig, ServerAuthData, TlsServerConfig},
1405            },
1406        };
1407        use rama_core::ServiceInput;
1408        use rama_crypto::cert::generate_server_auth;
1409        use rama_http::io::upgrade::handle_upgrade;
1410        use rama_http_backend::client::proxy::layer::HttpProxyConnectorLayer;
1411        use rama_net::http::TargetHttpVersion;
1412        use std::sync::Arc;
1413
1414        let (proxy_chain, proxy_key) =
1415            generate_server_auth(GeneratedServerAuthConfig::default()).expect("proxy auth");
1416        let proxy_trust = proxy_chain.last().expect("proxy trust anchor").clone();
1417        let (origin_chain, origin_key) =
1418            generate_server_auth(GeneratedServerAuthConfig::default()).expect("origin auth");
1419        let origin_trust = origin_chain.last().expect("origin trust anchor").clone();
1420
1421        let origin_server =
1422            TlsAcceptorLayer::new(TlsServerConfig::new().with_single_cert(ServerAuthData {
1423                cert_chain: origin_chain,
1424                private_key: origin_key,
1425                ocsp: None,
1426            }))
1427            .into_layer(EchoService::new());
1428        let (origin_done_tx, origin_done_rx) = tokio::sync::oneshot::channel();
1429        let origin_done_tx = Arc::new(parking_lot::Mutex::new(Some(origin_done_tx)));
1430
1431        let connect_version = Arc::new(parking_lot::Mutex::new(None));
1432        let observed_version = connect_version.clone();
1433        let proxy_http =
1434            HttpServer::auto(Executor::default()).service(service_fn(move |req: Request| {
1435                let origin_server = origin_server.clone();
1436                let origin_done_tx = origin_done_tx.clone();
1437                let observed_version = observed_version.clone();
1438                async move {
1439                    assert_eq!(req.method(), rama_http::Method::CONNECT);
1440                    *observed_version.lock() = Some(req.version());
1441                    let upgrade = handle_upgrade(&req);
1442                    tokio::spawn(async move {
1443                        let tunnel = upgrade.await.expect("server CONNECT upgrade");
1444                        // The client deliberately drops immediately after the
1445                        // handshake assertions, so the TLS server may finish
1446                        // with an EOF/close-notify error.
1447                        let _origin_result = origin_server.serve(tunnel).await;
1448                        if let Some(tx) = origin_done_tx.lock().take() {
1449                            tx.send(()).expect("origin completion receiver");
1450                        }
1451                    });
1452                    Ok::<_, Infallible>(Response::new(Body::empty()))
1453                }
1454            }));
1455        let proxy_server = TlsAcceptorLayer::new(
1456            TlsServerConfig::new()
1457                .with_single_cert(ServerAuthData {
1458                    cert_chain: proxy_chain,
1459                    private_key: proxy_key,
1460                    ocsp: None,
1461                })
1462                .with_alpn_http_2(),
1463        )
1464        .into_layer(proxy_http);
1465
1466        let (client_io, server_io) = tokio::io::duplex(64 * 1024);
1467        let client_io = Arc::new(parking_lot::Mutex::new(Some(client_io)));
1468        let transport = service_fn(move |input: ConnectRequest| {
1469            let conn = ServiceInput::new(client_io.lock().take().expect("one proxy connection"));
1470            async move { Ok::<_, ConnectionError>(EstablishedClientConnection { input, conn }) }
1471        });
1472
1473        let proxy_config = TlsClientConfig::new()
1474            .with_alpn_http_2()
1475            .with_server_name(crate::net::address::Host::from_static("localhost"))
1476            .try_with_server_trust_anchors([proxy_trust])
1477            .expect("proxy trust");
1478        let proxy_tls = TlsConnector::tunnel(transport, None).with_base_config(proxy_config);
1479        let proxy = HttpProxyConnectorLayer::default().into_layer(proxy_tls);
1480        let origin_config = TlsClientConfig::new()
1481            .with_alpn(Default::default())
1482            .with_server_name(crate::net::address::Host::from_static("localhost"))
1483            .try_with_server_trust_anchors([origin_trust])
1484            .expect("origin trust");
1485        let connector = TlsConnector::auto(proxy).with_base_config(origin_config);
1486
1487        let input = ConnectRequest::new(HostWithPort::try_from("localhost:443").unwrap())
1488            .with_application_protocol(Protocol::HTTPS);
1489        input
1490            .extensions
1491            .insert(ProxyRoute::Proxy("https://localhost:8443".parse().unwrap()));
1492        let client = async move {
1493            let established = connector.serve(input).await.expect("two TLS handshakes");
1494
1495            assert_eq!(*connect_version.lock(), Some(Version::HTTP_2));
1496            assert_eq!(
1497                established
1498                    .conn
1499                    .extensions()
1500                    .get_ref::<NegotiatedTlsParameters>()
1501                    .expect("origin TLS parameters")
1502                    .application_layer_protocol,
1503                None
1504            );
1505            assert!(
1506                established
1507                    .conn
1508                    .extensions()
1509                    .get_ref::<TargetHttpVersion>()
1510                    .is_none(),
1511                "proxy HTTP/2 must not leak past CONNECT into a no-ALPN origin"
1512            );
1513            drop(established);
1514        };
1515        let (proxy_result, ()) =
1516            Box::pin(tokio::time::timeout(Duration::from_secs(5), async move {
1517                tokio::join!(proxy_server.serve(ServiceInput::new(server_io)), client)
1518            }))
1519            .await
1520            .expect("proxy/origin exchange");
1521        proxy_result.expect("proxy server");
1522        tokio::time::timeout(Duration::from_secs(5), origin_done_rx)
1523            .await
1524            .expect("origin server shutdown")
1525            .expect("origin completion signal");
1526    }
1527
1528    #[cfg(feature = "boring")]
1529    #[tokio::test]
1530    async fn default_client_resumes_tls_sessions() {
1531        let exec = Executor::default();
1532        let resumed = Arc::new(parking_lot::Mutex::new(Vec::new()));
1533        let record = resumed.clone();
1534        let server = TlsAcceptorLayer::new(
1535            TlsServerConfig::new()
1536                .with_server_auth(
1537                    ServerAuthData::new_generated(GeneratedServerAuthConfig::default()).unwrap(),
1538                )
1539                .with_alpn_http_1(),
1540        )
1541        .with_session_resumption(true)
1542        .into_layer(
1543            MapInputLayer::new(move |stream: TlsStream<TcpStream>| {
1544                let params = stream.extensions().get_ref::<NegotiatedTlsParameters>();
1545                record.lock().push(params.and_then(|params| params.resumed));
1546                stream
1547            })
1548            .into_layer(HttpServer::new_http1(exec.clone()).service(service_fn(
1549                async |_: Request| {
1550                    // A fresh connection per request, so each one handshakes.
1551                    Ok::<_, Infallible>(
1552                        Response::builder()
1553                            .header(CONNECTION, "close")
1554                            .body(Body::empty())
1555                            .unwrap(),
1556                    )
1557                },
1558            ))),
1559        );
1560        let listener = TcpListener::bind_address(SocketAddress::local_ipv4(0), exec)
1561            .await
1562            .unwrap();
1563        let port = listener.local_addr().unwrap().port();
1564        let server = tokio::spawn(listener.serve(server));
1565
1566        let client = DefaultHttpWebClient::<Body>::default();
1567        for _ in 0..2 {
1568            let request = Request::builder()
1569                .uri(format!("https://localhost:{port}/"))
1570                .body(Body::empty())
1571                .unwrap();
1572            request
1573                .extensions()
1574                .insert(TlsServerVerify(ServerVerifyMode::Disable));
1575            let response = tokio::time::timeout(Duration::from_secs(5), client.serve(request))
1576                .await
1577                .unwrap()
1578                .unwrap();
1579            response.try_into_string().await.unwrap();
1580        }
1581        server.abort();
1582        assert_eq!(*resumed.lock(), [Some(false), Some(true)]);
1583    }
1584
1585    #[cfg(feature = "boring")]
1586    #[test]
1587    fn proxy_failure_cache_keeps_tls_client_future_bounded() {
1588        let client = EasyHttpWebClient::connector_builder()
1589            .with_default_transport_connector()
1590            .with_default_dns_connector()
1591            .without_tls_proxy_support()
1592            .with_proxy_support()
1593            .with_tls_support_using_boringssl_and_default_http_version(
1594                crate::tls::client::TlsClientConfig::default_http(),
1595                Version::HTTP_11,
1596            )
1597            .with_default_http_connector(Executor::default())
1598            .without_connection_pool()
1599            .build_client();
1600        let request = Request::builder()
1601            .uri("https://example.com")
1602            .body(Body::empty())
1603            .unwrap();
1604
1605        let future = client.serve(request);
1606        let future_size = std::mem::size_of_val(&future);
1607
1608        assert!(
1609            future_size < 64 * 1024,
1610            "easy TLS client future is unexpectedly large: {future_size} bytes"
1611        );
1612    }
1613
1614    #[tokio::test]
1615    async fn connection_is_in_use_until_response_body_is_consumed() {
1616        let client = EasyHttpWebClient::connector_builder()
1617            .with_custom_transport_connector(dummy_server())
1618            .without_dns_connector()
1619            .without_tls_proxy_support()
1620            .without_proxy_support()
1621            .without_tls_support()
1622            .with_default_http_connector(Executor::default())
1623            .with_connection_pool(HttpPooledConnectorConfig {
1624                max_streams_per_connection: NonZeroUsize::new(1),
1625                max_connections_total: NonZeroUsize::new(4),
1626                ..Default::default()
1627            })
1628            .build_client();
1629
1630        let req = || {
1631            Request::builder()
1632                .uri("http://example.com")
1633                .version(Version::HTTP_2)
1634                .body(Body::empty())
1635                .unwrap()
1636        };
1637
1638        // Get the first response but DO NOT consume its body yet: the connection
1639        // is logically still in use until the body is drained. Then issue a second
1640        // request before draining the first.
1641        let res1 = client.serve(req()).await.unwrap();
1642        let res2 = client.serve(req()).await.unwrap();
1643
1644        // Drain in reverse so `res1`'s body is still outstanding when `req2` runs.
1645        let out2 = res2.try_into_json::<Output>().await.unwrap();
1646        let out1 = res1.try_into_json::<Output>().await.unwrap();
1647
1648        assert_eq!(out1.conn, 0, "first request uses the first connection");
1649        // With `max_concurrent_streams = 1`, connection 0's response body is still
1650        // in flight, so the second request must NOT reuse it.
1651        assert_eq!(
1652            out2.conn, 1,
1653            "second request must not reuse a connection whose response body is still in flight"
1654        );
1655    }
1656
1657    // These things are already tested inside the pool itself, but here we add some high level tests
1658    // in case we ever swap the underlying pool implementation.
1659
1660    #[tokio::test]
1661    async fn default_pool_multiplexes_on_h2() {
1662        let client = EasyHttpWebClient::connector_builder()
1663            .with_custom_transport_connector(dummy_server())
1664            .without_dns_connector()
1665            .without_tls_proxy_support()
1666            .without_proxy_support()
1667            .without_tls_support()
1668            .with_default_http_connector(Executor::default())
1669            .with_default_connection_pool()
1670            .build_client();
1671
1672        let req = || {
1673            Request::builder()
1674                .uri("http://example.com")
1675                .version(Version::HTTP_2)
1676                .body(Body::empty())
1677                .unwrap()
1678        };
1679        let (res1, res2, res3) = tokio::join!(
1680            client.serve(req()),
1681            client.serve(req()),
1682            client.serve(req()),
1683        );
1684
1685        // Should only create single connection and send all requests over the same one
1686        for (i, res) in [res1, res2, res3].into_iter().enumerate() {
1687            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
1688            assert_eq!(out.conn, 0);
1689            assert_eq!(out.resp, i);
1690        }
1691    }
1692
1693    #[tokio::test]
1694    async fn default_pool_does_not_multiplexes_on_h1() {
1695        let client = EasyHttpWebClient::connector_builder()
1696            .with_custom_transport_connector(dummy_server())
1697            .without_dns_connector()
1698            .without_tls_proxy_support()
1699            .without_proxy_support()
1700            .without_tls_support()
1701            .with_default_http_connector(Executor::default())
1702            .with_default_connection_pool()
1703            .build_client();
1704
1705        let req = || {
1706            Request::builder()
1707                .uri("http://example.com")
1708                .version(Version::HTTP_11)
1709                .body(Body::empty())
1710                .unwrap()
1711        };
1712        let (res1, res2, res3) = tokio::join!(
1713            client.serve(req()),
1714            client.serve(req()),
1715            client.serve(req()),
1716        );
1717
1718        // Should create a new connection for each request since they are all inprogress at the same
1719        // time and h1 does not support multiplexing
1720        for (i, res) in [res1, res2, res3].into_iter().enumerate() {
1721            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
1722            assert_eq!(out.conn, i);
1723            assert_eq!(out.resp, 0);
1724        }
1725    }
1726
1727    #[tokio::test]
1728    async fn multiplex_on_h2_respects_limits() {
1729        let client = EasyHttpWebClient::connector_builder()
1730            .with_custom_transport_connector(dummy_server())
1731            .without_dns_connector()
1732            .without_tls_proxy_support()
1733            .without_proxy_support()
1734            .without_tls_support()
1735            .with_default_http_connector(Executor::default())
1736            .with_connection_pool(HttpPooledConnectorConfig {
1737                max_streams_per_connection: NonZeroUsize::new(2),
1738                ..Default::default()
1739            })
1740            .build_client();
1741
1742        let req = || {
1743            Request::builder()
1744                .uri("http://example.com")
1745                .version(Version::HTTP_2)
1746                .body(Body::empty())
1747                .unwrap()
1748        };
1749        let (res1, res2, res3, res4) = tokio::join!(
1750            client.serve(req()),
1751            client.serve(req()),
1752            client.serve(req()),
1753            client.serve(req()),
1754        );
1755
1756        // Should create a connection for every two request
1757        for (i, res) in [res1, res2, res3, res4].into_iter().enumerate() {
1758            let out = res.unwrap().try_into_json::<Output>().await.unwrap();
1759            assert_eq!(out.conn, i / 2);
1760            assert_eq!(out.resp, i % 2);
1761        }
1762    }
1763}
rama/http/client/mod.rs

rama/http/client/
mod.rs
