AWS SDK

/*

 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.

 * SPDX-License-Identifier: Apache-2.0

 */

use aws_smithy_async::future::timeout::TimedOutError;

use aws_smithy_async::rt::sleep::{default_async_sleep, AsyncSleep, SharedAsyncSleep};

use aws_smithy_runtime::client::http::connection_poisoning::CaptureSmithyConnection;

use aws_smithy_runtime_api::box_error::BoxError;

use aws_smithy_runtime_api::client::connection::ConnectionMetadata;

use aws_smithy_runtime_api::client::connector_metadata::ConnectorMetadata;

use aws_smithy_runtime_api::client::dns::ResolveDns;

use aws_smithy_runtime_api::client::http::{

    HttpClient, HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpClient,

    SharedHttpConnector,

};

use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};

use aws_smithy_runtime_api::client::result::ConnectorError;

use aws_smithy_runtime_api::client::runtime_components::{

    RuntimeComponents, RuntimeComponentsBuilder,

};

use aws_smithy_runtime_api::shared::IntoShared;

use aws_smithy_types::body::SdkBody;

use aws_smithy_types::config_bag::ConfigBag;

use aws_smithy_types::error::display::DisplayErrorContext;

use aws_smithy_types::retry::ErrorKind;

use client::connect::Connection;

use h2::Reason;

use http::{Extensions, Uri};

use hyper::rt::{Read, Write};

use hyper_util::client::legacy as client;

use hyper_util::client::legacy::connect::dns::Name;

use hyper_util::client::legacy::connect::{

    capture_connection, CaptureConnection, Connect, HttpInfo,

};

use hyper_util::rt::TokioExecutor;

use rustls::crypto::CryptoProvider;

use std::borrow::Cow;

use std::collections::HashMap;

use std::error::Error;

use std::future::Future;

use std::net::SocketAddr;

use std::pin::Pin;

use std::sync::RwLock;

use std::task::{Context, Poll};

use std::time::Duration;

use std::{fmt, vec};

#[derive(Debug, Eq, PartialEq, Clone, Copy)]

#[non_exhaustive]

pub enum CryptoMode {

    #[cfg(feature = "crypto-ring")]

    Ring,

    #[cfg(feature = "crypto-aws-lc")]

    AwsLc,

    #[cfg(feature = "crypto-aws-lc-fips")]

    AwsLcFips,

}

impl CryptoMode {

    fn provider(self) -> CryptoProvider {

        match self {

            #[cfg(feature = "crypto-aws-lc")]

            CryptoMode::AwsLc => rustls::crypto::aws_lc_rs::default_provider(),

            #[cfg(feature = "crypto-ring")]

            CryptoMode::Ring => rustls::crypto::ring::default_provider(),

            #[cfg(feature = "crypto-aws-lc-fips")]

            CryptoMode::AwsLcFips => {

                let provider = rustls::crypto::default_fips_provider();

                assert!(

                    provider.fips(),

                    "FIPS was requested but the provider did not support FIPS"

                );

                provider

            }

        }

    }

}

/// A bridge that allows our `ResolveDns` trait to work with Hyper's `Resolver` interface (based on tower)

#[derive(Clone)]

struct HyperUtilResolver<R> {

    resolver: R,

}

impl<R: ResolveDns + Clone + 'static> tower::Service<Name> for HyperUtilResolver<R> {

    type Response = vec::IntoIter<SocketAddr>;

    type Error = Box<dyn Error + Send + Sync>;

    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {

        Poll::Ready(Ok(()))

    }

    fn call(&mut self, req: Name) -> Self::Future {

        let resolver = self.resolver.clone();

        Box::pin(async move {

            let dns_entries = resolver.resolve_dns(req.as_str()).await?;

            Ok(dns_entries

                .into_iter()

                .map(|ip_addr| SocketAddr::new(ip_addr, 0))

                .collect::<Vec<_>>()

                .into_iter())

        })

    }

}

#[allow(unused_imports)]

mod cached_connectors {

    use client::connect::HttpConnector;

    use hyper_util::client::legacy as client;

    use hyper_util::client::legacy::connect::dns::GaiResolver;

    use crate::hyper_1_0::build_connector::make_tls;

    use crate::hyper_1_0::{CryptoMode, Inner};

    #[cfg(feature = "crypto-ring")]

    pub(crate) static HTTPS_NATIVE_ROOTS_RING: once_cell::sync::Lazy<

        hyper_rustls::HttpsConnector<HttpConnector>,

    > = once_cell::sync::Lazy::new(|| make_tls(GaiResolver::new(), CryptoMode::Ring.provider()));

    #[cfg(feature = "crypto-aws-lc")]

    pub(crate) static HTTPS_NATIVE_ROOTS_AWS_LC: once_cell::sync::Lazy<

        hyper_rustls::HttpsConnector<HttpConnector>,

    > = once_cell::sync::Lazy::new(|| make_tls(GaiResolver::new(), CryptoMode::AwsLc.provider()));

    #[cfg(feature = "crypto-aws-lc-fips")]

    pub(crate) static HTTPS_NATIVE_ROOTS_AWS_LC_FIPS: once_cell::sync::Lazy<

        hyper_rustls::HttpsConnector<HttpConnector>,

