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 JWT verification.

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/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/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/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/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/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/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

CLASS	DESCRIPTION
OAuth2Client	Acquire bearer tokens using client credentials for one configured resource.

OAuth2Client

OAuth2Client(*, token_url: str, client_id: str, client_secret: str | Callable[[], str], auth_method: Literal['client_secret_basic', 'client_secret_post'] = 'client_secret_basic', scopes: list[str] | None = None, audience: str | None = None, resource: str | None = None, timeout_seconds: float = 3)

Acquire bearer tokens using client credentials for one configured resource.

PARAMETER	DESCRIPTION
token_url	Trusted HTTPS OAuth token endpoint. TYPE: str
client_id	OAuth client identifier. TYPE: str
client_secret	Secret or loader invoked for each exchange attempt. TYPE: str | Callable[[], str]
auth_method	Client authentication method, by default client_secret_basic. client_secret_post sends credentials in the form body without an Authorization header. The client never switches methods automatically. TYPE: Literal['client_secret_basic', 'client_secret_post'] DEFAULT: 'client_secret_basic'
scopes	Scopes requested on every exchange. TYPE: list[str] DEFAULT: None
audience	Provider-specific audience request field, mutually exclusive with resource. TYPE: str DEFAULT: None
resource	RFC 8707 resource request field, mutually exclusive with audience. Must be an absolute URI without a fragment; URNs are supported. TYPE: str DEFAULT: None
timeout_seconds	Positive acquisition budget including retries, by default 3. TYPE: float DEFAULT: 3

Notes

Instances do not share tokens. Tokens are reacquired 30 seconds before expiration. Short-lived tokens and tokens without a lifetime are not cached. Configure timeouts on application-provided secret loaders.

Examples:

client = OAuth2Client(
    token_url="https://idp.example.com/token",
    client_id="orders",
    client_secret=load_secret,
    resource="https://inventory.example.com",
    scopes=["inventory:read"],
)
headers = client.auth_headers()

METHOD	DESCRIPTION
auth_headers	Return an Authorization header for this client's configured resource.
request	Send a synchronous HTTPS request using this resource's bearer token.

Source code in aws_lambda_powertools/utilities/auth_alpha/oauth2/client.py

def __init__(
    self,
    *,
    token_url: str,
    client_id: str,
    client_secret: str | Callable[[], str],
    auth_method: Literal["client_secret_basic", "client_secret_post"] = "client_secret_basic",
    scopes: list[str] | None = None,
    audience: str | None = None,
    resource: str | None = None,
    timeout_seconds: float = 3,
) -> None:
    self._token_url = https_url(token_url)
    if not is_nonempty_string(client_id):
        raise ValueError("A nonempty OAuth client ID is required")
    if not callable(client_secret) and (not isinstance(client_secret, str) or not client_secret):
        raise ValueError("client_secret must be a nonempty string or a callable")
    if auth_method not in ("client_secret_basic", "client_secret_post"):
        raise ValueError("auth_method must be client_secret_basic or client_secret_post")
    if audience is not None and resource is not None:
        raise ValueError("audience and resource are mutually exclusive")
    self._client_id = client_id
    self._client_secret = client_secret
    self._auth_method = auth_method
    self._scopes = required_scopes(scopes)
    self._timeout = finite_seconds(timeout_seconds, positive=True)
    self._fields = {"grant_type": "client_credentials"}
    if self._scopes:
        self._fields["scope"] = " ".join(self._scopes)
    for name, value in (("audience", audience), ("resource", resource)):
        if value is not None:
            if not is_nonempty_string(value):
                raise ValueError("Resource selection must be a nonempty string")
            self._fields[name] = value
    try:
        self._client_id.encode("utf-8")
        self._body = urlencode(self._fields).encode()
    except UnicodeError:
        raise ValueError("OAuth configuration must contain valid UTF-8 strings") from None
    if resource is not None:
        self._validate_resource(resource)
    self._http = HTTPClient()
    self._cached_token: _AccessToken | None = None
    self._flight: _Exchange | None = None
    self._lock = threading.Lock()

auth_headers

auth_headers() -> dict[str, str]

Return an Authorization header for this client's configured resource.

RAISES	DESCRIPTION
TokenExchangeError	A usable bearer token could not be obtained within the budget.

Examples:

headers = client.auth_headers()
response = http.request("GET", trusted_inventory_url, headers=headers)

Source code in aws_lambda_powertools/utilities/auth_alpha/oauth2/client.py

