fix(aws provider): Pass configured region to credentials provider (#1…

…2315)

* fix(aws provider): Pass configured region to credentials provider

Fixes: #12314

Relevant: #12313

Signed-off-by: Jesse Szwedko <jesse@szwedko.me>

src/aws/auth.rs

@@ -1,5 +1,7 @@
-use aws_config::{default_provider::credentials::default_provider, sts::AssumeRoleProviderBuilder};
-use aws_types::{credentials::SharedCredentialsProvider, Credentials};
+use aws_config::{
+    default_provider::credentials::DefaultCredentialsChain, sts::AssumeRoleProviderBuilder,
+};
+use aws_types::{credentials::SharedCredentialsProvider, region::Region, Credentials};
 use serde::{Deserialize, Serialize};
 
 /// Configuration for configuring authentication strategy for AWS.
@@ -28,7 +30,10 @@ pub enum AwsAuthentication {
 }
 
 impl AwsAuthentication {
-    pub async fn credentials_provider(&self) -> crate::Result<SharedCredentialsProvider> {
+    pub async fn credentials_provider(
+        &self,
+        region: Option<Region>,
+    ) -> crate::Result<SharedCredentialsProvider> {
         match self {
             Self::Static {
                 access_key_id,
@@ -41,12 +46,19 @@ impl AwsAuthentication {
             AwsAuthentication::File { .. } => {
                 Err("Overriding the credentials file is not supported.".into())
             }
-            AwsAuthentication::Role { assume_role } => Ok(SharedCredentialsProvider::new(
-                AssumeRoleProviderBuilder::new(assume_role)
-                    .build(default_credentials_provider().await),
-            )),
+            AwsAuthentication::Role { assume_role } => {
+                let mut credentials = AssumeRoleProviderBuilder::new(assume_role);
+
+                if let Some(ref region) = region {
+                    credentials = credentials.region(region.clone())
+                }
+
+                Ok(SharedCredentialsProvider::new(
+                    credentials.build(default_credentials_provider(region).await),
+                ))
+            }
             AwsAuthentication::Default {} => Ok(SharedCredentialsProvider::new(
-                default_credentials_provider().await,
+                default_credentials_provider(region).await,
             )),
         }
     }
@@ -60,8 +72,14 @@ impl AwsAuthentication {
     }
 }
 
-async fn default_credentials_provider() -> SharedCredentialsProvider {
-    SharedCredentialsProvider::new(default_provider().await)
+async fn default_credentials_provider(region: Option<Region>) -> SharedCredentialsProvider {
+    let mut credentials = DefaultCredentialsChain::builder();
+
+    if let Some(region) = region {
+        credentials = credentials.region(region)
+    }
+
+    SharedCredentialsProvider::new(credentials.build().await)
 }
 
 #[cfg(test)]

src/aws/mod.rs

@@ -82,7 +82,8 @@ pub async fn create_client<T: ClientBuilder>(
     proxy: &ProxyConfig,
     tls_options: &Option<TlsOptions>,
 ) -> crate::Result<T::Client> {
-    let mut config_builder = T::create_config_builder(auth.credentials_provider().await?);
+    let mut config_builder =
+        T::create_config_builder(auth.credentials_provider(region.clone()).await?);
 
     if let Some(endpoint_override) = endpoint {
         config_builder = T::with_endpoint_resolver(config_builder, endpoint_override);

src/sinks/aws_kinesis_firehose/integration_tests.rs

@@ -151,7 +151,7 @@ async fn ensure_elasticsearch_domain(domain_name: String) -> String {
         aws_sdk_elasticsearch::config::Builder::new()
             .credentials_provider(
                 AwsAuthentication::test_auth()
-                    .credentials_provider()
+                    .credentials_provider(test_region_endpoint().region())
                     .await
                     .unwrap(),
             )

src/sinks/elasticsearch/common.rs

@@ -77,7 +77,9 @@ impl ElasticsearchCommon {
 
         let aws_auth = match &config.auth {
             Some(ElasticsearchAuth::Basic { .. }) | None => None,
-            Some(ElasticsearchAuth::Aws(aws)) => Some(aws.credentials_provider().await?),
+            Some(ElasticsearchAuth::Aws(aws)) => {
+                Some(aws.credentials_provider(region.clone()).await?)
+            }
         };
 
         let compression = config.compression;

src/sources/aws_sqs/integration_tests.rs

@@ -51,7 +51,7 @@ async fn get_sqs_client() -> aws_sdk_sqs::Client {
     let config = aws_sdk_sqs::config::Builder::new()
         .credentials_provider(
             AwsAuthentication::test_auth()
-                .credentials_provider()
+                .credentials_provider(Some(Region::new("custom")))
                 .await
                 .unwrap(),
         )