    > = once_cell::sync::Lazy::new(|| {

        make_tls(GaiResolver::new(), CryptoMode::AwsLcFips.provider())

    });

    pub(super) fn cached_https(mode: Inner) -> hyper_rustls::HttpsConnector<HttpConnector> {

        match mode {

            #[cfg(feature = "crypto-ring")]

            Inner::Standard(CryptoMode::Ring) => HTTPS_NATIVE_ROOTS_RING.clone(),

            #[cfg(feature = "crypto-aws-lc")]

            Inner::Standard(CryptoMode::AwsLc) => HTTPS_NATIVE_ROOTS_AWS_LC.clone(),

            #[cfg(feature = "crypto-aws-lc-fips")]

            Inner::Standard(CryptoMode::AwsLcFips) => HTTPS_NATIVE_ROOTS_AWS_LC_FIPS.clone(),

            #[allow(unreachable_patterns)]

            Inner::Standard(_) => unreachable!("unexpected mode"),

            Inner::Custom(provider) => make_tls(GaiResolver::new(), provider),

        }

    }

}

mod build_connector {

    use crate::hyper_1_0::{HyperUtilResolver, Inner};

    use aws_smithy_runtime_api::client::dns::ResolveDns;

    use client::connect::HttpConnector;

    use hyper_util::client::legacy as client;

    use rustls::crypto::CryptoProvider;

    use std::sync::Arc;

    fn restrict_ciphers(base: CryptoProvider) -> CryptoProvider {

        let suites = &[

            rustls::CipherSuite::TLS13_AES_256_GCM_SHA384,

            rustls::CipherSuite::TLS13_AES_128_GCM_SHA256,

            // TLS1.2 suites

            rustls::CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,

            rustls::CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,

            rustls::CipherSuite::TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,

            rustls::CipherSuite::TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,

            rustls::CipherSuite::TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256,

        ];

        let supported_suites = suites

            .iter()

            .flat_map(|suite| {

                base.cipher_suites

                    .iter()

                    .find(|s| &s.suite() == suite)

                    .cloned()

            })

            .collect::<Vec<_>>();

        CryptoProvider {

            cipher_suites: supported_suites,

            ..base

        }

    }

    pub(crate) fn make_tls<R>(

        resolver: R,

        crypto_provider: CryptoProvider,

    ) -> hyper_rustls::HttpsConnector<HttpConnector<R>> {

        use hyper_rustls::ConfigBuilderExt;

        let mut base_connector = HttpConnector::new_with_resolver(resolver);

        base_connector.enforce_http(false);

        hyper_rustls::HttpsConnectorBuilder::new()

               .with_tls_config(

                rustls::ClientConfig::builder_with_provider(Arc::new(restrict_ciphers(crypto_provider)))

                    .with_safe_default_protocol_versions()

                    .expect("Error with the TLS configuration. Please file a bug report under https://github.com/smithy-lang/smithy-rs/issues.")

                    .with_native_roots().expect("error with TLS configuration.")

                    .with_no_client_auth()

            )

            .https_or_http()

            .enable_http1()

            .enable_http2()

            .wrap_connector(base_connector)

    }

    pub(super) fn https_with_resolver<R: ResolveDns>(

        crypto_provider: Inner,

        resolver: R,

    ) -> hyper_rustls::HttpsConnector<HttpConnector<HyperUtilResolver<R>>> {

        make_tls(HyperUtilResolver { resolver }, crypto_provider.provider())

    }

}

/// [`HttpConnector`] that uses [`hyper`] to make HTTP requests.

///

/// This connector also implements socket connect and read timeouts.

///

/// This shouldn't be used directly in most cases.

/// See the docs on [`HyperClientBuilder`] for examples of how

/// to customize the Hyper client.

#[derive(Debug)]

pub struct HyperConnector {

    adapter: Box<dyn HttpConnector>,

}

impl HyperConnector {

    /// Builder for a Hyper connector.

    pub fn builder() -> HyperConnectorBuilder {

        Default::default()

    }

}

impl HttpConnector for HyperConnector {

    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {

        self.adapter.call(request)

    }

}

/// Builder for [`HyperConnector`].

#[derive(Default, Debug)]

pub struct HyperConnectorBuilder<Crypto = CryptoUnset> {

    connector_settings: Option<HttpConnectorSettings>,

    sleep_impl: Option<SharedAsyncSleep>,

    client_builder: Option<hyper_util::client::legacy::Builder>,

    #[allow(unused)]

    crypto: Crypto,

}

#[derive(Default)]

#[non_exhaustive]

pub struct CryptoUnset {}

pub struct CryptoProviderSelected {

    crypto_provider: Inner,

}

#[derive(Clone)]

enum Inner {

    Standard(CryptoMode),

    #[allow(dead_code)]

    Custom(CryptoProvider),

}

impl Inner {

    fn provider(&self) -> CryptoProvider {

        match self {

            Inner::Standard(mode) => mode.provider(),

            Inner::Custom(provider) => provider.clone(),

        }

    }

}

#[cfg(any(feature = "crypto-aws-lc", feature = "crypto-ring"))]

impl HyperConnectorBuilder<CryptoProviderSelected> {

