fix(s3): align s3 handler additional creds providers

backend/ee-repo-ref.txt

@@ -1 +1 @@
-f667fcebdb5a2d1c22c1928d6a316ba57bb43d72
+150c3176f2c192366f2e8298ef6ece43780b529e

backend/windmill-common/src/s3_helpers.rs

@@ -117,10 +117,10 @@ pub struct S3Object {
 
 #[cfg(feature = "parquet")]
 pub async fn get_etag_or_empty(
-    object_store_resource: &ObjectStoreResource,
+    object_store_resource: &mut ObjectStoreResource,
     s3_object: S3Object,
 ) -> Option<String> {
-    let object_store_client = build_object_store_client(object_store_resource);
+    let object_store_client = build_object_store_client(object_store_resource).await;
     if object_store_client.is_err() {
         return None;
     }
@@ -166,11 +166,11 @@ pub fn render_endpoint(
 }
 
 #[cfg(feature = "parquet")]
-pub fn build_object_store_client(
+pub async fn build_object_store_client(
     resource_ref: &ObjectStoreResource,
 ) -> error::Result<Arc<dyn ObjectStore>> {
     match resource_ref {
-        ObjectStoreResource::S3(s3_resource_ref) => build_s3_client(&s3_resource_ref, None),
+        ObjectStoreResource::S3(s3_resource_ref) => build_s3_client(&s3_resource_ref).await,
         ObjectStoreResource::Azure(azure_blob_resource_ref) => {
             build_azure_blob_client(&azure_blob_resource_ref)
         }
@@ -225,10 +225,21 @@ use aws_config::{default_provider::credentials::DefaultCredentialsChain, Region}
 use object_store::CredentialProvider;
 
 #[cfg(feature = "parquet")]
-pub fn build_s3_client(
-    s3_resource_ref: &S3Resource,
-    credential_providers: Option<DefaultCredentialsChain>,
-) -> error::Result<Arc<dyn ObjectStore>> {
+pub async fn build_s3_client(s3_resource_ref: &S3Resource) -> error::Result<Arc<dyn ObjectStore>> {
+    let static_creds = s3_resource_ref.access_key.as_ref().is_some_and(|x| x != "")
+        || s3_resource_ref.secret_key.as_ref().is_some_and(|x| x != "");
+
+    let credentials_provider = if !static_creds {
+        Some(
+            DefaultCredentialsChain::builder()
+                .region(Region::new(s3_resource_ref.region.clone()))
+                .build()
+                .await,
+        )
+    } else {
+        None
+    };
+
     let s3_resource = s3_resource_ref.clone();
     let endpoint = render_endpoint(
         s3_resource.endpoint,
@@ -244,7 +255,7 @@ pub fn build_s3_client(
         .with_bucket_name(s3_resource.bucket)
         .with_endpoint(endpoint);
 
-    if let Some(credentials_provider) = credential_providers {
+    if let Some(credentials_provider) = credentials_provider {
         store_builder = store_builder.with_credentials(Arc::new(AwsCredentialAdapter {
             inner: credentials_provider,
         }));
@@ -399,18 +410,7 @@ pub async fn build_s3_client_from_settings(
 ) -> error::Result<Arc<dyn ObjectStore>> {
     let region = none_if_empty(settings.region)
         .unwrap_or_else(|| std::env::var("AWS_REGION").unwrap_or_else(|_| "us-east-1".to_string()));
-    let access_key = none_if_empty(settings.access_key);
-    let secret_key = none_if_empty(settings.secret_key);
-    let credentials_provider = if access_key.is_none() && secret_key.is_none() {
-        Some(
-            DefaultCredentialsChain::builder()
-                .region(Region::new(region.clone()))
-                .build()
-                .await,
-        )
-    } else {
-        None
-    };
+
     let s3_resource = S3Resource {
         endpoint: none_if_empty(settings.endpoint).unwrap_or_else(|| {
             std::env::var("S3_ENDPOINT").unwrap_or_else(|_| format!("s3.{region}.amazonaws.com"))
@@ -419,15 +419,15 @@ pub async fn build_s3_client_from_settings(
             std::env::var("S3_CACHE_BUCKET").unwrap_or_else(|_| "missingbucket".to_string())
         }),
         region,
-        access_key,
-        secret_key,
+        access_key: settings.access_key,
+        secret_key: settings.secret_key,
         use_ssl: !settings.allow_http.unwrap_or(true),
         path_style: settings.path_style,
         port: settings.port,
         token: None,
     };
 
-    build_s3_client(&s3_resource, credentials_provider)
+    build_s3_client(&s3_resource).await
 }
 
 #[cfg(feature = "parquet")]

backend/windmill-worker/src/common.rs

@@ -736,8 +736,8 @@ async fn arg_value_hash_additions(
                 .await;
         storage = s3_object.storage.clone();
 
-        if let Some(s3_resource) = s3_resource_opt.ok().flatten() {
-            let etag = get_etag_or_empty(&s3_resource, s3_object.clone()).await;
+        if let Some(mut s3_resource) = s3_resource_opt.ok().flatten() {
+            let etag = get_etag_or_empty(&mut s3_resource, s3_object.clone()).await;
             tracing::warn!("Enriching s3 arg value with etag: {:?}", etag);
             result.insert(s3_object.s3.clone(), etag.unwrap_or_default()); // TODO: maybe inject a random value to invalidate the cache?
         }
@@ -793,9 +793,9 @@ pub async fn get_cached_resource_value_if_valid(
                 return None;
             }
             for (s3_file_key, s3_file_etag) in s3_etags {
-                if let Some(object_store_resource) = object_store_resource_opt.clone() {
+                if let Some(mut object_store_resource) = object_store_resource_opt.clone() {
                     let etag = get_etag_or_empty(
-                        &object_store_resource,
+                        &mut object_store_resource,
                         S3Object {
                             s3: s3_file_key.clone(),
                             storage: cached_resource.storage.clone(),