Add support of Request-level credentials override in DefaultS3CrtAsyncClient

.changes/next-release/feature-AmazonS3-bf3403d.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "Amazon S3",
+    "contributor": "",
+    "description": "Added support of Request-level credentials override in DefaultS3CrtAsyncClient. See [#5354](https://github.com/aws/aws-sdk-java-v2/issues/5354)."
+}

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClient.java

@@ -423,24 +423,37 @@
                                                          existingHttpAttributes.toBuilder() :
                                                          SdkHttpExecutionAttributes.builder();
 
-            SdkHttpExecutionAttributes attributes =
-                builder.put(OPERATION_NAME,
-                            executionAttributes.getAttribute(SdkExecutionAttribute.OPERATION_NAME))
-                       .put(HTTP_CHECKSUM, executionAttributes.getAttribute(SdkInternalExecutionAttribute.HTTP_CHECKSUM))
-                       .put(SIGNING_REGION, executionAttributes.getAttribute(AwsSignerExecutionAttribute.SIGNING_REGION))
-                       .put(S3InternalSdkHttpExecutionAttribute.OBJECT_FILE_PATH,
-                            executionAttributes.getAttribute(OBJECT_FILE_PATH))
-                       .put(USE_S3_EXPRESS_AUTH, S3ExpressUtils.useS3ExpressAuthScheme(executionAttributes))
-                       .put(SIGNING_NAME, executionAttributes.getAttribute(SERVICE_SIGNING_NAME))
-                       .put(REQUEST_CHECKSUM_CALCULATION,
-                            executionAttributes.getAttribute(SdkInternalExecutionAttribute.REQUEST_CHECKSUM_CALCULATION))
-                       .put(RESPONSE_CHECKSUM_VALIDATION,
-                            executionAttributes.getAttribute(SdkInternalExecutionAttribute.RESPONSE_CHECKSUM_VALIDATION))
-                       .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_PATH,
-                            executionAttributes.getAttribute(RESPONSE_FILE_PATH))
-                       .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION,
-                            executionAttributes.getAttribute(RESPONSE_FILE_OPTION))
-                       .build();
+            builder.put(OPERATION_NAME,
+                        executionAttributes.getAttribute(SdkExecutionAttribute.OPERATION_NAME))
+                   .put(HTTP_CHECKSUM, executionAttributes.getAttribute(SdkInternalExecutionAttribute.HTTP_CHECKSUM))
+                   .put(SIGNING_REGION, executionAttributes.getAttribute(AwsSignerExecutionAttribute.SIGNING_REGION))
+                   .put(S3InternalSdkHttpExecutionAttribute.OBJECT_FILE_PATH,
+                        executionAttributes.getAttribute(OBJECT_FILE_PATH))
+                   .put(USE_S3_EXPRESS_AUTH, S3ExpressUtils.useS3ExpressAuthScheme(executionAttributes))
+                   .put(SIGNING_NAME, executionAttributes.getAttribute(SERVICE_SIGNING_NAME))
+                   .put(REQUEST_CHECKSUM_CALCULATION,
+                        executionAttributes.getAttribute(SdkInternalExecutionAttribute.REQUEST_CHECKSUM_CALCULATION))
+                   .put(RESPONSE_CHECKSUM_VALIDATION,
+                        executionAttributes.getAttribute(SdkInternalExecutionAttribute.RESPONSE_CHECKSUM_VALIDATION))
+                   .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_PATH,
+                        executionAttributes.getAttribute(RESPONSE_FILE_PATH))
+                   .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION,
+                        executionAttributes.getAttribute(RESPONSE_FILE_OPTION));
+
+            SdkRequest request = context.request();
+            if (request instanceof AwsRequest) {
+                ((AwsRequest) request).overrideConfiguration().ifPresent(config -> {
+                    AwsRequestOverrideConfiguration awsConfig = (AwsRequestOverrideConfiguration) config;
+                    awsConfig.credentialsIdentityProvider().ifPresent(credentialsProvider -> {
+                        CrtCredentialsProviderAdapter adapter =
+                            new CrtCredentialsProviderAdapter(credentialsProvider);
+                        builder.put(S3InternalSdkHttpExecutionAttribute.CRT_CREDENTIALS_PROVIDER_ADAPTER,
+                                    adapter);
+                    });
+                });
+            }
+
+            SdkHttpExecutionAttributes attributes = builder.build();
 
             // We rely on CRT to perform checksum validation, disable SDK flexible checksum implementation
             executionAttributes.putAttribute(SdkInternalExecutionAttribute.HTTP_CHECKSUM, null);
