Pass new request into signer for canonical_request

packages/aws-sdk-signers/src/aws_sdk_signers/signers.py

@@ -441,7 +441,7 @@ async def sign(
         # Construct core signing components
         canonical_request = await self.canonical_request(
             signing_properties=signing_properties,
-            request=request,
+            request=new_request,
         )
         string_to_sign = await self.string_to_sign(
             canonical_request=canonical_request,
@@ -453,7 +453,7 @@ async def sign(
             signing_properties=new_signing_properties,
         )
 
-        signing_fields = await self._normalize_signing_fields(request=request)
+        signing_fields = await self._normalize_signing_fields(request=new_request)
         credential_scope = await self._scope(signing_properties=new_signing_properties)
         credential = f"{identity.access_key_id}/{credential_scope}"
         authorization = await self.generate_authorization_field(

packages/aws-sdk-signers/tests/unit/test_signers.py

@@ -1,6 +1,7 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+import copy
 import re
 import typing
 from datetime import UTC, datetime
@@ -76,6 +77,23 @@ def test_sign(
         authorization_field = signed_request.fields["authorization"]
         assert SIGV4_RE.match(authorization_field.as_string())
 
+    def test_sign_doesnt_modify_original_request(
+        self,
+        aws_identity: AWSCredentialIdentity,
+        aws_request: AWSRequest,
+        signing_properties: SigV4SigningProperties,
+    ) -> None:
+        original_request = copy.deepcopy(aws_request)
+        signed_request = self.SIGV4_SYNC_SIGNER.sign(
+            signing_properties=signing_properties,
+            request=aws_request,
+            identity=aws_identity,
+        )
+        assert isinstance(signed_request, AWSRequest)
+        assert signed_request is not aws_request
+        assert aws_request.fields == original_request.fields
+        assert signed_request.fields != aws_request.fields
+
     @typing.no_type_check
     def test_sign_with_invalid_identity(
         self, aws_request: AWSRequest, signing_properties: SigV4SigningProperties
@@ -127,6 +145,23 @@ async def test_sign(
         authorization_field = signed_request.fields["authorization"]
         assert SIGV4_RE.match(authorization_field.as_string())
 
+    async def test_sign_doesnt_modify_original_request(
+        self,
+        aws_identity: AWSCredentialIdentity,
+        aws_request: AWSRequest,
+        signing_properties: SigV4SigningProperties,
+    ) -> None:
+        original_request = copy.deepcopy(aws_request)
+        signed_request = await self.SIGV4_ASYNC_SIGNER.sign(
+            signing_properties=signing_properties,
+            request=aws_request,
+            identity=aws_identity,
+        )
+        assert isinstance(signed_request, AWSRequest)
+        assert signed_request is not aws_request
+        assert aws_request.fields == original_request.fields
+        assert signed_request.fields != aws_request.fields
+
     @typing.no_type_check
     async def test_sign_with_invalid_identity(
         self, aws_request: AWSRequest, signing_properties: SigV4SigningProperties