fix(python, rust): remove imds calls from profile auth and region (#2442

)

# Description
The AWS SDK uses EC2 instance metadata in the default provider chain,
the profile chain and the region provider

# Related Issue(s)
<!---
For example:

- closes #106
--->
- closes #2377 
# Documentation

<!---
Share links to useful documentation
--->

crates/aws/src/credentials.rs

@@ -1,48 +1,73 @@
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 use aws_config::{
-    ecs::EcsCredentialsProvider, environment::EnvironmentVariableCredentialsProvider,
-    imds::credentials::ImdsCredentialsProvider, meta::credentials::CredentialsProviderChain,
-    profile::ProfileFileCredentialsProvider, provider_config::ProviderConfig,
+    ecs::EcsCredentialsProvider,
+    environment::{EnvironmentVariableCredentialsProvider, EnvironmentVariableRegionProvider},
+    imds::credentials::ImdsCredentialsProvider,
+    meta::{credentials::CredentialsProviderChain, region::RegionProviderChain},
+    profile::ProfileFileCredentialsProvider,
+    provider_config::ProviderConfig,
     web_identity_token::WebIdentityTokenCredentialsProvider,
 };
 use aws_credential_types::provider::{self, ProvideCredentials};
 use tracing::Instrument;
 
+const IMDS_PROVIDER_NAME: &str = "Ec2InstanceMetadata";
+
 #[derive(Debug)]
 pub struct ConfiguredCredentialChain {
     provider_chain: CredentialsProviderChain,
 }
 
