Auth (alpha)

CLASS	DESCRIPTION
JWTVerifier	Verify JWT access tokens for a configured issuer and resource audience.

JWTVerifier

JWTVerifier(*, issuer: str, audience: str | list[str], algorithms: list[str], jwks: dict[str, Any] | None = None, jwks_uri: str | None = None, required_claims: list[str] | None = None, expected_claims: Mapping[str, str] | None = None, expected_headers: Mapping[str, str] | None = None, clock_skew_seconds: float = 60, timeout_seconds: float = 3, jwks_max_age_seconds: float = 300, unknown_kid_cooldown_seconds: float = 300)

Bases: Verifier

Verify JWT access tokens for a configured issuer and resource audience.

PARAMETER	DESCRIPTION
issuer	Exact trusted HTTPS issuer. Discovery must advertise this issuer. TYPE: str
audience	Accepted resource audiences; at least one must match the token. TYPE: str | list[str]
algorithms	Explicit allowlist of asymmetric signing algorithms. TYPE: list[str]
jwks	Static key-set snapshot. Its rotation is the application's responsibility. TYPE: dict DEFAULT: None
jwks_uri	HTTPS key-set endpoint, mutually exclusive with jwks. Without either, discover keys from the configured issuer. TYPE: str DEFAULT: None
required_claims	Claims required in addition to iss, aud, and exp. TYPE: list[str] DEFAULT: None
expected_claims	Exact, case-sensitive string values required in verified claims, for example {"token_use": "access"}. Missing values are rejected. TYPE: Mapping[str, str] DEFAULT: None
expected_headers	Exact string values required in the signed header, for example {"typ": "at+jwt"}. These checks cannot weaken signature validation. TYPE: Mapping[str, str] DEFAULT: None
clock_skew_seconds	Nonnegative allowance for temporal claims, by default 60. TYPE: float DEFAULT: 60
timeout_seconds	Positive discovery/key-fetch and refresh-wait budget, by default 3. TYPE: float DEFAULT: 3
jwks_max_age_seconds	Positive maximum lifetime of fetched keys, by default 300. TYPE: float DEFAULT: 300
unknown_kid_cooldown_seconds	Nonnegative interval between unknown-key refreshes, by default 300. TYPE: float DEFAULT: 300

RAISES	DESCRIPTION
ValueError	Configuration is invalid or weakens the required verification profile.

Examples:

verifier = JWTVerifier(
    issuer="https://idp.example.com/",
    audience="https://orders.example.com",
    algorithms=["RS256"],
    required_claims=["sub"],
)
claims = verifier.verify(token)

METHOD	DESCRIPTION
any_of	Route an untrusted issuer claim only to explicitly configured verifiers.
authorize	Return an API Gateway authorizer response for the current request.
cognito	Verify resource-bound Cognito access tokens, never Cognito ID tokens.
prefetch	Populate an absent or expired remote key set; static keys need no I/O.
require	Create Event Handler middleware enforcing token validity and all scopes.
verify	Return verified access-token claims.
verify_authorization_header	Verify the JWT carried by an HTTP Authorization header.

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/verifier.py

def __init__(
    self,
    *,
    issuer: str,
    audience: str | list[str],
    algorithms: list[str],
    jwks: dict[str, Any] | None = None,
    jwks_uri: str | None = None,
    required_claims: list[str] | None = None,
    expected_claims: Mapping[str, str] | None = None,
    expected_headers: Mapping[str, str] | None = None,
    clock_skew_seconds: float = 60,
    timeout_seconds: float = 3,
    jwks_max_age_seconds: float = 300,
    unknown_kid_cooldown_seconds: float = 300,
) -> None:
    self._issuer = https_url(issuer, issuer=True)
    self._audience = string_list([audience] if isinstance(audience, str) else audience, nonempty=True)
    self._algorithms = string_list(algorithms, nonempty=True)
    if not set(self._algorithms) <= _ASYMMETRIC_ALGORITHMS:
        raise ValueError("Only asymmetric JWT signing algorithms are supported")
    if jwks is not None and jwks_uri is not None:
        raise ValueError("jwks and jwks_uri are mutually exclusive")
    self._jwks = copy_key_set(jwks) if jwks is not None else None
    self._jwks_uri = https_url(jwks_uri) if jwks_uri is not None else None
    self._timeout = finite_seconds(timeout_seconds, positive=True)
    max_age = finite_seconds(jwks_max_age_seconds, positive=True)
    cooldown = finite_seconds(unknown_kid_cooldown_seconds)
    self._cache = shared_cache(self._issuer, self._jwks_uri, max_age, cooldown) if jwks is None else None
    additional_claims = string_list(required_claims if required_claims is not None else [])
    self._required_claims = sorted({"iss", "aud", "exp"} | set(additional_claims))
    self._expected_claims = string_mapping(expected_claims)
    self._expected_headers = string_mapping(expected_headers)
    self._clock_skew = finite_seconds(clock_skew_seconds)
    self._cognito_client_id: str | None = None

