fix(s3): allow Object Lock on versioned buckets and reject invalid checksums (#2024)

crates/rio/src/checksum.rs

@@ -546,7 +546,26 @@ pub fn get_content_checksum(headers: &HeaderMap) -> Result<Option<Checksum>, std
         return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid checksum"));
     }
 
+    if checksum_type == ChecksumType::INVALID {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            crate::errors::ChecksumMismatch {
+                want: "valid checksum header".to_string(),
+                got: "invalid or duplicate checksum headers".to_string(),
+            },
+        ));
+    }
+
     let checksum = Checksum::new_with_type(checksum_type, &value);
+    if checksum.is_none() && !value.is_empty() {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            crate::errors::ChecksumMismatch {
+                want: value,
+                got: "invalid checksum value".to_string(),
+            },
+        ));
+    }
     Ok(checksum)
 }
 

rustfs/src/app/multipart_usecase.rs

@@ -609,7 +609,7 @@ impl DefaultMultipartUsecase {
             let mut hrd = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?;
 
             if let Err(err) = hrd.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
-                return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
+                return Err(ApiError::from(err).into());
             }
 
             let compress_reader = CompressReader::new(hrd, CompressionAlgorithm::default());
@@ -622,7 +622,7 @@ impl DefaultMultipartUsecase {
         let mut reader = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?;
 
         if let Err(err) = reader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), size < 0) {
-            return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
+            return Err(ApiError::from(err).into());
         }
 
         let has_ssec = sse_customer_algorithm.is_some();

rustfs/src/app/object_usecase.rs

@@ -472,7 +472,7 @@ impl DefaultObjectUsecase {
             let mut hrd = HashReader::new(reader, size as i64, size as i64, md5hex, sha256hex, false).map_err(ApiError::from)?;
 
             if let Err(err) = hrd.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
-                return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
+                return Err(ApiError::from(err).into());
             }
 
             opts.want_checksum = hrd.checksum();
@@ -491,7 +491,7 @@ impl DefaultObjectUsecase {
 
         if size >= 0 {
             if let Err(err) = reader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
-                return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
+                return Err(ApiError::from(err).into());
             }
 
             opts.want_checksum = reader.checksum();
@@ -788,16 +788,21 @@ impl DefaultObjectUsecase {
             Ok(_) => {}
             Err(err) => {
                 if err == StorageError::ConfigNotFound {
+                    // AWS S3 allows enabling Object Lock on existing buckets if versioning
+                    // is already enabled. Reject only when versioning is not enabled.
+                    if !BucketVersioningSys::enabled(&bucket).await {
+                        return Err(S3Error::with_message(
+                            S3ErrorCode::InvalidBucketState,
+                            "Object Lock configuration cannot be enabled on existing buckets".to_string(),
+                        ));
+                    }
+                } else {
+                    warn!("get_object_lock_config err {:?}", err);
                     return Err(S3Error::with_message(
-                        S3ErrorCode::InvalidBucketState,
-                        "Object Lock configuration cannot be enabled on existing buckets".to_string(),
+                        S3ErrorCode::InternalError,
+                        "Failed to get bucket ObjectLockConfiguration".to_string(),
                     ));
                 }
-                warn!("get_object_lock_config err {:?}", err);
-                return Err(S3Error::with_message(
-                    S3ErrorCode::InternalError,
-                    "Failed to get bucket ObjectLockConfiguration".to_string(),
-                ));
             }
         };
 
@@ -3524,7 +3529,7 @@ impl DefaultObjectUsecase {
         let mut hreader = HashReader::new(reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?;
 
         if let Err(err) = hreader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
-            return Err(ApiError::from(StorageError::other(format!("add_checksum error={err:?}"))).into());
+            return Err(ApiError::from(err).into());
         }
 
         let decoder = CompressionFormat::from_extension(&ext).get_decoder(hreader).map_err(|e| {

scripts/s3-tests/implemented_tests.txt

@@ -29,8 +29,10 @@
 # - Raw requests: Authenticated and anonymous
 # - Conditional writes: If-Match/If-None-Match for PUT/Copy
 # - Atomic read/write: Concurrent read/write consistency
+# - Object Lock: Enable on versioned existing buckets
+# - Checksum: SHA256 and CRC64NVME validation on PutObject
 #
-# Total: 376 tests
+# Total: 379 tests
 
 test_basic_key_count
 test_bucket_create_naming_bad_short_one
@@ -456,3 +458,10 @@ test_atomic_dual_write_1mb
 test_atomic_dual_write_4mb
 test_atomic_dual_write_8mb
 test_atomic_multipart_upload_write
+
+# Object Lock tests
+test_object_lock_put_obj_lock_enable_after_create
+
+# Checksum validation tests
+test_object_checksum_sha256
+test_object_checksum_crc64nvme

scripts/s3-tests/unimplemented_tests.txt

@@ -6,16 +6,12 @@
 #
 # Unimplemented features:
 # - Bucket Logging: Access logging
-# - Object Lock: Enable after create
-# - Checksum: Full checksum validation
+# - Checksum: POST Object form upload checksum
 # - Bucket Ownership Controls
 
 # Failed tests
 test_bucket_create_delete_bucket_ownership
 test_bucket_logging_owner
-test_object_checksum_crc64nvme
-test_object_checksum_sha256
-test_object_lock_put_obj_lock_enable_after_create
 test_post_object_upload_checksum
 test_put_bucket_logging
 test_put_bucket_logging_errors