Fix SigV4 signature for URIs with literal query parameters (#674)

* Fix SigV4 signature for URIs with literal query parameters
* Add unit tests for _format_canonical_query in SigV4Signer

Author: Alan4506

packages/aws-sdk-signers/.changes/next-release/aws-sdk-signers-bugfix-23a4eca5165f48efb1524b95f8498299.json

@@ -0,0 +1,4 @@
+{
+  "type": "bugfix",
+  "description": "Fixed SigV4 signature computation for URIs with literal query parameters (e.g., ?sync). parse_qsl was silently dropping query keys without values, causing InvalidSignatureException."
+}

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

@@ -315,7 +315,7 @@ def _format_canonical_query(self, *, query: str | None) -> str:
         if query is None:
             return ""
 
-        query_params = parse_qsl(qs=query)
+        query_params = parse_qsl(qs=query, keep_blank_values=True)
         query_parts = (
             (quote(string=key, safe=""), quote(string=value, safe=""))
             for key, value in query_params
@@ -695,7 +695,7 @@ async def _format_canonical_query(self, *, query: str | None) -> str:
         if query is None:
             return ""
 
-        query_params = parse_qsl(qs=query)
+        query_params = parse_qsl(qs=query, keep_blank_values=True)
         query_parts = (
             (quote(string=key, safe=""), quote(string=value, safe=""))
             for key, value in query_params

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

@@ -125,6 +125,18 @@ def test_sign_with_expired_identity(
                 identity=identity,
             )
 
+    def test_format_canonical_query_keeps_blank_values(self) -> None:
+        canonical_query = self.SIGV4_SYNC_SIGNER._format_canonical_query(  # pyright: ignore[reportPrivateUsage]
+            query="foo=bar&baz="
+        )
+        assert canonical_query == "baz=&foo=bar"
+
+    def test_format_canonical_query_with_literal_query_param(self) -> None:
+        canonical_query = self.SIGV4_SYNC_SIGNER._format_canonical_query(  # pyright: ignore[reportPrivateUsage]
+            query="sync"
+        )
+        assert canonical_query == "sync="
+
 
 class UnreadableAsyncStream:
     def __aiter__(self) -> typing.Self:
@@ -231,3 +243,15 @@ async def test_sign_event_stream(
         assert "X-Amz-Content-SHA256" in signed.fields
         payload_hash = signed.fields["X-Amz-Content-SHA256"].as_string()
         assert payload_hash == "STREAMING-AWS4-HMAC-SHA256-EVENTS"
+
+    async def test_format_canonical_query_keeps_blank_values(self) -> None:
+        canonical_query = await self.SIGV4_ASYNC_SIGNER._format_canonical_query(  # pyright: ignore[reportPrivateUsage]
+            query="foo=bar&baz="
+        )
+        assert canonical_query == "baz=&foo=bar"
+
+    async def test_format_canonical_query_with_literal_query_param(self) -> None:
+        canonical_query = await self.SIGV4_ASYNC_SIGNER._format_canonical_query(  # pyright: ignore[reportPrivateUsage]
+            query="sync"
+        )
+        assert canonical_query == "sync="