Fix Trailer based Http Checksum for Async Request body with variable chunk size

.changes/next-release/bugfix-AWSSDKforJavav2-5ecdce1.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Fixed issue where request used to fail while calculating Trailer based checksum for Async File Request body."
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/ChecksumCalculatingAsyncRequestBody.java

@@ -189,7 +189,8 @@ public void onNext(ByteBuffer byteBuffer) {
                     ByteBuffer allocatedBuffer = getFinalChecksumAppendedChunk(byteBuffer);
                     wrapped.onNext(allocatedBuffer);
                 } else {
-                    wrapped.onNext(byteBuffer);
+                    ByteBuffer allocatedBuffer = appendChunkSizeAndFinalByte(byteBuffer);
+                    wrapped.onNext(allocatedBuffer);
                 }
             } catch (SdkException sdkException) {
                 this.subscription.cancel();
@@ -215,6 +216,14 @@ private ByteBuffer getFinalChecksumAppendedChunk(ByteBuffer byteBuffer) {
             return checksumAppendedBuffer;
         }
 
+        private ByteBuffer appendChunkSizeAndFinalByte(ByteBuffer byteBuffer) {
+            ByteBuffer contentChunk = createChunk(byteBuffer, false);
+            ByteBuffer checksumAppendedBuffer = ByteBuffer.allocate(contentChunk.remaining());
+            checksumAppendedBuffer.put(contentChunk);
+            checksumAppendedBuffer.flip();
+            return checksumAppendedBuffer;
+        }
+
         @Override
         public void onError(Throwable t) {
             wrapped.onError(t);

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBody.java

@@ -47,12 +47,13 @@
  */
 @SdkInternalApi
 public final class FileAsyncRequestBody implements AsyncRequestBody {
-    private static final Logger log = Logger.loggerFor(FileAsyncRequestBody.class);
 
     /**
      * Default size (in bytes) of ByteBuffer chunks read from the file and delivered to the subscriber.
      */
-    private static final int DEFAULT_CHUNK_SIZE = 16 * 1024;
+    public static final int DEFAULT_CHUNK_SIZE = 16 * 1024;
+
+    private static final Logger log = Logger.loggerFor(FileAsyncRequestBody.class);
 
     /**
      * File to read.

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/interceptor/AsyncRequestBodyHttpChecksumTrailerInterceptor.java

@@ -25,6 +25,8 @@
 import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
 import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
 import software.amazon.awssdk.core.internal.async.ChecksumCalculatingAsyncRequestBody;
+import software.amazon.awssdk.core.internal.async.FileAsyncRequestBody;
+import software.amazon.awssdk.core.internal.io.AwsUnsignedChunkedEncodingInputStream;
 import software.amazon.awssdk.core.internal.util.ChunkContentUtils;
 import software.amazon.awssdk.core.internal.util.HttpChecksumUtils;
 import software.amazon.awssdk.http.Header;
@@ -97,7 +99,11 @@ public SdkHttpRequest modifyHttpRequest(Context.ModifyHttpRequest context,
     private static SdkHttpRequest updateHeadersForTrailerChecksum(Context.ModifyHttpRequest context, ChecksumSpecs checksum,
                                                                   long checksumContentLength, long originalContentLength) {
 
-        long chunkLength = ChunkContentUtils.calculateChunkLength(originalContentLength);
+        long chunkLength = isFileAsyncRequestBody(context)
+                           ? AwsUnsignedChunkedEncodingInputStream
+                               .calculateStreamContentLength(originalContentLength, FileAsyncRequestBody.DEFAULT_CHUNK_SIZE)
+                           : ChunkContentUtils.calculateChunkLength(originalContentLength);
+
         return context.httpRequest().copy(r ->
                 r.putHeader(HttpChecksumConstant.HEADER_FOR_TRAILER_REFERENCE, checksum.headerName())
                         .putHeader("Content-encoding", HttpChecksumConstant.AWS_CHUNKED_HEADER)
@@ -106,4 +112,9 @@ private static SdkHttpRequest updateHeadersForTrailerChecksum(Context.ModifyHttp
                         .putHeader(Header.CONTENT_LENGTH,
                                    Long.toString(chunkLength + checksumContentLength)));
     }
+
+    private static boolean isFileAsyncRequestBody(Context.ModifyHttpRequest context) {
+        return context.asyncRequestBody().isPresent() && context.asyncRequestBody().get() instanceof FileAsyncRequestBody;
+    }
+
 }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/interceptor/SyncHttpChecksumInTrailerInterceptor.java

@@ -29,6 +29,7 @@
 import software.amazon.awssdk.core.interceptor.Context;
 import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
 import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
+import software.amazon.awssdk.core.internal.io.AwsChunkedEncodingInputStream;
 import software.amazon.awssdk.core.internal.io.AwsUnsignedChunkedEncodingInputStream;
 import software.amazon.awssdk.core.internal.util.HttpChecksumResolver;
 import software.amazon.awssdk.core.internal.util.HttpChecksumUtils;
@@ -73,7 +74,7 @@ public Optional<RequestBody> modifyHttpContent(Context.ModifyHttpRequest context
             RequestBody.fromContentProvider(
                 streamProvider,
                 AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength(
-                    requestBody.optionalContentLength().orElse(0L))
+                    requestBody.optionalContentLength().orElse(0L), AwsChunkedEncodingInputStream.DEFAULT_CHUNK_SIZE)
                 + checksumContentLength,
                 requestBody.contentType()));
     }
@@ -112,7 +113,8 @@ public SdkHttpRequest modifyHttpRequest(Context.ModifyHttpRequest context, Execu
                   .putHeader("x-amz-decoded-content-length", Long.toString(originalContentLength))
                   .putHeader(CONTENT_LENGTH,
                              Long.toString(AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength(
-                                 originalContentLength) + checksumContentLength)));
+                                 originalContentLength, AwsChunkedEncodingInputStream.DEFAULT_CHUNK_SIZE)
+                                           + checksumContentLength)));
     }
 
 

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/io/AwsChunkedEncodingInputStream.java

@@ -39,7 +39,7 @@
 @SdkInternalApi
 public abstract class AwsChunkedEncodingInputStream extends SdkInputStream {
 
-    protected static final int DEFAULT_CHUNK_SIZE = 128 * 1024;
+    public static final int DEFAULT_CHUNK_SIZE = 128 * 1024;
     protected static final int SKIP_BUFFER_SIZE = 256 * 1024;
     protected static final String CRLF = "\r\n";
     protected static final byte[] FINAL_CHUNK = new byte[0];

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/io/AwsUnsignedChunkedEncodingInputStream.java

@@ -68,17 +68,19 @@ private static long calculateChunkLength(long originalContentLength) {
                 + CRLF.length();
     }
 
-    public static long calculateStreamContentLength(long originalLength) {
-        if (originalLength < 0) {
-            throw new IllegalArgumentException("Non negative content length expected.");
+    public static long calculateStreamContentLength(long originalLength, long defaultChunkSize) {
+        if (originalLength < 0 || defaultChunkSize == 0) {
+            throw new IllegalArgumentException(originalLength + ", " + defaultChunkSize + "Args <= 0 not expected");
         }
 
-        long maxSizeChunks = originalLength / DEFAULT_CHUNK_SIZE;
-        long remainingBytes = originalLength % DEFAULT_CHUNK_SIZE;
+        long maxSizeChunks = originalLength / defaultChunkSize;
+        long remainingBytes = originalLength % defaultChunkSize;
 
-        return maxSizeChunks * calculateChunkLength(DEFAULT_CHUNK_SIZE)
-                + (remainingBytes > 0 ? calculateChunkLength(remainingBytes) : 0)
-                + calculateChunkLength(0);
+        long allChunks = maxSizeChunks * calculateChunkLength(defaultChunkSize);
+        long remainingInChunk = remainingBytes > 0 ? calculateChunkLength(remainingBytes) : 0;
+        long lastByteSize = "0".length() + CRLF.length();
+
+        return allChunks +  remainingInChunk + lastByteSize;
     }
 
     @Override

core/sdk-core/src/test/java/software/amazon/awssdk/core/checksum/AwsChunkedEncodingInputStreamTest.java

@@ -55,10 +55,11 @@ public void readAwsUnsignedChunkedEncodingInputStream() throws IOException {
     public void lengthsOfCalculateByChecksumCalculatingInputStream(){
 
         String initialString = "Hello world";
-        long calculateChunkLength = AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength(initialString.length());
+        long calculateChunkLength = AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength(initialString.length(),
+                                                                                                       AwsChunkedEncodingInputStream.DEFAULT_CHUNK_SIZE);
         long checksumContentLength = AwsUnsignedChunkedEncodingInputStream.calculateChecksumContentLength(
                 SHA256_ALGORITHM, SHA256_HEADER_NAME);
-        assertThat(calculateChunkLength).isEqualTo(21);
+        assertThat(calculateChunkLength).isEqualTo(19);
         assertThat(checksumContentLength).isEqualTo(69);
     }
 

services/s3/src/it/java/software/amazon/awssdk/services/s3/checksum/HttpChecksumIntegrationTest.java

@@ -20,9 +20,14 @@
 import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
 
 import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
 import java.io.InputStreamReader;
+import java.io.RandomAccessFile;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.stream.Collectors;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
@@ -303,7 +308,7 @@ public void asyncValidUnsignedTrailerChecksumCalculatedBySdkClient() throws Inte
     }
 
     @Test
-    public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient() throws InterruptedException {
+    public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient_withSmallRequestBody() throws InterruptedException {
         s3Async.putObject(PutObjectRequest.builder()
                                           .bucket(BUCKET)
                                           .key(KEY)
@@ -320,6 +325,23 @@ public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient() throws
         assertThat(response).isEqualTo("Hello world");
     }
 
+    @Test
+    public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient_withHugeRequestBody() throws InterruptedException {
+        s3Async.putObject(PutObjectRequest.builder()
+                                          .bucket(BUCKET)
+                                          .key(KEY)
+                                          .checksumAlgorithm(ChecksumAlgorithm.CRC32)
+                                          .build(), AsyncRequestBody.fromString(createDataSize(HUGE_MSG_SIZE))).join();
+        assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
+        assertThat(interceptor.requestChecksumInHeader()).isNull();
+
+        String response = s3Async.getObject(GetObjectRequest.builder().bucket(BUCKET)
+                                                            .key(KEY).checksumMode(ChecksumMode.ENABLED)
+                                                            .build(), AsyncResponseTransformer.toBytes()).join().asUtf8String();
+        assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
+        assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
+        assertThat(response).isEqualTo(createDataSize(HUGE_MSG_SIZE));
+    }
 
 
     @Disabled("Http Async Signing is not supported for S3")
@@ -416,4 +438,118 @@ public void syncUnsignedPayloadMultiPartForHugeMessage() throws InterruptedExcep
         assertThat(interceptor.responseValidation()).isNull();
         assertThat(text).isEqualTo(createDataSize(HUGE_MSG_SIZE));
     }
+
+
+    @Test
+    public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient_withSmallFileRequestBody() throws InterruptedException, IOException {
+        File randomFileOfFixedLength = getRandomFileOfFixedLength(10);
+        s3Async.putObject(PutObjectRequest.builder()
+                                          .bucket(BUCKET)
+                                          .key(KEY)
+                                          .checksumAlgorithm(ChecksumAlgorithm.CRC32)
+                                          .build(), AsyncRequestBody.fromFile(randomFileOfFixedLength.toPath())).join();
+        assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
+        assertThat(interceptor.requestChecksumInHeader()).isNull();
+
+        String response = s3Async.getObject(GetObjectRequest.builder().bucket(BUCKET)
+                                                            .key(KEY).checksumMode(ChecksumMode.ENABLED)
+                                                            .build(), AsyncResponseTransformer.toBytes()).join().asUtf8String();
+        assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
+        assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
+
+        byte[] bytes = Files.readAllBytes(randomFileOfFixedLength.toPath());
+        assertThat(response).isEqualTo(new String (bytes));
+
+
+    }
+
+    @Test
+    public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient_withHugeFileRequestBody()
+        throws IOException {
+
+        File randomFileOfFixedLength = getRandomFileOfFixedLength(17);
+        s3Async.putObject(PutObjectRequest.builder()
+                                          .bucket(BUCKET)
+                                          .key(KEY)
+                                          .checksumAlgorithm(ChecksumAlgorithm.CRC32)
+                                          .build(), AsyncRequestBody.fromFile(randomFileOfFixedLength.toPath())).join();
+        assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
+        assertThat(interceptor.requestChecksumInHeader()).isNull();
+
+        String response = s3Async.getObject(GetObjectRequest.builder().bucket(BUCKET)
+                                                            .key(KEY).checksumMode(ChecksumMode.ENABLED)
+                                                            .build(), AsyncResponseTransformer.toBytes()).join().asUtf8String();
+        assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
+        assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
+
+        byte[] bytes = Files.readAllBytes(randomFileOfFixedLength.toPath());
+        assertThat(response).isEqualTo(new String (bytes));
+
+    }
+
+    @Test
+    public void syncValidUnsignedTrailerChecksumCalculatedBySdkClient_withSmallFileRequestBody() throws InterruptedException,
+                                                                                                        IOException {
+
+        File randomFileOfFixedLength = getRandomFileOfFixedLength(10);
+
+        s3Https.putObject(PutObjectRequest.builder()
+                                          .bucket(BUCKET)
+                                          .key(KEY)
+                                          .checksumAlgorithm(ChecksumAlgorithm.CRC32)
+                                          .build(), RequestBody.fromFile(randomFileOfFixedLength.toPath()));
+
+        assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
+        assertThat(interceptor.requestChecksumInHeader()).isNull();
+
+        ResponseInputStream<GetObjectResponse> s3HttpsObject =
+            s3Https.getObject(GetObjectRequest.builder().bucket(BUCKET).key(KEY).checksumMode(ChecksumMode.ENABLED).build());
+        String text = new BufferedReader(
+            new InputStreamReader(s3HttpsObject, StandardCharsets.UTF_8))
+            .lines()
+            .collect(Collectors.joining("\n"));
+        assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
+        assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
+        byte[] bytes = Files.readAllBytes(randomFileOfFixedLength.toPath());
+        assertThat(text).isEqualTo(new String(bytes));
+    }
+
+
+    @Test
+    public void syncValidUnsignedTrailerChecksumCalculatedBySdkClient_withHugeFileRequestBody() throws InterruptedException,
+                                                                                                        IOException {
+
+        File randomFileOfFixedLength = getRandomFileOfFixedLength(34);
+
+        s3Https.putObject(PutObjectRequest.builder()
+                                          .bucket(BUCKET)
+                                          .key(KEY)
+                                          .checksumAlgorithm(ChecksumAlgorithm.CRC32)
+                                          .build(), RequestBody.fromFile(randomFileOfFixedLength.toPath()));
+
+        assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
+        assertThat(interceptor.requestChecksumInHeader()).isNull();
+
+        ResponseInputStream<GetObjectResponse> s3HttpsObject =
+            s3Https.getObject(GetObjectRequest.builder().bucket(BUCKET).key(KEY).checksumMode(ChecksumMode.ENABLED).build());
+        String text = new BufferedReader(
+            new InputStreamReader(s3HttpsObject, StandardCharsets.UTF_8))
+            .lines()
+            .collect(Collectors.joining("\n"));
+        assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
+        assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
+        byte[] bytes = Files.readAllBytes(randomFileOfFixedLength.toPath());
+        assertThat(text).isEqualTo(new String(bytes));
+    }
+
+    private File getRandomFileOfFixedLength(int sizeInKb) throws IOException {
+        int objectSize = sizeInKb * 1024  ;
+        final File tempFile = File.createTempFile("s3-object-file-", ".tmp");
+        try (RandomAccessFile f = new RandomAccessFile(tempFile, "rw")) {
+            f.setLength(objectSize  );
+        }
+        tempFile.deleteOnExit();
+        return tempFile;
+    }
+
 }