feat(auth): add JWT verification and API Gateway authorization

aws_lambda_powertools/utilities/auth/__init__.py

@@ -0,0 +1,26 @@
+"""JWT access-token verification for AWS Lambda."""
+
+from __future__ import annotations
+
+import importlib
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.auth._middleware import AuthErrorContext as AuthErrorContext
+    from aws_lambda_powertools.utilities.auth.exceptions import AuthFailureReason as AuthFailureReason
+    from aws_lambda_powertools.utilities.auth.verifier import JWTVerifier as JWTVerifier
+
+__all__ = ["AuthErrorContext", "AuthFailureReason", "JWTVerifier"]
+
+
+def __getattr__(name: str) -> object:
+    modules = {"AuthErrorContext": "_middleware", "AuthFailureReason": "exceptions", "JWTVerifier": "verifier"}
+    if name in modules:
+        value = getattr(importlib.import_module(f"{__name__}.{modules[name]}"), name)
+        globals()[name] = value
+        return value
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(__all__))

aws_lambda_powertools/utilities/auth/_authorization.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from aws_lambda_powertools.utilities.auth._validation import string_list
+from aws_lambda_powertools.utilities.auth.exceptions import (
+    AuthError,
+    AuthFailureReason,
+    InvalidClaimsError,
+    InvalidTokenError,
+)
+
+
+class MissingTokenError(InvalidTokenError):
+    """No authorization header was supplied."""
+
+    reason = AuthFailureReason.MISSING_TOKEN
+
+
+class ForbiddenError(AuthError):
+    """A verified caller does not have permission for this operation."""
+
+    reason = AuthFailureReason.FORBIDDEN
+
+
+class InsufficientScopeError(ForbiddenError):
+    """A verified caller is missing a required scope."""
+
+    reason = AuthFailureReason.INSUFFICIENT_SCOPE
+
+
+def bearer_token(value: Any) -> str:
+    if value is None:
+        raise MissingTokenError()
+    if not isinstance(value, str):
+        raise InvalidTokenError()
+    parts = value.split()
+    if len(parts) != 2 or parts[0].lower() != "bearer":
+        raise InvalidTokenError()
+    return parts[1]
+
+
+def header_token(headers: Any, multi_value_headers: Any = None) -> str:
+    values = _authorization_values(headers)
+    multi_values = _authorization_values(multi_value_headers)
+    if multi_values:
+        entries = multi_values[0]
+        if not isinstance(entries, list) or len(entries) != 1:
+            raise InvalidTokenError()
+        if values and values[0] != entries[0]:
+            raise InvalidTokenError()
+        return bearer_token(entries[0])
+    return bearer_token(values[0] if values else None)
+
+
+def _authorization_values(headers: Any) -> list[Any]:
+    if headers is None:
+        return []
+    if not isinstance(headers, Mapping):
+        raise InvalidTokenError()
+    values = [value for name, value in headers.items() if isinstance(name, str) and name.lower() == "authorization"]
+    if len(values) > 1:
+        raise InvalidTokenError()
+    return values
+
+
+def valid_scope(value: str) -> bool:
+    return bool(value) and all(33 <= ord(character) <= 126 and character not in {'"', "\\"} for character in value)
+
+
+def required_scopes(scopes: list[str] | None) -> tuple[str, ...]:
+    values = string_list(scopes if scopes is not None else [])
+    if not all(valid_scope(value) for value in values):
+        raise ValueError("Scopes must be valid OAuth scope tokens")
+    return values
+
+
+def enforce_scopes(claims: dict[str, Any], expected: tuple[str, ...]) -> None:
+    value: Any = next((claims[name] for name in ("scope", "scp", "scopes") if name in claims), [])
+    if isinstance(value, str):
+        values = [part for part in value.split(" ") if part]
+    elif isinstance(value, list):
+        values = value
+    else:
+        raise InvalidClaimsError()
+    if any(not isinstance(scope, str) or not valid_scope(scope) for scope in values):
+        raise InvalidClaimsError()
+    if not set(expected).issubset(values):
+        raise InsufficientScopeError()

aws_lambda_powertools/utilities/auth/_authorizer.py