@@ -468,11 +481,6 @@
                     throw new UnsupportedOperationException("Request-level signer override is not supported");
                 }
 
-                // TODO: support request-level credential override
-                if (overrideConfiguration.credentialsIdentityProvider().isPresent()) {
-                    throw new UnsupportedOperationException("Request-level credentials override is not supported");
-                }
-
                 if (!CollectionUtils.isNullOrEmpty(overrideConfiguration.metricPublishers())) {
                     throw new UnsupportedOperationException("Request-level Metric Publishers override is not supported");
                 }

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java

@@ -43,6 +43,7 @@
 import software.amazon.awssdk.core.checksums.RequestChecksumCalculation;
 import software.amazon.awssdk.core.checksums.ResponseChecksumValidation;
 import software.amazon.awssdk.core.interceptor.trait.HttpChecksum;
+import software.amazon.awssdk.crt.auth.credentials.CredentialsProvider;
 import software.amazon.awssdk.crt.auth.signing.AwsSigningConfig;
 import software.amazon.awssdk.crt.http.HttpHeader;
 import software.amazon.awssdk.crt.http.HttpProxyEnvironmentVariableSetting;
@@ -186,6 +187,9 @@
             requestOptions = requestOptions.withResponseFileOption(responseFileOption);
         }
 
+        CrtCredentialsProviderAdapter requestCredentialsAdapter =
+            httpExecutionAttributes.getAttribute(S3InternalSdkHttpExecutionAttribute.CRT_CREDENTIALS_PROVIDER_ADAPTER);
+
         try {
             S3MetaRequestWrapper requestWrapper = new S3MetaRequestWrapper(crtS3Client.makeMetaRequest(requestOptions));
             s3MetaRequestFuture.complete(requestWrapper);
@@ -196,16 +200,30 @@
             if (observable != null) {
                 observable.subscribe(requestWrapper);
             }
+        } catch (Throwable t) {
+            if (requestCredentialsAdapter != null) {
+                requestCredentialsAdapter.close();
+            }
+            throw t;
         } finally {
             signingConfig.close();
         }
 