any_of classmethod

any_of(*verifiers: JWTVerifier) -> Verifier

Route an untrusted issuer claim only to explicitly configured verifiers.

Unknown issuers trigger no discovery. Duplicate issuer configurations are rejected. The returned verifier has the same verification, middleware, authorizer, and prefetch interface.

Examples:

combined = JWTVerifier.any_of(corporate_verifier, cognito_verifier)
claims = combined.verify(token)

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/verifier.py

@classmethod
def any_of(cls, *verifiers: JWTVerifier) -> Verifier:
    """Route an untrusted issuer claim only to explicitly configured verifiers.

    Unknown issuers trigger no discovery. Duplicate issuer configurations
    are rejected. The returned verifier has the same verification,
    middleware, authorizer, and prefetch interface.

    Examples
    --------
    ```python
    combined = JWTVerifier.any_of(corporate_verifier, cognito_verifier)
    claims = combined.verify(token)
    ```
    """
    if not verifiers or any(not isinstance(verifier, JWTVerifier) for verifier in verifiers):
        raise ValueError("At least one issuer-specific JWTVerifier is required")
    issuers = {verifier._issuer: verifier for verifier in verifiers}
    if len(issuers) != len(verifiers):
        raise ValueError("Duplicate issuer configurations are ambiguous")
    return _IssuerVerifier(issuers)

authorize

