Allow list-objects-v2 to run against an S3 Express bucket (#3388)

## Motivation and Context
Adds an implementation spike to allow `list-objects-v2` (possibly
others, haven't tested yet) to run against an S3 Express bucket.

## Description
This PR implements two ingredients, `S3ExpressIdentityProvider` and
`S3ExpressSigner`. `S3ExpressIdentityProvider` uses an internal S3
client to obtain an S3 Express session token that is passed to
`S3ExpressSigner`. `S3ExpressSigner` then signs a request with that
token, using effectively sigv4 but with session token omitted and an
extra header added instead, `x-amz-s3session-token`.

In addition, this PR supports presigning for S3 Express. Similarly to
signing headers, presigning for S3 Express excludes a query param
`X-Amz-Security-Token` and instead uses `X-Amz-S3session-Token` for the
signing query params. The following screeshot shows that a presigned URL
from `get_object` works for an S3 Express bucket:
<p align="center">
<img width="600" alt="chain-provider-ext-timeout-2"
src="https://github.com/smithy-lang/smithy-rs/assets/15333866/40d7bb53-d936-4d0d-8f95-0323725e2111">
</p>

Some implementation details:
- Since `S3ExpressIdentityProvider` passes an S3 Express bucket name for
S3's `create_session` API to obtain an S3 Express session token, it
needs to obtain the bucket name from somewhere.
`S3ExpressIdentityProvider::ProvideCredentials` I put previously did not
have enough arguments for us to figure this out, so I switched to
`S3ExpressIdentityProvider::ResolveIdentity` that takes enough
arguments.
- `SigV4Signer::sign_http_request` did not allow calling code to pass a
configured `SigningSettings`; The signer needs to exclude a header
`x-amz-security-token` and include `x-amz-s3session-token`. To make this
happen, I made `sigv4::extract_operation_config` and `sigv4::settings`
public APIs (previously private).
- One area I haven't quite figured out yet is how to configure the inner
S3 client to call `create_session`. The changes in this PR inherits
runtime components & config bag from the "outer" S3 client, but
customers may want to configure the inner S3 client in a more flexible
manner (e.g. operation timeout).

## Testing
To lock the behavior at this time, I added a connection recording test
for `list-objects-v2`.

----

_By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice._

---------

Co-authored-by: John DiSanti <[email protected]>
Co-authored-by: AWS SDK Rust Bot <[email protected]>
Co-authored-by: AWS SDK Rust Bot <[email protected]>
Co-authored-by: Zelda Hessler <[email protected]>

aws/rust-runtime/aws-inlineable/src/s3_express.rs

@@ -5,6 +5,8 @@
 
 /// Supporting code for S3 Express auth
 pub(crate) mod auth {
+    use aws_runtime::auth::sigv4::SigV4Signer;
+    use aws_sigv4::http_request::{SignatureLocation, SigningSettings};
     use aws_smithy_runtime_api::box_error::BoxError;
     use aws_smithy_runtime_api::client::auth::{
         AuthScheme, AuthSchemeEndpointConfig, AuthSchemeId, Sign,
@@ -56,15 +58,37 @@ pub(crate) mod auth {
     impl Sign for S3ExpressSigner {
         fn sign_http_request(
             &self,
-            _request: &mut HttpRequest,
-            _identity: &Identity,
-            _auth_scheme_endpoint_config: AuthSchemeEndpointConfig<'_>,
-            _runtime_components: &RuntimeComponents,
-            _config_bag: &ConfigBag,
+            request: &mut HttpRequest,
+            identity: &Identity,
+            auth_scheme_endpoint_config: AuthSchemeEndpointConfig<'_>,
+            runtime_components: &RuntimeComponents,
+            config_bag: &ConfigBag,
         ) -> Result<(), BoxError> {
-            todo!()
+            let operation_config =
+                SigV4Signer::extract_operation_config(auth_scheme_endpoint_config, config_bag)?;
+            let mut settings = SigV4Signer::signing_settings(&operation_config);
+            override_session_token_name(&mut settings)?;
+
+            SigV4Signer.sign_http_request(
+                request,
+                identity,
+                settings,
+                &operation_config,
+                runtime_components,
+                config_bag,
+            )
         }
     }
+
+    fn override_session_token_name(settings: &mut SigningSettings) -> Result<(), BoxError> {
+        let session_token_name_override = match settings.signature_location {
+            SignatureLocation::Headers => Some("x-amz-s3session-token"),
+            SignatureLocation::QueryParams => Some("X-Amz-S3session-Token"),
+            _ => { return Err(BoxError::from("`SignatureLocation` adds a new variant, which needs to be handled in a separate match arm")) },
+        };
+        settings.session_token_name_override = session_token_name_override;
+        Ok(())
+    }
 }
 
 /// Supporting code for S3 Express identity cache
@@ -80,11 +104,21 @@ pub(crate) mod identity_cache {
 
 /// Supporting code for S3 Express identity provider
 pub(crate) mod identity_provider {
+    use std::time::SystemTime;
+
     use crate::s3_express::identity_cache::S3ExpressIdentityCache;
+    use crate::types::SessionCredentials;
+    use aws_credential_types::provider::error::CredentialsError;
+    use aws_credential_types::Credentials;
+    use aws_smithy_runtime_api::box_error::BoxError;
+    use aws_smithy_runtime_api::client::endpoint::EndpointResolverParams;
     use aws_smithy_runtime_api::client::identity::{
-        IdentityCacheLocation, IdentityFuture, ResolveIdentity,
+        Identity, IdentityCacheLocation, IdentityFuture, ResolveCachedIdentity, ResolveIdentity,
+    };
+    use aws_smithy_runtime_api::client::interceptors::SharedInterceptor;
+    use aws_smithy_runtime_api::client::runtime_components::{
+        GetIdentityResolver, RuntimeComponents,
     };
-    use aws_smithy_runtime_api::client::runtime_components::RuntimeComponents;
     use aws_smithy_types::config_bag::ConfigBag;
 
     #[derive(Debug)]
@@ -95,10 +129,93 @@ pub(crate) mod identity_provider {
     #[derive(Default)]
     pub(crate) struct Builder;
 
+    impl TryFrom<SessionCredentials> for Credentials {
+        type Error = BoxError;
+
+        fn try_from(session_creds: SessionCredentials) -> Result<Self, Self::Error> {
+            Ok(Credentials::new(
+                session_creds.access_key_id,
+                session_creds.secret_access_key,
+                Some(session_creds.session_token),
+                Some(SystemTime::try_from(session_creds.expiration).map_err(|_| {
+                    CredentialsError::unhandled(
+                        "credential expiration time cannot be represented by a SystemTime",
+                    )
+                })?),
+                "s3express",
+            ))
+        }
+    }
+
     impl DefaultS3ExpressIdentityProvider {
         pub(crate) fn builder() -> Builder {
             Builder
         }
+
+        async fn identity<'a>(
+            &'a self,
+            runtime_components: &'a RuntimeComponents,
+            config_bag: &'a ConfigBag,
+        ) -> Result<Identity, BoxError> {
+            let bucket_name = self.bucket_name(config_bag)?;
+
+            let sigv4_identity_resolver = runtime_components
+                .identity_resolver(aws_runtime::auth::sigv4::SCHEME_ID)
+                .ok_or("identity resolver for sigv4 should be set for S3")?;
+            let _aws_identity = runtime_components
+                .identity_cache()
+                .resolve_cached_identity(sigv4_identity_resolver, runtime_components, config_bag)
+                .await?;
+
+            // TODO(S3Express): use both `bucket_name` and `aws_identity` as part of `S3ExpressIdentityCache` implementation
+
+            let express_session_credentials = self
+                .express_session_credentials(bucket_name, runtime_components, config_bag)
+                .await?;
+
+            let data = Credentials::try_from(express_session_credentials)?;
+
+            Ok(Identity::new(data.clone(), data.expiry()))
+        }
+
+        fn bucket_name<'a>(&'a self, config_bag: &'a ConfigBag) -> Result<&'a str, BoxError> {
+            let params = config_bag
+                .load::<EndpointResolverParams>()
+                .expect("endpoint resolver params must be set");
+            let params = params
+                .get::<crate::config::endpoint::Params>()
+                .expect("`Params` should be wrapped in `EndpointResolverParams`");
+            params
+                .bucket()
+                .ok_or("A bucket was not set in endpoint params".into())
+        }
+
+        async fn express_session_credentials<'a>(
+            &'a self,
+            bucket_name: &'a str,
+            runtime_components: &'a RuntimeComponents,
+            config_bag: &'a ConfigBag,
+        ) -> Result<SessionCredentials, BoxError> {
+            let mut config_builder = crate::config::Builder::from_config_bag(config_bag);
+
+            // inherits all runtime components from a current S3 operation but clears out
+            // out interceptors configured for that operation
+            let mut rc_builder = runtime_components.to_builder();
+            rc_builder.set_interceptors(std::iter::empty::<SharedInterceptor>());
+            config_builder.runtime_components = rc_builder;
+
+            let client = crate::Client::from_conf(config_builder.build());
+            let response = client
+                .create_session()
+                .bucket(bucket_name)
+                .session_mode(crate::types::SessionMode::ReadWrite)
+                .send()
+                .await?;
+
+            response
+                .credentials
+                .ok_or("no session credentials in response".into())
+        }
     }
 
     impl Builder {
@@ -112,10 +229,10 @@ pub(crate) mod identity_provider {
     impl ResolveIdentity for DefaultS3ExpressIdentityProvider {
         fn resolve_identity<'a>(
             &'a self,
-            _runtime_components: &'a RuntimeComponents,
-            _config_bag: &'a ConfigBag,
+            runtime_components: &'a RuntimeComponents,
+            config_bag: &'a ConfigBag,
         ) -> IdentityFuture<'a> {
-            todo!()
+            IdentityFuture::new(async move { self.identity(runtime_components, config_bag).await })
         }
 
         fn cache_location(&self) -> IdentityCacheLocation {

aws/rust-runtime/aws-runtime/src/auth/sigv4.rs

@@ -72,7 +72,8 @@ impl SigV4Signer {
         Self
     }
 
-    fn settings(operation_config: &SigV4OperationSigningConfig) -> SigningSettings {
+    /// Creates a [`SigningSettings`] from the given `operation_config`.
+    pub fn signing_settings(operation_config: &SigV4OperationSigningConfig) -> SigningSettings {
         super::settings(operation_config)
     }
 
@@ -117,10 +118,11 @@ impl SigV4Signer {
             .expect("all required fields set"))
     }
 
-    fn extract_operation_config<'a>(
+    /// Extracts a [`SigV4OperationSigningConfig`].
+    pub fn extract_operation_config<'a>(
         auth_scheme_endpoint_config: AuthSchemeEndpointConfig<'a>,
         config_bag: &'a ConfigBag,
-    ) -> Result<Cow<'a, SigV4OperationSigningConfig>, SigV4SigningError> {
+    ) -> Result<Cow<'a, SigV4OperationSigningConfig>, BoxError> {
         let operation_config = config_bag
             .load::<SigV4OperationSigningConfig>()
             .ok_or(SigV4SigningError::MissingOperationSigningConfig)?;
@@ -141,28 +143,27 @@ impl SigV4Signer {
             }
         }
     }
-}
 
-impl Sign for SigV4Signer {
-    fn sign_http_request(
+    /// Signs the given `request`.
+    ///
+    /// This is a helper used by [`Sign::sign_http_request`] and will be useful if calling code
+    /// needs to pass a configured `settings`.
+    ///
+    /// TODO(S3Express): Make this method more user friendly, possibly returning a builder
+    ///  instead of taking these input parameters. The builder will have a `sign` method that
+    ///  does what this method body currently does.
+    pub fn sign_http_request(
         &self,
         request: &mut HttpRequest,
         identity: &Identity,
-        auth_scheme_endpoint_config: AuthSchemeEndpointConfig<'_>,
+        settings: SigningSettings,
+        operation_config: &SigV4OperationSigningConfig,
         runtime_components: &RuntimeComponents,
-        config_bag: &ConfigBag,
+        #[allow(unused_variables)] config_bag: &ConfigBag,
     ) -> Result<(), BoxError> {
-        let operation_config =
-            Self::extract_operation_config(auth_scheme_endpoint_config, config_bag)?;
         let request_time = runtime_components.time_source().unwrap_or_default().now();
-
-        if identity.data::<Credentials>().is_none() {
-            return Err(SigV4SigningError::WrongIdentityType(identity.clone()).into());
-        };
-
-        let settings = Self::settings(&operation_config);
         let signing_params =
-            Self::signing_params(settings, identity, &operation_config, request_time)?;
+            Self::signing_params(settings, identity, operation_config, request_time)?;
 
         let (signing_instructions, _signature) = {
             // A body that is already in memory can be signed directly. A body that is not in memory
@@ -218,6 +219,35 @@ impl Sign for SigV4Signer {
     }
 }
 
+impl Sign for SigV4Signer {
+    fn sign_http_request(
+        &self,
+        request: &mut HttpRequest,
+        identity: &Identity,
+        auth_scheme_endpoint_config: AuthSchemeEndpointConfig<'_>,
+        runtime_components: &RuntimeComponents,
+        config_bag: &ConfigBag,
+    ) -> Result<(), BoxError> {
+        if identity.data::<Credentials>().is_none() {
+            return Err(SigV4SigningError::WrongIdentityType(identity.clone()).into());
+        };
+
+        let operation_config =
+            Self::extract_operation_config(auth_scheme_endpoint_config, config_bag)?;
+
+        let settings = Self::signing_settings(&operation_config);
+
+        self.sign_http_request(
+            request,
+            identity,
+            settings,
+            &operation_config,
+            runtime_components,
+            config_bag,
+        )
+    }
+}
+
 #[cfg(feature = "event-stream")]
 mod event_stream {
     use aws_sigv4::event_stream::{sign_empty_message, sign_message};

aws/rust-runtime/aws-sigv4/src/http_request/canonical_request.rs

@@ -10,8 +10,8 @@ use crate::http_request::settings::UriPathNormalizationMode;
 use crate::http_request::sign::SignableRequest;
 use crate::http_request::uri_path_normalization::normalize_uri_path;
 use crate::http_request::url_escape::percent_encode_path;
-use crate::http_request::PercentEncodingMode;
 use crate::http_request::{PayloadChecksumKind, SignableBody, SignatureLocation, SigningParams};
+use crate::http_request::{PercentEncodingMode, SigningSettings};
 use crate::sign::v4::sha256_hex_string;
 use crate::SignatureVersion;
 use aws_smithy_http::query_writer::QueryWriter;
@@ -218,7 +218,7 @@ impl<'a> CanonicalRequest<'a> {
         let creq = CanonicalRequest {
             method: req.method(),
             path,
-            params: Self::params(req.uri(), &values),
+            params: Self::params(req.uri(), &values, params.settings()),
             headers: canonical_headers,
             values,
         };
@@ -250,6 +250,11 @@ impl<'a> CanonicalRequest<'a> {
 
         Self::insert_host_header(&mut canonical_headers, req.uri());
 
+        let token_header_name = params
+            .settings()
+            .session_token_name_override
+            .unwrap_or(header::X_AMZ_SECURITY_TOKEN);
+
         if params.settings().signature_location == SignatureLocation::Headers {
             let creds = params
                 .credentials()
@@ -259,7 +264,7 @@ impl<'a> CanonicalRequest<'a> {
             if let Some(security_token) = creds.session_token() {
                 let mut sec_header = HeaderValue::from_str(security_token)?;
                 sec_header.set_sensitive(true);
-                canonical_headers.insert(header::X_AMZ_SECURITY_TOKEN, sec_header);
+                canonical_headers.insert(token_header_name, sec_header);
             }
 
             if params.settings().payload_checksum_kind == PayloadChecksumKind::XAmzSha256 {
@@ -283,7 +288,7 @@ impl<'a> CanonicalRequest<'a> {
             }
 
             if params.settings().session_token_mode == SessionTokenMode::Exclude
-                && name == HeaderName::from_static(header::X_AMZ_SECURITY_TOKEN)
+                && name == HeaderName::from_static(token_header_name)
             {
                 continue;
             }
@@ -320,7 +325,11 @@ impl<'a> CanonicalRequest<'a> {
         }
     }
 
-    fn params(uri: &Uri, values: &SignatureValues<'_>) -> Option<String> {
+    fn params(
+        uri: &Uri,
+        values: &SignatureValues<'_>,
+        settings: &SigningSettings,
+    ) -> Option<String> {
         let mut params: Vec<(Cow<'_, str>, Cow<'_, str>)> =
             form_urlencoded::parse(uri.query().unwrap_or_default().as_bytes()).collect();
         fn add_param<'a>(params: &mut Vec<(Cow<'a, str>, Cow<'a, str>)>, k: &'a str, v: &'a str) {
@@ -345,7 +354,13 @@ impl<'a> CanonicalRequest<'a> {
             );
 
             if let Some(security_token) = values.security_token {
-                add_param(&mut params, param::X_AMZ_SECURITY_TOKEN, security_token);
+                add_param(
+                    &mut params,
+                    settings
+                        .session_token_name_override
+                        .unwrap_or(param::X_AMZ_SECURITY_TOKEN),
+                    security_token,
+                );
             }
         }
         // Sort by param name, and then by param value

aws/rust-runtime/aws-sigv4/src/http_request/settings.rs

@@ -37,6 +37,10 @@ pub struct SigningSettings {
     /// canonical request. Other services require only it to be added after
     /// calculating the signature.
     pub session_token_mode: SessionTokenMode,
+
+    /// Some services require an alternative session token header or query param instead of
+    /// `x-amz-security-token` or `X-Amz-Security-Token`.
+    pub session_token_name_override: Option<&'static str>,
 }
 
 /// HTTP payload checksum type
@@ -133,6 +137,7 @@ impl Default for SigningSettings {
             excluded_headers,
             uri_path_normalization_mode: UriPathNormalizationMode::Enabled,
             session_token_mode: SessionTokenMode::Include,
+            session_token_name_override: None,
         }
     }
 }