sigV4a

Author: stephenbawks

aws_lambda_powertools/utilities/auth/aws_auth.py

@@ -4,6 +4,7 @@
 from typing import Optional
 
 import botocore.session
+from botocore import crt
 from botocore.auth import SigV4Auth
 from botocore.awsrequest import AWSRequest
 from botocore.credentials import Credentials, ReadOnlyCredentials
@@ -21,6 +22,7 @@ class AWSServicePrefix(Enum):
     RESTAPI = "execute-api"
     HTTPAPI = "apigateway"
     APPSYNC = "appsync"
+    S3 = "s3"
 
 
 class AuthProvider(Enum):
@@ -36,6 +38,7 @@ class AuthProvider(Enum):
 class AWSSigV4Auth:
     """
     Authenticating Requests (AWS Signature Version 4)
+    Requests that were signed with SigV4 will have SignatureVersion set to AWS4-HMAC-SHA256
 
     Args:
         url (str): URL
@@ -107,3 +110,80 @@ def __init__(
 
         def __call__(self):
             return self.signed_request
+
+
+class AWSSigV4aAuth:
+    """
+    Authenticating Requests (AWS Signature Version 4a)
+    Requests that were signed with SigV4A will have a SignatureVersion set to AWS4-ECDSA-P256-SHA256
+
+    Args:
+        url (str): URL
+        region (str): AWS region
+        body (str, optional): Request body
+        params (dict, optional): Request parameters
+        headers (dict, optional): Request headers
+        method (str, optional): Request method
+        service (str, optional): AWS service
+        access_key (str, optional): AWS access key
+        secret_key (str, optional): AWS secret key
+        token (str, optional): AWS session token
+
+    Returns:
+        SigV4aAuth: SigV4aAuth instance
+
+    Examples
+    --------
+    **Using default credentials**
+    >>> from aws_lambda_powertools.utilities.iam import AWSSigV4aAuth
+    >>> auth = AWSSigV4aAuth(region="us-east-2", service=AWSServicePrefix.LATTICE, url="https://test-fake-service.vpc-lattice-svcs.us-east-2.on.aws")
+    """
+
+    def __init__(
+        self,
+        url: str,
+        region: str,
+        body: Optional[str] = None,
+        params: Optional[dict] = None,
+        headers: Optional[dict] = None,
+        method: Optional[str] = "GET",
+        service: Enum = AWSServicePrefix.LATTICE,
+        access_key: Optional[str] = None,
+        secret_key: Optional[str] = None,
+        token: Optional[str] = None,
+    ):
+        self.service = service.value
+        self.region = region
+        self.method = method
+        self.url = url
+        self.data = body
+        self.params = params
+        self.headers = headers
+
+        self.credentials: Credentials | ReadOnlyCredentials
+
+        if access_key and secret_key and token:
+            self.access_key = access_key
+            self.secret_key = secret_key
+            self.token = token
+            self.credentials = Credentials(access_key=self.access_key, secret_key=self.secret_key, token=self.token)
+        else:
+            credentials = botocore.session.Session().get_credentials()
+            self.credentials = credentials.get_frozen_credentials()
+
+        if self.headers is None:
+            self.headers = {"Content-Type": "application/json"}
+
+        signer = crt.auth.CrtSigV4AsymAuth(self.credentials, self.service, self.region)
+
+        request = AWSRequest(method=self.method, url=self.url, data=self.data, params=self.params, headers=self.headers)
+
+        if self.service == AWSServicePrefix.LATTICE.value:
+            # payload signing is not supported for vpc-lattice-svcs
+            request.context["payload_signing_enabled"] = False
+
+        signer.add_auth(request)
+        self.signed_request = request.prepare()
+
+        def __call__(self):
+            return self.signed_request

poetry.lock

@@ -48,6 +48,7 @@ typing-extensions = "^4.11.0"
 datadog-lambda = { version = ">=4.77,<6.0", optional = true }
 aws-encryption-sdk = { version = "^3.1.1", optional = true }
 jsonpath-ng = { version = "^1.6.0", optional = true }
+awscrt = "^0.20.9"
 
 [tool.poetry.dev-dependencies]
 coverage = { extras = ["toml"], version = "^7.5" }
@@ -122,6 +123,10 @@ types-redis = "^4.6.0.7"
 testcontainers = { extras = ["redis"], version = "^3.7.1" }
 multiprocess = "^0.70.16"
 
+
+[tool.poetry.group.dev-dependencies.dependencies]
+botocore = "^1.34.94"
+
 [tool.coverage.run]
 source = ["aws_lambda_powertools"]
 omit = [