authorize(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.

PARAMETER	DESCRIPTION
event	REST TOKEN/REQUEST or HTTP REQUEST authorizer event. TYPE: dict | DictWrapper
scopes	Every listed scope must be present in the token. TYPE: list[str] DEFAULT: None
response_format	Response format configured in Gateway, by default iam. TYPE: Literal['iam', 'simple'] DEFAULT: 'iam'
context_claims	Selected scalar claims to include; no claims are copied by default. TYPE: list[str] DEFAULT: None
on_error	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. TYPE: Callable DEFAULT: None

Examples:

return verifier.authorize(
    event, scopes=["orders:read"], response_format="iam", context_claims=["sub"],
)

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/base.py

@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_alpha.jwt.integrations.api_gateway import authorize_event

    return authorize_event(self, event, scopes, response_format, context_claims, on_error)

cognito classmethod

cognito(*, user_pool_id: str, client_id: str, audience: str | list[str], **options: Any) -> JWTVerifier

Verify resource-bound Cognito access tokens, never Cognito ID tokens.

Additional keyword arguments configure caching, static keys and claim requirements in the same way as JWTVerifier.

PARAMETER	DESCRIPTION
user_pool_id	Cognito user pool identifier, including its Region. TYPE: str
client_id	App client identifier required in the client_id claim. TYPE: str
audience	Resource audience required in aud. Request resource binding when obtaining the access token. TYPE: str | list[str]

Examples:

verifier = JWTVerifier.cognito(
    user_pool_id="us-east-1_abc123",
    client_id="orders-client",
    audience="https://orders.example.com",
)

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/verifier.py

@classmethod
def cognito(
    cls,
    *,
    user_pool_id: str,
    client_id: str,
    audience: str | list[str],
    **options: Any,
) -> JWTVerifier:
    """Verify resource-bound Cognito access tokens, never Cognito ID tokens.

    Additional keyword arguments configure caching, static keys and claim
    requirements in the same way as ``JWTVerifier``.

    Parameters
    ----------
    user_pool_id : str
        Cognito user pool identifier, including its Region.
    client_id : str
        App client identifier required in the ``client_id`` claim.
    audience : str | list[str]
        Resource audience required in ``aud``. Request resource binding
        when obtaining the access token.

    Examples
    --------
    ```python
    verifier = JWTVerifier.cognito(
        user_pool_id="us-east-1_abc123",
        client_id="orders-client",
        audience="https://orders.example.com",
    )
    ```
    """
    if not isinstance(user_pool_id, str) or not re.fullmatch(
        r"[a-z]{2}(?:-[a-z]+)+-\d+_[A-Za-z0-9]+",
        user_pool_id,
    ):
        raise ValueError("A valid Cognito user pool ID is required")
    if not is_nonempty_string(client_id):
        raise ValueError("A nonempty Cognito app client ID is required")
    if {"issuer", "algorithms", "jwks_uri"} & options.keys():
        raise ValueError("Cognito issuer, algorithm and JWKS endpoint cannot be overridden")
    region = user_pool_id.split("_", 1)[0]
    domain = "amazonaws.com.cn" if region.startswith("cn-") else "amazonaws.com"
    issuer = f"https://cognito-idp.{region}.{domain}/{user_pool_id}"
    if options.get("jwks") is None:
        options["jwks_uri"] = issuer + "/.well-known/jwks.json"
    verifier = cls(issuer=issuer, audience=audience, algorithms=["RS256"], **options)
    verifier._cognito_client_id = client_id
    return verifier

prefetch

prefetch() -> None

Populate an absent or expired remote key set; static keys need no I/O.

RAISES	DESCRIPTION
JWKSFetchError	Trusted keys could not be fetched within the configured budget.

Examples:

verifier.prefetch()  # Optional initialization work outside the handler.

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/verifier.py

@sanitize_errors
def prefetch(self) -> None:
    """Populate an absent or expired remote key set; static keys need no I/O.

    Raises
    ------
    JWKSFetchError
        Trusted keys could not be fetched within the configured budget.

    Examples
    --------
    ```python
    verifier.prefetch()  # Optional initialization work outside the handler.
    ```
    """
    if self._cache is not None:
        self._cache.get_keys(None, Deadline(self._timeout))

require

require(*, 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.

PARAMETER	DESCRIPTION
scopes	Every listed scope must be present in the token. TYPE: list[str] DEFAULT: None
authorize	Additional policy receiving verified claims; must return True. TYPE: Callable DEFAULT: None
on_error	Receives status_code, headers, a fixed reason, and retryable, and returns an Event Handler Response. No automatic logging is performed. TYPE: Callable DEFAULT: None

Examples:

@app.get("/orders", middlewares=[verifier.require(scopes=["orders:read"])])
def orders():
    return {"subject": app.context["claims"]["sub"]}

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/base.py

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_alpha.jwt.integrations.event_handler import AuthMiddleware

    return AuthMiddleware(self, scopes, authorize, on_error)

verify

verify(token: str) -> dict[str, Any]

Return verified access-token claims.

PARAMETER	DESCRIPTION
token	JWT access token without the Bearer prefix. TYPE: str

RETURNS	DESCRIPTION
dict[str, Any]	Claims after signature, issuer, resource, and time validation.

RAISES	DESCRIPTION
InvalidTokenError	Token, key, signature, or required claims are invalid.
JWKSFetchError	Current trusted keys could not be obtained.

Examples:

claims = verifier.verify(token)
subject = claims["sub"]

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/verifier.py

@sanitize_errors
def verify(self, token: str) -> dict[str, Any]:
    """Return verified access-token claims.

    Parameters
    ----------
    token : str
        JWT access token without the ``Bearer`` prefix.

    Returns
    -------
    dict[str, Any]
        Claims after signature, issuer, resource, and time validation.

    Raises
    ------
    InvalidTokenError
        Token, key, signature, or required claims are invalid.
    JWKSFetchError
        Current trusted keys could not be obtained.

    Examples
    --------
    ```python
    claims = verifier.verify(token)
    subject = claims["sub"]
    ```
    """
    header = self._header(token)
    key = self._signing_key(header)
    try:
        claims = jwt.decode(
            token,
            key.key,
            algorithms=self._algorithms,
            issuer=self._issuer,
            audience=self._audience,
            options={
                "require": self._required_claims,
                "verify_exp": False,
                "verify_nbf": False,
                "verify_iat": False,
            },
        )
    except jwt.InvalidSignatureError:
        raise InvalidSignatureError() from None
    except (jwt.PyJWTError, TypeError, ValueError, OverflowError, RecursionError):
        raise InvalidClaimsError() from None
    self._validate_times(claims)
    self._validate_profile(claims, header)
    if self._cognito_client_id is not None:
        if claims.get("token_use") != "access" or claims.get("client_id") != self._cognito_client_id:
            raise InvalidClaimsError()
    return claims

verify_authorization_header

verify_authorization_header(value: str | None) -> dict[str, Any]

Verify the JWT carried by an HTTP Authorization header.

The scheme is case-insensitive. Missing headers, malformed values, and schemes other than Bearer raise InvalidTokenError.

PARAMETER	DESCRIPTION
value	Raw HTTP Authorization header value. TYPE: str | None

RETURNS	DESCRIPTION
dict[str, Any]	Claims after token verification.

RAISES	DESCRIPTION
InvalidTokenError	Header or token is missing, malformed, or invalid.
JWKSFetchError	Current trusted signing keys could not be obtained.

Examples:

authorization = event.get("headers", {}).get("authorization")
claims = verifier.verify_authorization_header(authorization)

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/_internal/base.py

@sanitize_errors
def verify_authorization_header(self, value: str | None) -> dict[str, Any]:
    """Verify the JWT carried by an HTTP Authorization header.

    The scheme is case-insensitive. Missing headers, malformed values, and
    schemes other than Bearer raise ``InvalidTokenError``.

    Parameters
    ----------
    value : str | None
        Raw HTTP Authorization header value.

    Returns
    -------
    dict[str, Any]
        Claims after token verification.

    Raises
    ------
    InvalidTokenError
        Header or token is missing, malformed, or invalid.
    JWKSFetchError
        Current trusted signing keys could not be obtained.

    Examples
    --------
    ```python
    authorization = event.get("headers", {}).get("authorization")
    claims = verifier.verify_authorization_header(authorization)
    ```
    """
    from aws_lambda_powertools.utilities.auth_alpha.jwt._internal.authorization import bearer_token

    return self.verify(bearer_token(value))

Mapped HTTP failure available to a route's custom error response callback.

Credential-free errors raised by the Auth utility.

CLASS	DESCRIPTION
AuthError	Base error with a fixed message that never includes credential material.
AuthFailureReason	Stable, credential-free reasons suitable for application logs and metrics.
InvalidClaimsError	A required claim is missing or a claim does not match the token profile.
InvalidSignatureError	The access token signature does not match the configured signing key.
InvalidTokenError	The bearer token could not be verified.
JWKSFetchError	Required signing keys could not be retrieved or refreshed.
TokenExpiredError	The access token has expired beyond the configured clock tolerance.

AuthError

AuthError()

Bases: Exception

Base error with a fixed message that never includes credential material.

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py

def __init__(self) -> None:
    super().__init__(self.message)

AuthFailureReason

Bases: str, Enum

Stable, credential-free reasons suitable for application logs and metrics.

InvalidClaimsError

InvalidClaimsError()

Bases: InvalidTokenError

A required claim is missing or a claim does not match the token profile.

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py

def __init__(self) -> None:
    super().__init__(self.message)

InvalidSignatureError

InvalidSignatureError()

Bases: InvalidTokenError

The access token signature does not match the configured signing key.

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py

def __init__(self) -> None:
    super().__init__(self.message)

InvalidTokenError

InvalidTokenError()

Bases: AuthError

The bearer token could not be verified.

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py

def __init__(self) -> None:
    super().__init__(self.message)

JWKSFetchError

JWKSFetchError()

Bases: AuthError

Required signing keys could not be retrieved or refreshed.

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py

def __init__(self) -> None:
    super().__init__(self.message)

TokenExpiredError

TokenExpiredError()

Bases: InvalidTokenError

The access token has expired beyond the configured clock tolerance.

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/exceptions.py

def __init__(self) -> None:
    super().__init__(self.message)

Helpers for application tests that intentionally bypass token verification.

FUNCTION	DESCRIPTION
mock_claims	Temporarily return supplied claims without cryptography or network calls.

mock_claims

mock_claims(verifier: Verifier, claims: dict[str, Any]) -> Iterator[None]

Temporarily return supplied claims without cryptography or network calls.

This helper bypasses the verifier's security checks. Use it only in application tests; retain separate tests for real token verification.

Examples:

with mock_claims(verifier, {"sub": "test-user", "scope": "orders:read"}):
    response = app.resolve(event, context)

Source code in aws_lambda_powertools/utilities/auth_alpha/jwt/testing.py

@contextmanager
def mock_claims(verifier: Verifier, claims: dict[str, Any]) -> Iterator[None]:
    """Temporarily return supplied claims without cryptography or network calls.

    This helper bypasses the verifier's security checks. Use it only in
    application tests; retain separate tests for real token verification.

    Examples
    --------
    ```python
    with mock_claims(verifier, {"sub": "test-user", "scope": "orders:read"}):
        response = app.resolve(event, context)
    ```
    """
    snapshot = copy.deepcopy(claims)
    with patch.object(verifier, "verify", side_effect=lambda token: copy.deepcopy(snapshot)):
        yield