@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+import math
+import re
+from typing import TYPE_CHECKING, Any, Literal
+
+from aws_lambda_powertools.utilities.auth._authorization import (
+    ForbiddenError,
+    bearer_token,
+    enforce_scopes,
+    header_token,
+    required_scopes,
+)
+from aws_lambda_powertools.utilities.auth._validation import string_list
+from aws_lambda_powertools.utilities.auth.exceptions import AuthError, InvalidClaimsError, InvalidTokenError
+from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import APIGatewayAuthorizerResponseV2
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from aws_lambda_powertools.utilities.auth._base import Verifier
+
+_ARN = re.compile(r"arn:[a-z0-9-]+:execute-api:[a-z0-9-]+:\d{12}:[a-z0-9]+/[^/]+/[A-Z]+/.*")
+
+
+def authorize_event(
+    verifier: Verifier,
+    event: dict[str, Any] | DictWrapper,
+    scopes: list[str] | None,
+    response_format: Literal["iam", "simple"],
+    context_claims: list[str] | None,
+    on_error: Callable[[AuthError], None] | None,
+) -> dict[str, Any]:
+    raw = event.raw_event if isinstance(event, DictWrapper) else event
+    _validate_event(raw, response_format)
+    arn = _request_arn(raw) if response_format == "iam" else None
+    expected = required_scopes(scopes)
+    selected = string_list(context_claims if context_claims is not None else [])
+    if "claims" in selected:
+        raise ValueError("claims is reserved in API Gateway authorizer context")
+    claims = _verified_claims(verifier, raw, expected, on_error, require_principal=response_format == "iam")
+    context = _context(claims, selected) if claims is not None else {}
+    if response_format == "simple":
+        return APIGatewayAuthorizerResponseV2(authorize=claims is not None, context=context).asdict()
+    return _iam_response(claims, arn, context)
+
+
+def _validate_event(raw: dict[str, Any], response_format: str) -> None:
+    if not isinstance(raw, dict) or raw.get("type") not in ("TOKEN", "REQUEST"):
+        raise ValueError("An API Gateway TOKEN or REQUEST authorizer event is required")
+    if response_format not in ("iam", "simple"):
+        raise ValueError("response_format must be iam or simple")
+    if response_format == "simple" and (raw.get("version") != "2.0" or raw["type"] != "REQUEST"):
+        raise ValueError("Simple authorizer responses require HTTP API payload version 2.0")
+
+
+def _verified_claims(
+    verifier: Verifier,
+    raw: dict[str, Any],
+    expected: tuple[str, ...],
+    on_error: Callable[[AuthError], None] | None,
+    *,
+    require_principal: bool,
+) -> dict[str, Any] | None:
+    try:
+        candidate = verifier.verify(_token(raw))
+        enforce_scopes(candidate, expected)
+        if require_principal:
+            _validate_principal(candidate)
+        return candidate
+    except AuthError as error:
+        # The callback observes a public, sanitized error, including failures
+        # raised while the caller is already handling another exception.
+        error.__context__ = None
+        error.__cause__ = None
+        if on_error is not None:
+            on_error(error)
+        if isinstance(error, (InvalidTokenError, ForbiddenError)):
+            return None
+        raise
+
+
+def _token(raw: dict[str, Any]) -> str:
+    if raw["type"] == "TOKEN":
+        return bearer_token(raw.get("authorizationToken"))
+    return header_token(raw.get("headers"), raw.get("multiValueHeaders"))
+
+
+def _validate_principal(claims: dict[str, Any]) -> None:
+    if not isinstance(claims.get("sub"), str) or not claims["sub"].strip():
+        raise InvalidClaimsError()
+
+
+def _iam_response(claims: dict[str, Any] | None, arn: str | None, context: dict[str, Any]) -> dict[str, Any]:
+    # Preserve the exact supplied resource, including its partition and encoded
+    # path. Route builders normalize paths and cannot represent every ARN here.
+    result: dict[str, Any] = {
+        "principalId": claims["sub"] if claims is not None else "unauthorized",
+        "policyDocument": {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Action": "execute-api:Invoke",
+                    "Effect": "Allow" if claims is not None else "Deny",
+                    "Resource": [arn],
+                },
+            ],
+        },
+    }
+    if context:
+        result["context"] = context
+    return result
+
+
+def _request_arn(event: dict[str, Any]) -> str:
+    arn = event.get("routeArn") if event.get("version") == "2.0" else event.get("methodArn")
+    if (
+        not isinstance(arn, str)
+        or len(arn) > 512
+        or not _ARN.fullmatch(arn)
+        or any(character in arn for character in ("*", "?", "\r", "\n"))
+    ):
+        raise ValueError("A concrete API Gateway method or route ARN of at most 512 characters is required")
+    return arn
+
+
+def _context(claims: dict[str, Any], selected: tuple[str, ...]) -> dict[str, Any]:
+    context = {}
+    for name in selected:
+        value = claims.get(name)
+        if isinstance(value, (str, bool, int)) or isinstance(value, float) and math.isfinite(value):
+            context[name] = value
+    return context