    pub fn build_from_resolver<R: ResolveDns + Clone + 'static>(

        self,

        resolver: R,

    ) -> HyperConnector {

        let connector =

            build_connector::https_with_resolver(self.crypto.crypto_provider.clone(), resolver);

        self.build(connector)

    }

}

impl<Any> HyperConnectorBuilder<Any> {

    /// Create a [`HyperConnector`] from this builder and a given connector.

    pub(crate) fn build<C>(self, tcp_connector: C) -> HyperConnector

    where

        C: Send + Sync + 'static,

        C: Clone,

        C: tower::Service<Uri>,

        C::Response: Read + Write + Connection + Send + Sync + Unpin,

        C: Connect,

        C::Future: Unpin + Send + 'static,

        C::Error: Into<BoxError>,

    {

        let client_builder =

            self.client_builder

                .unwrap_or(hyper_util::client::legacy::Builder::new(

                    TokioExecutor::new(),

                ));

        let sleep_impl = self.sleep_impl.or_else(default_async_sleep);

        let (connect_timeout, read_timeout) = self

            .connector_settings

            .map(|c| (c.connect_timeout(), c.read_timeout()))

            .unwrap_or((None, None));

        let connector = match connect_timeout {

            Some(duration) => timeout_middleware::ConnectTimeout::new(

                tcp_connector,

                sleep_impl

                    .clone()

                    .expect("a sleep impl must be provided in order to have a connect timeout"),

                duration,

            ),

            None => timeout_middleware::ConnectTimeout::no_timeout(tcp_connector),

        };

        let base = client_builder.build(connector);

        let read_timeout = match read_timeout {

            Some(duration) => timeout_middleware::HttpReadTimeout::new(

                base,

                sleep_impl.expect("a sleep impl must be provided in order to have a read timeout"),

                duration,

            ),

            None => timeout_middleware::HttpReadTimeout::no_timeout(base),

        };

        HyperConnector {

            adapter: Box::new(Adapter {

                client: read_timeout,

            }),

        }

    }

    /// Set the async sleep implementation used for timeouts

    ///

    /// Calling this is only necessary for testing or to use something other than

    /// [`default_async_sleep`].

    pub fn sleep_impl(mut self, sleep_impl: impl AsyncSleep + 'static) -> Self {

        self.sleep_impl = Some(sleep_impl.into_shared());

        self

    }

    /// Set the async sleep implementation used for timeouts

    ///

    /// Calling this is only necessary for testing or to use something other than

    /// [`default_async_sleep`].

    pub fn set_sleep_impl(&mut self, sleep_impl: Option<SharedAsyncSleep>) -> &mut Self {

        self.sleep_impl = sleep_impl;

        self

    }

    /// Configure the HTTP settings for the `HyperAdapter`

    pub fn connector_settings(mut self, connector_settings: HttpConnectorSettings) -> Self {

        self.connector_settings = Some(connector_settings);

        self

    }

    /// Configure the HTTP settings for the `HyperAdapter`

    pub fn set_connector_settings(

        &mut self,

        connector_settings: Option<HttpConnectorSettings>,

    ) -> &mut Self {

        self.connector_settings = connector_settings;

        self

    }

    /// Override the Hyper client [`Builder`](hyper_util::client::legacy::Builder) used to construct this client.

    ///

    /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.

    pub(crate) fn hyper_builder(

        mut self,

        hyper_builder: hyper_util::client::legacy::Builder,

    ) -> Self {

        self.set_hyper_builder(Some(hyper_builder));

        self

    }

    /// Override the Hyper client [`Builder`](hyper_util::client::legacy::Builder) used to construct this client.

    ///

    /// This enables changing settings like forcing HTTP2 and modifying other default client behavior.

    pub(crate) fn set_hyper_builder(

        &mut self,

        hyper_builder: Option<hyper_util::client::legacy::Builder>,

    ) -> &mut Self {

        self.client_builder = hyper_builder;

        self

    }

}

/// Adapter to use a Hyper 1.0-based Client as an `HttpConnector`

///

/// This adapter also enables TCP `CONNECT` and HTTP `READ` timeouts via [`HyperConnector::builder`].

struct Adapter<C> {

    client: timeout_middleware::HttpReadTimeout<

        hyper_util::client::legacy::Client<timeout_middleware::ConnectTimeout<C>, SdkBody>,

    >,

}

impl<C> fmt::Debug for Adapter<C> {

    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

        f.debug_struct("Adapter")

            .field("client", &"** hyper client **")

            .finish()

    }

}

/// Extract a smithy connection from a hyper CaptureConnection

fn extract_smithy_connection(capture_conn: &CaptureConnection) -> Option<ConnectionMetadata> {

    let capture_conn = capture_conn.clone();

    if let Some(conn) = capture_conn.clone().connection_metadata().as_ref() {

        let mut extensions = Extensions::new();

        conn.get_extras(&mut extensions);

        let http_info = extensions.get::<HttpInfo>();

        let mut builder = ConnectionMetadata::builder()

            .proxied(conn.is_proxied())

            .poison_fn(move || match capture_conn.connection_metadata().as_ref() {

                Some(conn) => conn.poison(),

                None => tracing::trace!("no connection existed to poison"),

            });

        builder

            .set_local_addr(http_info.map(|info| info.local_addr()))

            .set_remote_addr(http_info.map(|info| info.remote_addr()));

        let smithy_connection = builder.build();

        Some(smithy_connection)

    } else {

        None

    }

}