-impl ConfiguredCredentialChain {
-    pub fn new(disable_imds: bool, imds_timeout: u64, config: Option<ProviderConfig>) -> Self {
-        let conf = config.unwrap_or_default();
+#[derive(Debug)]
+pub struct NoOpCredentials {}
+
+pub fn new_region_provider(
+    configuration: &ProviderConfig,
+    disable_imds: bool,
+    imds_timeout: u64,
+) -> RegionProviderChain {
+    let env_provider = EnvironmentVariableRegionProvider::new();
+    let profile_file = aws_config::profile::region::Builder::default()
+        .configure(configuration)
+        .build();
+    if disable_imds {
+        return RegionProviderChain::first_try(env_provider).or_else(profile_file);
+    }
+
+    RegionProviderChain::first_try(env_provider)
+        .or_else(profile_file)
+        .or_else(
+            aws_config::imds::region::Builder::default()
+                .configure(configuration)
+                .imds_client(
+                    aws_config::imds::Client::builder()
+                        .connect_timeout(Duration::from_millis(imds_timeout))
+                        .read_timeout(Duration::from_millis(imds_timeout))
+                        .build(),
+                )
+                .build(),
+        )
+}
 
+impl ConfiguredCredentialChain {
+    pub fn new(disable_imds: bool, imds_timeout: u64, conf: &ProviderConfig) -> Self {
+        let imds_provider = Self::build_imds_provider(conf, disable_imds, imds_timeout);
         let env_provider = EnvironmentVariableCredentialsProvider::default();
         let profile_provider = ProfileFileCredentialsProvider::builder()
-            .configure(&conf)
+            .configure(conf)
+            .with_custom_provider(IMDS_PROVIDER_NAME, imds_provider.clone())
             .build();
         let web_identity_token_provider = WebIdentityTokenCredentialsProvider::builder()
-            .configure(&conf)
+            .configure(conf)
             .build();
-        let ecs_provider = EcsCredentialsProvider::builder().configure(&conf).build();
+        let ecs_provider = EcsCredentialsProvider::builder().configure(conf).build();
 
-        let mut provider_chain = CredentialsProviderChain::first_try("Environment", env_provider)
+        let provider_chain = CredentialsProviderChain::first_try("Environment", env_provider)
             .or_else("Profile", profile_provider)
             .or_else("WebIdentityToken", web_identity_token_provider)
-            .or_else("EcsContainer", ecs_provider);
-        if !disable_imds {
-            let imds_provider = ImdsCredentialsProvider::builder()
-                .configure(&conf)
-                .imds_client(
-                    aws_config::imds::Client::builder()
-                        .connect_timeout(Duration::from_millis(imds_timeout))
-                        .read_timeout(Duration::from_millis(imds_timeout))
-                        .build(),
-                )
-                .build();
-            provider_chain = provider_chain.or_else("Ec2InstanceMetadata", imds_provider);
-        }
+            .or_else("EcsContainer", ecs_provider)
+            .or_else(IMDS_PROVIDER_NAME, imds_provider);
 
         Self { provider_chain }
     }
@@ -53,6 +78,27 @@ impl ConfiguredCredentialChain {
             .instrument(tracing::debug_span!("provide_credentials", provider = %"default_chain"))
             .await
     }
+
+    fn build_imds_provider(
+        conf: &ProviderConfig,
+        disable_imds: bool,
+        imds_timeout: u64,
+    ) -> Arc<dyn ProvideCredentials> {
+        if disable_imds {
+            return Arc::new(NoOpCredentials {});
+        }
+
+        let imds_provider = ImdsCredentialsProvider::builder()
+            .configure(conf)
+            .imds_client(
+                aws_config::imds::Client::builder()
+                    .connect_timeout(Duration::from_millis(imds_timeout))
+                    .read_timeout(Duration::from_millis(imds_timeout))
+                    .build(),
+            )
+            .build();
+        Arc::new(imds_provider)
+    }
 }
 
 impl ProvideCredentials for ConfiguredCredentialChain {
@@ -65,3 +111,14 @@ impl ProvideCredentials for ConfiguredCredentialChain {
         aws_credential_types::provider::future::ProvideCredentials::new(self.credentials())
     }
 }
+
+impl ProvideCredentials for NoOpCredentials {
+    fn provide_credentials<'a>(&'a self) -> provider::future::ProvideCredentials<'a>
+    where
+        Self: 'a,
+    {
+        aws_credential_types::provider::future::ProvideCredentials::new(std::future::ready(Err(
+            provider::error::CredentialsError::not_loaded_no_source(),
+        )))
+    }
+}

crates/aws/src/storage.rs

@@ -1,5 +1,6 @@
 //! AWS S3 storage backend.
 
+use aws_config::provider_config::ProviderConfig;
 use aws_config::{Region, SdkConfig};
 use bytes::Bytes;
 use deltalake_core::storage::object_store::{
@@ -166,11 +167,15 @@ impl S3StorageOptions {
             .unwrap_or(false);
         let disable_imds = str_option(options, s3_constants::AWS_EC2_METADATA_DISABLED)
             .map(|val| str_is_truthy(&val))
-            .unwrap_or(false);
+            .unwrap_or(true);
         let imds_timeout =
             Self::u64_or_default(options, s3_constants::AWS_EC2_METADATA_TIMEOUT, 100);
-        let credentials_provider =
-            crate::credentials::ConfiguredCredentialChain::new(disable_imds, imds_timeout, None);
+        let provider_config = ProviderConfig::default();
+        let credentials_provider = crate::credentials::ConfiguredCredentialChain::new(
+            disable_imds,
+            imds_timeout,
+            &provider_config,
+        );
         #[cfg(feature = "native-tls")]
         let sdk_config = execute_sdk_future(
             aws_config::from_env()
@@ -179,13 +184,23 @@ impl S3StorageOptions {
                         .map(|val| str_is_truthy(&val))
                         .unwrap_or(false),
                 ))
+                .region(crate::credentials::new_region_provider(
+                    &provider_config,
+                    disable_imds,
+                    imds_timeout,
+                ))
                 .credentials_provider(credentials_provider)
                 .load(),
         )?;
         #[cfg(feature = "rustls")]
         let sdk_config = execute_sdk_future(
             aws_config::from_env()
                 .credentials_provider(credentials_provider)
+                .region(crate::credentials::new_region_provider(
+                    &provider_config,
+                    disable_imds,
+                    imds_timeout,
+                ))
                 .load(),
         )?;
 
@@ -445,7 +460,7 @@ pub mod s3_constants {
     pub const AWS_S3_ALLOW_UNSAFE_RENAME: &str = "AWS_S3_ALLOW_UNSAFE_RENAME";
 
     /// If set to "true", disables the imds client
-    /// Defaults to "false"
+    /// Defaults to "true"
     pub const AWS_EC2_METADATA_DISABLED: &str = "AWS_EC2_METADATA_DISABLED";
 
     /// The timeout in milliseconds for the EC2 metadata endpoint
@@ -793,7 +808,7 @@ mod tests {
                 default_time.as_micros(),
             );
             assert!(disabled_time < enabled_time);
-            assert!(disabled_time < default_time);
+            assert!(default_time < enabled_time);
         })
         .await;
     }