aws_lambda_powertools/utilities/auth/_base.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any, Literal
+
+from aws_lambda_powertools.utilities.auth._errors import sanitize_errors
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from aws_lambda_powertools.event_handler import Response
+    from aws_lambda_powertools.utilities.auth._middleware import AuthErrorContext, AuthMiddleware
+    from aws_lambda_powertools.utilities.auth.exceptions import AuthError
+    from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class Verifier(ABC):
+    """Shared verification interface used by issuer-specific and routed verifiers."""
+
+    @abstractmethod
+    def verify(self, token: str) -> dict[str, Any]:
+        """Return verified claims or raise an Auth utility error."""
+
+    @abstractmethod
+    def prefetch(self) -> None:
+        """Populate remote key caches without accepting a token."""
+
+    def require(
+        self,
+        *,
+        scopes: list[str] | None = None,
+        authorize: Callable[[dict[str, Any]], bool] | None = None,
+        on_error: Callable[[AuthErrorContext], Response] | None = None,
+    ) -> AuthMiddleware:
+        """Create Event Handler middleware enforcing token validity and all scopes.
+
+        Successful verification stores claims in ``app.context["claims"]``
+        while the downstream middleware and handler execute. Claims are
+        removed when they return or raise.
+        Missing/invalid tokens return 401, missing permissions return 403, and
+        unavailable signing keys return 503. A custom error callback replaces
+        the response, never execution of the protected handler.
+
+        Parameters
+        ----------
+        scopes : list[str], optional
+            Every listed scope must be present in the token.
+        authorize : Callable, optional
+            Additional policy receiving verified claims; must return True.
+        on_error : Callable, optional
+            Receives status_code, headers, a fixed reason, and retryable, and
+            returns an Event Handler Response. No automatic logging is performed.
+
+        Examples
+        --------
+        ```python
+        @app.get("/orders", middlewares=[verifier.require(scopes=["orders:read"])])
+        def orders():
+            return {"subject": app.context["claims"]["sub"]}
+        ```
+        """
+        from aws_lambda_powertools.utilities.auth._middleware import AuthMiddleware
+
+        return AuthMiddleware(self, scopes, authorize, on_error)
+
+    @sanitize_errors
+    def authorize(
+        self,
+        event: dict[str, Any] | DictWrapper,
+        *,
+        scopes: list[str] | None = None,
+        response_format: Literal["iam", "simple"] = "iam",
+        context_claims: list[str] | None = None,
+        on_error: Callable[[AuthError], None] | None = None,
+    ) -> dict[str, Any]:
+        """Return an API Gateway authorizer response for the current request.
+
+        IAM allows require a nonempty ``sub`` and target the supplied ARN only.
+        Simple responses require payload version 2.0 and must also be enabled
+        in the Gateway deployment. Disable Gateway result caching when each
+        request must be verified; this method cannot change Gateway's TTL.
+
+        Parameters
+        ----------
+        event : dict | DictWrapper
+            REST TOKEN/REQUEST or HTTP REQUEST authorizer event.
+        scopes : list[str], optional
+            Every listed scope must be present in the token.
+        response_format : Literal["iam", "simple"]
+            Response format configured in Gateway, by default iam.
+        context_claims : list[str], optional
+            Selected scalar claims to include; no claims are copied by default.
+        on_error : Callable, optional
+            Records a failure using the error's fixed reason and retryable fields.
+            Its return value is ignored: invalid credentials still deny access,
+            and unavailable keys still raise JWKSFetchError. Callback exceptions
+            fail the invocation. No automatic logging is performed.
+
+        Examples
+        --------
+        ```python
+        return verifier.authorize(
+            event, scopes=["orders:read"], response_format="iam", context_claims=["sub"],
+        )
+        ```
+        """
+        from aws_lambda_powertools.utilities.auth._authorizer import authorize_event
+
+        return authorize_event(self, event, scopes, response_format, context_claims, on_error)

aws_lambda_powertools/utilities/auth/_deadline.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import time
+
+from aws_lambda_powertools.utilities.auth._validation import finite_seconds
+
+
+class RequestError(Exception):
+    """Internal, credential-free transport failure."""
+
+    def __init__(self, *, retryable: bool = False) -> None:
+        self.retryable = retryable
+        super().__init__("Authentication endpoint request failed")
+
+
+class Deadline:
+    """One monotonic budget shared across a fetch and any subsequent requests."""
+
+    def __init__(self, seconds: float) -> None:
+        self._expires_at = time.monotonic() + finite_seconds(seconds, positive=True)
+
+    def remaining(self) -> float:
+        remaining = self._expires_at - time.monotonic()
+        if remaining <= 0:
+            raise RequestError(retryable=True)
+        return remaining