fix: enforce user must specify access_key and secret_key using aws auth

src/connector/src/aws_auth.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use aws_config::default_provider::credentials::DefaultCredentialsChain;
 use aws_config::default_provider::region::DefaultRegionChain;
 use aws_config::sts::AssumeRoleProvider;
 use aws_credential_types::provider::SharedCredentialsProvider;
@@ -86,24 +85,12 @@ impl AwsAuthProps {
         }
     }
 
-    async fn build_credential_provider(&self) -> anyhow::Result<SharedCredentialsProvider> {
-        if self.access_key.is_some() && self.secret_key.is_some() {
-            Ok(SharedCredentialsProvider::new(
-                aws_credential_types::Credentials::from_keys(
-                    self.access_key.as_ref().unwrap(),
-                    self.secret_key.as_ref().unwrap(),
-                    self.session_token.clone(),
-                ),
-            ))
-        } else {
-            let region = self.build_region().await?;
-            let mut chain = DefaultCredentialsChain::builder().region(region);
-
-            if let Some(profile_name) = self.profile.as_ref() {
-                chain = chain.profile_name(profile_name)
-            }
-            Ok(SharedCredentialsProvider::new(chain.build().await))
-        }
+    fn build_credential_provider(&self) -> SharedCredentialsProvider {
+        SharedCredentialsProvider::new(aws_credential_types::Credentials::from_keys(
+            self.access_key.as_ref().unwrap_or(&"".into()),
+            self.secret_key.as_ref().unwrap_or(&"".into()),
+            self.session_token.clone(),
+        ))
     }
 
     async fn with_role_provider(
@@ -127,7 +114,7 @@ impl AwsAuthProps {
     pub async fn build_config(&self) -> anyhow::Result<SdkConfig> {
         let region = self.build_region().await?;
         let credentials_provider = self
-            .with_role_provider(self.build_credential_provider().await?)
+            .with_role_provider(self.build_credential_provider())
             .await?;
         let config_loader = aws_config::from_env()
             .region(region)

src/connector/src/aws_utils.rs

@@ -16,6 +16,7 @@ use std::collections::HashMap;
 use std::time::Duration;
 
 use aws_config::timeout::TimeoutConfig;
+use aws_sdk_s3::error::DisplayErrorContext;
 use aws_sdk_s3::{client as s3_client, config as s3_config};
 use risingwave_common::error::ErrorCode::InternalError;
 use risingwave_common::error::{Result, RwError};
@@ -117,12 +118,9 @@ pub async fn load_file_descriptor_from_s3(
     location: &Url,
     config: &AwsAuthProps,
 ) -> Result<Vec<u8>> {
-    let bucket = location.domain().ok_or_else(|| {
-        RwError::from(InternalError(format!(
-            "Illegal Protobuf schema path {}",
-            location
-        )))
-    })?;
+    let bucket = location
+        .domain()
+        .ok_or_else(|| RwError::from(InternalError(format!("Illegal file path {}", location))))?;
     let key = location.path().replace('/', "");
     let sdk_config = config.build_config().await?;
     let s3_client = s3_client(&sdk_config, Some(default_conn_config()));
@@ -132,13 +130,18 @@ pub async fn load_file_descriptor_from_s3(
         .key(&key)
         .send()
         .await
-        .map_err(|e| RwError::from(InternalError(e.to_string())))?;
+        .map_err(|e| {
+            RwError::from(InternalError(format!(
+                "get file {} err:{}",
+                location,
+                DisplayErrorContext(e)
+            )))
+        })?;
 
-    let body = response.body.collect().await.map_err(|e| {
-        RwError::from(InternalError(format!(
-            "Read Protobuf schema file from s3 {}",
-            e
-        )))
-    })?;
+    let body = response
+        .body
+        .collect()
+        .await
+        .map_err(|e| RwError::from(InternalError(format!("Read file from s3 {}", e))))?;
     Ok(body.into_bytes().to_vec())
 }

src/connector/src/source/filesystem/s3/enumerator.rs

@@ -12,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::Context;
+use anyhow::{anyhow, Context};
 use async_trait::async_trait;
 use aws_sdk_s3::client::Client;
+use aws_sdk_s3::error::DisplayErrorContext;
 use itertools::Itertools;
 
 use crate::aws_auth::AwsAuthProps;
@@ -106,7 +107,10 @@ impl SplitEnumerator for S3SplitEnumerator {
             if let Some(continuation_token) = next_continuation_token.take() {
                 req = req.continuation_token(continuation_token);
             }
-            let mut res = req.send().await?;
+            let mut res = req
+                .send()
+                .await
+                .map_err(|e| anyhow!(DisplayErrorContext(e)))?;
             objects.extend(res.contents.take().unwrap_or_default());
             if res.is_truncated() {
                 next_continuation_token = Some(res.next_continuation_token.unwrap())