feat(auth): SigV4 Signing for calls that are taking IAM Auth

aws_lambda_powertools/utilities/auth/__init__.py

@@ -0,0 +1,5 @@
+"""Advanced feature flags utility"""
+
+from .aws_auth import ServicePrefix, SigV4aAuth, SigV4Auth
+
+__all__ = ["ServicePrefix", "SigV4Auth", "SigV4aAuth"]

aws_lambda_powertools/utilities/auth/aws_auth.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import json
+import os
+from enum import Enum
+from typing import Optional
+
+import botocore.session
+from botocore import crt
+from botocore.awsrequest import AWSRequest
+
+
+class ServicePrefix(Enum):
+    """
+    AWS Service Prefixes - Enumerations of the supported service proxy types
+    URLs:
+        https://docs.aws.amazon.com/service-authorization/latest/reference/reference_policies_actions-resources-contextkeys.html
+    """
+
+    LATTICE = "vpc-lattice-svcs"
+    RESTAPI = "execute-api"
+    HTTPAPI = "apigateway"
+    APPSYNC = "appsync"
+
+
+class SigV4Auth:
+    """
+    Authenticating Requests (AWS Signature Version 4)
+    Requests that were signed with SigV4 will have SignatureVersion set to AWS4-HMAC-SHA256
+
+    Args:
+        url (str): URL
+        service (ServicePrefix): AWS service Prefix
+        region (str, Optional): AWS region
+        body (dict, optional): Request body
+        params (dict, optional): Request parameters
+        headers (dict, optional): Request headers
+        method (str, optional): Request method
+
+    Returns:
+        SigV4Auth: SigV4Auth instance
+
+    Examples
+    --------
+    >>> from aws_lambda_powertools.utilities.auth import SigV4Auth, ServicePrefix
+    >>> prepped = SigV4Auth.prepare_request(region="us-east-2", service=ServicePrefix.LATTICE, url="https://test-fake-service.vpc-lattice-svcs.us-east-2.on.aws")
+    """
+
+    @staticmethod
+    def prepare_request(
+        url: str,
+        service: ServicePrefix,
+        region: Optional[str],
+        body: Optional[dict] = None,
+        params: Optional[dict] = None,
+        headers: Optional[dict] = None,
+        method: Optional[str] = "GET",
+    ):
+        if region is None:
+            region = os.environ.get("AWS_REGION")
+
+        if body is not None:
+            body = json.dumps(body)
+        else:
+            body = json.dumps({})
+
+        credentials = botocore.session.Session().get_credentials()
+
+        signer = crt.auth.CrtSigV4Auth(credentials, service.value, region)
+
+        if headers is None:
+            headers = {"Content-Type": "application/json"}
+
+        request = AWSRequest(method=method, url=url, data=body, params=params, headers=headers)
+
+        if service.value == "vpc-lattice-svcs":
+            # payload signing is not supported for vpc-lattice-svcs
+            request.context["payload_signing_enabled"] = False
+
+        signer.add_auth(request)
+        return request.prepare()
+
+
+class SigV4aAuth:
+    """
+    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
+        service (ServicePrefix): AWS service Prefix
+        region (str, Optional): AWS region
+        body (dict, optional): Request body
+        params (dict, optional): Request parameters
+        headers (dict, optional): Request headers
+        method (str, optional): Request method
+
+    Returns:
+        SigV4aAuth: SigV4aAuth instance
+
+    Examples
+    --------
+    >>> from aws_lambda_powertools.utilities.iam import SigV4aAuth, ServicePrefix
+    >>> prepped = SigV4aAuth.prepare_request(region="us-east-2", service=ServicePrefix.LATTICE, url="https://test-fake-service.vpc-lattice-svcs.us-east-2.on.aws")
+    """
+
+    @staticmethod
+    def prepare_request(
+        url: str,
+        service: ServicePrefix,
+        region: Optional[str] = "*",
+        body: Optional[dict] = None,
+        params: Optional[dict] = None,
+        headers: Optional[dict] = None,
+        method: Optional[str] = "GET",
+    ):
+        if body is not None:
+            body = json.dumps(body)
+        else:
+            body = json.dumps({})
+
+        credentials = botocore.session.Session().get_credentials()
+
+        signer = crt.auth.CrtSigV4AsymAuth(credentials, service.value, region)
+
+        if headers is None:
+            headers = {"Content-Type": "application/json"}
+
+        request = AWSRequest(method=method, url=url, data=body, params=params, headers=headers)
+
+        if service.value == "vpc-lattice-svcs":
+            # payload signing is not supported for vpc-lattice-svcs
+            request.context["payload_signing_enabled"] = False
+
+        signer.add_auth(request)
+        return request.prepare()

mypy.ini

@@ -36,6 +36,9 @@ ignore_missing_imports = True
 [mypy-botocore.response]
 ignore_missing_imports = True
 
+[mypy-botocore.*]
+ignore_missing_imports = True
+
 [mypy-boto3.dynamodb.conditions]
 ignore_missing_imports = True
 

pyproject.toml

@@ -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 = [