@@ -0,0 +1,26 @@
+"""Alpha authentication and authorization utilities for AWS Lambda."""
+
+from __future__ import annotations
+
+import importlib
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.auth_alpha.jwt import AuthErrorContext as AuthErrorContext
+    from aws_lambda_powertools.utilities.auth_alpha.jwt import AuthFailureReason as AuthFailureReason
+    from aws_lambda_powertools.utilities.auth_alpha.jwt import JWTVerifier as JWTVerifier
+
+__all__ = ["AuthErrorContext", "AuthFailureReason", "JWTVerifier"]
+
+
+def __getattr__(name: str) -> object:
+    modules = {"AuthErrorContext": "jwt", "AuthFailureReason": "jwt", "JWTVerifier": "jwt"}
+    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__))
@@ -0,0 +1 @@
+
@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import time
+
+from aws_lambda_powertools.utilities.auth_alpha._internal.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
@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any
+
+import urllib3
+from urllib3.connection import HTTPConnection
+
+from aws_lambda_powertools.utilities.auth_alpha._internal.deadline import Deadline, RequestError
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+_MAX_JSON_BYTES = 1024 * 1024
+
+
+class HTTPClient:
+    """HTTPS transport with bounded JSON responses and no implicit redirects/retries."""
+
+    def __init__(self) -> None:
+        self.pool = urllib3.PoolManager(cert_reqs="CERT_REQUIRED")
+
+    def json_request(
+        self,
+        method: str,
+        url: str,
+        deadline: Deadline,
+        *,
+        body: bytes | None = None,
+        headers: Mapping[str, str] | None = None,
+    ) -> tuple[int, dict[str, Any]]:
+        response = None
+        try:
+            response = self.pool.request(
+                method,
+                url,
+                body=body,
+                headers=headers,
+                timeout=urllib3.Timeout(total=deadline.remaining()),
+                retries=False,
+                redirect=False,
+                preload_content=False,
+            )
+            if response.status != 200:
+                deadline.remaining()
+                return response.status, {}
+            data = self._read_json(response, deadline)
+            return response.status, data
+        except (urllib3.exceptions.HTTPError, OSError):
+            raise RequestError(retryable=True) from None
+        finally:
+            if response is not None:
+                response.close()
+                response.release_conn()
+
+    @staticmethod
+    def _read_json(response: urllib3.response.BaseHTTPResponse, deadline: Deadline) -> dict[str, Any]:
+        chunks = bytearray()
+        while True:
+            remaining = deadline.remaining()
+            connection = response.connection
+            if isinstance(connection, HTTPConnection) and connection.sock is not None:
+                connection.sock.settimeout(remaining)
+            chunk = response.read1(min(65536, _MAX_JSON_BYTES + 1 - len(chunks)), decode_content=False)
+            deadline.remaining()
+            if not chunk:
+                break
+            chunks.extend(chunk)
+            if len(chunks) > _MAX_JSON_BYTES:
+                raise RequestError()
+        try:
+            data = json.loads(chunks)
+        except (ValueError, UnicodeError, RecursionError):
+            raise RequestError() from None
+        if not isinstance(data, dict):
+            raise RequestError()
+        return data
@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import math
+from collections.abc import Mapping
+from typing import Any
+from urllib.parse import urlsplit
+
+
+def https_url(value: str, *, issuer: bool = False) -> str:
+    """Validate configured URLs without echoing their contents in errors."""
+    try:
+        parts = urlsplit(value)
+        valid = isinstance(value, str) and all(
+            (
+                _valid_url_characters(value),
+                parts.scheme == "https",
+                bool(parts.hostname),
+                parts.username is None,
+                parts.password is None,
+                not parts.fragment,
+                not issuer or not parts.query,
+            ),
+        )
+        _ = parts.port  # Accessing the property validates a supplied port.
+    except (AttributeError, TypeError, ValueError):
+        valid = False
+    if not valid:
+        raise ValueError("An HTTPS URL without user information or a fragment is required") from None
+    return value
+
+
+def _valid_url_characters(value: str) -> bool:
+    return not any(character.isspace() or ord(character) < 32 for character in value)
+
+
+def finite_seconds(value: float, *, positive: bool = False) -> float:
+    """Validate a duration; booleans and non-finite values are not durations."""
+    try:
+        valid = type(value) in (int, float) and math.isfinite(value) and value >= 0 and (not positive or value > 0)
+    except OverflowError:
+        valid = False
+    if not valid:
+        message = "A finite positive duration is required" if positive else "A finite nonnegative duration is required"
+        raise ValueError(message)
+    return value
+
+
+def string_list(values: list[str] | tuple[str, ...], *, nonempty: bool = False) -> tuple[str, ...]:
+    """Copy a sequence of nonempty strings so configuration cannot be mutated."""
+    if not isinstance(values, (list, tuple)) or (nonempty and not values):
+        raise ValueError("A list of nonempty strings is required")
+    if not all(is_nonempty_string(value) for value in values):
+        raise ValueError("A list of nonempty strings is required")
+    return tuple(dict.fromkeys(values))
+
+
+def is_nonempty_string(value: Any) -> bool:
+    return isinstance(value, str) and bool(value.strip())
+
+
+def string_mapping(values: Mapping[str, str] | None) -> dict[str, str]:
+    """Copy exact token-profile constraints without exposing their contents."""
+    if values is None:
+        return {}
+    if not isinstance(values, Mapping) or not all(
+        is_nonempty_string(name) and is_nonempty_string(value) for name, value in values.items()
+    ):
+        raise ValueError("Expected claims and headers must map nonempty strings to nonempty strings")
+    return dict(values)
@@ -0,0 +1,32 @@
+"""JWT access-token verification."""
+
+from __future__ import annotations
+
+import importlib
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.auth_alpha.jwt.exceptions import AuthFailureReason as AuthFailureReason
+    from aws_lambda_powertools.utilities.auth_alpha.jwt.integrations.event_handler import (
+        AuthErrorContext as AuthErrorContext,
+    )
+    from aws_lambda_powertools.utilities.auth_alpha.jwt.verifier import JWTVerifier as JWTVerifier
+
+__all__ = ["AuthErrorContext", "AuthFailureReason", "JWTVerifier"]
+
+
+def __getattr__(name: str) -> object:
+    modules = {
+        "AuthErrorContext": "integrations.event_handler",
+        "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__))
@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from aws_lambda_powertools.utilities.auth_alpha._internal.validation import string_list
+from aws_lambda_powertools.utilities.auth_alpha.jwt.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()