feat(aws_s3): Add S3 compatibility environment variables for non-AWS services

crates/aws_s3/src/storage.rs

@@ -9,6 +9,7 @@ use async_trait::async_trait;
 use aws_config::retry::RetryConfig;
 use aws_sdk_s3::{
     operation::{
+        create_multipart_upload::builders::CreateMultipartUploadFluentBuilder,
         head_object::{
             HeadObjectError,
             HeadObjectOutput,
@@ -26,6 +27,8 @@ use aws_sdk_s3::{
     Client,
 };
 use aws_utils::{
+    are_checksums_disabled,
+    is_sse_disabled,
     must_s3_config_from_env,
     s3::S3Client,
 };
@@ -151,6 +154,27 @@ impl<RT: Runtime> S3Storage<RT> {
         let bucket_name = s3_bucket_name(&use_case)?;
         S3Storage::new_with_prefix(bucket_name, key_prefix, runtime).await
     }
+
+    /// Helper method to configure multipart upload builder with optional AWS headers
+    /// for S3 compatibility with non-AWS services
+    fn configure_multipart_upload_builder(
+        &self,
+        mut upload_builder: CreateMultipartUploadFluentBuilder,
+    ) -> CreateMultipartUploadFluentBuilder {
+        // Add server-side encryption if not disabled for S3 compatibility
+        if !is_sse_disabled() {
+            upload_builder = upload_builder.server_side_encryption(ServerSideEncryption::Aes256);
+        }
+
+        // Add checksum algorithm if not disabled for S3 compatibility
+        if !are_checksums_disabled() {
+            // Because we're using multipart uploads, we're really specifying the part checksum
+            // algorithm here, so it needs to match what we use for each part.
+            upload_builder = upload_builder.checksum_algorithm(ChecksumAlgorithm::Crc32);
+        }
+
+        upload_builder
+    }
 }
 
 async fn s3_client() -> Result<Client, anyhow::Error> {
@@ -216,15 +240,15 @@ impl<RT: Runtime> Storage for S3Storage<RT> {
     async fn start_upload(&self) -> anyhow::Result<Box<BufferedUpload>> {
         let key: ObjectKey = self.runtime.new_uuid_v4().to_string().try_into()?;
         let s3_key = S3Key(self.key_prefix.clone() + &key);
-        let output = self
+        let upload_builder = self
             .client
             .create_multipart_upload()
             .bucket(self.bucket.clone())
-            .key(&s3_key.0)
-            .server_side_encryption(ServerSideEncryption::Aes256)
-            // Because we're using multipart uploads, we're really specifying the part checksum
-            // algorithm here, so it needs to match what we use for each part.
-            .checksum_algorithm(ChecksumAlgorithm::Crc32)
+            .key(&s3_key.0);
+
+        let upload_builder = self.configure_multipart_upload_builder(upload_builder);
+
+        let output = upload_builder
             .send()
             .await
             .context("Failed to create multipart upload")?;
@@ -254,15 +278,15 @@ impl<RT: Runtime> Storage for S3Storage<RT> {
     async fn start_client_driven_upload(&self) -> anyhow::Result<ClientDrivenUploadToken> {
         let key: ObjectKey = self.runtime.new_uuid_v4().to_string().try_into()?;
         let s3_key = S3Key(self.key_prefix.clone() + &key);
-        let output = self
+        let upload_builder = self
             .client
             .create_multipart_upload()
             .bucket(self.bucket.clone())
-            .key(&s3_key.0)
-            .server_side_encryption(ServerSideEncryption::Aes256)
-            // Because we're using multipart uploads, we're really specifying the part checksum
-            // algorithm here, so it needs to match what we use for each part.
-            .checksum_algorithm(ChecksumAlgorithm::Crc32)
+            .key(&s3_key.0);
+
+        let upload_builder = self.configure_multipart_upload_builder(upload_builder);
+
+        let output = upload_builder
             .send()
             .await
             .context("Failed to create multipart upload")?;
@@ -565,15 +589,20 @@ impl<RT: Runtime> S3Upload<RT> {
         let part_number = self.next_part_number()?;
         crate::metrics::log_aws_s3_part_upload_size_bytes(data.len());
 
-        let builder = self
+        let mut builder = self
             .client
             .upload_part()
-            .checksum_algorithm(ChecksumAlgorithm::Crc32)
             .body(ByteStream::from(data))
             .bucket(self.bucket.clone())
             .key(&self.s3_key.0)
             .part_number(Into::<u16>::into(part_number) as i32)
             .upload_id(self.upload_id.to_string());
+
+        // Add checksum algorithm if not disabled for S3 compatibility
+        if !are_checksums_disabled() {
+            builder = builder.checksum_algorithm(ChecksumAlgorithm::Crc32);
+        }
+
         Ok(UploadPart {
             part_number,
             builder,

crates/aws_utils/src/lib.rs

@@ -33,6 +33,20 @@ static AWS_S3_FORCE_PATH_STYLE: LazyLock<bool> = LazyLock::new(|| {
         .unwrap_or_default()
 });
 
+static AWS_S3_DISABLE_SSE: LazyLock<bool> = LazyLock::new(|| {
+    env::var("AWS_S3_DISABLE_SSE")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or_default()
+});
+
+static AWS_S3_DISABLE_CHECKSUMS: LazyLock<bool> = LazyLock::new(|| {
+    env::var("AWS_S3_DISABLE_CHECKSUMS")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or_default()
+});
+
 /// Similar aws_config::from_env but returns an error if credentials or
 /// region is are not. It also doesn't spew out log lines every time
 /// credentials are accessed.
@@ -62,3 +76,13 @@ pub async fn must_s3_config_from_env() -> anyhow::Result<S3ConfigBuilder> {
     s3_config_builder = s3_config_builder.force_path_style(*AWS_S3_FORCE_PATH_STYLE);
     Ok(s3_config_builder)
 }
+
+/// Returns true if server-side encryption headers should be disabled
+pub fn is_sse_disabled() -> bool {
+    *AWS_S3_DISABLE_SSE
+}
+
+/// Returns true if checksum headers should be disabled
+pub fn are_checksums_disabled() -> bool {
+    *AWS_S3_DISABLE_CHECKSUMS
+}