Append aws-chunked to Content-Encoding header if present (#2859)

.changes/next-release/bugfix-s3-10151.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "``s3``",
+  "description": "boto3 no longer overwrites user supplied `Content-Encoding` with `aws-chunked` when user also supplies `ChecksumAlgorithm`."
+}

botocore/httpchecksum.py

@@ -331,7 +331,12 @@ def _apply_request_trailer_checksum(request):
         return
 
     headers["Transfer-Encoding"] = "chunked"
-    headers["Content-Encoding"] = "aws-chunked"
+    if "Content-Encoding" in headers:
+        # We need to preserve the existing content encoding and add
+        # aws-chunked as a new content encoding.
+        headers["Content-Encoding"] += ",aws-chunked"
+    else:
+        headers["Content-Encoding"] = "aws-chunked"
     headers["X-Amz-Trailer"] = location_name
 
     content_length = determine_content_length(body)

tests/functional/test_s3.py

@@ -2141,6 +2141,27 @@ def test_checksums_included_in_expected_operations(
         assert "Content-MD5" in stub.requests[-1].headers
 
 
+@pytest.mark.parametrize(
+    "content_encoding, expected_header",
+    [("foo", b"foo,aws-chunked"), (None, b"aws-chunked")],
+)
+def test_checksum_content_encoding(content_encoding, expected_header):
+    op_kwargs = {
+        "Bucket": "mybucket",
+        "Key": "mykey",
+        "Body": b"foo",
+        "ChecksumAlgorithm": "sha256",
+    }
+    if content_encoding is not None:
+        op_kwargs["ContentEncoding"] = content_encoding
+    s3 = _create_s3_client()
+    with ClientHTTPStubber(s3) as http_stubber:
+        http_stubber.add_response()
+        s3.put_object(**op_kwargs)
+        request_headers = http_stubber.requests[-1].headers
+        assert request_headers["Content-Encoding"] == expected_header
+
+
 def _s3_addressing_test_cases():
     # The default behavior for sigv2. DNS compatible buckets
     yield dict(

tests/unit/test_httpchecksum.py

@@ -301,6 +301,34 @@ def test_apply_request_checksum_flex_header_trailer_explicit_digest(self):
         # The body should not have been wrapped
         self.assertIsInstance(request["body"], bytes)
 
+    def test_apply_request_checksum_content_encoding_preset(self):
+        request = self._build_request(b"")
+        request["context"]["checksum"] = {
+            "request_algorithm": {
+                "in": "trailer",
+                "algorithm": "crc32",
+                "name": "x-amz-checksum-crc32",
+            }
+        }
+        request["headers"]["Content-Encoding"] = "foo"
+        apply_request_checksum(request)
+        # The content encoding should only have been appended
+        self.assertEqual(
+            request["headers"]["Content-Encoding"], "foo,aws-chunked"
+        )
+
+    def test_apply_request_checksum_content_encoding_default(self):
+        request = self._build_request(b"")
+        request["context"]["checksum"] = {
+            "request_algorithm": {
+                "in": "trailer",
+                "algorithm": "crc32",
+                "name": "x-amz-checksum-crc32",
+            }
+        }
+        apply_request_checksum(request)
+        self.assertEqual(request["headers"]["Content-Encoding"], "aws-chunked")
+
     def test_response_checksum_algorithm_no_model(self):
         request = self._build_request(b"")
         operation_model = self._make_operation_model()