@sanitize_errors
def auth_headers(self) -> dict[str, str]:
    """Return an Authorization header for this client's configured resource.

    Raises
    ------
    TokenExchangeError
        A usable bearer token could not be obtained within the budget.

    Examples
    --------
    ```python
    headers = client.auth_headers()
    response = http.request("GET", trusted_inventory_url, headers=headers)
    ```
    """
    try:
        token = self._get_token(Deadline(self._timeout))
    except RequestError as error:
        raise TokenExchangeError(retryable=error.retryable) from None
    return {"Authorization": f"Bearer {token.value}"}

request

request(method: str, url: str, *, timeout: float = 5, headers: Mapping[str, str] | None = None, **options: Any) -> urllib3.response.BaseHTTPResponse

Send a synchronous HTTPS request using this resource's bearer token.

Only trusted destination URLs should be supplied. Redirects and retries are disabled, and an existing Authorization header is rejected. body, fields, json, encode_multipart and multipart_boundary are forwarded to urllib3.

PARAMETER	DESCRIPTION
method	HTTP method. TYPE: str
url	Trusted HTTPS destination for this resource's credentials. TYPE: str
timeout	Positive downstream timeout, separate from acquisition, by default 5. TYPE: float DEFAULT: 5
headers	Additional headers with HTTP token names, excluding Authorization. TYPE: Mapping[str, str] DEFAULT: None

RETURNS	DESCRIPTION
BaseHTTPResponse	Downstream response; inspect its status before consuming its body.

RAISES	DESCRIPTION
TokenExchangeError	Token acquisition failed.
DownstreamRequestError	Downstream transport failed; the request is never replayed automatically.
ValueError	Request configuration is invalid.

Examples:

response = client.request("GET", "https://inventory.example.com/items")
if response.status == 200:
    items = response.json()

Source code in aws_lambda_powertools/utilities/auth_alpha/oauth2/client.py

@sanitize_errors
def request(
    self,
    method: str,
    url: str,
    *,
    timeout: float = 5,
    headers: Mapping[str, str] | None = None,
    **options: Any,
) -> urllib3.response.BaseHTTPResponse:
    """Send a synchronous HTTPS request using this resource's bearer token.

    Only trusted destination URLs should be supplied. Redirects and retries
    are disabled, and an existing Authorization header is rejected.
    ``body``, ``fields``, ``json``, ``encode_multipart`` and
    ``multipart_boundary`` are forwarded to urllib3.

    Parameters
    ----------
    method : str
        HTTP method.
    url : str
        Trusted HTTPS destination for this resource's credentials.
    timeout : float
        Positive downstream timeout, separate from acquisition, by default 5.
    headers : Mapping[str, str], optional
        Additional headers with HTTP token names, excluding Authorization.

    Returns
    -------
    urllib3.response.BaseHTTPResponse
        Downstream response; inspect its status before consuming its body.

    Raises
    ------
    TokenExchangeError
        Token acquisition failed.
    DownstreamRequestError
        Downstream transport failed; the request is never replayed automatically.
    ValueError
        Request configuration is invalid.

    Examples
    --------
    ```python
    response = client.request("GET", "https://inventory.example.com/items")
    if response.status == 200:
        items = response.json()
    ```
    """
    target = https_url(url)
    duration = finite_seconds(timeout, positive=True)
    allowed = {"body", "fields", "json", "encode_multipart", "multipart_boundary"}
    if not options.keys() <= allowed:
        raise ValueError("Unsupported authenticated request option")
    if not isinstance(method, str) or not re.fullmatch(r"[A-Za-z]+", method):
        raise ValueError("A valid HTTP method is required")
    request_headers = self._request_headers(headers)
    request_headers.update(self.auth_headers())
    deadline = Deadline(duration)
    try:
        return self._http.request(
            method.upper(),
            target,
            deadline,
            headers=request_headers,
            **options,
        )
    except (urllib3.exceptions.HTTPError, OSError, ValueError, TypeError, RequestError):
        raise DownstreamRequestError() from None

Credential-free errors raised by outbound OAuth requests.

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.
DownstreamRequestError	The authenticated HTTP operation could not complete.
TokenExchangeError	A usable bearer token could not be obtained.

AuthError

AuthError()

Bases: Exception

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

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

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

AuthFailureReason

Bases: str, Enum

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

DownstreamRequestError

DownstreamRequestError()

Bases: AuthError

The authenticated HTTP operation could not complete.

The server may already have performed the operation. Callers must decide whether replay is safe; this error does not advise automatic retries.

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

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

TokenExchangeError

TokenExchangeError(*, retryable: bool = False)

Bases: AuthError

A usable bearer token could not be obtained.

retryable indicates a transient endpoint failure or acquisition timeout. Invalid responses, rejected credentials, and secret-loader failures are not retried automatically. Exception messages never include provider details.

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

def __init__(self, *, retryable: bool = False) -> None:
    self.retryable = retryable
    super().__init__()