HDDS-10633. Support Content-MD5 header for checking object integrity when uploading object

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java

@@ -111,6 +111,27 @@ public long getClientID() {
     return clientID;
   }
 
+  @VisibleForTesting
+  public KeyDataStreamOutput() {
+    super(null);
+    this.config = new OzoneClientConfig();
+    OmKeyInfo info = new OmKeyInfo.Builder().setKeyName("test").build();
+    blockDataStreamOutputEntryPool =
+        new BlockDataStreamOutputEntryPool(
+            config,
+            null,
+            null,
+            null, 0,
+            false, info,
+            false,
+            null,
+            0L);
+
+    this.writeOffset = 0;
+    this.clientID = 0L;
+    this.atomicKeyCreation = false;
+  }
+
   @SuppressWarnings({"parameternumber", "squid:S00107"})
   public KeyDataStreamOutput(
       OzoneClientConfig config,

hadoop-ozone/dist/src/main/smoketest/s3/MultipartUpload.robot

@@ -405,3 +405,55 @@ Check Bucket Ownership Verification
     ${uploadID}=    Execute and checkrc                                            echo '${uploadID}' | jq -r '.UploadId'    0
 
     Execute AWSS3APICli with bucket owner check                                    abort-multipart-upload --bucket ${BUCKET} --key ${PREFIX}/mpu/aborttest --upload-id ${uploadID}  ${correct_owner}
+
+Test Multipart Upload Part with Content-MD5 header
+    # Create test file for multipart upload
+                                Execute                    echo "Multipart Upload Part Test" > /tmp/mpu_md5testfile
+    ${md5_hash} =               Execute                    md5sum /tmp/mpu_md5testfile | awk '{print $1}'
+    ${md5_base64} =             Execute                    openssl dgst -md5 -binary /tmp/mpu_md5testfile | base64
+
+    # Initialize multipart upload
+    ${uploadID} =               Initiate MPU               ${BUCKET}    ${PREFIX}/mpu/md5test/key1
+
+    # Upload part with correct Content-MD5 header
+    ${result} =                 Execute AWSS3APICli        upload-part --bucket ${BUCKET} --key ${PREFIX}/mpu/md5test/key1 --part-number 1 --body /tmp/mpu_md5testfile --upload-id ${uploadID} --content-md5 ${md5_base64}
+                                Should contain             ${result}    ETag
+    ${eTag1} =                  Execute and checkrc        echo '${result}' | jq -r '.ETag' | tr -d '"'    0
+                                Should Be Equal            ${eTag1}     ${md5_hash}
+
+    # List parts to verify upload
+    ${result} =                 Execute AWSS3APICli        list-parts --bucket ${BUCKET} --key ${PREFIX}/mpu/md5test/key1 --upload-id ${uploadID}
+    ${part_etag} =              Execute and checkrc        echo '${result}' | jq -r '.Parts[0].ETag' | tr -d '"'    0
+                                Should Be Equal            ${part_etag}    ${md5_hash}
+
+    # Complete the multipart upload
+    ${parts} =                  Set Variable               {ETag=\"${eTag1}\",PartNumber=1}
+                                Complete MPU               ${BUCKET}    ${PREFIX}/mpu/md5test/key1    ${uploadID}    ${parts}
+
+    # Verify the final object
+    ${result} =                 Execute AWSS3APICli        get-object --bucket ${BUCKET} --key ${PREFIX}/mpu/md5test/key1 /tmp/mpu_md5testfile.result
+    Compare files               /tmp/mpu_md5testfile       /tmp/mpu_md5testfile.result
+
+Test Multipart Upload Part with wrong Content-MD5 header
+    # Create test file for multipart upload
+                                Execute                    echo "Multipart Upload Part Wrong MD5 Test" > /tmp/mpu_md5testfile2
+
+    # Calculate wrong MD5 (from different content)
+    ${wrong_md5_hash} =         Execute                    echo -n "wrong content for mpu" | md5sum | awk '{print $1}'
+    ${wrong_md5_base64} =       Execute                    echo -n "wrong content for mpu" | openssl dgst -md5 -binary | base64
+
+    # Initialize multipart upload
+    ${uploadID} =               Initiate MPU               ${BUCKET}    ${PREFIX}/mpu/md5test/key2
+
+    # Upload part with wrong Content-MD5 header - should fail
+    ${result} =                 Execute AWSS3APICli and checkrc    upload-part --bucket ${BUCKET} --key ${PREFIX}/mpu/md5test/key2 --part-number 1 --body /tmp/mpu_md5testfile2 --upload-id ${uploadID} --content-md5 ${wrong_md5_base64}    255
+                                Should contain             ${result}    BadDigest
+
+    # Verify no parts were uploaded
+    ${result} =                 Execute AWSS3APICli        list-parts --bucket ${BUCKET} --key ${PREFIX}/mpu/md5test/key2 --upload-id ${uploadID}
+    ${parts_count} =            Execute and checkrc        echo '${result}' | jq -r '.Parts | length'    0
+                                Should Be Equal            ${parts_count}    0
+
+    # Abort the multipart upload (cleanup)
+                                Abort MPU                  ${BUCKET}    ${PREFIX}/mpu/md5test/key2    ${uploadID}
+

hadoop-ozone/dist/src/main/smoketest/s3/objectputget.robot

@@ -316,3 +316,26 @@ Check Bucket Ownership Verification
     # create directory
     Execute                                             touch /tmp/emptyfile
     Execute AWSS3APICli with bucket owner check         put-object --bucket ${BUCKET} --key ${PREFIX}/bucketownercondition/key=value/dir/ --body /tmp/emptyfile  ${correct_owner}
+
+Put object with Content-MD5 header
+                                Execute                    echo "bar" > /tmp/md5testfile
+    ${md5_hash} =               Execute                    md5sum /tmp/md5testfile | awk '{print $1}'
+    ${md5_base64} =             Execute                    openssl dgst -md5 -binary /tmp/md5testfile | base64
+    ${result} =                 Execute AWSS3APICli        put-object --bucket ${BUCKET} --key ${PREFIX}/putobject/md5test/key1 --body /tmp/md5testfile --content-md5 ${md5_base64}
+                                Should contain             ${result}    ETag
+    ${etag} =                   Execute and checkrc        echo '${result}' | jq -r '.ETag' | tr -d '"'    0
+                                Should Be Equal            ${etag}      ${md5_hash}
+    ${result} =                 Execute AWSS3APICli        get-object --bucket ${BUCKET} --key ${PREFIX}/putobject/md5test/key1 /tmp/md5testfile.result
+    ${etag} =                   Execute and checkrc        echo '${result}' | jq -r '.ETag' | tr -d '"'    0
+                                Should Be Equal            ${etag}      ${md5_hash}
+    Compare files               /tmp/md5testfile           /tmp/md5testfile.result
+
+Put object with wrong Content-MD5 header
+                                Execute                    echo "bar" > /tmp/md5testfile2
+    ${wrong_md5_base64} =       Execute                    echo -n "wrong" | openssl dgst -md5 -binary | base64
+    ${result} =                 Execute AWSS3APICli and checkrc    put-object --bucket ${BUCKET} --key ${PREFIX}/putobject/md5test/key2 --body /tmp/md5testfile2 --content-md5 ${wrong_md5_base64}    255
+                                Should contain             ${result}    BadDigest
+    # Verify the object was not uploaded
+    ${result} =                 Execute AWSS3APICli and checkrc    head-object --bucket ${BUCKET} --key ${PREFIX}/putobject/md5test/key2    255
+                                Should contain             ${result}    404
+

hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java

@@ -90,6 +90,7 @@
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
@@ -376,6 +377,163 @@ public void testPutObject() {
     assertEquals("37b51d194a7513e45b56f6524f2d51f2", putObjectResult.getETag());
   }
 
+  @Test
+  public void testPutObjectWithMD5Header() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    final String content = "bar";
+    s3Client.createBucket(bucketName);
+
+    byte[] contentBytes = content.getBytes(StandardCharsets.UTF_8);
+    byte[] md5Bytes = calculateDigest(new ByteArrayInputStream(contentBytes), 0, contentBytes.length);
+    String md5Base64 = Base64.getEncoder().encodeToString(md5Bytes);
+
+    InputStream is = new ByteArrayInputStream(contentBytes);
+    ObjectMetadata objectMetadata = new ObjectMetadata();
+    objectMetadata.setContentMD5(md5Base64);
+    objectMetadata.setContentLength(contentBytes.length);
+
+    PutObjectResult putObjectResult = s3Client.putObject(bucketName, keyName, is, objectMetadata);
+    assertEquals("37b51d194a7513e45b56f6524f2d51f2", putObjectResult.getETag());
+
+    S3Object object = s3Client.getObject(bucketName, keyName);
+    assertEquals(content.length(), object.getObjectMetadata().getContentLength());
+    assertEquals("37b51d194a7513e45b56f6524f2d51f2", object.getObjectMetadata().getETag());
+  }
+
+  @Test
+  public void testPutObjectWithWrongMD5Header() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    final String content = "bar";
+    s3Client.createBucket(bucketName);
+
+    // Use wrong content to calculate MD5
+    byte[] wrongContentBytes = "wrong".getBytes(StandardCharsets.UTF_8);
+    byte[] wrongMd5Bytes = calculateDigest(new ByteArrayInputStream(wrongContentBytes), 0, wrongContentBytes.length);
+    String wrongMd5Base64 = Base64.getEncoder().encodeToString(wrongMd5Bytes);
+
+    byte[] contentBytes = content.getBytes(StandardCharsets.UTF_8);
+    InputStream is = new ByteArrayInputStream(contentBytes);
+    ObjectMetadata objectMetadata = new ObjectMetadata();
+    objectMetadata.setContentMD5(wrongMd5Base64);
+    objectMetadata.setContentLength(contentBytes.length);
+
+    AmazonServiceException ase = assertThrows(AmazonServiceException.class,
+        () -> s3Client.putObject(bucketName, keyName, is, objectMetadata));
+
+    assertEquals(ErrorType.Client, ase.getErrorType());
+    assertEquals(400, ase.getStatusCode());
+    assertEquals("BadDigest", ase.getErrorCode());
+
+    // Verify the object was not uploaded
+    assertFalse(s3Client.doesObjectExist(bucketName, keyName));
+  }
+
+  @Test
+  public void testMultipartUploadWithMD5Header() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    s3Client.createBucket(bucketName);
+
+    // Initiate multipart upload
+    InitiateMultipartUploadRequest initRequest = new InitiateMultipartUploadRequest(bucketName, keyName);
+    InitiateMultipartUploadResult initResponse = s3Client.initiateMultipartUpload(initRequest);
+    String uploadId = initResponse.getUploadId();
+
+    // Prepare part data
+    String part1Content = "part1data";
+    byte[] part1Bytes = part1Content.getBytes(StandardCharsets.UTF_8);
+    byte[] part1Md5Bytes = calculateDigest(new ByteArrayInputStream(part1Bytes), 0, part1Bytes.length);
+    String part1Md5Base64 = Base64.getEncoder().encodeToString(part1Md5Bytes);
+
+    // Upload part 1 with MD5
+    InputStream part1InputStream = new ByteArrayInputStream(part1Bytes);
+    ObjectMetadata metadata = new ObjectMetadata();
+    metadata.setContentMD5(part1Md5Base64);
+    metadata.setContentLength(part1Bytes.length);
+
+    UploadPartRequest uploadRequest = new UploadPartRequest()
+        .withBucketName(bucketName)
+        .withKey(keyName)
+        .withUploadId(uploadId)
+        .withPartNumber(1)
+        .withInputStream(part1InputStream)
+        .withPartSize(part1Bytes.length)
+        .withObjectMetadata(metadata);
+
+    UploadPartResult uploadResult = s3Client.uploadPart(uploadRequest);
+
+    // Verify ETag
+    String expectedETag = DatatypeConverter.printHexBinary(part1Md5Bytes).toLowerCase();
+    assertEquals(expectedETag, uploadResult.getPartETag().getETag());
+
+    // Complete multipart upload
+    List<PartETag> partETags = new ArrayList<>();
+    partETags.add(uploadResult.getPartETag());
+
+    CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(
+        bucketName, keyName, uploadId, partETags);
+    s3Client.completeMultipartUpload(completeRequest);
+
+    // Verify object was uploaded
+    S3Object object = s3Client.getObject(bucketName, keyName);
+    try (S3ObjectInputStream s3is = object.getObjectContent();
+         ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
+      IOUtils.copy(s3is, bos);
+      assertEquals(part1Content, bos.toString("UTF-8"));
+    }
+  }
+
+  @Test
+  public void testMultipartUploadPartWithWrongMD5Header() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    s3Client.createBucket(bucketName);
+
+    // Initiate multipart upload
+    InitiateMultipartUploadRequest initRequest = new InitiateMultipartUploadRequest(bucketName, keyName);
+    InitiateMultipartUploadResult initResponse = s3Client.initiateMultipartUpload(initRequest);
+    String uploadId = initResponse.getUploadId();
+
+    // Prepare part data with wrong MD5
+    String partContent = "partdata";
+    byte[] partBytes = partContent.getBytes(StandardCharsets.UTF_8);
+
+    byte[] wrongMd5Bytes = calculateDigest(
+        new ByteArrayInputStream("wrongdata".getBytes(StandardCharsets.UTF_8)), 0, 9);
+    String wrongMd5Base64 = Base64.getEncoder().encodeToString(wrongMd5Bytes);
+
+    // Upload part with wrong MD5 should fail
+    InputStream partInputStream = new ByteArrayInputStream(partBytes);
+    ObjectMetadata metadata = new ObjectMetadata();
+    metadata.setContentMD5(wrongMd5Base64);
+    metadata.setContentLength(partBytes.length);
+
+    UploadPartRequest uploadRequest = new UploadPartRequest()
+        .withBucketName(bucketName)
+        .withKey(keyName)
+        .withUploadId(uploadId)
+        .withPartNumber(1)
+        .withInputStream(partInputStream)
+        .withPartSize(partBytes.length)
+        .withObjectMetadata(metadata);
+
+    AmazonServiceException ase = assertThrows(AmazonServiceException.class,
+        () -> s3Client.uploadPart(uploadRequest));
+
+    assertEquals(ErrorType.Client, ase.getErrorType());
+    assertEquals(400, ase.getStatusCode());
+    assertEquals("BadDigest", ase.getErrorCode());
+
+    // Abort the multipart upload
+    AbortMultipartUploadRequest abortRequest = new AbortMultipartUploadRequest(bucketName, keyName, uploadId);
+    s3Client.abortMultipartUpload(abortRequest);
+
+    // Verify object was not created
+    assertFalse(s3Client.doesObjectExist(bucketName, keyName));
+  }
+
   @Test
   public void testPutDoubleSlashPrefixObject() throws IOException {
     final String bucketName = getBucketName();