Skip to content

Commit e0c75c3

Browse files
joviegasjoviegas
committed
Fix Trailer based Http Checksum for Async Request body created from File
1 parent a9d5cce commit e0c75c3

File tree

8 files changed

+179
-17
lines changed

8 files changed

+179
-17
lines changed

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

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -189,7 +189,8 @@ public void onNext(ByteBuffer byteBuffer) {
189189
ByteBuffer allocatedBuffer = getFinalChecksumAppendedChunk(byteBuffer);
190190
wrapped.onNext(allocatedBuffer);
191191
} else {
192-
wrapped.onNext(byteBuffer);
192+
ByteBuffer allocatedBuffer = appendChunkSizeAndFinalByte(byteBuffer);
193+
wrapped.onNext(allocatedBuffer);
193194
}
194195
} catch (SdkException sdkException) {
195196
this.subscription.cancel();
@@ -215,6 +216,14 @@ private ByteBuffer getFinalChecksumAppendedChunk(ByteBuffer byteBuffer) {
215216
return checksumAppendedBuffer;
216217
}
217218

219+
private ByteBuffer appendChunkSizeAndFinalByte(ByteBuffer byteBuffer) {
220+
ByteBuffer contentChunk = createChunk(byteBuffer, false);
221+
ByteBuffer checksumAppendedBuffer = ByteBuffer.allocate(contentChunk.remaining());
222+
checksumAppendedBuffer.put(contentChunk);
223+
checksumAppendedBuffer.flip();
224+
return checksumAppendedBuffer;
225+
}
226+
218227
@Override
219228
public void onError(Throwable t) {
220229
wrapped.onError(t);

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

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,12 +47,13 @@
4747
*/
4848
@SdkInternalApi
4949
public final class FileAsyncRequestBody implements AsyncRequestBody {
50-
private static final Logger log = Logger.loggerFor(FileAsyncRequestBody.class);
5150

5251
/**
5352
* Default size (in bytes) of ByteBuffer chunks read from the file and delivered to the subscriber.
5453
*/
55-
private static final int DEFAULT_CHUNK_SIZE = 16 * 1024;
54+
public static final int DEFAULT_CHUNK_SIZE = 16 * 1024;
55+
56+
private static final Logger log = Logger.loggerFor(FileAsyncRequestBody.class);
5657

5758
/**
5859
* File to read.

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

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
2626
import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
2727
import software.amazon.awssdk.core.internal.async.ChecksumCalculatingAsyncRequestBody;
28+
import software.amazon.awssdk.core.internal.async.FileAsyncRequestBody;
29+
import software.amazon.awssdk.core.internal.io.AwsUnsignedChunkedEncodingInputStream;
2830
import software.amazon.awssdk.core.internal.util.ChunkContentUtils;
2931
import software.amazon.awssdk.core.internal.util.HttpChecksumUtils;
3032
import software.amazon.awssdk.http.Header;
@@ -97,7 +99,11 @@ public SdkHttpRequest modifyHttpRequest(Context.ModifyHttpRequest context,
9799
private static SdkHttpRequest updateHeadersForTrailerChecksum(Context.ModifyHttpRequest context, ChecksumSpecs checksum,
98100
long checksumContentLength, long originalContentLength) {
99101

100-
long chunkLength = ChunkContentUtils.calculateChunkLength(originalContentLength);
102+
long chunkLength = isFileAsyncRequestBody(context)
103+
? AwsUnsignedChunkedEncodingInputStream
104+
.calculateStreamContentLength(originalContentLength, FileAsyncRequestBody.DEFAULT_CHUNK_SIZE)
105+
: ChunkContentUtils.calculateChunkLength(originalContentLength);
106+
101107
return context.httpRequest().copy(r ->
102108
r.putHeader(HttpChecksumConstant.HEADER_FOR_TRAILER_REFERENCE, checksum.headerName())
103109
.putHeader("Content-encoding", HttpChecksumConstant.AWS_CHUNKED_HEADER)
@@ -106,4 +112,9 @@ private static SdkHttpRequest updateHeadersForTrailerChecksum(Context.ModifyHttp
106112
.putHeader(Header.CONTENT_LENGTH,
107113
Long.toString(chunkLength + checksumContentLength)));
108114
}
115+
116+
private static boolean isFileAsyncRequestBody(Context.ModifyHttpRequest context) {
117+
return context.asyncRequestBody().isPresent() && context.asyncRequestBody().get() instanceof FileAsyncRequestBody;
118+
}
119+
109120
}

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import software.amazon.awssdk.core.interceptor.Context;
3030
import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
3131
import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
32+
import software.amazon.awssdk.core.internal.io.AwsChunkedEncodingInputStream;
3233
import software.amazon.awssdk.core.internal.io.AwsUnsignedChunkedEncodingInputStream;
3334
import software.amazon.awssdk.core.internal.util.HttpChecksumResolver;
3435
import software.amazon.awssdk.core.internal.util.HttpChecksumUtils;
@@ -73,7 +74,7 @@ public Optional<RequestBody> modifyHttpContent(Context.ModifyHttpRequest context
7374
RequestBody.fromContentProvider(
7475
streamProvider,
7576
AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength(
76-
requestBody.optionalContentLength().orElse(0L))
77+
requestBody.optionalContentLength().orElse(0L), AwsChunkedEncodingInputStream.DEFAULT_CHUNK_SIZE)
7778
+ checksumContentLength,
7879
requestBody.contentType()));
7980
}
@@ -112,7 +113,8 @@ public SdkHttpRequest modifyHttpRequest(Context.ModifyHttpRequest context, Execu
112113
.putHeader("x-amz-decoded-content-length", Long.toString(originalContentLength))
113114
.putHeader(CONTENT_LENGTH,
114115
Long.toString(AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength(
115-
originalContentLength) + checksumContentLength)));
116+
originalContentLength, AwsChunkedEncodingInputStream.DEFAULT_CHUNK_SIZE)
117+
+ checksumContentLength)));
116118
}
117119

118120

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@
3939
@SdkInternalApi
4040
public abstract class AwsChunkedEncodingInputStream extends SdkInputStream {
4141

42-
protected static final int DEFAULT_CHUNK_SIZE = 128 * 1024;
42+
public static final int DEFAULT_CHUNK_SIZE = 128 * 1024;
4343
protected static final int SKIP_BUFFER_SIZE = 256 * 1024;
4444
protected static final String CRLF = "\r\n";
4545
protected static final byte[] FINAL_CHUNK = new byte[0];

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

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -68,17 +68,19 @@ private static long calculateChunkLength(long originalContentLength) {
6868
+ CRLF.length();
6969
}
7070

71-
public static long calculateStreamContentLength(long originalLength) {
72-
if (originalLength < 0) {
73-
throw new IllegalArgumentException("Non negative content length expected.");
71+
public static long calculateStreamContentLength(long originalLength, long defaultChunkSize) {
72+
if (originalLength < 0 || defaultChunkSize == 0) {
73+
throw new IllegalArgumentException(originalLength + ", " + defaultChunkSize + "Args <= 0 not expected");
7474
}
7575

76-
long maxSizeChunks = originalLength / DEFAULT_CHUNK_SIZE;
77-
long remainingBytes = originalLength % DEFAULT_CHUNK_SIZE;
76+
long maxSizeChunks = originalLength / defaultChunkSize;
77+
long remainingBytes = originalLength % defaultChunkSize;
7878

79-
return maxSizeChunks * calculateChunkLength(DEFAULT_CHUNK_SIZE)
80-
+ (remainingBytes > 0 ? calculateChunkLength(remainingBytes) : 0)
81-
+ calculateChunkLength(0);
79+
long allChunks = maxSizeChunks * calculateChunkLength(defaultChunkSize);
80+
long remainingInChunk = remainingBytes > 0 ? calculateChunkLength(remainingBytes) : 0;
81+
long lastByteSize = "0".length() + CRLF.length();
82+
83+
return allChunks + remainingInChunk + lastByteSize;
8284
}
8385

8486
@Override

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,8 @@ public void readAwsUnsignedChunkedEncodingInputStream() throws IOException {
5555
public void lengthsOfCalculateByChecksumCalculatingInputStream(){
5656

5757
String initialString = "Hello world";
58-
long calculateChunkLength = AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength(initialString.length());
58+
long calculateChunkLength = AwsUnsignedChunkedEncodingInputStream.calculateStreamContentLength(initialString.length(),
59+
AwsChunkedEncodingInputStream.DEFAULT_CHUNK_SIZE);
5960
long checksumContentLength = AwsUnsignedChunkedEncodingInputStream.calculateChecksumContentLength(
6061
SHA256_ALGORITHM, SHA256_HEADER_NAME);
6162
assertThat(calculateChunkLength).isEqualTo(21);

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

Lines changed: 137 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,14 @@
2020
import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName;
2121

2222
import java.io.BufferedReader;
23+
import java.io.File;
24+
import java.io.IOException;
2325
import java.io.InputStreamReader;
26+
import java.io.RandomAccessFile;
2427
import java.net.URI;
2528
import java.nio.charset.StandardCharsets;
29+
import java.nio.file.Files;
30+
import java.nio.file.Paths;
2631
import java.util.stream.Collectors;
2732
import org.junit.jupiter.api.AfterAll;
2833
import org.junit.jupiter.api.AfterEach;
@@ -303,7 +308,7 @@ public void asyncValidUnsignedTrailerChecksumCalculatedBySdkClient() throws Inte
303308
}
304309

305310
@Test
306-
public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient() throws InterruptedException {
311+
public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient_withSmallRequestBody() throws InterruptedException {
307312
s3Async.putObject(PutObjectRequest.builder()
308313
.bucket(BUCKET)
309314
.key(KEY)
@@ -320,6 +325,23 @@ public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient() throws
320325
assertThat(response).isEqualTo("Hello world");
321326
}
322327

328+
@Test
329+
public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient_withHugeRequestBody() throws InterruptedException {
330+
s3Async.putObject(PutObjectRequest.builder()
331+
.bucket(BUCKET)
332+
.key(KEY)
333+
.checksumAlgorithm(ChecksumAlgorithm.CRC32)
334+
.build(), AsyncRequestBody.fromString(createDataSize(HUGE_MSG_SIZE))).join();
335+
assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
336+
assertThat(interceptor.requestChecksumInHeader()).isNull();
337+
338+
String response = s3Async.getObject(GetObjectRequest.builder().bucket(BUCKET)
339+
.key(KEY).checksumMode(ChecksumMode.ENABLED)
340+
.build(), AsyncResponseTransformer.toBytes()).join().asUtf8String();
341+
assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
342+
assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
343+
assertThat(response).isEqualTo(createDataSize(HUGE_MSG_SIZE));
344+
}
323345

324346

325347
@Disabled("Http Async Signing is not supported for S3")
@@ -416,4 +438,118 @@ public void syncUnsignedPayloadMultiPartForHugeMessage() throws InterruptedExcep
416438
assertThat(interceptor.responseValidation()).isNull();
417439
assertThat(text).isEqualTo(createDataSize(HUGE_MSG_SIZE));
418440
}
441+
442+
443+
@Test
444+
public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient_withSmallFileRequestBody() throws InterruptedException, IOException {
445+
File randomFileOfFixedLength = getRandomFileOfFixedLength(10);
446+
s3Async.putObject(PutObjectRequest.builder()
447+
.bucket(BUCKET)
448+
.key(KEY)
449+
.checksumAlgorithm(ChecksumAlgorithm.CRC32)
450+
.build(), AsyncRequestBody.fromFile(randomFileOfFixedLength.toPath())).join();
451+
assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
452+
assertThat(interceptor.requestChecksumInHeader()).isNull();
453+
454+
String response = s3Async.getObject(GetObjectRequest.builder().bucket(BUCKET)
455+
.key(KEY).checksumMode(ChecksumMode.ENABLED)
456+
.build(), AsyncResponseTransformer.toBytes()).join().asUtf8String();
457+
assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
458+
assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
459+
460+
byte[] bytes = Files.readAllBytes(randomFileOfFixedLength.toPath());
461+
assertThat(response).isEqualTo(new String (bytes));
462+
463+
464+
}
465+
466+
@Test
467+
public void asyncHttpsValidUnsignedTrailerChecksumCalculatedBySdkClient_withHugeFileRequestBody()
468+
throws IOException {
469+
470+
File randomFileOfFixedLength = getRandomFileOfFixedLength(17);
471+
s3Async.putObject(PutObjectRequest.builder()
472+
.bucket(BUCKET)
473+
.key(KEY)
474+
.checksumAlgorithm(ChecksumAlgorithm.CRC32)
475+
.build(), AsyncRequestBody.fromFile(randomFileOfFixedLength.toPath())).join();
476+
assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
477+
assertThat(interceptor.requestChecksumInHeader()).isNull();
478+
479+
String response = s3Async.getObject(GetObjectRequest.builder().bucket(BUCKET)
480+
.key(KEY).checksumMode(ChecksumMode.ENABLED)
481+
.build(), AsyncResponseTransformer.toBytes()).join().asUtf8String();
482+
assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
483+
assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
484+
485+
byte[] bytes = Files.readAllBytes(randomFileOfFixedLength.toPath());
486+
assertThat(response).isEqualTo(new String (bytes));
487+
488+
}
489+
490+
@Test
491+
public void syncValidUnsignedTrailerChecksumCalculatedBySdkClient_withSmallFileRequestBody() throws InterruptedException,
492+
IOException {
493+
494+
File randomFileOfFixedLength = getRandomFileOfFixedLength(10);
495+
496+
s3Https.putObject(PutObjectRequest.builder()
497+
.bucket(BUCKET)
498+
.key(KEY)
499+
.checksumAlgorithm(ChecksumAlgorithm.CRC32)
500+
.build(), RequestBody.fromFile(randomFileOfFixedLength.toPath()));
501+
502+
assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
503+
assertThat(interceptor.requestChecksumInHeader()).isNull();
504+
505+
ResponseInputStream<GetObjectResponse> s3HttpsObject =
506+
s3Https.getObject(GetObjectRequest.builder().bucket(BUCKET).key(KEY).checksumMode(ChecksumMode.ENABLED).build());
507+
String text = new BufferedReader(
508+
new InputStreamReader(s3HttpsObject, StandardCharsets.UTF_8))
509+
.lines()
510+
.collect(Collectors.joining("\n"));
511+
assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
512+
assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
513+
byte[] bytes = Files.readAllBytes(randomFileOfFixedLength.toPath());
514+
assertThat(text).isEqualTo(new String(bytes));
515+
}
516+
517+
518+
@Test
519+
public void syncValidUnsignedTrailerChecksumCalculatedBySdkClient_withHugeFileRequestBody() throws InterruptedException,
520+
IOException {
521+
522+
File randomFileOfFixedLength = getRandomFileOfFixedLength(34);
523+
524+
s3Https.putObject(PutObjectRequest.builder()
525+
.bucket(BUCKET)
526+
.key(KEY)
527+
.checksumAlgorithm(ChecksumAlgorithm.CRC32)
528+
.build(), RequestBody.fromFile(randomFileOfFixedLength.toPath()));
529+
530+
assertThat(interceptor.requestChecksumInTrailer()).isEqualTo("x-amz-checksum-crc32");
531+
assertThat(interceptor.requestChecksumInHeader()).isNull();
532+
533+
ResponseInputStream<GetObjectResponse> s3HttpsObject =
534+
s3Https.getObject(GetObjectRequest.builder().bucket(BUCKET).key(KEY).checksumMode(ChecksumMode.ENABLED).build());
535+
String text = new BufferedReader(
536+
new InputStreamReader(s3HttpsObject, StandardCharsets.UTF_8))
537+
.lines()
538+
.collect(Collectors.joining("\n"));
539+
assertThat(interceptor.validationAlgorithm()).isEqualTo(Algorithm.CRC32);
540+
assertThat(interceptor.responseValidation()).isEqualTo(ChecksumValidation.VALIDATED);
541+
byte[] bytes = Files.readAllBytes(randomFileOfFixedLength.toPath());
542+
assertThat(text).isEqualTo(new String(bytes));
543+
}
544+
545+
private File getRandomFileOfFixedLength(int sizeInKb) throws IOException {
546+
int objectSize = sizeInKb * 1024 ;
547+
final File tempFile = File.createTempFile("s3-object-file-", ".tmp");
548+
try (RandomAccessFile f = new RandomAccessFile(tempFile, "rw")) {
549+
f.setLength(objectSize );
550+
}
551+
tempFile.deleteOnExit();
552+
return tempFile;
553+
}
554+
419555
}

0 commit comments

Comments
 (0)