Build SigningConfig from AuthScheme for each request (#428)

* Add default signing config for each request

Signed-off-by: Ankit Saurabh <sauraank@amazon.co.uk>

* Build SigningConfig from AuthScheme

Use the AuthScheme obtained from the EndpointResolver to build the SigningConfig
for each request. Also extend the initializer for SigningConfig to accept the
additional parameters: service name, signing algorithm, and the use_double_uri_encode
flag.

The AuthScheme will now validate the `scheme_name` field (i.e. signing algorithm)
on parsing and store it as a `SigningAlgorithm`.

Signed-off-by: Alessandro Passaro <alexpax@amazon.co.uk>

* Log auth_scheme

Signed-off-by: Alessandro Passaro <alexpax@amazon.co.uk>

* Parse signingRegionSet if signingRegion is not present

Signed-off-by: Alessandro Passaro <alexpax@amazon.co.uk>

* Make SigningConfig not Clone

Signed-off-by: Alessandro Passaro <alexpax@amazon.co.uk>

* Fix order of error fields

Signed-off-by: James Bornholt <bornholt@amazon.com>

---------

Signed-off-by: Ankit Saurabh <sauraank@amazon.co.uk>
Signed-off-by: Alessandro Passaro <alexpax@amazon.co.uk>
Signed-off-by: James Bornholt <bornholt@amazon.com>
Co-authored-by: Ankit Saurabh <sauraank@amazon.co.uk>
Co-authored-by: James Bornholt <bornholt@amazon.com>

mountpoint-s3-client/src/endpoint_config.rs

@@ -1,6 +1,7 @@
 use std::os::unix::prelude::OsStrExt;
 
 use mountpoint_s3_crt::{
+    auth::signing_config::SigningAlgorithm,
     common::{allocator::Allocator, uri::Uri},
     s3::endpoint_resolver::{RequestContext, ResolvedEndpoint, ResolverError, RuleEngine},
 };
@@ -18,7 +19,7 @@ pub enum AddressingStyle {
 #[derive(Debug, Clone)]
 pub struct AuthScheme {
     disable_double_encoding: bool,
-    scheme_name: String,
+    scheme_name: SigningAlgorithm,
     signing_name: String,
     signing_region: String,
 }
@@ -40,8 +41,8 @@ impl AuthScheme {
     }
 
     /// Get the name of [AuthScheme]
-    pub fn scheme_name(&self) -> &str {
-        &self.scheme_name
+    pub fn scheme_name(&self) -> SigningAlgorithm {
+        self.scheme_name
     }
 }
 
@@ -207,30 +208,33 @@ impl ResolvedEndpointInfo {
         // {\"authSchemes\":[{\"disableDoubleEncoding\":true,\"name\":\"sigv4\",\"signingName\":\"s3\",\"signingRegion\":\"us-east-2\"}]}
         let endpoint_properties = self.0.get_properties();
         let auth_scheme_data: serde_json::Value = serde_json::from_slice(endpoint_properties.as_bytes())?;
-        let auth_scheme_value = auth_scheme_data
-            .get("authSchemes")
-            .ok_or_else(|| EndpointError::InvalidAuthScheme("AuthScheme".to_owned()))?;
-        let auth_scheme_value = &auth_scheme_value[0];
-        let disable_double_encoding = auth_scheme_value
-            .get("disableDoubleEncoding")
-            .and_then(|t| t.as_bool())
-            .ok_or_else(|| EndpointError::InvalidAuthScheme("disableDoubleEncoding".to_owned()))?;
-        let scheme_name = auth_scheme_value
-            .get("name")
-            .and_then(|t| t.as_str())
-            .ok_or_else(|| EndpointError::InvalidAuthScheme("name".to_owned()))?;
-        let signing_name = auth_scheme_value
-            .get("signingName")
-            .and_then(|t| t.as_str())
-            .ok_or_else(|| EndpointError::InvalidAuthScheme("signingName".to_owned()))?;
+        let auth_scheme_value = auth_scheme_data["authSchemes"]
+            .get(0)
+            .ok_or_else(|| EndpointError::MissingAuthSchemeField("authSchemes"))?;
+        let disable_double_encoding = auth_scheme_value["disableDoubleEncoding"]
+            .as_bool()
+            .ok_or_else(|| EndpointError::MissingAuthSchemeField("disableDoubleEncoding"))?;
+        let scheme_name = auth_scheme_value["name"]
+            .as_str()
+            .ok_or_else(|| EndpointError::MissingAuthSchemeField("name"))?;
+        let scheme_name = match scheme_name {
+            "sigv4" => SigningAlgorithm::SigV4,
+            "sigv4a" => SigningAlgorithm::SigV4A,
+            _ => return Err(EndpointError::InvalidAuthSchemeField("name", scheme_name.to_owned())),
+        };
+
+        let signing_name = auth_scheme_value["signingName"]
+            .as_str()
+            .ok_or_else(|| EndpointError::MissingAuthSchemeField("signingName"))?;
         let signing_region = auth_scheme_value
             .get("signingRegion")
+            .or_else(|| auth_scheme_value["signingRegionSet"].get(0))
             .and_then(|t| t.as_str())
-            .unwrap_or("*");
+            .ok_or_else(|| EndpointError::MissingAuthSchemeField("signingRegion or signingRegionSet"))?;
 
         Ok(AuthScheme {
             disable_double_encoding,
-            scheme_name: scheme_name.to_owned(),
+            scheme_name,
             signing_name: signing_name.to_owned(),
             signing_region: signing_region.to_owned(),
         })
@@ -246,7 +250,9 @@ pub enum EndpointError {
     #[error("Endpoint properties could not be parsed")]
     ParseError(#[from] serde_json::Error),
     #[error("AuthScheme field missing: {0}")]
-    InvalidAuthScheme(String),
+    MissingAuthSchemeField(&'static str),
+    #[error("invalid value {1} for AuthScheme field {0}")]
+    InvalidAuthSchemeField(&'static str, String),
 }
 
 #[derive(Debug, Error)]

mountpoint-s3-client/src/s3_crt_client.rs

@@ -12,6 +12,7 @@ use std::time::{Duration, Instant};
 use mountpoint_s3_crt::auth::credentials::{
     CredentialsProvider, CredentialsProviderChainDefaultOptions, CredentialsProviderProfileOptions,
 };
+use mountpoint_s3_crt::auth::signing_config::SigningConfig;
 use mountpoint_s3_crt::common::allocator::Allocator;
 use mountpoint_s3_crt::common::uri::Uri;
 use mountpoint_s3_crt::http::request_response::{Header, Headers, Message};
@@ -197,6 +198,7 @@ struct S3CrtClientInner {
     request_payer: Option<String>,
     part_size: usize,
     bucket_owner: Option<String>,
+    credentials_provider: Option<CredentialsProvider>,
 }
 
 impl S3CrtClientInner {
@@ -255,11 +257,6 @@ impl S3CrtClientInner {
 
         let endpoint_config = config.endpoint_config;
 
-        if let Some(credentials_provider) = credentials_provider {
-            let signing_config = init_default_signing_config(endpoint_config.get_region(), credentials_provider);
-            client_config.signing_config(signing_config);
-        }
-
         client_config
             .client_bootstrap(client_bootstrap)
             .retry_strategy(retry_strategy);
@@ -291,6 +288,7 @@ impl S3CrtClientInner {
             request_payer: config.request_payer,
             part_size: config.part_size,
             bucket_owner: config.bucket_owner,
+            credentials_provider,
         })
     }
 
@@ -302,6 +300,27 @@ impl S3CrtClientInner {
         let endpoint = self.endpoint_config.resolve_for_bucket(bucket)?;
         let uri = endpoint.uri()?;
         trace!(?uri, "resolved endpoint");
+
+        let signing_config = if let Some(credentials_provider) = &self.credentials_provider {
+            let auth_scheme = match endpoint.auth_scheme() {
+                Ok(auth_scheme) => auth_scheme,
+                Err(e) => {
+                    error!(error=?e, "no auth scheme for endpoint");
+                    return Err(e.into());
+                }
+            };
+            trace!(?auth_scheme, "resolved auth scheme");
+            Some(init_default_signing_config(
+                auth_scheme.signing_region(),
+                auth_scheme.scheme_name(),
+                auth_scheme.signing_name(),
+                !auth_scheme.disable_double_encoding(),
+                credentials_provider.clone(),
+            ))
+        } else {
+            None
+        };
+
         let hostname = uri.host_name().to_str().unwrap();
         let path_prefix = uri.path().to_os_string().into_string().unwrap();
         let port = uri.host_port();
@@ -330,6 +349,7 @@ impl S3CrtClientInner {
             uri,
             path_prefix,
             checksum_config: None,
+            signing_config,
         })
     }
 
@@ -338,6 +358,9 @@ impl S3CrtClientInner {
         if let Some(checksum_config) = message.checksum_config {
             options.checksum_config(checksum_config);
         }
+        if let Some(signing_config) = message.signing_config {
+            options.signing_config(signing_config);
+        }
         options
             .message(message.inner)
             .endpoint(message.uri)
@@ -601,6 +624,7 @@ struct S3Message {
     uri: Uri,
     path_prefix: String,
     checksum_config: Option<ChecksumConfig>,
+    signing_config: Option<SigningConfig>,
 }
 
 impl S3Message {

mountpoint-s3-crt/src/auth/credentials.rs

@@ -59,6 +59,11 @@ pub struct CredentialsProvider {
     pub(crate) inner: NonNull<aws_credentials_provider>,
 }
 
+// SAFETY: aws_credentials_provider is thread-safe.
+unsafe impl Send for CredentialsProvider {}
+// SAFETY: aws_credentials_provider is thread-safe.
+unsafe impl Sync for CredentialsProvider {}
+
 impl CredentialsProvider {
     /// Creates the default credential provider chain as used by most AWS SDKs
     pub fn new_chain_default(

mountpoint-s3-crt/src/auth/signing_config.rs

@@ -1,12 +1,11 @@
 //! Configuration for signing requests to AWS APIs
 
 use crate::auth::credentials::CredentialsProvider;
-use mountpoint_s3_crt_sys::aws_signing_config_aws;
+use mountpoint_s3_crt_sys::{aws_signing_algorithm, aws_signing_config_aws};
 use std::ffi::OsString;
 use std::fmt::Debug;
 use std::marker::PhantomPinned;
 use std::pin::Pin;
-use std::sync::Arc;
 
 pub(crate) struct SigningConfigInner {
     /// The raw `aws_signing_config` for this config
@@ -15,9 +14,12 @@ pub(crate) struct SigningConfigInner {
     /// An owned copy of the region string, since the `aws_signing_config_aws` holds a pointer to it.
     pub(crate) region: OsString,
 
-    /// An owned CredentialProvider, since the holds `aws_signing_config_aws` a pointer to it inner.
+    /// An owned CredentialProvider, since the `aws_signing_config_aws` holds a pointer to it.
     pub(crate) credentials_provider: CredentialsProvider,
 
+    /// An owned copy of the service string, since the `aws_signing_config_aws` holds a pointer to it.
+    pub(crate) service: OsString,
+
     /// This forces the struct to be !Unpin, because the signing config can contain pointers to itself
     pub(crate) _pinned: PhantomPinned,
 }
@@ -30,15 +32,31 @@ impl Debug for SigningConfigInner {
     }
 }
 
-/// Wrap the SigningConfigInner struct into a Pin<Box<_>>, so that it cannot be moved. Then wrap
-/// in an Arc so that we can make copies. If there were a way to convert Pin<Box<_>> into Pin<Arc<_>>
-/// then we could avoid needing both.
-#[derive(Debug, Clone)]
-pub struct SigningConfig(pub(crate) Arc<Pin<Box<SigningConfigInner>>>);
+/// Wrap the SigningConfigInner struct into a Pin<Box<_>>, so that it cannot be moved.
+#[derive(Debug)]
+pub struct SigningConfig(pub(crate) Pin<Box<SigningConfigInner>>);
 
 impl SigningConfig {
     /// Get out the inner pointer to the signing config
     pub(crate) fn to_inner_ptr(&self) -> *const aws_signing_config_aws {
-        &Pin::as_ref(&self.0).get_ref().inner
+        &self.0.as_ref().get_ref().inner
+    }
+}
+
+/// The version of the AWS signing process.
+#[derive(Debug, Copy, Clone)]
+pub enum SigningAlgorithm {
+    /// Signature Version 4 (SigV4)
+    SigV4,
+    /// Signature Version 4 Asymmetric (SigV4A)
+    SigV4A,
+}
+
+impl From<SigningAlgorithm> for aws_signing_algorithm {
+    fn from(typ: SigningAlgorithm) -> Self {
+        match typ {
+            SigningAlgorithm::SigV4 => aws_signing_algorithm::AWS_SIGNING_ALGORITHM_V4,
+            SigningAlgorithm::SigV4A => aws_signing_algorithm::AWS_SIGNING_ALGORITHM_V4_ASYMMETRIC,
+        }
     }
 }

mountpoint-s3-crt/src/s3/client.rs

@@ -1,7 +1,7 @@
 //! A client for high-throughput access to Amazon S3
 
 use crate::auth::credentials::CredentialsProvider;
-use crate::auth::signing_config::{SigningConfig, SigningConfigInner};
+use crate::auth::signing_config::{SigningAlgorithm, SigningConfig, SigningConfigInner};
 use crate::common::allocator::Allocator;
 use crate::common::error::Error;
 use crate::common::thread::ThreadId;
@@ -18,7 +18,6 @@ use std::marker::PhantomPinned;
 use std::os::unix::prelude::OsStrExt;
 use std::pin::Pin;
 use std::ptr::NonNull;
-use std::sync::Arc;
 use std::time::Duration;
 
 /// A client for high-throughput access to Amazon S3
@@ -30,7 +29,7 @@ pub struct Client {
     /// Hold on to an owned copy of the configuration so that it doesn't get dropped while the
     /// client still exists. This is because the client config holds ownership of some strings
     /// (like the region) that could still be used while the client exists.
-    config: ClientConfig,
+    _config: ClientConfig,
 }
 
 // SAFETY: We assume that the CRT allows making requests using the same client from multiple threads.
@@ -47,9 +46,6 @@ pub struct ClientConfig {
     /// The [ClientBootstrap] to use to create connections to S3
     client_bootstrap: Option<ClientBootstrap>,
 
-    /// The [SigningConfig] configuration for signing API requests to S3
-    signing_config: Option<SigningConfig>,
-
     /// The [RetryStrategy] to use to reschedule failed requests to S3. This is reference counted,
     /// so we only need to hold onto it until this [ClientConfig] is consumed, at which point the
     /// client will take ownership.
@@ -62,14 +58,6 @@ impl ClientConfig {
         Default::default()
     }
 
-    /// Signing options to be used for each request. Leave out to not sign requests.
-    pub fn signing_config(&mut self, signing_config: SigningConfig) -> &mut Self {
-        // Safety: Cast the signing config to mut pointer since we know creating the client won't modify it.
-        self.inner.signing_config = signing_config.to_inner_ptr() as *mut aws_signing_config_aws;
-        self.signing_config = Some(signing_config);
-        self
-    }
-
     /// Client bootstrap used for common staples such as event loop group, host resolver, etc.
     pub fn client_bootstrap(&mut self, client_bootstrap: ClientBootstrap) -> &mut Self {
         self.inner.client_bootstrap = client_bootstrap.inner.as_ptr();
@@ -280,7 +268,7 @@ impl MetaRequestOptions {
 
     /// Set the signing config used for this message. Not public because we copy it from the client
     /// when making a request.
-    fn signing_config(&mut self, signing_config: SigningConfig) -> &mut Self {
+    pub fn signing_config(&mut self, signing_config: SigningConfig) -> &mut Self {
         // SAFETY: we aren't moving out of the struct.
         let options = unsafe { Pin::get_unchecked_mut(Pin::as_mut(&mut self.0)) };
         options.signing_config = Some(signing_config);
@@ -565,21 +553,16 @@ impl Client {
         // guaranteed to be a valid allocator because of the type-safe wrapper.
         let inner = unsafe { aws_s3_client_new(allocator.inner.as_ptr(), &config.inner).ok_or_last_error()? };
 
-        Ok(Self { inner, config })
+        Ok(Self { inner, _config: config })
     }
 
     /// Make a meta request to S3 using this [Client]. A meta request is an HTTP request that
     /// the CRT might internally split up into multiple requests for performance.
-    pub fn make_meta_request(&self, mut options: MetaRequestOptions) -> Result<MetaRequest, Error> {
+    pub fn make_meta_request(&self, options: MetaRequestOptions) -> Result<MetaRequest, Error> {
         // SAFETY: The inner struct pointed to by MetaRequestOptions will live as long as the
         // request does, since we only drop it in the shutdown callback. That struct owns everything
         // related to the request, like the message, signing config, etc.
         unsafe {
-            // The client holds a copy of the signing config, copy it again for this request.
-            if let Some(signing_config) = self.config.signing_config.as_ref() {
-                options.signing_config(signing_config.clone());
-            }
-
             // Unpin the options (we won't move out of it, nor will the callbacks).
             let options = Pin::into_inner_unchecked(options.0);
 
@@ -1054,26 +1037,40 @@ impl From<aws_s3_request_type> for RequestType {
 
 /// Create a new [SigningConfig] with the default configuration for signing S3 requests to a region
 /// using the given [CredentialsProvider]
-pub fn init_default_signing_config(region: &str, credentials_provider: CredentialsProvider) -> SigningConfig {
+pub fn init_default_signing_config(
+    region: &str,
+    algorithm: SigningAlgorithm,
+    service: &str,
+    use_double_uri_encode: bool,
+    credentials_provider: CredentialsProvider,
+) -> SigningConfig {
     let mut signing_config = Box::new(SigningConfigInner {
         inner: Default::default(),
         region: region.to_owned().into(),
         credentials_provider,
+        service: service.to_owned().into(),
         _pinned: Default::default(),
     });
 
     let credentials_provider = signing_config.credentials_provider.inner.as_ptr();
-    // SAFETY: we copied the region into the signing_config (`region.to_owned()` above), so the byte
-    // cursor we create here will point to bytes that are valid as long as this SigningConfig is.
+    // SAFETY: `region` and `service` are owned by signing_config (see e.g. `region.to_owned()` above),
+    // so the byte cursors we create here will point to bytes that are valid as long as this SigningConfig is.
     // singing_config owns `credential_provider` that is valid as long as this SingingConfig is.
     unsafe {
         let region_cursor = signing_config.region.as_aws_byte_cursor();
-
         aws_s3_init_default_signing_config(&mut signing_config.inner, region_cursor, credentials_provider);
+
+        let service_cursor = signing_config.service.as_aws_byte_cursor();
+        signing_config.inner.service = service_cursor;
     }
-    signing_config.inner.flags.set_use_double_uri_encode(false as u32);
 
-    SigningConfig(Arc::new(Box::into_pin(signing_config)))
+    signing_config
+        .inner
+        .flags
+        .set_use_double_uri_encode(use_double_uri_encode as u32);
+    signing_config.inner.algorithm = algorithm.into();
+
+    SigningConfig(Box::into_pin(signing_config))
 }
 
 /// The checksum configuration.