Added option to disable payload signing. This is necessary e.g. for AWS Lattice which does not support payload signing.

Author: Jan Stiborek

CHANGELOG.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Added
+- `httpx_auth.AWS4Auth` - added option to disable payload signing to allow use this Auth class with AWS Lattice service.
 
 ## [0.23.1] - 2025-01-07
 ### Fixed

httpx_auth/_aws.py

@@ -8,7 +8,7 @@
 import hmac
 from collections import defaultdict
 from posixpath import normpath
-from typing import Generator
+from typing import Generator, Sequence, Optional
 from urllib.parse import quote
 
 import httpx
@@ -22,7 +22,15 @@ class AWS4Auth(httpx.Auth):
     requires_request_body = True
 
     def __init__(
-        self, access_id: str, secret_key: str, region: str, service: str, **kwargs
+        self,
+        access_id: str,
+        secret_key: str,
+        region: str,
+        service: str,
+        security_token: Optional[str] = None,
+        include_headers: Sequence[str] = tuple(),
+        enable_payload_signing: bool = True,
+        **kwargs,
     ):
         """
 
@@ -34,6 +42,9 @@ def __init__(
         :param service: The name of the service you're connecting to, as per endpoints at:
         http://docs.aws.amazon.com/general/latest/gr/rande.html
         e.g. elasticbeanstalk.
+        :param enable_payload_signing: Whether to include payload hash in signature
+        AWS Lattice service does not support payload signing - https://docs.aws.amazon.com/vpc-lattice/latest/ug/sigv4-authenticated-requests.html
+        Setting this parameter to False will set x-amz-content-sha256 header value to "UNSIGNED-PAYLOAD"
         :param security_token: Used for the x-amz-security-token header, for use with STS temporary credentials.
         :param include_headers: Set of headers to include in the canonical and signed headers, in addition to:
          * host
@@ -48,12 +59,9 @@ def __init__(
         self.access_id = access_id
         self.region = region
         self.service = service
-
-        self.security_token = kwargs.get("security_token")
-
-        self.include_headers = {
-            header.lower() for header in kwargs.get("include_headers", [])
-        }
+        self.enable_payload_signing = enable_payload_signing
+        self.security_token = security_token
+        self.include_headers = {header.lower() for header in include_headers}
 
     def auth_flow(
         self, request: httpx.Request
@@ -69,9 +77,13 @@ def auth_flow(
         # The x-amz-content-sha256 header is required for all AWS Signature Version 4 requests.
         # It provides a hash of the request payload.
         # If there is no payload, you must provide the hash of an empty string.
-        request.headers["x-amz-content-sha256"] = hashlib.sha256(
-            request.read()
-        ).hexdigest()
+        # This does not apply to AWS Lattice which does not support payload signing.
+        # In this case the value of this header must be set to "UNSIGNED-PAYLOAD".
+        if self.enable_payload_signing:
+            content_hash_digest = hashlib.sha256(request.read()).hexdigest()
+        else:
+            content_hash_digest = "UNSIGNED-PAYLOAD"
+        request.headers["x-amz-content-sha256"] = content_hash_digest
 
         # https://docs.aws.amazon.com/AmazonS3/latest/API/sig-v4-header-based-auth.html
         # if you are using temporary security credentials, you need to include x-amz-security-token in your request.

tests/aws_signature_v4/test_aws4auth_async.py

@@ -843,3 +843,36 @@ async def test_aws_auth_without_path(httpx_mock: HTTPXMock):
 
     async with httpx.AsyncClient() as client:
         await client.get("https://authorized_only", auth=auth)
+
+
+@time_machine.travel("2018-10-11T15:05:05.663979+00:00", tick=False)
+@pytest.mark.asyncio
+async def test_aws_auth_with_payload_signing_disabled(httpx_mock: HTTPXMock):
+
+    auth = httpx_auth.AWS4Auth(
+        access_id="access_id",
+        secret_key="wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
+        region="us-east-1",
+        service="vpc-lattice-svcs",
+        enable_payload_signing=False,
+        include_headers={"x-xyz-my-super-cool-header"},
+    )
+
+    httpx_mock.add_response(
+        url="https://authorized_only",
+        method="POST",
+        match_headers={
+            "x-amz-content-sha256": "UNSIGNED-PAYLOAD",
+            "Authorization": "AWS4-HMAC-SHA256 Credential=access_id/20181011/us-east-1/vpc-lattice-svcs/aws4_request, SignedHeaders=host;x-amz-content-sha256;x-amz-date;x-xyz-my-super-cool-header, Signature=1a9ad454247bd09ddaf8c096d4b1f767625c68fcb1f73aa2f8e1c9a80b9d2231",
+            "x-amz-date": "20181011T150505Z",
+            "x-xyz-my-super-cool-header": "1",
+        },
+    )
+
+    async with httpx.AsyncClient() as client:
+        await client.post(
+            "https://authorized_only",
+            auth=auth,
+            content="some-data-here",
+            headers={"x-xyz-my-super-cool-header": "1"},
+        )