impl<C> HttpConnector for Adapter<C>

where

    C: Clone + Send + Sync + 'static,

    C: tower::Service<Uri>,

    C::Response: Connection + Read + Write + Unpin + 'static,

    timeout_middleware::ConnectTimeout<C>: Connect,

    C::Future: Unpin + Send + 'static,

    C::Error: Into<BoxError>,

{

    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {

        let mut request = match request.try_into_http1x() {

            Ok(request) => request,

            Err(err) => {

                return HttpConnectorFuture::ready(Err(ConnectorError::user(err.into())));

            }

        };

        let capture_connection = capture_connection(&mut request);

        if let Some(capture_smithy_connection) =

            request.extensions().get::<CaptureSmithyConnection>()

        {

            capture_smithy_connection

                .set_connection_retriever(move || extract_smithy_connection(&capture_connection));

        }

        let mut client = self.client.clone();

        use tower::Service;

        let fut = client.call(request);

        HttpConnectorFuture::new(async move {

            let response = fut

                .await

                .map_err(downcast_error)?

                .map(SdkBody::from_body_1_x);

            match HttpResponse::try_from(response) {

                Ok(response) => Ok(response),

                Err(err) => Err(ConnectorError::other(err.into(), None)),

            }

        })

    }

}

/// Downcast errors coming out of hyper into an appropriate `ConnectorError`

fn downcast_error(err: BoxError) -> ConnectorError {

    // is a `TimedOutError` (from aws_smithy_async::timeout) in the chain? if it is, this is a timeout

    if find_source::<TimedOutError>(err.as_ref()).is_some() {

        return ConnectorError::timeout(err);

    }

    // is the top of chain error actually already a `ConnectorError`? return that directly

    let err = match err.downcast::<ConnectorError>() {

        Ok(connector_error) => return *connector_error,

        Err(box_error) => box_error,

    };

    // generally, the top of chain will probably be a hyper error. Go through a set of hyper specific

    // error classifications

    let err = match find_source::<hyper::Error>(err.as_ref()) {

        Some(hyper_error) => return to_connector_error(hyper_error)(err),

        None => err,

    };

    // otherwise, we have no idea!

    ConnectorError::other(err, None)

}

/// Convert a [`hyper::Error`] into a [`ConnectorError`]

fn to_connector_error(err: &hyper::Error) -> fn(BoxError) -> ConnectorError {

    if err.is_timeout() || find_source::<timeout_middleware::HttpTimeoutError>(err).is_some() {

        return ConnectorError::timeout;

    }

    if err.is_user() {

        return ConnectorError::user;

    }

    if err.is_closed() || err.is_canceled() || find_source::<std::io::Error>(err).is_some() {

        return ConnectorError::io;

    }

    // We sometimes receive this from S3: hyper::Error(IncompleteMessage)

    if err.is_incomplete_message() {

        return |err: BoxError| ConnectorError::other(err, Some(ErrorKind::TransientError));

    }

    if let Some(h2_err) = find_source::<h2::Error>(err) {

        if h2_err.is_go_away()

            || (h2_err.is_reset() && h2_err.reason() == Some(Reason::REFUSED_STREAM))

        {

            return ConnectorError::io;

        }

    }

    tracing::warn!(err = %DisplayErrorContext(&err), "unrecognized error from Hyper. If this error should be retried, please file an issue.");

    |err: BoxError| ConnectorError::other(err, None)

}

fn find_source<'a, E: Error + 'static>(err: &'a (dyn Error + 'static)) -> Option<&'a E> {

    let mut next = Some(err);

    while let Some(err) = next {

        if let Some(matching_err) = err.downcast_ref::<E>() {

            return Some(matching_err);

        }

        next = err.source();

    }

    None

}

// TODO(https://github.com/awslabs/aws-sdk-rust/issues/1090): CacheKey must also include ptr equality to any

// runtime components that are used—sleep_impl as a base (unless we prohibit overriding sleep impl)

// If we decide to put a DnsResolver in RuntimeComponents, then we'll need to handle that as well.

#[derive(Clone, Debug, Eq, PartialEq, Hash)]

struct CacheKey {

    connect_timeout: Option<Duration>,

    read_timeout: Option<Duration>,

}

impl From<&HttpConnectorSettings> for CacheKey {

    fn from(value: &HttpConnectorSettings) -> Self {

        Self {

            connect_timeout: value.connect_timeout(),

            read_timeout: value.read_timeout(),

        }

    }

}

struct HyperClient<F> {

    connector_cache: RwLock<HashMap<CacheKey, SharedHttpConnector>>,

    client_builder: hyper_util::client::legacy::Builder,

    tcp_connector_fn: F,

}

impl<F> fmt::Debug for HyperClient<F> {

    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

        f.debug_struct("HyperClient")

            .field("connector_cache", &self.connector_cache)

            .field("client_builder", &self.client_builder)

            .finish()

    }

}

impl<C, F> HttpClient for HyperClient<F>

