fix: skip CRT Content-Length when chunked

.changes/next-release/bugfix-AWSCRTHTTPClient-89a932e.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT HTTP Client",
+    "contributor": "",
+    "description": "Do not set the Content-Length header on a request that already carries Transfer-Encoding. Emitting both violates RFC 7230 and was rejected by the underlying CRT layer; this aligns the CRT client with the Netty client's existing behavior for chunked requests."
+}

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/request/CrtRequestAdapter.java

@@ -108,9 +108,11 @@ private static List<HttpHeader> createAsyncHttpHeaderList(URI uri, AsyncExecuteR
             crtHeaderList.add(new HttpHeader(Header.CONNECTION, Header.KEEP_ALIVE_VALUE));
         }
 
-        // Set Content-Length if needed
+        // Set Content-Length if needed, but never alongside Transfer-Encoding. RFC 7230 forbids sending both
         Optional<Long> contentLength = sdkExecuteRequest.requestContentPublisher().contentLength();
-        if (!sdkRequest.firstMatchingHeader(Header.CONTENT_LENGTH).isPresent() && contentLength.isPresent()) {
+        if (!sdkRequest.firstMatchingHeader(Header.CONTENT_LENGTH).isPresent()
+            && !sdkRequest.firstMatchingHeader(Header.TRANSFER_ENCODING).isPresent()
+            && contentLength.isPresent()) {
             crtHeaderList.add(new HttpHeader(Header.CONTENT_LENGTH, Long.toString(contentLength.get())));
         }
 
@@ -135,8 +137,6 @@ private static List<HttpHeader> createHttpHeaderList(URI uri, HttpExecuteRequest
             crtHeaderList.add(new HttpHeader(Header.CONNECTION, Header.KEEP_ALIVE_VALUE));
         }
 
-        // We assume content length was set by the caller if a stream was present, so don't set it here.
-
         // Add the rest of the Headers
         sdkRequest.forEachHeader((key, value) -> value.stream().map(val -> new HttpHeader(key, val)).forEach(crtHeaderList::add));
 

http-clients/aws-crt-client/src/test/java/software/amazon/awssdk/http/crt/internal/request/CrtRequestAdapterTest.java

@@ -0,0 +1,115 @@
+/*
+ * 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.http.crt.internal.request;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static software.amazon.awssdk.http.HttpTestUtils.createProvider;
+
+import java.net.URI;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.crt.http.HttpHeader;
+import software.amazon.awssdk.crt.http.HttpRequestBase;
+import software.amazon.awssdk.http.Header;
+import software.amazon.awssdk.http.HttpExecuteRequest;
+import software.amazon.awssdk.http.Protocol;
+import software.amazon.awssdk.http.SdkHttpFullRequest;
+import software.amazon.awssdk.http.SdkHttpMethod;
+import software.amazon.awssdk.http.async.AsyncExecuteRequest;
+import software.amazon.awssdk.http.async.SdkHttpContentPublisher;
+import software.amazon.awssdk.http.crt.internal.CrtAsyncRequestContext;
+import software.amazon.awssdk.http.crt.internal.CrtRequestContext;
+
+public class CrtRequestAdapterTest {
+
+    @Test
+    public void toAsyncCrtRequest_transferEncodingPresent_doesNotAddContentLength() {
+        SdkHttpFullRequest sdkRequest = requestBuilder().putHeader(Header.TRANSFER_ENCODING, "chunked").build();
+        SdkHttpContentPublisher publisher = createProvider("content-with-known-length");
+
+        HttpRequestBase crtRequest = toAsyncCrtRequest(sdkRequest, publisher);
+
+        assertThat(headerNames(crtRequest)).contains(Header.TRANSFER_ENCODING)
+                                           .doesNotContain(Header.CONTENT_LENGTH);
+    }
+
+    @Test
+    public void toAsyncCrtRequest_noTransferEncoding_addsContentLengthFromPublisher() {
+        SdkHttpFullRequest sdkRequest = requestBuilder().build();
+        SdkHttpContentPublisher publisher = createProvider("content-with-known-length");
+
+        HttpRequestBase crtRequest = toAsyncCrtRequest(sdkRequest, publisher);
+
+        assertThat(headerNames(crtRequest)).contains(Header.CONTENT_LENGTH)
+                                           .doesNotContain(Header.TRANSFER_ENCODING);
+    }
+
+    @Test
+    public void toCrtRequest_transferEncodingPresent_doesNotAddContentLength() {
+        SdkHttpFullRequest sdkRequest = requestBuilder().putHeader(Header.TRANSFER_ENCODING, "chunked").build();
+
+        HttpRequestBase crtRequest = toCrtRequest(sdkRequest);
+
+        assertThat(headerNames(crtRequest)).contains(Header.TRANSFER_ENCODING)
+                                           .doesNotContain(Header.CONTENT_LENGTH);
+    }
+
+    @Test
+    public void toCrtRequest_contentLengthOnRequest_isPreserved() {
+        SdkHttpFullRequest sdkRequest = requestBuilder().putHeader(Header.CONTENT_LENGTH, "42").build();
+
+        HttpRequestBase crtRequest = toCrtRequest(sdkRequest);
+
+        assertThat(headerNames(crtRequest)).contains(Header.CONTENT_LENGTH)
+                                           .doesNotContain(Header.TRANSFER_ENCODING);
+    }
+
+    private static SdkHttpFullRequest.Builder requestBuilder() {
+        return SdkHttpFullRequest.builder()
+                                 .uri(URI.create("http://localhost:8080"))
+                                 .method(SdkHttpMethod.POST)
+                                 .encodedPath("/");
+    }
+
+    private static HttpRequestBase toAsyncCrtRequest(SdkHttpFullRequest sdkRequest, SdkHttpContentPublisher publisher) {
+        AsyncExecuteRequest asyncRequest = AsyncExecuteRequest.builder()
+                                                              .request(sdkRequest)
+                                                              .requestContentPublisher(publisher)
+                                                              .build();
+        CrtAsyncRequestContext context = CrtAsyncRequestContext.builder()
+                                                               .request(asyncRequest)
+                                                               .readBufferSize(2000)
+                                                               .protocol(Protocol.HTTP1_1)
+                                                               .build();
+        return CrtRequestAdapter.toAsyncCrtRequest(context);
+    }
+
+    private static HttpRequestBase toCrtRequest(SdkHttpFullRequest sdkRequest) {
+        HttpExecuteRequest executeRequest = HttpExecuteRequest.builder()
+                                                              .request(sdkRequest)
+                                                              .build();
+        CrtRequestContext context = CrtRequestContext.builder()
+                                                     .request(executeRequest)
+                                                     .readBufferSize(2000)
+                                                     .build();
+        return CrtRequestAdapter.toCrtRequest(context);
+    }
+
+    private static List<String> headerNames(HttpRequestBase crtRequest) {
+        return crtRequest.getHeaders().stream().map(HttpHeader::getName).collect(Collectors.toList());
+    }
+}

test/protocol-tests/src/test/java/software/amazon/awssdk/protocol/tests/chunkedencoding/TransferEncodingChunkedFunctionalTests.java

@@ -18,16 +18,16 @@
 
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
 import static com.github.tomakehurst.wiremock.client.WireMock.anyUrl;
+import static com.github.tomakehurst.wiremock.client.WireMock.binaryEqualTo;
 import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
 import static com.github.tomakehurst.wiremock.client.WireMock.post;
 import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
-import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
-import static com.github.tomakehurst.wiremock.client.WireMock.verify;
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
 import static software.amazon.awssdk.http.Header.CONTENT_LENGTH;
 import static software.amazon.awssdk.http.Header.TRANSFER_ENCODING;
 import static software.amazon.awssdk.utils.FunctionalUtils.invokeSafely;
 
-import com.github.tomakehurst.wiremock.junit.WireMockRule;
+import com.github.tomakehurst.wiremock.junit5.WireMockExtension;
 import io.reactivex.Flowable;
 import java.io.ByteArrayInputStream;
 import java.io.FilterInputStream;
@@ -40,58 +40,152 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
 import org.apache.commons.lang3.RandomStringUtils;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.reactivestreams.Subscriber;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.sync.RequestBody;
 import software.amazon.awssdk.http.ContentStreamProvider;
+import software.amazon.awssdk.http.SdkHttpClient;
+import software.amazon.awssdk.http.SdkHttpConfigurationOption;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.http.crt.AwsCrtAsyncHttpClient;
+import software.amazon.awssdk.http.crt.AwsCrtHttpClient;
 import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
 import software.amazon.awssdk.services.protocolrestjson.ProtocolRestJsonAsyncClient;
 import software.amazon.awssdk.services.protocolrestjson.ProtocolRestJsonClient;
 import software.amazon.awssdk.services.protocolrestjson.model.StreamingInputOperationChunkedEncodingRequest;
+import software.amazon.awssdk.utils.AttributeMap;
 
 public final class TransferEncodingChunkedFunctionalTests {
-    @Rule
-    public WireMockRule wireMock = new WireMockRule(0);
-
-    @Test
-    public void apacheClientStreamingOperation_withoutContentLength_addsTransferEncodingDoesNotAddContentLength() {
+    // A single-chunk body and a body large enough to span multiple HTTP chunks (streamed in STREAM_BUFFER_SIZE pieces).
+    private static final int SMALL_BODY_SIZE = 1024;
+    private static final int MULTI_CHUNK_BODY_SIZE = 1024 * 1024;
+    private static final int STREAM_BUFFER_SIZE = 128 * 1024;
+
+    @RegisterExtension
+    static WireMockExtension wireMock = WireMockExtension.newInstance()
+                                                         .options(wireMockConfig().dynamicPort().dynamicHttpsPort())
+                                                         .build();
+
+    @BeforeEach
+    public void setUp() {
         stubSuccessfulResponse();
-        try (ProtocolRestJsonClient client = ProtocolRestJsonClient.builder()
-                                                                   .httpClient(ApacheHttpClient.builder().build())
-                                                                   .endpointOverride(URI.create("http://localhost:" + wireMock.port()))
-                                                                   .build()) {
-            TestContentProvider provider = new TestContentProvider(RandomStringUtils.random(1000).getBytes(StandardCharsets.UTF_8));
-            RequestBody requestBody = RequestBody.fromContentProvider(provider, "binary/octet-stream");
-            client.streamingInputOperationChunkedEncoding(StreamingInputOperationChunkedEncodingRequest.builder().build(), requestBody);
+    }
 
-            verify(postRequestedFor(anyUrl()).withHeader(TRANSFER_ENCODING, equalTo("chunked")));
-            verify(postRequestedFor(anyUrl()).withoutHeader(CONTENT_LENGTH));
-        }
+    // The clients are tested over HTTPS so the request takes the default unsigned-payload path. Over plaintext HTTP the
+    // SDK forces payload signing, which buffers the body and attaches a Content-Length that conflicts with
+    // Transfer-Encoding: chunked - a configuration real chunked callers never use. Each client is exercised with a
+    // single-chunk body and a multi-chunk body; the large case additionally proves the streamed body arrives intact.
+    private static Stream<Arguments> asyncClients() {
+        return Stream.of(
+            Arguments.of("Netty, small body", trustAllAsyncClient(NettyNioAsyncHttpClient::builder), SMALL_BODY_SIZE),
+            Arguments.of("Netty, large body", trustAllAsyncClient(NettyNioAsyncHttpClient::builder), MULTI_CHUNK_BODY_SIZE),
+            Arguments.of("CRT, small body", trustAllAsyncClient(AwsCrtAsyncHttpClient::builder), SMALL_BODY_SIZE),
+            Arguments.of("CRT, large body", trustAllAsyncClient(AwsCrtAsyncHttpClient::builder), MULTI_CHUNK_BODY_SIZE)
+        );
     }
 
-    @Test
-    public void nettyClientStreamingOperation_withoutContentLength_addsTransferEncodingDoesNotAddContentLength() {
-        stubSuccessfulResponse();
-        try (ProtocolRestJsonAsyncClient client = ProtocolRestJsonAsyncClient.builder()
-                                                                             .httpClient(NettyNioAsyncHttpClient.create())
-                                                                             .endpointOverride(URI.create("http://localhost:" + wireMock.port()))
-                                                                             .build()) {
+    private static Stream<Arguments> syncClients() {
+        return Stream.of(
+            Arguments.of("Apache, small body", trustAllSyncClient(ApacheHttpClient::builder), SMALL_BODY_SIZE),
+            Arguments.of("Apache, large body", trustAllSyncClient(ApacheHttpClient::builder), MULTI_CHUNK_BODY_SIZE),
+            Arguments.of("CRT, small body", trustAllSyncClient(AwsCrtHttpClient::builder), SMALL_BODY_SIZE),
+            Arguments.of("CRT, large body", trustAllSyncClient(AwsCrtHttpClient::builder), MULTI_CHUNK_BODY_SIZE)
+        );
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("asyncClients")
+    public void asyncClientStreamingOperation_withoutContentLength_usesChunkedEncodingAndStreamsBody(
+        String description, Supplier<SdkAsyncHttpClient> httpClient, int bodySize) {
+        byte[] body = randomBody(bodySize);
+        try (ProtocolRestJsonAsyncClient client = asyncClient(httpClient.get())) {
             client.streamingInputOperationChunkedEncoding(StreamingInputOperationChunkedEncodingRequest.builder().build(),
-                                                          customAsyncRequestBodyWithoutContentLength()).join();
+                                                          streamingBodyWithoutContentLength(body)).join();
+
+            verifyChunkedRequest(body);
+        }
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("syncClients")
+    public void syncClientStreamingOperation_withoutContentLength_usesChunkedEncodingAndStreamsBody(
+        String description, Supplier<SdkHttpClient> httpClient, int bodySize) {
+        byte[] body = randomBody(bodySize);
+        try (ProtocolRestJsonClient client = syncClient(httpClient.get())) {
+            RequestBody requestBody = RequestBody.fromContentProvider(new TestContentProvider(body), "binary/octet-stream");
+            client.streamingInputOperationChunkedEncoding(StreamingInputOperationChunkedEncodingRequest.builder().build(), requestBody);
 
-            verify(postRequestedFor(anyUrl()).withHeader(TRANSFER_ENCODING, equalTo("chunked")));
+            verifyChunkedRequest(body);
         }
     }
 
+    private void verifyChunkedRequest(byte[] body) {
+        wireMock.verify(postRequestedFor(anyUrl()).withHeader(TRANSFER_ENCODING, equalTo("chunked")));
+        wireMock.verify(postRequestedFor(anyUrl()).withoutHeader(CONTENT_LENGTH));
+        wireMock.verify(postRequestedFor(anyUrl()).withRequestBody(binaryEqualTo(body)));
+    }
+
+    private static byte[] randomBody(int size) {
+        return RandomStringUtils.randomAscii(size).getBytes(StandardCharsets.UTF_8);
+    }
+
+    private static Supplier<SdkAsyncHttpClient> trustAllAsyncClient(Supplier<SdkAsyncHttpClient.Builder<?>> builder) {
+        return () -> trustAllClient(builder.get());
+    }
+
+    private static Supplier<SdkHttpClient> trustAllSyncClient(Supplier<SdkHttpClient.Builder<?>> builder) {
+        return () -> trustAllClient(builder.get());
+    }
+
+    private ProtocolRestJsonAsyncClient asyncClient(SdkAsyncHttpClient httpClient) {
+        return ProtocolRestJsonAsyncClient.builder()
+                                          .httpClient(httpClient)
+                                          .endpointOverride(URI.create("https://localhost:" + wireMock.getHttpsPort()))
+                                          .build();
+    }
+
+    private ProtocolRestJsonClient syncClient(SdkHttpClient httpClient) {
+        return ProtocolRestJsonClient.builder()
+                                     .httpClient(httpClient)
+                                     .endpointOverride(URI.create("https://localhost:" + wireMock.getHttpsPort()))
+                                     .build();
+    }
+
     private void stubSuccessfulResponse() {
-        stubFor(post(anyUrl()).willReturn(aResponse().withStatus(200)));
+        wireMock.stubFor(post(anyUrl()).willReturn(aResponse().withStatus(200)));
     }
 
-    private AsyncRequestBody customAsyncRequestBodyWithoutContentLength() {
+    private static SdkAsyncHttpClient trustAllClient(SdkAsyncHttpClient.Builder<?> builder) {
+        return builder.buildWithDefaults(trustAll());
+    }
+
+    private static SdkHttpClient trustAllClient(SdkHttpClient.Builder<?> builder) {
+        return builder.buildWithDefaults(trustAll());
+    }
+
+    private static AttributeMap trustAll() {
+        return AttributeMap.builder()
+                           .put(SdkHttpConfigurationOption.TRUST_ALL_CERTIFICATES, true)
+                           .build();
+    }
+
+    // A streaming body that reports no content length, forcing chunked transfer encoding. The payload is split into
+    // multiple buffers so a large body is emitted as more than one HTTP chunk.
+    private AsyncRequestBody streamingBodyWithoutContentLength(byte[] body) {
+        List<ByteBuffer> buffers = new ArrayList<>();
+        for (int offset = 0; offset < body.length; offset += STREAM_BUFFER_SIZE) {
+            int end = Math.min(offset + STREAM_BUFFER_SIZE, body.length);
+            buffers.add(ByteBuffer.wrap(body, offset, end - offset));
+        }
         return new AsyncRequestBody() {
             @Override
             public Optional<Long> contentLength() {
@@ -100,8 +194,7 @@ public Optional<Long> contentLength() {
 
             @Override
             public void subscribe(Subscriber<? super ByteBuffer> s) {
-                Flowable.fromPublisher(AsyncRequestBody.fromBytes("Random text".getBytes()))
-                        .subscribe(s);
+                Flowable.fromIterable(buffers).subscribe(s);
             }
         };
     }