Fixes after merge

Author: dagnir

.changes/next-release/feature-AWSSDKForJavav2-3b0176a.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Add support for payload signing of async streaming requests signed with SigV4 using default `AwsV4HttpSigner` (using `AwsV4HttpSigner.create()`). Note, requests using the `http` URI scheme will not be signed regardless of the value of `AwsV4FamilyHttpSigner.PAYLOAD_SIGNING_ENABLED` to remain consistent with existing behavior. This may change in a future release."
+}

.changes/next-release/feature-AWSSDKforJavav2-e386e28.json

@@ -69,7 +69,7 @@ public SignedRequest sign(SignRequest<? extends AwsCredentialsIdentity> request)
 
     @Override
     public CompletableFuture<AsyncSignedRequest> signAsync(AsyncSignRequest<? extends AwsCredentialsIdentity> request) {
-        Checksummer checksummer = checksummer(request, null, checksumStore(request));
+        Checksummer checksummer = asyncChecksummer(request, checksumStore(request));
         V4Properties v4Properties = v4Properties(request);
         V4RequestSigner v4RequestSigner = v4RequestSigner(request, v4Properties);
         V4PayloadSigner payloadSigner = v4PayloadAsyncSigner(request, v4Properties);
@@ -81,7 +81,7 @@ private static V4Properties v4Properties(BaseSignRequest<?, ? extends AwsCredent
         Clock signingClock = request.requireProperty(SIGNING_CLOCK, Clock.systemUTC());
         Instant signingInstant = signingClock.instant();
         AwsCredentialsIdentity credentials = sanitizeCredentials(request.identity());
-        String regionName = request.requireProperty(AwsV4HttpSigner.REGION_NAME);
+        String regionName = request.requireProperty(REGION_NAME);
         String serviceSigningName = request.requireProperty(SERVICE_SIGNING_NAME);
         CredentialScope credentialScope = new CredentialScope(regionName, serviceSigningName, signingInstant);
         boolean doubleUrlEncode = request.requireProperty(DOUBLE_URL_ENCODE, true);
@@ -129,9 +129,29 @@ private static V4RequestSigner v4RequestSigner(
         return requestSigner.apply(v4Properties);
     }
 
+    // TODO: remove this once we consolidate the behavior for plaintext HTTP signing for sync and async
+    private static Checksummer asyncChecksummer(BaseSignRequest<?, ? extends AwsCredentialsIdentity> request,
+                                                PayloadChecksumStore checksumStore) {
+        boolean shouldTreatAsUnsigned = asyncShouldTreatAsUnsigned(request);
+
+        // set the override to false if it should be treated as unsigned, otherwise, null should be passed so that the normal
+        // check for payload signing is done.
+        Boolean overridePayloadSigning = shouldTreatAsUnsigned ? false : null;
+
+        return checksummer(request, overridePayloadSigning, checksumStore);
+    }
+
+    // TODO: remove this once we consolidate the behavior for plaintext HTTP signing for sync and async
+    private static boolean asyncShouldTreatAsUnsigned(BaseSignRequest<?, ? extends AwsCredentialsIdentity> request) {
+        boolean isHttp = !"https".equals(request.request().protocol());
+        boolean isPayloadSigning = isPayloadSigning(request);
+        boolean isChunkEncoding = request.requireProperty(CHUNK_ENCODING_ENABLED, false);
+
+        return isHttp && isPayloadSigning && isChunkEncoding;
+    }
+
     private static V4PayloadSigner v4PayloadSigner(
-        SignRequest<? extends AwsCredentialsIdentity> request,
-        V4Properties properties) {
+        BaseSignRequest<?, ? extends AwsCredentialsIdentity> request, V4Properties properties) {
 
         boolean isPayloadSigning = isPayloadSigning(request);
         boolean isEventStreaming = isEventStreaming(request.request());
@@ -162,6 +182,7 @@ private static V4PayloadSigner v4PayloadSigner(
         return V4PayloadSigner.create();
     }
 
+    // TODO: remove this once we consolidate the behavior for plaintext HTTP signing for sync and async
     private static V4PayloadSigner v4PayloadAsyncSigner(
         AsyncSignRequest<? extends AwsCredentialsIdentity> request,
         V4Properties properties) {
@@ -183,10 +204,19 @@ private static V4PayloadSigner v4PayloadAsyncSigner(
             throw new UnsupportedOperationException("Unsigned payload is not supported with event-streaming.");
         }
 
-        if (useChunkEncoding(isPayloadSigning, isChunkEncoding, isTrailing || isFlexible)) {
+        // Note: this check is done after we check if the request is eventstreaming, during which we just use the same logic
+        // as sync to determine if the body should be signed. If it's not eventstreaming, then async needs to treat this
+        // request differently to maintain current behavior re: plain HTTP requests.
+        boolean nonEvenstreamingPayloadSigning = isPayloadSigning;
+        if (asyncShouldTreatAsUnsigned(request)) {
+            nonEvenstreamingPayloadSigning = false;
+        }
+
+        if (useChunkEncoding(nonEvenstreamingPayloadSigning, isChunkEncoding, isTrailing || isFlexible)) {
             return AwsChunkedV4PayloadSigner.builder()
                                             .credentialScope(properties.getCredentialScope())
                                             .chunkSize(DEFAULT_CHUNK_SIZE_IN_BYTES)
+                                            .checksumStore(checksumStore(request))
                                             .checksumAlgorithm(request.property(CHECKSUM_ALGORITHM))
                                             .build();
         }

core/http-auth-aws/src/main/java/software/amazon/awssdk/http/auth/aws/internal/signer/DefaultAwsV4HttpSigner.java

@@ -4,6 +4,7 @@
 import static software.amazon.awssdk.http.auth.aws.signer.AwsV4HttpSigner.SERVICE_SIGNING_NAME;
 import static software.amazon.awssdk.http.auth.spi.signer.HttpSigner.SIGNING_CLOCK;
 
+import io.reactivex.Flowable;
 import java.io.ByteArrayInputStream;
 import java.net.URI;
 import java.nio.ByteBuffer;
@@ -13,6 +14,7 @@
 import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.util.function.Consumer;
+import org.reactivestreams.Publisher;
 import software.amazon.awssdk.http.SdkHttpMethod;
 import software.amazon.awssdk.http.SdkHttpRequest;
 import software.amazon.awssdk.http.auth.spi.signer.AsyncSignRequest;
@@ -58,10 +60,8 @@ public static <T extends AwsCredentialsIdentity> AsyncSignRequest<T> generateBas
         Consumer<? super SdkHttpRequest.Builder> requestOverrides,
         Consumer<? super AsyncSignRequest.Builder<T>> signRequestOverrides
     ) {
-        SimplePublisher<ByteBuffer> publisher = new SimplePublisher<>();
 
-        publisher.send(ByteBuffer.wrap(testPayload()));
-        publisher.complete();
+        Publisher<ByteBuffer> publisher = Flowable.just(ByteBuffer.wrap(testPayload()));
 
         return AsyncSignRequest.builder(credentials)
                                .request(SdkHttpRequest.builder()

core/http-auth-aws/src/test/java/software/amazon/awssdk/http/auth/aws/TestUtils.java

@@ -30,17 +30,21 @@
 import static software.amazon.awssdk.http.auth.aws.signer.AwsV4HttpSigner.PAYLOAD_SIGNING_ENABLED;
 import static software.amazon.awssdk.http.auth.spi.signer.SdkInternalHttpSignerProperty.CHECKSUM_STORE;
 
+import io.reactivex.Flowable;
 import java.io.IOException;
 import java.net.URI;
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
-import java.util.Optional;
+import java.util.List;
+import java.util.stream.Collectors;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
+import org.reactivestreams.Publisher;
 import software.amazon.awssdk.checksums.SdkChecksum;
 import software.amazon.awssdk.checksums.spi.ChecksumAlgorithm;
 import software.amazon.awssdk.http.Header;
@@ -709,7 +713,9 @@ void sign_WithPayloadSigningTrueAndChunkEncodingTrueAndHttp_SignsPayload() {
     }
 
     @Test
-    void signAsync_WithPayloadSigningTrueAndChunkEncodingTrueAndHttp_IgnoresPayloadSigning() {
+    @Disabled("Fallback to signing is disabled to match pre-SRA behavior")
+    // TODO: Enable this test once we figure out what the expected behavior is post SRA. See JAVA-8078
+    void signAsync_WithPayloadSigningTrueAndChunkEncodingTrueAndHttp_RespectsPayloadSigning() {
         AsyncSignRequest<? extends AwsCredentialsIdentity> request = generateBasicAsyncRequest(
             AwsCredentialsIdentity.create("access", "secret"),
             httpRequest -> httpRequest.uri(URI.create("http://demo.us-east-1.amazonaws.com")),
@@ -727,6 +733,8 @@ void signAsync_WithPayloadSigningTrueAndChunkEncodingTrueAndHttp_IgnoresPayloadS
     }
 
     @Test
+    @Disabled("Fallback to signing is disabled to match pre-SRA behavior")
+    // TODO: Enable this test once we figure out what the expected behavior is post SRA. See JAVA-8078
     void sign_WithPayloadSigningFalseAndChunkEncodingTrueAndHttp_SignsPayload() {
         SignRequest<? extends AwsCredentialsIdentity> request = generateBasicRequest(
             AwsCredentialsIdentity.create("access", "secret"),
@@ -745,7 +753,9 @@ void sign_WithPayloadSigningFalseAndChunkEncodingTrueAndHttp_SignsPayload() {
     }
 
     @Test
-    void signAsync_WithPayloadSigningFalseAndChunkEncodingTrueAndHttp_DoesNotFallBackToPayloadSigning() {
+    @Disabled("Fallback to signing is disabled to match pre-SRA behavior")
+    // TODO: Enable this test once we figure out what the expected behavior is post SRA. See JAVA-8078
+    void signAsync_WithPayloadSigningFalseAndChunkEncodingTrueAndHttp_FallsBackToPayloadSigning() {
         AsyncSignRequest<? extends AwsCredentialsIdentity> request = generateBasicAsyncRequest(
             AwsCredentialsIdentity.create("access", "secret"),
             httpRequest -> httpRequest.uri(URI.create("http://demo.us-east-1.amazonaws.com")),
@@ -783,7 +793,9 @@ void sign_WithPayloadSigningFalseAndChunkEncodingTrueAndFlexibleChecksumAndHttp_
     }
 
     @Test
-    void signAsync_WithPayloadSigningFalseAndChunkEncodingTrueAndFlexibleChecksumAndHttp_DoesNotFallBackToPayloadSigning() {
+    @Disabled("Fallback to signing is disabled to match pre-SRA behavior")
+    // TODO: Enable this test once we figure out what the expected behavior is post SRA. See JAVA-8078
+    void signAsync_WithPayloadSigningFalseAndChunkEncodingTrueAndFlexibleChecksumAndHttp_FallsBackToPayloadSigning() {
         AsyncSignRequest<? extends AwsCredentialsIdentity> request = generateBasicAsyncRequest(
             AwsCredentialsIdentity.create("access", "secret"),
             httpRequest -> httpRequest.uri(URI.create("http://demo.us-east-1.amazonaws.com")),
@@ -900,9 +912,61 @@ void sign_withPayloadSigningTrue_chunkEncodingFalse_withChecksum_cacheEmpty_stor
         assertThat(cache.getChecksumValue(CRC32)).isEqualTo(crc32Value);
     }
 
+    @Test
+    void signAsync_WithPayloadSigningFalse_chunkEncodingTrue_cacheEmpty_storesComputedChecksum() throws IOException {
+        PayloadChecksumStore cache = PayloadChecksumStore.create();
+
+        AsyncSignRequest<? extends AwsCredentialsIdentity> request = generateBasicAsyncRequest(
+            AwsCredentialsIdentity.create("access", "secret"),
+            httpRequest -> httpRequest.uri(URI.create("http://demo.us-east-1.amazonaws.com")),
+            signRequest -> signRequest
+                .putProperty(PAYLOAD_SIGNING_ENABLED, false)
+                .putProperty(CHUNK_ENCODING_ENABLED, true)
+                .putProperty(CHECKSUM_ALGORITHM, CRC32)
+                .putProperty(CHECKSUM_STORE, cache)
+        );
+
+        AsyncSignedRequest signedRequest = signer.signAsync(request).join();
+
+        getAllItems(signedRequest.payload().get());
+        assertThat(cache.getChecksumValue(CRC32)).isEqualTo(computeChecksum(CRC32, testPayload()));
+    }
+
+    @Test
+    void signAsync_WithPayloadSigningFalse_chunkEncodingTrue_cacheContainsChecksum_usesCachedValue() throws IOException {
+        PayloadChecksumStore cache = PayloadChecksumStore.create();
+
+        byte[] checksumValue = "my-checksum".getBytes(StandardCharsets.UTF_8);
+        cache.putChecksumValue(CRC32, checksumValue);
+
+        AsyncSignRequest<? extends AwsCredentialsIdentity> request = generateBasicAsyncRequest(
+            AwsCredentialsIdentity.create("access", "secret"),
+            httpRequest -> httpRequest.uri(URI.create("http://demo.us-east-1.amazonaws.com")),
+            signRequest -> signRequest
+                .putProperty(PAYLOAD_SIGNING_ENABLED, false)
+                .putProperty(CHUNK_ENCODING_ENABLED, true)
+                .putProperty(CHECKSUM_ALGORITHM, CRC32)
+                .putProperty(CHECKSUM_STORE, cache)
+        );
+
+        AsyncSignedRequest signedRequest = signer.signAsync(request).join();
+
+        List<ByteBuffer> content = getAllItems(signedRequest.payload().get());
+        String contentAsString = content.stream().map(DefaultAwsV4HttpSignerTest::bufferAsString).collect(Collectors.joining());
+        assertThat(contentAsString).contains("x-amz-checksum-crc32:" + BinaryUtils.toBase64(checksumValue) + "\r\n");
+    }
+
     private static byte[] computeChecksum(ChecksumAlgorithm algorithm, byte[] data) {
         SdkChecksum checksum = SdkChecksum.forAlgorithm(algorithm);
         checksum.update(data, 0, data.length);
         return checksum.getChecksumBytes();
     }
+
+    private List<ByteBuffer> getAllItems(Publisher<ByteBuffer> publisher) {
+        return Flowable.fromPublisher(publisher).toList().blockingGet();
+    }
+
+    private static String bufferAsString(ByteBuffer buffer) {
+        return StandardCharsets.UTF_8.decode(buffer.duplicate()).toString();
+    }
 }