Decouple auth from http

This decouples auth from HTTP, allowing it to be defined centrally.

A number of changes have been made to various interfaces. Notably
identity resolvers are generally expected to have zero-arg
constructors, instead getting everything they need from their
properties where possible.

Construction of event signers has been moved into AuthScheme. Right
now it takes the entire context of the request during construction,
but we should consider passing identity and signing properties in
at signing time like normal signers so that they can be reused.

Author: JordonPhillips

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

@@ -2,21 +2,14 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from dataclasses import dataclass
-from datetime import UTC, datetime
+from datetime import datetime
 
-from .interfaces.identity import Identity
+from .interfaces.identity import AWSCredentialsIdentity
 
 
 @dataclass(kw_only=True)
-class AWSCredentialIdentity(Identity):
+class AWSCredentialIdentity(AWSCredentialsIdentity):
     access_key_id: str
     secret_access_key: str
     session_token: str | None = None
     expiration: datetime | None = None
-
-    @property
-    def is_expired(self) -> bool:
-        """Whether the identity is expired."""
-        if self.expiration is None:
-            return False
-        return self.expiration < datetime.now(UTC)

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

@@ -3,39 +3,39 @@
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import UTC, datetime
 from typing import Protocol, runtime_checkable
 
 
+@runtime_checkable
 class Identity(Protocol):
     """An entity available to the client representing who the user is."""
 
-    # The expiration time of the identity. If time zone is provided,
-    # it is updated to UTC. The value must always be in UTC.
     expiration: datetime | None = None
+    """The expiration time of the identity.
+
+    If time zone is provided, it is updated to UTC. The value must always be in UTC.
+    """
 
     @property
     def is_expired(self) -> bool:
         """Whether the identity is expired."""
-        ...
+        if self.expiration is None:
+            return False
+        return datetime.now(tz=UTC) >= self.expiration
 
 
 @runtime_checkable
-class AWSCredentialsIdentity(Protocol):
+class AWSCredentialsIdentity(Identity, Protocol):
     """AWS Credentials Identity."""
 
-    # The access key ID.
     access_key_id: str
+    """A unique identifier for an AWS user or role."""
 
-    # The secret access key.
     secret_access_key: str
+    """A secret key used in conjunction with the access key ID to authenticate
+    programmatic access to AWS services."""
 
-    # The session token.
-    session_token: str | None
-
-    expiration: datetime | None = None
-
-    @property
-    def is_expired(self) -> bool:
-        """Whether the identity is expired."""
-        ...
+    session_token: str | None = None
+    """A temporary token used to specify the current session for the supplied
+    credentials."""

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

@@ -15,7 +15,6 @@
 from urllib.parse import parse_qsl, quote
 
 from ._http import AWSRequest, Field, URI
-from ._identity import AWSCredentialIdentity
 from ._io import AsyncBytesReader
 from .exceptions import AWSSDKWarning, MissingExpectedParameterException
 from .interfaces.identity import AWSCredentialsIdentity as _AWSCredentialsIdentity
@@ -55,27 +54,27 @@ class SigV4Signer:
     def sign(
         self,
         *,
-        signing_properties: SigV4SigningProperties,
-        http_request: AWSRequest,
-        identity: AWSCredentialIdentity,
+        request: AWSRequest,
+        identity: _AWSCredentialsIdentity,
+        properties: SigV4SigningProperties,
     ) -> AWSRequest:
         """Generate and apply a SigV4 Signature to a copy of the supplied request.
 
-        :param signing_properties: SigV4SigningProperties to define signing primitives
-            such as the target service, region, and date.
-        :param http_request: An AWSRequest to sign prior to sending to the service.
+        :param request: An AWSRequest to sign prior to sending to the service.
         :param identity: A set of credentials representing an AWS Identity or role
             capacity.
+        :param properties: SigV4SigningProperties to define signing primitives such as
+            the target service, region, and date.
         """
         # Copy and prepopulate any missing values in the
         # supplied request and signing properties.
         self._validate_identity(identity=identity)
         new_signing_properties = self._normalize_signing_properties(
-            signing_properties=signing_properties
+            signing_properties=properties
         )
         assert "date" in new_signing_properties
 
