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

rama/http/client/
mod.rs