where

    F: Fn() -> C + Send + Sync,

    C: Clone + Send + Sync + 'static,

    C: tower::Service<Uri>,

    C::Response: Connection + Read + Write + Send + Sync + Unpin + 'static,

    C::Future: Unpin + Send + 'static,

    C::Error: Into<BoxError>,

{

    fn http_connector(

        &self,

        settings: &HttpConnectorSettings,

        components: &RuntimeComponents,

    ) -> SharedHttpConnector {

        let key = CacheKey::from(settings);

        let mut connector = self.connector_cache.read().unwrap().get(&key).cloned();

        if connector.is_none() {

            let mut cache = self.connector_cache.write().unwrap();

            // Short-circuit if another thread already wrote a connector to the cache for this key

            if !cache.contains_key(&key) {

                let mut builder = HyperConnector::builder()

                    .hyper_builder(self.client_builder.clone())

                    .connector_settings(settings.clone());

                builder.set_sleep_impl(components.sleep_impl());

                let start = components.time_source().map(|ts| ts.now());

                let tcp_connector = (self.tcp_connector_fn)();

                let end = components.time_source().map(|ts| ts.now());

                if let (Some(start), Some(end)) = (start, end) {

                    if let Ok(elapsed) = end.duration_since(start) {

                        tracing::debug!("new TCP connector created in {:?}", elapsed);

                    }

                }

                let connector = SharedHttpConnector::new(builder.build(tcp_connector));

                cache.insert(key.clone(), connector);

            }

            connector = cache.get(&key).cloned();

        }

        connector.expect("cache populated above")

    }

    fn validate_base_client_config(

        &self,

        _: &RuntimeComponentsBuilder,

        _: &ConfigBag,

    ) -> Result<(), BoxError> {

        // Initialize the TCP connector at this point so that native certs load

        // at client initialization time instead of upon first request. We do it

        // here rather than at construction so that it won't run if this is not

        // the selected HTTP client for the base config (for example, if this was

        // the default HTTP client, and it was overridden by a later plugin).

        let _ = (self.tcp_connector_fn)();

        Ok(())

    }

    fn connector_metadata(&self) -> Option<ConnectorMetadata> {

        Some(ConnectorMetadata::new("hyper", Some(Cow::Borrowed("1.x"))))

    }

}

/// Builder for a hyper-backed [`HttpClient`] implementation.

///

/// This builder can be used to customize the underlying TCP connector used, as well as

/// hyper client configuration.

///

/// # Examples

///

/// Construct a Hyper client with the RusTLS TLS implementation.

/// This can be useful when you want to share a Hyper connector between multiple

/// generated Smithy clients.

#[derive(Clone, Default, Debug)]

pub struct HyperClientBuilder<Crypto = CryptoUnset> {

    client_builder: Option<hyper_util::client::legacy::Builder>,

    crypto_provider: Crypto,

}

impl HyperClientBuilder<CryptoProviderSelected> {

    /// Create a hyper client using RusTLS for TLS

    ///

    /// The trusted certificates will be loaded later when this becomes the selected

    /// HTTP client for a Smithy client.

    pub fn build_https(self) -> SharedHttpClient {

        let crypto = self.crypto_provider.crypto_provider;

        build_with_fn(self.client_builder, move || {

            cached_connectors::cached_https(crypto.clone())

        })

    }

    /// Create a hyper client using a custom DNS resolver

    pub fn build_with_resolver(

        self,

        resolver: impl ResolveDns + Clone + 'static,

    ) -> SharedHttpClient {

        build_with_fn(self.client_builder, move || {

            build_connector::https_with_resolver(

                self.crypto_provider.crypto_provider.clone(),

                resolver.clone(),

            )

        })

    }

}

impl HyperClientBuilder<CryptoUnset> {

    /// Creates a new builder.

    pub fn new() -> Self {

        Self::default()

    }

    pub fn crypto_mode(self, provider: CryptoMode) -> HyperClientBuilder<CryptoProviderSelected> {

        HyperClientBuilder {

            client_builder: self.client_builder,

            crypto_provider: CryptoProviderSelected {

                crypto_provider: Inner::Standard(provider),

            },

        }

    }

    /// This interface will be broken in the future

    ///

    /// This exposes `CryptoProvider` from `rustls` directly and this API has no stability guarantee.

    #[cfg(crypto_unstable)]

    pub fn crypto_provider_unstable(

        self,

        provider: CryptoProvider,

    ) -> HyperClientBuilder<CryptoProviderSelected> {

        HyperClientBuilder {

            client_builder: self.client_builder,

            crypto_provider: CryptoProviderSelected {

                crypto_provider: Inner::Custom(provider),

            },

        }

    }

}

fn build_with_fn<C, F>(

    client_builder: Option<hyper_util::client::legacy::Builder>,

    tcp_connector_fn: F,

) -> SharedHttpClient

where

    F: Fn() -> C + Send + Sync + 'static,

    C: Clone + Send + Sync + 'static,

    C: tower::Service<Uri>,

    C::Response: Connection + Read + Write + Send + Sync + Unpin + 'static,

    C::Future: Unpin + Send + 'static,

    C::Error: Into<BoxError>,

    C: Connect,

