fix(s3): improve S3 API compatibility for versioning, SSE, and policy (#2013)

rustfs/src/app/bucket_usecase.rs

@@ -168,17 +168,28 @@ impl DefaultBucketUsecase {
 
         counter!("rustfs_create_bucket_total").increment(1);
 
-        store
+        let make_result = store
             .make_bucket(
                 &bucket,
                 &MakeBucketOptions {
-                    force_create: false, // TODO: force support
+                    force_create: false,
                     lock_enabled: object_lock_enabled_for_bucket.is_some_and(|v| v),
                     ..Default::default()
                 },
             )
-            .await
-            .map_err(ApiError::from)?;
+            .await;
+
+        match make_result {
+            Ok(()) => {}
+            Err(StorageError::BucketExists(_)) => {
+                // Per S3 spec: CreateBucket for a bucket you already own returns 200 OK
+                let output = CreateBucketOutput::default();
+                let result = Ok(S3Response::new(output));
+                let _ = helper.complete(&result);
+                return result;
+            }
+            Err(e) => return Err(ApiError::from(e).into()),
+        }
 
         let owner = default_owner();
         let mut stored_acl = stored_acl_from_grant_headers(

rustfs/src/app/multipart_usecase.rs

@@ -625,18 +625,27 @@ impl DefaultMultipartUsecase {
             return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
         }
 
-        let server_side_encryption = fi
-            .user_defined
-            .get("x-amz-server-side-encryption")
-            .map(|s| {
-                ServerSideEncryption::from_str(s)
-                    .map_err(|e| ApiError::from(StorageError::other(format!("Invalid server-side encryption: {e}"))))
-            })
-            .transpose()?;
-        let ssekms_key_id = fi
-            .user_defined
-            .get("x-amz-server-side-encryption-aws-kms-key-id")
-            .map(|s| s.to_string());
+        let has_ssec = sse_customer_algorithm.is_some();
+        // When SSE-C headers are present, skip managed-encryption metadata to avoid
+        // false conflict: the bucket default SSE config stored in multipart metadata
+        // should not block a legitimate SSE-C upload part.
+        let (server_side_encryption, ssekms_key_id) = if has_ssec {
+            (None, None)
+        } else {
+            let sse = fi
+                .user_defined
+                .get("x-amz-server-side-encryption")
+                .map(|s| {
+                    ServerSideEncryption::from_str(s)
+                        .map_err(|e| ApiError::from(StorageError::other(format!("Invalid server-side encryption: {e}"))))
+                })
+                .transpose()?;
+            let key_id = fi
+                .user_defined
+                .get("x-amz-server-side-encryption-aws-kms-key-id")
+                .map(|s| s.to_string());
+            (sse, key_id)
+        };
         let part_key = fi.user_defined.get("x-rustfs-encryption-key").cloned();
         let part_nonce = fi.user_defined.get("x-rustfs-encryption-iv").cloned();
         let encryption_request = EncryptionRequest {

rustfs/src/app/object_usecase.rs

@@ -550,14 +550,21 @@ impl DefaultObjectUsecase {
         // Fast in-memory update for immediate quota consistency
         rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await;
 
-        let put_version = obj_info.version_id.map(|v| v.to_string());
+        let raw_version = obj_info.version_id.map(|v| v.to_string());
 
         helper = helper.object(obj_info.clone());
-        if let Some(version_id) = &put_version {
+        if let Some(version_id) = &raw_version {
             helper = helper.version_id(version_id.clone());
         }
 
-        Self::spawn_cache_invalidation(bucket.clone(), key.clone(), put_version.clone());
+        Self::spawn_cache_invalidation(bucket.clone(), key.clone(), raw_version.clone());
+
+        // Per S3 spec: only return VersionId when versioning is Enabled (not Suspended or default)
+        let put_version = if BucketVersioningSys::prefix_enabled(&bucket, &key).await {
+            raw_version
+        } else {
+            None
+        };
 
         let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
 

rustfs/src/storage/access.rs

@@ -628,11 +628,14 @@ impl S3Access for FS {
     }
 
     /// Checks whether the CreateMultipartUpload request has accesses to the resources.
-    ///
-    /// This method returns `Ok(())` by default.
-    async fn create_multipart_upload(&self, _req: &mut S3Request<CreateMultipartUploadInput>) -> S3Result<()> {
+    async fn create_multipart_upload(&self, req: &mut S3Request<CreateMultipartUploadInput>) -> S3Result<()> {
         license_check().map_err(|er| s3_error!(AccessDenied, "{:?}", er.to_string()))?;
-        Ok(())
+
+        let req_info = ext_req_info_mut(&mut req.extensions)?;
+        req_info.bucket = Some(req.input.bucket.clone());
+        req_info.object = Some(req.input.key.clone());
+
+        authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await
     }
 
     /// Checks whether the DeleteBucket request has accesses to the resources.

rustfs/src/storage/sse.rs

@@ -204,7 +204,19 @@ async fn prepare_sse_configuration(
         }))
     } else if let Err(e) = bucket_sse_config_result {
         match e {
-            Error::ConfigNotFound => Ok(None),
+            Error::ConfigNotFound => {
+                // The bucket has no SSE config. If the user explicitly requested
+                // aws:kms, we must honor that — return the explicit SSE header so
+                // downstream logic can try (and fail if KMS is unavailable).
+                if let Some(sse) = server_side_encryption {
+                    Ok(Some(SseConfiguration {
+                        effective_sse: sse,
+                        effective_kms_key_id: ssekms_key_id,
+                    }))
+                } else {
+                    Ok(None)
+                }
+            }
             _ => Err(ApiError::from(e)),
         }
     } else {