-        new_request = self._generate_new_request(request=http_request)
+        new_request = self._generate_new_request(request=request)
         self._apply_required_fields(
             request=new_request,
             signing_properties=new_signing_properties,
@@ -164,7 +163,7 @@ def _signature(
     def _hash(self, key: bytes, value: str) -> bytes:
         return hmac.new(key=key, msg=value.encode(), digestmod=sha256).digest()
 
-    def _validate_identity(self, *, identity: AWSCredentialIdentity) -> None:
+    def _validate_identity(self, *, identity: _AWSCredentialsIdentity) -> None:
         """Perform runtime and expiration checks before attempting signing."""
         if not isinstance(identity, _AWSCredentialsIdentity):  # pyright: ignore
             raise ValueError(
@@ -195,7 +194,7 @@ def _apply_required_fields(
         *,
         request: AWSRequest,
         signing_properties: SigV4SigningProperties,
-        identity: AWSCredentialIdentity,
+        identity: _AWSCredentialsIdentity,
     ) -> None:
         # Apply required X-Amz-Date if neither X-Amz-Date nor Date are present.
         if "Date" not in request.fields and "X-Amz-Date" not in request.fields:
@@ -427,26 +426,26 @@ class AsyncSigV4Signer:
     async def sign(
         self,
         *,
-        signing_properties: SigV4SigningProperties,
-        http_request: AWSRequest,
-        identity: AWSCredentialIdentity,
+        request: AWSRequest,
+        identity: _AWSCredentialsIdentity,
+        properties: SigV4SigningProperties,
     ) -> AWSRequest:
         """Generate and apply a SigV4 Signature to a copy of the supplied request.
 
-        :param signing_properties: SigV4SigningProperties to define signing primitives
-            such as the target service, region, and date.
-        :param http_request: An AWSRequest to sign prior to sending to the service.
+        :param request: An AWSRequest to sign prior to sending to the service.
         :param identity: A set of credentials representing an AWS Identity or role
             capacity.
+        :param properties: SigV4SigningProperties to define signing primitives such as
+            the target service, region, and date.
         """
         # Copy and prepopulate any missing values in the
         # supplied request and signing properties.
 
         await self._validate_identity(identity=identity)
         new_signing_properties = await self._normalize_signing_properties(
-            signing_properties=signing_properties
+            signing_properties=properties
         )
-        new_request = await self._generate_new_request(request=http_request)
+        new_request = await self._generate_new_request(request=request)
         await self._apply_required_fields(
             request=new_request,
             signing_properties=new_signing_properties,
@@ -455,7 +454,7 @@ async def sign(
 
         # Construct core signing components
         canonical_request = await self.canonical_request(
-            signing_properties=signing_properties,
+            signing_properties=properties,
             request=new_request,
         )
         string_to_sign = await self.string_to_sign(
@@ -535,7 +534,7 @@ async def _signature(
     async def _hash(self, key: bytes, value: str) -> bytes:
         return hmac.new(key=key, msg=value.encode(), digestmod=sha256).digest()
 
-    async def _validate_identity(self, *, identity: AWSCredentialIdentity) -> None:
+    async def _validate_identity(self, *, identity: _AWSCredentialsIdentity) -> None:
         """Perform runtime and expiration checks before attempting signing."""
         if not isinstance(identity, _AWSCredentialsIdentity):  # pyright: ignore
             raise ValueError(
@@ -566,7 +565,7 @@ async def _apply_required_fields(
         *,
         request: AWSRequest,
         signing_properties: SigV4SigningProperties,
-        identity: AWSCredentialIdentity,
+        identity: _AWSCredentialsIdentity,
     ) -> None:
         # Apply required X-Amz-Date if neither X-Amz-Date nor Date are present.
         if "Date" not in request.fields and "X-Amz-Date" not in request.fields:
@@ -804,26 +803,27 @@ class AsyncEventSigner:
     def __init__(
         self,
         *,
-        signing_properties: SigV4SigningProperties,
-        identity: AWSCredentialIdentity,
+        properties: SigV4SigningProperties,
+        identity: _AWSCredentialsIdentity,
         initial_signature: bytes,
+        event_encoder_cls: type["EventHeaderEncoder"],
     ):
-        self._signing_properties = signing_properties
-        self._identity = identity
         self._prior_signature = initial_signature
         self._signing_lock = asyncio.Lock()
+        self._event_encoder_cls = event_encoder_cls
+        self._properties = properties
+        self._identity = identity
 
-    async def sign_event(
+    async def sign(
         self,
         *,
-        event_message: "EventMessage",
-        event_encoder_cls: type["EventHeaderEncoder"],
+        event: "EventMessage",
     ) -> "EventMessage":
         async with self._signing_lock:
             # Copy and prepopulate any missing values in the
             # signing properties.
             new_signing_properties = SigV4SigningProperties(  # type: ignore
-                **self._signing_properties
+                **self._properties
             )
             # TODO: If date is in properties, parse a datetime from it.
             date_obj = datetime.datetime.now(datetime.UTC)
@@ -834,11 +834,11 @@ async def sign_event(
 
             timestamp = new_signing_properties["date"]
             headers: dict[str, str | bytes | datetime.datetime] = {":date": date_obj}
-            encoder = event_encoder_cls()
+            encoder = self._event_encoder_cls()
             encoder.encode_headers(headers)
             encoded_headers = encoder.get_result()
 
-            payload = event_message.encode()
+            payload = event.encode()
 
             string_to_sign = await self._event_string_to_sign(
                 timestamp=timestamp,
@@ -848,19 +848,20 @@ async def sign_event(
                 prior_signature=self._prior_signature,
             )
             event_signature = await self._sign_event(
+                identity=self._identity,
                 timestamp=timestamp,
                 string_to_sign=string_to_sign,
-                signing_properties=new_signing_properties,
+                properties=new_signing_properties,
             )
             headers[":chunk-signature"] = event_signature
 
-            event_message.headers = headers
-            event_message.payload = payload
+            event.headers = headers
+            event.payload = payload
 
             # set new prior signature before releasing the lock
             self._prior_signature = hexlify(event_signature)
 
-        return event_message
+        return event
 
     async def _event_string_to_sign(
         self,
@@ -885,13 +886,14 @@ async def _sign_event(
         *,
         timestamp: str,
         string_to_sign: str,
-        signing_properties: SigV4SigningProperties,
+        identity: _AWSCredentialsIdentity,
+        properties: SigV4SigningProperties,
     ) -> bytes:
-        key = self._identity.secret_access_key.encode("utf-8")
+        key = identity.secret_access_key.encode("utf-8")
         today = timestamp[:8].encode("utf-8")
         k_date = self._hash(b"AWS4" + key, today)
-        k_region = self._hash(k_date, signing_properties["region"].encode("utf-8"))
-        k_service = self._hash(k_region, signing_properties["service"].encode("utf-8"))
+        k_region = self._hash(k_date, properties["region"].encode("utf-8"))
+        k_service = self._hash(k_region, properties["service"].encode("utf-8"))
         k_signing = self._hash(k_service, b"aws4_request")
         return self._hash(k_signing, string_to_sign.encode("utf-8"))
 

packages/aws-sdk-signers/tests/unit/auth/test_sigv4.py

@@ -112,8 +112,8 @@ def _test_signature_version_4_sync(test_case_name: str, signer: SigV4Signer) ->
     assert test_case.string_to_sign == actual_string_to_sign
     with pytest.warns(AWSSDKWarning):
         signed_request = signer.sign(
-            signing_properties=signing_props,
-            http_request=request,
+            properties=signing_props,
+            request=request,
             identity=test_case.credentials,
         )
     assert (
@@ -151,8 +151,8 @@ async def _test_signature_version_4_async(
     assert test_case.string_to_sign == actual_string_to_sign
     with pytest.warns(AWSSDKWarning):
         signed_request = await signer.sign(
-            signing_properties=signing_props,
-            http_request=request,
+            properties=signing_props,
+            request=request,
             identity=test_case.credentials,
         )
     assert (