{

    SharedHttpClient::new(HyperClient {

        connector_cache: RwLock::new(HashMap::new()),

        client_builder: client_builder

            .unwrap_or_else(|| hyper_util::client::legacy::Builder::new(TokioExecutor::new())),

        tcp_connector_fn,

    })

}

mod timeout_middleware {

    use std::error::Error;

    use std::fmt::Formatter;

    use std::future::Future;

    use std::pin::Pin;

    use std::task::{Context, Poll};

    use std::time::Duration;

    use http::Uri;

    use pin_project_lite::pin_project;

    use aws_smithy_async::future::timeout::{TimedOutError, Timeout};

    use aws_smithy_async::rt::sleep::Sleep;

    use aws_smithy_async::rt::sleep::{AsyncSleep, SharedAsyncSleep};

    use aws_smithy_runtime_api::box_error::BoxError;

    #[derive(Debug)]

    pub(crate) struct HttpTimeoutError {

        kind: &'static str,

        duration: Duration,

    }

    impl std::fmt::Display for HttpTimeoutError {

        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {

            write!(

                f,

                "{} timeout occurred after {:?}",

                self.kind, self.duration

            )

        }

    }

    impl Error for HttpTimeoutError {

        // We implement the `source` function as returning a `TimedOutError` because when `downcast_error`

        // or `find_source` is called with an `HttpTimeoutError` (or another error wrapping an `HttpTimeoutError`)

        // this method will be checked to determine if it's a timeout-related error.

        fn source(&self) -> Option<&(dyn Error + 'static)> {

            Some(&TimedOutError)

        }

    }

    /// Timeout wrapper that will timeout on the initial TCP connection

    ///

    /// # Stability

    /// This interface is unstable.

    #[derive(Clone, Debug)]

    pub(super) struct ConnectTimeout<I> {

        inner: I,

        timeout: Option<(SharedAsyncSleep, Duration)>,

    }

    impl<I> ConnectTimeout<I> {

        /// Create a new `ConnectTimeout` around `inner`.

        ///

        /// Typically, `I` will implement [`hyper_util::client::legacy::connect::Connect`].

        pub(crate) fn new(inner: I, sleep: SharedAsyncSleep, timeout: Duration) -> Self {

            Self {

                inner,

                timeout: Some((sleep, timeout)),

            }

        }

        pub(crate) fn no_timeout(inner: I) -> Self {

            Self {

                inner,

                timeout: None,

            }

        }

    }

    #[derive(Clone, Debug)]

    pub(crate) struct HttpReadTimeout<I> {

        inner: I,

        timeout: Option<(SharedAsyncSleep, Duration)>,

    }

    impl<I> HttpReadTimeout<I> {

        /// Create a new `HttpReadTimeout` around `inner`.

        ///

        /// Typically, `I` will implement [`tower::Service<http::Request<SdkBody>>`].

        pub(crate) fn new(inner: I, sleep: SharedAsyncSleep, timeout: Duration) -> Self {

            Self {

                inner,

                timeout: Some((sleep, timeout)),

            }

        }

        pub(crate) fn no_timeout(inner: I) -> Self {

            Self {

                inner,

                timeout: None,

            }

        }

    }

    pin_project! {

        /// Timeout future for Tower services

        ///

        /// Timeout future to handle timing out, mapping errors, and the possibility of not timing out

        /// without incurring an additional allocation for each timeout layer.

        #[project = MaybeTimeoutFutureProj]

        pub enum MaybeTimeoutFuture<F> {

            Timeout {

                #[pin]

                timeout: Timeout<F, Sleep>,

                error_type: &'static str,

                duration: Duration,

            },

            NoTimeout {

                #[pin]

                future: F

            }

        }

    }

    impl<F, T, E> Future for MaybeTimeoutFuture<F>

    where

        F: Future<Output = Result<T, E>>,

        E: Into<BoxError>,

    {

        type Output = Result<T, BoxError>;

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {

            let (timeout_future, kind, &mut duration) = match self.project() {

                MaybeTimeoutFutureProj::NoTimeout { future } => {

                    return future.poll(cx).map_err(|err| err.into());

                }

                MaybeTimeoutFutureProj::Timeout {

                    timeout,

                    error_type,

                    duration,

                } => (timeout, error_type, duration),

            };

            match timeout_future.poll(cx) {

                Poll::Ready(Ok(response)) => Poll::Ready(response.map_err(|err| err.into())),

                Poll::Ready(Err(_timeout)) => {

                    Poll::Ready(Err(HttpTimeoutError { kind, duration }.into()))

                }

                Poll::Pending => Poll::Pending,

            }

        }

    }

    impl<I> tower::Service<Uri> for ConnectTimeout<I>

    where

        I: tower::Service<Uri>,

        I::Error: Into<BoxError>,

    {

        type Response = I::Response;

        type Error = BoxError;

        type Future = MaybeTimeoutFuture<I::Future>;

        fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {

            self.inner.poll_ready(cx).map_err(|err| err.into())

        }

        fn call(&mut self, req: Uri) -> Self::Future {

            match &self.timeout {

                Some((sleep, duration)) => {

                    let sleep = sleep.sleep(*duration);

                    MaybeTimeoutFuture::Timeout {

                        timeout: Timeout::new(self.inner.call(req), sleep),

                        error_type: "HTTP connect",

                        duration: *duration,

                    }

                }

                None => MaybeTimeoutFuture::NoTimeout {

                    future: self.inner.call(req),

                },

            }

        }

    }