+        if (requestCredentialsAdapter != null) {
+            executeFuture.whenComplete((result, error) -> requestCredentialsAdapter.close());
+        }
+
         return executeFuture;
     }
 
     private AwsSigningConfig awsSigningConfig(Region signingRegion, SdkHttpExecutionAttributes httpExecutionAttributes) {
+        CrtCredentialsProviderAdapter requestAdapter =
+            httpExecutionAttributes.getAttribute(S3InternalSdkHttpExecutionAttribute.CRT_CREDENTIALS_PROVIDER_ADAPTER);
+        CredentialsProvider effectiveCredentials =
+            requestAdapter != null ? requestAdapter.crtCredentials() : s3ClientOptions.getCredentialsProvider();
+
         AwsSigningConfig defaultS3SigningConfig =
-            AwsSigningConfig.getDefaultS3SigningConfig(s3ClientOptions.getRegion(), s3ClientOptions.getCredentialsProvider());
+            AwsSigningConfig.getDefaultS3SigningConfig(s3ClientOptions.getRegion(), effectiveCredentials);
 
         // Override the region only if the signing region has changed from the previously configured region.
         if (signingRegion != null && !s3ClientOptions.getRegion().equals(signingRegion.id())) {

services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3InternalSdkHttpExecutionAttribute.java

@@ -64,6 +64,9 @@ public final class S3InternalSdkHttpExecutionAttribute<T> extends SdkHttpExecuti
     public static final S3InternalSdkHttpExecutionAttribute<S3MetaRequestOptions.ResponseFileOption> RESPONSE_FILE_OPTION =
         new S3InternalSdkHttpExecutionAttribute<>(S3MetaRequestOptions.ResponseFileOption.class);
 
+    public static final S3InternalSdkHttpExecutionAttribute<CrtCredentialsProviderAdapter> CRT_CREDENTIALS_PROVIDER_ADAPTER =
+        new S3InternalSdkHttpExecutionAttribute<>(CrtCredentialsProviderAdapter.class);
+
     private S3InternalSdkHttpExecutionAttribute(Class<T> valueClass) {
         super(valueClass);
     }

services/s3/src/test/java/software/amazon/awssdk/services/s3/crt/S3CrtRequestLevelCredentialsWireMockTest.java

@@ -0,0 +1,142 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.services.s3.crt;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.containing;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.head;
+import static com.github.tomakehurst.wiremock.client.WireMock.put;
+import static com.github.tomakehurst.wiremock.client.WireMock.putRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.verify;
+
+import com.github.tomakehurst.wiremock.client.WireMock;
+import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo;
+import com.github.tomakehurst.wiremock.junit5.WireMockTest;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.async.AsyncRequestBody;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.crt.Log;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+
+/**
+ * WireMock tests verifying that request-level credential overrides are used for signing
+ * with the S3 CRT client. Verifies the Authorization header contains the expected access key.
+ */
+@WireMockTest
+@Timeout(10)
+public class S3CrtRequestLevelCredentialsWireMockTest {
+
+    private static final String BUCKET = "my-bucket";
+    private static final String KEY = "my-key";
+    private static final String PATH = String.format("/%s/%s", BUCKET, KEY);
+    private static final byte[] CONTENT = "hello".getBytes(StandardCharsets.UTF_8);
+
+    private static final StaticCredentialsProvider CLIENT_CREDENTIALS =
+        StaticCredentialsProvider.create(AwsBasicCredentials.create("clientAccessKey", "clientSecretKey"));
+
+    private static final StaticCredentialsProvider REQUEST_CREDENTIALS =
+        StaticCredentialsProvider.create(AwsBasicCredentials.create("requestAccessKey", "requestSecretKey"));
+
+    private S3AsyncClient s3;
+
+    @BeforeAll
+    public static void setUpBeforeAll() {
+        System.setProperty("aws.crt.debugnative", "true");
+        Log.initLoggingToStdout(Log.LogLevel.Warn);
+    }
+
+    @BeforeEach
+    public void setup(WireMockRuntimeInfo wiremock) {
+        stubFor(head(urlPathEqualTo(PATH))
+                    .willReturn(WireMock.aResponse().withStatus(200)
+                                        .withHeader("ETag", "etag")
+                                        .withHeader("Content-Length",
+                                                    Integer.toString(CONTENT.length))));
+        stubFor(get(urlPathEqualTo(PATH))
+                    .willReturn(WireMock.aResponse().withStatus(200)
+                                        .withHeader("Content-Type", "text/plain")
+                                        .withBody(CONTENT)));
+        stubFor(put(urlPathEqualTo(PATH))
+                    .willReturn(WireMock.aResponse().withStatus(200)
+                                        .withHeader("ETag", "etag")));
+
+        s3 = S3AsyncClient.crtBuilder()
+                          .endpointOverride(URI.create("http://localhost:" + wiremock.getHttpPort()))
+                          .credentialsProvider(CLIENT_CREDENTIALS)
+                          .forcePathStyle(true)
+                          .region(Region.US_EAST_1)
+                          .build();
+    }
+
+    @AfterEach
+    public void tearDown() {
+        s3.close();
+    }
+
+    @Test
+    void getObject_withRequestLevelCredentials_shouldSignWithOverrideCredentials() {
+        s3.getObject(
+            b -> b.bucket(BUCKET).key(KEY)
+                  .overrideConfiguration(o -> o.credentialsProvider(REQUEST_CREDENTIALS)),
+            AsyncResponseTransformer.toBytes()).join();
+
+        verify(getRequestedFor(urlPathEqualTo(PATH))
+                   .withHeader("Authorization", containing("Credential=requestAccessKey/")));
+    }
+
+    @Test
+    void getObject_withoutRequestLevelCredentials_shouldSignWithClientCredentials() {
+        s3.getObject(
+            b -> b.bucket(BUCKET).key(KEY),
+            AsyncResponseTransformer.toBytes()).join();
+
+        verify(getRequestedFor(urlPathEqualTo(PATH))
+                   .withHeader("Authorization", containing("Credential=clientAccessKey/")));
+    }
+
+    @Test
+    void putObject_withRequestLevelCredentials_shouldSignWithOverrideCredentials() {
+        s3.putObject(
+            b -> b.bucket(BUCKET).key(KEY)
+                  .overrideConfiguration(o -> o.credentialsProvider(REQUEST_CREDENTIALS)),
+            AsyncRequestBody.fromString("hello")).join();
+
+        verify(putRequestedFor(urlPathEqualTo(PATH))
+                   .withHeader("Authorization", containing("Credential=requestAccessKey/")));
+    }
+
+    @Test
+    void putObject_withoutRequestLevelCredentials_shouldSignWithClientCredentials() {
+        s3.putObject(
+            b -> b.bucket(BUCKET).key(KEY),
+            AsyncRequestBody.fromString("hello")).join();
+
+        verify(putRequestedFor(urlPathEqualTo(PATH))
+                   .withHeader("Authorization", containing("Credential=clientAccessKey/")));
+    }
+}

services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java

@@ -16,6 +16,7 @@
 package software.amazon.awssdk.services.s3.internal.crt;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -27,6 +28,7 @@
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_CHECKSUM_VALIDATION;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_PATH;
+import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.CRT_CREDENTIALS_PROVIDER_ADAPTER;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.SIGNING_NAME;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.SIGNING_REGION;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.USE_S3_EXPRESS_AUTH;
@@ -50,6 +52,7 @@
 import software.amazon.awssdk.core.checksums.RequestChecksumCalculation;
 import software.amazon.awssdk.core.checksums.ResponseChecksumValidation;
 import software.amazon.awssdk.core.interceptor.trait.HttpChecksum;
+import software.amazon.awssdk.crt.auth.credentials.CredentialsProvider;
 import software.amazon.awssdk.crt.auth.signing.AwsSigningConfig;
 import software.amazon.awssdk.crt.http.HttpProxyEnvironmentVariableSetting;
 import software.amazon.awssdk.crt.http.HttpRequest;
@@ -60,7 +63,6 @@
 import software.amazon.awssdk.crt.s3.S3ClientOptions;
 import software.amazon.awssdk.crt.s3.S3MetaRequest;
 import software.amazon.awssdk.crt.s3.S3MetaRequestOptions;
-import software.amazon.awssdk.crt.s3.S3MetaRequestResponseHandler;
 import software.amazon.awssdk.http.SdkHttpMethod;
 import software.amazon.awssdk.http.SdkHttpRequest;
 import software.amazon.awssdk.http.async.AsyncExecuteRequest;
@@ -623,6 +625,51 @@
         assertThat(actual.getResponseFileOption()).isEqualTo(S3MetaRequestOptions.ResponseFileOption.CREATE_OR_APPEND);
     }
 
+    @Test
+    public void execute_withRequestLevelCredentials_shouldCloseAdapterOnCompletion() {
+        CrtCredentialsProviderAdapter adapter = Mockito.mock(CrtCredentialsProviderAdapter.class);
+        when(adapter.crtCredentials()).thenReturn(Mockito.mock(CredentialsProvider.class));
+        S3MetaRequest metaRequest = Mockito.mock(S3MetaRequest.class);
+        when(s3Client.makeMetaRequest(any(S3MetaRequestOptions.class))).thenReturn(metaRequest);
+
+        AsyncExecuteRequest asyncExecuteRequest =
+            getExecuteRequestBuilder()
+                .putHttpExecutionAttribute(OPERATION_NAME, "GetObject")
+                .putHttpExecutionAttribute(SIGNING_REGION, Region.US_WEST_2)
+                .putHttpExecutionAttribute(SIGNING_NAME, "s3")
+                .putHttpExecutionAttribute(CRT_CREDENTIALS_PROVIDER_ADAPTER, adapter)
+                .build();
+
+        CompletableFuture<Void> future = asyncHttpClient.execute(asyncExecuteRequest);
+
+        Mockito.verify(adapter, Mockito.never()).close();
+
+        future.complete(null);
+
+        Mockito.verify(adapter).close();
+    }
+
+    @Test
+    void execute_whenMakeMetaRequestThrows_shouldCloseAdapter() {
+        CrtCredentialsProviderAdapter adapter = Mockito.mock(CrtCredentialsProviderAdapter.class);
+        when(adapter.crtCredentials()).thenReturn(Mockito.mock(CredentialsProvider.class));
+        when(s3Client.makeMetaRequest(any(S3MetaRequestOptions.class)))
+            .thenThrow(new RuntimeException("boom"));
+
+        AsyncExecuteRequest asyncExecuteRequest =
+            getExecuteRequestBuilder()
+                .putHttpExecutionAttribute(OPERATION_NAME, "GetObject")
+                .putHttpExecutionAttribute(SIGNING_REGION, Region.US_WEST_2)
+                .putHttpExecutionAttribute(SIGNING_NAME, "s3")
+                .putHttpExecutionAttribute(CRT_CREDENTIALS_PROVIDER_ADAPTER, adapter)
+                .build();
+
+        assertThatThrownBy(() -> asyncHttpClient.execute(asyncExecuteRequest))
+            .isInstanceOf(RuntimeException.class);
+
+        Mockito.verify(adapter).close();
+    }
+
     private AsyncExecuteRequest.Builder getExecuteRequestBuilder() {
         return getExecuteRequestBuilder(443);
     }