aws_smithy_http_client/
hyper_legacy.rs

1/*
2 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3 * SPDX-License-Identifier: Apache-2.0
4 */
5
6use crate::hyper_legacy::timeout_middleware::HttpTimeoutError;
7use aws_smithy_async::future::timeout::TimedOutError;
8use aws_smithy_async::rt::sleep::{default_async_sleep, AsyncSleep, SharedAsyncSleep};
9use aws_smithy_runtime_api::box_error::BoxError;
10use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
11use aws_smithy_runtime_api::client::connection::ConnectionMetadata;
12use aws_smithy_runtime_api::client::connector_metadata::ConnectorMetadata;
13use aws_smithy_runtime_api::client::http::{
14    HttpClient, HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpClient,
15    SharedHttpConnector,
16};
17use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};
18use aws_smithy_runtime_api::client::result::ConnectorError;
19use aws_smithy_runtime_api::client::runtime_components::{
20    RuntimeComponents, RuntimeComponentsBuilder,
21};
22use aws_smithy_runtime_api::shared::IntoShared;
23use aws_smithy_types::body::SdkBody;
24use aws_smithy_types::config_bag::ConfigBag;
25use aws_smithy_types::error::display::DisplayErrorContext;
26use aws_smithy_types::retry::ErrorKind;
27use h2_0_3::Reason;
28use hyper_0_14::client::connect::{capture_connection, CaptureConnection, Connection, HttpInfo};
29use std::borrow::Cow;
30use std::collections::HashMap;
31use std::error::Error;
32use std::fmt;
33use std::sync::RwLock;
34use std::time::Duration;
35use tokio::io::{AsyncRead, AsyncWrite};
36
37#[cfg(feature = "legacy-rustls-ring")]
38mod default_connector {
39    use aws_smithy_async::rt::sleep::SharedAsyncSleep;
40    use aws_smithy_runtime_api::client::http::HttpConnectorSettings;
41    use legacy_hyper_rustls as hyper_rustls;
42    use legacy_rustls as rustls;
43    use std::sync::LazyLock;
44
45    // Creating a `with_native_roots` HTTP client takes 300ms on OS X. Cache this so that we
46    // don't need to repeatedly incur that cost.
47    pub(crate) static HTTPS_NATIVE_ROOTS: LazyLock<
48        hyper_rustls::HttpsConnector<hyper_0_14::client::HttpConnector>,
49    > = LazyLock::new(default_tls);
50
51    fn default_tls() -> hyper_rustls::HttpsConnector<hyper_0_14::client::HttpConnector> {
52        use legacy_rustls::client::WantsTransparencyPolicyOrClientCert;
53        use legacy_rustls::{ClientConfig, ConfigBuilder, WantsVerifier};
54        use rustls_native_certs;
55        // polyfill with_native_roots from https://docs.rs/hyper-rustls/0.24.2/src/hyper_rustls/config.rs.html#22-70
56        // to use the new rustls_native_certs, since rustls_native_certs 0.6 depends on rustls-pemfile which is deprecated
57        fn with_native_roots(
58            this: ConfigBuilder<ClientConfig, WantsVerifier>,
59        ) -> ConfigBuilder<ClientConfig, WantsTransparencyPolicyOrClientCert> {
60            let mut roots = rustls::RootCertStore::empty();
61            let mut valid_count = 0;
62            let mut invalid_count = 0;
63
64            for cert in
65                rustls_native_certs::load_native_certs().expect("could not load platform certs")
66            {
67                let cert = rustls::Certificate(cert.to_vec());
68                match roots.add(&cert) {
69                    Ok(_) => valid_count += 1,
70                    Err(err) => {
71                        tracing::trace!("invalid cert der {:?}", cert.0);
72                        tracing::debug!("certificate parsing failed: {:?}", err);
73                        invalid_count += 1
74                    }
75                }
76            }
77            tracing::debug!(
78                "with_native_roots processed {} valid and {} invalid certs",
79                valid_count,
80                invalid_count
81            );
82            assert!(!roots.is_empty(), "no CA certificates found");
83
84            this.with_root_certificates(roots)
85        }
86        hyper_rustls::HttpsConnectorBuilder::new()
87               .with_tls_config(
88                with_native_roots(rustls::ClientConfig::builder()
89                    .with_cipher_suites(&[
90                        // TLS1.3 suites
91                        rustls::cipher_suite::TLS13_AES_256_GCM_SHA384,
92                        rustls::cipher_suite::TLS13_AES_128_GCM_SHA256,
93                        // TLS1.2 suites
94                        rustls::cipher_suite::TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
95                        rustls::cipher_suite::TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
96                        rustls::cipher_suite::TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
97                        rustls::cipher_suite::TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
98                        rustls::cipher_suite::TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256,
99                    ])
100                    .with_safe_default_kx_groups()
101                    .with_safe_default_protocol_versions()
102                    .expect("Error with the TLS configuration. Please file a bug report under https://github.com/smithy-lang/smithy-rs/issues."))
103                    .with_no_client_auth()
104            )
105            .https_or_http()
106            .enable_http1()
107            .enable_http2()
108            .build()
109    }
110
111    pub(super) fn base(
112        settings: &HttpConnectorSettings,
113        sleep: Option<SharedAsyncSleep>,
114    ) -> super::HyperConnectorBuilder {
115        let mut hyper = super::HyperConnector::builder().connector_settings(settings.clone());
116        if let Some(sleep) = sleep {
117            hyper = hyper.sleep_impl(sleep);
118        }
119        hyper
120    }
121
122    /// Return a default HTTPS connector backed by the `rustls` crate.
123    ///
124    /// It requires a minimum TLS version of 1.2.
125    /// It allows you to connect to both `http` and `https` URLs.
126    pub(super) fn https() -> hyper_rustls::HttpsConnector<hyper_0_14::client::HttpConnector> {
127        HTTPS_NATIVE_ROOTS.clone()
128    }
129}
130
131/// Given `HttpConnectorSettings` and an `SharedAsyncSleep`, create a `SharedHttpConnector` from defaults depending on what cargo features are activated.
132pub fn default_connector(
133    settings: &HttpConnectorSettings,
134    sleep: Option<SharedAsyncSleep>,
135) -> Option<SharedHttpConnector> {
136    #[cfg(feature = "legacy-rustls-ring")]
137    {
138        tracing::trace!(settings = ?settings, sleep = ?sleep, "creating a new default connector");
139        let hyper = default_connector::base(settings, sleep).build_https();
140        Some(SharedHttpConnector::new(hyper))
141    }
142    #[cfg(not(feature = "legacy-rustls-ring"))]
143    {
144        tracing::trace!(settings = ?settings, sleep = ?sleep, "no default connector available");
145        None
146    }
147}
148
149/// Creates a hyper-backed HTTPS client from defaults depending on what cargo features are activated.
150pub fn default_client() -> Option<SharedHttpClient> {
151    #[cfg(feature = "legacy-rustls-ring")]
152    {
153        tracing::trace!("creating a new default hyper 0.14.x client");
154        Some(HyperClientBuilder::new().build_https())
155    }
156    #[cfg(not(feature = "legacy-rustls-ring"))]
157    {
158        tracing::trace!("no default connector available");
159        None
160    }
161}
162
163/// [`HttpConnector`] that uses [`hyper_0_14`] to make HTTP requests.
164///
165/// This connector also implements socket connect and read timeouts.
166///
167/// This shouldn't be used directly in most cases.
168/// See the docs on [`HyperClientBuilder`] for examples of how
169/// to customize the Hyper client.
170#[derive(Debug)]
171pub struct HyperConnector {
172    adapter: Box<dyn HttpConnector>,
173}
174
175impl HyperConnector {
176    /// Builder for a Hyper connector.
177    pub fn builder() -> HyperConnectorBuilder {
178        Default::default()
179    }
180}
181
182impl HttpConnector for HyperConnector {
183    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
184        self.adapter.call(request)
185    }
186}
187
188/// Builder for [`HyperConnector`].
189#[derive(Default, Debug)]
190pub struct HyperConnectorBuilder {
191    connector_settings: Option<HttpConnectorSettings>,
192    sleep_impl: Option<SharedAsyncSleep>,
193    client_builder: Option<hyper_0_14::client::Builder>,
194}
195
196impl HyperConnectorBuilder {
197    /// Create a [`HyperConnector`] from this builder and a given connector.
198    pub fn build<C>(self, tcp_connector: C) -> HyperConnector
199    where
200        C: Clone + Send + Sync + 'static,
201        C: hyper_0_14::service::Service<http_02x::Uri>,
202        C::Response: Connection + AsyncRead + AsyncWrite + Send + Unpin + 'static,
203        C::Future: Unpin + Send + 'static,
204        C::Error: Into<BoxError>,
205    {
206        let client_builder = self.client_builder.unwrap_or_default();
207        let sleep_impl = self.sleep_impl.or_else(default_async_sleep);
208        let (connect_timeout, read_timeout) = self
209            .connector_settings
210            .map(|c| (c.connect_timeout(), c.read_timeout()))
211            .unwrap_or((None, None));
212
213        let connector = match connect_timeout {
214            Some(duration) => timeout_middleware::ConnectTimeout::new(
215                tcp_connector,
216                sleep_impl
217                    .clone()
218                    .expect("a sleep impl must be provided in order to have a connect timeout"),
219                duration,
220            ),
221            None => timeout_middleware::ConnectTimeout::no_timeout(tcp_connector),
222        };
223        let base = client_builder.build(connector);
224        let read_timeout = match read_timeout {
225            Some(duration) => timeout_middleware::HttpReadTimeout::new(
226                base,
227                sleep_impl.expect("a sleep impl must be provided in order to have a read timeout"),
228                duration,
229            ),
230            None => timeout_middleware::HttpReadTimeout::no_timeout(base),
231        };
232        HyperConnector {
233            adapter: Box::new(Adapter {
234                client: read_timeout,
235            }),
236        }
237    }
238
239    /// Create a [`HyperConnector`] with the default rustls HTTPS implementation.
240    #[cfg(feature = "legacy-rustls-ring")]
241    pub fn build_https(self) -> HyperConnector {
242        self.build(default_connector::https())
243    }
244
245    /// Set the async sleep implementation used for timeouts
246    ///
247    /// Calling this is only necessary for testing or to use something other than
248    /// [`default_async_sleep`].
249    pub fn sleep_impl(mut self, sleep_impl: impl AsyncSleep + 'static) -> Self {
250        self.sleep_impl = Some(sleep_impl.into_shared());
251        self
252    }
253
254    /// Set the async sleep implementation used for timeouts
255    ///
256    /// Calling this is only necessary for testing or to use something other than
257    /// [`default_async_sleep`].
258    pub fn set_sleep_impl(&mut self, sleep_impl: Option<SharedAsyncSleep>) -> &mut Self {
259        self.sleep_impl = sleep_impl;
260        self
261    }
262
263    /// Configure the HTTP settings for the `HyperAdapter`
264    pub fn connector_settings(mut self, connector_settings: HttpConnectorSettings) -> Self {
265        self.connector_settings = Some(connector_settings);
266        self
267    }
268
269    /// Configure the HTTP settings for the `HyperAdapter`
270    pub fn set_connector_settings(
271        &mut self,
272        connector_settings: Option<HttpConnectorSettings>,
273    ) -> &mut Self {
274        self.connector_settings = connector_settings;
275        self
276    }
277
278    /// Override the Hyper client [`Builder`](hyper_0_14::client::Builder) used to construct this client.
279    ///
280    /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.
281    pub fn hyper_builder(mut self, hyper_builder: hyper_0_14::client::Builder) -> Self {
282        self.client_builder = Some(hyper_builder);
283        self
284    }
285
286    /// Override the Hyper client [`Builder`](hyper_0_14::client::Builder) used to construct this client.
287    ///
288    /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.
289    pub fn set_hyper_builder(
290        &mut self,
291        hyper_builder: Option<hyper_0_14::client::Builder>,
292    ) -> &mut Self {
293        self.client_builder = hyper_builder;
294        self
295    }
296}
297
298/// Adapter from a [`hyper_0_14::Client`] to [`HttpConnector`].
299///
300/// This adapter also enables TCP `CONNECT` and HTTP `READ` timeouts via [`HyperConnector::builder`].
301struct Adapter<C> {
302    client: timeout_middleware::HttpReadTimeout<
303        hyper_0_14::Client<timeout_middleware::ConnectTimeout<C>, SdkBody>,
304    >,
305}
306
307impl<C> fmt::Debug for Adapter<C> {
308    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
309        f.debug_struct("Adapter")
310            .field("client", &"** hyper client **")
311            .finish()
312    }
313}
314
315/// Extract a smithy connection from a hyper CaptureConnection
316fn extract_smithy_connection(capture_conn: &CaptureConnection) -> Option<ConnectionMetadata> {
317    let capture_conn = capture_conn.clone();
318    if let Some(conn) = capture_conn.clone().connection_metadata().as_ref() {
319        let mut extensions = http_02x::Extensions::new();
320        conn.get_extras(&mut extensions);
321        let http_info = extensions.get::<HttpInfo>();
322        let mut builder = ConnectionMetadata::builder()
323            .proxied(conn.is_proxied())
324            .poison_fn(move || match capture_conn.connection_metadata().as_ref() {
325                Some(conn) => conn.poison(),
326                None => tracing::trace!("no connection existed to poison"),
327            });
328
329        builder
330            .set_local_addr(http_info.map(|info| info.local_addr()))
331            .set_remote_addr(http_info.map(|info| info.remote_addr()));
332
333        let smithy_connection = builder.build();
334
335        Some(smithy_connection)
336    } else {
337        None
338    }
339}
340
341impl<C> HttpConnector for Adapter<C>
342where
343    C: Clone + Send + Sync + 'static,
344    C: hyper_0_14::service::Service<http_02x::Uri>,
345    C::Response: Connection + AsyncRead + AsyncWrite + Send + Unpin + 'static,
346    C::Future: Unpin + Send + 'static,
347    C::Error: Into<BoxError>,
348{
349    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
350        use hyper_0_14::service::Service;
351
352        let mut request = match request.try_into_http02x() {
353            Ok(request) => request,
354            Err(err) => {
355                return HttpConnectorFuture::ready(Err(ConnectorError::other(err.into(), None)));
356            }
357        };
358        let capture_connection = capture_connection(&mut request);
359        if let Some(capture_smithy_connection) =
360            request.extensions().get::<CaptureSmithyConnection>()
361        {
362            capture_smithy_connection
363                .set_connection_retriever(move || extract_smithy_connection(&capture_connection));
364        }
365        let mut client = self.client.clone();
366        let fut = client.call(request);
367        HttpConnectorFuture::new(async move {
368            match fut.await {
369                Ok(response) => HttpResponse::try_from(response.map(SdkBody::from_body_0_4))
370                    .map_err(|err| ConnectorError::other(err.into(), None)),
371                Err(err) => Err(downcast_error(err)),
372            }
373        })
374    }
375}
376
377/// Downcast errors coming out of hyper into an appropriate `ConnectorError`
378fn downcast_error(err: BoxError) -> ConnectorError {
379    // is a `TimedOutError` (from aws_smithy_async::timeout) in the chain? if it is, this is a timeout
380    if find_source::<TimedOutError>(err.as_ref()).is_some() {
381        return ConnectorError::timeout(err);
382    }
383    // is the top of chain error actually already a `ConnectorError`? return that directly
384    let err = match err.downcast::<ConnectorError>() {
385        Ok(connector_error) => return *connector_error,
386        Err(box_error) => box_error,
387    };
388    // generally, the top of chain will probably be a hyper error. Go through a set of hyper specific
389    // error classifications
390    let err = match err.downcast::<hyper_0_14::Error>() {
391        Ok(hyper_error) => return to_connector_error(*hyper_error),
392        Err(box_error) => box_error,
393    };
394
395    // otherwise, we have no idea!
396    ConnectorError::other(err, None)
397}
398
399/// Convert a [`hyper_0_14::Error`] into a [`ConnectorError`]
400fn to_connector_error(err: hyper_0_14::Error) -> ConnectorError {
401    if err.is_timeout() || find_source::<HttpTimeoutError>(&err).is_some() {
402        return ConnectorError::timeout(err.into());
403    }
404    if err.is_user() {
405        return ConnectorError::user(err.into());
406    }
407    if err.is_closed() || err.is_canceled() || find_source::<std::io::Error>(&err).is_some() {
408        return ConnectorError::io(err.into());
409    }
410    // We sometimes receive this from S3: hyper::Error(IncompleteMessage)
411    if err.is_incomplete_message() {
412        return ConnectorError::other(err.into(), Some(ErrorKind::TransientError));
413    }
414    if let Some(h2_err) = find_source::<h2_0_3::Error>(&err) {
415        if h2_err.is_go_away()
416            || (h2_err.is_reset() && h2_err.reason() == Some(Reason::REFUSED_STREAM))
417        {
418            return ConnectorError::io(err.into());
419        }
420    }
421
422    tracing::warn!(err = %DisplayErrorContext(&err), "unrecognized error from Hyper. If this error should be retried, please file an issue.");
423    ConnectorError::other(err.into(), None)
424}
425
426fn find_source<'a, E: Error + 'static>(err: &'a (dyn Error + 'static)) -> Option<&'a E> {
427    let mut next = Some(err);
428    while let Some(err) = next {
429        if let Some(matching_err) = err.downcast_ref::<E>() {
430            return Some(matching_err);
431        }
432        next = err.source();
433    }
434    None
435}
436
437#[derive(Clone, Debug, Eq, PartialEq, Hash)]
438struct CacheKey {
439    connect_timeout: Option<Duration>,
440    read_timeout: Option<Duration>,
441}
442
443impl From<&HttpConnectorSettings> for CacheKey {
444    fn from(value: &HttpConnectorSettings) -> Self {
445        Self {
446            connect_timeout: value.connect_timeout(),
447            read_timeout: value.read_timeout(),
448        }
449    }
450}
451
452struct HyperClient<F> {
453    connector_cache: RwLock<HashMap<CacheKey, SharedHttpConnector>>,
454    client_builder: hyper_0_14::client::Builder,
455    tcp_connector_fn: F,
456}
457
458impl<F> fmt::Debug for HyperClient<F> {
459    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
460        f.debug_struct("HyperClient")
461            .field("connector_cache", &self.connector_cache)
462            .field("client_builder", &self.client_builder)
463            .finish()
464    }
465}
466
467impl<C, F> HttpClient for HyperClient<F>
468where
469    F: Fn() -> C + Send + Sync,
470    C: Clone + Send + Sync + 'static,
471    C: hyper_0_14::service::Service<http_02x::Uri>,
472    C::Response: Connection + AsyncRead + AsyncWrite + Send + Unpin + 'static,
473    C::Future: Unpin + Send + 'static,
474    C::Error: Into<BoxError>,
475{
476    fn http_connector(
477        &self,
478        settings: &HttpConnectorSettings,
479        components: &RuntimeComponents,
480    ) -> SharedHttpConnector {
481        let key = CacheKey::from(settings);
482        let mut connector = self.connector_cache.read().unwrap().get(&key).cloned();
483        if connector.is_none() {
484            let mut cache = self.connector_cache.write().unwrap();
485            // Short-circuit if another thread already wrote a connector to the cache for this key
486            if !cache.contains_key(&key) {
487                let mut builder = HyperConnector::builder()
488                    .hyper_builder(self.client_builder.clone())
489                    .connector_settings(settings.clone());
490                builder.set_sleep_impl(components.sleep_impl());
491                let start = components.time_source().map(|ts| ts.now());
492                let tcp_connector = (self.tcp_connector_fn)();
493                let end = components.time_source().map(|ts| ts.now());
494                if let (Some(start), Some(end)) = (start, end) {
495                    if let Ok(elapsed) = end.duration_since(start) {
496                        tracing::debug!("new TCP connector created in {:?}", elapsed);
497                    }
498                }
499                let connector = SharedHttpConnector::new(builder.build(tcp_connector));
500                cache.insert(key.clone(), connector);
501            }
502            connector = cache.get(&key).cloned();
503        }
504
505        connector.expect("cache populated above")
506    }
507
508    fn validate_base_client_config(
509        &self,
510        _: &RuntimeComponentsBuilder,
511        _: &ConfigBag,
512    ) -> Result<(), BoxError> {
513        // Initialize the TCP connector at this point so that native certs load
514        // at client initialization time instead of upon first request. We do it
515        // here rather than at construction so that it won't run if this is not
516        // the selected HTTP client for the base config (for example, if this was
517        // the default HTTP client, and it was overridden by a later plugin).
518        let _ = (self.tcp_connector_fn)();
519        Ok(())
520    }
521
522    fn connector_metadata(&self) -> Option<ConnectorMetadata> {
523        Some(ConnectorMetadata::new("hyper", Some(Cow::Borrowed("0.x"))))
524    }
525}
526
527/// Builder for a hyper-backed [`HttpClient`] implementation.
528///
529/// This builder can be used to customize the underlying TCP connector used, as well as
530/// hyper client configuration.
531///
532/// # Examples
533///
534/// Construct a Hyper client with the default TLS implementation (rustls).
535/// This can be useful when you want to share a Hyper connector between multiple
536/// generated Smithy clients.
537///
538/// ```no_run,ignore
539/// use aws_smithy_http_client::hyper_014::HyperClientBuilder;
540///
541/// let http_client = HyperClientBuilder::new().build_https();
542///
543/// // This connector can then be given to a generated service Config
544/// let config = my_service_client::Config::builder()
545///     .endpoint_url("http://localhost:1234")
546///     .http_client(http_client)
547///     .build();
548/// let client = my_service_client::Client::from_conf(config);
549/// ```
550///
551/// ## Use a Hyper client with WebPKI roots
552///
553/// A use case for where you may want to use the [`HyperClientBuilder`] is when
554/// setting Hyper client settings that aren't otherwise exposed by the `Config`
555/// builder interface. Some examples include changing:
556///
557/// - Hyper client settings
558/// - Allowed TLS cipher suites
559/// - Using an alternative TLS connector library (not the default, rustls)
560/// - CA trust root certificates (illustrated using WebPKI below)
561///
562/// ```no_run,ignore
563/// use aws_smithy_http_client::hyper_014::HyperClientBuilder;
564///
565/// let https_connector = hyper_rustls::HttpsConnectorBuilder::new()
566///     .with_webpki_roots()
567///     .https_only()
568///     .enable_http1()
569///     .enable_http2()
570///     .build();
571/// let http_client = HyperClientBuilder::new().build(https_connector);
572///
573/// // This connector can then be given to a generated service Config
574/// let config = my_service_client::Config::builder()
575///     .endpoint_url("https://example.com")
576///     .http_client(http_client)
577///     .build();
578/// let client = my_service_client::Client::from_conf(config);
579/// ```
580#[derive(Clone, Default, Debug)]
581pub struct HyperClientBuilder {
582    client_builder: Option<hyper_0_14::client::Builder>,
583}
584
585impl HyperClientBuilder {
586    /// Creates a new builder.
587    pub fn new() -> Self {
588        Self::default()
589    }
590
591    /// Override the Hyper client [`Builder`](hyper_0_14::client::Builder) used to construct this client.
592    ///
593    /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.
594    pub fn hyper_builder(mut self, hyper_builder: hyper_0_14::client::Builder) -> Self {
595        self.client_builder = Some(hyper_builder);
596        self
597    }
598
599    /// Override the Hyper client [`Builder`](hyper_0_14::client::Builder) used to construct this client.
600    ///
601    /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.
602    pub fn set_hyper_builder(
603        &mut self,
604        hyper_builder: Option<hyper_0_14::client::Builder>,
605    ) -> &mut Self {
606        self.client_builder = hyper_builder;
607        self
608    }
609
610    /// Create a hyper client with the default rustls HTTPS implementation.
611    ///
612    /// The trusted certificates will be loaded later when this becomes the selected
613    /// HTTP client for a Smithy client.
614    #[cfg(feature = "legacy-rustls-ring")]
615    pub fn build_https(self) -> SharedHttpClient {
616        self.build_with_fn(default_connector::https)
617    }
618
619    /// Create a [`SharedHttpClient`] from this builder and a given connector.
620    ///
621    #[cfg_attr(
622        feature = "legacy-rustls-ring",
623        doc = "Use [`build_https`](HyperClientBuilder::build_https) if you don't want to provide a custom TCP connector."
624    )]
625    pub fn build<C>(self, tcp_connector: C) -> SharedHttpClient
626    where
627        C: Clone + Send + Sync + 'static,
628        C: hyper_0_14::service::Service<http_02x::Uri>,
629        C::Response: Connection + AsyncRead + AsyncWrite + Send + Unpin + 'static,
630        C::Future: Unpin + Send + 'static,
631        C::Error: Into<BoxError>,
632    {
633        self.build_with_fn(move || tcp_connector.clone())
634    }
635
636    fn build_with_fn<C, F>(self, tcp_connector_fn: F) -> SharedHttpClient
637    where
638        F: Fn() -> C + Send + Sync + 'static,
639        C: Clone + Send + Sync + 'static,
640        C: hyper_0_14::service::Service<http_02x::Uri>,
641        C::Response: Connection + AsyncRead + AsyncWrite + Send + Unpin + 'static,
642        C::Future: Unpin + Send + 'static,
643        C::Error: Into<BoxError>,
644    {
645        SharedHttpClient::new(HyperClient {
646            connector_cache: RwLock::new(HashMap::new()),
647            client_builder: self.client_builder.unwrap_or_default(),
648            tcp_connector_fn,
649        })
650    }
651}
652
653mod timeout_middleware {
654    use aws_smithy_async::future::timeout::{TimedOutError, Timeout};
655    use aws_smithy_async::rt::sleep::Sleep;
656    use aws_smithy_async::rt::sleep::{AsyncSleep, SharedAsyncSleep};
657    use aws_smithy_runtime_api::box_error::BoxError;
658    use pin_project_lite::pin_project;
659    use std::error::Error;
660    use std::fmt::Formatter;
661    use std::future::Future;
662    use std::pin::Pin;
663    use std::task::{Context, Poll};
664    use std::time::Duration;
665
666    #[derive(Debug)]
667    pub(crate) struct HttpTimeoutError {
668        kind: &'static str,
669        duration: Duration,
670    }
671
672    impl std::fmt::Display for HttpTimeoutError {
673        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
674            write!(
675                f,
676                "{} timeout occurred after {:?}",
677                self.kind, self.duration
678            )
679        }
680    }
681
682    impl Error for HttpTimeoutError {
683        // We implement the `source` function as returning a `TimedOutError` because when `downcast_error`
684        // or `find_source` is called with an `HttpTimeoutError` (or another error wrapping an `HttpTimeoutError`)
685        // this method will be checked to determine if it's a timeout-related error.
686        fn source(&self) -> Option<&(dyn Error + 'static)> {
687            Some(&TimedOutError)
688        }
689    }
690
691    /// Timeout wrapper that will timeout on the initial TCP connection
692    ///
693    /// # Stability
694    /// This interface is unstable.
695    #[derive(Clone, Debug)]
696    pub(super) struct ConnectTimeout<I> {
697        inner: I,
698        timeout: Option<(SharedAsyncSleep, Duration)>,
699    }
700
701    impl<I> ConnectTimeout<I> {
702        /// Create a new `ConnectTimeout` around `inner`.
703        ///
704        /// Typically, `I` will implement [`hyper_0_14::client::connect::Connect`].
705        pub(crate) fn new(inner: I, sleep: SharedAsyncSleep, timeout: Duration) -> Self {
706            Self {
707                inner,
708                timeout: Some((sleep, timeout)),
709            }
710        }
711
712        pub(crate) fn no_timeout(inner: I) -> Self {
713            Self {
714                inner,
715                timeout: None,
716            }
717        }
718    }
719
720    #[derive(Clone, Debug)]
721    pub(crate) struct HttpReadTimeout<I> {
722        inner: I,
723        timeout: Option<(SharedAsyncSleep, Duration)>,
724    }
725
726    impl<I> HttpReadTimeout<I> {
727        /// Create a new `HttpReadTimeout` around `inner`.
728        ///
729        /// Typically, `I` will implement [`hyper_0_14::service::Service<http::Request<SdkBody>>`].
730        pub(crate) fn new(inner: I, sleep: SharedAsyncSleep, timeout: Duration) -> Self {
731            Self {
732                inner,
733                timeout: Some((sleep, timeout)),
734            }
735        }
736
737        pub(crate) fn no_timeout(inner: I) -> Self {
738            Self {
739                inner,
740                timeout: None,
741            }
742        }
743    }
744
745    pin_project! {
746        /// Timeout future for Tower services
747        ///
748        /// Timeout future to handle timing out, mapping errors, and the possibility of not timing out
749        /// without incurring an additional allocation for each timeout layer.
750        #[project = MaybeTimeoutFutureProj]
751        pub enum MaybeTimeoutFuture<F> {
752            Timeout {
753                #[pin]
754                timeout: Timeout<F, Sleep>,
755                error_type: &'static str,
756                duration: Duration,
757            },
758            NoTimeout {
759                #[pin]
760                future: F
761            }
762        }
763    }
764
765    impl<F, T, E> Future for MaybeTimeoutFuture<F>
766    where
767        F: Future<Output = Result<T, E>>,
768        E: Into<BoxError>,
769    {
770        type Output = Result<T, BoxError>;
771
772        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
773            let (timeout_future, kind, &mut duration) = match self.project() {
774                MaybeTimeoutFutureProj::NoTimeout { future } => {
775                    return future.poll(cx).map_err(|err| err.into());
776                }
777                MaybeTimeoutFutureProj::Timeout {
778                    timeout,
779                    error_type,
780                    duration,
781                } => (timeout, error_type, duration),
782            };
783            match timeout_future.poll(cx) {
784                Poll::Ready(Ok(response)) => Poll::Ready(response.map_err(|err| err.into())),
785                Poll::Ready(Err(_timeout)) => {
786                    Poll::Ready(Err(HttpTimeoutError { kind, duration }.into()))
787                }
788                Poll::Pending => Poll::Pending,
789            }
790        }
791    }
792
793    impl<I> hyper_0_14::service::Service<http_02x::Uri> for ConnectTimeout<I>
794    where
795        I: hyper_0_14::service::Service<http_02x::Uri>,
796        I::Error: Into<BoxError>,
797    {
798        type Response = I::Response;
799        type Error = BoxError;
800        type Future = MaybeTimeoutFuture<I::Future>;
801
802        fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
803            self.inner.poll_ready(cx).map_err(|err| err.into())
804        }
805
806        fn call(&mut self, req: http_02x::Uri) -> Self::Future {
807            match &self.timeout {
808                Some((sleep, duration)) => {
809                    let sleep = sleep.sleep(*duration);
810                    MaybeTimeoutFuture::Timeout {
811                        timeout: Timeout::new(self.inner.call(req), sleep),
812                        error_type: "HTTP connect",
813                        duration: *duration,
814                    }
815                }
816                None => MaybeTimeoutFuture::NoTimeout {
817                    future: self.inner.call(req),
818                },
819            }
820        }
821    }
822
823    impl<I, B> hyper_0_14::service::Service<http_02x::Request<B>> for HttpReadTimeout<I>
824    where
825        I: hyper_0_14::service::Service<http_02x::Request<B>, Error = hyper_0_14::Error>,
826    {
827        type Response = I::Response;
828        type Error = BoxError;
829        type Future = MaybeTimeoutFuture<I::Future>;
830
831        fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
832            self.inner.poll_ready(cx).map_err(|err| err.into())
833        }
834
835        fn call(&mut self, req: http_02x::Request<B>) -> Self::Future {
836            match &self.timeout {
837                Some((sleep, duration)) => {
838                    let sleep = sleep.sleep(*duration);
839                    MaybeTimeoutFuture::Timeout {
840                        timeout: Timeout::new(self.inner.call(req), sleep),
841                        error_type: "HTTP read",
842                        duration: *duration,
843                    }
844                }
845                None => MaybeTimeoutFuture::NoTimeout {
846                    future: self.inner.call(req),
847                },
848            }
849        }
850    }
851
852    #[cfg(test)]
853    pub(crate) mod test {
854        use crate::hyper_014::HyperConnector;
855        use aws_smithy_async::assert_elapsed;
856        use aws_smithy_async::future::never::Never;
857        use aws_smithy_async::rt::sleep::{SharedAsyncSleep, TokioSleep};
858        use aws_smithy_runtime_api::box_error::BoxError;
859        use aws_smithy_runtime_api::client::http::HttpConnectorSettings;
860        use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
861        use aws_smithy_runtime_api::client::result::ConnectorError;
862        use aws_smithy_types::error::display::DisplayErrorContext;
863        use hyper_0_14::client::connect::{Connected, Connection};
864        use std::future::Future;
865        use std::pin::Pin;
866        use std::task::{Context, Poll};
867        use std::time::Duration;
868        use tokio::io::ReadBuf;
869        use tokio::io::{AsyncRead, AsyncWrite};
870        use tokio::net::TcpStream;
871
872        #[allow(unused)]
873        fn connect_timeout_is_correct<T: Send + Sync + Clone + 'static>() {
874            is_send_sync::<super::ConnectTimeout<T>>();
875        }
876
877        #[allow(unused)]
878        fn is_send_sync<T: Send + Sync>() {}
879
880        /// A service that will never return whatever it is you want
881        ///
882        /// Returned futures will return Pending forever
883        #[non_exhaustive]
884        #[derive(Clone, Default, Debug)]
885        pub(crate) struct NeverConnects;
886        impl hyper_0_14::service::Service<http_02x::Uri> for NeverConnects {
887            type Response = TcpStream;
888            type Error = ConnectorError;
889            type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
890
891            fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
892                Poll::Ready(Ok(()))
893            }
894
895            fn call(&mut self, _uri: http_02x::Uri) -> Self::Future {
896                Box::pin(async move {
897                    Never::new().await;
898                    unreachable!()
899                })
900            }
901        }
902
903        /// A service that will connect but never send any data
904        #[derive(Clone, Debug, Default)]
905        struct NeverReplies;
906        impl hyper_0_14::service::Service<http_02x::Uri> for NeverReplies {
907            type Response = EmptyStream;
908            type Error = BoxError;
909            type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
910
911            fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
912                Poll::Ready(Ok(()))
913            }
914
915            fn call(&mut self, _req: http_02x::Uri) -> Self::Future {
916                std::future::ready(Ok(EmptyStream))
917            }
918        }
919
920        /// A stream that will never return or accept any data
921        #[non_exhaustive]
922        #[derive(Debug, Default)]
923        struct EmptyStream;
924        impl AsyncRead for EmptyStream {
925            fn poll_read(
926                self: Pin<&mut Self>,
927                _cx: &mut Context<'_>,
928                _buf: &mut ReadBuf<'_>,
929            ) -> Poll<std::io::Result<()>> {
930                Poll::Pending
931            }
932        }
933        impl AsyncWrite for EmptyStream {
934            fn poll_write(
935                self: Pin<&mut Self>,
936                _cx: &mut Context<'_>,
937                _buf: &[u8],
938            ) -> Poll<Result<usize, std::io::Error>> {
939                Poll::Pending
940            }
941
942            fn poll_flush(
943                self: Pin<&mut Self>,
944                _cx: &mut Context<'_>,
945            ) -> Poll<Result<(), std::io::Error>> {
946                Poll::Pending
947            }
948
949            fn poll_shutdown(
950                self: Pin<&mut Self>,
951                _cx: &mut Context<'_>,
952            ) -> Poll<Result<(), std::io::Error>> {
953                Poll::Pending
954            }
955        }
956        impl Connection for EmptyStream {
957            fn connected(&self) -> Connected {
958                Connected::new()
959            }
960        }
961
962        #[tokio::test]
963        async fn http_connect_timeout_works() {
964            let tcp_connector = NeverConnects::default();
965            let connector_settings = HttpConnectorSettings::builder()
966                .connect_timeout(Duration::from_secs(1))
967                .build();
968            let hyper = HyperConnector::builder()
969                .connector_settings(connector_settings)
970                .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
971                .build(tcp_connector)
972                .adapter;
973            let now = tokio::time::Instant::now();
974            tokio::time::pause();
975            let resp = hyper
976                .call(HttpRequest::get("https://static-uri.com").unwrap())
977                .await
978                .unwrap_err();
979            assert!(
980                resp.is_timeout(),
981                "expected resp.is_timeout() to be true but it was false, resp == {:?}",
982                resp
983            );
984            let message = DisplayErrorContext(&resp).to_string();
985            let expected =
986                "timeout: error trying to connect: HTTP connect timeout occurred after 1s";
987            assert!(
988                message.contains(expected),
989                "expected '{message}' to contain '{expected}'"
990            );
991            assert_elapsed!(now, Duration::from_secs(1));
992        }
993
994        #[tokio::test]
995        async fn http_read_timeout_works() {
996            let tcp_connector = NeverReplies;
997            let connector_settings = HttpConnectorSettings::builder()
998                .connect_timeout(Duration::from_secs(1))
999                .read_timeout(Duration::from_secs(2))
1000                .build();
1001            let hyper = HyperConnector::builder()
1002                .connector_settings(connector_settings)
1003                .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))
1004                .build(tcp_connector)
1005                .adapter;
1006            let now = tokio::time::Instant::now();
1007            tokio::time::pause();
1008            let err = hyper
1009                .call(HttpRequest::get("https://fake-uri.com").unwrap())
1010                .await
1011                .unwrap_err();
1012            assert!(
1013                err.is_timeout(),
1014                "expected err.is_timeout() to be true but it was false, err == {err:?}",
1015            );
1016            let message = format!("{}", DisplayErrorContext(&err));
1017            let expected = "timeout: HTTP read timeout occurred after 2s";
1018            assert!(
1019                message.contains(expected),
1020                "expected '{message}' to contain '{expected}'"
1021            );
1022            assert_elapsed!(now, Duration::from_secs(2));
1023        }
1024    }
1025}
1026
1027#[cfg(test)]
1028mod test {
1029    use crate::hyper_legacy::timeout_middleware::test::NeverConnects;
1030    use crate::hyper_legacy::{HyperClientBuilder, HyperConnector};
1031    use aws_smithy_async::time::SystemTimeSource;
1032    use aws_smithy_runtime_api::box_error::BoxError;
1033    use aws_smithy_runtime_api::client::http::{HttpClient, HttpConnectorSettings};
1034    use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
1035    use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
1036    use hyper_0_14::client::connect::{Connected, Connection};
1037    use std::io::{Error, ErrorKind};
1038    use std::pin::Pin;
1039    use std::sync::atomic::{AtomicU32, Ordering};
1040    use std::sync::Arc;
1041    use std::task::{Context, Poll};
1042    use std::time::Duration;
1043    use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
1044
1045    #[tokio::test]
1046    async fn connector_selection() {
1047        // Create a client that increments a count every time it creates a new HyperConnector
1048        let creation_count = Arc::new(AtomicU32::new(0));
1049        let http_client = HyperClientBuilder::new().build_with_fn({
1050            let count = creation_count.clone();
1051            move || {
1052                count.fetch_add(1, Ordering::Relaxed);
1053                NeverConnects::default()
1054            }
1055        });
1056
1057        // This configuration should result in 4 separate connectors with different timeout settings
1058        let settings = [
1059            HttpConnectorSettings::builder()
1060                .connect_timeout(Duration::from_secs(3))
1061                .build(),
1062            HttpConnectorSettings::builder()
1063                .read_timeout(Duration::from_secs(3))
1064                .build(),
1065            HttpConnectorSettings::builder()
1066                .connect_timeout(Duration::from_secs(3))
1067                .read_timeout(Duration::from_secs(3))
1068                .build(),
1069            HttpConnectorSettings::builder()
1070                .connect_timeout(Duration::from_secs(5))
1071                .read_timeout(Duration::from_secs(3))
1072                .build(),
1073        ];
1074
1075        // Kick off thousands of parallel tasks that will try to create a connector
1076        let components = RuntimeComponentsBuilder::for_tests()
1077            .with_time_source(Some(SystemTimeSource::new()))
1078            .build()
1079            .unwrap();
1080        let mut handles = Vec::new();
1081        for setting in &settings {
1082            for _ in 0..1000 {
1083                let client = http_client.clone();
1084                handles.push(tokio::spawn({
1085                    let setting = setting.clone();
1086                    let components = components.clone();
1087                    async move {
1088                        let _ = client.http_connector(&setting, &components);
1089                    }
1090                }));
1091            }
1092        }
1093        for handle in handles {
1094            handle.await.unwrap();
1095        }
1096
1097        // Verify only 4 connectors were created amidst the chaos
1098        assert_eq!(4, creation_count.load(Ordering::Relaxed));
1099    }
1100
1101    #[tokio::test]
1102    async fn hyper_io_error() {
1103        let connector = TestConnection {
1104            inner: HangupStream,
1105        };
1106        let adapter = HyperConnector::builder().build(connector).adapter;
1107        let err = adapter
1108            .call(HttpRequest::get("https://socket-hangup.com").unwrap())
1109            .await
1110            .expect_err("socket hangup");
1111        assert!(err.is_io(), "{:?}", err);
1112    }
1113
1114    // ---- machinery to make a Hyper connector that responds with an IO Error
1115    #[derive(Clone)]
1116    struct HangupStream;
1117
1118    impl Connection for HangupStream {
1119        fn connected(&self) -> Connected {
1120            Connected::new()
1121        }
1122    }
1123
1124    impl AsyncRead for HangupStream {
1125        fn poll_read(
1126            self: Pin<&mut Self>,
1127            _cx: &mut Context<'_>,
1128            _buf: &mut ReadBuf<'_>,
1129        ) -> Poll<std::io::Result<()>> {
1130            Poll::Ready(Err(Error::new(
1131                ErrorKind::ConnectionReset,
1132                "connection reset",
1133            )))
1134        }
1135    }
1136
1137    impl AsyncWrite for HangupStream {
1138        fn poll_write(
1139            self: Pin<&mut Self>,
1140            _cx: &mut Context<'_>,
1141            _buf: &[u8],
1142        ) -> Poll<Result<usize, Error>> {
1143            Poll::Pending
1144        }
1145
1146        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1147            Poll::Pending
1148        }
1149
1150        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
1151            Poll::Pending
1152        }
1153    }
1154
1155    #[derive(Clone)]
1156    struct TestConnection<T> {
1157        inner: T,
1158    }
1159
1160    impl<T> hyper_0_14::service::Service<http_02x::Uri> for TestConnection<T>
1161    where
1162        T: Clone + Connection,
1163    {
1164        type Response = T;
1165        type Error = BoxError;
1166        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
1167
1168        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
1169            Poll::Ready(Ok(()))
1170        }
1171
1172        fn call(&mut self, _req: http_02x::Uri) -> Self::Future {
1173            std::future::ready(Ok(self.inner.clone()))
1174        }
1175    }
1176}
aws_smithy_http_client/hyper_legacy.rs

aws_smithy_http_client/
hyper_legacy.rs