    impl<I, B> tower::Service<http::Request<B>> for HttpReadTimeout<I>

    where

        I: tower::Service<http::Request<B>>,

        I::Error: Send + Sync + Error + 'static,

    {

        type Response = I::Response;

        type Error = BoxError;

        type Future = MaybeTimeoutFuture<I::Future>;

        fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {

            self.inner.poll_ready(cx).map_err(|err| err.into())

        }

        fn call(&mut self, req: http::Request<B>) -> Self::Future {

            match &self.timeout {

                Some((sleep, duration)) => {

                    let sleep = sleep.sleep(*duration);

                    MaybeTimeoutFuture::Timeout {

                        timeout: Timeout::new(self.inner.call(req), sleep),

                        error_type: "HTTP read",

                        duration: *duration,

                    }

                }

                None => MaybeTimeoutFuture::NoTimeout {

                    future: self.inner.call(req),

                },

            }

        }

    }

    #[cfg(test)]

    pub(crate) mod test {

        use std::time::Duration;

        use hyper::rt::ReadBufCursor;

        use hyper_util::client::legacy::connect::Connected;

        use hyper_util::rt::TokioIo;

        use tokio::net::TcpStream;

        use aws_smithy_async::assert_elapsed;

        use aws_smithy_async::future::never::Never;

        use aws_smithy_async::rt::sleep::{SharedAsyncSleep, TokioSleep};

        use aws_smithy_types::error::display::DisplayErrorContext;

        use super::super::*;

        #[allow(unused)]

        fn connect_timeout_is_correct<T: Send + Sync + Clone + 'static>() {

            is_send_sync::<super::ConnectTimeout<T>>();

        }

        #[allow(unused)]

        fn is_send_sync<T: Send + Sync>() {}

        /// A service that will never return whatever it is you want

        ///

        /// Returned futures will return Pending forever

        #[non_exhaustive]

        #[derive(Clone, Default, Debug)]

        pub(crate) struct NeverConnects;

        impl tower::Service<Uri> for NeverConnects {

            type Response = TokioIo<TcpStream>;

            type Error = ConnectorError;

            type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

            fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {

                Poll::Ready(Ok(()))

            }

            fn call(&mut self, _uri: Uri) -> Self::Future {

                Box::pin(async move {

                    Never::new().await;

                    unreachable!()

                })

            }

        }

        /// A service that will connect but never send any data

        #[derive(Clone, Debug, Default)]

        struct NeverReplies;

        impl tower::Service<Uri> for NeverReplies {

            type Response = EmptyStream;

            type Error = BoxError;

            type Future = std::future::Ready<Result<Self::Response, Self::Error>>;

            fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {

                Poll::Ready(Ok(()))

            }

            fn call(&mut self, _req: Uri) -> Self::Future {

                std::future::ready(Ok(EmptyStream))

            }

        }

        /// A stream that will never return or accept any data

        #[non_exhaustive]

        #[derive(Debug, Default)]

        struct EmptyStream;

        impl Read for EmptyStream {

            fn poll_read(

                self: Pin<&mut Self>,

                _cx: &mut Context<'_>,

                _buf: ReadBufCursor<'_>,

            ) -> Poll<Result<(), std::io::Error>> {

                Poll::Pending

            }

        }

        impl Write for EmptyStream {

            fn poll_write(

                self: Pin<&mut Self>,

                _cx: &mut Context<'_>,

                _buf: &[u8],

            ) -> Poll<Result<usize, std::io::Error>> {

                Poll::Pending

            }

            fn poll_flush(

                self: Pin<&mut Self>,

                _cx: &mut Context<'_>,

            ) -> Poll<Result<(), std::io::Error>> {

                Poll::Pending

            }

            fn poll_shutdown(

                self: Pin<&mut Self>,

                _cx: &mut Context<'_>,

            ) -> Poll<Result<(), std::io::Error>> {

                Poll::Pending

            }

        }

        impl Connection for EmptyStream {

            fn connected(&self) -> Connected {

                Connected::new()

            }

        }

        #[tokio::test]

        async fn http_connect_timeout_works() {

            let tcp_connector = NeverConnects::default();

            let connector_settings = HttpConnectorSettings::builder()

                .connect_timeout(Duration::from_secs(1))

                .build();

            let hyper = HyperConnector::builder()

                .connector_settings(connector_settings)

                .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))

                .build(tcp_connector)

                .adapter;

            let now = tokio::time::Instant::now();

            tokio::time::pause();

            let resp = hyper

                .call(HttpRequest::get("https://static-uri.com").unwrap())

                .await

                .unwrap_err();

            assert!(

                resp.is_timeout(),

                "expected resp.is_timeout() to be true but it was false, resp == {:?}",

                resp

            );

            let message = DisplayErrorContext(&resp).to_string();

            let expected =

                "timeout: client error (Connect): HTTP connect timeout occurred after 1s";

            assert!(

                message.contains(expected),

                "expected '{message}' to contain '{expected}'"

            );

            assert_elapsed!(now, Duration::from_secs(1));

        }

        #[tokio::test]

        async fn http_read_timeout_works() {

            let tcp_connector = NeverReplies;

            let connector_settings = HttpConnectorSettings::builder()

                .connect_timeout(Duration::from_secs(1))

                .read_timeout(Duration::from_secs(2))

                .build();

            let hyper = HyperConnector::builder()

                .connector_settings(connector_settings)

                .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))

                .build(tcp_connector)

                .adapter;

            let now = tokio::time::Instant::now();

            tokio::time::pause();

            let err = hyper

                .call(HttpRequest::get("https://fake-uri.com").unwrap())

                .await

                .unwrap_err();

            assert!(

                err.is_timeout(),

                "expected err.is_timeout() to be true but it was false, err == {err:?}",

            );

            let message = format!("{}", DisplayErrorContext(&err));

            let expected = "timeout: HTTP read timeout occurred after 2s";

            assert!(

                message.contains(expected),

                "expected '{message}' to contain '{expected}'"

            );

            assert_elapsed!(now, Duration::from_secs(2));

        }

    }

}

#[cfg(test)]

mod test {

    use std::io::{Error, ErrorKind};

    use std::pin::Pin;

    use std::sync::atomic::{AtomicU32, Ordering};

    use std::sync::Arc;

    use std::task::{Context, Poll};

    use http::Uri;

    use hyper::rt::ReadBufCursor;

    use hyper_util::client::legacy::connect::Connected;

    use aws_smithy_async::time::SystemTimeSource;

    use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;

    use crate::hyper_1_0::timeout_middleware::test::NeverConnects;

    use super::*;

    #[tokio::test]

    async fn connector_selection() {

        // Create a client that increments a count every time it creates a new HyperConnector

        let creation_count = Arc::new(AtomicU32::new(0));

        let http_client = build_with_fn(None, {

            let count = creation_count.clone();

            move || {

                count.fetch_add(1, Ordering::Relaxed);

                NeverConnects

            }

        });

        // This configuration should result in 4 separate connectors with different timeout settings

        let settings = [

            HttpConnectorSettings::builder()

                .connect_timeout(Duration::from_secs(3))

                .build(),

            HttpConnectorSettings::builder()

                .read_timeout(Duration::from_secs(3))

                .build(),

            HttpConnectorSettings::builder()

                .connect_timeout(Duration::from_secs(3))

                .read_timeout(Duration::from_secs(3))

                .build(),

            HttpConnectorSettings::builder()

                .connect_timeout(Duration::from_secs(5))

                .read_timeout(Duration::from_secs(3))

                .build(),

        ];

        // Kick off thousands of parallel tasks that will try to create a connector

        let components = RuntimeComponentsBuilder::for_tests()

            .with_time_source(Some(SystemTimeSource::new()))

            .build()

            .unwrap();

        let mut handles = Vec::new();

        for setting in &settings {

            for _ in 0..1000 {

                let client = http_client.clone();

                handles.push(tokio::spawn({

                    let setting = setting.clone();

                    let components = components.clone();

                    async move {

                        let _ = client.http_connector(&setting, &components);

                    }

                }));

            }

        }

        for handle in handles {

            handle.await.unwrap();

        }

        // Verify only 4 connectors were created amidst the chaos

        assert_eq!(4, creation_count.load(Ordering::Relaxed));

    }

    #[tokio::test]

    async fn hyper_io_error() {

        let connector = TestConnection {

            inner: HangupStream,

        };

        let adapter = HyperConnector::builder().build(connector).adapter;

        let err = adapter

            .call(HttpRequest::get("https://socket-hangup.com").unwrap())

            .await

            .expect_err("socket hangup");

        assert!(err.is_io(), "unexpected error type: {:?}", err);

    }

    // ---- machinery to make a Hyper connector that responds with an IO Error

    #[derive(Clone)]

    struct HangupStream;

    impl Connection for HangupStream {

        fn connected(&self) -> Connected {

            Connected::new()

        }

    }

    impl Read for HangupStream {

        fn poll_read(

            self: Pin<&mut Self>,

            _cx: &mut Context<'_>,

            _buf: ReadBufCursor<'_>,

        ) -> Poll<std::io::Result<()>> {

            Poll::Ready(Err(Error::new(

                ErrorKind::ConnectionReset,

                "connection reset",

            )))

        }

    }

    impl Write for HangupStream {

        fn poll_write(

            self: Pin<&mut Self>,

            _cx: &mut Context<'_>,

            _buf: &[u8],

        ) -> Poll<Result<usize, Error>> {

            Poll::Pending

        }

        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {

            Poll::Pending

        }

        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {

            Poll::Pending

        }

    }

    #[derive(Clone)]

    struct TestConnection<T> {

        inner: T,

    }

    impl<T> tower::Service<Uri> for TestConnection<T>

    where

        T: Clone + Connection,

    {

        type Response = T;

        type Error = BoxError;

        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;

        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {

            Poll::Ready(Ok(()))

        }

        fn call(&mut self, _req: Uri) -> Self::Future {

            std::future::ready(Ok(self.inner.clone()))

        }

    }

}