API Reference

HTTP route classification

The paths below use the default RouterPrefixes and are relative to the path where the host mounts create_auth_router(). A host may override the prefixes, so deployed absolute paths can differ.

JAFAAL's registered clients are trusted, statically configured, first-party public clients. Calling a route an OAuth endpoint describes its wire contract; it does not make JAFAAL a third-party authorization service. The OIDC RP routes communicate with an upstream identity provider. Every other route is a JAFAAL extension, even when it implements a standard such as WebAuthn internally.

OAuth authorization-server endpoints

Method	Path	Contract
GET	/auth/authorize	RFC 6749 authorization endpoint; authorization code only, with PKCE S256 required
POST	/auth/token	RFC 6749 token endpoint; authorization-code and refresh-token grants
POST	/auth/introspect	RFC 7662 token introspection
POST	/auth/revoke	RFC 7009 token revocation
GET	/.well-known/oauth-authorization-server	RFC 8414 authorization-server metadata
GET	/.well-known/jwks.json	RFC 7517 JSON Web Key Set used to verify asymmetric tokens

Upstream OIDC relying-party integration

Method	Path	Contract
GET	/public/idp/callback/{idp_slug}	OAuth/OIDC callback from a configured upstream provider
POST	/auth/idp/step-up/reauth/{idp_id}	JAFAAL extension that starts fresh upstream OIDC authentication for step-up

These routes are part of JAFAAL's role as an OAuth client and OpenID Connect Relying Party. They are not OpenID Provider endpoints.

JAFAAL extension endpoints

Methods	Paths	Purpose
POST	/auth/login	First-party password login; not the OAuth resource-owner password grant
POST	/auth/mfa/verify	Complete a pending login with TOTP or a backup code
POST	/auth/refresh	Native cookie/header alias for the refresh-token grant; OAuth clients use /auth/token
POST	/auth/logout	Delete the current JAFAAL session; not an OIDC logout endpoint
POST	/auth/password/change	Change the authenticated user's password after step-up
POST	/auth/password/renew	Replace a password marked as requiring change
POST	/auth/password/user/{user_id}	Administrative password reset
POST	/auth/password-reset/request, /auth/password-reset/confirm	Request and consume a password-reset token
POST	/auth/sign-up/request, /auth/sign-up/confirm	Create a local account and confirm its email token
GET	/auth/sessions/user/{user_id}	List a user's sessions
DELETE	/auth/sessions/{session_id}/user/{user_id}, /auth/sessions/user/{user_id}	Revoke one or all of a user's sessions
GET, POST	/auth/api-keys	List or create API keys
PATCH	/auth/api-keys/{api_key_id}/revoke	Revoke an API key while retaining its record
DELETE	/auth/api-keys/{api_key_id}	Delete an API key
GET	/public/idp	List enabled upstream identity providers for a login picker
GET, POST	/auth/idp	List or create configured identity providers
GET	/auth/idp/templates	List built-in identity-provider templates
PUT, DELETE	/auth/idp/{idp_id}	Update or delete an identity provider
POST	/auth/webauthn/register/begin, /auth/webauthn/register/complete	Register a passkey
GET	/auth/webauthn/credentials	List the authenticated user's passkeys
POST	/auth/webauthn/credentials/{credential_pk}/delete	Delete a passkey after step-up
POST	/auth/webauthn/mfa/begin, /auth/webauthn/mfa/complete	Complete a pending password login with a passkey second factor
POST	/public/webauthn/authenticate/begin, /public/webauthn/authenticate/complete	Passwordless passkey login

Login, MFA, passkey, password, signup, session, API-key, and administrative routes are not OAuth grants or endpoints. Their JAFAAL domain failures use the {"detail", "code"} contract; request-schema validation may use FastAPI's native HTTP 422 detail array.

OAuth protocol routes never use that FastAPI validation body. Missing, empty, malformed, non-text, and repeated query or form parameters use the OAuth {"error", "error_description"} shape. /auth/authorize renders that JSON without redirecting until one registered client and redirect URI validate; later errors are sent to that validated URI with error, error_description, iss, and an unambiguous state. See Client integration: OAuth protocol errors.

Python API

The curated public API is everything exported from the top-level jafaal package. Each symbol below is generated from its source docstring.

Authentication package.

Provides the FastAPI router, JWT token issuance and validation, password hashing, scope enforcement, API-key validation, and the progressive-lockout stores used during login and MFA verification.

Persistence-bearing concerns (identity providers, IdP link tokens, MFA backup codes, OAuth state) live in dedicated sub-packages and expose their own models, schemas, and CRUD modules.

Exports

• Password hashing: PasswordHasher, PasswordPolicyError, get_password_hasher
• JWT: TokenManager, TokenType, get_token_manager
• Security dependencies: AuthContext, oauth2_scheme, validate_access_token, check_auth_scopes, get_sub_from_access_token, get_sid_from_access_token, get_sub_from_refresh_token, get_sid_from_refresh_token, validate_access_token_or_api_key,

Scope enforcement (check_scopes) is provided by :mod:jafaal.dependencies, which resolves the full principal. - Schemas: MFALoginRequest, MFARequiredResponse, TokenResponseWeb, TokenResponseMobile, LogoutResponse - Stores: PendingMFALogin, FailedLoginAttempts, StepUpAttempts, get_pending_mfa_store, get_failed_login_attempts, get_step_up_attempts, cleanup_expired_pending_mfa_logins, clear_pending_mfa_for_user - Helpers: authenticate_user, complete_login, create_tokens - User model mixins: UserMixin, IntPKUserMixin, UUIDPKUserMixin (extensible base for the host app's user table)

PasswordHasher

PasswordHasher provides secure password hashing, verification, and password policy enforcement.

This class encapsulates password hashing logic, verification, and secure password generation according to strong password policies. It supports pluggable hashers and ensures that generated or validated passwords meet complexity requirements (uppercase, lowercase, digit, punctuation).

Attributes:

Name	Type	Description
UPPER	str	All uppercase ASCII letters.
LOWER	str	All lowercase ASCII letters.
DIGITS	str	All ASCII digits.
PUNCTUATION	str	All ASCII punctuation characters.
ALL	str	Combination of all allowed characters.

Methods:

Name	Description
__init__	Argon2Hasher | None = None): Initializes the PasswordHasher with an optional custom hasher.
hash_password	str) -> str: Hashes a plain text password using the configured password hashing algorithm.
verify_password	str, hashed_password: str) -> bool: Verifies if a plain password matches the given hashed password.
verify_and_update	str, hashed_password: str) -> tuple[bool, str | None]: Verifies a password and updates the hash if the algorithm or parameters have changed.
generate_password	int = 8) -> str: Generates a secure random password of specified length, ensuring complexity.
validate_password	str, min_length: int = 8) -> None: Validates that a password meets the required security policy, raising PasswordPolicyError if not.
is_valid_password	str, min_length: int = 8) -> bool: Checks if a password meets the specified minimum length and password policy requirements.

Example

try: PasswordHasher.validate_password("weak") except PasswordPolicyError as e: print("Oops:", e)

Source code in jafaal/_internal/password_hasher.py

class PasswordHasher:
    """
    PasswordHasher provides secure password hashing, verification, and password policy enforcement.

    This class encapsulates password hashing logic, verification, and secure password generation
    according to strong password policies. It supports pluggable hashers and ensures that generated
    or validated passwords meet complexity requirements (uppercase, lowercase, digit, punctuation).

    Attributes:
        UPPER (str): All uppercase ASCII letters.
        LOWER (str): All lowercase ASCII letters.
        DIGITS (str): All ASCII digits.
        PUNCTUATION (str): All ASCII punctuation characters.
        ALL (str): Combination of all allowed characters.

    Methods:
        __init__(hasher: Argon2Hasher | None = None):
            Initializes the PasswordHasher with an optional custom hasher.

        hash_password(password: str) -> str:
            Hashes a plain text password using the configured password hashing algorithm.

        verify_password(plain_password: str, hashed_password: str) -> bool:
            Verifies if a plain password matches the given hashed password.

        verify_and_update(plain_password: str, hashed_password: str) -> tuple[bool, str | None]:
            Verifies a password and updates the hash if the algorithm or parameters have changed.

        generate_password(length: int = 8) -> str:
            Generates a secure random password of specified length, ensuring complexity.

        validate_password(password: str, min_length: int = 8) -> None:
            Validates that a password meets the required security policy, raising PasswordPolicyError if not.

        is_valid_password(password: str, min_length: int = 8) -> bool:
            Checks if a password meets the specified minimum length and password policy requirements.

    Example:
        try:
            PasswordHasher.validate_password("weak")
        except PasswordPolicyError as e:
            print("Oops:", e)
    """

    # Character classes
    UPPER = string.ascii_uppercase
    LOWER = string.ascii_lowercase
    DIGITS = string.digits
    PUNCTUATION = string.punctuation
    ALL = UPPER + LOWER + DIGITS + PUNCTUATION

    def __init__(
        self,
        hasher: (Argon2Hasher | Iterable[object] | PasswordHash | None) = None,
    ):
        """
        Initialize the password hasher configuration.
        Args:
            hasher (Argon2Hasher | Iterable[object] | PasswordHash | None, optional):
                The hasher(s) to use for password hashing. Can be:
                - None: Uses the strongest recommended configuration.
                - PasswordHash: Uses the provided PasswordHash instance.
                - Argon2Hasher: Uses the single hasher instance.
                - Iterable: Uses a list of hasher instances.
        Raises:
            TypeError: If the provided hasher is not of a supported type.
        """

        if hasher is None:
            # Default: strongest recommended config
            self._password_hash = PasswordHash.recommended()
        elif isinstance(hasher, PasswordHash):
            # Already a PasswordHash instance
            self._password_hash = hasher
        elif isinstance(hasher, Argon2Hasher):
            # Single hasher instance
            self._password_hash = PasswordHash([hasher])
        elif isinstance(hasher, Iterable):
            # Iterable of hashers
            self._password_hash = PasswordHash(cast(list[HasherProtocol], list(hasher)))
        else:
            raise TypeError(
                f"Unsupported hasher type: {type(hasher).__name__}. "
                "Must be Argon2Hasher, Iterable, PasswordHash, or None."
            )

        # Pre-compute the dummy hash now so dummy_verify() costs exactly one
        # verify on every call — including the first. Otherwise the first
        # "user not found" login would additionally pay the (deliberately slow)
        # hash and be measurably slower than the steady-state "found, wrong
        # password" branch, re-opening the username-enumeration timing side
        # channel that dummy_verify() exists to close.
        self._dummy_hash = self._password_hash.hash(secrets.token_urlsafe(32))

    def hash_password(self, password: str) -> str:
        """
        Hashes the provided password using the configured password hashing algorithm.

        The password is NFKC-normalized first (NIST SP 800-63B §5.1.1.2), so a
        passphrase enrolled on one platform verifies on another.

        Args:
            password (str): The plain text password to be hashed.

        Returns:
            str: The resulting hashed password.
        """
        return self._password_hash.hash(normalize_password(password))

    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        """
        Verifies whether the provided plain text password matches the given hashed password.

        Args:
            plain_password (str): The plain text password to verify.
            hashed_password (str): The hashed password to compare against.

        Returns:
            bool: True if the plain password matches the hashed password, False otherwise.
        """
        return self._password_hash.verify(normalize_password(plain_password), hashed_password)

    def verify_and_update(self, plain_password: str, hashed_password: str) -> tuple[bool, str | None]:
        """
        Verifies a plain password against a hashed password and updates the hash if necessary.

        Args:
            plain_password (str): The plain text password to verify.
            hashed_password (str): The hashed password to verify against.

        Returns:
            tuple[bool, str | None]: A tuple where the first element is a boolean indicating
            whether the password is correct, and the second element is the updated hash if
            the hash algorithm has changed or None otherwise.
        """
        return self._password_hash.verify_and_update(normalize_password(plain_password), hashed_password)

    def dummy_verify(self) -> None:
        """Run a constant-time-equivalent password verify against a dummy hash.

        Used by the login and MFA-verify endpoints on the "username/user
        not found" branch to equalise wall-clock latency with the
        "found, wrong password" branch. Without this, an unauthenticated
        attacker can enumerate valid usernames by measuring response
        time, because Argon2 is deliberately tuned to hundreds of
        milliseconds and a fast bail-out on the not-found branch is
        trivially distinguishable from a real verify.

        The dummy hash is pre-computed once at construction (see
        :meth:`__init__`), so this call always costs exactly one verify —
        the first invocation is not slower than steady state. A fresh
        random password is verified against it, so the result is always
        ``False``; the return value is ignored (the call exists purely
        for its timing side effect).
        """
        # Verify a random string against the pre-computed dummy hash: it is
        # guaranteed to return False but performs the full Argon2 verify work.
        self._password_hash.verify(secrets.token_urlsafe(32), self._dummy_hash)

    @staticmethod
    def generate_password(length: int = 8) -> str:
        """
        Generate a secure random password of specified length.

        The generated password will contain at least one uppercase letter, one lowercase letter,
        one digit, and one punctuation character to ensure complexity. The remaining characters
        are randomly selected from all allowed character sets.

        Args:
            length (int): The desired length of the password. Must be at least 8.

        Returns:
            str: A randomly generated password meeting the specified criteria.

        Raises:
            PasswordPolicyError: If the requested length is less than 8.
        """
        if length < 8:
            raise PasswordPolicyError(f"Requested length {length!r} is too short; must be ≥ 8.")

        # Guarantee at least one from each category
        chars = [
            secrets.choice(PasswordHasher.UPPER),
            secrets.choice(PasswordHasher.LOWER),
            secrets.choice(PasswordHasher.DIGITS),
            secrets.choice(PasswordHasher.PUNCTUATION),
        ]
        for _ in range(length - 4):
            chars.append(secrets.choice(PasswordHasher.ALL))
        secrets.SystemRandom().shuffle(chars)
        return "".join(chars)

    @staticmethod
    def validate_password(
        password: str,
        min_length: int = 8,
        policy_type: str = "strict",
        max_length: int | None = None,
    ) -> None:
        """
        Validates whether the given password meets the required security policy.

        Args:
            password (str): The password string to validate.
            min_length (int, optional): The minimum required length for the password. Defaults to 8.
            policy_type (str, optional): The password policy type to enforce.
                - "strict": Requires uppercase, lowercase, digit, and special character.
                - "length_only": Only enforces minimum/maximum length.
                Defaults to "strict".
            max_length (int | None, optional): The maximum accepted length. When
                provided, the password is rejected before hashing if it exceeds
                this bound. ``None`` (the default) enforces no maximum.

        Raises:
            PasswordPolicyError: If the password does not meet the policy requirements.

        Notes:
            - NIST SP 800-63B advises against imposing composition rules and in
              favour of length plus breach screening. ``"length_only"`` is the
              standards-aligned choice and the shipped default; pair it with a
              longer ``min_length`` and a breached-password check. ``"strict"``
              remains available for hosts bound by legacy composition
              requirements, but SP 800-63B-4 §3.1.1.2 states verifiers SHALL NOT
              impose them.
            - The password is never truncated: Argon2 accepts it whole, so
              ``max_length`` is the only upper bound and exists purely to cap
              hashing work.
        """
        if len(password) < min_length:
            raise PasswordPolicyError(f"Password is too short (got {len(password)}, need ≥ {min_length}).")

        # Bound the length before any (deliberately slow) hashing work and to keep
        # long passphrases supported (NIST SP 800-63B) without accepting unbounded
        # input.
        if max_length is not None and len(password) > max_length:
            raise PasswordPolicyError(f"Password is too long (got {len(password)}, allowed ≤ {max_length}).")

        # For length_only policy, only length is enforced
        if policy_type == "length_only":
            return

        # For strict policy, enforce complexity requirements
        if policy_type == "strict":
            if not any(c.isupper() for c in password):
                raise PasswordPolicyError("Password must contain at least one uppercase letter (A-Z).")

            if not any(c.islower() for c in password):
                raise PasswordPolicyError("Password must contain at least one lowercase letter (a-z).")

            if not any(c.isdigit() for c in password):
                raise PasswordPolicyError("Password must contain at least one digit (0-9).")

            if not any(c in PasswordHasher.PUNCTUATION for c in password):
                raise PasswordPolicyError(
                    f"Password must contain at least one special character ({PasswordHasher.PUNCTUATION})."
                )
        else:
            raise PasswordPolicyError(
                f"Unknown password policy type: {policy_type!r}. Supported types: 'strict', 'length_only'."
            )

    @staticmethod
    def is_valid_password(
        password: str,
        min_length: int = 8,
        policy_type: str = "strict",
        max_length: int | None = None,
    ) -> bool:
        """
        Checks if the provided password meets the specified minimum length and password policy requirements.

        Args:
            password (str): The password string to validate.
            min_length (int, optional): The minimum required length for the password. Defaults to 8.
            policy_type (str, optional): The password policy type to enforce. Defaults to "strict".
            max_length (int | None, optional): The maximum accepted length, or ``None`` for no maximum.

        Returns:
            bool: True if the password is valid according to the policy, False otherwise.
        """
        try:
            PasswordHasher.validate_password(password, min_length, policy_type, max_length)
            return True
        except PasswordPolicyError:
            return False

__init__

__init__(hasher=None)

Initialize the password hasher configuration. Args: hasher (Argon2Hasher | Iterable[object] | PasswordHash | None, optional): The hasher(s) to use for password hashing. Can be: - None: Uses the strongest recommended configuration. - PasswordHash: Uses the provided PasswordHash instance. - Argon2Hasher: Uses the single hasher instance. - Iterable: Uses a list of hasher instances. Raises: TypeError: If the provided hasher is not of a supported type.

Source code in jafaal/_internal/password_hasher.py

def __init__(
    self,
    hasher: (Argon2Hasher | Iterable[object] | PasswordHash | None) = None,
):
    """
    Initialize the password hasher configuration.
    Args:
        hasher (Argon2Hasher | Iterable[object] | PasswordHash | None, optional):
            The hasher(s) to use for password hashing. Can be:
            - None: Uses the strongest recommended configuration.
            - PasswordHash: Uses the provided PasswordHash instance.
            - Argon2Hasher: Uses the single hasher instance.
            - Iterable: Uses a list of hasher instances.
    Raises:
        TypeError: If the provided hasher is not of a supported type.
    """

    if hasher is None:
        # Default: strongest recommended config
        self._password_hash = PasswordHash.recommended()
    elif isinstance(hasher, PasswordHash):
        # Already a PasswordHash instance
        self._password_hash = hasher
    elif isinstance(hasher, Argon2Hasher):
        # Single hasher instance
        self._password_hash = PasswordHash([hasher])
    elif isinstance(hasher, Iterable):
        # Iterable of hashers
        self._password_hash = PasswordHash(cast(list[HasherProtocol], list(hasher)))
    else:
        raise TypeError(
            f"Unsupported hasher type: {type(hasher).__name__}. "
            "Must be Argon2Hasher, Iterable, PasswordHash, or None."
        )

    # Pre-compute the dummy hash now so dummy_verify() costs exactly one
    # verify on every call — including the first. Otherwise the first
    # "user not found" login would additionally pay the (deliberately slow)
    # hash and be measurably slower than the steady-state "found, wrong
    # password" branch, re-opening the username-enumeration timing side
    # channel that dummy_verify() exists to close.
    self._dummy_hash = self._password_hash.hash(secrets.token_urlsafe(32))

hash_password

hash_password(password)

Hashes the provided password using the configured password hashing algorithm.

The password is NFKC-normalized first (NIST SP 800-63B §5.1.1.2), so a passphrase enrolled on one platform verifies on another.

Parameters:

Name	Type	Description	Default
password	str	The plain text password to be hashed.	required

Returns:

Name	Type	Description
str	str	The resulting hashed password.

Source code in jafaal/_internal/password_hasher.py

def hash_password(self, password: str) -> str:
    """
    Hashes the provided password using the configured password hashing algorithm.

    The password is NFKC-normalized first (NIST SP 800-63B §5.1.1.2), so a
    passphrase enrolled on one platform verifies on another.

    Args:
        password (str): The plain text password to be hashed.

    Returns:
        str: The resulting hashed password.
    """
    return self._password_hash.hash(normalize_password(password))

verify_password

verify_password(plain_password, hashed_password)

Verifies whether the provided plain text password matches the given hashed password.

Parameters:

Name	Type	Description	Default
plain_password	str	The plain text password to verify.	required
hashed_password	str	The hashed password to compare against.	required

Returns:

Name	Type	Description
bool	bool	True if the plain password matches the hashed password, False otherwise.

Source code in jafaal/_internal/password_hasher.py

def verify_password(self, plain_password: str, hashed_password: str) -> bool:
    """
    Verifies whether the provided plain text password matches the given hashed password.

    Args:
        plain_password (str): The plain text password to verify.
        hashed_password (str): The hashed password to compare against.

    Returns:
        bool: True if the plain password matches the hashed password, False otherwise.
    """
    return self._password_hash.verify(normalize_password(plain_password), hashed_password)

verify_and_update

verify_and_update(plain_password, hashed_password)

Verifies a plain password against a hashed password and updates the hash if necessary.

Parameters:

Name	Type	Description	Default
plain_password	str	The plain text password to verify.	required
hashed_password	str	The hashed password to verify against.	required

Returns:

Type	Description
bool	tuple[bool, str | None]: A tuple where the first element is a boolean indicating
str | None	whether the password is correct, and the second element is the updated hash if
tuple[bool, str | None]	the hash algorithm has changed or None otherwise.

Source code in jafaal/_internal/password_hasher.py

def verify_and_update(self, plain_password: str, hashed_password: str) -> tuple[bool, str | None]:
    """
    Verifies a plain password against a hashed password and updates the hash if necessary.

    Args:
        plain_password (str): The plain text password to verify.
        hashed_password (str): The hashed password to verify against.

    Returns:
        tuple[bool, str | None]: A tuple where the first element is a boolean indicating
        whether the password is correct, and the second element is the updated hash if
        the hash algorithm has changed or None otherwise.
    """
    return self._password_hash.verify_and_update(normalize_password(plain_password), hashed_password)

dummy_verify

dummy_verify()

Run a constant-time-equivalent password verify against a dummy hash.

Used by the login and MFA-verify endpoints on the "username/user not found" branch to equalise wall-clock latency with the "found, wrong password" branch. Without this, an unauthenticated attacker can enumerate valid usernames by measuring response time, because Argon2 is deliberately tuned to hundreds of milliseconds and a fast bail-out on the not-found branch is trivially distinguishable from a real verify.

The dummy hash is pre-computed once at construction (see :meth:__init__), so this call always costs exactly one verify — the first invocation is not slower than steady state. A fresh random password is verified against it, so the result is always False; the return value is ignored (the call exists purely for its timing side effect).

Source code in jafaal/_internal/password_hasher.py

def dummy_verify(self) -> None:
    """Run a constant-time-equivalent password verify against a dummy hash.

    Used by the login and MFA-verify endpoints on the "username/user
    not found" branch to equalise wall-clock latency with the
    "found, wrong password" branch. Without this, an unauthenticated
    attacker can enumerate valid usernames by measuring response
    time, because Argon2 is deliberately tuned to hundreds of
    milliseconds and a fast bail-out on the not-found branch is
    trivially distinguishable from a real verify.

    The dummy hash is pre-computed once at construction (see
    :meth:`__init__`), so this call always costs exactly one verify —
    the first invocation is not slower than steady state. A fresh
    random password is verified against it, so the result is always
    ``False``; the return value is ignored (the call exists purely
    for its timing side effect).
    """
    # Verify a random string against the pre-computed dummy hash: it is
    # guaranteed to return False but performs the full Argon2 verify work.
    self._password_hash.verify(secrets.token_urlsafe(32), self._dummy_hash)

generate_password staticmethod

generate_password(length=8)

Generate a secure random password of specified length.

The generated password will contain at least one uppercase letter, one lowercase letter, one digit, and one punctuation character to ensure complexity. The remaining characters are randomly selected from all allowed character sets.

Parameters:

Name	Type	Description	Default
length	int	The desired length of the password. Must be at least 8.	8

Returns:

Name	Type	Description
str	str	A randomly generated password meeting the specified criteria.

Raises:

Type	Description
PasswordPolicyError	If the requested length is less than 8.

Source code in jafaal/_internal/password_hasher.py

@staticmethod
def generate_password(length: int = 8) -> str:
    """
    Generate a secure random password of specified length.

    The generated password will contain at least one uppercase letter, one lowercase letter,
    one digit, and one punctuation character to ensure complexity. The remaining characters
    are randomly selected from all allowed character sets.

    Args:
        length (int): The desired length of the password. Must be at least 8.

    Returns:
        str: A randomly generated password meeting the specified criteria.

    Raises:
        PasswordPolicyError: If the requested length is less than 8.
    """
    if length < 8:
        raise PasswordPolicyError(f"Requested length {length!r} is too short; must be ≥ 8.")

    # Guarantee at least one from each category
    chars = [
        secrets.choice(PasswordHasher.UPPER),
        secrets.choice(PasswordHasher.LOWER),
        secrets.choice(PasswordHasher.DIGITS),
        secrets.choice(PasswordHasher.PUNCTUATION),
    ]
    for _ in range(length - 4):
        chars.append(secrets.choice(PasswordHasher.ALL))
    secrets.SystemRandom().shuffle(chars)
    return "".join(chars)

validate_password staticmethod

validate_password(
    password,
    min_length=8,
    policy_type="strict",
    max_length=None,
)

Validates whether the given password meets the required security policy.

Parameters:

Name	Type	Description	Default
password	str	The password string to validate.	required
min_length	int	The minimum required length for the password. Defaults to 8.	8
policy_type	str	The password policy type to enforce. - "strict": Requires uppercase, lowercase, digit, and special character. - "length_only": Only enforces minimum/maximum length. Defaults to "strict".	'strict'
max_length	int | None	The maximum accepted length. When provided, the password is rejected before hashing if it exceeds this bound. None (the default) enforces no maximum.	None

Raises:

Type	Description
PasswordPolicyError	If the password does not meet the policy requirements.

Notes

• NIST SP 800-63B advises against imposing composition rules and in favour of length plus breach screening. "length_only" is the standards-aligned choice and the shipped default; pair it with a longer min_length and a breached-password check. "strict" remains available for hosts bound by legacy composition requirements, but SP 800-63B-4 §3.1.1.2 states verifiers SHALL NOT impose them.
• The password is never truncated: Argon2 accepts it whole, so max_length is the only upper bound and exists purely to cap hashing work.

Source code in jafaal/_internal/password_hasher.py

@staticmethod
def validate_password(
    password: str,
    min_length: int = 8,
    policy_type: str = "strict",
    max_length: int | None = None,
) -> None:
    """
    Validates whether the given password meets the required security policy.

    Args:
        password (str): The password string to validate.
        min_length (int, optional): The minimum required length for the password. Defaults to 8.
        policy_type (str, optional): The password policy type to enforce.
            - "strict": Requires uppercase, lowercase, digit, and special character.
            - "length_only": Only enforces minimum/maximum length.
            Defaults to "strict".
        max_length (int | None, optional): The maximum accepted length. When
            provided, the password is rejected before hashing if it exceeds
            this bound. ``None`` (the default) enforces no maximum.

    Raises:
        PasswordPolicyError: If the password does not meet the policy requirements.

    Notes:
        - NIST SP 800-63B advises against imposing composition rules and in
          favour of length plus breach screening. ``"length_only"`` is the
          standards-aligned choice and the shipped default; pair it with a
          longer ``min_length`` and a breached-password check. ``"strict"``
          remains available for hosts bound by legacy composition
          requirements, but SP 800-63B-4 §3.1.1.2 states verifiers SHALL NOT
          impose them.
        - The password is never truncated: Argon2 accepts it whole, so
          ``max_length`` is the only upper bound and exists purely to cap
          hashing work.
    """
    if len(password) < min_length:
        raise PasswordPolicyError(f"Password is too short (got {len(password)}, need ≥ {min_length}).")

    # Bound the length before any (deliberately slow) hashing work and to keep
    # long passphrases supported (NIST SP 800-63B) without accepting unbounded
    # input.
    if max_length is not None and len(password) > max_length:
        raise PasswordPolicyError(f"Password is too long (got {len(password)}, allowed ≤ {max_length}).")

    # For length_only policy, only length is enforced
    if policy_type == "length_only":
        return

    # For strict policy, enforce complexity requirements
    if policy_type == "strict":
        if not any(c.isupper() for c in password):
            raise PasswordPolicyError("Password must contain at least one uppercase letter (A-Z).")

        if not any(c.islower() for c in password):
            raise PasswordPolicyError("Password must contain at least one lowercase letter (a-z).")

        if not any(c.isdigit() for c in password):
            raise PasswordPolicyError("Password must contain at least one digit (0-9).")

        if not any(c in PasswordHasher.PUNCTUATION for c in password):
            raise PasswordPolicyError(
                f"Password must contain at least one special character ({PasswordHasher.PUNCTUATION})."
            )
    else:
        raise PasswordPolicyError(
            f"Unknown password policy type: {policy_type!r}. Supported types: 'strict', 'length_only'."
        )

is_valid_password staticmethod

is_valid_password(
    password,
    min_length=8,
    policy_type="strict",
    max_length=None,
)

Checks if the provided password meets the specified minimum length and password policy requirements.

Parameters:

Name	Type	Description	Default
password	str	The password string to validate.	required
min_length	int	The minimum required length for the password. Defaults to 8.	8
policy_type	str	The password policy type to enforce. Defaults to "strict".	'strict'
max_length	int | None	The maximum accepted length, or None for no maximum.	None

Returns:

Name	Type	Description
bool	bool	True if the password is valid according to the policy, False otherwise.

Source code in jafaal/_internal/password_hasher.py

@staticmethod
def is_valid_password(
    password: str,
    min_length: int = 8,
    policy_type: str = "strict",
    max_length: int | None = None,
) -> bool:
    """
    Checks if the provided password meets the specified minimum length and password policy requirements.

    Args:
        password (str): The password string to validate.
        min_length (int, optional): The minimum required length for the password. Defaults to 8.
        policy_type (str, optional): The password policy type to enforce. Defaults to "strict".
        max_length (int | None, optional): The maximum accepted length, or ``None`` for no maximum.

    Returns:
        bool: True if the password is valid according to the policy, False otherwise.
    """
    try:
        PasswordHasher.validate_password(password, min_length, policy_type, max_length)
        return True
    except PasswordPolicyError:
        return False

PasswordPolicyError

Bases: UnprocessableError

A password failed the configured policy (422).

Source code in jafaal/exceptions.py

class PasswordPolicyError(UnprocessableError):
    """A password failed the configured policy (422)."""

    code = "password_policy"
    default_detail = "The password does not meet the required policy."

FailedLoginAttempts

Track failed login attempts with progressive lockout.

Two independent dimensions guard the login endpoint:

• Per-username (5/10/20 → 5m/30m/24h) bounds brute-force against a single account. Because it keys on the username, anyone who knows a username can trip it — i.e. it is also a targeted-lockout (DoS) lever: an attacker can lock a known account out by submitting bad passwords for it. This is inherent to per-account lockout.
• Per-source-IP (50/100/250 → 15m/1h/24h) bounds how many accounts a single IP can lock out by spraying failures across many usernames, so the targeted-lockout lever above is not cheap at scale — an attacker must rotate IPs. It is reset on any successful login from the IP (so a busy shared egress rarely trips it) and gated by :attr:~jafaal.settings.AuthSettings.login_ip_lockout_enabled. It relies on an accurate client IP, so configure trusted_proxies behind a reverse proxy (otherwise every client shares the proxy's address).

Attributes:

Name	Type	Description
_state_override		Explicit provider (tests); None resolves the process-wide provider lazily at call time.
_lockout		Per-username progressive-lockout helper.
_ip_lockout		Per-source-IP progressive-lockout helper.

Source code in jafaal/_internal/security_stores.py

class FailedLoginAttempts:
    """
    Track failed login attempts with progressive lockout.

    Two independent dimensions guard the login endpoint:

    * **Per-username** (5/10/20 → 5m/30m/24h) bounds brute-force against a single
      account. Because it keys on the username, anyone who knows a username can
      trip it — i.e. it is also a *targeted-lockout* (DoS) lever: an attacker can
      lock a known account out by submitting bad passwords for it. This is
      inherent to per-account lockout.
    * **Per-source-IP** (50/100/250 → 15m/1h/24h) bounds how many accounts a
      single IP can lock out by *spraying* failures across many usernames, so the
      targeted-lockout lever above is not cheap at scale — an attacker must
      rotate IPs. It is reset on any successful login from the IP (so a busy
      shared egress rarely trips it) and gated by
      :attr:`~jafaal.settings.AuthSettings.login_ip_lockout_enabled`. It relies on
      an accurate client IP, so configure ``trusted_proxies`` behind a reverse
      proxy (otherwise every client shares the proxy's address).

    Attributes:
        _state_override: Explicit provider (tests); ``None`` resolves the
            process-wide provider lazily at call time.
        _lockout: Per-username progressive-lockout helper.
        _ip_lockout: Per-source-IP progressive-lockout helper.
    """

    def __init__(self, state: StateStore | None = None) -> None:
        self._state_override = state
        self._lockout = _ProgressiveLockout(
            self._get_state,
            name="login",
            display_name="Login",
            thresholds=_LOGIN_LOCKOUT_THRESHOLDS,
            attempts_ttl_seconds=24 * 60 * 60,
            normalize_key=normalize_username_key,
        )
        self._ip_lockout = _ProgressiveLockout(
            self._get_state,
            name="login_ip",
            display_name="Login (per-IP)",
            thresholds=_LOGIN_IP_LOCKOUT_THRESHOLDS,
            attempts_ttl_seconds=60 * 60,
            subject="ip",
        )

    def _get_state(self) -> StateStore:
        return self._state_override if self._state_override is not None else get_state_store()

    def _ip_lockout_enabled(self) -> bool:
        return jafaal_settings.get_settings().login_ip_lockout_enabled

    # --- per-username lockout ---
    def is_locked_out(self, username: str) -> bool:
        """Check if a username is locked out from failed logins."""
        return self._lockout.is_locked_out(username)

    def get_lockout_time(self, username: str) -> datetime | None:
        """Get the lockout expiry for a username, if locked out."""
        return self._lockout.get_lockout_time(username)

    def record_failed_attempt(self, username: str) -> int:
        """Record a failed login and return the current attempt count."""
        return self._lockout.record_failed_attempt(username)

    def reset_attempts(self, username: str) -> None:
        """Clear the failed-attempt counter on successful login."""
        self._lockout.reset_attempts(username)

    # --- per-source-IP backoff ---
    def is_ip_locked_out(self, ip: str) -> bool:
        """Check if a source IP is under the per-IP failed-login backoff."""
        if not self._ip_lockout_enabled():
            return False
        return self._ip_lockout.is_locked_out(ip)

    def get_ip_lockout_time(self, ip: str) -> datetime | None:
        """Get the per-IP backoff expiry for a source IP, if active."""
        if not self._ip_lockout_enabled():
            return None
        return self._ip_lockout.get_lockout_time(ip)

    def record_ip_failure(self, ip: str) -> int:
        """Record a failed login against the source IP; return the count (0 if disabled)."""
        if not self._ip_lockout_enabled():
            return 0
        return self._ip_lockout.record_failed_attempt(ip)

    def reset_ip_attempts(self, ip: str) -> None:
        """Deliberately a no-op — see below.

        The per-username counter is reset on a successful login because
        authenticating *as that user* proves the failures were that user
        fumbling their own password. No equivalent proof exists per IP: an
        address is shared by many accounts, so a success from it says nothing
        about the failures against other usernames.

        Resetting it made the per-IP tier — the only bound on the targeted
        lockout DoS the username tier enables — trivially defeatable: spray 49
        failures at victims, log in once to an account you own, repeat, and lock
        out arbitrarily many accounts from one address. The counter instead
        decays on its own ``attempts_ttl_seconds`` window, which is what keeps a
        legitimate NAT gateway from accumulating failures forever.

        Kept as a method (rather than removed) so the call site still reads as
        "success handling", with the reasoning attached to the behaviour.
        """
        return

    def clear_all(self) -> None:
        """Clear all failed-login records (per-username and per-IP)."""
        self._lockout.clear_all()
        self._ip_lockout.clear_all()

is_locked_out

is_locked_out(username)

Check if a username is locked out from failed logins.

Source code in jafaal/_internal/security_stores.py

def is_locked_out(self, username: str) -> bool:
    """Check if a username is locked out from failed logins."""
    return self._lockout.is_locked_out(username)

get_lockout_time

get_lockout_time(username)

Get the lockout expiry for a username, if locked out.

Source code in jafaal/_internal/security_stores.py

def get_lockout_time(self, username: str) -> datetime | None:
    """Get the lockout expiry for a username, if locked out."""
    return self._lockout.get_lockout_time(username)

record_failed_attempt

record_failed_attempt(username)

Record a failed login and return the current attempt count.

Source code in jafaal/_internal/security_stores.py

def record_failed_attempt(self, username: str) -> int:
    """Record a failed login and return the current attempt count."""
    return self._lockout.record_failed_attempt(username)

reset_attempts

reset_attempts(username)

Clear the failed-attempt counter on successful login.

Source code in jafaal/_internal/security_stores.py

def reset_attempts(self, username: str) -> None:
    """Clear the failed-attempt counter on successful login."""
    self._lockout.reset_attempts(username)

is_ip_locked_out

is_ip_locked_out(ip)

Check if a source IP is under the per-IP failed-login backoff.

Source code in jafaal/_internal/security_stores.py

def is_ip_locked_out(self, ip: str) -> bool:
    """Check if a source IP is under the per-IP failed-login backoff."""
    if not self._ip_lockout_enabled():
        return False
    return self._ip_lockout.is_locked_out(ip)

get_ip_lockout_time

get_ip_lockout_time(ip)

Get the per-IP backoff expiry for a source IP, if active.

Source code in jafaal/_internal/security_stores.py

def get_ip_lockout_time(self, ip: str) -> datetime | None:
    """Get the per-IP backoff expiry for a source IP, if active."""
    if not self._ip_lockout_enabled():
        return None
    return self._ip_lockout.get_lockout_time(ip)

record_ip_failure

record_ip_failure(ip)

Record a failed login against the source IP; return the count (0 if disabled).

Source code in jafaal/_internal/security_stores.py

def record_ip_failure(self, ip: str) -> int:
    """Record a failed login against the source IP; return the count (0 if disabled)."""
    if not self._ip_lockout_enabled():
        return 0
    return self._ip_lockout.record_failed_attempt(ip)

reset_ip_attempts

reset_ip_attempts(ip)

Deliberately a no-op — see below.

The per-username counter is reset on a successful login because authenticating as that user proves the failures were that user fumbling their own password. No equivalent proof exists per IP: an address is shared by many accounts, so a success from it says nothing about the failures against other usernames.

Resetting it made the per-IP tier — the only bound on the targeted lockout DoS the username tier enables — trivially defeatable: spray 49 failures at victims, log in once to an account you own, repeat, and lock out arbitrarily many accounts from one address. The counter instead decays on its own attempts_ttl_seconds window, which is what keeps a legitimate NAT gateway from accumulating failures forever.

Kept as a method (rather than removed) so the call site still reads as "success handling", with the reasoning attached to the behaviour.

Source code in jafaal/_internal/security_stores.py

def reset_ip_attempts(self, ip: str) -> None:
    """Deliberately a no-op — see below.

    The per-username counter is reset on a successful login because
    authenticating *as that user* proves the failures were that user
    fumbling their own password. No equivalent proof exists per IP: an
    address is shared by many accounts, so a success from it says nothing
    about the failures against other usernames.

    Resetting it made the per-IP tier — the only bound on the targeted
    lockout DoS the username tier enables — trivially defeatable: spray 49
    failures at victims, log in once to an account you own, repeat, and lock
    out arbitrarily many accounts from one address. The counter instead
    decays on its own ``attempts_ttl_seconds`` window, which is what keeps a
    legitimate NAT gateway from accumulating failures forever.

    Kept as a method (rather than removed) so the call site still reads as
    "success handling", with the reasoning attached to the behaviour.
    """
    return

clear_all

clear_all()

Clear all failed-login records (per-username and per-IP).

Source code in jafaal/_internal/security_stores.py

def clear_all(self) -> None:
    """Clear all failed-login records (per-username and per-IP)."""
    self._lockout.clear_all()
    self._ip_lockout.clear_all()

PendingLogin dataclass

A password-verified login awaiting its second factor.

Attributes:

Name	Type	Description
user_id	UserId	The user who completed the password step.
username	str	The username as supplied at login, used for the MFA lockout key and for audit records.
client_id	str	The registered client the login was started for. The second factor must be completed against the same client: the client's registration decides token delivery (cookie vs body) and the scope ceiling, so letting the ticket be redeemed by a different one would let a login begun for a narrow, body-delivery client finish as a wide, cookie-delivery one.
scope	tuple[str, ...]	The scope requested on the password step (RFC 6749 §3.3), re-applied when the second factor completes the login. Carried here rather than re-read from the second-factor request for the same reason as client_id: a value the caller re-supplies at step two is a value it can widen at step two. Empty means "whatever this client and user are entitled to".
auth_request	str | None	The pending authorization request this login is completing (/auth/authorize without an idp), or None for a direct login. Carried for the same reason again: the second factor must finish the authorization request the password step started, not one the caller names afterwards.

Source code in jafaal/_internal/security_stores.py

@dataclass(frozen=True)
class PendingLogin:
    """A password-verified login awaiting its second factor.

    Attributes:
        user_id: The user who completed the password step.
        username: The username as supplied at login, used for the MFA lockout
            key and for audit records.
        client_id: The registered client the login was started for. The second
            factor must be completed against the same client: the client's
            registration decides token delivery (cookie vs body) and the scope
            ceiling, so letting the ticket be redeemed by a different one would
            let a login begun for a narrow, body-delivery client finish as a
            wide, cookie-delivery one.
        scope: The ``scope`` requested on the password step (RFC 6749 §3.3),
            re-applied when the second factor completes the login. Carried here
            rather than re-read from the second-factor request for the same
            reason as ``client_id``: a value the caller re-supplies at step two
            is a value it can widen at step two. Empty means "whatever this
            client and user are entitled to".
        auth_request: The pending authorization request this login is completing
            (``/auth/authorize`` without an ``idp``), or ``None`` for a direct
            login. Carried for the same reason again: the second factor must
            finish the authorization request the password step started, not one
            the caller names afterwards.
    """

    user_id: UserId
    username: str
    client_id: str
    scope: tuple[str, ...] = ()
    auth_request: str | None = None

PendingMFALogin

Manage pending MFA logins plus per-username MFA failure lockout (5/10/15).

A pending login is addressed by an opaque, single-use mfa_token minted when the password step succeeds and returned to that caller only. The username is not an address: it is public (or guessable), so keying the pending record on it would mean anyone holding a valid one-time code could complete a login during the window a legitimate password step opened — collapsing two factors back to one. Possession of the ticket is the proof that the password factor was satisfied by this caller.

The ticket survives failed code attempts (a user may mistype) and is consumed atomically by :meth:claim_pending_login on success; the MFA lockout tiers bound how many attempts it can absorb.

Attributes:

Name	Type	Description
PENDING_MFA_TTL_SECONDS	int	TTL for pending MFA entries.
_state_override		Explicit provider (tests); None resolves lazily.
_lockout		Shared progressive-lockout helper for MFA failures.

Source code in jafaal/_internal/security_stores.py

class PendingMFALogin:
    """
    Manage pending MFA logins plus per-username MFA failure lockout (5/10/15).

    A pending login is addressed by an **opaque, single-use ``mfa_token``**
    minted when the password step succeeds and returned to that caller only. The
    username is *not* an address: it is public (or guessable), so keying the
    pending record on it would mean anyone holding a valid one-time code could
    complete a login during the window a legitimate password step opened —
    collapsing two factors back to one. Possession of the ticket is the proof
    that the password factor was satisfied *by this caller*.

    The ticket survives failed code attempts (a user may mistype) and is
    consumed atomically by :meth:`claim_pending_login` on success; the MFA
    lockout tiers bound how many attempts it can absorb.

    Attributes:
        PENDING_MFA_TTL_SECONDS: TTL for pending MFA entries.
        _state_override: Explicit provider (tests); ``None`` resolves lazily.
        _lockout: Shared progressive-lockout helper for MFA failures.
    """

    PENDING_MFA_TTL_SECONDS: int = 300

    def __init__(self, state: StateStore | None = None) -> None:
        self._state_override = state
        self._lockout = _ProgressiveLockout(
            self._get_state,
            name="mfa",
            display_name="MFA",
            thresholds=_MFA_LOCKOUT_THRESHOLDS,
            attempts_ttl_seconds=2 * 60 * 60,
            normalize_key=normalize_username_key,
        )

    def _get_state(self) -> StateStore:
        return self._state_override if self._state_override is not None else get_state_store()

    def _pending_key(self, mfa_token: str) -> str:
        # The stored key is a digest of the ticket, so the state store never
        # holds anything that could be replayed as the ticket itself.
        return f"{_key_prefix()}:mfa:pending:{hashing.sha256_hex(mfa_token)}"

    @staticmethod
    def _encode(
        user_id: UserId,
        username: str,
        client_id: str,
        scope: Sequence[str],
        auth_request: str | None,
    ) -> bytes:
        """Serialise a pending login for storage."""
        return json.dumps(
            {
                "uid": str(user_id),
                "un": username,
                "cid": client_id,
                "sc": list(scope),
                "ar": auth_request,
            }
        ).encode()

    @staticmethod
    def _decode(raw: bytes) -> PendingLogin | None:
        """Parse a stored pending login, or ``None`` if it is unusable."""
        try:
            payload = json.loads(raw.decode())
            # The id is stored in its string form and coerced back to the host
            # user table's primary-key type (``int`` or ``uuid.UUID``) on read,
            # so the store works for both integer- and UUID-keyed hosts.
            # ``sc`` / ``ar`` are read defensively: an entry written by an older
            # release predates them, and an in-flight login must not be
            # invalidated by a deploy.
            return PendingLogin(
                coerce_user_id(payload["uid"]),
                payload["un"],
                payload["cid"],
                tuple(payload.get("sc") or ()),
                payload.get("ar"),
            )
        except (TypeError, ValueError, KeyError, AttributeError):
            return None

    def add_pending_login(
        self,
        username: str,
        user_id: UserId,
        client_id: str,
        scope: Sequence[str] = (),
        auth_request: str | None = None,
    ) -> str:
        """Record a pending MFA login and return its opaque ticket.

        Args:
            username: The username that just passed the password step.
            user_id: The user the pending login belongs to.
            client_id: The registered client the login was started for; the
                second factor must be completed against the same one.
            scope: The ``scope`` requested on the password step, re-applied when
                the second factor completes the login.
            auth_request: The pending authorization request being completed, if
                the login came from ``/auth/authorize``.

        Returns:
            The ``mfa_token`` to hand to the caller; it must be presented to
            complete the second factor.
        """
        mfa_token = secrets.token_urlsafe(32)
        try:
            self._get_state().set(
                self._pending_key(mfa_token),
                self._encode(user_id, username, client_id, scope, auth_request),
                ttl_seconds=self.PENDING_MFA_TTL_SECONDS,
            )
        except StateStoreUnavailableError as err:
            _raise_store_unavailable("add pending MFA login", err)
        return mfa_token

    def get_pending_login(self, mfa_token: str) -> PendingLogin | None:
        """Resolve a pending MFA login from its ticket, evicting corrupt entries."""
        pending_key = self._pending_key(mfa_token)
        try:
            raw = self._get_state().get(pending_key)
        except StateStoreUnavailableError as err:
            _raise_store_unavailable("get pending MFA login", err)
        if raw is None:
            return None
        pending = self._decode(raw)
        if pending is None:
            try:
                self._get_state().delete(pending_key)
            except StateStoreUnavailableError as err:
                _raise_store_unavailable("delete invalid pending MFA login", err)
        return pending

    def claim_pending_login(self, mfa_token: str) -> PendingLogin | None:
        """Atomically consume a pending MFA login, so one ticket logs in once."""
        try:
            raw = self._get_state().get_and_delete(self._pending_key(mfa_token))
        except StateStoreUnavailableError as err:
            _raise_store_unavailable("claim pending MFA login", err)
        if raw is None:
            return None
        return self._decode(raw)

    def delete_pending_login(self, mfa_token: str) -> None:
        """Remove the pending MFA login addressed by ``mfa_token``."""
        try:
            self._get_state().delete(self._pending_key(mfa_token))
        except StateStoreUnavailableError as err:
            _raise_store_unavailable("delete pending MFA login", err)

    def clear_for_user(self, user_id: UserId) -> int:
        """Remove every pending MFA login entry tied to a user ID."""
        target = str(user_id)
        removed = 0
        state = self._get_state()
        try:
            for pending_key in list(state.iter_keys(f"{_key_prefix()}:mfa:pending:")):
                raw = state.get(pending_key)
                if raw is None:
                    continue
                pending = self._decode(raw)
                if pending is not None and str(pending.user_id) == target:
                    state.delete(pending_key)
                    removed += 1
        except StateStoreUnavailableError as err:
            _raise_store_unavailable("clear pending MFA logins for user", err)
        return removed

    def has_pending_login(self, mfa_token: str) -> bool:
        """Check whether ``mfa_token`` addresses a valid pending MFA login."""
        return self.get_pending_login(mfa_token) is not None

    def cleanup_expired(self) -> int:
        """Return zero because the backend expires pending entries by TTL."""
        return 0

    def is_locked_out(self, username: str) -> bool:
        """Check if a username is locked out from MFA attempts."""
        return self._lockout.is_locked_out(username)

    def get_lockout_time(self, username: str) -> datetime | None:
        """Get the MFA lockout expiry for a username, if locked out."""
        return self._lockout.get_lockout_time(username)

    def record_failed_attempt(self, username: str) -> int:
        """Record a failed MFA attempt and return the current count."""
        return self._lockout.record_failed_attempt(username)

    def reset_attempts(self, username: str) -> None:
        """Reset the MFA failure counter after a successful verification."""
        self._lockout.reset_attempts(username)

    def clear_all(self) -> None:
        """Clear all pending logins and MFA failure records."""
        try:
            self._get_state().delete_prefix(f"{_key_prefix()}:mfa:pending:")
        except StateStoreUnavailableError as err:
            _raise_store_unavailable("clear pending MFA logins", err)
        self._lockout.clear_all()

add_pending_login

add_pending_login(
    username,
    user_id,
    client_id,
    scope=(),
    auth_request=None,
)

Record a pending MFA login and return its opaque ticket.

Parameters:

Name	Type	Description	Default
username	str	The username that just passed the password step.	required
user_id	UserId	The user the pending login belongs to.	required
client_id	str	The registered client the login was started for; the second factor must be completed against the same one.	required
scope	Sequence[str]	The scope requested on the password step, re-applied when the second factor completes the login.	()
auth_request	str | None	The pending authorization request being completed, if the login came from /auth/authorize.	None

Returns:

Type	Description
str	The mfa_token to hand to the caller; it must be presented to
str	complete the second factor.

Source code in jafaal/_internal/security_stores.py

def add_pending_login(
    self,
    username: str,
    user_id: UserId,
    client_id: str,
    scope: Sequence[str] = (),
    auth_request: str | None = None,
) -> str:
    """Record a pending MFA login and return its opaque ticket.

    Args:
        username: The username that just passed the password step.
        user_id: The user the pending login belongs to.
        client_id: The registered client the login was started for; the
            second factor must be completed against the same one.
        scope: The ``scope`` requested on the password step, re-applied when
            the second factor completes the login.
        auth_request: The pending authorization request being completed, if
            the login came from ``/auth/authorize``.

    Returns:
        The ``mfa_token`` to hand to the caller; it must be presented to
        complete the second factor.
    """
    mfa_token = secrets.token_urlsafe(32)
    try:
        self._get_state().set(
            self._pending_key(mfa_token),
            self._encode(user_id, username, client_id, scope, auth_request),
            ttl_seconds=self.PENDING_MFA_TTL_SECONDS,
        )
    except StateStoreUnavailableError as err:
        _raise_store_unavailable("add pending MFA login", err)
    return mfa_token

get_pending_login

get_pending_login(mfa_token)

Resolve a pending MFA login from its ticket, evicting corrupt entries.

Source code in jafaal/_internal/security_stores.py

def get_pending_login(self, mfa_token: str) -> PendingLogin | None:
    """Resolve a pending MFA login from its ticket, evicting corrupt entries."""
    pending_key = self._pending_key(mfa_token)
    try:
        raw = self._get_state().get(pending_key)
    except StateStoreUnavailableError as err:
        _raise_store_unavailable("get pending MFA login", err)
    if raw is None:
        return None
    pending = self._decode(raw)
    if pending is None:
        try:
            self._get_state().delete(pending_key)
        except StateStoreUnavailableError as err:
            _raise_store_unavailable("delete invalid pending MFA login", err)
    return pending

claim_pending_login

claim_pending_login(mfa_token)

Atomically consume a pending MFA login, so one ticket logs in once.

Source code in jafaal/_internal/security_stores.py

def claim_pending_login(self, mfa_token: str) -> PendingLogin | None:
    """Atomically consume a pending MFA login, so one ticket logs in once."""
    try:
        raw = self._get_state().get_and_delete(self._pending_key(mfa_token))
    except StateStoreUnavailableError as err:
        _raise_store_unavailable("claim pending MFA login", err)
    if raw is None:
        return None
    return self._decode(raw)

delete_pending_login

delete_pending_login(mfa_token)

Remove the pending MFA login addressed by mfa_token.

Source code in jafaal/_internal/security_stores.py

def delete_pending_login(self, mfa_token: str) -> None:
    """Remove the pending MFA login addressed by ``mfa_token``."""
    try:
        self._get_state().delete(self._pending_key(mfa_token))
    except StateStoreUnavailableError as err:
        _raise_store_unavailable("delete pending MFA login", err)

clear_for_user

clear_for_user(user_id)

Remove every pending MFA login entry tied to a user ID.

Source code in jafaal/_internal/security_stores.py

def clear_for_user(self, user_id: UserId) -> int:
    """Remove every pending MFA login entry tied to a user ID."""
    target = str(user_id)
    removed = 0
    state = self._get_state()
    try:
        for pending_key in list(state.iter_keys(f"{_key_prefix()}:mfa:pending:")):
            raw = state.get(pending_key)
            if raw is None:
                continue
            pending = self._decode(raw)
            if pending is not None and str(pending.user_id) == target:
                state.delete(pending_key)
                removed += 1
    except StateStoreUnavailableError as err:
        _raise_store_unavailable("clear pending MFA logins for user", err)
    return removed

has_pending_login

has_pending_login(mfa_token)

Check whether mfa_token addresses a valid pending MFA login.

Source code in jafaal/_internal/security_stores.py

def has_pending_login(self, mfa_token: str) -> bool:
    """Check whether ``mfa_token`` addresses a valid pending MFA login."""
    return self.get_pending_login(mfa_token) is not None

cleanup_expired

cleanup_expired()

Return zero because the backend expires pending entries by TTL.

Source code in jafaal/_internal/security_stores.py

def cleanup_expired(self) -> int:
    """Return zero because the backend expires pending entries by TTL."""
    return 0

is_locked_out

is_locked_out(username)

Check if a username is locked out from MFA attempts.

Source code in jafaal/_internal/security_stores.py

def is_locked_out(self, username: str) -> bool:
    """Check if a username is locked out from MFA attempts."""
    return self._lockout.is_locked_out(username)

get_lockout_time

get_lockout_time(username)

Get the MFA lockout expiry for a username, if locked out.

Source code in jafaal/_internal/security_stores.py

def get_lockout_time(self, username: str) -> datetime | None:
    """Get the MFA lockout expiry for a username, if locked out."""
    return self._lockout.get_lockout_time(username)

record_failed_attempt

record_failed_attempt(username)

Record a failed MFA attempt and return the current count.

Source code in jafaal/_internal/security_stores.py

def record_failed_attempt(self, username: str) -> int:
    """Record a failed MFA attempt and return the current count."""
    return self._lockout.record_failed_attempt(username)

reset_attempts

reset_attempts(username)

Reset the MFA failure counter after a successful verification.

Source code in jafaal/_internal/security_stores.py

def reset_attempts(self, username: str) -> None:
    """Reset the MFA failure counter after a successful verification."""
    self._lockout.reset_attempts(username)

clear_all

clear_all()

Clear all pending logins and MFA failure records.

Source code in jafaal/_internal/security_stores.py

def clear_all(self) -> None:
    """Clear all pending logins and MFA failure records."""
    try:
        self._get_state().delete_prefix(f"{_key_prefix()}:mfa:pending:")
    except StateStoreUnavailableError as err:
        _raise_store_unavailable("clear pending MFA logins", err)
    self._lockout.clear_all()

StepUpAttempts

Track failed step-up verification attempts (5/10/15 → 5m/30m/2h).

Keys are stable user identifiers (e.g. user:{user_id}).

Attributes:

Name	Type	Description
_state_override		Explicit provider (tests); None resolves lazily.
_lockout		Shared progressive-lockout helper.

Source code in jafaal/_internal/security_stores.py

class StepUpAttempts:
    """
    Track failed step-up verification attempts (5/10/15 → 5m/30m/2h).

    Keys are stable user identifiers (e.g. ``user:{user_id}``).

    Attributes:
        _state_override: Explicit provider (tests); ``None`` resolves lazily.
        _lockout: Shared progressive-lockout helper.
    """

    def __init__(self, state: StateStore | None = None) -> None:
        self._state_override = state
        self._lockout = _ProgressiveLockout(
            self._get_state,
            name="step_up",
            display_name="Step-up",
            thresholds=_STEP_UP_LOCKOUT_THRESHOLDS,
            attempts_ttl_seconds=2 * 60 * 60,
        )

    def _get_state(self) -> StateStore:
        return self._state_override if self._state_override is not None else get_state_store()

    def is_locked_out(self, key: str) -> bool:
        """Check if a user key is locked out from step-up."""
        return self._lockout.is_locked_out(key)

    def get_lockout_time(self, key: str) -> datetime | None:
        """Get the step-up lockout expiry for a user key, if locked out."""
        return self._lockout.get_lockout_time(key)

    def record_failed_attempt(self, key: str) -> int:
        """Record a failed step-up attempt and return the current count."""
        return self._lockout.record_failed_attempt(key)

    def reset_attempts(self, key: str) -> None:
        """Reset the step-up failure counter for a user key."""
        self._lockout.reset_attempts(key)

    def clear_all(self) -> None:
        """Clear all step-up failure records."""
        self._lockout.clear_all()

is_locked_out

is_locked_out(key)

Check if a user key is locked out from step-up.

Source code in jafaal/_internal/security_stores.py

def is_locked_out(self, key: str) -> bool:
    """Check if a user key is locked out from step-up."""
    return self._lockout.is_locked_out(key)

get_lockout_time

get_lockout_time(key)

Get the step-up lockout expiry for a user key, if locked out.

Source code in jafaal/_internal/security_stores.py

def get_lockout_time(self, key: str) -> datetime | None:
    """Get the step-up lockout expiry for a user key, if locked out."""
    return self._lockout.get_lockout_time(key)

record_failed_attempt

record_failed_attempt(key)

Record a failed step-up attempt and return the current count.

Source code in jafaal/_internal/security_stores.py

def record_failed_attempt(self, key: str) -> int:
    """Record a failed step-up attempt and return the current count."""
    return self._lockout.record_failed_attempt(key)

reset_attempts

reset_attempts(key)

Reset the step-up failure counter for a user key.

Source code in jafaal/_internal/security_stores.py

def reset_attempts(self, key: str) -> None:
    """Reset the step-up failure counter for a user key."""
    self._lockout.reset_attempts(key)

clear_all

clear_all()

Clear all step-up failure records.

Source code in jafaal/_internal/security_stores.py

def clear_all(self) -> None:
    """Clear all step-up failure records."""
    self._lockout.clear_all()

StepUpStore

Bases: Protocol

Contract for step-up lockout stores. Keys are stable user identifiers.

Source code in jafaal/_internal/security_stores.py

@runtime_checkable
class StepUpStore(Protocol):
    """Contract for step-up lockout stores. Keys are stable user identifiers."""

    def is_locked_out(self, key: str) -> bool: ...

    def get_lockout_time(self, key: str) -> datetime | None: ...

    def record_failed_attempt(self, key: str) -> int: ...

    def reset_attempts(self, key: str) -> None: ...

    def clear_all(self) -> None: ...

TokenManager

Issue, decode, and validate JWTs (and mint CSRF tokens) for user sessions.

Signs with either HS256 (symmetric, the default — a shared secret) or an asymmetric RSA/EC algorithm (RS256/ES256/…), where a private key signs and the corresponding public key is published at the JWKS endpoint so resource servers verify statelessly. The algorithm is pinned via :data:jafaal.settings.ALLOWED_ALGORITHMS and the same allow-list is passed to jwt.decode so it cannot drift (blocking alg=none and algorithm-confusion). Asymmetric tokens carry the active key's RFC 7638 thumbprint as kid. Validation failures raise a :class:~jafaal.exceptions.JafaalError (mapped to HTTP 401 at the router edge); the constructor raises :class:ValueError for an algorithm outside the allow-list or missing asymmetric key material.

Attributes:

Name	Type	Description
algorithm		The JWT signing algorithm.

Source code in jafaal/_internal/token_manager.py

class TokenManager:
    """Issue, decode, and validate JWTs (and mint CSRF tokens) for user sessions.

    Signs with either ``HS256`` (symmetric, the default — a shared secret) or an
    asymmetric RSA/EC algorithm (``RS256``/``ES256``/…), where a private key
    signs and the corresponding public key is published at the JWKS endpoint so
    resource servers verify statelessly. The algorithm is pinned via
    :data:`jafaal.settings.ALLOWED_ALGORITHMS` and the same allow-list is passed
    to ``jwt.decode`` so it cannot drift (blocking ``alg=none`` and
    algorithm-confusion). Asymmetric tokens carry the active key's RFC 7638
    thumbprint as ``kid``. Validation failures raise a
    :class:`~jafaal.exceptions.JafaalError` (mapped to HTTP 401 at the router
    edge); the constructor raises :class:`ValueError` for an algorithm outside
    the allow-list or missing asymmetric key material.

    Attributes:
        algorithm: The JWT signing algorithm.
    """

    def __init__(
        self,
        secret_key: str,
        algorithm: str = "HS256",
        *,
        access_token_expire_minutes: int = 15,
        refresh_token_expire_days: int = 7,
        issuer: str = "",
        audience: str = "",
        secret_key_fallbacks: tuple[str, ...] = (),
        private_key: str = "",
        private_key_fallbacks: tuple[str, ...] = (),
        leeway_seconds: int = 0,
        client_id: str = "",
    ):
        """
        Initializes the TokenManager with the provided secret key and settings.

        Args:
            secret_key (str): The secret key used for signing and verifying
                tokens.
            algorithm (str, optional): The algorithm to use for token
                operations. Defaults to "HS256". Must be a member of
                :data:`jafaal.settings.ALLOWED_ALGORITHMS` so that the
                allow-list passed to ``jwt.decode`` cannot drift from the
                signing algorithm.
            access_token_expire_minutes (int): Access-token lifetime in minutes.
            refresh_token_expire_days (int): Refresh-token lifetime in days.
            issuer (str): JWT ``iss`` claim value.
            audience (str): JWT ``aud`` claim value.
            secret_key_fallbacks (tuple[str, ...]): Additional keys accepted
                when *verifying* a token (never used to sign). Lets tokens
                issued before a ``secret_key`` rotation keep validating during
                the overlap window.
            private_key (str): PEM private key used to sign JWTs when
                ``algorithm`` is asymmetric (RSA/EC); ignored for HS256.
            private_key_fallbacks (tuple[str, ...]): Verify-only public/private
                PEM keys kept in the published JWKS during a signing-key
                rotation overlap.
            leeway_seconds (int): Clock-skew tolerance, in seconds, applied to
                the ``exp`` / ``nbf`` claims during validation. ``0`` is strict;
                a small value avoids spurious 401s when the issuing and
                validating clocks differ slightly.
            client_id (str): Value of the ``client_id`` claim RFC 9068 requires
                on an access token.
        """
        if algorithm not in jafaal_settings.ALLOWED_ALGORITHMS:
            raise ValueError(
                f"algorithm={algorithm!r} is not in the JWT allow-list {sorted(jafaal_settings.ALLOWED_ALGORITHMS)}."
            )
        self.secret_key = secret_key
        self.algorithm = algorithm
        self.access_token_expire_minutes = access_token_expire_minutes
        self.refresh_token_expire_days = refresh_token_expire_days
        self.issuer = issuer
        self.audience = audience
        self.leeway_seconds = leeway_seconds
        self.client_id = client_id

        self._is_symmetric: bool = algorithm not in jwk_keys.ASYMMETRIC_ALGORITHMS
        self._sign_key: Any
        self._sign_header: dict[str, str]
        self._encode_algorithms: list[str] | None
        self._decode_keys: list[Any]
        self._verify_keys: list[Any]
        self._verify_keyset: KeySet | None

        if self._is_symmetric:
            # HS256: sign and verify with the shared secret. Verification also
            # accepts any rotation fallbacks so a token signed before a
            # secret_key rotation still validates during the overlap window;
            # signing always uses the primary key.
            self._sign_key = OctKey.import_key(secret_key)
            self._sign_header = {"alg": algorithm}
            self._encode_algorithms = None
            self._decode_keys = [self._sign_key, *(OctKey.import_key(fallback) for fallback in secret_key_fallbacks)]
            self._verify_keys = []
            self._verify_keyset = None
        else:
            # Asymmetric: sign with the private key; verify/publish with the
            # public key(s). The token header carries the active key's RFC 7638
            # thumbprint as ``kid`` so verifiers and the JWKS agree on it, and
            # fallback public keys stay in the JWKS during a rotation overlap.
            if not private_key:
                raise ValueError(f"algorithm={algorithm!r} is asymmetric and requires a private_key.")
            self._sign_key = jwk_keys.import_private_signing_key(private_key, algorithm)
            self._sign_header = {"alg": algorithm, "kid": self._sign_key.thumbprint()}
            self._encode_algorithms = [algorithm]
            self._verify_keys = [
                jwk_keys.public_verification_key(self._sign_key, algorithm),
                *(jwk_keys.import_verification_key(fallback, algorithm) for fallback in private_key_fallbacks),
            ]
            self._verify_keyset = KeySet(self._verify_keys)
            self._decode_keys = []

    def get_token_claim(self, token: str, claim: str) -> str | list[str] | int:
        """
        Retrieves a specific claim from a decoded JWT token.

        Args:
            token (str): The JWT token string to decode.
            claim (str): The name of the claim to retrieve from the token.

        Returns:
            str | list[str] | int: The value of the requested claim, which can
                be a string, list of strings, or integer.

        Raises:
            JafaalError: If the claim is not found in the token or if there
                is an error retrieving the claim.
        """
        try:
            # Decode the token
            payload = self.decode_token(token)

            # Get the claim from the payload and return it
            return payload.claims[claim]
        except KeyError as err:
            logger.error(f"Claim '{claim}' not found in token: {err}", exc_info=err, extra={"token": "[REDACTED]"})
            raise jafaal_exceptions.InvalidTokenError(f"Claim '{claim}' is missing in the token.") from err
        except jafaal_exceptions.JafaalError:
            # decode_token already raised a properly-formed 401; re-raise as-is.
            raise
        except Exception as err:
            logger.error(
                f"Unexpected error retrieving claim: {type(err).__name__}", exc_info=err, extra={"token": "[REDACTED]"}
            )
            raise jafaal_exceptions.InvalidTokenError("Unable to retrieve claim") from err

    def decode_token(self, token: str) -> Token:
        """
        Decodes a JWT token and returns the parsed Token object.

        The ``algorithms`` allow-list (pinned to this manager's algorithm) is
        always passed to ``jwt.decode``: without it joserfc would trust whatever
        algorithm the token header advertises (``none`` or an
        algorithm-confusion variant), bypassing the signature check.

        In symmetric (HS256) mode the token is verified against the primary key
        then each rotation fallback. In asymmetric mode it is verified against
        the public-key set, which joserfc selects by the header ``kid`` (the
        active key plus any rotation fallbacks).

        Args:
            token (str): The JWT token to decode.

        Returns:
            joserfc.jwt.Token: The decoded token (use ``.claims`` to access
                payload claims).

        Raises:
            JafaalError: If the token cannot be decoded, raises an HTTP 401
                Unauthorized exception.
        """
        if self._is_symmetric:
            return self._decode_symmetric(token)
        return self._decode_asymmetric(token)

    def _decode_symmetric(self, token: str) -> Token:
        """Verify an HS256 token against the primary key then rotation fallbacks."""
        last_signature_err: BadSignatureError | None = None
        for key in self._decode_keys:
            try:
                return jwt.decode(token, key, algorithms=[self.algorithm])
            except BadSignatureError as sig_err:
                # Wrong key: try the next rotation fallback before giving up.
                last_signature_err = sig_err
                continue
            except InvalidPayloadError as payload_err:
                logger.error(
                    f"Invalid token payload: {payload_err}", exc_info=payload_err, extra={"token": "[REDACTED]"}
                )
                raise jafaal_exceptions.InvalidTokenError("Invalid token payload") from payload_err
            except DecodeError as decode_err:
                logger.error(f"Error decoding token: {decode_err}", exc_info=decode_err, extra={"token": "[REDACTED]"})
                raise jafaal_exceptions.InvalidTokenError("Unable to decode token") from decode_err
            except Exception as err:
                logger.error(
                    f"Unexpected error decoding token: {type(err).__name__}",
                    exc_info=err,
                    extra={"token": "[REDACTED]"},
                )
                raise jafaal_exceptions.InvalidTokenError("Unable to decode token") from err
        # Signature did not match the primary key or any rotation fallback.
        logger.error(
            "Token signature did not match any active signing key",
            exc_info=last_signature_err,
            extra={"token": "[REDACTED]"},
        )
        raise jafaal_exceptions.InvalidTokenError("Unable to decode token") from last_signature_err

    def _decode_asymmetric(self, token: str) -> Token:
        """Verify an asymmetric token against the public-key set (selected by ``kid``)."""
        keyset = self._verify_keyset
        if keyset is None:  # pragma: no cover - always built in asymmetric mode
            raise jafaal_exceptions.InvalidTokenError("Unable to decode token")
        try:
            return jwt.decode(token, keyset, algorithms=[self.algorithm])
        except BadSignatureError as sig_err:
            logger.error(
                "Token signature did not match any active signing key",
                exc_info=sig_err,
                extra={"token": "[REDACTED]"},
            )
            raise jafaal_exceptions.InvalidTokenError("Unable to decode token") from sig_err
        except InvalidPayloadError as payload_err:
            logger.error(f"Invalid token payload: {payload_err}", exc_info=payload_err, extra={"token": "[REDACTED]"})
            raise jafaal_exceptions.InvalidTokenError("Invalid token payload") from payload_err
        except DecodeError as decode_err:
            logger.error(f"Error decoding token: {decode_err}", exc_info=decode_err, extra={"token": "[REDACTED]"})
            raise jafaal_exceptions.InvalidTokenError("Unable to decode token") from decode_err
        except jafaal_exceptions.JafaalError:
            raise
        except Exception as err:
            logger.error(
                f"Unexpected error decoding token: {type(err).__name__}",
                exc_info=err,
                extra={"token": "[REDACTED]"},
            )
            raise jafaal_exceptions.InvalidTokenError("Unable to decode token") from err

    def validate_token_expiration(
        self,
        token: str,
        expected_type: TokenType,
    ) -> None:
        """
        Validates expiration, required claims, and type of a JWT.

        Checks that the token contains all essential claims, is not expired or
        used before its valid time, and that it names the expected token use.
        This prevents refresh tokens from being used as access tokens and vice
        versa.

        Args:
            token: The JWT token to validate.
            expected_type: The expected token type
                (``TokenType.ACCESS`` or
                ``TokenType.REFRESH``).

        Raises:
            JafaalError: If the token is missing required claims, expired,
                not yet valid, contains invalid claims, or has the wrong type.
        """
        try:
            # Define required claims. ``leeway`` applies a small clock-skew
            # tolerance to the time-based claims (``exp`` / ``nbf``) so slightly
            # skewed nodes do not spuriously reject otherwise-valid tokens.
            # The token-use claim is checked separately below.
            claims_requests = jwt.JWTClaimsRegistry(
                leeway=self.leeway_seconds,
                sid={"essential": True},
                iss={
                    "essential": True,
                    "value": self.issuer,
                },
                aud={
                    "essential": True,
                    "value": self.audience,
                },
                sub={"essential": True},
                scope={"essential": True},
                iat={"essential": True},
                nbf={"essential": True},
                exp={"essential": True},
                jti={"essential": True},
            )

            # Decode the token to get the payload
            payload = self.decode_token(token)
            _validate_typ_header(payload.header, expected_type)
            _validate_token_use(payload.claims, expected_type)

            # Validate token claims (incl. expiration and typ)
            claims_requests.validate(payload.claims)
        except MissingClaimError as missing_err:
            logger.error(f"JWT missing claim error: {missing_err}", exc_info=missing_err, extra={"token": "[REDACTED]"})
            raise jafaal_exceptions.InvalidTokenError("Token is missing required claims.") from missing_err
        except ExpiredTokenError as expired_err:
            raise jafaal_exceptions.TokenExpiredError("Token is expired.") from expired_err
        except InvalidTokenError as invalid_err:
            logger.error(
                f"JWT is not valid yet error: {invalid_err}", exc_info=invalid_err, extra={"token": "[REDACTED]"}
            )
            raise jafaal_exceptions.InvalidTokenError("Token is not valid yet.") from invalid_err
        except InsecureClaimError as insecure_err:
            logger.error(
                f"JWT insecure claim error: {insecure_err}", exc_info=insecure_err, extra={"token": "[REDACTED]"}
            )
            raise jafaal_exceptions.InvalidTokenError("Token has insecure claims.") from insecure_err
        except InvalidClaimError as claims_err:
            logger.error(
                f"JWT claims validation error: {claims_err}", exc_info=claims_err, extra={"token": "[REDACTED]"}
            )
            raise jafaal_exceptions.InvalidTokenError("Token has invalid claims.") from claims_err
        except jafaal_exceptions.JafaalError:
            # decode_token already raised a properly-formed 401; re-raise as-is.
            raise
        except Exception as err:
            logger.error(
                f"Unexpected error validating token: {type(err).__name__}", exc_info=err, extra={"token": "[REDACTED]"}
            )
            raise jafaal_exceptions.InvalidTokenError("Token expired or invalid.") from err

    def validate_access_expiration_logged(self, access_token: str) -> None:
        """
        Validate an access token's expiration and log failures consistently.

        Wraps :meth:`validate_token_expiration` for ``TokenType.ACCESS`` and
        applies the shared logging policy used by the access-token validation
        dependency and ``IdentityService``: expired tokens log at ``debug``
        (an expected, routine condition) while all other validation failures
        log at ``error``. The original :class:`JafaalError` is re-raised
        unchanged so callers keep the same 401 semantics.

        Args:
            access_token: The raw JWT access token to validate.

        Raises:
            JafaalError: 401 if the token is missing claims, expired, not
                yet valid, or otherwise invalid.
        """
        try:
            self.validate_token_expiration(access_token, TokenType.ACCESS)
        except jafaal_exceptions.JafaalError as http_err:
            is_expired = isinstance(http_err, jafaal_exceptions.TokenExpiredError)
            logger.log(
                logging.DEBUG if is_expired else logging.ERROR,
                f"Access token validation failed: {http_err.detail}",
                exc_info=None if is_expired else http_err,
                extra={"access_token": "[REDACTED]"},
            )
            raise

    def create_token(
        self,
        session_id: str,
        user: jafaal_ports.UserProtocol,
        token_type: TokenType,
        client: jafaal_settings.OAuthClient | None = None,
        requested_scope: Sequence[str] | None = None,
    ) -> tuple[datetime, str]:
        """
        Creates a JWT token for a user session with appropriate access scope
        and expiration.

        Args:
            session_id (str): The unique identifier for the session.
            user (jafaal_ports.UserProtocol): The user object containing user
                details.
            token_type (TokenType): The type of token to create (access or
                refresh).
            client: The registered client the token is being issued to. Its
                scope ceiling narrows the user's grants and its ``client_id``
                becomes the RFC 9068 §2.2 ``client_id`` claim. ``None`` falls
                back to the deployment-wide identifier.
            requested_scope: The ``scope`` the client asked for (RFC 6749 §3.3),
                applied as a final narrowing bound. ``None`` means "everything
                this client and user are entitled to".

        Returns:
            tuple[datetime, str]: A tuple containing the token's expiration
                datetime and the encoded JWT token string.

        Raises:
            ValueError: If required parameters are missing or invalid.
        """
        # Three bounds, applied in order, each of which can only ever remove:
        #
        # 1. the host's ScopeResolver port decides what the *account* holds (the
        #    built-in default is the catalog's is_superuser two-tier mapping), so
        #    an application with a richer authorisation model stamps its own
        #    grants without patching the token minter;
        # 2. the registered client's ceiling caps what *this client* may ever
        #    carry; and
        # 3. the client's own ``scope`` request caps what it asked for on *this*
        #    exchange, so a client that deliberately asks for less gets less.
        scope = jafaal_ports.get_scope_resolver().scopes_for(user)
        if client is not None:
            scope = client.narrow(scope)
        scope = jafaal_scopes.narrow_to_requested(scope, requested_scope)

        # Set now
        issued_at = datetime.now(UTC)
        lifetime = (
            timedelta(days=self.refresh_token_expire_days)
            if token_type == TokenType.REFRESH
            else timedelta(minutes=self.access_token_expire_minutes)
        )
        exp = issued_at + lifetime
        now = int(issued_at.timestamp())

        claims: dict[str, Any] = {
            "sid": session_id,
            "iss": self.issuer,
            "aud": self.audience,
            "iat": now,
            # Backdated by a few seconds. ``nbf == iat`` is correct in the
            # abstract and a reliable interop failure in practice: a *resource
            # server* verifying with a stock JWT library and a clock a second
            # behind ours rejects a token that was minted milliseconds earlier,
            # and it has no ``leeway_seconds`` of ours to fall back on. RFC 7519
            # §4.1.5 anticipates exactly this ("implementers MAY provide for
            # some small leeway"); applying it at issuance rather than asking
            # every verifier to configure it is what makes the token portable.
            # Small enough not to widen any meaningful attack window: the token
            # is already valid from the instant it is handed out.
            "nbf": now - _NBF_BACKDATE_SECONDS,
            "exp": exp,
            "jti": str(uuid.uuid4()),
            # RFC 9068 / RFC 7519 shapes, so a resource server verifying against
            # the published JWKS with a stock JWT library reads what it expects:
            # ``sub`` is a string (RFC 7519 §4.1.2 defines it as StringOrURI),
            # ``scope`` is space-delimited (RFC 6749 §3.3), and ``client_id`` is
            # present (RFC 9068 §2.2). ``coerce_user_id`` converts ``sub`` back
            # to the host user table's primary-key type on the way in.
            "sub": str(user.id),
            "scope": " ".join(scope),
            "client_id": client.client_id if client is not None else self.client_id,
            TOKEN_USE_CLAIM: token_type.value,
        }
        # The media type goes in the JOSE ``typ`` header, not a payload claim.
        header = {**self._sign_header, "typ": _TYP_HEADER_BY_TOKEN_TYPE[token_type]}

        encoded_token = jwt.encode(
            header,
            claims.copy(),
            self._sign_key,
            algorithms=self._encode_algorithms,
        )

        # Return the expiration and the encoded token
        return exp, encoded_token

    @staticmethod
    def create_csrf_token() -> str:
        """
        Generate a secure random CSRF (Cross-Site Request Forgery) token.

        Returns:
            str: A URL-safe, securely generated random string suitable for use
                as a CSRF token.
        """
        return secrets.token_urlsafe(32)

    def jwks(self) -> dict[str, Any]:
        """Return the JSON Web Key Set of public verification keys.

        Empty (``{"keys": []}``) in symmetric (HS256) mode, which has no public
        key to publish. In asymmetric mode it contains the active signing key's
        public JWK plus any rotation fallbacks, each tagged with its ``kid``
        (RFC 7638 thumbprint), ``use: "sig"``, and ``alg`` — exactly what a
        resource server needs to verify JAFAAL's access tokens statelessly.
        """
        return {"keys": [jwk_keys.jwk_entry(key, self.algorithm) for key in self._verify_keys]}

__init__

__init__(
    secret_key,
    algorithm="HS256",
    *,
    access_token_expire_minutes=15,
    refresh_token_expire_days=7,
    issuer="",
    audience="",
    secret_key_fallbacks=(),
    private_key="",
    private_key_fallbacks=(),
    leeway_seconds=0,
    client_id="",
)

Initializes the TokenManager with the provided secret key and settings.

Parameters:

Name	Type	Description	Default
secret_key	str	The secret key used for signing and verifying tokens.	required
algorithm	str	The algorithm to use for token operations. Defaults to "HS256". Must be a member of :data:jafaal.settings.ALLOWED_ALGORITHMS so that the allow-list passed to jwt.decode cannot drift from the signing algorithm.	'HS256'
access_token_expire_minutes	int	Access-token lifetime in minutes.	15
refresh_token_expire_days	int	Refresh-token lifetime in days.	7
issuer	str	JWT iss claim value.	''
audience	str	JWT aud claim value.	''
secret_key_fallbacks	tuple[str, ...]	Additional keys accepted when verifying a token (never used to sign). Lets tokens issued before a secret_key rotation keep validating during the overlap window.	()
private_key	str	PEM private key used to sign JWTs when algorithm is asymmetric (RSA/EC); ignored for HS256.	''
private_key_fallbacks	tuple[str, ...]	Verify-only public/private PEM keys kept in the published JWKS during a signing-key rotation overlap.	()
leeway_seconds	int	Clock-skew tolerance, in seconds, applied to the exp / nbf claims during validation. 0 is strict; a small value avoids spurious 401s when the issuing and validating clocks differ slightly.	0
client_id	str	Value of the client_id claim RFC 9068 requires on an access token.	''

Source code in jafaal/_internal/token_manager.py

def __init__(
    self,
    secret_key: str,
    algorithm: str = "HS256",
    *,
    access_token_expire_minutes: int = 15,
    refresh_token_expire_days: int = 7,
    issuer: str = "",
    audience: str = "",
    secret_key_fallbacks: tuple[str, ...] = (),
    private_key: str = "",
    private_key_fallbacks: tuple[str, ...] = (),
    leeway_seconds: int = 0,
    client_id: str = "",
):
    """
    Initializes the TokenManager with the provided secret key and settings.

    Args:
        secret_key (str): The secret key used for signing and verifying
            tokens.
        algorithm (str, optional): The algorithm to use for token
            operations. Defaults to "HS256". Must be a member of
            :data:`jafaal.settings.ALLOWED_ALGORITHMS` so that the
            allow-list passed to ``jwt.decode`` cannot drift from the
            signing algorithm.
        access_token_expire_minutes (int): Access-token lifetime in minutes.
        refresh_token_expire_days (int): Refresh-token lifetime in days.
        issuer (str): JWT ``iss`` claim value.
        audience (str): JWT ``aud`` claim value.
        secret_key_fallbacks (tuple[str, ...]): Additional keys accepted
            when *verifying* a token (never used to sign). Lets tokens
            issued before a ``secret_key`` rotation keep validating during
            the overlap window.
        private_key (str): PEM private key used to sign JWTs when
            ``algorithm`` is asymmetric (RSA/EC); ignored for HS256.
        private_key_fallbacks (tuple[str, ...]): Verify-only public/private
            PEM keys kept in the published JWKS during a signing-key
            rotation overlap.
        leeway_seconds (int): Clock-skew tolerance, in seconds, applied to
            the ``exp`` / ``nbf`` claims during validation. ``0`` is strict;
            a small value avoids spurious 401s when the issuing and
            validating clocks differ slightly.
        client_id (str): Value of the ``client_id`` claim RFC 9068 requires
            on an access token.
    """
    if algorithm not in jafaal_settings.ALLOWED_ALGORITHMS:
        raise ValueError(
            f"algorithm={algorithm!r} is not in the JWT allow-list {sorted(jafaal_settings.ALLOWED_ALGORITHMS)}."
        )
    self.secret_key = secret_key
    self.algorithm = algorithm
    self.access_token_expire_minutes = access_token_expire_minutes
    self.refresh_token_expire_days = refresh_token_expire_days
    self.issuer = issuer
    self.audience = audience
    self.leeway_seconds = leeway_seconds
    self.client_id = client_id

    self._is_symmetric: bool = algorithm not in jwk_keys.ASYMMETRIC_ALGORITHMS
    self._sign_key: Any
    self._sign_header: dict[str, str]
    self._encode_algorithms: list[str] | None
    self._decode_keys: list[Any]
    self._verify_keys: list[Any]
    self._verify_keyset: KeySet | None

    if self._is_symmetric:
        # HS256: sign and verify with the shared secret. Verification also
        # accepts any rotation fallbacks so a token signed before a
        # secret_key rotation still validates during the overlap window;
        # signing always uses the primary key.
        self._sign_key = OctKey.import_key(secret_key)
        self._sign_header = {"alg": algorithm}
        self._encode_algorithms = None
        self._decode_keys = [self._sign_key, *(OctKey.import_key(fallback) for fallback in secret_key_fallbacks)]
        self._verify_keys = []
        self._verify_keyset = None
    else:
        # Asymmetric: sign with the private key; verify/publish with the
        # public key(s). The token header carries the active key's RFC 7638
        # thumbprint as ``kid`` so verifiers and the JWKS agree on it, and
        # fallback public keys stay in the JWKS during a rotation overlap.
        if not private_key:
            raise ValueError(f"algorithm={algorithm!r} is asymmetric and requires a private_key.")
        self._sign_key = jwk_keys.import_private_signing_key(private_key, algorithm)
        self._sign_header = {"alg": algorithm, "kid": self._sign_key.thumbprint()}
        self._encode_algorithms = [algorithm]
        self._verify_keys = [
            jwk_keys.public_verification_key(self._sign_key, algorithm),
            *(jwk_keys.import_verification_key(fallback, algorithm) for fallback in private_key_fallbacks),
        ]
        self._verify_keyset = KeySet(self._verify_keys)
        self._decode_keys = []

get_token_claim

get_token_claim(token, claim)

Retrieves a specific claim from a decoded JWT token.

Parameters:

Name	Type	Description	Default
token	str	The JWT token string to decode.	required
claim	str	The name of the claim to retrieve from the token.	required

Returns:

Type	Description
str | list[str] | int	str | list[str] | int: The value of the requested claim, which can be a string, list of strings, or integer.

Raises:

Type	Description
JafaalError	If the claim is not found in the token or if there is an error retrieving the claim.

Source code in jafaal/_internal/token_manager.py

def get_token_claim(self, token: str, claim: str) -> str | list[str] | int:
    """
    Retrieves a specific claim from a decoded JWT token.

    Args:
        token (str): The JWT token string to decode.
        claim (str): The name of the claim to retrieve from the token.

    Returns:
        str | list[str] | int: The value of the requested claim, which can
            be a string, list of strings, or integer.

    Raises:
        JafaalError: If the claim is not found in the token or if there
            is an error retrieving the claim.
    """
    try:
        # Decode the token
        payload = self.decode_token(token)

        # Get the claim from the payload and return it
        return payload.claims[claim]
    except KeyError as err:
        logger.error(f"Claim '{claim}' not found in token: {err}", exc_info=err, extra={"token": "[REDACTED]"})
        raise jafaal_exceptions.InvalidTokenError(f"Claim '{claim}' is missing in the token.") from err
    except jafaal_exceptions.JafaalError:
        # decode_token already raised a properly-formed 401; re-raise as-is.
        raise
    except Exception as err:
        logger.error(
            f"Unexpected error retrieving claim: {type(err).__name__}", exc_info=err, extra={"token": "[REDACTED]"}
        )
        raise jafaal_exceptions.InvalidTokenError("Unable to retrieve claim") from err

decode_token

decode_token(token)

Decodes a JWT token and returns the parsed Token object.

The algorithms allow-list (pinned to this manager's algorithm) is always passed to jwt.decode: without it joserfc would trust whatever algorithm the token header advertises (none or an algorithm-confusion variant), bypassing the signature check.

In symmetric (HS256) mode the token is verified against the primary key then each rotation fallback. In asymmetric mode it is verified against the public-key set, which joserfc selects by the header kid (the active key plus any rotation fallbacks).

Parameters:

Name	Type	Description	Default
token	str	The JWT token to decode.	required

Returns:

Type	Description
Token	joserfc.jwt.Token: The decoded token (use .claims to access payload claims).

Raises:

Type	Description
JafaalError	If the token cannot be decoded, raises an HTTP 401 Unauthorized exception.

Source code in jafaal/_internal/token_manager.py

def decode_token(self, token: str) -> Token:
    """
    Decodes a JWT token and returns the parsed Token object.

    The ``algorithms`` allow-list (pinned to this manager's algorithm) is
    always passed to ``jwt.decode``: without it joserfc would trust whatever
    algorithm the token header advertises (``none`` or an
    algorithm-confusion variant), bypassing the signature check.

    In symmetric (HS256) mode the token is verified against the primary key
    then each rotation fallback. In asymmetric mode it is verified against
    the public-key set, which joserfc selects by the header ``kid`` (the
    active key plus any rotation fallbacks).

    Args:
        token (str): The JWT token to decode.

    Returns:
        joserfc.jwt.Token: The decoded token (use ``.claims`` to access
            payload claims).

    Raises:
        JafaalError: If the token cannot be decoded, raises an HTTP 401
            Unauthorized exception.
    """
    if self._is_symmetric:
        return self._decode_symmetric(token)
    return self._decode_asymmetric(token)

validate_token_expiration

validate_token_expiration(token, expected_type)

Validates expiration, required claims, and type of a JWT.

Checks that the token contains all essential claims, is not expired or used before its valid time, and that it names the expected token use. This prevents refresh tokens from being used as access tokens and vice versa.

Parameters:

Name	Type	Description	Default
token	str	The JWT token to validate.	required
expected_type	TokenType	The expected token type (TokenType.ACCESS or TokenType.REFRESH).	required

Raises:

Type	Description
JafaalError	If the token is missing required claims, expired, not yet valid, contains invalid claims, or has the wrong type.

Source code in jafaal/_internal/token_manager.py

def validate_token_expiration(
    self,
    token: str,
    expected_type: TokenType,
) -> None:
    """
    Validates expiration, required claims, and type of a JWT.

    Checks that the token contains all essential claims, is not expired or
    used before its valid time, and that it names the expected token use.
    This prevents refresh tokens from being used as access tokens and vice
    versa.

    Args:
        token: The JWT token to validate.
        expected_type: The expected token type
            (``TokenType.ACCESS`` or
            ``TokenType.REFRESH``).

    Raises:
        JafaalError: If the token is missing required claims, expired,
            not yet valid, contains invalid claims, or has the wrong type.
    """
    try:
        # Define required claims. ``leeway`` applies a small clock-skew
        # tolerance to the time-based claims (``exp`` / ``nbf``) so slightly
        # skewed nodes do not spuriously reject otherwise-valid tokens.
        # The token-use claim is checked separately below.
        claims_requests = jwt.JWTClaimsRegistry(
            leeway=self.leeway_seconds,
            sid={"essential": True},
            iss={
                "essential": True,
                "value": self.issuer,
            },
            aud={
                "essential": True,
                "value": self.audience,
            },
            sub={"essential": True},
            scope={"essential": True},
            iat={"essential": True},
            nbf={"essential": True},
            exp={"essential": True},
            jti={"essential": True},
        )

        # Decode the token to get the payload
        payload = self.decode_token(token)
        _validate_typ_header(payload.header, expected_type)
        _validate_token_use(payload.claims, expected_type)

        # Validate token claims (incl. expiration and typ)
        claims_requests.validate(payload.claims)
    except MissingClaimError as missing_err:
        logger.error(f"JWT missing claim error: {missing_err}", exc_info=missing_err, extra={"token": "[REDACTED]"})
        raise jafaal_exceptions.InvalidTokenError("Token is missing required claims.") from missing_err
    except ExpiredTokenError as expired_err:
        raise jafaal_exceptions.TokenExpiredError("Token is expired.") from expired_err
    except InvalidTokenError as invalid_err:
        logger.error(
            f"JWT is not valid yet error: {invalid_err}", exc_info=invalid_err, extra={"token": "[REDACTED]"}
        )
        raise jafaal_exceptions.InvalidTokenError("Token is not valid yet.") from invalid_err
    except InsecureClaimError as insecure_err:
        logger.error(
            f"JWT insecure claim error: {insecure_err}", exc_info=insecure_err, extra={"token": "[REDACTED]"}
        )
        raise jafaal_exceptions.InvalidTokenError("Token has insecure claims.") from insecure_err
    except InvalidClaimError as claims_err:
        logger.error(
            f"JWT claims validation error: {claims_err}", exc_info=claims_err, extra={"token": "[REDACTED]"}
        )
        raise jafaal_exceptions.InvalidTokenError("Token has invalid claims.") from claims_err
    except jafaal_exceptions.JafaalError:
        # decode_token already raised a properly-formed 401; re-raise as-is.
        raise
    except Exception as err:
        logger.error(
            f"Unexpected error validating token: {type(err).__name__}", exc_info=err, extra={"token": "[REDACTED]"}
        )
        raise jafaal_exceptions.InvalidTokenError("Token expired or invalid.") from err

validate_access_expiration_logged

validate_access_expiration_logged(access_token)

Validate an access token's expiration and log failures consistently.

Wraps :meth:validate_token_expiration for TokenType.ACCESS and applies the shared logging policy used by the access-token validation dependency and IdentityService: expired tokens log at debug (an expected, routine condition) while all other validation failures log at error. The original :class:JafaalError is re-raised unchanged so callers keep the same 401 semantics.

Parameters:

Name	Type	Description	Default
access_token	str	The raw JWT access token to validate.	required

Raises:

Type	Description
JafaalError	401 if the token is missing claims, expired, not yet valid, or otherwise invalid.

Source code in jafaal/_internal/token_manager.py

def validate_access_expiration_logged(self, access_token: str) -> None:
    """
    Validate an access token's expiration and log failures consistently.

    Wraps :meth:`validate_token_expiration` for ``TokenType.ACCESS`` and
    applies the shared logging policy used by the access-token validation
    dependency and ``IdentityService``: expired tokens log at ``debug``
    (an expected, routine condition) while all other validation failures
    log at ``error``. The original :class:`JafaalError` is re-raised
    unchanged so callers keep the same 401 semantics.

    Args:
        access_token: The raw JWT access token to validate.

    Raises:
        JafaalError: 401 if the token is missing claims, expired, not
            yet valid, or otherwise invalid.
    """
    try:
        self.validate_token_expiration(access_token, TokenType.ACCESS)
    except jafaal_exceptions.JafaalError as http_err:
        is_expired = isinstance(http_err, jafaal_exceptions.TokenExpiredError)
        logger.log(
            logging.DEBUG if is_expired else logging.ERROR,
            f"Access token validation failed: {http_err.detail}",
            exc_info=None if is_expired else http_err,
            extra={"access_token": "[REDACTED]"},
        )
        raise

create_token

create_token(
    session_id,
    user,
    token_type,
    client=None,
    requested_scope=None,
)

Creates a JWT token for a user session with appropriate access scope and expiration.

Parameters:

Name	Type	Description	Default
session_id	str	The unique identifier for the session.	required
user	UserProtocol	The user object containing user details.	required
token_type	TokenType	The type of token to create (access or refresh).	required
client	OAuthClient | None	The registered client the token is being issued to. Its scope ceiling narrows the user's grants and its client_id becomes the RFC 9068 §2.2 client_id claim. None falls back to the deployment-wide identifier.	None
requested_scope	Sequence[str] | None	The scope the client asked for (RFC 6749 §3.3), applied as a final narrowing bound. None means "everything this client and user are entitled to".	None

Returns:

Type	Description
tuple[datetime, str]	tuple[datetime, str]: A tuple containing the token's expiration datetime and the encoded JWT token string.

Raises:

Type	Description
ValueError	If required parameters are missing or invalid.

Source code in jafaal/_internal/token_manager.py

def create_token(
    self,
    session_id: str,
    user: jafaal_ports.UserProtocol,
    token_type: TokenType,
    client: jafaal_settings.OAuthClient | None = None,
    requested_scope: Sequence[str] | None = None,
) -> tuple[datetime, str]:
    """
    Creates a JWT token for a user session with appropriate access scope
    and expiration.

    Args:
        session_id (str): The unique identifier for the session.
        user (jafaal_ports.UserProtocol): The user object containing user
            details.
        token_type (TokenType): The type of token to create (access or
            refresh).
        client: The registered client the token is being issued to. Its
            scope ceiling narrows the user's grants and its ``client_id``
            becomes the RFC 9068 §2.2 ``client_id`` claim. ``None`` falls
            back to the deployment-wide identifier.
        requested_scope: The ``scope`` the client asked for (RFC 6749 §3.3),
            applied as a final narrowing bound. ``None`` means "everything
            this client and user are entitled to".

    Returns:
        tuple[datetime, str]: A tuple containing the token's expiration
            datetime and the encoded JWT token string.

    Raises:
        ValueError: If required parameters are missing or invalid.
    """
    # Three bounds, applied in order, each of which can only ever remove:
    #
    # 1. the host's ScopeResolver port decides what the *account* holds (the
    #    built-in default is the catalog's is_superuser two-tier mapping), so
    #    an application with a richer authorisation model stamps its own
    #    grants without patching the token minter;
    # 2. the registered client's ceiling caps what *this client* may ever
    #    carry; and
    # 3. the client's own ``scope`` request caps what it asked for on *this*
    #    exchange, so a client that deliberately asks for less gets less.
    scope = jafaal_ports.get_scope_resolver().scopes_for(user)
    if client is not None:
        scope = client.narrow(scope)
    scope = jafaal_scopes.narrow_to_requested(scope, requested_scope)

    # Set now
    issued_at = datetime.now(UTC)
    lifetime = (
        timedelta(days=self.refresh_token_expire_days)
        if token_type == TokenType.REFRESH
        else timedelta(minutes=self.access_token_expire_minutes)
    )
    exp = issued_at + lifetime
    now = int(issued_at.timestamp())

    claims: dict[str, Any] = {
        "sid": session_id,
        "iss": self.issuer,
        "aud": self.audience,
        "iat": now,
        # Backdated by a few seconds. ``nbf == iat`` is correct in the
        # abstract and a reliable interop failure in practice: a *resource
        # server* verifying with a stock JWT library and a clock a second
        # behind ours rejects a token that was minted milliseconds earlier,
        # and it has no ``leeway_seconds`` of ours to fall back on. RFC 7519
        # §4.1.5 anticipates exactly this ("implementers MAY provide for
        # some small leeway"); applying it at issuance rather than asking
        # every verifier to configure it is what makes the token portable.
        # Small enough not to widen any meaningful attack window: the token
        # is already valid from the instant it is handed out.
        "nbf": now - _NBF_BACKDATE_SECONDS,
        "exp": exp,
        "jti": str(uuid.uuid4()),
        # RFC 9068 / RFC 7519 shapes, so a resource server verifying against
        # the published JWKS with a stock JWT library reads what it expects:
        # ``sub`` is a string (RFC 7519 §4.1.2 defines it as StringOrURI),
        # ``scope`` is space-delimited (RFC 6749 §3.3), and ``client_id`` is
        # present (RFC 9068 §2.2). ``coerce_user_id`` converts ``sub`` back
        # to the host user table's primary-key type on the way in.
        "sub": str(user.id),
        "scope": " ".join(scope),
        "client_id": client.client_id if client is not None else self.client_id,
        TOKEN_USE_CLAIM: token_type.value,
    }
    # The media type goes in the JOSE ``typ`` header, not a payload claim.
    header = {**self._sign_header, "typ": _TYP_HEADER_BY_TOKEN_TYPE[token_type]}

    encoded_token = jwt.encode(
        header,
        claims.copy(),
        self._sign_key,
        algorithms=self._encode_algorithms,
    )

    # Return the expiration and the encoded token
    return exp, encoded_token

create_csrf_token staticmethod

create_csrf_token()

Generate a secure random CSRF (Cross-Site Request Forgery) token.

Returns:

Name	Type	Description
str	str	A URL-safe, securely generated random string suitable for use as a CSRF token.

Source code in jafaal/_internal/token_manager.py

@staticmethod
def create_csrf_token() -> str:
    """
    Generate a secure random CSRF (Cross-Site Request Forgery) token.

    Returns:
        str: A URL-safe, securely generated random string suitable for use
            as a CSRF token.
    """
    return secrets.token_urlsafe(32)

jwks

jwks()

Return the JSON Web Key Set of public verification keys.

Empty ({"keys": []}) in symmetric (HS256) mode, which has no public key to publish. In asymmetric mode it contains the active signing key's public JWK plus any rotation fallbacks, each tagged with its kid (RFC 7638 thumbprint), use: "sig", and alg — exactly what a resource server needs to verify JAFAAL's access tokens statelessly.

Source code in jafaal/_internal/token_manager.py

def jwks(self) -> dict[str, Any]:
    """Return the JSON Web Key Set of public verification keys.

    Empty (``{"keys": []}``) in symmetric (HS256) mode, which has no public
    key to publish. In asymmetric mode it contains the active signing key's
    public JWK plus any rotation fallbacks, each tagged with its ``kid``
    (RFC 7638 thumbprint), ``use: "sig"``, and ``alg`` — exactly what a
    resource server needs to verify JAFAAL's access tokens statelessly.
    """
    return {"keys": [jwk_keys.jwk_entry(key, self.algorithm) for key in self._verify_keys]}

AuthenticationError

Bases: JafaalError

The caller could not be authenticated (401).

Carries an RFC 6750 §3 WWW-Authenticate challenge. The base case — no credential was supplied — is a bare Bearer, because §3 says the error parameter MUST NOT be sent when the request contained no authentication information. Subclasses that represent a presented but unusable credential set :attr:bearer_error (almost always invalid_token, per §3.1), which is what lets a client tell "log in" from "refresh and retry" without parsing prose.

Source code in jafaal/exceptions.py

class AuthenticationError(JafaalError):
    """The caller could not be authenticated (401).

    Carries an RFC 6750 §3 ``WWW-Authenticate`` challenge. The base case — no
    credential was supplied — is a bare ``Bearer``, because §3 says the
    ``error`` parameter MUST NOT be sent when the request contained no
    authentication information. Subclasses that represent a *presented but
    unusable* credential set :attr:`bearer_error` (almost always
    ``invalid_token``, per §3.1), which is what lets a client tell "log in" from
    "refresh and retry" without parsing prose.
    """

    code = "authentication_error"
    status_code = 401
    default_detail = "Authentication required."

    #: RFC 6750 §3.1 error code for the challenge, or ``None`` for a bare
    #: ``Bearer`` (no credential was presented, so there is nothing wrong with).
    bearer_error: str | None = None

    #: Auth scheme named in the challenge. Overridden by API-key errors.
    auth_scheme: str = "Bearer"

    def __init__(self, detail: str | None = None, *, headers: dict[str, str] | None = None) -> None:
        super().__init__(detail, headers=headers)
        if headers is None:
            self.headers = self._challenge()

    def _challenge(self) -> dict[str, str]:
        """Build the ``WWW-Authenticate`` challenge for this failure."""
        challenge = self.auth_scheme
        if self.bearer_error:
            challenge += f' error="{self.bearer_error}", error_description="{_quote_safe(self.detail)}"'
        return {"WWW-Authenticate": challenge}

AuthorizationError

Bases: JafaalError

The caller is authenticated but not permitted (403).

Source code in jafaal/exceptions.py

class AuthorizationError(JafaalError):
    """The caller is authenticated but not permitted (403)."""

    code = "authorization_error"
    status_code = 403
    default_detail = "You do not have permission to perform this action."

ConflictError

Bases: JafaalError

The request conflicts with the current state (409).

Source code in jafaal/exceptions.py

class ConflictError(JafaalError):
    """The request conflicts with the current state (409)."""

    code = "conflict"
    status_code = 409
    default_detail = "The request conflicts with the current state."

IdentityProviderError

Bases: UpstreamError

An external identity provider returned an error (502).

Source code in jafaal/exceptions.py

class IdentityProviderError(UpstreamError):
    """An external identity provider returned an error (502)."""

    code = "identity_provider_error"
    default_detail = "The identity provider returned an error."

IdentityProviderTimeoutError

Bases: UpstreamTimeoutError

An external identity provider timed out (504).

Source code in jafaal/exceptions.py

class IdentityProviderTimeoutError(UpstreamTimeoutError):
    """An external identity provider timed out (504)."""

    code = "identity_provider_timeout"
    default_detail = "The identity provider timed out."

InactiveAccountError

Bases: AuthenticationError

The credential is well-formed but its account can no longer be used.

A deactivated (or deleted) account is a credential-validation failure, not an authorization one: RFC 6750 §3.1 puts "the access token is revoked… or otherwise invalid" under invalid_token with a 401. Answering 403 would tell a bearer that its token is still good and only the permission is missing, and answering 404 would make the resource server's account state observable to whoever holds a stale token.

Source code in jafaal/exceptions.py

class InactiveAccountError(AuthenticationError):
    """The credential is well-formed but its account can no longer be used.

    A deactivated (or deleted) account is a **credential-validation** failure,
    not an authorization one: RFC 6750 §3.1 puts "the access token is
    revoked… or otherwise invalid" under ``invalid_token`` with a 401. Answering
    403 would tell a bearer that its token is still good and only the permission
    is missing, and answering 404 would make the resource server's account state
    observable to whoever holds a stale token.
    """

    code = "inactive_account"
    default_detail = "This account is not active."
    bearer_error = "invalid_token"

InternalError

Bases: JafaalError

An unexpected internal error (500).

Source code in jafaal/exceptions.py

class InternalError(JafaalError):
    """An unexpected internal error (500)."""

    code = "internal_error"
    status_code = 500
    default_detail = "An internal error occurred."

InvalidApiKeyError

Bases: AuthenticationError

The supplied API key is unknown, revoked, or malformed.

Source code in jafaal/exceptions.py

class InvalidApiKeyError(AuthenticationError):
    """The supplied API key is unknown, revoked, or malformed."""

    code = "invalid_api_key"
    default_detail = "The API key is invalid."
    # Not an IANA-registered auth scheme, but neither is the X-API-Key header it
    # answers; naming it keeps the challenge honest about what to retry with.
    auth_scheme = "ApiKey"

InvalidCredentialsError

Bases: AuthenticationError

Username/password (or equivalent) did not verify.

Source code in jafaal/exceptions.py

class InvalidCredentialsError(AuthenticationError):
    """Username/password (or equivalent) did not verify."""

    code = "invalid_credentials"
    default_detail = "Unable to authenticate with provided credentials."

InvalidMFACodeError

Bases: InvalidRequestError

A supplied TOTP/backup MFA code did not verify (400).

Source code in jafaal/exceptions.py

class InvalidMFACodeError(InvalidRequestError):
    """A supplied TOTP/backup MFA code did not verify (400)."""

    code = "invalid_mfa_code"
    default_detail = "Invalid MFA code."

InvalidRequestError

Bases: JafaalError

The request is malformed or semantically invalid (400).

Source code in jafaal/exceptions.py

class InvalidRequestError(JafaalError):
    """The request is malformed or semantically invalid (400)."""

    code = "invalid_request"
    status_code = 400
    default_detail = "The request is invalid."

InvalidTokenError

Bases: AuthenticationError

A JWT is malformed, has a bad signature, or fails claim validation.

Source code in jafaal/exceptions.py

class InvalidTokenError(AuthenticationError):
    """A JWT is malformed, has a bad signature, or fails claim validation."""

    code = "invalid_token"
    default_detail = "The token is invalid."
    bearer_error = "invalid_token"

JafaalError

Bases: Exception

Base class for all JAFAAL domain errors.

Source code in jafaal/exceptions.py

class JafaalError(Exception):
    """Base class for all JAFAAL domain errors."""

    code: str = "error"
    status_code: int = 500
    default_detail: str = "An unexpected error occurred."
    headers: dict[str, str] | None = None

    def __init__(self, detail: str | None = None, *, headers: dict[str, str] | None = None) -> None:
        self.detail = detail if detail is not None else self.default_detail
        if headers is not None:
            self.headers = headers
        super().__init__(self.detail)

MissingScopeError

Bases: AuthorizationError

The principal lacks one or more required scopes.

Carries an RFC 6750 §3 WWW-Authenticate challenge built from the scopes the endpoint requires: Bearer error="insufficient_scope", scope="users:read users:write". The scope attribute is a space-delimited list per RFC 6749 §3.3 — a client parsing the challenge to decide what to re-request needs that exact shape, so it is built here rather than at each raise site.

Source code in jafaal/exceptions.py

class MissingScopeError(AuthorizationError):
    """The principal lacks one or more required scopes.

    Carries an RFC 6750 §3 ``WWW-Authenticate`` challenge built from the scopes
    the endpoint requires:
    ``Bearer error="insufficient_scope", scope="users:read users:write"``. The
    ``scope`` attribute is a **space-delimited** list per RFC 6749 §3.3 — a
    client parsing the challenge to decide what to re-request needs that exact
    shape, so it is built here rather than at each raise site.
    """

    code = "missing_scope"
    default_detail = "You do not have the required permissions."

    def __init__(
        self,
        detail: str | None = None,
        *,
        missing: frozenset[str] | set[str] | None = None,
        required: frozenset[str] | set[str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> None:
        self.missing: frozenset[str] = frozenset(missing or ())
        self.required: frozenset[str] = frozenset(required or ()) or self.missing
        super().__init__(detail, headers=headers or self._challenge())

    def _challenge(self) -> dict[str, str]:
        """Build the RFC 6750 ``insufficient_scope`` challenge header."""
        challenge = 'Bearer error="insufficient_scope"'
        if self.required:
            challenge += f', scope="{" ".join(sorted(self.required))}"'
        return {"WWW-Authenticate": challenge}

NotFoundError

Bases: JafaalError

The requested resource does not exist (404).

Source code in jafaal/exceptions.py

class NotFoundError(JafaalError):
    """The requested resource does not exist (404)."""

    code = "not_found"
    status_code = 404
    default_detail = "The requested resource was not found."

PasswordChangeRequiredError

Bases: AuthenticationError

The password is correct but must be replaced before it can be used.

Raised when a credential was written with must_change=True — an operator seeding the first administrator, or a CLI reset-password. Such a password is known to whoever set it, so allowing it to stay in use indefinitely would make a bootstrap credential a permanent one.

Distinct from :class:InvalidCredentialsError on purpose: the caller needs to know the password was right and that the remedy is to replace it, not to retry. There is no enumeration concern in the distinction, because reaching it already required presenting the correct password.

Source code in jafaal/exceptions.py

class PasswordChangeRequiredError(AuthenticationError):
    """The password is correct but must be replaced before it can be used.

    Raised when a credential was written with ``must_change=True`` — an operator
    seeding the first administrator, or a CLI ``reset-password``. Such a password
    is known to whoever set it, so allowing it to stay in use indefinitely would
    make a bootstrap credential a permanent one.

    Distinct from :class:`InvalidCredentialsError` on purpose: the caller needs
    to know the password was right and that the remedy is to replace it, not to
    retry. There is no enumeration concern in the distinction, because reaching
    it already required presenting the correct password.
    """

    code = "password_change_required"
    default_detail = "This password must be changed before you can sign in."

PreconditionFailedError

Bases: JafaalError

A precondition for the request was not met (412).

Source code in jafaal/exceptions.py

class PreconditionFailedError(JafaalError):
    """A precondition for the request was not met (412)."""

    code = "precondition_failed"
    status_code = 412
    default_detail = "A precondition for this request was not met."

RateLimitedError

Bases: JafaalError

The caller has been rate limited (429).

retry_after (seconds) is surfaced as the Retry-After header.

Source code in jafaal/exceptions.py

class RateLimitedError(JafaalError):
    """The caller has been rate limited (429).

    ``retry_after`` (seconds) is surfaced as the ``Retry-After`` header.
    """

    code = "rate_limited"
    status_code = 429
    default_detail = "Too many requests. Please try again later."

    def __init__(
        self,
        detail: str | None = None,
        *,
        retry_after: int | None = None,
        headers: dict[str, str] | None = None,
    ) -> None:
        self.retry_after = retry_after
        merged = dict(headers) if headers else {}
        if retry_after is not None and "Retry-After" not in merged:
            merged["Retry-After"] = str(retry_after)
        super().__init__(detail, headers=merged or None)

ServiceUnavailableError

Bases: JafaalError

A required dependency is temporarily unavailable (503).

Source code in jafaal/exceptions.py

class ServiceUnavailableError(JafaalError):
    """A required dependency is temporarily unavailable (503)."""

    code = "service_unavailable"
    status_code = 503
    default_detail = "The service is temporarily unavailable."

SessionExpiredError

Bases: AuthenticationError

The server-side session is missing or expired.

Source code in jafaal/exceptions.py

class SessionExpiredError(AuthenticationError):
    """The server-side session is missing or expired."""

    code = "session_expired"
    default_detail = "The session has expired. Please log in again."
    bearer_error = "invalid_token"

StaleRefreshTokenError

Bases: AuthenticationError

A rotated/replayed refresh token was presented; clear the cookie.

Replaces the old ClearRefreshTokenCookieHTTPException: the edge handler reads :attr:clear_refresh_cookie and emits the refresh-cookie deletion headers on the response.

Source code in jafaal/exceptions.py

class StaleRefreshTokenError(AuthenticationError):
    """A rotated/replayed refresh token was presented; clear the cookie.

    Replaces the old ``ClearRefreshTokenCookieHTTPException``: the edge handler
    reads :attr:`clear_refresh_cookie` and emits the refresh-cookie deletion
    headers on the response.
    """

    code = "stale_refresh_token"
    default_detail = "The refresh token is no longer valid. Please log in again."
    bearer_error = "invalid_token"
    clear_refresh_cookie = True

StepUpReauthRequiredError

Bases: AuthenticationError

Step-up needs a fresh identity-provider re-authentication.

Raised for an SSO-only account (no local password and no MFA) that has at least one usable identity-provider link: a valid access token alone cannot satisfy step-up, so the caller must complete a fresh IdP re-authentication to obtain a single-use step-up grant and then retry the operation. reauth_idp_ids lists the linked providers eligible for re-authentication.

Aligns with RFC 9470 (OAuth 2.0 Step-Up Authentication): the WWW-Authenticate header advertises insufficient_user_authentication so a standards-aware client knows to trigger a stronger authentication.

Source code in jafaal/exceptions.py

class StepUpReauthRequiredError(AuthenticationError):
    """Step-up needs a fresh identity-provider re-authentication.

    Raised for an SSO-only account (no local password and no MFA) that has at
    least one usable identity-provider link: a valid access token alone cannot
    satisfy step-up, so the caller must complete a fresh IdP re-authentication
    to obtain a single-use step-up grant and then retry the operation.
    ``reauth_idp_ids`` lists the linked providers eligible for re-authentication.

    Aligns with RFC 9470 (OAuth 2.0 Step-Up Authentication): the
    ``WWW-Authenticate`` header advertises ``insufficient_user_authentication``
    so a standards-aware client knows to trigger a stronger authentication.
    """

    code = "step_up_reauth_required"
    default_detail = "Re-authenticate with your identity provider to continue."
    headers = {"WWW-Authenticate": 'Bearer error="insufficient_user_authentication"'}

    def __init__(
        self,
        detail: str | None = None,
        *,
        reauth_idp_ids: list[int] | tuple[int, ...] | None = None,
        headers: dict[str, str] | None = None,
    ) -> None:
        self.reauth_idp_ids: list[int] = list(reauth_idp_ids or ())
        super().__init__(detail, headers=headers)

StoreUnavailableError

Bases: ServiceUnavailableError

A security state store (lockout counters, MFA secret) is unreachable.

Unifies the former AuthSecurityStoreUnavailableError and MFASecretStoreUnavailableError.

Source code in jafaal/exceptions.py

class StoreUnavailableError(ServiceUnavailableError):
    """A security state store (lockout counters, MFA secret) is unreachable.

    Unifies the former ``AuthSecurityStoreUnavailableError`` and
    ``MFASecretStoreUnavailableError``.
    """

    code = "store_unavailable"
    default_detail = "A required storage backend is unavailable."

TokenExpiredError

Bases: AuthenticationError

A JWT (access/refresh) has expired.

Source code in jafaal/exceptions.py

class TokenExpiredError(AuthenticationError):
    """A JWT (access/refresh) has expired."""

    code = "token_expired"
    default_detail = "The token has expired."
    bearer_error = "invalid_token"

UnprocessableError

Bases: InvalidRequestError

The request is well-formed but cannot be processed (422).

Source code in jafaal/exceptions.py

class UnprocessableError(InvalidRequestError):
    """The request is well-formed but cannot be processed (422)."""

    code = "unprocessable"
    status_code = 422
    default_detail = "The request could not be processed."

UpstreamError

Bases: JafaalError

An upstream provider returned a bad response (502).

Source code in jafaal/exceptions.py

class UpstreamError(JafaalError):
    """An upstream provider returned a bad response (502)."""

    code = "upstream_error"
    status_code = 502
    default_detail = "The upstream provider returned an error."

UpstreamTimeoutError

Bases: UpstreamError

An upstream provider timed out (504).

Source code in jafaal/exceptions.py

class UpstreamTimeoutError(UpstreamError):
    """An upstream provider timed out (504)."""

    code = "upstream_timeout"
    status_code = 504
    default_detail = "The upstream provider timed out."

RouterPrefixes dataclass

Sub-prefixes for the aggregated routers (relative to the host's API root).

Defaults assume the aggregate is mounted under /api/v1 and match the path assumptions in :class:~jafaal.settings.AuthSettings.

Source code in jafaal/factory.py

@dataclass(frozen=True)
class RouterPrefixes:
    """Sub-prefixes for the aggregated routers (relative to the host's API root).

    Defaults assume the aggregate is mounted under ``/api/v1`` and match the
    path assumptions in :class:`~jafaal.settings.AuthSettings`.
    """

    auth: str = "/auth"
    sessions: str = "/auth/sessions"
    api_keys: str = "/auth/api-keys"
    identity_providers: str = "/auth/idp"
    identity_providers_public: str = "/public/idp"
    password_reset: str = "/auth"
    sign_up: str = "/auth"
    webauthn: str = "/auth/webauthn"
    webauthn_public: str = "/public/webauthn"

Base

Bases: DeclarativeBase

JAFAAL's convenience declarative base.

Under Option B the host owns the registry: define your own :class:~sqlalchemy.orm.DeclarativeBase, build Users on it, and pass it to :func:map_models. Use this default base only if you would rather not own one — build the user model on it and call map_models() without a base.

Source code in jafaal/orm.py

class Base(DeclarativeBase):
    """JAFAAL's convenience declarative base.

    Under Option B the **host** owns the registry: define your own
    :class:`~sqlalchemy.orm.DeclarativeBase`, build ``Users`` on it, and pass it
    to :func:`map_models`. Use this default base only if you would rather not own
    one — build the user model on it and call ``map_models()`` without a base.
    """

AccountLocked dataclass

Progressive lockout was applied to a login / MFA / step-up subject.

Emitted best-effort when a lockout tier trips so the host can notify the account owner. subject is the locked value (a username or an IP address), subject_kind distinguishes the two, and store names the flow ("Login" / "MFA" / "Step-up").

Source code in jafaal/ports.py

@dataclass(frozen=True)
class AccountLocked:
    """Progressive lockout was applied to a login / MFA / step-up subject.

    Emitted best-effort when a lockout tier trips so the host can notify the
    account owner. ``subject`` is the locked value (a username or an IP address),
    ``subject_kind`` distinguishes the two, and ``store`` names the flow
    (``"Login"`` / ``"MFA"`` / ``"Step-up"``).
    """

    subject: str
    subject_kind: str
    store: str
    failed_attempts: int
    lockout_label: str

AuthenticatorChanged dataclass

An authentication factor was added to or removed from an account.

Covers TOTP enable/disable, backup-code regeneration, and passkey registration/deletion. Binding or unbinding an authenticator changes how the account can be signed into, so the owner has to hear about it out of band — an attacker who enrols their own factor (or strips the victim's) otherwise does so in total silence. remaining_factors lets a host warn loudly when an account is left with none.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class AuthenticatorChanged:
    """An authentication factor was added to or removed from an account.

    Covers TOTP enable/disable, backup-code regeneration, and passkey
    registration/deletion. Binding or unbinding an authenticator changes *how
    the account can be signed into*, so the owner has to hear about it out of
    band — an attacker who enrols their own factor (or strips the victim's)
    otherwise does so in total silence. ``remaining_factors`` lets a host warn
    loudly when an account is left with none.
    """

    user_id: Any
    username: str
    #: ``"totp"``, ``"backup_codes"``, or ``"passkey"``.
    factor: str
    #: ``"added"`` or ``"removed"``.
    change: str
    remaining_factors: int | None = None

AuthEventSink

Bases: Protocol

Host-owned delivery of JAFAAL's outbound notifications.

JAFAAL performs the security-critical work (mint/hash/store token, single-use + expiry, enumeration-safe response) and emits these events; the host delivers them via email/SMS/websocket/queue/log. All methods are awaited best-effort — for enumeration-safe flows JAFAAL swallows and logs delivery failures so they cannot change the HTTP response or leak whether an account exists.

Source code in jafaal/ports.py

class AuthEventSink(Protocol):
    """Host-owned delivery of JAFAAL's outbound notifications.

    JAFAAL performs the security-critical work (mint/hash/store token,
    single-use + expiry, enumeration-safe response) and emits these events; the
    host delivers them via email/SMS/websocket/queue/log. All methods are
    awaited best-effort — for enumeration-safe flows JAFAAL swallows and logs
    delivery failures so they cannot change the HTTP response or leak whether an
    account exists.
    """

    async def on_password_reset_requested(self, event: PasswordResetRequested) -> None: ...

    async def on_email_verification_requested(self, event: EmailVerificationRequested) -> None: ...

    async def on_signup_pending_admin_approval(self, event: SignupPendingAdminApproval) -> None: ...

    async def on_signup_approved(self, event: SignupApproved) -> None: ...

    # --- Security events (best-effort, fire-and-forget) ---
    # Emitted from the auth flow via jafaal.ports.dispatch_event / adispatch_event,
    # which skip a sink that does not implement the method — so a host sink written
    # before these existed keeps working without change.

    async def on_new_device_login(self, event: NewDeviceLogin) -> None: ...

    async def on_account_locked(self, event: AccountLocked) -> None: ...

    async def on_refresh_token_theft_detected(self, event: RefreshTokenTheftDetected) -> None: ...

    async def on_idp_account_linked(self, event: IdpAccountLinked) -> None: ...

    async def on_authenticator_changed(self, event: AuthenticatorChanged) -> None: ...

EmailVerificationRequested dataclass

A sign-up needs email verification; deliver token to email.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class EmailVerificationRequested:
    """A sign-up needs email verification; deliver ``token`` to ``email``."""

    user_id: Any
    email: str
    display_name: str | None
    token: str
    expires_at: datetime
    locale: str | None

IdpAccountLinked dataclass

An identity provider was linked to an existing account by matching email.

Emitted when an SSO login adopts a pre-existing local account rather than creating one — a new way to sign in that the owner did not initiate from a session they already held. Tell them out of band, so a link they did not expect is visible rather than silent.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class IdpAccountLinked:
    """An identity provider was linked to an existing account by matching email.

    Emitted when an SSO login adopts a pre-existing local account rather than
    creating one — a new way to sign in that the owner did not initiate from a
    session they already held. Tell them out of band, so a link they did not
    expect is visible rather than silent.
    """

    user_id: Any
    username: str
    idp_name: str
    idp_slug: str
    email: str

IdpIdentity dataclass

An identity resolved from an external identity provider.

Handed to the host so it can provision (or sync) its own user row with its own profile shape/defaults. claims carries the raw mapped IdP claims so the host can pick whatever additional fields it wants.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class IdpIdentity:
    """An identity resolved from an external identity provider.

    Handed to the host so it can provision (or sync) its own user row with its
    own profile shape/defaults. ``claims`` carries the raw mapped IdP claims so
    the host can pick whatever additional fields it wants.
    """

    subject: str
    idp_id: int
    email: str | None
    email_verified: bool
    suggested_username: str
    display_name: str | None
    claims: Mapping[str, Any]

NewDeviceLogin dataclass

A user signed in from a device/browser not seen on any prior session.

Emitted best-effort after a successful login so the host can alert the user ("new sign-in from …"). device_description is a human-readable summary parsed from the User-Agent (browser + OS).

Source code in jafaal/ports.py

@dataclass(frozen=True)
class NewDeviceLogin:
    """A user signed in from a device/browser not seen on any prior session.

    Emitted best-effort after a successful login so the host can alert the user
    ("new sign-in from …"). ``device_description`` is a human-readable summary
    parsed from the User-Agent (browser + OS).
    """

    user_id: Any
    username: str
    ip: str | None
    device_description: str
    session_id: str

NullAuthEventSink

Default no-op sink — a host that skips these flows implements nothing.

Source code in jafaal/ports.py

class NullAuthEventSink:
    """Default no-op sink — a host that skips these flows implements nothing."""

    async def on_password_reset_requested(self, event: PasswordResetRequested) -> None:
        return None

    async def on_email_verification_requested(self, event: EmailVerificationRequested) -> None:
        return None

    async def on_signup_pending_admin_approval(self, event: SignupPendingAdminApproval) -> None:
        return None

    async def on_signup_approved(self, event: SignupApproved) -> None:
        return None

    async def on_new_device_login(self, event: NewDeviceLogin) -> None:
        return None

    async def on_account_locked(self, event: AccountLocked) -> None:
        return None

    async def on_refresh_token_theft_detected(self, event: RefreshTokenTheftDetected) -> None:
        return None

    async def on_idp_account_linked(self, event: IdpAccountLinked) -> None:
        return None

    async def on_authenticator_changed(self, event: AuthenticatorChanged) -> None:
        return None

NullPasswordBreachChecker

Default checker that treats every password as not breached (no-op).

Source code in jafaal/ports.py

class NullPasswordBreachChecker:
    """Default checker that treats every password as not breached (no-op)."""

    def is_breached(self, password: str) -> bool:
        return False

PasswordBreachChecker

Bases: Protocol

Host-owned check for whether a password appears in a breach corpus/blocklist.

Consulted during sign-up and password change, after the length/complexity policy passes and before the password is hashed. Return True to reject the password. This is the NIST SP 800-63B-recommended companion to the length_only policy — a common implementation is an HIBP k-anonymity range query or a local blocklist.

It runs in the request path (synchronously), so keep it fast and fail open (return False) on an upstream error, so a breach-service outage cannot block all password changes.

Source code in jafaal/ports.py

@runtime_checkable
class PasswordBreachChecker(Protocol):
    """Host-owned check for whether a password appears in a breach corpus/blocklist.

    Consulted during sign-up and password change, *after* the length/complexity
    policy passes and *before* the password is hashed. Return ``True`` to reject
    the password. This is the NIST SP 800-63B-recommended companion to the
    ``length_only`` policy — a common implementation is an HIBP k-anonymity range
    query or a local blocklist.

    It runs in the request path (synchronously), so keep it fast and fail open
    (return ``False``) on an upstream error, so a breach-service outage cannot
    block all password changes.
    """

    def is_breached(self, password: str) -> bool: ...

PasswordPolicy dataclass

Minimum-length policy resolved by user tier, plus the policy type.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class PasswordPolicy:
    """Minimum-length policy resolved by user tier, plus the policy type."""

    min_length_regular: int
    min_length_admin: int
    password_type: str

    def min_length_for(self, *, is_superuser: bool) -> int:
        """Return the minimum length for an admin/superuser or regular account."""
        return self.min_length_admin if is_superuser else self.min_length_regular

min_length_for

min_length_for(*, is_superuser)

Return the minimum length for an admin/superuser or regular account.

Source code in jafaal/ports.py

def min_length_for(self, *, is_superuser: bool) -> int:
    """Return the minimum length for an admin/superuser or regular account."""
    return self.min_length_admin if is_superuser else self.min_length_regular

PasswordResetRequested dataclass

A password reset was requested; deliver token to email.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class PasswordResetRequested:
    """A password reset was requested; deliver ``token`` to ``email``."""

    user_id: Any
    email: str
    display_name: str | None
    token: str
    expires_at: datetime
    locale: str | None

RefreshTokenTheftDetected dataclass

A rotated refresh token was replayed past the grace window (likely theft).

Emitted when reuse detection invalidates a token family, so the host can force a re-login notification / security alert for the affected user.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class RefreshTokenTheftDetected:
    """A rotated refresh token was replayed past the grace window (likely theft).

    Emitted when reuse detection invalidates a token family, so the host can
    force a re-login notification / security alert for the affected user.
    """

    user_id: Any
    token_family_id: str

ScopeResolver

Bases: Protocol

Host-owned mapping from a user to the scopes their tokens carry.

JAFAAL's default (:class:TieredScopeResolver) is deliberately simple: two tiers, keyed on is_superuser. That covers the common case and nothing else — it cannot express "this user is a billing admin", per-organisation roles, or any grant that is not a boolean on the user row. Authorisation models are application domain, not authentication plumbing, so the mapping is a port: implement it and JAFAAL stamps whatever scopes you return into the tokens it mints.

The resolver runs at token issuance (login, refresh, SSO/PKCE exchange), and again per request when :attr:~jafaal.settings.AuthSettings.reauthorize_scopes_per_request is set — where the result is intersected with the token's existing scopes, so re-resolution can only ever narrow a live token's authority, never widen it.

Keep it fast and side-effect free; it is on the login path. It is called with the user alone — a resolver that needs more (roles from another table, say) should read them through its own session or cache rather than expect one to be passed in, since JAFAAL calls it from several transaction contexts.

Source code in jafaal/ports.py

@runtime_checkable
class ScopeResolver(Protocol):
    """Host-owned mapping from a user to the scopes their tokens carry.

    JAFAAL's default (:class:`TieredScopeResolver`) is deliberately simple: two
    tiers, keyed on ``is_superuser``. That covers the common case and nothing
    else — it cannot express "this user is a billing admin", per-organisation
    roles, or any grant that is not a boolean on the user row. Authorisation
    models are application domain, not authentication plumbing, so the mapping is
    a port: implement it and JAFAAL stamps whatever scopes you return into the
    tokens it mints.

    The resolver runs at token issuance (login, refresh, SSO/PKCE exchange), and
    again per request when
    :attr:`~jafaal.settings.AuthSettings.reauthorize_scopes_per_request` is set —
    where the result is *intersected* with the token's existing scopes, so
    re-resolution can only ever narrow a live token's authority, never widen it.

    Keep it fast and side-effect free; it is on the login path. It is called with
    the user alone — a resolver that needs more (roles from another table, say)
    should read them through its own session or cache rather than expect one to
    be passed in, since JAFAAL calls it from several transaction contexts.
    """

    def scopes_for(self, user: UserProtocol) -> tuple[str, ...]:
        """Return the scopes to stamp into ``user``'s tokens."""
        ...

scopes_for

scopes_for(user)

Return the scopes to stamp into user's tokens.

Source code in jafaal/ports.py

def scopes_for(self, user: UserProtocol) -> tuple[str, ...]:
    """Return the scopes to stamp into ``user``'s tokens."""
    ...

SettingsProvider

Bases: Protocol

Host-owned dynamic settings (password policy + sign-up toggles).

Read-only configuration, independent of the caller's transaction, so these methods take no session — a DB-backed adapter manages its own read (JAFAAL's :func:jafaal.orm.session_scope is available) and a static adapter simply returns constants.

Source code in jafaal/ports.py

class SettingsProvider(Protocol):
    """Host-owned dynamic settings (password policy + sign-up toggles).

    Read-only configuration, independent of the caller's transaction, so these
    methods take no session — a DB-backed adapter manages its own read (JAFAAL's
    :func:`jafaal.orm.session_scope` is available) and a static adapter simply
    returns constants.
    """

    def get_password_policy(self) -> PasswordPolicy:
        """Return the active password policy."""
        ...

    def get_signup_config(self) -> SignupConfig:
        """Return the active sign-up configuration."""
        ...

get_password_policy

get_password_policy()

Return the active password policy.

Source code in jafaal/ports.py

def get_password_policy(self) -> PasswordPolicy:
    """Return the active password policy."""
    ...

get_signup_config

get_signup_config()

Return the active sign-up configuration.

Source code in jafaal/ports.py

def get_signup_config(self) -> SignupConfig:
    """Return the active sign-up configuration."""
    ...

SignupApproved dataclass

A pending sign-up was approved; notify the user.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class SignupApproved:
    """A pending sign-up was approved; notify the user."""

    user_id: Any
    email: str
    display_name: str | None
    locale: str | None

SignupConfig dataclass

Host sign-up toggles.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class SignupConfig:
    """Host sign-up toggles."""

    enabled: bool
    require_email_verification: bool
    require_admin_approval: bool

SignupPendingAdminApproval dataclass

A newly verified sign-up is awaiting admin approval.

JAFAAL emits one event with the new user's context; the host fans out to whichever admins it wants, in whatever locale.

Source code in jafaal/ports.py

@dataclass(frozen=True)
class SignupPendingAdminApproval:
    """A newly verified sign-up is awaiting admin approval.

    JAFAAL emits one event with the new user's context; the host fans out to
    whichever admins it wants, in whatever locale.
    """

    user_id: Any
    username: str
    display_name: str | None

TieredScopeResolver

Default resolver: the :class:~jafaal.scopes.ScopeCatalog's two tiers.

Returns the catalog's admin tuple for a user whose is_superuser attribute is truthy and regular otherwise. A model without that attribute gets regular, so the two-tier default is a convenience rather than a schema requirement — the host adds the column if it wants the split, or installs its own resolver if its authorisation model is richer than a boolean.

Source code in jafaal/ports.py

class TieredScopeResolver:
    """Default resolver: the :class:`~jafaal.scopes.ScopeCatalog`'s two tiers.

    Returns the catalog's ``admin`` tuple for a user whose ``is_superuser``
    attribute is truthy and ``regular`` otherwise. A model without that attribute
    gets ``regular``, so the two-tier default is a *convenience* rather than a
    schema requirement — the host adds the column if it wants the split, or
    installs its own resolver if its authorisation model is richer than a
    boolean.
    """

    def scopes_for(self, user: UserProtocol) -> tuple[str, ...]:
        """Return the catalog tier matching the user's superuser flag."""
        import jafaal.scopes as jafaal_scopes

        catalog = jafaal_scopes.get_scope_catalog()
        return catalog.admin if is_superuser(user) else catalog.regular

scopes_for

scopes_for(user)

Return the catalog tier matching the user's superuser flag.

Source code in jafaal/ports.py

def scopes_for(self, user: UserProtocol) -> tuple[str, ...]:
    """Return the catalog tier matching the user's superuser flag."""
    import jafaal.scopes as jafaal_scopes

    catalog = jafaal_scopes.get_scope_catalog()
    return catalog.admin if is_superuser(user) else catalog.regular

UserProtocol

Bases: Protocol

The user attributes JAFAAL reads across the boundary.

Satisfied by a host user model built on :class:jafaal.UserMixin: id, username, email, is_active, is_verified, and the mfa_enabled property. JAFAAL never reads app-specific profile fields.

is_superuser is deliberately not here. It is an authorisation concept, and authorisation is the host's domain: it is read only by the default :class:TieredScopeResolver, which is one swappable implementation of the :class:ScopeResolver port. A host whose model has no such flag — because it uses roles, organisations, or per-tenant grants — supplies its own resolver and never needs the column. Requiring it on the protocol would have made a two-tier authorisation model a condition of using the library.

Source code in jafaal/ports.py

@runtime_checkable
class UserProtocol(Protocol):
    """The user attributes JAFAAL reads across the boundary.

    Satisfied by a host user model built on :class:`jafaal.UserMixin`: ``id``,
    ``username``, ``email``, ``is_active``, ``is_verified``, and the
    ``mfa_enabled`` property. JAFAAL never reads app-specific profile fields.

    ``is_superuser`` is deliberately **not** here. It is an authorisation
    concept, and authorisation is the host's domain: it is read only by the
    default :class:`TieredScopeResolver`, which is one swappable implementation
    of the :class:`ScopeResolver` port. A host whose model has no such flag —
    because it uses roles, organisations, or per-tenant grants — supplies its own
    resolver and never needs the column. Requiring it on the protocol would have
    made a two-tier authorisation model a condition of using the library.
    """

    id: Any
    username: str
    email: str
    is_active: bool
    is_verified: bool

    @property
    def mfa_enabled(self) -> bool: ...

UserRepository

Bases: Protocol

Host-owned persistence for the user table.

Methods run inside the caller's transaction and therefore take the active :class:~sqlalchemy.orm.Session. Implementations return objects satisfying :class:UserProtocol (typically the host's Users ORM instance).

Source code in jafaal/ports.py

class UserRepository(Protocol):
    """Host-owned persistence for the user table.

    Methods run inside the caller's transaction and therefore take the active
    :class:`~sqlalchemy.orm.Session`. Implementations return objects satisfying
    :class:`UserProtocol` (typically the host's ``Users`` ORM instance).
    """

    def get_by_id(self, user_id: Any, db: Session) -> UserProtocol | None:
        """Return the user with ``user_id``, or ``None``."""
        ...

    def get_by_email(self, email: str, db: Session) -> UserProtocol | None:
        """Return the user with ``email``, or ``None``."""
        ...

    def get_by_username(self, username: str, db: Session) -> UserProtocol | None:
        """Return the user with ``username``, or ``None``."""
        ...

    def create_local_user(
        self,
        username: str,
        email: str,
        db: Session,
        *,
        is_active: bool,
        is_verified: bool,
    ) -> UserProtocol:
        """Create a user row for a local sign-up and return it.

        JAFAAL validates the password and persists the credential separately in
        its own ``users_local_credentials`` table; the host only creates the
        user/profile row here (with the given active/verified state and any
        host-specific defaults). ``username``/``email`` are passed as supplied;
        the host applies its own normalization and uniqueness checks.

        **Flush, do not commit.** The primary key must be populated for the
        credential write that follows, but the two rows have to land in one
        transaction: committing here leaves a credential-less account squatting
        the username and email if anything downstream fails. JAFAAL commits both
        together when it writes the credential.
        """
        ...

    def provision_from_idp(self, identity: IdpIdentity, db: Session) -> UserProtocol:
        """Create a user row from an IdP identity and return it.

        The account has no local password (JAFAAL persists no credential).
        Profile shape and defaults are the host's concern.
        """
        ...

    def sync_from_idp(self, user_id: Any, claims: Mapping[str, Any], db: Session) -> None:
        """Optionally sync host-owned profile fields from refreshed IdP claims.

        ``claims`` is the mapped IdP claim dict (e.g. ``email``, ``name``), plus
        ``email_verified`` so the host can apply its own policy. Called on
        subsequent logins when IdP→user sync is enabled; the host decides which
        fields to update and resolves any email conflicts.

        ``email`` is present **only when the provider asserted it verified** —
        JAFAAL withholds an unverified address rather than hand the host
        something it might write onto the user row, since the local email is
        where password resets are delivered.
        """
        ...

    def set_email_verified(self, user_id: Any, db: Session, *, activate: bool) -> None:
        """Mark the user's email address as verified.

        When ``activate`` is ``True`` the account is also activated (used when
        email verification is the last gate before login); when ``False`` the
        account stays inactive (e.g. admin approval is still pending).
        """
        ...

get_by_id

get_by_id(user_id, db)

Return the user with user_id, or None.

Source code in jafaal/ports.py

def get_by_id(self, user_id: Any, db: Session) -> UserProtocol | None:
    """Return the user with ``user_id``, or ``None``."""
    ...

get_by_email

get_by_email(email, db)

Return the user with email, or None.

Source code in jafaal/ports.py

def get_by_email(self, email: str, db: Session) -> UserProtocol | None:
    """Return the user with ``email``, or ``None``."""
    ...

get_by_username

get_by_username(username, db)

Return the user with username, or None.

Source code in jafaal/ports.py

def get_by_username(self, username: str, db: Session) -> UserProtocol | None:
    """Return the user with ``username``, or ``None``."""
    ...

create_local_user

create_local_user(
    username, email, db, *, is_active, is_verified
)

Create a user row for a local sign-up and return it.

JAFAAL validates the password and persists the credential separately in its own users_local_credentials table; the host only creates the user/profile row here (with the given active/verified state and any host-specific defaults). username/email are passed as supplied; the host applies its own normalization and uniqueness checks.

Flush, do not commit. The primary key must be populated for the credential write that follows, but the two rows have to land in one transaction: committing here leaves a credential-less account squatting the username and email if anything downstream fails. JAFAAL commits both together when it writes the credential.

Source code in jafaal/ports.py

def create_local_user(
    self,
    username: str,
    email: str,
    db: Session,
    *,
    is_active: bool,
    is_verified: bool,
) -> UserProtocol:
    """Create a user row for a local sign-up and return it.

    JAFAAL validates the password and persists the credential separately in
    its own ``users_local_credentials`` table; the host only creates the
    user/profile row here (with the given active/verified state and any
    host-specific defaults). ``username``/``email`` are passed as supplied;
    the host applies its own normalization and uniqueness checks.

    **Flush, do not commit.** The primary key must be populated for the
    credential write that follows, but the two rows have to land in one
    transaction: committing here leaves a credential-less account squatting
    the username and email if anything downstream fails. JAFAAL commits both
    together when it writes the credential.
    """
    ...

provision_from_idp

provision_from_idp(identity, db)

Create a user row from an IdP identity and return it.

The account has no local password (JAFAAL persists no credential). Profile shape and defaults are the host's concern.

Source code in jafaal/ports.py

def provision_from_idp(self, identity: IdpIdentity, db: Session) -> UserProtocol:
    """Create a user row from an IdP identity and return it.

    The account has no local password (JAFAAL persists no credential).
    Profile shape and defaults are the host's concern.
    """
    ...

sync_from_idp

sync_from_idp(user_id, claims, db)

Optionally sync host-owned profile fields from refreshed IdP claims.

claims is the mapped IdP claim dict (e.g. email, name), plus email_verified so the host can apply its own policy. Called on subsequent logins when IdP→user sync is enabled; the host decides which fields to update and resolves any email conflicts.

email is present only when the provider asserted it verified — JAFAAL withholds an unverified address rather than hand the host something it might write onto the user row, since the local email is where password resets are delivered.

Source code in jafaal/ports.py

def sync_from_idp(self, user_id: Any, claims: Mapping[str, Any], db: Session) -> None:
    """Optionally sync host-owned profile fields from refreshed IdP claims.

    ``claims`` is the mapped IdP claim dict (e.g. ``email``, ``name``), plus
    ``email_verified`` so the host can apply its own policy. Called on
    subsequent logins when IdP→user sync is enabled; the host decides which
    fields to update and resolves any email conflicts.

    ``email`` is present **only when the provider asserted it verified** —
    JAFAAL withholds an unverified address rather than hand the host
    something it might write onto the user row, since the local email is
    where password resets are delivered.
    """
    ...

set_email_verified

set_email_verified(user_id, db, *, activate)

Mark the user's email address as verified.

When activate is True the account is also activated (used when email verification is the last gate before login); when False the account stays inactive (e.g. admin approval is still pending).

Source code in jafaal/ports.py

def set_email_verified(self, user_id: Any, db: Session, *, activate: bool) -> None:
    """Mark the user's email address as verified.

    When ``activate`` is ``True`` the account is also activated (used when
    email verification is the last gate before login); when ``False`` the
    account stays inactive (e.g. admin approval is still pending).
    """
    ...

NoOpRateLimiter

Default limiter that enforces nothing (returns the endpoint unchanged).

Source code in jafaal/rate_limit.py

class NoOpRateLimiter:
    """Default limiter that enforces nothing (returns the endpoint unchanged)."""

    def limit(self, category: str) -> Callable[[F], F]:
        def decorator(func: F) -> F:
            return func

        return decorator

RateLimiter

Bases: Protocol

Maps a JAFAAL rate-limit category to an endpoint decorator.

The host implementation resolves the category ("sensitive" / "write") to a concrete budget and returns the decorator its limiter uses (e.g. slowapi's Limiter.limit("10/minute")).

Source code in jafaal/rate_limit.py

@runtime_checkable
class RateLimiter(Protocol):
    """Maps a JAFAAL rate-limit category to an endpoint decorator.

    The host implementation resolves the category (``"sensitive"`` / ``"write"``)
    to a concrete budget and returns the decorator its limiter uses (e.g.
    ``slowapi``'s ``Limiter.limit("10/minute")``).
    """

    def limit(self, category: str) -> Callable[[F], F]: ...

AuthorizationRedirectResponse

Bases: BaseModel

Where to send the browser once a local authorization request completes.

Returned by /auth/login (and the second-factor endpoints) when the request carried an auth_request handle from /auth/authorize. The login then produces an RFC 6749 §4.1.2 authorization response rather than a token response: the URL carries a single-use code, the client's state, and iss — and no token, which is the point of the code flow.

The host's login page navigates to it; JAFAAL does not redirect on its own, because the page is script and needs to clear its own state first.

Attributes:

Name	Type	Description
redirect_to	StrictStr	The client's registered redirect_uri with the authorization response parameters appended.

Source code in jafaal/schema.py

class AuthorizationRedirectResponse(BaseModel):
    """Where to send the browser once a local authorization request completes.

    Returned by ``/auth/login`` (and the second-factor endpoints) when the
    request carried an ``auth_request`` handle from ``/auth/authorize``. The
    login then produces an RFC 6749 §4.1.2 *authorization response* rather than
    a token response: the URL carries a single-use ``code``, the client's
    ``state``, and ``iss`` — and no token, which is the point of the code flow.

    The host's login page navigates to it; JAFAAL does not redirect on its own,
    because the page is script and needs to clear its own state first.

    Attributes:
        redirect_to: The client's registered ``redirect_uri`` with the
            authorization response parameters appended.
    """

    model_config = ConfigDict(extra="forbid")

    redirect_to: StrictStr

LogoutResponse

Bases: BaseModel

Response payload returned by the logout endpoint.

Attributes:

Name	Type	Description
message	StrictStr	Human-readable confirmation message.

Source code in jafaal/schema.py

class LogoutResponse(BaseModel):
    """
    Response payload returned by the logout endpoint.

    Attributes:
        message: Human-readable confirmation message.
    """

    model_config = ConfigDict(extra="forbid")

    message: StrictStr

MFALoginRequest

Bases: BaseModel

Schema for MFA login requests.

Attributes:

Name	Type	Description
mfa_token	StrictStr	The opaque, single-use ticket returned by /auth/login in :class:MFARequiredResponse. It proves this caller satisfied the password factor. The username is deliberately not accepted here: it is public or guessable, so addressing the pending login by username would let anyone holding a valid one-time code finish a login that somebody else's password step opened.
mfa_code	StrictStr	Either a 6-digit TOTP code or a backup code.

Source code in jafaal/schema.py

class MFALoginRequest(BaseModel):
    """
    Schema for MFA login requests.

    Attributes:
        mfa_token: The opaque, single-use ticket returned by ``/auth/login``
            in :class:`MFARequiredResponse`. It proves *this caller* satisfied
            the password factor. The username is deliberately **not** accepted
            here: it is public or guessable, so addressing the pending login by
            username would let anyone holding a valid one-time code finish a
            login that somebody else's password step opened.
        mfa_code: Either a 6-digit TOTP code or a backup code.
    """

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    mfa_token: StrictStr = Field(..., min_length=1, max_length=512)
    mfa_code: StrictStr = Field(
        ...,
        pattern=r"^(\d{6}|[A-Z0-9]{4}-[A-Z0-9]{4})$",
    )
    client_id: StrictStr = Field(
        ...,
        max_length=256,
        description="The registered client the login was started for; decides token delivery and scope.",
    )

MFARequiredResponse

Bases: BaseModel

Response indicating MFA verification is required.

Attributes:

Name	Type	Description
mfa_required	StrictBool	Indicates whether MFA is required.
mfa_token	StrictStr	Opaque, single-use ticket proving the password factor was satisfied by this caller. Hold it in memory (never persist it) and present it to /auth/mfa/verify or the WebAuthn second-factor endpoints. It expires in five minutes.
username	StrictStr	Username for which MFA is required, echoed back for display. It is not a credential and does not address the pending login.
message	StrictStr	Message describing the requirement.

Source code in jafaal/schema.py

class MFARequiredResponse(BaseModel):
    """
    Response indicating MFA verification is required.

    Attributes:
        mfa_required: Indicates whether MFA is required.
        mfa_token: Opaque, single-use ticket proving the password factor was
            satisfied by this caller. Hold it in memory (never persist it) and
            present it to ``/auth/mfa/verify`` or the WebAuthn second-factor
            endpoints. It expires in five minutes.
        username: Username for which MFA is required, echoed back for display.
            It is *not* a credential and does not address the pending login.
        message: Message describing the requirement.
    """

    model_config = ConfigDict(extra="forbid")

    mfa_required: StrictBool = True
    mfa_token: StrictStr
    username: StrictStr
    message: StrictStr = "MFA verification required"

PasswordChangeRequest

Bases: StepUpVerification

Self-service password change, on top of the step-up factors.

Attributes:

Name	Type	Description
new_password	StrictStr	The replacement password, held to the account tier's policy and screened against the installed breach corpus.
revoke_other_sessions	bool	Whether to end the caller's other sessions.

Source code in jafaal/schema.py

class PasswordChangeRequest(StepUpVerification):
    """Self-service password change, on top of the step-up factors.

    Attributes:
        new_password: The replacement password, held to the account tier's
            policy and screened against the installed breach corpus.
        revoke_other_sessions: Whether to end the caller's other sessions.
    """

    new_password: StrictStr = Field(
        ...,
        min_length=1,
        max_length=PASSWORD_FIELD_MAX_LENGTH,
        description="The new password.",
    )
    revoke_other_sessions: bool = Field(
        default=True,
        description=(
            "End every other session for this account (default). 'Change my password' is what a user does "
            "when they think they are compromised, so leaving the attacker's session live is the one "
            "outcome that makes it pointless. Set false for a routine rotation."
        ),
    )

PasswordChangeResponse

Bases: BaseModel

Result of a successful password change.

Source code in jafaal/schema.py

class PasswordChangeResponse(BaseModel):
    """Result of a successful password change."""

    model_config = ConfigDict(extra="forbid")

    message: StrictStr = Field(default="Password changed", description="Human-readable confirmation.")
    revoked_sessions: int = Field(
        default=0,
        description="How many other sessions were ended. The caller's own session is preserved.",
    )

SignUpRequest

Bases: BaseModel

Minimal local sign-up request.

JAFAAL only needs credentials to create the account and its password; any additional profile fields a host collects at sign-up are the host's own concern (its UserRepository fills them). username/email are passed to the host repository as supplied.

Source code in jafaal/schema.py

class SignUpRequest(BaseModel):
    """Minimal local sign-up request.

    JAFAAL only needs credentials to create the account and its password; any
    additional profile fields a host collects at sign-up are the host's own
    concern (its ``UserRepository`` fills them). ``username``/``email`` are
    passed to the host repository as supplied.
    """

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    username: StrictStr = Field(..., min_length=1, max_length=250)
    email: StrictStr = Field(..., min_length=3, max_length=250)
    # Bounded by the shared transport limit; the policy minimum/maximum comes
    # from PasswordSettings, applied by validate_and_hash_for_user.
    password: StrictStr = Field(..., min_length=1, max_length=PASSWORD_FIELD_MAX_LENGTH)

StepUpVerification

Bases: BaseModel

Generic step-up verification payload.

Used by sensitive account-level operations (API-key creation, MFA backup-code regeneration, IdP unlink, ...) to require fresh proof of identity beyond a valid access token. Accounts with a local password must supply current_password; when MFA is enabled an mfa_code is also required. An SSO-only account with no MFA has no factor to verify, so these operations are refused until MFA is enrolled — step-up fails closed rather than passing on a valid access token alone.

Attributes:

Name	Type	Description
current_password	StrictStr | None	Caller's existing password. Required when the account has a local password; may be omitted for SSO-only accounts (which must then satisfy step-up via MFA).
mfa_code	StrictStr | None	TOTP or backup code, required when MFA is enabled.

Source code in jafaal/schema.py

class StepUpVerification(BaseModel):
    """Generic step-up verification payload.

    Used by sensitive account-level operations (API-key creation, MFA
    backup-code regeneration, IdP unlink, ...) to require fresh proof of
    identity beyond a valid access token. Accounts with a local password must
    supply ``current_password``; when MFA is enabled an ``mfa_code`` is also
    required. An SSO-only account with no MFA has no factor to verify, so these
    operations are refused until MFA is enrolled — step-up fails closed rather
    than passing on a valid access token alone.

    Attributes:
        current_password: Caller's existing password. Required when the account
            has a local password; may be omitted for SSO-only accounts (which
            must then satisfy step-up via MFA).
        mfa_code: TOTP or backup code, required when MFA is enabled.
    """

    model_config = ConfigDict(extra="forbid", validate_assignment=True)

    current_password: StrictStr | None = Field(
        default=None,
        min_length=1,
        max_length=PASSWORD_FIELD_MAX_LENGTH,
        description="Current password (step-up verification). Required when the account has a local password.",
    )
    mfa_code: StrictStr | None = Field(
        default=None,
        max_length=32,
        description="TOTP or backup code, required when MFA is enabled",
    )

TokenIntrospectionResponse

Bases: BaseModel

RFC 7662 token introspection response.

active is the only guaranteed field; the rest are populated only for an active token. token_use and sid are JAFAAL extensions (§2.2 permits them), reporting the token's own use and its session id.

The extension is named token_use rather than typ for the same reason the payload claim is: §2.2 already defines token_type as the RFC 6749 §7.1 type (Bearer), and RFC 9068 uses typ for the JOSE header's media type. A third spelling of "type" meaning a third thing is how a client reads the wrong one.

Attributes:

Name	Type	Description
active	StrictBool	Whether the token is currently valid.
sub	StrictStr | None	Subject (user) identifier.
scope	StrictStr | None	Space-delimited granted scopes.
token_use	StrictStr | None	JAFAAL token use (access or refresh); mirrors the token's own token_use claim.
token_type	StrictStr | None	Bearer for an active token.
client_id	StrictStr | None	OAuth client identifier the token was issued to.
exp	StrictInt | None	Expiry (epoch seconds).
iat	StrictInt | None	Issued-at (epoch seconds).
nbf	StrictInt | None	Not-before (epoch seconds).
iss	StrictStr | None	Issuer.
aud	StrictStr | None	Audience.
jti	StrictStr | None	Token identifier.
sid	StrictStr | None	Session identifier (JAFAAL extension).

Source code in jafaal/schema.py

class TokenIntrospectionResponse(BaseModel):
    """RFC 7662 token introspection response.

    ``active`` is the only guaranteed field; the rest are populated only for an
    active token. ``token_use`` and ``sid`` are JAFAAL extensions (§2.2 permits
    them), reporting the token's own use and its session id.

    The extension is named ``token_use`` rather than ``typ`` for the same reason
    the payload claim is: §2.2 already defines ``token_type`` as the RFC 6749
    §7.1 type (``Bearer``), and RFC 9068 uses ``typ`` for the JOSE *header*'s
    media type. A third spelling of "type" meaning a third thing is how a client
    reads the wrong one.

    Attributes:
        active: Whether the token is currently valid.
        sub: Subject (user) identifier.
        scope: Space-delimited granted scopes.
        token_use: JAFAAL token use (``access`` or ``refresh``); mirrors the
            token's own ``token_use`` claim.
        token_type: ``Bearer`` for an active token.
        client_id: OAuth client identifier the token was issued to.
        exp: Expiry (epoch seconds).
        iat: Issued-at (epoch seconds).
        nbf: Not-before (epoch seconds).
        iss: Issuer.
        aud: Audience.
        jti: Token identifier.
        sid: Session identifier (JAFAAL extension).
    """

    model_config = ConfigDict(extra="forbid")

    active: StrictBool
    sub: StrictStr | None = None
    scope: StrictStr | None = None
    token_use: StrictStr | None = None
    token_type: StrictStr | None = None
    client_id: StrictStr | None = None
    exp: StrictInt | None = None
    iat: StrictInt | None = None
    nbf: StrictInt | None = None
    iss: StrictStr | None = None
    aud: StrictStr | None = None
    jti: StrictStr | None = None
    sid: StrictStr | None = None

TokenResponseMobile

Bases: BaseModel

The RFC 6749 §5.1 token response, for token_delivery="body".

Attributes:

Name	Type	Description
session_id	StrictStr	Session identifier (JAFAAL extension).
access_token	StrictStr	Bearer access token.
refresh_token	StrictStr	Refresh token.
token_type	Literal['Bearer']	Always Bearer (RFC 6750 §4).
expires_in	StrictInt	Seconds until the access token expires.
refresh_token_expires_in	StrictInt	Seconds until the refresh token expires (JAFAAL extension).
scope	StrictStr | None	Space-delimited scopes the access token actually carries.

Source code in jafaal/schema.py

class TokenResponseMobile(BaseModel):
    """The RFC 6749 §5.1 token response, for ``token_delivery="body"``.

    Attributes:
        session_id: Session identifier (JAFAAL extension).
        access_token: Bearer access token.
        refresh_token: Refresh token.
        token_type: Always ``Bearer`` (RFC 6750 §4).
        expires_in: Seconds until the access token expires.
        refresh_token_expires_in: Seconds until the refresh token expires
            (JAFAAL extension).
        scope: Space-delimited scopes the access token actually carries.
    """

    model_config = ConfigDict(extra="forbid")

    session_id: StrictStr
    access_token: StrictStr
    refresh_token: StrictStr
    token_type: Literal["Bearer"] = "Bearer"
    expires_in: StrictInt
    refresh_token_expires_in: StrictInt
    scope: StrictStr | None = None

TokenResponseWeb

Bases: BaseModel

Token response for a client registered with token_delivery="cookie".

The RFC 6749 §5.1 response minus refresh_token, which is delivered as an HttpOnly, SameSite=Strict cookie instead (RFC 9700 §7.2: do not hand a refresh token to page script). session_id, csrf_token and refresh_token_expires_in are JAFAAL extensions, which §5.1 permits.

Attributes:

Name	Type	Description
session_id	StrictStr	Session identifier.
access_token	StrictStr	Bearer access token.
csrf_token	StrictStr	CSRF token bound to the session.
token_type	Literal['Bearer']	Always Bearer (RFC 6750 §4).
expires_in	StrictInt	Seconds until the access token expires.
refresh_token_expires_in	StrictInt	Seconds until the refresh token expires.
scope	StrictStr | None	Space-delimited scopes the access token actually carries.

Source code in jafaal/schema.py

class TokenResponseWeb(BaseModel):
    """Token response for a client registered with ``token_delivery="cookie"``.

    The RFC 6749 §5.1 response minus ``refresh_token``, which is delivered as an
    ``HttpOnly``, ``SameSite=Strict`` cookie instead (RFC 9700 §7.2: do not hand
    a refresh token to page script). ``session_id``, ``csrf_token`` and
    ``refresh_token_expires_in`` are JAFAAL extensions, which §5.1 permits.

    Attributes:
        session_id: Session identifier.
        access_token: Bearer access token.
        csrf_token: CSRF token bound to the session.
        token_type: Always ``Bearer`` (RFC 6750 §4).
        expires_in: Seconds until the access token expires.
        refresh_token_expires_in: Seconds until the refresh token expires.
        scope: Space-delimited scopes the access token actually carries.
    """

    model_config = ConfigDict(extra="forbid")

    session_id: StrictStr
    access_token: StrictStr
    csrf_token: StrictStr
    token_type: Literal["Bearer"] = "Bearer"
    expires_in: StrictInt
    refresh_token_expires_in: StrictInt
    scope: StrictStr | None = None

ScopeCatalog dataclass

Scopes a token carries per access tier, plus Swagger descriptions.

Attributes:

Name	Type	Description
regular	tuple[str, ...]	Scopes stamped into a non-superuser's token.
admin	tuple[str, ...]	Scopes stamped into a superuser's token (a superset of regular).
descriptions	Mapping[str, str]	Scope -> human description, shown in the Swagger Authorize dialog. Every minted scope must be described.

Source code in jafaal/scopes.py

@dataclass(frozen=True)
class ScopeCatalog:
    """Scopes a token carries per access tier, plus Swagger descriptions.

    Attributes:
        regular: Scopes stamped into a non-superuser's token.
        admin: Scopes stamped into a superuser's token (a superset of
            ``regular``).
        descriptions: Scope -> human description, shown in the Swagger
            ``Authorize`` dialog. Every minted scope must be described.
    """

    regular: tuple[str, ...]
    admin: tuple[str, ...]
    descriptions: Mapping[str, str]

    def validate(self) -> None:
        """Assert the catalog is internally consistent.

        Every scope advertised in ``descriptions`` must be one the server
        actually mints, and every minted scope must be advertised — a drift
        means either the UI offers a scope that is never enforced or a token
        carries an undocumented scope. ``regular`` must also be a subset of
        ``admin``.

        Raises:
            ValueError: If the catalog is inconsistent.
        """
        minted = frozenset(self.admin)
        described = frozenset(self.descriptions)
        undescribed = minted - described
        unminted = described - minted
        if undescribed or unminted:
            raise ValueError(
                "ScopeCatalog descriptions are out of sync with the scope tuples: "
                f"minted-but-undeclared={sorted(undescribed)}, "
                f"declared-but-never-minted={sorted(unminted)}"
            )
        extra_regular = frozenset(self.regular) - minted
        if extra_regular:
            raise ValueError(f"ScopeCatalog regular scopes are not a subset of admin scopes: {sorted(extra_regular)}")

    def extend(
        self,
        *,
        regular: tuple[str, ...] = (),
        admin: tuple[str, ...] = (),
        descriptions: Mapping[str, str] | None = None,
    ) -> ScopeCatalog:
        """Return a new catalog with the host's application scopes added on top.

        Args:
            regular: Extra scopes for the regular (and, implicitly, admin) tier.
            admin: Extra scopes for the admin tier (include the ``regular`` ones
                too, plus any admin-only scopes).
            descriptions: Descriptions for the added scopes.

        Returns:
            A new :class:`ScopeCatalog` combining JAFAAL's scopes with the host's.
        """
        return ScopeCatalog(
            regular=self.regular + tuple(regular),
            admin=self.admin + tuple(admin),
            descriptions={**self.descriptions, **(descriptions or {})},
        )

validate

validate()

Assert the catalog is internally consistent.

Every scope advertised in descriptions must be one the server actually mints, and every minted scope must be advertised — a drift means either the UI offers a scope that is never enforced or a token carries an undocumented scope. regular must also be a subset of admin.

Raises:

Type	Description
ValueError	If the catalog is inconsistent.

Source code in jafaal/scopes.py

def validate(self) -> None:
    """Assert the catalog is internally consistent.

    Every scope advertised in ``descriptions`` must be one the server
    actually mints, and every minted scope must be advertised — a drift
    means either the UI offers a scope that is never enforced or a token
    carries an undocumented scope. ``regular`` must also be a subset of
    ``admin``.

    Raises:
        ValueError: If the catalog is inconsistent.
    """
    minted = frozenset(self.admin)
    described = frozenset(self.descriptions)
    undescribed = minted - described
    unminted = described - minted
    if undescribed or unminted:
        raise ValueError(
            "ScopeCatalog descriptions are out of sync with the scope tuples: "
            f"minted-but-undeclared={sorted(undescribed)}, "
            f"declared-but-never-minted={sorted(unminted)}"
        )
    extra_regular = frozenset(self.regular) - minted
    if extra_regular:
        raise ValueError(f"ScopeCatalog regular scopes are not a subset of admin scopes: {sorted(extra_regular)}")

extend

extend(*, regular=(), admin=(), descriptions=None)

Return a new catalog with the host's application scopes added on top.

Parameters:

Name	Type	Description	Default
regular	tuple[str, ...]	Extra scopes for the regular (and, implicitly, admin) tier.	()
admin	tuple[str, ...]	Extra scopes for the admin tier (include the regular ones too, plus any admin-only scopes).	()
descriptions	Mapping[str, str] | None	Descriptions for the added scopes.	None

Returns:

Type	Description
ScopeCatalog	A new :class:ScopeCatalog combining JAFAAL's scopes with the host's.

Source code in jafaal/scopes.py

def extend(
    self,
    *,
    regular: tuple[str, ...] = (),
    admin: tuple[str, ...] = (),
    descriptions: Mapping[str, str] | None = None,
) -> ScopeCatalog:
    """Return a new catalog with the host's application scopes added on top.

    Args:
        regular: Extra scopes for the regular (and, implicitly, admin) tier.
        admin: Extra scopes for the admin tier (include the ``regular`` ones
            too, plus any admin-only scopes).
        descriptions: Descriptions for the added scopes.

    Returns:
        A new :class:`ScopeCatalog` combining JAFAAL's scopes with the host's.
    """
    return ScopeCatalog(
        regular=self.regular + tuple(regular),
        admin=self.admin + tuple(admin),
        descriptions={**self.descriptions, **(descriptions or {})},
    )

ApiKeySettings dataclass

API-key format and transport policy.

Attributes:

Name	Type	Description
prefix	str	Prefix for generated API keys (<prefix>_<token>).
allow_query_param	bool	Whether API keys may be supplied via the ?api_key= query string. Off by default: credentials in query strings appear in access logs, proxy histories, and browser history. Enable only for integrations that genuinely cannot set a header.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class ApiKeySettings:
    """API-key format and transport policy.

    Attributes:
        prefix: Prefix for generated API keys (``<prefix>_<token>``).
        allow_query_param: Whether API keys may be supplied via the
            ``?api_key=`` query string. Off by default: credentials in query
            strings appear in access logs, proxy histories, and browser history.
            Enable only for integrations that genuinely cannot set a header.
    """

    prefix: str = "jafaal"
    allow_query_param: bool = False

AuditSettings dataclass

Privacy policy for the jafaal.audit stream.

Attributes:

Name	Type	Description
include_pii	bool	When True (default), audit records carry direct identifiers (plaintext username, client IP, email) — the signal a SIEM needs to spot targeted brute-force. Set False to drop them (substituting a one-way username hash) for PII-minimal retention.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class AuditSettings:
    """Privacy policy for the ``jafaal.audit`` stream.

    Attributes:
        include_pii: When ``True`` (default), audit records carry direct
            identifiers (plaintext username, client IP, email) — the signal a
            SIEM needs to spot targeted brute-force. Set ``False`` to drop them
            (substituting a one-way username hash) for PII-minimal retention.
    """

    include_pii: bool = True

AuthSettings dataclass

Immutable, host-supplied configuration for the auth library.

Only values describing the deployment as a whole live here; everything else is grouped by concern (see the module docstring).

Attributes:

Name	Type	Description
secrets	Secrets	Key material and its rotation fallbacks. The one required group.
base_url	str	Public base URL of the host app. Used to build SSO redirect and error URLs, and as the default JWT issuer/audience, WebAuthn RP ID and origin, and CSRF trusted origin.
app_name	str	Human-readable application name; used as the MFA TOTP issuer shown in authenticator apps and as the default WebAuthn RP name.
environment	str	Deployment environment. Must be one of :data:KNOWN_ENVIRONMENTS; the :data:DEPLOYED_ENVIRONMENTS members are treated as deployed, which drives the cookie Secure flag, the cookie-name prefix, and the two fail-closed startup guards. An unrecognised value is rejected at construction rather than silently treated as local. Defaults to the safest value, so forgetting to set it cannot weaken a deployment.
store_key_prefix	str	Namespace prefix for state-store keys (lockout counters, MFA setup secrets, WebAuthn challenges, ...).
login_token_url	str	URL FastAPI's Swagger Authorize dialog posts the username/password form to. Cosmetic — it configures the OAuth2PasswordBearer scheme's tokenUrl and nothing else; JAFAAL does not implement the OAuth password grant.
login_ui_url	str	Absolute URL of the host's login page, used by /auth/authorize when no idp is named. JAFAAL is an authorization server with no user interface of its own: it redirects the browser here with an auth_request parameter, the host's page collects the credentials and posts them to /auth/login with that parameter, and JAFAAL answers with the redirect back to the client. Empty (the default) means local login is not offered at the authorization endpoint, and idp is then required.
login_ip_lockout_enabled	bool	When True (default), a per-source-IP backoff bounds how many accounts one IP can lock out by spraying failed logins across usernames (the per-account lockout is otherwise a cheap targeted-DoS lever). Relies on an accurate client IP (configure network.trusted_proxies behind a reverse proxy); disable if a shared egress IP causes false lockouts.
allow_in_memory_state_store_when_deployed	bool	Permit the process-local in-memory :class:~jafaal.state_store.StateStore in a deployed environment. Off by default: startup refuses, because a multi-worker/replica deployment would fragment progressive-lockout and TOTP-replay state per worker. Set True only for a genuine single-worker deployment.
allow_no_rate_limit_when_deployed	bool	Permit a deployed environment to run with the no-op rate limiter. Off by default; mirror of the above.
tokens	TokenSettings	JWT issuance and revocation policy.
sessions	SessionSettings	Session lifetime and refresh-cookie delivery.
passwords	PasswordSettings	Argon2 cost and length bounds.
mfa	MfaSettings	TOTP replay policy.
webauthn	WebAuthnSettings	Passkey Relying-Party identity and ceremony policy.
sso	SsoSettings	Identity-provider flows and step-up.
network	NetworkSettings	Proxy trust and SSRF policy.
rate_limits	RateLimitSettings	Canonical request budgets.
api_keys	ApiKeySettings	API-key format and transport policy.
audit	AuditSettings	Audit-stream privacy policy.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class AuthSettings:
    """Immutable, host-supplied configuration for the auth library.

    Only values describing the deployment as a whole live here; everything else
    is grouped by concern (see the module docstring).

    Attributes:
        secrets: Key material and its rotation fallbacks. The one required
            group.
        base_url: Public base URL of the host app. Used to build SSO redirect
            and error URLs, and as the default JWT issuer/audience, WebAuthn RP
            ID and origin, and CSRF trusted origin.
        app_name: Human-readable application name; used as the MFA TOTP issuer
            shown in authenticator apps and as the default WebAuthn RP name.
        environment: Deployment environment. Must be one of
            :data:`KNOWN_ENVIRONMENTS`; the :data:`DEPLOYED_ENVIRONMENTS`
            members are treated as deployed, which drives the cookie ``Secure``
            flag, the cookie-name prefix, and the two fail-closed startup
            guards. An unrecognised value is rejected at construction rather
            than silently treated as local. Defaults to the safest value, so
            forgetting to set it cannot weaken a deployment.
        store_key_prefix: Namespace prefix for state-store keys (lockout
            counters, MFA setup secrets, WebAuthn challenges, ...).
        login_token_url: URL FastAPI's Swagger *Authorize* dialog posts the
            username/password form to. Cosmetic — it configures the
            ``OAuth2PasswordBearer`` scheme's ``tokenUrl`` and nothing else;
            JAFAAL does not implement the OAuth password grant.
        login_ui_url: Absolute URL of the **host's** login page, used by
            ``/auth/authorize`` when no ``idp`` is named. JAFAAL is an
            authorization server with no user interface of its own: it redirects
            the browser here with an ``auth_request`` parameter, the host's page
            collects the credentials and posts them to ``/auth/login`` with that
            parameter, and JAFAAL answers with the redirect back to the client.
            Empty (the default) means local login is not offered at the
            authorization endpoint, and ``idp`` is then required.
        login_ip_lockout_enabled: When ``True`` (default), a per-source-IP
            backoff bounds how many accounts one IP can lock out by spraying
            failed logins across usernames (the per-account lockout is
            otherwise a cheap targeted-DoS lever). Relies on an accurate client
            IP (configure ``network.trusted_proxies`` behind a reverse proxy);
            disable if a shared egress IP causes false lockouts.
        allow_in_memory_state_store_when_deployed: Permit the process-local
            in-memory :class:`~jafaal.state_store.StateStore` in a *deployed*
            environment. Off by default: startup refuses, because a
            multi-worker/replica deployment would fragment progressive-lockout
            and TOTP-replay state per worker. Set ``True`` only for a genuine
            single-worker deployment.
        allow_no_rate_limit_when_deployed: Permit a deployed environment to run
            with the no-op rate limiter. Off by default; mirror of the above.
        tokens: JWT issuance and revocation policy.
        sessions: Session lifetime and refresh-cookie delivery.
        passwords: Argon2 cost and length bounds.
        mfa: TOTP replay policy.
        webauthn: Passkey Relying-Party identity and ceremony policy.
        sso: Identity-provider flows and step-up.
        network: Proxy trust and SSRF policy.
        rate_limits: Canonical request budgets.
        api_keys: API-key format and transport policy.
        audit: Audit-stream privacy policy.
    """

    secrets: Secrets

    # --- deployment identity ---
    base_url: str = ""
    app_name: str = "Jafaal"
    environment: str = "production"
    store_key_prefix: str = "jafaal:auth"
    login_token_url: str = "/api/v1/auth/login"
    login_ui_url: str = ""

    # --- registered public clients (RFC 8252) ---
    # Empty by default: a deployment that only serves its own first-party web
    # frontend needs none. Register one per native app that drives the
    # authorization-code flow; the authorization endpoint refuses any
    # client_id / redirect_uri pair not listed here.
    oauth_clients: tuple[OAuthClient, ...] = ()

    # --- deployment-wide security toggles ---
    login_ip_lockout_enabled: bool = True
    allow_in_memory_state_store_when_deployed: bool = False
    allow_no_rate_limit_when_deployed: bool = False

    # --- grouped configuration ---
    tokens: TokenSettings = field(default_factory=TokenSettings)
    sessions: SessionSettings = field(default_factory=SessionSettings)
    passwords: PasswordSettings = field(default_factory=PasswordSettings)
    mfa: MfaSettings = field(default_factory=MfaSettings)
    webauthn: WebAuthnSettings = field(default_factory=WebAuthnSettings)
    sso: SsoSettings = field(default_factory=SsoSettings)
    network: NetworkSettings = field(default_factory=NetworkSettings)
    rate_limits: RateLimitSettings = field(default_factory=RateLimitSettings)
    api_keys: ApiKeySettings = field(default_factory=ApiKeySettings)
    audit: AuditSettings = field(default_factory=AuditSettings)

    def __post_init__(self) -> None:
        if self.environment not in KNOWN_ENVIRONMENTS:
            # Rejected rather than defaulted: ``is_deployed`` gates the cookie
            # ``Secure`` flag, the cookie name prefix, and the two fail-closed
            # startup guards, so an unrecognised value (a typo such as "prod")
            # would silently disable all four in production.
            raise ValueError(
                f"AuthSettings.environment={self.environment!r} is not a recognised environment. "
                f"Use one of {sorted(KNOWN_ENVIRONMENTS)} — {sorted(DEPLOYED_ENVIRONMENTS)} are treated "
                "as deployed (refresh cookies get Secure, and startup fails closed without a distributed "
                "state store and an enforcing rate limiter)."
            )
        # Cross-group rule: the signing algorithm and the key material that
        # backs it are configured separately, so their consistency can only be
        # checked here.
        if self.tokens.is_asymmetric:
            if not self.secrets.private_key:
                raise ValueError(
                    f"tokens.algorithm={self.tokens.algorithm!r} is asymmetric and requires secrets.private_key (PEM)."
                )
            try:
                jwk_keys.import_private_signing_key(self.secrets.private_key, self.tokens.algorithm)
            except ValueError as err:
                raise ValueError(f"Secrets.private_key is invalid: {err}") from err
            for index, fallback in enumerate(self.secrets.private_key_fallbacks):
                try:
                    jwk_keys.import_verification_key(fallback, self.tokens.algorithm)
                except ValueError as err:
                    raise ValueError(f"Secrets.private_key_fallbacks[{index}] is invalid: {err}") from err
        elif self.secrets.private_key or self.secrets.private_key_fallbacks:
            raise ValueError(
                "Secrets.private_key / private_key_fallbacks are set but "
                f"tokens.algorithm={self.tokens.algorithm!r} is symmetric (HS256). Use an asymmetric "
                "algorithm (e.g. RS256 / ES256) or remove the keys."
            )
        seen_client_ids: set[str] = set()
        for client in self.oauth_clients:
            if client.client_id in seen_client_ids:
                raise ValueError(
                    f"AuthSettings.oauth_clients contains duplicate client_id {client.client_id!r}; "
                    "the first match would silently win and the second's redirect_uris would never apply."
                )
            seen_client_ids.add(client.client_id)

    def oauth_client(self, client_id: str) -> OAuthClient | None:
        """Return the registered client with ``client_id``, or ``None``.

        Args:
            client_id: The identifier presented by the caller.

        Returns:
            The matching :class:`OAuthClient`, or ``None`` when unregistered.
        """
        for client in self.oauth_clients:
            if hmac.compare_digest(client.client_id, client_id):
                return client
        return None

    def __repr__(self) -> str:
        """Render the settings, delegating secret redaction to each group's repr."""
        return _redacting_repr(self)

    # --- derived values that span groups ---

    @property
    def is_deployed(self) -> bool:
        """Whether the environment is a deployed one.

        True for every name in :data:`DEPLOYED_ENVIRONMENTS`. The value is
        validated at construction, so an unrecognised environment can never
        silently fall through to ``False``.
        """
        return self.environment in DEPLOYED_ENVIRONMENTS

    @property
    def resolved_issuer(self) -> str:
        """JWT issuer, falling back to :attr:`base_url` when unset."""
        return self.tokens.issuer or self.base_url

    @property
    def resolved_audience(self) -> str:
        """JWT audience, falling back to :attr:`base_url` when unset."""
        return self.tokens.audience or self.base_url

    @property
    def resolved_client_id(self) -> str:
        """``client_id`` claim value, falling back to the resolved audience.

        RFC 9068 §2.2 requires ``client_id`` on an access token. JAFAAL issues
        first-party tokens and has no client registry, so the audience (i.e. the
        application the token is for) is the meaningful identifier unless the
        host sets one explicitly.
        """
        return self.tokens.client_id or self.resolved_audience

    @property
    def _base_url_origin(self) -> tuple[str, ...]:
        """The scheme+host+port origin of :attr:`base_url`, or empty when unusable."""
        parsed = urlparse(self.base_url)
        if parsed.scheme and parsed.netloc:
            return (f"{parsed.scheme}://{parsed.netloc}",)
        return ()

    @property
    def resolved_csrf_trusted_origins(self) -> tuple[str, ...]:
        """Origins permitted to drive the cookie-issuing web flows.

        Returns the explicit :attr:`SessionSettings.csrf_trusted_origins` when
        set, otherwise the origin of :attr:`base_url`. Empty when neither is
        available, in which case the ``Origin`` comparison is skipped and only
        the browser's ``Sec-Fetch-Site`` signal is enforced.
        """
        return self.sessions.csrf_trusted_origins or self._base_url_origin

    @property
    def effective_refresh_cookie_name(self) -> str:
        """Refresh-cookie name including any ``__Secure-`` / ``__Host-`` prefix.

        The prefix is applied only in a *deployed* environment, where the cookie
        is served with ``Secure`` — browsers reject ``__Secure-`` / ``__Host-``
        cookies that arrive without it, which would otherwise break local http
        development. Reads and writes of the refresh cookie must go through this
        name so the set/clear/read sides stay in lockstep.
        """
        if self.sessions.refresh_cookie_prefix and self.is_deployed:
            return f"{self.sessions.refresh_cookie_prefix}{self.sessions.refresh_cookie_name}"
        return self.sessions.refresh_cookie_name

    @property
    def resolved_webauthn_rp_id(self) -> str:
        """WebAuthn Relying Party ID, falling back to the ``base_url`` host.

        Empty when neither is available — the WebAuthn endpoints treat that as a
        misconfiguration and fail fast.
        """
        if self.webauthn.rp_id:
            return self.webauthn.rp_id
        return urlparse(self.base_url).hostname or ""

    @property
    def resolved_webauthn_rp_name(self) -> str:
        """WebAuthn Relying Party display name, falling back to :attr:`app_name`."""
        return self.webauthn.rp_name or self.app_name

    @property
    def resolved_webauthn_origins(self) -> tuple[str, ...]:
        """Expected WebAuthn origins, falling back to the origin of ``base_url``."""
        return self.webauthn.origins or self._base_url_origin

is_deployed property

is_deployed

Whether the environment is a deployed one.

True for every name in :data:DEPLOYED_ENVIRONMENTS. The value is validated at construction, so an unrecognised environment can never silently fall through to False.

resolved_issuer property

resolved_issuer

JWT issuer, falling back to :attr:base_url when unset.

resolved_audience property

resolved_audience

JWT audience, falling back to :attr:base_url when unset.

resolved_client_id property

resolved_client_id

client_id claim value, falling back to the resolved audience.

RFC 9068 §2.2 requires client_id on an access token. JAFAAL issues first-party tokens and has no client registry, so the audience (i.e. the application the token is for) is the meaningful identifier unless the host sets one explicitly.

resolved_csrf_trusted_origins property

resolved_csrf_trusted_origins

Origins permitted to drive the cookie-issuing web flows.

Returns the explicit :attr:SessionSettings.csrf_trusted_origins when set, otherwise the origin of :attr:base_url. Empty when neither is available, in which case the Origin comparison is skipped and only the browser's Sec-Fetch-Site signal is enforced.

effective_refresh_cookie_name property

effective_refresh_cookie_name

Refresh-cookie name including any __Secure- / __Host- prefix.

The prefix is applied only in a deployed environment, where the cookie is served with Secure — browsers reject __Secure- / __Host- cookies that arrive without it, which would otherwise break local http development. Reads and writes of the refresh cookie must go through this name so the set/clear/read sides stay in lockstep.

resolved_webauthn_rp_id property

resolved_webauthn_rp_id

WebAuthn Relying Party ID, falling back to the base_url host.

Empty when neither is available — the WebAuthn endpoints treat that as a misconfiguration and fail fast.

resolved_webauthn_rp_name property

resolved_webauthn_rp_name

WebAuthn Relying Party display name, falling back to :attr:app_name.

resolved_webauthn_origins property

resolved_webauthn_origins

Expected WebAuthn origins, falling back to the origin of base_url.

oauth_client

oauth_client(client_id)

Return the registered client with client_id, or None.

Parameters:

Name	Type	Description	Default
client_id	str	The identifier presented by the caller.	required

Returns:

Type	Description
OAuthClient | None	The matching :class:OAuthClient, or None when unregistered.

Source code in jafaal/settings.py

def oauth_client(self, client_id: str) -> OAuthClient | None:
    """Return the registered client with ``client_id``, or ``None``.

    Args:
        client_id: The identifier presented by the caller.

    Returns:
        The matching :class:`OAuthClient`, or ``None`` when unregistered.
    """
    for client in self.oauth_clients:
        if hmac.compare_digest(client.client_id, client_id):
            return client
    return None

__repr__

__repr__()

Render the settings, delegating secret redaction to each group's repr.

Source code in jafaal/settings.py

def __repr__(self) -> str:
    """Render the settings, delegating secret redaction to each group's repr."""
    return _redacting_repr(self)

MfaSettings dataclass

TOTP replay-protection policy.

Attributes:

Name	Type	Description
totp_replay_fail_open	bool	TOTP single-use replay protection is defense-in-depth on top of the (unchanged) TOTP signature check, but it needs the shared state store. When False (default) an unreachable store fails closed: the code is rejected as a 503 rather than accepted without replay protection. True prefers availability — the code is accepted and the degraded check is logged and audited.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class MfaSettings:
    """TOTP replay-protection policy.

    Attributes:
        totp_replay_fail_open: TOTP single-use replay protection is
            defense-in-depth on top of the (unchanged) TOTP signature check, but
            it needs the shared state store. When ``False`` (default) an
            unreachable store fails *closed*: the code is rejected as a 503
            rather than accepted without replay protection. ``True`` prefers
            availability — the code is accepted and the degraded check is logged
            and audited.
    """

    totp_replay_fail_open: bool = False

NetworkSettings dataclass

Proxy trust, SSRF policy, and the outbound user agent.

Attributes:

Name	Type	Description
trusted_proxies	tuple[str, ...]	Peers and forwarding hops whose X-Forwarded-For / X-Real-IP headers are honoured. Empty by default, which trusts only the direct TCP peer — the safe default: proxy headers from arbitrary clients are ignored, so a client cannot spoof its source IP. Behind a reverse proxy list every hop your infrastructure adds (e.g. both the CDN egress ranges and the reverse proxy), because the forwarded chain is resolved right-to-left and stops at the first hop that is not listed. ("*",) trusts every hop (only safe when a trusted proxy always overwrites the header).
ssrf_allowed_hosts	tuple[str, ...]	Hosts/CIDRs exempted from the SSRF private-address guard on outbound OIDC calls.
user_agent	str	User-Agent header sent on outbound OIDC HTTP calls.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class NetworkSettings:
    """Proxy trust, SSRF policy, and the outbound user agent.

    Attributes:
        trusted_proxies: Peers and forwarding hops whose ``X-Forwarded-For`` /
            ``X-Real-IP`` headers are honoured. Empty by default, which trusts
            only the direct TCP peer — the safe default: proxy headers from
            arbitrary clients are ignored, so a client cannot spoof its source
            IP. Behind a reverse proxy list **every** hop your infrastructure
            adds (e.g. both the CDN egress ranges and the reverse proxy),
            because the forwarded chain is resolved right-to-left and stops at
            the first hop that is not listed. ``("*",)`` trusts every hop (only
            safe when a trusted proxy always overwrites the header).
        ssrf_allowed_hosts: Hosts/CIDRs exempted from the SSRF private-address
            guard on outbound OIDC calls.
        user_agent: ``User-Agent`` header sent on outbound OIDC HTTP calls.
    """

    trusted_proxies: tuple[str, ...] = ()
    ssrf_allowed_hosts: tuple[str, ...] = ()
    user_agent: str = "Jafaal (OIDC Client)"

OAuthClient dataclass

A registered public client, and the policy that applies to its tokens.

The client is the unit of policy. Delivery mode and scope ceiling are properties of the registration, never of the request: a value a caller can choose per-request is a value an attacker can choose, and JAFAAL previously carried both in a header (X-Client-Type) that had to be defended with mismatch detection. Configuration removes the attack surface instead of guarding it.

Registration also supplies the one thing RFC 9700 §4.1 makes mandatory and that is impossible without it: exact redirect_uri matching. Without a registered list there is nothing to match against, and an authorization code can be steered to an attacker-controlled target.

Clients are public (RFC 8252): a native app or browser cannot keep a secret, so PKCE — not client authentication — binds a code to its requester.

Attributes:

Name	Type	Description
client_id	str	The identifier the client sends as client_id. Any stable opaque string; a reverse-DNS name (com.example.app) is conventional for native apps.
redirect_uris	tuple[str, ...]	Every URI the client may receive an authorization code at, matched exactly (byte-for-byte, per RFC 9700 §4.1.3 — no prefix, wildcard, or path-suffix matching). A plain-http URI is accepted only for loopback (RFC 8252 §7.3). Not required for a client that only uses the direct login and refresh endpoints.
token_delivery	str	"body" (default, RFC 6749 §5.1) or "cookie" (RFC 9700 §7.2 — refresh token as an HttpOnly cookie, CSRF token in the body). Browser clients want cookie; everything else wants body.
scopes	tuple[str, ...]	Ceiling on what this client's tokens may carry, intersected with what the host's :class:~jafaal.ports.ScopeResolver grants the user. Empty (default) means "whatever the user holds", which is correct for a first-party app that is the application. Set it for any client you do not control: without a ceiling, registering a client grants it the user's entire account, and JAFAAL has no consent screen to ask the user about it.
name	str	Human-readable label, used in logs and audit records.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class OAuthClient:
    """A registered public client, and the policy that applies to its tokens.

    The client is the **unit of policy**. Delivery mode and scope ceiling are
    properties of the registration, never of the request: a value a caller can
    choose per-request is a value an attacker can choose, and JAFAAL previously
    carried both in a header (``X-Client-Type``) that had to be defended with
    mismatch detection. Configuration removes the attack surface instead of
    guarding it.

    Registration also supplies the one thing RFC 9700 §4.1 makes mandatory and
    that is impossible without it: **exact** ``redirect_uri`` matching. Without a
    registered list there is nothing to match against, and an authorization code
    can be steered to an attacker-controlled target.

    Clients are public (RFC 8252): a native app or browser cannot keep a secret,
    so PKCE — not client authentication — binds a code to its requester.

    Attributes:
        client_id: The identifier the client sends as ``client_id``. Any stable
            opaque string; a reverse-DNS name (``com.example.app``) is
            conventional for native apps.
        redirect_uris: Every URI the client may receive an authorization code
            at, matched **exactly** (byte-for-byte, per RFC 9700 §4.1.3 — no
            prefix, wildcard, or path-suffix matching). A plain-``http`` URI is
            accepted only for loopback (RFC 8252 §7.3). Not required for a
            client that only uses the direct login and refresh endpoints.
        token_delivery: ``"body"`` (default, RFC 6749 §5.1) or ``"cookie"``
            (RFC 9700 §7.2 — refresh token as an ``HttpOnly`` cookie, CSRF token
            in the body). Browser clients want ``cookie``; everything else wants
            ``body``.
        scopes: Ceiling on what this client's tokens may carry, intersected with
            what the host's :class:`~jafaal.ports.ScopeResolver` grants the user.
            Empty (default) means "whatever the user holds", which is correct for
            a first-party app that *is* the application. Set it for any client
            you do not control: without a ceiling, registering a client grants it
            the user's entire account, and JAFAAL has no consent screen to ask
            the user about it.
        name: Human-readable label, used in logs and audit records.
    """

    client_id: str
    redirect_uris: tuple[str, ...] = ()
    token_delivery: str = "body"
    scopes: tuple[str, ...] = ()
    name: str = ""

    def __post_init__(self) -> None:
        if not self.client_id:
            raise ValueError("OAuthClient.client_id is required.")
        if self.token_delivery not in TOKEN_DELIVERY_MODES:
            raise ValueError(
                f"OAuthClient(client_id={self.client_id!r}).token_delivery must be one of "
                f"{sorted(TOKEN_DELIVERY_MODES)} (got {self.token_delivery!r})."
            )
        for uri in self.redirect_uris:
            _validate_redirect_uri(self.client_id, uri)

    def permits(self, redirect_uri: str) -> bool:
        """Return whether ``redirect_uri`` is registered for this client.

        Compared in constant time and byte-for-byte. RFC 9700 §4.1.3 requires
        exact matching precisely because every relaxation (prefix, wildcard,
        sub-path) has been used to exfiltrate authorization codes.
        """
        matched = False
        for registered in self.redirect_uris:
            matched |= hmac.compare_digest(registered, redirect_uri)
        return matched

    @property
    def uses_cookie_delivery(self) -> bool:
        """Whether this client's refresh token is delivered as a cookie.

        The single predicate every delivery decision branches on — cookie write,
        CSRF binding, off-site rejection, and the ``202`` MFA status — so those
        four cannot drift apart.
        """
        return self.token_delivery == "cookie"

    def narrow(self, granted: tuple[str, ...]) -> tuple[str, ...]:
        """Intersect ``granted`` with this client's ceiling.

        Strictly narrowing: a client can only ever receive less than the user
        holds, never more. An empty ceiling means no narrowing.

        Args:
            granted: The scopes the user is entitled to.

        Returns:
            The scopes this client's token may carry.
        """
        if not self.scopes:
            return granted
        ceiling = set(self.scopes)
        return tuple(scope for scope in granted if scope in ceiling)

uses_cookie_delivery property

uses_cookie_delivery

Whether this client's refresh token is delivered as a cookie.

The single predicate every delivery decision branches on — cookie write, CSRF binding, off-site rejection, and the 202 MFA status — so those four cannot drift apart.

permits

permits(redirect_uri)

Return whether redirect_uri is registered for this client.

Compared in constant time and byte-for-byte. RFC 9700 §4.1.3 requires exact matching precisely because every relaxation (prefix, wildcard, sub-path) has been used to exfiltrate authorization codes.

Source code in jafaal/settings.py

def permits(self, redirect_uri: str) -> bool:
    """Return whether ``redirect_uri`` is registered for this client.

    Compared in constant time and byte-for-byte. RFC 9700 §4.1.3 requires
    exact matching precisely because every relaxation (prefix, wildcard,
    sub-path) has been used to exfiltrate authorization codes.
    """
    matched = False
    for registered in self.redirect_uris:
        matched |= hmac.compare_digest(registered, redirect_uri)
    return matched

narrow

narrow(granted)

Intersect granted with this client's ceiling.

Strictly narrowing: a client can only ever receive less than the user holds, never more. An empty ceiling means no narrowing.

Parameters:

Name	Type	Description	Default
granted	tuple[str, ...]	The scopes the user is entitled to.	required

Returns:

Type	Description
tuple[str, ...]	The scopes this client's token may carry.

Source code in jafaal/settings.py

def narrow(self, granted: tuple[str, ...]) -> tuple[str, ...]:
    """Intersect ``granted`` with this client's ceiling.

    Strictly narrowing: a client can only ever receive less than the user
    holds, never more. An empty ceiling means no narrowing.

    Args:
        granted: The scopes the user is entitled to.

    Returns:
        The scopes this client's token may carry.
    """
    if not self.scopes:
        return granted
    ceiling = set(self.scopes)
    return tuple(scope for scope in granted if scope in ceiling)

PasswordSettings dataclass

Argon2 cost parameters and the accepted password length bound.

Attributes:

Name	Type	Description
argon2_time_cost	int	Argon2 time cost (iterations).
argon2_memory_cost	int	Argon2 memory cost, in KiB.
argon2_parallelism	int	Argon2 parallelism (lanes).
max_length	int	Maximum accepted password length, enforced before hashing so an unauthenticated caller cannot force unbounded Argon2 work. Must be at least 64 so long passphrases are accepted (NIST SP 800-63B). Passwords are never truncated.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class PasswordSettings:
    """Argon2 cost parameters and the accepted password length bound.

    Attributes:
        argon2_time_cost: Argon2 time cost (iterations).
        argon2_memory_cost: Argon2 memory cost, in KiB.
        argon2_parallelism: Argon2 parallelism (lanes).
        max_length: Maximum accepted password length, enforced *before* hashing
            so an unauthenticated caller cannot force unbounded Argon2 work.
            Must be at least 64 so long passphrases are accepted (NIST
            SP 800-63B). Passwords are never truncated.
    """

    argon2_time_cost: int = 3
    argon2_memory_cost: int = 65536
    argon2_parallelism: int = 4
    max_length: int = 128

    def __post_init__(self) -> None:
        if self.argon2_time_cost <= 0:
            raise ValueError("PasswordSettings.argon2_time_cost must be positive.")
        if self.argon2_memory_cost <= 0:
            raise ValueError("PasswordSettings.argon2_memory_cost must be positive.")
        if self.argon2_parallelism <= 0:
            raise ValueError("PasswordSettings.argon2_parallelism must be positive.")
        if self.max_length < 64:
            raise ValueError(
                "PasswordSettings.max_length must be at least 64 so long passphrases are "
                "accepted (NIST SP 800-63B recommends allowing at least 64 characters)."
            )
        if self.max_length > PASSWORD_FIELD_MAX_LENGTH:
            raise ValueError(
                f"PasswordSettings.max_length ({self.max_length}) exceeds the transport bound "
                f"PASSWORD_FIELD_MAX_LENGTH ({PASSWORD_FIELD_MAX_LENGTH}); the request schemas would "
                "reject such a password before the policy ever saw it."
            )

RateLimitSettings dataclass

Canonical request budgets for the host's :class:~jafaal.rate_limit.RateLimiter.

Attributes:

Name	Type	Description
sensitive	str	Budget for sensitive endpoints (login, MFA, password reset, sign-up, OAuth, API-key minting).
write	str	Budget for write endpoints (logout, refresh, session and API-key revocation, introspection).

Source code in jafaal/settings.py

@dataclass(frozen=True)
class RateLimitSettings:
    """Canonical request budgets for the host's :class:`~jafaal.rate_limit.RateLimiter`.

    Attributes:
        sensitive: Budget for sensitive endpoints (login, MFA, password reset,
            sign-up, OAuth, API-key minting).
        write: Budget for write endpoints (logout, refresh, session and API-key
            revocation, introspection).
    """

    sensitive: str = "10/minute"
    write: str = "30/minute"

Secrets dataclass

Key material, and the rotation fallbacks that keep it rotatable.

Attributes:

Name	Type	Description
secret_key	str	HMAC key. Signs and verifies HS256 JWTs, and is stretched (HKDF, per purpose) into the keys that MAC every stored token digest — refresh tokens, CSRF tokens, API keys, reset/sign-up/link tokens, the WebAuthn user handle. Required regardless of algorithm.
fernet_key	str	Fernet key used to encrypt at-rest tokens (IdP client secrets, MFA secrets, rotated refresh tokens). A url-safe base64 32-byte key as produced by cryptography.fernet.Fernet.generate_key.
secret_key_fallbacks	tuple[str, ...]	Additional HMAC keys accepted when verifying (never used to sign or to write a new digest), so credentials issued before a secret_key rotation stay valid during the overlap.
fernet_key_fallbacks	tuple[str, ...]	Additional Fernet keys accepted when decrypting (never used to encrypt), enabling fernet_key rotation without a bulk re-encrypt.
private_key	str	PEM private key used to sign JWTs when :attr:TokenSettings.algorithm is asymmetric (must be empty for HS256).
private_key_fallbacks	tuple[str, ...]	Verify-only keys (PEM, public or private) kept in the published JWKS during a signing-key rotation overlap.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class Secrets:
    """Key material, and the rotation fallbacks that keep it rotatable.

    Attributes:
        secret_key: HMAC key. Signs and verifies HS256 JWTs, and is stretched
            (HKDF, per purpose) into the keys that MAC every stored token digest
            — refresh tokens, CSRF tokens, API keys, reset/sign-up/link tokens,
            the WebAuthn user handle. Required regardless of ``algorithm``.
        fernet_key: Fernet key used to encrypt at-rest tokens (IdP client
            secrets, MFA secrets, rotated refresh tokens). A url-safe base64
            32-byte key as produced by ``cryptography.fernet.Fernet.generate_key``.
        secret_key_fallbacks: Additional HMAC keys accepted when *verifying*
            (never used to sign or to write a new digest), so credentials issued
            before a ``secret_key`` rotation stay valid during the overlap.
        fernet_key_fallbacks: Additional Fernet keys accepted when *decrypting*
            (never used to encrypt), enabling ``fernet_key`` rotation without a
            bulk re-encrypt.
        private_key: PEM private key used to sign JWTs when
            :attr:`TokenSettings.algorithm` is asymmetric (must be empty for
            HS256).
        private_key_fallbacks: Verify-only keys (PEM, public or private) kept in
            the published JWKS during a signing-key rotation overlap.
    """

    secret_key: str = field(repr=False)
    fernet_key: str = field(repr=False)
    secret_key_fallbacks: tuple[str, ...] = field(default=(), repr=False)
    fernet_key_fallbacks: tuple[str, ...] = field(default=(), repr=False)
    private_key: str = field(default="", repr=False)
    private_key_fallbacks: tuple[str, ...] = field(default=(), repr=False)

    def __post_init__(self) -> None:
        if not self.secret_key:
            raise ValueError("Secrets.secret_key is required.")
        if len(self.secret_key) < MIN_SECRET_KEY_LENGTH:
            raise ValueError(
                f"Secrets.secret_key is too short (got {len(self.secret_key)} characters, "
                f"need at least {MIN_SECRET_KEY_LENGTH}). HS256 requires a high-entropy key; "
                "generate one with e.g. secrets.token_urlsafe(32)."
            )
        if not self.fernet_key:
            raise ValueError("Secrets.fernet_key is required.")
        try:
            Fernet(self.fernet_key.encode())
        except Exception as err:
            raise ValueError(
                "Secrets.fernet_key is not a valid Fernet key. Expected a url-safe "
                "base64-encoded 32-byte key, e.g. cryptography.fernet.Fernet.generate_key()."
            ) from err
        for index, fallback in enumerate(self.secret_key_fallbacks):
            if len(fallback) < MIN_SECRET_KEY_LENGTH:
                raise ValueError(
                    f"Secrets.secret_key_fallbacks[{index}] is too short (got {len(fallback)} "
                    f"characters, need at least {MIN_SECRET_KEY_LENGTH})."
                )
        for index, fallback in enumerate(self.fernet_key_fallbacks):
            try:
                Fernet(fallback.encode())
            except Exception as err:
                raise ValueError(f"Secrets.fernet_key_fallbacks[{index}] is not a valid Fernet key.") from err

    def __repr__(self) -> str:
        """Render with every key-bearing field as ``<redacted>``."""
        return _redacting_repr(self)

__repr__

__repr__()

Render with every key-bearing field as <redacted>.

Source code in jafaal/settings.py

def __repr__(self) -> str:
    """Render with every key-bearing field as ``<redacted>``."""
    return _redacting_repr(self)

SessionSettings dataclass

Session lifetime, revocation strictness, and refresh-cookie delivery.

Attributes:

Name	Type	Description
idle_timeout_enabled	bool	Whether idle-session expiry is enforced. The absolute lifetime below is always enforced and does not depend on this flag.
idle_timeout_hours	int	Idle-session timeout, in hours.
absolute_timeout_hours	int	Hard ceiling on how long a session may live, measured from created_at and always enforced. A session's expires_at is capped at this deadline on every rotation, so refreshing cannot slide the window forward indefinitely — without the cap, a client that refreshes once per refresh_token_expire_days keeps one login alive forever, which is exactly the unbounded refresh-token lifetime RFC 9700 §4.14.2 warns against. Defaults to 30 days: long enough not to be user-hostile, finite enough that a stolen session eventually dies on its own.
strict_binding	bool	When True, every access-token-authenticated request verifies the token's sid session still exists and is valid, so logout / single-session revocation is immediate instead of bounded by the access-token lifetime. Off by default (stateless access-token validation); adds one indexed session lookup per request. A deactivated user is rejected immediately regardless.
refresh_cookie_name	str	Name of the refresh-token cookie.
refresh_cookie_path	str	Path scope of the refresh-token cookie. Must line up with where the auth router is mounted, or web sessions silently fail to refresh.
refresh_cookie_prefix	str	Optional cookie-name-prefix hardening, applied only in a deployed environment (browsers reject these prefixes on a cookie that arrives without Secure, which would break local http development). "__Secure-" asserts the cookie was set with Secure; "__Host-" additionally binds it to the exact host and requires refresh_cookie_path="/".
csrf_trusted_origins	tuple[str, ...]	Origins allowed to drive the web refresh flow and the cookie-issuing login endpoints. Defaults to the origin of base_url; set explicitly when the frontend is served from a different origin than the API.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class SessionSettings:
    """Session lifetime, revocation strictness, and refresh-cookie delivery.

    Attributes:
        idle_timeout_enabled: Whether *idle*-session expiry is enforced. The
            absolute lifetime below is always enforced and does not depend on
            this flag.
        idle_timeout_hours: Idle-session timeout, in hours.
        absolute_timeout_hours: Hard ceiling on how long a session may live,
            measured from ``created_at`` and **always enforced**. A session's
            ``expires_at`` is capped at this deadline on every rotation, so
            refreshing cannot slide the window forward indefinitely — without
            the cap, a client that refreshes once per ``refresh_token_expire_days``
            keeps one login alive forever, which is exactly the unbounded
            refresh-token lifetime RFC 9700 §4.14.2 warns against. Defaults to
            30 days: long enough not to be user-hostile, finite enough that a
            stolen session eventually dies on its own.
        strict_binding: When ``True``, every access-token-authenticated request
            verifies the token's ``sid`` session still exists and is valid, so
            logout / single-session revocation is immediate instead of bounded
            by the access-token lifetime. Off by default (stateless
            access-token validation); adds one indexed session lookup per
            request. A deactivated *user* is rejected immediately regardless.
        refresh_cookie_name: Name of the refresh-token cookie.
        refresh_cookie_path: Path scope of the refresh-token cookie. Must line
            up with where the auth router is mounted, or web sessions silently
            fail to refresh.
        refresh_cookie_prefix: Optional cookie-name-prefix hardening, applied
            only in a *deployed* environment (browsers reject these prefixes on
            a cookie that arrives without ``Secure``, which would break local
            http development). ``"__Secure-"`` asserts the cookie was set with
            ``Secure``; ``"__Host-"`` additionally binds it to the exact host
            and requires ``refresh_cookie_path="/"``.
        csrf_trusted_origins: Origins allowed to drive the web refresh flow and
            the cookie-issuing login endpoints. Defaults to the origin of
            ``base_url``; set explicitly when the frontend is served from a
            different origin than the API.
    """

    idle_timeout_enabled: bool = False
    idle_timeout_hours: int = 1
    absolute_timeout_hours: int = 720
    strict_binding: bool = False
    refresh_cookie_name: str = "jafaal_refresh_token"
    refresh_cookie_path: str = "/api/v1/auth"
    refresh_cookie_prefix: str = ""
    csrf_trusted_origins: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        if self.idle_timeout_hours <= 0:
            raise ValueError("SessionSettings.idle_timeout_hours must be positive.")
        if self.absolute_timeout_hours <= 0:
            raise ValueError(
                "SessionSettings.absolute_timeout_hours must be positive: a session's lifetime is "
                "always bounded, so there is no value meaning 'never expires'."
            )
        if self.refresh_cookie_prefix not in ("", "__Secure-", "__Host-"):
            raise ValueError(
                "SessionSettings.refresh_cookie_prefix must be '', '__Secure-', or '__Host-' "
                f"(got {self.refresh_cookie_prefix!r})."
            )
        if self.refresh_cookie_prefix == "__Host-" and self.refresh_cookie_path != "/":
            raise ValueError(
                "SessionSettings.refresh_cookie_prefix='__Host-' requires refresh_cookie_path='/': "
                "the __Host- prefix mandates Path=/ and no Domain. Use '__Secure-' to keep a "
                "path-scoped refresh cookie."
            )

SsoSettings dataclass

Identity-provider flows: transport policy and step-up.

Notably absent is anything describing where to send the browser after a flow. Post-login, post-link and post-step-up targets all come from the initiating request's redirect_uri, matched exactly against the calling :class:OAuthClient's registration. A configured frontend path would be a second, weaker redirect rule sitting beside the exact-match one — and the weaker rule is the one an attacker would use.

Attributes:

Name	Type	Description
idp_require_https	bool	When True (default), identity-provider endpoints (the browser-facing authorization endpoint and the server-side token, userinfo, JWKS, discovery and revocation URLs) must use https, refusing to transmit authorization codes, tokens and client credentials in cleartext. Set False only for local or self-hosted development.
step_up_idp_reauth_enabled	bool	Whether an SSO-only account (no local password and no MFA) may satisfy step-up verification by re-authenticating at a linked identity provider. When False such accounts are refused sensitive operations (step-up fails closed).
step_up_reauth_max_age_seconds	int	Maximum age of the IdP authentication (the ID token auth_time claim) accepted as "fresh" for step-up. Also sent as the OIDC max_age so the provider re-prompts.
step_up_grant_ttl_seconds	int	Lifetime of the single-use step-up grant minted after a successful IdP re-authentication; the caller must retry the sensitive operation within this window.
id_token_leeway_seconds	int	Clock-skew tolerance applied to an IdP ID token's exp/iat/nbf. These clocks belong to someone else, and a strict 0 rejects a token whose iat is a single second ahead of ours — OIDC Core §3.1.3.7 (10) anticipates an implementer allowance.
max_response_bytes	int	Largest response body accepted from an identity provider (discovery, JWKS, userinfo). Timeouts bound how long JAFAAL waits, not how much it accepts, and the JWKS is cached — so without a cap one hostile response is a persistent memory cost.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class SsoSettings:
    """Identity-provider flows: transport policy and step-up.

    Notably absent is anything describing *where* to send the browser after a
    flow. Post-login, post-link and post-step-up targets all come from the
    initiating request's ``redirect_uri``, matched exactly against the calling
    :class:`OAuthClient`'s registration. A configured frontend path would be a
    second, weaker redirect rule sitting beside the exact-match one — and the
    weaker rule is the one an attacker would use.

    Attributes:
        idp_require_https: When ``True`` (default), identity-provider endpoints
            (the browser-facing authorization endpoint and the server-side
            token, userinfo, JWKS, discovery and revocation URLs) must use
            ``https``, refusing to transmit authorization codes, tokens and
            client credentials in cleartext. Set ``False`` only for local or
            self-hosted development.
        step_up_idp_reauth_enabled: Whether an SSO-only account (no local
            password and no MFA) may satisfy step-up verification by
            re-authenticating at a linked identity provider. When ``False`` such
            accounts are refused sensitive operations (step-up fails closed).
        step_up_reauth_max_age_seconds: Maximum age of the IdP authentication
            (the ID token ``auth_time`` claim) accepted as "fresh" for step-up.
            Also sent as the OIDC ``max_age`` so the provider re-prompts.
        step_up_grant_ttl_seconds: Lifetime of the single-use step-up grant
            minted after a successful IdP re-authentication; the caller must
            retry the sensitive operation within this window.
        id_token_leeway_seconds: Clock-skew tolerance applied to an IdP ID
            token's ``exp``/``iat``/``nbf``. These clocks belong to someone
            else, and a strict ``0`` rejects a token whose ``iat`` is a single
            second ahead of ours — OIDC Core §3.1.3.7 (10) anticipates an
            implementer allowance.
        max_response_bytes: Largest response body accepted from an identity
            provider (discovery, JWKS, userinfo). Timeouts bound how long JAFAAL
            waits, not how much it accepts, and the JWKS is cached — so without
            a cap one hostile response is a persistent memory cost.
    """

    idp_require_https: bool = True
    step_up_idp_reauth_enabled: bool = True
    step_up_reauth_max_age_seconds: int = 300
    step_up_grant_ttl_seconds: int = 120
    id_token_leeway_seconds: int = 60
    max_response_bytes: int = 1024 * 1024

    def __post_init__(self) -> None:
        if self.step_up_reauth_max_age_seconds <= 0:
            raise ValueError("SsoSettings.step_up_reauth_max_age_seconds must be positive.")
        if self.step_up_grant_ttl_seconds <= 0:
            raise ValueError("SsoSettings.step_up_grant_ttl_seconds must be positive.")
        if self.id_token_leeway_seconds < 0:
            raise ValueError("SsoSettings.id_token_leeway_seconds must be non-negative.")
        if self.max_response_bytes <= 0:
            raise ValueError("SsoSettings.max_response_bytes must be positive.")

TokenSettings dataclass

JWT issuance, lifetimes, and the opt-in immediate-revocation controls.

Attributes:

Name	Type	Description
algorithm	str	JWT signing algorithm. HS256 (default, symmetric) or an asymmetric RSA/EC algorithm (RS256/384/512, PS256/384/512, ES256/384/512), which signs with :attr:Secrets.private_key and publishes the public key at the JWKS endpoint.
access_token_expire_minutes	int	Access-token lifetime, in minutes.
refresh_token_expire_days	int	Refresh-token lifetime, in days.
issuer	str	JWT iss claim. Defaults to base_url when empty (see :attr:AuthSettings.resolved_issuer).
audience	str	JWT aud claim. Defaults to base_url when empty (see :attr:AuthSettings.resolved_audience).
client_id	str	Value of the client_id claim RFC 9068 requires on an access token. JAFAAL is a first-party issuer with no client registry, so this defaults to the resolved audience.
leeway_seconds	int	Clock-skew tolerance applied to the exp/nbf claims of JAFAAL's own JWTs. 0 (default) is strict; a small value avoids spurious 401s across slightly skewed nodes. Kept small so an expired token is not honoured for long.
denylist_enabled	bool	When True, revoked access-token jti values are recorded and checked per request so /revoke kills an access token immediately (one state-store lookup per request). Off by default: access tokens are short-lived and lapse at expiry, and revoking the refresh token (which deletes the session) is the always-effective revocation path.
reauthorize_scopes_per_request	bool	When True, an access token's scopes are intersected with what the host's :class:~jafaal.ports.ScopeResolver grants the account on this request, so a demotion applies immediately rather than at token expiry. Strictly narrowing — a token never gains a scope it was not issued with. Off by default; adds no query (the user row is already loaded every request), only the narrowing. Applies to access tokens only: API keys are narrowed unconditionally, because they are long-lived (expires_at is optional) and so cannot rely on expiry to shed stale authority.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class TokenSettings:
    """JWT issuance, lifetimes, and the opt-in immediate-revocation controls.

    Attributes:
        algorithm: JWT signing algorithm. ``HS256`` (default, symmetric) or an
            asymmetric RSA/EC algorithm (RS256/384/512, PS256/384/512,
            ES256/384/512), which signs with :attr:`Secrets.private_key` and
            publishes the public key at the JWKS endpoint.
        access_token_expire_minutes: Access-token lifetime, in minutes.
        refresh_token_expire_days: Refresh-token lifetime, in days.
        issuer: JWT ``iss`` claim. Defaults to ``base_url`` when empty
            (see :attr:`AuthSettings.resolved_issuer`).
        audience: JWT ``aud`` claim. Defaults to ``base_url`` when empty
            (see :attr:`AuthSettings.resolved_audience`).
        client_id: Value of the ``client_id`` claim RFC 9068 requires on an
            access token. JAFAAL is a first-party issuer with no client
            registry, so this defaults to the resolved audience.
        leeway_seconds: Clock-skew tolerance applied to the ``exp``/``nbf``
            claims of JAFAAL's own JWTs. ``0`` (default) is strict; a small
            value avoids spurious 401s across slightly skewed nodes. Kept small
            so an expired token is not honoured for long.
        denylist_enabled: When ``True``, revoked access-token ``jti`` values are
            recorded and checked per request so ``/revoke`` kills an access
            token immediately (one state-store lookup per request). Off by
            default: access tokens are short-lived and lapse at expiry, and
            revoking the refresh token (which deletes the session) is the
            always-effective revocation path.
        reauthorize_scopes_per_request: When ``True``, an access token's scopes
            are intersected with what the host's
            :class:`~jafaal.ports.ScopeResolver` grants the account on *this*
            request, so a demotion applies immediately rather than at token
            expiry. Strictly narrowing — a token never gains a scope it was not
            issued with. Off by default; adds no query (the user row is already
            loaded every request), only the narrowing. Applies to access tokens
            only: API keys are narrowed unconditionally, because they are
            long-lived (``expires_at`` is optional) and so cannot rely on expiry
            to shed stale authority.
    """

    algorithm: str = "HS256"
    access_token_expire_minutes: int = 15
    refresh_token_expire_days: int = 7
    issuer: str = ""
    audience: str = ""
    client_id: str = ""
    leeway_seconds: int = 0
    denylist_enabled: bool = False
    reauthorize_scopes_per_request: bool = False

    def __post_init__(self) -> None:
        if self.algorithm not in ALLOWED_ALGORITHMS:
            raise ValueError(
                f"TokenSettings.algorithm={self.algorithm!r} is not in the allow-list {sorted(ALLOWED_ALGORITHMS)}"
            )
        if self.access_token_expire_minutes <= 0:
            raise ValueError("TokenSettings.access_token_expire_minutes must be positive.")
        if self.refresh_token_expire_days <= 0:
            raise ValueError("TokenSettings.refresh_token_expire_days must be positive.")
        if self.leeway_seconds < 0:
            raise ValueError("TokenSettings.leeway_seconds must be non-negative.")

    @property
    def is_asymmetric(self) -> bool:
        """Whether :attr:`algorithm` signs with a private key rather than a shared secret."""
        return self.algorithm in jwk_keys.ASYMMETRIC_ALGORITHMS

is_asymmetric property

is_asymmetric

Whether :attr:algorithm signs with a private key rather than a shared secret.

WebAuthnSettings dataclass

Relying-Party identity and ceremony policy for passkeys.

Attributes:

Name	Type	Description
rp_id	str	Relying Party ID — the registrable domain passkeys are scoped to (e.g. "example.com"; no scheme or port). Must be a registrable suffix of every origin the app is served from. Defaults to the base_url host.
rp_name	str	Human-readable Relying Party name shown by the authenticator. Defaults to app_name.
origins	tuple[str, ...]	Exact origins (scheme + host + port) a ceremony may complete from. Defaults to the origin of base_url.
user_verification	str	User-verification requirement for the second-factor ceremony: "required" forces the authenticator to verify the user (PIN/biometric), "preferred" (default) verifies where supported, "discouraged" skips it. The passwordless ceremony always requires user verification regardless of this value — there the passkey is the entire authentication.
attestation	str	Attestation conveyance requested at registration. "none" (default) asks for no attestation statement (best for privacy and interoperability); "direct" requests it so the host can inspect the authenticator model. JAFAAL does not verify attestation certificates — request "direct" only if the host processes them.
second_factor_enabled	bool	When True, a user with registered passkeys must present one as a second factor after a successful password login. Off by default; passwordless authentication is always available regardless.
passkey_login_satisfies_mfa	bool	Whether a passwordless passkey login completes on its own for an account that also has TOTP enrolled. True (default) treats the ceremony as already multi-factor: the passwordless path always demands user verification, so the authenticator has checked possession and a PIN/biometric before it will sign. False refuses the shortcut and makes such an account log in with its password + TOTP, for deployments whose policy names TOTP specifically rather than "two factors". Accounts without TOTP are unaffected either way.
challenge_ttl_seconds	int	Lifetime of a challenge held in the state store before it must be redeemed. Kept short (a ceremony is interactive and immediate) to bound replay of a leaked challenge.

Source code in jafaal/settings.py

@dataclass(frozen=True)
class WebAuthnSettings:
    """Relying-Party identity and ceremony policy for passkeys.

    Attributes:
        rp_id: Relying Party ID — the registrable domain passkeys are scoped to
            (e.g. ``"example.com"``; no scheme or port). Must be a registrable
            suffix of every origin the app is served from. Defaults to the
            ``base_url`` host.
        rp_name: Human-readable Relying Party name shown by the authenticator.
            Defaults to ``app_name``.
        origins: Exact origins (scheme + host + port) a ceremony may complete
            from. Defaults to the origin of ``base_url``.
        user_verification: User-verification requirement for the *second-factor*
            ceremony: ``"required"`` forces the authenticator to verify the user
            (PIN/biometric), ``"preferred"`` (default) verifies where supported,
            ``"discouraged"`` skips it. The passwordless ceremony always
            requires user verification regardless of this value — there the
            passkey is the entire authentication.
        attestation: Attestation conveyance requested at registration. ``"none"``
            (default) asks for no attestation statement (best for privacy and
            interoperability); ``"direct"`` requests it so the host can inspect
            the authenticator model. JAFAAL does not verify attestation
            certificates — request ``"direct"`` only if the host processes them.
        second_factor_enabled: When ``True``, a user with registered passkeys
            must present one as a second factor after a successful password
            login. Off by default; passwordless authentication is always
            available regardless.
        passkey_login_satisfies_mfa: Whether a passwordless passkey login
            completes on its own for an account that also has TOTP enrolled.
            ``True`` (default) treats the ceremony as already multi-factor: the
            passwordless path always demands user verification, so the
            authenticator has checked possession *and* a PIN/biometric before it
            will sign. ``False`` refuses the shortcut and makes such an account
            log in with its password + TOTP, for deployments whose policy names
            TOTP specifically rather than "two factors". Accounts without TOTP
            are unaffected either way.
        challenge_ttl_seconds: Lifetime of a challenge held in the state store
            before it must be redeemed. Kept short (a ceremony is interactive
            and immediate) to bound replay of a leaked challenge.
    """

    rp_id: str = ""
    rp_name: str = ""
    origins: tuple[str, ...] = ()
    user_verification: str = "preferred"
    attestation: str = "none"
    second_factor_enabled: bool = False
    passkey_login_satisfies_mfa: bool = True
    challenge_ttl_seconds: int = 300

    def __post_init__(self) -> None:
        if self.user_verification not in ("required", "preferred", "discouraged"):
            raise ValueError(
                "WebAuthnSettings.user_verification must be 'required', 'preferred', or "
                f"'discouraged' (got {self.user_verification!r})."
            )
        if self.attestation not in ("none", "direct"):
            raise ValueError(f"WebAuthnSettings.attestation must be 'none' or 'direct' (got {self.attestation!r}).")
        if self.challenge_ttl_seconds <= 0:
            raise ValueError("WebAuthnSettings.challenge_ttl_seconds must be positive.")

InMemoryStateStore

Process-local :class:StateStore backed by a dict with per-key TTL expiry.

Correct for a single process only — it is not shared across workers or replicas, so a multi-worker/replica deployment must configure a distributed backend instead. Access is guarded by a lock because FastAPI runs sync handlers in a threadpool.

Source code in jafaal/state_store.py

class InMemoryStateStore:
    """Process-local :class:`StateStore` backed by a dict with per-key TTL expiry.

    Correct for a single process only — it is not shared across workers or
    replicas, so a multi-worker/replica deployment must configure a distributed
    backend instead. Access is guarded by a lock because FastAPI runs sync
    handlers in a threadpool.
    """

    def __init__(self) -> None:
        self._data: dict[str, tuple[bytes, float | None]] = {}
        self._lock = threading.Lock()

    @staticmethod
    def _is_expired(expiry: float | None) -> bool:
        return expiry is not None and expiry <= time.monotonic()

    def _live_value(self, key: str) -> bytes | None:
        """Return the unexpired value for ``key``, evicting it if it has expired."""
        entry = self._data.get(key)
        if entry is None:
            return None
        value, expiry = entry
        if self._is_expired(expiry):
            del self._data[key]
            return None
        return value

    def get(self, key: str) -> bytes | None:
        with self._lock:
            return self._live_value(key)

    def set(self, key: str, value: bytes, ttl_seconds: int | None = None) -> None:
        with self._lock:
            expiry = time.monotonic() + ttl_seconds if ttl_seconds is not None else None
            self._data[key] = (value, expiry)

    def delete(self, key: str) -> None:
        with self._lock:
            self._data.pop(key, None)

    def delete_prefix(self, prefix: str) -> int:
        with self._lock:
            matching = [key for key in self._data if key.startswith(prefix)]
            for key in matching:
                del self._data[key]
            return len(matching)

    def get_and_delete(self, key: str) -> bytes | None:
        with self._lock:
            value = self._live_value(key)
            if value is not None:
                del self._data[key]
            return value

    def set_if_absent(self, key: str, value: bytes, ttl_seconds: int) -> bool:
        with self._lock:
            # ``_live_value`` evicts an expired entry, so an expired claim is
            # correctly treated as absent and can be re-claimed.
            if self._live_value(key) is not None:
                return False
            self._data[key] = (value, time.monotonic() + ttl_seconds)
            return True

    def increment(self, key: str, ttl_seconds: int) -> int:
        with self._lock:
            current = self._live_value(key)
            count = (int(current.decode()) if current is not None else 0) + 1
            self._data[key] = (str(count).encode(), time.monotonic() + ttl_seconds)
            return count

    def iter_keys(self, prefix: str) -> Iterator[str]:
        with self._lock:
            # Snapshot live matching keys under the lock; ``list(self._data)``
            # guards against the eviction that ``_live_value`` performs while
            # scanning.
            live_keys = [
                key for key in list(self._data) if key.startswith(prefix) and self._live_value(key) is not None
            ]
        return iter(live_keys)

    def record_tiered_failure(
        self,
        counter_key: str,
        gate_key: str,
        tiers: tuple[tuple[int, int], ...],
        counter_ttl_seconds: int,
    ) -> TieredFailureOutcome:
        now = int(time.time())
        with self._lock:
            # Already locked: return the current count without incrementing, so
            # a locked-out caller cannot keep inflating the counter.
            gate_bytes = self._live_value(gate_key)
            if gate_bytes is not None:
                gate_until = int(gate_bytes.decode())
                if gate_until > now:
                    counter_bytes = self._live_value(counter_key)
                    count = int(counter_bytes.decode()) if counter_bytes is not None else 0
                    return TieredFailureOutcome(count, gate_until, False)
                self._data.pop(gate_key, None)  # expired gate

            counter_bytes = self._live_value(counter_key)
            count = (int(counter_bytes.decode()) if counter_bytes is not None else 0) + 1
            self._data[counter_key] = (str(count).encode(), time.monotonic() + counter_ttl_seconds)

            lock_seconds = 0
            for threshold, tier_lock_seconds in tiers:  # ascending; last match wins
                if count >= threshold:
                    lock_seconds = tier_lock_seconds

            if lock_seconds > 0:
                gate_until = now + lock_seconds
                self._data[gate_key] = (str(gate_until).encode(), time.monotonic() + lock_seconds)
                return TieredFailureOutcome(count, gate_until, True)
            return TieredFailureOutcome(count, None, False)

StateStore

Bases: Protocol

Ephemeral keyed state (counters, TTL flags, small blobs).

The single seam through which the auth/MFA stores read and write short-lived shared state, so a store never needs to know whether it is backed by a process-local dict or Redis.

Source code in jafaal/state_store.py

@runtime_checkable
class StateStore(Protocol):
    """Ephemeral keyed state (counters, TTL flags, small blobs).

    The single seam through which the auth/MFA stores read and write short-lived
    shared state, so a store never needs to know whether it is backed by a
    process-local dict or Redis.
    """

    def get(self, key: str) -> bytes | None: ...

    def set(self, key: str, value: bytes, ttl_seconds: int | None = None) -> None: ...

    def delete(self, key: str) -> None: ...

    def delete_prefix(self, prefix: str) -> int: ...

    def get_and_delete(self, key: str) -> bytes | None: ...

    def set_if_absent(self, key: str, value: bytes, ttl_seconds: int) -> bool:
        """Atomically create ``key`` only if it does not already exist.

        The single-writer primitive behind "claim this exactly once" semantics
        (single-use TOTP timesteps, and any future one-shot marker). It must be
        atomic in the backend: a check-then-``set`` pair in calling code lets two
        concurrent requests both observe "absent" and both proceed, which is
        precisely the race this primitive exists to remove.

        Args:
            key: The key to claim.
            value: The value to store when the claim succeeds.
            ttl_seconds: Lifetime of the claim; it expires automatically so the
                key space stays bounded.

        Returns:
            True when *this* call created the key (the caller owns the claim),
            False when it already existed (someone else claimed it first).
        """
        ...

    def increment(self, key: str, ttl_seconds: int) -> int:
        """Atomically add 1 to the counter at ``key`` and return the new value.

        The counter carries ``ttl_seconds`` TTL, so it is self-expiring — used by
        the reference rate limiter for fixed-window request counting.
        """
        ...

    def iter_keys(self, prefix: str) -> Iterator[str]: ...

    def record_tiered_failure(
        self,
        counter_key: str,
        gate_key: str,
        tiers: tuple[tuple[int, int], ...],
        counter_ttl_seconds: int,
    ) -> TieredFailureOutcome: ...

set_if_absent

set_if_absent(key, value, ttl_seconds)

Atomically create key only if it does not already exist.

The single-writer primitive behind "claim this exactly once" semantics (single-use TOTP timesteps, and any future one-shot marker). It must be atomic in the backend: a check-then-set pair in calling code lets two concurrent requests both observe "absent" and both proceed, which is precisely the race this primitive exists to remove.

Parameters:

Name	Type	Description	Default
key	str	The key to claim.	required
value	bytes	The value to store when the claim succeeds.	required
ttl_seconds	int	Lifetime of the claim; it expires automatically so the key space stays bounded.	required

Returns:

Type	Description
bool	True when this call created the key (the caller owns the claim),
bool	False when it already existed (someone else claimed it first).

Source code in jafaal/state_store.py

def set_if_absent(self, key: str, value: bytes, ttl_seconds: int) -> bool:
    """Atomically create ``key`` only if it does not already exist.

    The single-writer primitive behind "claim this exactly once" semantics
    (single-use TOTP timesteps, and any future one-shot marker). It must be
    atomic in the backend: a check-then-``set`` pair in calling code lets two
    concurrent requests both observe "absent" and both proceed, which is
    precisely the race this primitive exists to remove.

    Args:
        key: The key to claim.
        value: The value to store when the claim succeeds.
        ttl_seconds: Lifetime of the claim; it expires automatically so the
            key space stays bounded.

    Returns:
        True when *this* call created the key (the caller owns the claim),
        False when it already existed (someone else claimed it first).
    """
    ...

increment

increment(key, ttl_seconds)

Atomically add 1 to the counter at key and return the new value.

The counter carries ttl_seconds TTL, so it is self-expiring — used by the reference rate limiter for fixed-window request counting.

Source code in jafaal/state_store.py

def increment(self, key: str, ttl_seconds: int) -> int:
    """Atomically add 1 to the counter at ``key`` and return the new value.

    The counter carries ``ttl_seconds`` TTL, so it is self-expiring — used by
    the reference rate limiter for fixed-window request counting.
    """
    ...

StateStoreUnavailableError

Bases: StoreUnavailableError

Raised by a :class:StateStore when its backing store is unreachable.

Lets the auth/MFA stores react to an infrastructure outage (surface a 503, or swallow a best-effort cleanup) without importing anything about the concrete backend. :class:InMemoryStateStore never raises it.

Source code in jafaal/state_store.py

class StateStoreUnavailableError(StoreUnavailableError):
    """Raised by a :class:`StateStore` when its backing store is unreachable.

    Lets the auth/MFA stores react to an infrastructure outage (surface a 503,
    or swallow a best-effort cleanup) without importing anything about the
    concrete backend. :class:`InMemoryStateStore` never raises it.
    """

TieredFailureOutcome dataclass

Result of an atomic tiered-lockout increment.

Attributes:

Name	Type	Description
count	int	The failure counter value after this attempt.
locked_until_epoch	int | None	Wall-clock epoch (seconds) the lock is active until, or None when not locked.
newly_locked	bool	True only when this call created (or renewed) the lock.

Source code in jafaal/state_store.py

@dataclass(frozen=True)
class TieredFailureOutcome:
    """Result of an atomic tiered-lockout increment.

    Attributes:
        count: The failure counter value after this attempt.
        locked_until_epoch: Wall-clock epoch (seconds) the lock is active until,
            or ``None`` when not locked.
        newly_locked: True only when *this* call created (or renewed) the lock.
    """

    count: int
    locked_until_epoch: int | None
    newly_locked: bool

IntPKUserMixin

Bases: UserMixin

User columns with an auto-incrementing integer primary key.

Use for applications that prefer compact, sequential identifiers.

Source code in jafaal/user_model.py

class IntPKUserMixin(UserMixin):
    """User columns with an auto-incrementing integer primary key.

    Use for applications that prefer compact, sequential identifiers.
    """

    id: Mapped[int] = mapped_column(
        primary_key=True,
        autoincrement=True,
    )

UserMixin

Auth-relevant user columns, excluding the primary key.

This mixin is not mapped on its own. Combine it with a concrete primary-key mixin (:class:IntPKUserMixin or :class:UUIDPKUserMixin) and the application's declarative Base to produce a mapped user model. Host applications add their own profile columns and relationships on the concrete subclass.

Source code in jafaal/user_model.py

class UserMixin:
    """Auth-relevant user columns, excluding the primary key.

    This mixin is not mapped on its own. Combine it with a concrete
    primary-key mixin (:class:`IntPKUserMixin` or :class:`UUIDPKUserMixin`)
    and the application's declarative ``Base`` to produce a mapped user
    model. Host applications add their own profile columns and relationships
    on the concrete subclass.
    """

    username: Mapped[str] = mapped_column(
        String(250),
        unique=True,
        index=True,
        nullable=False,
        comment="Unique login handle",
    )
    email: Mapped[str] = mapped_column(
        # RFC 5321 caps an e-mail address at 254 characters.
        String(254),
        unique=True,
        index=True,
        nullable=False,
        comment="Unique e-mail address",
    )
    is_active: Mapped[bool] = mapped_column(
        default=True,
        nullable=False,
        comment="Whether the account may authenticate",
    )
    is_superuser: Mapped[bool] = mapped_column(
        default=False,
        nullable=False,
        comment="Whether the account holds administrative scope",
    )
    is_verified: Mapped[bool] = mapped_column(
        default=False,
        nullable=False,
        comment="Whether the e-mail address has been verified",
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
        comment="Row creation timestamp (UTC)",
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
        nullable=False,
        comment="Last-update timestamp (UTC)",
    )

    # ------------------------------------------------------------------
    # Reverse relationships to JAFAAL's auth-owned tables.
    #
    # Declared via ``declared_attr`` so any host user model composing this
    # mixin automatically gets the counterparts that JAFAAL's models
    # ``back_populates``. All resolve by class name within JAFAAL's single
    # registry (:data:`jafaal.orm.Base`); the host declares none of them.
    # ------------------------------------------------------------------

    @declared_attr
    def users_sessions(cls) -> Mapped[list[UsersSessions]]:
        return relationship("UsersSessions", back_populates="users", cascade="all, delete-orphan")

    @declared_attr
    def users_api_keys(cls) -> Mapped[list[UsersApiKeys]]:
        return relationship("UsersApiKeys", back_populates="users", cascade="all, delete-orphan")

    @declared_attr
    def password_reset_tokens(cls) -> Mapped[list[PasswordResetToken]]:
        return relationship("PasswordResetToken", back_populates="users", cascade="all, delete-orphan")

    @declared_attr
    def sign_up_tokens(cls) -> Mapped[list[SignUpToken]]:
        return relationship("SignUpToken", back_populates="users", cascade="all, delete-orphan")

    @declared_attr
    def user_identity_providers(cls) -> Mapped[list[IdentityLink]]:
        return relationship("IdentityLink", back_populates="users", cascade="all, delete-orphan")

    @declared_attr
    def idp_link_tokens(cls) -> Mapped[list[IdpLinkToken]]:
        return relationship("IdpLinkToken", back_populates="users", cascade="all, delete-orphan")

    @declared_attr
    def oauth_states(cls) -> Mapped[list[OAuthState]]:
        return relationship("OAuthState", back_populates="users", cascade="all, delete-orphan")

    @declared_attr
    def mfa_backup_codes(cls) -> Mapped[list[MFABackupCode]]:
        return relationship("MFABackupCode", back_populates="users", cascade="all, delete-orphan")

    @declared_attr
    def webauthn_credentials(cls) -> Mapped[list[WebAuthnCredential]]:
        return relationship("WebAuthnCredential", back_populates="users", cascade="all, delete-orphan")

    @declared_attr
    def auth_mfa(cls) -> Mapped[UsersMFA | None]:
        return relationship("UsersMFA", back_populates="users", uselist=False, cascade="all, delete-orphan")

    @declared_attr
    def local_credential(cls) -> Mapped[LocalCredential | None]:
        return relationship("LocalCredential", back_populates="users", uselist=False, cascade="all, delete-orphan")

    @property
    def mfa_enabled(self) -> bool:
        """Return ``True`` when MFA is active for this user."""
        return bool(self.auth_mfa and self.auth_mfa.mfa_enabled)

mfa_enabled property

mfa_enabled

Return True when MFA is active for this user.

UUIDPKUserMixin

Bases: UserMixin

User columns with a UUID primary key.

The identifier defaults to a random UUID4 generated application-side, so it is available before the row is flushed and does not leak account counts or creation order.

Source code in jafaal/user_model.py

class UUIDPKUserMixin(UserMixin):
    """User columns with a UUID primary key.

    The identifier defaults to a random UUID4 generated application-side, so
    it is available before the row is flushed and does not leak account
    counts or creation order.
    """

    id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        primary_key=True,
        default=uuid.uuid4,
    )

AuthContext dataclass

Unified authentication context.

Carries the resolved user identity and scopes regardless of whether authentication was via JWT or API key.

Attributes:

Name	Type	Description
user_id	UserId	Authenticated user's ID.
scopes	list[str]	List of granted scope strings.
auth_type	str	Source of authentication ("jwt" or "api_key").

Source code in jafaal/_internal/internal_dependencies.py

@dataclass
class AuthContext:
    """
    Unified authentication context.

    Carries the resolved user identity and scopes
    regardless of whether authentication was via JWT
    or API key.

    Attributes:
        user_id: Authenticated user's ID.
        scopes: List of granted scope strings.
        auth_type: Source of authentication
            (``"jwt"`` or ``"api_key"``).
    """

    user_id: jafaal_orm.UserId
    scopes: list[str]
    auth_type: str

get_password_hasher

get_password_hasher()

Return the process-wide password hasher.

When JAFAAL is configured, the hasher is built from the Argon2 cost parameters in :class:~jafaal.settings.AuthSettings (argon2_time_cost / argon2_memory_cost / argon2_parallelism), cached, and transparently rebuilt if :func:jafaal.configure is called again (mirroring get_token_manager). Before configuration it falls back to the default-cost singleton, so isolated password hashing/verification works without installing settings. Argon2 hashes are self-describing, so a hash produced at one cost still verifies (and is transparently upgraded via verify_and_update) at another.

Returns:

Name	Type	Description
PasswordHasher	PasswordHasher	The active password hasher.

Source code in jafaal/_internal/password_hasher.py

def get_password_hasher() -> PasswordHasher:
    """Return the process-wide password hasher.

    When JAFAAL is configured, the hasher is built from the Argon2 cost
    parameters in :class:`~jafaal.settings.AuthSettings` (``argon2_time_cost`` /
    ``argon2_memory_cost`` / ``argon2_parallelism``), cached, and transparently
    rebuilt if :func:`jafaal.configure` is called again (mirroring
    ``get_token_manager``). Before configuration it falls back to the
    default-cost singleton, so isolated password hashing/verification works
    without installing settings. Argon2 hashes are self-describing, so a hash
    produced at one cost still verifies (and is transparently upgraded via
    ``verify_and_update``) at another.

    Returns:
        PasswordHasher: The active password hasher.
    """
    global _settings_password_hasher, _settings_password_hasher_generation
    if not jafaal_settings.is_configured():
        return password_hasher
    generation = jafaal_settings.settings_generation()
    if _settings_password_hasher is None or _settings_password_hasher_generation != generation:
        settings = jafaal_settings.get_settings()
        _settings_password_hasher = PasswordHasher(
            hasher=Argon2Hasher(
                time_cost=settings.passwords.argon2_time_cost,
                memory_cost=settings.passwords.argon2_memory_cost,
                parallelism=settings.passwords.argon2_parallelism,
            )
        )
        _settings_password_hasher_generation = generation
    return _settings_password_hasher

cleanup_expired_pending_mfa_logins

cleanup_expired_pending_mfa_logins()

Evict all expired pending MFA login entries (no-op; TTL-managed).

Returns the number of entries evicted (always 0 — the backend expires pending entries by TTL).

Source code in jafaal/_internal/security_stores.py

def cleanup_expired_pending_mfa_logins() -> int:
    """Evict all expired pending MFA login entries (no-op; TTL-managed).

    Returns the number of entries evicted (always ``0`` — the backend expires
    pending entries by TTL).
    """
    return pending_mfa_store.cleanup_expired()

clear_pending_mfa_for_user

clear_pending_mfa_for_user(user_id)

Remove pending MFA login entries for a user across credential changes.

Called from password-change paths so that an attacker who already submitted the now-rotated password and is sitting at the pending-MFA step cannot still complete the login. Storage outages are logged and swallowed because the password rotation itself must remain successful; pending entries expire naturally after their 5-minute TTL.

Parameters:

Name	Type	Description	Default
user_id	UserId	User ID whose pending MFA entries should be removed.	required

Returns:

Type	Description
int	Number of pending MFA entries removed (zero on storage outage).

Source code in jafaal/_internal/security_stores.py

def clear_pending_mfa_for_user(user_id: UserId) -> int:
    """
    Remove pending MFA login entries for a user across credential changes.

    Called from password-change paths so that an attacker who already submitted
    the now-rotated password and is sitting at the pending-MFA step cannot still
    complete the login. Storage outages are logged and swallowed because the
    password rotation itself must remain successful; pending entries expire
    naturally after their 5-minute TTL.

    Args:
        user_id: User ID whose pending MFA entries should be removed.

    Returns:
        Number of pending MFA entries removed (zero on storage outage).
    """
    try:
        return pending_mfa_store.clear_for_user(user_id)
    except AuthSecurityStoreUnavailableError as err:
        logger.warning(
            "Failed to clear pending MFA entries during password change; entries will expire naturally via TTL",
            exc_info=err,
        )
        return 0

get_failed_login_attempts

get_failed_login_attempts()

Dependency injection for failed-login attempt storage.

Source code in jafaal/_internal/security_stores.py

def get_failed_login_attempts() -> FailedLoginStore:
    """Dependency injection for failed-login attempt storage."""
    return failed_login_attempts

get_pending_mfa_store

get_pending_mfa_store()

Dependency injection for pending MFA storage.

Source code in jafaal/_internal/security_stores.py

def get_pending_mfa_store() -> PendingMFAStore:
    """Dependency injection for pending MFA storage."""
    return pending_mfa_store

get_step_up_attempts

get_step_up_attempts()

Dependency injection for step-up attempt tracking.

Source code in jafaal/_internal/security_stores.py

def get_step_up_attempts() -> StepUpStore:
    """Dependency injection for step-up attempt tracking."""
    return step_up_attempts

get_token_manager

get_token_manager()

Return a process-wide :class:TokenManager built from settings.

The instance is cached and transparently rebuilt if :func:jafaal.configure is called again (detected via the settings generation counter).

Returns:

Name	Type	Description
TokenManager	TokenManager	Token manager bound to the installed AuthSettings.

Source code in jafaal/_internal/token_manager.py

def get_token_manager() -> TokenManager:
    """Return a process-wide :class:`TokenManager` built from settings.

    The instance is cached and transparently rebuilt if :func:`jafaal.configure`
    is called again (detected via the settings generation counter).

    Returns:
        TokenManager: Token manager bound to the installed ``AuthSettings``.
    """
    global _token_manager, _token_manager_generation
    generation = jafaal_settings.settings_generation()
    if _token_manager is None or _token_manager_generation != generation:
        settings = jafaal_settings.get_settings()
        _token_manager = TokenManager(
            settings.secrets.secret_key,
            settings.tokens.algorithm,
            access_token_expire_minutes=settings.tokens.access_token_expire_minutes,
            refresh_token_expire_days=settings.tokens.refresh_token_expire_days,
            issuer=settings.resolved_issuer,
            audience=settings.resolved_audience,
            secret_key_fallbacks=settings.secrets.secret_key_fallbacks,
            private_key=settings.secrets.private_key,
            private_key_fallbacks=settings.secrets.private_key_fallbacks,
            leeway_seconds=settings.tokens.leeway_seconds,
            client_id=settings.resolved_client_id,
        )
        _token_manager_generation = generation
    return _token_manager

jafaal_exception_handler async

jafaal_exception_handler(request, exc)

Translate a :class:JafaalError into a JSON HTTP response.

Emits JAFAAL's {"detail", "code"} shape, or the RFC 6749 §5.2 {"error", "error_description"} shape for an :class:~jafaal.exceptions.OAuthError (see :func:_body), with the error's status code and header hints (e.g. WWW-Authenticate, Retry-After). When exc.clear_refresh_cookie is set (stale refresh token) the refresh-cookie deletion headers are added to the response.

Token-endpoint errors also carry Cache-Control: no-store per RFC 6749 §5.1, so an intermediary never caches an authorization failure.

Source code in jafaal/error_handler.py

async def jafaal_exception_handler(request: Request, exc: JafaalError) -> JSONResponse:
    """Translate a :class:`JafaalError` into a JSON HTTP response.

    Emits JAFAAL's ``{"detail", "code"}`` shape, or the RFC 6749 §5.2
    ``{"error", "error_description"}`` shape for an
    :class:`~jafaal.exceptions.OAuthError` (see :func:`_body`), with the error's
    status code and header hints (e.g. ``WWW-Authenticate``, ``Retry-After``).
    When ``exc.clear_refresh_cookie`` is set (stale refresh token) the
    refresh-cookie deletion headers are added to the response.

    Token-endpoint errors also carry ``Cache-Control: no-store`` per RFC 6749
    §5.1, so an intermediary never caches an authorization failure.
    """
    headers = dict(exc.headers) if exc.headers else {}
    if isinstance(exc, OAuthError):
        headers.setdefault("Cache-Control", "no-store")
        headers.setdefault("Pragma", "no-cache")
    response = JSONResponse(
        status_code=exc.status_code,
        content=_body(exc),
        headers=headers or None,
    )
    if getattr(exc, "clear_refresh_cookie", False):
        # Imported lazily: jafaal.utils pulls in the ORM layer, which is not
        # mapped until jafaal.map_models() runs.
        import jafaal.utils as jafaal_utils

        jafaal_utils.clear_refresh_token_cookies(response)
    return response

register_exception_handlers

register_exception_handlers(app)

Install JAFAAL's edge exception handler on the host FastAPI app.

Call once at startup, before serving requests. Safe to install up-front: it is a no-op until the core raises a :class:JafaalError.

Source code in jafaal/error_handler.py

def register_exception_handlers(app: FastAPI) -> None:
    """Install JAFAAL's edge exception handler on the host FastAPI app.

    Call once at startup, before serving requests. Safe to install up-front: it
    is a no-op until the core raises a :class:`JafaalError`.
    """
    # Starlette types handlers as ``Callable[[Request, Exception], ...]``; ours
    # narrows ``exc`` to ``JafaalError`` (safe — it is only dispatched for that
    # exception type), which Starlette's broad signature cannot express.
    app.add_exception_handler(JafaalError, jafaal_exception_handler)  # type: ignore[arg-type]

create_auth_router

create_auth_router(
    *,
    app=None,
    rate_limiter=None,
    prefixes=None,
    verify=True,
)

Build the aggregated JAFAAL auth router.

Parameters:

Name	Type	Description	Default
app	FastAPI | None	When provided, the :class:~jafaal.exceptions.JafaalError exception handler is registered on it. Omit it and call :func:~jafaal.error_handler.register_exception_handlers yourself if you assemble the app differently.	None
rate_limiter	RateLimiter | None	The host's rate limiter. Installed before the sub-routers are imported so their decorators resolve it. Defaults to the no-op limiter already in effect (no enforcement).	None
prefixes	RouterPrefixes | None	Override the default sub-prefixes (keep them in lockstep with :class:~jafaal.settings.AuthSettings path fields).	None
verify	bool	Run :func:verify_configuration first (default), so a missing host adapter fails here with one clear message instead of surfacing as a RuntimeError on the first request that needs it. Set False only when the router is built before the adapters are installed (then call :func:verify_configuration yourself once they are).	True

Returns:

Type	Description
APIRouter	An APIRouter the host mounts under its API root, e.g.:: app.include_router(create_auth_router(app=app), prefix="/api/v1")

Raises:

Type	Description
RuntimeError	If verify is set and a required component is missing, or a deployed environment is running on the in-memory state store or without an enforcing rate limiter.

Source code in jafaal/factory.py

def create_auth_router(
    *,
    app: FastAPI | None = None,
    rate_limiter: RateLimiter | None = None,
    prefixes: RouterPrefixes | None = None,
    verify: bool = True,
) -> APIRouter:
    """Build the aggregated JAFAAL auth router.

    Args:
        app: When provided, the :class:`~jafaal.exceptions.JafaalError` exception
            handler is registered on it. Omit it and call
            :func:`~jafaal.error_handler.register_exception_handlers` yourself if
            you assemble the app differently.
        rate_limiter: The host's rate limiter. Installed before the sub-routers
            are imported so their decorators resolve it. Defaults to the no-op
            limiter already in effect (no enforcement).
        prefixes: Override the default sub-prefixes (keep them in lockstep with
            :class:`~jafaal.settings.AuthSettings` path fields).
        verify: Run :func:`verify_configuration` first (default), so a missing
            host adapter fails here with one clear message instead of surfacing
            as a ``RuntimeError`` on the first request that needs it. Set
            ``False`` only when the router is built before the adapters are
            installed (then call :func:`verify_configuration` yourself once they
            are).

    Returns:
        An ``APIRouter`` the host mounts under its API root, e.g.::

            app.include_router(create_auth_router(app=app), prefix="/api/v1")

    Raises:
        RuntimeError: If ``verify`` is set and a required component is missing,
            or a deployed environment is running on the in-memory state store or
            without an enforcing rate limiter.
    """
    if rate_limiter is not None:
        configure_rate_limiter(rate_limiter)
    if app is not None:
        register_exception_handlers(app)
    prefixes = prefixes or RouterPrefixes()

    _warn_on_insecure_defaults()
    _warn_on_router_prefix_mismatch(prefixes)
    if verify:
        verify_configuration()
    else:
        # The deployment guards are not optional even when the adapter check is
        # deferred: they are what keeps a deployed environment from silently
        # running unprotected.
        _ensure_state_store_safe_for_deployment()
        _ensure_rate_limiter_safe_for_deployment()

    # Import the sub-routers here, after installing the limiter. The rate-limit
    # decorators bind the configured limiter lazily (on first request, re-binding
    # when it is reconfigured), so this import order is no longer load-bearing —
    # it is kept only for clarity.
    from jafaal.api_keys.router import router as api_keys_router
    from jafaal.identity_providers.public_router import router as idp_public_router
    from jafaal.identity_providers.router import router as idp_router
    from jafaal.jwks import router as jwks_router
    from jafaal.metadata import create_metadata_router
    from jafaal.password_reset_tokens.router import router as password_reset_router
    from jafaal.router import router as auth_router
    from jafaal.sessions.router import router as sessions_router
    from jafaal.sign_up_tokens.router import router as sign_up_router
    from jafaal.webauthn.router import public_router as webauthn_public_router
    from jafaal.webauthn.router import router as webauthn_router

    aggregate = APIRouter()
    aggregate.include_router(auth_router, prefix=prefixes.auth, tags=["auth"])
    aggregate.include_router(sessions_router, prefix=prefixes.sessions, tags=["sessions"])
    aggregate.include_router(api_keys_router, prefix=prefixes.api_keys, tags=["api_keys"])
    aggregate.include_router(idp_router, prefix=prefixes.identity_providers, tags=["identity_providers"])
    aggregate.include_router(
        idp_public_router,
        prefix=prefixes.identity_providers_public,
        tags=["identity_providers"],
    )
    aggregate.include_router(password_reset_router, prefix=prefixes.password_reset, tags=["password_reset"])
    aggregate.include_router(sign_up_router, prefix=prefixes.sign_up, tags=["sign_up"])
    aggregate.include_router(webauthn_router, prefix=prefixes.webauthn, tags=["webauthn"])
    aggregate.include_router(webauthn_public_router, prefix=prefixes.webauthn_public, tags=["webauthn"])
    # JWKS is mounted at the aggregate root (no sub-prefix) so it lands at
    # ``<api-root>/.well-known/jwks.json`` — the URL to advertise as the
    # ``jwks_uri`` for resource servers verifying asymmetric tokens. The RFC 8414
    # discovery document sits beside it and points at both.
    #
    # It stays on the aggregate router rather than moving to the RFC 8414 §3
    # issuer-derived path, because the endpoint URLs *inside* the document are
    # built from the mount, and the mount is only knowable from the request path
    # of a route that is itself mounted. An app-level absolute route would serve
    # the document from the right place with the wrong endpoints in it, which is
    # strictly worse. A host that needs the spec location mounts a second copy —
    # ``create_metadata_router(prefixes.auth, path=issuer_derived_metadata_path())``
    # — or serves ``get_authorization_server_metadata()`` there itself.
    aggregate.include_router(jwks_router)
    aggregate.include_router(create_metadata_router(prefixes.auth))
    return aggregate

shutdown async

shutdown(*, drain_events_timeout=5.0)

Release JAFAAL's process-wide resources. Call from the ASGI lifespan.

JAFAAL holds three things that outlive a request and are not garbage: the pooled HTTP client used for outbound OIDC calls (keep-alive sockets to every configured identity provider), the optional background maintenance thread, and any in-flight :class:~jafaal.ports.AuthEventSink deliveries. Without an explicit shutdown those leak on reload, keep a process alive longer than it should, and drop notifications that were mid-flight::

@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
    jafaal.maintenance.start_background_scheduler()
    yield
    await jafaal.shutdown()

Never raises: shutdown must not be able to fail a clean stop.

Parameters:

Name	Type	Description	Default
drain_events_timeout	float	Seconds to wait for pending event deliveries.	5.0

Source code in jafaal/factory.py

async def shutdown(*, drain_events_timeout: float = 5.0) -> None:
    """Release JAFAAL's process-wide resources. Call from the ASGI lifespan.

    JAFAAL holds three things that outlive a request and are not garbage: the
    pooled HTTP client used for outbound OIDC calls (keep-alive sockets to every
    configured identity provider), the optional background maintenance thread,
    and any in-flight :class:`~jafaal.ports.AuthEventSink` deliveries. Without an
    explicit shutdown those leak on reload, keep a process alive longer than it
    should, and drop notifications that were mid-flight::

        @contextlib.asynccontextmanager
        async def lifespan(app: FastAPI):
            jafaal.maintenance.start_background_scheduler()
            yield
            await jafaal.shutdown()

    Never raises: shutdown must not be able to fail a clean stop.

    Args:
        drain_events_timeout: Seconds to wait for pending event deliveries.
    """
    import jafaal.maintenance as jafaal_maintenance
    from jafaal.identity_providers.service import idp_service

    jafaal_maintenance.stop_background_scheduler()
    try:
        await idp_service.aclose()
    except Exception as err:  # pragma: no cover - defensive
        logger.warning(f"Error closing the identity-provider HTTP client: {type(err).__name__}", exc_info=err)
    if not jafaal_ports.wait_for_pending_events(drain_events_timeout):
        logger.warning(f"JAFAAL shut down with AuthEventSink deliveries still in flight after {drain_events_timeout}s")

verify_configuration

verify_configuration()

Assert that every required host-supplied component is installed.

JAFAAL resolves several host adapters lazily, so a missing one otherwise surfaces as a RuntimeError on the first request that needs it. Call this once at startup (e.g. in a FastAPI lifespan handler) to fail fast with a single, clear message listing everything that is missing.

Checks the components JAFAAL cannot default: the ORM model mapping (:func:jafaal.map_models), the settings object, the session factory, the user repository, and the settings provider. The event sink, state store, rate limiter, and scope catalog all have working defaults and so are not required here. Also enforces :func:_ensure_state_store_safe_for_deployment.

Raises:

Type	Description
RuntimeError	If any required component is missing (the message enumerates all of them), or if the in-memory state store is used in a deployed environment without the opt-out.

Source code in jafaal/factory.py

def verify_configuration() -> None:
    """Assert that every required host-supplied component is installed.

    JAFAAL resolves several host adapters lazily, so a missing one otherwise
    surfaces as a ``RuntimeError`` on the first request that needs it. Call this
    once at startup (e.g. in a FastAPI lifespan handler) to fail fast with a
    single, clear message listing everything that is missing.

    Checks the components JAFAAL cannot default: the ORM model mapping
    (:func:`jafaal.map_models`), the settings object, the session factory, the
    user repository, and the settings provider. The event sink, state store,
    rate limiter, and scope catalog all have working defaults and so are not
    required here. Also enforces
    :func:`_ensure_state_store_safe_for_deployment`.

    Raises:
        RuntimeError: If any required component is missing (the message
            enumerates all of them), or if the in-memory state store is used in a
            deployed environment without the opt-out.
    """
    missing: list[str] = []
    if not jafaal_orm.is_models_mapped():
        missing.append("ORM models — call jafaal.map_models(YourBase) after defining your Users model")
    if not jafaal_settings.is_configured():
        missing.append("AuthSettings — call jafaal.configure(AuthSettings(...))")
    if not jafaal_orm.is_sessionmaker_configured():
        missing.append("session factory — call jafaal.configure_sessionmaker(sessionmaker(bind=engine))")
    if not jafaal_ports.is_user_repository_configured():
        missing.append("UserRepository — call jafaal.configure_user_repository(...)")
    if not jafaal_ports.is_settings_provider_configured():
        missing.append("SettingsProvider — call jafaal.configure_settings_provider(...)")
    if missing:
        raise RuntimeError(
            "JAFAAL is not fully configured; the following required components are missing:\n"
            + "\n".join(f"  - {item}" for item in missing)
        )
    _ensure_state_store_safe_for_deployment()
    _ensure_rate_limiter_safe_for_deployment()

get_jwks

get_jwks()

Return the JWK Set of public keys that verify JAFAAL's JWTs.

Returns:

Type	Description
dict[str, Any]	A JWK Set dict ({"keys": [...]}). Empty in HS256 (symmetric) mode —
dict[str, Any]	callers serving this over HTTP should check
dict[str, Any]	attr:~jafaal.settings.TokenSettings.is_asymmetric first, as the
dict[str, Any]	packaged route does.

Raises:

Type	Description
RuntimeError	If JAFAAL has not been configured yet.

Source code in jafaal/jwks.py

def get_jwks() -> dict[str, Any]:
    """Return the JWK Set of public keys that verify JAFAAL's JWTs.

    Returns:
        A JWK Set dict (``{"keys": [...]}``). Empty in HS256 (symmetric) mode —
        callers serving this over HTTP should check
        :attr:`~jafaal.settings.TokenSettings.is_asymmetric` first, as the
        packaged route does.

    Raises:
        RuntimeError: If JAFAAL has not been configured yet.
    """
    return jafaal_token_manager.get_token_manager().jwks()

create_metadata_router

create_metadata_router(
    auth_prefix="/auth", *, path=METADATA_PATH
)

Build the router serving the RFC 8414 discovery document.

The core auth prefix is injected rather than imported because it is a deployment choice owned by :class:jafaal.RouterPrefixes; the advertised endpoint URLs must follow wherever the host actually mounted the router.

Parameters:

Name	Type	Description	Default
auth_prefix	str	Prefix the core auth router is mounted under.	'/auth'
path	str	Route path to expose the document at. Defaults to the aggregate root; :func:jafaal.create_auth_router passes the issuer-derived path when it registers the route on the host application.	METADATA_PATH

Returns:

Name	Type	Description
An	APIRouter	class:~fastapi.APIRouter exposing path.

Source code in jafaal/metadata.py

def create_metadata_router(auth_prefix: str = "/auth", *, path: str = METADATA_PATH) -> APIRouter:
    """Build the router serving the RFC 8414 discovery document.

    The core auth prefix is injected rather than imported because it is a
    deployment choice owned by :class:`jafaal.RouterPrefixes`; the advertised
    endpoint URLs must follow wherever the host actually mounted the router.

    Args:
        auth_prefix: Prefix the core auth router is mounted under.
        path: Route path to expose the document at. Defaults to the aggregate
            root; :func:`jafaal.create_auth_router` passes the issuer-derived
            path when it registers the route on the host application.

    Returns:
        An :class:`~fastapi.APIRouter` exposing ``path``.
    """
    router = APIRouter()

    @router.get(path, tags=["metadata"])
    def authorization_server_metadata(request: Request, response: Response) -> dict[str, Any]:
        """Serve this deployment's OAuth 2.0 authorization-server metadata."""
        response.headers["Cache-Control"] = f"public, max-age={_METADATA_CACHE_MAX_AGE_SECONDS}"
        return get_authorization_server_metadata(
            api_root=_resolve_api_root(request),
            auth_prefix=auth_prefix,
        )

    return router

get_authorization_server_metadata

get_authorization_server_metadata(
    *, api_root, auth_prefix="/auth"
)

Build the RFC 8414 metadata document for this deployment.

Parameters:

Name	Type	Description	Default
api_root	str	Absolute URL the aggregate auth router is mounted at, e.g. https://app.example/api/v1. Endpoint URLs are built from it.	required
auth_prefix	str	Prefix the core auth router is mounted under, i.e. :attr:jafaal.RouterPrefixes.auth.	'/auth'

Returns:

Type	Description
dict[str, Any]	The metadata document, ready to be serialised as JSON.

Raises:

Type	Description
RuntimeError	If JAFAAL has not been configured yet.

Source code in jafaal/metadata.py

def get_authorization_server_metadata(*, api_root: str, auth_prefix: str = "/auth") -> dict[str, Any]:
    """Build the RFC 8414 metadata document for this deployment.

    Args:
        api_root: Absolute URL the aggregate auth router is mounted at, e.g.
            ``https://app.example/api/v1``. Endpoint URLs are built from it.
        auth_prefix: Prefix the core auth router is mounted under, i.e.
            :attr:`jafaal.RouterPrefixes.auth`.

    Returns:
        The metadata document, ready to be serialised as JSON.

    Raises:
        RuntimeError: If JAFAAL has not been configured yet.
    """
    settings = jafaal_settings.get_settings()
    auth_root = _join_url(api_root, auth_prefix)
    metadata: dict[str, Any] = {
        "issuer": settings.resolved_issuer,
        "authorization_endpoint": _join_url(auth_root, "/authorize"),
        "token_endpoint": _join_url(auth_root, "/token"),
        "introspection_endpoint": _join_url(auth_root, "/introspect"),
        "revocation_endpoint": _join_url(auth_root, "/revoke"),
        "grant_types_supported": ["authorization_code", "refresh_token"],
        "response_types_supported": ["code"],
        "response_modes_supported": ["query"],
        # The catalog tiers, plus the introspection capability. AUTH_INTROSPECT
        # is deliberately outside the tiers (it is granted to a service API key,
        # never minted into a user's token), but a client reading this document
        # still has to be able to learn the scope it must obtain to call the
        # advertised introspection_endpoint.
        "scopes_supported": sorted(set(jafaal_scopes.get_scope_catalog().admin) | {jafaal_scopes.AUTH_INTROSPECT}),
        # First-party public clients (RFC 8252): the token endpoint
        # authenticates the *user* and binds the code with PKCE, never a client
        # credential.
        "token_endpoint_auth_methods_supported": ["none"],
        # ``introspection_endpoint_auth_methods_supported`` is deliberately
        # absent. RFC 8414 §2 draws its values from the IANA "OAuth Token
        # Endpoint Authentication Methods" registry — all of which describe
        # *client* authentication — and JAFAAL protects introspection with a
        # scoped access token instead (RFC 7662 §2.1's "separate OAuth 2.0
        # access token" model), which has no registered value. Emitting
        # ``"bearer"`` would be an unregistered token that a strict client may
        # reject, and ``"none"`` would claim the endpoint is unprotected. The
        # required scope is discoverable via ``scopes_supported`` instead.
        # RFC 6749 §3.2.1: a public client identifies itself with ``client_id``
        # and authenticates with nothing, which is the registry's ``none``.
        # ``/revoke`` requires that ``client_id`` and checks the token was issued
        # to it (RFC 7009 §5), but identification is not authentication and the
        # registry has no value for it.
        "revocation_endpoint_auth_methods_supported": ["none"],
        # PKCE is mandatory, and ``plain`` is refused.
        "code_challenge_methods_supported": ["S256"],
        # RFC 9207: every authorization response carries ``iss``, so a client
        # talking to more than one authorization server can tell which one
        # answered and refuse a code steered in from another (the mix-up
        # attack). Advertising it is what lets a client *require* the check.
        "authorization_response_iss_parameter_supported": True,
    }
    # ``jwks_uri`` only when there is something to publish. Under HS256 the key
    # set is empty, and RFC 8414 §2 describes ``jwks_uri`` as the location of
    # the issuer's signing keys — pointing a verifier at a document that
    # contains none tells it the keys were rotated away, not that stateless
    # verification was never on offer. Omitting the field says the latter.
    if settings.tokens.is_asymmetric:
        metadata["jwks_uri"] = _join_url(api_root, "/.well-known/jwks.json")
    return metadata

issuer_derived_metadata_path

issuer_derived_metadata_path()

Return the RFC 8414 §3 metadata path for the configured issuer.

§3 forms the URL by inserting /.well-known/oauth-authorization-server between the issuer's host and its path component — so an issuer of https://app.example/api/v1 publishes at /.well-known/oauth-authorization-server/api/v1. The naive <issuer>/.well-known/... is not the spec location whenever the issuer carries a path, which it does for every deployment mounted under an API prefix.

Returns:

Type	Description
str	The absolute path to register on the host application.

Source code in jafaal/metadata.py

def issuer_derived_metadata_path() -> str:
    """Return the RFC 8414 §3 metadata path for the configured issuer.

    §3 forms the URL by inserting ``/.well-known/oauth-authorization-server``
    between the issuer's host and its path component — so an issuer of
    ``https://app.example/api/v1`` publishes at
    ``/.well-known/oauth-authorization-server/api/v1``. The naive
    ``<issuer>/.well-known/...`` is *not* the spec location whenever the issuer
    carries a path, which it does for every deployment mounted under an API
    prefix.

    Returns:
        The absolute path to register on the host application.
    """
    issuer_path = urlparse(jafaal_settings.get_settings().resolved_issuer).path.rstrip("/")
    return f"{METADATA_PATH}{issuer_path}"

autonomous_session

autonomous_session()

A separate session that commits independently of the caller's transaction.

The deliberate exception to "the caller owns the transaction", for writes whose durability must not depend on the surrounding request succeeding. There is exactly one such case in JAFAAL today: claiming a single-use OAuth state. Replay protection has to stick — if the claim were rolled back when the callback later fails, an attacker could deliberately fail the flow to release the state and replay the authorization code.

Committing it separately also keeps the caller's transaction (and its pooled connection) from being held open across the several outbound HTTP calls the SSO callback then makes.

Yields:

Type	Description
Generator[Session]	A fresh session, committed on clean exit and rolled back on failure.

Source code in jafaal/orm.py

@contextmanager
def autonomous_session() -> Generator[Session]:
    """A separate session that commits independently of the caller's transaction.

    The deliberate exception to "the caller owns the transaction", for writes
    whose durability must not depend on the surrounding request succeeding.
    There is exactly one such case in JAFAAL today: claiming a single-use OAuth
    state. Replay protection has to *stick* — if the claim were rolled back when
    the callback later fails, an attacker could deliberately fail the flow to
    release the state and replay the authorization code.

    Committing it separately also keeps the caller's transaction (and its pooled
    connection) from being held open across the several outbound HTTP calls the
    SSO callback then makes.

    Yields:
        A fresh session, committed on clean exit and rolled back on failure.
    """
    db = get_sessionmaker()()
    try:
        with unit_of_work(db):
            yield db
    finally:
        db.close()

configure_sessionmaker

configure_sessionmaker(factory)

Install the host's session factory.

Call once at startup with a sessionmaker bound to the application's engine. Both the :func:get_db request dependency and background maintenance tasks obtain sessions from it.

Parameters:

Name	Type	Description	Default
factory	sessionmaker[Session]	A configured sessionmaker.	required

Source code in jafaal/orm.py

def configure_sessionmaker(factory: sessionmaker[Session]) -> None:
    """Install the host's session factory.

    Call once at startup with a ``sessionmaker`` bound to the application's
    engine. Both the :func:`get_db` request dependency and background
    maintenance tasks obtain sessions from it.

    Args:
        factory: A configured ``sessionmaker``.
    """
    _session_factory.configure(factory)

get_active_base

get_active_base()

Return the declarative base JAFAAL's models are mapped onto.

Model modules call this at import time to obtain their base, so importing a model module (or any CRUD/router that imports one) before :func:map_models is a configuration error.

Raises:

Type	Description
RuntimeError	If :func:map_models has not been called yet.

Source code in jafaal/orm.py

def get_active_base() -> type[DeclarativeBase]:
    """Return the declarative base JAFAAL's models are mapped onto.

    Model modules call this at import time to obtain their base, so importing a
    model module (or any CRUD/router that imports one) before :func:`map_models`
    is a configuration error.

    Raises:
        RuntimeError: If :func:`map_models` has not been called yet.
    """
    if _active_base is None:
        raise RuntimeError(
            "JAFAAL's models are not mapped yet. Call "
            "jafaal.map_models(YourBase, user_model=YourUserClass) once at startup — after "
            "defining your user model and before create_auth_router() or any DB use. "
            "Omit the base to use jafaal.orm.Base."
        )
    return _active_base

is_models_mapped

is_models_mapped()

Return whether :func:map_models has been called.

Source code in jafaal/orm.py

def is_models_mapped() -> bool:
    """Return whether :func:`map_models` has been called."""
    return _active_base is not None

map_models

map_models(base=None, *, user_model=None)

Define and map JAFAAL's companion tables into base's registry.

Call once at startup, after defining your user model and before :func:jafaal.create_auth_router or any database use. JAFAAL's models are mapped onto the base you pass, so your user model and JAFAAL's tables share one registry — which is what resolves the users.id foreign keys.

Parameters:

Name	Type	Description	Default
base	type[DeclarativeBase] | None	The host's :class:~sqlalchemy.orm.DeclarativeBase subclass; your user model must be built on it. Omit it to use JAFAAL's own :data:Base (the convenience default).	None
user_model	type | None	The host's user class. Passing it explicitly is what lets the class be called anything — Account, Member, Person. Omitted, JAFAAL falls back to whichever class is mapped to the users table, which is unambiguous but silently constrains the host's schema; pass it.	None

Raises:

Type	Description
RuntimeError	If called again with a different base or user model, or if a model references a class that is not mapped on base.

Source code in jafaal/orm.py

def map_models(base: type[DeclarativeBase] | None = None, *, user_model: type | None = None) -> None:
    """Define and map JAFAAL's companion tables into ``base``'s registry.

    Call once at startup, **after** defining your user model and **before**
    :func:`jafaal.create_auth_router` or any database use. JAFAAL's models are
    mapped onto the base you pass, so your user model and JAFAAL's tables share
    one registry — which is what resolves the ``users.id`` foreign keys.

    Args:
        base: The host's :class:`~sqlalchemy.orm.DeclarativeBase` subclass; your
            user model must be built on it. Omit it to use JAFAAL's own
            :data:`Base` (the convenience default).
        user_model: The host's user class. Passing it explicitly is what lets the
            class be called anything — ``Account``, ``Member``, ``Person``.
            Omitted, JAFAAL falls back to whichever class is mapped to the
            ``users`` table, which is unambiguous but silently constrains the
            host's schema; pass it.

    Raises:
        RuntimeError: If called again with a different base or user model, or if
            a model references a class that is not mapped on ``base``.
    """
    global _active_base
    target = base if base is not None else Base
    if _active_base is not None:
        if _active_base is not target:
            raise RuntimeError("jafaal.map_models() was already called with a different base; call it once at startup.")
        if user_model is not None and _user_model.is_configured() and _user_model.get() is not user_model:
            raise RuntimeError(
                "jafaal.map_models() was already called with a different user_model; call it once at startup."
            )
        return
    _active_base = target
    if user_model is not None:
        _user_model.configure(user_model)
    try:
        for module_name in _MODEL_MODULES:
            importlib.import_module(module_name)
        # Resolve every mapper/relationship now so misconfiguration (e.g. no
        # user model mapped on this base) fails fast at startup, not on first
        # query.
        target.registry.configure()
    except Exception:
        _active_base = None  # let the host fix the problem and retry
        _user_model.reset()
        raise

savepoint

savepoint(db)

Run a block inside a SAVEPOINT so a failed flush stays recoverable.

A statement that fails mid-transaction (a UNIQUE violation on flush, say) leaves SQLAlchemy's transaction in a pending-rollback state: every later statement on that session raises until someone unwinds it. Because JAFAAL runs inside the caller's transaction it must not unwind the whole thing — that would discard the host's pending work too — so a CRUD helper that wants to catch a constraint violation and translate it (e.g. into a 409) brackets the flush in a savepoint and rolls back only that.

Delegates to Session.begin_nested() used as a context manager, which is the only form that works: unwinding the savepoint by hand (nested.rollback() in an except) leaves the parent transaction holding the captured flush exception, so the caller's later commit() still raises PendingRollbackError. SQLAlchemy's own __exit__ clears it. tests/test_transactions.py pins this behaviour.

Parameters:

Name	Type	Description	Default
db	Session	The active session.	required

Yields:

Type	Description
Generator[Session]	The same session, for convenience.

Source code in jafaal/orm.py

@contextmanager
def savepoint(db: Session) -> Generator[Session]:
    """Run a block inside a SAVEPOINT so a failed flush stays recoverable.

    A statement that fails mid-transaction (a UNIQUE violation on flush, say)
    leaves SQLAlchemy's transaction in a pending-rollback state: every later
    statement on that session raises until someone unwinds it. Because JAFAAL
    runs inside the *caller's* transaction it must not unwind the whole thing —
    that would discard the host's pending work too — so a CRUD helper that wants
    to *catch* a constraint violation and translate it (e.g. into a 409) brackets
    the flush in a savepoint and rolls back only that.

    Delegates to ``Session.begin_nested()`` used as a context manager, which is
    the only form that works: unwinding the savepoint by hand
    (``nested.rollback()`` in an ``except``) leaves the *parent* transaction
    holding the captured flush exception, so the caller's later ``commit()``
    still raises ``PendingRollbackError``. SQLAlchemy's own ``__exit__`` clears
    it. ``tests/test_transactions.py`` pins this behaviour.

    Args:
        db: The active session.

    Yields:
        The same session, for convenience.
    """
    with db.begin_nested():
        yield db

session_scope

session_scope()

Context manager yielding a session for non-request (background) work.

Used by the maintenance tasks. The caller is responsible for committing (typically via :func:unit_of_work); the session is closed on exit, rolling back any pending transaction.

Source code in jafaal/orm.py

@contextmanager
def session_scope() -> Generator[Session]:
    """Context manager yielding a session for non-request (background) work.

    Used by the maintenance tasks. The caller is responsible for committing
    (typically via :func:`unit_of_work`); the session is closed on exit, rolling
    back any pending transaction.
    """
    db = get_sessionmaker()()
    try:
        yield db
    finally:
        db.close()

unit_of_work

unit_of_work(db)

Commit db on clean exit, roll it back on failure.

The supported way for a host to compose JAFAAL calls with its own writes::

with jafaal.unit_of_work(db):
    user = repo.create_local_user(...)
    identity_service.set_local_password_hash(user.id, hashed)
    db.add(MyProfile(user_id=user.id))
# one commit; any failure rolls back all three

Re-entrant: when a unit of work is already open on this session, the inner block joins it and the outermost scope decides the outcome. That is what makes "JAFAAL never commits under you" hold even when a JAFAAL service calls another internally.

Parameters:

Name	Type	Description	Default
db	Session	The session to own for the duration of the block.	required

Yields:

Type	Description
Generator[Session]	The same session, for convenience.

Raises:

Type	Description
Exception	Whatever the wrapped block raised, after rolling back.

Source code in jafaal/orm.py

@contextmanager
def unit_of_work(db: Session) -> Generator[Session]:
    """Commit ``db`` on clean exit, roll it back on failure.

    The supported way for a **host** to compose JAFAAL calls with its own writes::

        with jafaal.unit_of_work(db):
            user = repo.create_local_user(...)
            identity_service.set_local_password_hash(user.id, hashed)
            db.add(MyProfile(user_id=user.id))
        # one commit; any failure rolls back all three

    Re-entrant: when a unit of work is already open on this session, the inner
    block joins it and the *outermost* scope decides the outcome. That is what
    makes "JAFAAL never commits under you" hold even when a JAFAAL service calls
    another internally.

    Args:
        db: The session to own for the duration of the block.

    Yields:
        The same session, for convenience.

    Raises:
        Exception: Whatever the wrapped block raised, after rolling back.
    """
    if db.info.get(_UOW_FLAG):
        yield db
        return
    db.info[_UOW_FLAG] = True
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.info.pop(_UOW_FLAG, None)

configure_event_sink

configure_event_sink(sink)

Install the host's :class:AuthEventSink (defaults to a no-op sink).

Source code in jafaal/ports.py

def configure_event_sink(sink: AuthEventSink) -> None:
    """Install the host's :class:`AuthEventSink` (defaults to a no-op sink)."""
    _event_sink.configure(sink)

configure_password_breach_checker

configure_password_breach_checker(checker)

Install the host's :class:PasswordBreachChecker (defaults to a no-op).

Source code in jafaal/ports.py

def configure_password_breach_checker(checker: PasswordBreachChecker) -> None:
    """Install the host's :class:`PasswordBreachChecker` (defaults to a no-op)."""
    _password_breach_checker.configure(checker)

configure_scope_resolver

configure_scope_resolver(resolver)

Install the host's :class:ScopeResolver.

Call once at startup, before tokens are issued. Defaults to :class:TieredScopeResolver (the is_superuser two-tier mapping).

Parameters:

Name	Type	Description	Default
resolver	ScopeResolver	The host's scope-resolution adapter.	required

Source code in jafaal/ports.py

def configure_scope_resolver(resolver: ScopeResolver) -> None:
    """Install the host's :class:`ScopeResolver`.

    Call once at startup, before tokens are issued. Defaults to
    :class:`TieredScopeResolver` (the ``is_superuser`` two-tier mapping).

    Args:
        resolver: The host's scope-resolution adapter.
    """
    _scope_resolver.configure(resolver)

configure_settings_provider

configure_settings_provider(provider)

Install the host's :class:SettingsProvider. Call once at startup.

Source code in jafaal/ports.py

def configure_settings_provider(provider: SettingsProvider) -> None:
    """Install the host's :class:`SettingsProvider`. Call once at startup."""
    _settings_provider.configure(provider)

configure_user_repository

configure_user_repository(repository)

Install the host's :class:UserRepository. Call once at startup.

Source code in jafaal/ports.py

def configure_user_repository(repository: UserRepository) -> None:
    """Install the host's :class:`UserRepository`. Call once at startup."""
    _user_repository.configure(repository)

get_event_sink

get_event_sink()

Return the installed :class:AuthEventSink (NullAuthEventSink by default).

Source code in jafaal/ports.py

def get_event_sink() -> AuthEventSink:
    """Return the installed :class:`AuthEventSink` (``NullAuthEventSink`` by default)."""
    return _event_sink.get()

get_password_breach_checker

get_password_breach_checker()

Return the installed :class:PasswordBreachChecker (no-op by default).

Source code in jafaal/ports.py

def get_password_breach_checker() -> PasswordBreachChecker:
    """Return the installed :class:`PasswordBreachChecker` (no-op by default)."""
    return _password_breach_checker.get()

get_scope_resolver

get_scope_resolver()

Return the installed :class:ScopeResolver (:class:TieredScopeResolver by default).

Source code in jafaal/ports.py

def get_scope_resolver() -> ScopeResolver:
    """Return the installed :class:`ScopeResolver` (:class:`TieredScopeResolver` by default)."""
    return _scope_resolver.get()

get_settings_provider

get_settings_provider()

Return the installed :class:SettingsProvider.

Raises:

Type	Description
RuntimeError	If none has been configured.

Source code in jafaal/ports.py

def get_settings_provider() -> SettingsProvider:
    """Return the installed :class:`SettingsProvider`.

    Raises:
        RuntimeError: If none has been configured.
    """
    return _settings_provider.get()

get_user_repository

get_user_repository()

Return the installed :class:UserRepository.

Raises:

Type	Description
RuntimeError	If none has been configured.

Source code in jafaal/ports.py

def get_user_repository() -> UserRepository:
    """Return the installed :class:`UserRepository`.

    Raises:
        RuntimeError: If none has been configured.
    """
    return _user_repository.get()

reset_ports

reset_ports()

Clear all installed adapters. Intended for test isolation.

Source code in jafaal/ports.py

def reset_ports() -> None:
    """Clear all installed adapters. Intended for test isolation."""
    _user_repository.reset()
    _settings_provider.reset()
    _event_sink.reset()
    _password_breach_checker.reset()
    _scope_resolver.reset()

configure_rate_limiter

configure_rate_limiter(limiter)

Install the host-provided rate limiter.

Call this before the routers are imported (create_auth_router() does so automatically) so the endpoint decorators resolve the configured limiter.

Parameters:

Name	Type	Description	Default
limiter	RateLimiter	A :class:RateLimiter implementation.	required

Source code in jafaal/rate_limit.py

def configure_rate_limiter(limiter: RateLimiter) -> None:
    """Install the host-provided rate limiter.

    Call this before the routers are imported (``create_auth_router()`` does so
    automatically) so the endpoint decorators resolve the configured limiter.

    Args:
        limiter: A :class:`RateLimiter` implementation.
    """
    _rate_limiter.configure(limiter)

get_rate_limiter

get_rate_limiter()

Return the configured rate limiter (the no-op default until configured).

Source code in jafaal/rate_limit.py

def get_rate_limiter() -> RateLimiter:
    """Return the configured rate limiter (the no-op default until configured)."""
    return _rate_limiter.get()

reset_rate_limiter

reset_rate_limiter()

Reset to the no-op limiter. Intended for tests.

:func:limit binds the configured limiter lazily and watches the limiter slot's generation counter, so resetting (or reconfiguring) the limiter re-binds every decorated route on its next request — no import-order juggling required.

Source code in jafaal/rate_limit.py

def reset_rate_limiter() -> None:
    """Reset to the no-op limiter. Intended for tests.

    :func:`limit` binds the configured limiter lazily and watches the limiter
    slot's generation counter, so resetting (or reconfiguring) the limiter
    re-binds every decorated route on its next request — no import-order
    juggling required.
    """
    _rate_limiter.reset()

configure_scopes

configure_scopes(catalog)

Install the host's scope catalog (JAFAAL's scopes extended with app scopes).

Parameters:

Name	Type	Description	Default
catalog	ScopeCatalog	The full catalog, typically DEFAULT_SCOPE_CATALOG.extend(...).	required

Raises:

Type	Description
ValueError	If the catalog is inconsistent (see :meth:ScopeCatalog.validate).

Source code in jafaal/scopes.py

def configure_scopes(catalog: ScopeCatalog) -> None:
    """Install the host's scope catalog (JAFAAL's scopes extended with app scopes).

    Args:
        catalog: The full catalog, typically ``DEFAULT_SCOPE_CATALOG.extend(...)``.

    Raises:
        ValueError: If the catalog is inconsistent (see :meth:`ScopeCatalog.validate`).
    """
    catalog.validate()
    _scope_catalog.configure(catalog)

get_scope_catalog

get_scope_catalog()

Return the configured scope catalog (JAFAAL's own until configured).

Source code in jafaal/scopes.py

def get_scope_catalog() -> ScopeCatalog:
    """Return the configured scope catalog (JAFAAL's own until configured)."""
    return _scope_catalog.get()

reset_scopes

reset_scopes()

Reset to JAFAAL's own catalog. Intended for tests.

Source code in jafaal/scopes.py

def reset_scopes() -> None:
    """Reset to JAFAAL's own catalog. Intended for tests."""
    _scope_catalog.reset()

configure

configure(settings)

Install the host-supplied :class:AuthSettings for the process.

Call once at application startup, before serving requests. Re-calling replaces the settings and invalidates any settings-derived caches.

Parameters:

Name	Type	Description	Default
settings	AuthSettings	The fully-built, validated settings instance.	required

Source code in jafaal/settings.py

def configure(settings: AuthSettings) -> None:
    """Install the host-supplied :class:`AuthSettings` for the process.

    Call once at application startup, before serving requests. Re-calling
    replaces the settings and invalidates any settings-derived caches.

    Args:
        settings: The fully-built, validated settings instance.
    """
    if not isinstance(settings, AuthSettings):
        raise TypeError(f"expected AuthSettings, got {type(settings).__name__}")
    _slot.configure(settings)

get_settings

get_settings()

Return the installed :class:AuthSettings.

Raises:

Type	Description
RuntimeError	If :func:configure has not been called.

Source code in jafaal/settings.py

def get_settings() -> AuthSettings:
    """Return the installed :class:`AuthSettings`.

    Raises:
        RuntimeError: If :func:`configure` has not been called.
    """
    return _slot.get()

reset

reset()

Clear the installed settings. Intended for test isolation.

Source code in jafaal/settings.py

def reset() -> None:
    """Clear the installed settings. Intended for test isolation."""
    _slot.reset()

configure_state_store

configure_state_store(store)

Install the host-provided state store (e.g. a Redis-backed adapter).

Parameters:

Name	Type	Description	Default
store	StateStore	A :class:StateStore implementation.	required

Source code in jafaal/state_store.py

def configure_state_store(store: StateStore) -> None:
    """Install the host-provided state store (e.g. a Redis-backed adapter).

    Args:
        store: A :class:`StateStore` implementation.
    """
    _state_store.configure(store)

get_state_store

get_state_store()

Return the configured state store (the in-memory default until configured).

Source code in jafaal/state_store.py

def get_state_store() -> StateStore:
    """Return the configured state store (the in-memory default until configured)."""
    return _state_store.get()

reset_state_store

reset_state_store()

Reset to a fresh in-memory store. Intended for tests.

Source code in jafaal/state_store.py

def reset_state_store() -> None:
    """Reset to a fresh in-memory store. Intended for tests."""
    _state_store.reset()

get_sid_from_access_token

get_sid_from_access_token(
    request, access_token, identity_service
)

Retrieve the session ID from the access token.

Resolves and caches the :class:~jafaal.principal.Principal on request.state then extracts the session ID from the :class:~jafaal.principal.AccessTokenCred.

Parameters:

Name	Type	Description	Default
request	Request	Current HTTP request for state caching.	required
access_token	Annotated[str, Depends(get_access_token)]	JWT from the Authorization header.	required
identity_service	Annotated[IdentityService, Depends(get_identity_service)]	Per-request IdentityService.	required

Returns:

Name	Type	Description
str	str	Session ID (sid claim) from the token.

Raises:

Type	Description
JafaalError	401 if the token is invalid, expired, or the credential type is unexpected.

Source code in jafaal/_internal/internal_dependencies.py

def get_sid_from_access_token(
    request: Request,
    access_token: Annotated[str, Depends(get_access_token)],
    identity_service: Annotated[
        jafaal_identity_service.IdentityService,
        Depends(jafaal_identity_service.get_identity_service),
    ],
) -> str:
    """Retrieve the session ID from the access token.

    Resolves and caches the :class:`~jafaal.principal.Principal`
    on ``request.state`` then extracts the session ID from the
    :class:`~jafaal.principal.AccessTokenCred`.

    Args:
        request: Current HTTP request for state caching.
        access_token: JWT from the Authorization header.
        identity_service: Per-request IdentityService.

    Returns:
        str: Session ID (``sid`` claim) from the token.

    Raises:
        JafaalError: 401 if the token is invalid,
            expired, or the credential type is unexpected.
    """
    principal = _resolve_and_cache_principal(access_token, request, identity_service)
    cred = principal.credential
    if not isinstance(cred, AccessTokenCred):
        raise jafaal_exceptions.InvalidTokenError("Invalid credential type for session ID")
    return cred.session_id

get_sid_from_refresh_token

get_sid_from_refresh_token(validated)

Retrieves the session ID ('sid') from a validated refresh token.

Parameters:

Name	Type	Description	Default
validated	Annotated[ValidatedRefreshToken, Depends(get_validated_refresh_token)]	The validated refresh token and its claims.	required

Returns:

Type	Description
str	The session ID associated with the provided refresh token.

Source code in jafaal/_internal/internal_dependencies.py

def get_sid_from_refresh_token(
    validated: Annotated[ValidatedRefreshToken, Depends(get_validated_refresh_token)],
) -> str:
    """
    Retrieves the session ID ('sid') from a validated refresh token.

    Args:
        validated: The validated refresh token and its claims.

    Returns:
        The session ID associated with the provided refresh token.
    """
    return validated.session_id

get_sub_from_access_token

get_sub_from_access_token(
    request, access_token, identity_service
)

Retrieve the user ID from the access token.

Resolves and caches the :class:~jafaal.principal.Principal on request.state then returns principal.user_id. Subsequent calls within the same request hit the cache instead of issuing another DB lookup.

Parameters:

Name	Type	Description	Default
request	Request	Current HTTP request for state caching.	required
access_token	Annotated[str, Depends(get_access_token)]	JWT from the Authorization header.	required
identity_service	Annotated[IdentityService, Depends(get_identity_service)]	Per-request IdentityService.	required

Returns:

Name	Type	Description
int	UserId	Authenticated user's primary key.

Raises:

Type	Description
JafaalError	401 if the token is invalid, expired, or the user is not found.

Source code in jafaal/_internal/internal_dependencies.py

def get_sub_from_access_token(
    request: Request,
    access_token: Annotated[str, Depends(get_access_token)],
    identity_service: Annotated[
        jafaal_identity_service.IdentityService,
        Depends(jafaal_identity_service.get_identity_service),
    ],
) -> jafaal_orm.UserId:
    """Retrieve the user ID from the access token.

    Resolves and caches the :class:`~jafaal.principal.Principal`
    on ``request.state`` then returns ``principal.user_id``.
    Subsequent calls within the same request hit the cache
    instead of issuing another DB lookup.

    Args:
        request: Current HTTP request for state caching.
        access_token: JWT from the Authorization header.
        identity_service: Per-request IdentityService.

    Returns:
        int: Authenticated user's primary key.

    Raises:
        JafaalError: 401 if the token is invalid,
            expired, or the user is not found.
    """
    principal = _resolve_and_cache_principal(access_token, request, identity_service)
    return principal.user_id

get_sub_from_refresh_token

get_sub_from_refresh_token(validated)

Retrieves the user ID ('sub' claim) from a validated refresh token.

Parameters:

Name	Type	Description	Default
validated	Annotated[ValidatedRefreshToken, Depends(get_validated_refresh_token)]	The validated refresh token and its claims.	required

Returns:

Type	Description
UserId	The user ID associated with the provided refresh token.

Source code in jafaal/_internal/internal_dependencies.py

def get_sub_from_refresh_token(
    validated: Annotated[ValidatedRefreshToken, Depends(get_validated_refresh_token)],
) -> jafaal_orm.UserId:
    """
    Retrieves the user ID ('sub' claim) from a validated refresh token.

    Args:
        validated: The validated refresh token and its claims.

    Returns:
        The user ID associated with the provided refresh token.
    """
    return validated.user_id

validate_access_token_or_api_key async

validate_access_token_or_api_key(
    request,
    identity_service,
    access_token=Depends(oauth2_scheme),
    api_key_header=Depends(header_api_key_scheme),
    api_key_query=Query(None, alias="api_key"),
)

Accept either a JWT bearer token or an API key.

API keys should be supplied via the X-API-Key request header. Query-string delivery (?api_key=) is disabled by default because credentials in query strings appear in access logs, proxy histories, and browser history. It can be enabled via the allow_api_key_query_param setting on :class:~jafaal.settings.AuthSettings for self-hosted deployments that require it (e.g. webhook integrations that cannot set custom headers).

Tries JWT first (Authorization: Bearer header). If none is present, falls back to the X-API-Key header, then the ?api_key= query parameter if allowed. Raises 401 if none is supplied.

Delegates to :class:~jafaal.identity_service.IdentityService for credential resolution and caches the resolved :class:~jafaal.principal.Principal on request.state.principal so that other dependencies in the same request can share the result without a second DB lookup.

Parameters:

Name	Type	Description	Default
request	Request	The current HTTP request.	required
identity_service	Annotated[IdentityService, Depends(get_identity_service)]	Per-request IdentityService.	required
access_token	str | None	Optional Bearer token from the Authorization header.	Depends(oauth2_scheme)
api_key_header	str | None	Optional API key from the X-API-Key header.	Depends(header_api_key_scheme)
api_key_query	str | None	Optional API key from the ?api_key= query parameter (only honoured when AuthSettings.allow_api_key_query_param is True).	Query(None, alias='api_key')

Returns:

Type	Description
AuthContext	AuthContext with resolved user_id, scopes, and
AuthContext	auth_type ("jwt" or "api_key").

Raises:

Type	Description
JafaalError	401 if no valid credential is provided.

Source code in jafaal/_internal/internal_dependencies.py

async def validate_access_token_or_api_key(
    request: Request,
    identity_service: Annotated[
        jafaal_identity_service.IdentityService,
        Depends(jafaal_identity_service.get_identity_service),
    ],
    access_token: str | None = Depends(oauth2_scheme),
    api_key_header: str | None = Depends(header_api_key_scheme),
    api_key_query: str | None = Query(None, alias="api_key"),
) -> "AuthContext":
    """Accept either a JWT bearer token or an API key.

    API keys should be supplied via the ``X-API-Key`` request header.
    Query-string delivery (``?api_key=``) is disabled by default because
    credentials in query strings appear in access logs, proxy histories,
    and browser history. It can be enabled via the
    ``allow_api_key_query_param`` setting on
    :class:`~jafaal.settings.AuthSettings` for self-hosted deployments
    that require it (e.g. webhook integrations that cannot set custom
    headers).

    Tries JWT first (Authorization: Bearer header). If none is
    present, falls back to the ``X-API-Key`` header, then the
    ``?api_key=`` query parameter if allowed. Raises 401 if none
    is supplied.

    Delegates to :class:`~jafaal.identity_service.IdentityService`
    for credential resolution and caches the resolved
    :class:`~jafaal.principal.Principal` on
    ``request.state.principal`` so that other dependencies in
    the same request can share the result without a second DB
    lookup.

    Args:
        request: The current HTTP request.
        identity_service: Per-request IdentityService.
        access_token: Optional Bearer token from the
            Authorization header.
        api_key_header: Optional API key from the
            ``X-API-Key`` header.
        api_key_query: Optional API key from the
            ``?api_key=`` query parameter (only honoured
            when ``AuthSettings.allow_api_key_query_param`` is ``True``).

    Returns:
        AuthContext with resolved user_id, scopes, and
        auth_type (``"jwt"`` or ``"api_key"``).

    Raises:
        JafaalError: 401 if no valid credential is
            provided.
    """
    # --- Cache check: return early if Principal already resolved ---
    cached: Principal | None = getattr(request.state, "principal", None)
    if cached is not None:
        auth_type = "api_key" if cached.is_api_key() else "jwt"
        return AuthContext(
            user_id=cached.user_id,
            scopes=list(cached.scopes),
            auth_type=auth_type,
        )

    # --- JWT path ---
    if access_token is not None:
        principal = identity_service.resolve_from_access_token(access_token)
        request.state.principal = principal
        return AuthContext(
            user_id=principal.user_id,
            scopes=list(principal.scopes),
            auth_type="jwt",
        )

    # --- API key path ---
    settings = jafaal_settings.get_settings()
    raw_key = api_key_header
    if raw_key is None and api_key_query is not None and settings.api_keys.allow_query_param:
        logger.warning(
            "API key supplied via query string (?api_key=). "
            "This is a security risk: credentials appear in access logs "
            "and browser history. Set X-API-Key header instead."
        )
        raw_key = api_key_query
    if raw_key is not None:
        principal = identity_service.resolve_from_api_key(raw_key, request)
        request.state.principal = principal
        return AuthContext(
            user_id=principal.user_id,
            scopes=list(principal.scopes),
            auth_type="api_key",
        )

    raise jafaal_exceptions.AuthenticationError("Not authenticated. Provide a Bearer token or an API key.")

configure_api_key_scopes

configure_api_key_scopes(scopes)

Install the scopes an API key is allowed to grant.

Call once at startup, before serving requests.

Parameters:

Name	Type	Description	Default
scopes	Iterable[str]	The scope strings API keys may carry.	required

Source code in jafaal/api_keys/utils.py

def configure_api_key_scopes(scopes: Iterable[str]) -> None:
    """Install the scopes an API key is allowed to grant.

    Call once at startup, before serving requests.

    Args:
        scopes: The scope strings API keys may carry.
    """
    _supported_api_key_scopes.configure(frozenset(scopes))

get_api_key_scopes

get_api_key_scopes()

Return the configured API-key scope allow-list (empty until configured).

Source code in jafaal/api_keys/utils.py

def get_api_key_scopes() -> frozenset[str]:
    """Return the configured API-key scope allow-list (empty until configured)."""
    return _supported_api_key_scopes.get()

reset_api_key_scopes

reset_api_key_scopes()

Reset the API-key scope allow-list to empty. Intended for tests.

Source code in jafaal/api_keys/utils.py

def reset_api_key_scopes() -> None:
    """Reset the API-key scope allow-list to empty. Intended for tests."""
    _supported_api_key_scopes.reset()

check_auth_scopes

check_auth_scopes(auth, security_scopes)

Validate scopes from a unified AuthContext.

Use this in place of :func:check_scopes on endpoints that accept both JWT and API key auth. The underlying AuthContext is resolved by :func:validate_access_token_or_api_key, which goes through IdentityService (asserting the user exists and is active).

Parameters:

Name	Type	Description	Default
auth	Annotated[AuthContext, Depends(validate_access_token_or_api_key)]	Resolved AuthContext from validate_access_token_or_api_key.	required
security_scopes	SecurityScopes	Required scopes for the endpoint.	required

Raises:

Type	Description
MissingScopeError	403 if any required scope is missing from the AuthContext.

Source code in jafaal/dependencies.py

def check_auth_scopes(
    auth: Annotated[
        AuthContext,
        Depends(validate_access_token_or_api_key),
    ],
    security_scopes: SecurityScopes,
) -> None:
    """Validate scopes from a unified AuthContext.

    Use this in place of :func:`check_scopes` on endpoints that accept both
    JWT and API key auth. The underlying ``AuthContext`` is resolved by
    :func:`validate_access_token_or_api_key`, which goes through
    ``IdentityService`` (asserting the user exists and is active).

    Args:
        auth: Resolved AuthContext from validate_access_token_or_api_key.
        security_scopes: Required scopes for the endpoint.

    Raises:
        MissingScopeError: 403 if any required scope is missing from the
            AuthContext.
    """
    missing = set(security_scopes.scopes) - set(auth.scopes)
    if missing:
        jafaal_audit.record(
            jafaal_audit.Event.SCOPE_DENIED,
            outcome=jafaal_audit.Outcome.BLOCKED,
            level=logging.WARNING,
            user_id=auth.user_id,
            auth_type=auth.auth_type,
            missing=sorted(missing),
            required=sorted(security_scopes.scopes),
        )
        raise jafaal_exceptions.MissingScopeError(
            f"Unauthorized Access - Missing permissions: {' '.join(sorted(missing))}",
            missing=missing,
            required=set(security_scopes.scopes),
        )

clear_password

clear_password(user_id, db)

Remove a user's local password credential, leaving the account SSO-only.

A no-op when the account already has no local credential. Carries the same "no step-up, no authorization check" warning as :func:set_password, and like it, participates in the caller's transaction without committing and revokes everything the removed password could still reach — leaving those alive would make the removal cosmetic.

Leaves the account able to authenticate only through a linked identity provider or a registered passkey. Clearing the password of an account with neither locks the user out.

Parameters:

Name	Type	Description	Default
user_id	UserId	The user whose credential should be removed.	required
db	Session	Active database session.	required

Source code in jafaal/identity_service.py

def clear_password(user_id: UserId, db: Session) -> None:
    """Remove a user's local password credential, leaving the account SSO-only.

    A no-op when the account already has no local credential. Carries the same
    "no step-up, no authorization check" warning as :func:`set_password`, and
    like it, participates in the caller's transaction without committing and
    revokes everything the removed password could still reach — leaving those
    alive would make the removal cosmetic.

    Leaves the account able to authenticate only through a linked identity
    provider or a registered passkey. Clearing the password of an account with
    neither locks the user out.

    Args:
        user_id: The user whose credential should be removed.
        db: Active database session.
    """
    DefaultIdentityService(
        db,
        jafaal_token_manager.get_token_manager(),
        jafaal_password_hasher.get_password_hasher(),
    ).clear_local_password(user_id)

    credential_sweep.revoke_derived_credentials(user_id, db, reason="password_cleared")

    jafaal_audit.record(
        jafaal_audit.Event.PASSWORD_CHANGED,
        level=logging.WARNING,
        user_id=user_id,
        actor="host",
        cleared=True,
    )

set_password

set_password(
    user_id,
    password,
    db,
    *,
    human_chosen=True,
    must_change=False,
)

Set or replace a user's local password, outside an HTTP request.

The out-of-band half of credential management: seeding the first administrator (which no HTTP flow can do — sign-up cannot grant is_superuser), a CLI reset-password, or a migration importing accounts. Ordinary self-service registration should go through /auth/sign-up, which additionally gets enumeration safety and the verification/approval gates.

.. warning:: This performs no step-up verification and no authorization check — it cannot, having no request to authenticate. JAFAAL's own routes reach the same credential store through step-up-gated endpoints; this is the equivalent for code that has already established the caller is allowed to do it. Never wire it to an HTTP handler without that gate.

Revokes everything the previous password could still reach — sessions, API keys, outstanding reset tokens, pending MFA tickets, step-up grants, live passkey-registration challenges — exactly as every other password path does. That is what makes an operator-driven reset evict an intruder rather than merely inconvenience them, so it is not optional. On an account created moments ago there is nothing to revoke and the sweep is a no-op.

Participates in the caller's transaction and does not commit — wrap it in :func:jafaal.unit_of_work (or your own session.begin()) so the credential lands atomically with whatever else you are writing::

with jafaal.unit_of_work(db):
    user = repo.create_local_user("admin", "admin@example.com", db,
                                  is_active=True, is_verified=True)
    jafaal.set_password(user.id, os.environ["ADMIN_PASSWORD"], db)

Parameters:

Name	Type	Description	Default
user_id	UserId	The user to write the credential for. The account's tier selects the admin or regular minimum length.	required
password	str	The plaintext password.	required
db	Session	Active database session.	required
human_chosen	bool	Whether a person picked this password (the default). Human passwords are held to the host's policy and screened against the installed breach corpus. Pass False only for a high-entropy secret your own code generated, where a composition policy is meaningless and a breach corpus cannot contain it — never to make a policy rejection go away for a password a person supplied.	True
must_change	bool	Require the account owner to replace this password before they can sign in. Recommended whenever you chose the password — seeding an administrator, or a support-desk reset — because such a password is known to whoever set it. Login then fails with :class:~jafaal.PasswordChangeRequiredError until a new password is written. You must have a way for the user to set one: JAFAAL ships no change-password endpoint, so wire one that calls this function, or leave the account's e-mail reachable so /auth/password-reset can serve as the escape hatch. Setting this with neither in place locks the account out.	False

Raises:

Type	Description
NotFoundError	If user_id does not resolve to a user.
PasswordPolicyError	If the password is human-chosen and fails the configured policy or appears in the installed breach corpus.

Source code in jafaal/identity_service.py

def set_password(
    user_id: UserId,
    password: str,
    db: Session,
    *,
    human_chosen: bool = True,
    must_change: bool = False,
) -> None:
    """Set or replace a user's local password, outside an HTTP request.

    The out-of-band half of credential management: seeding the first
    administrator (which no HTTP flow can do — sign-up cannot grant
    ``is_superuser``), a CLI ``reset-password``, or a migration importing
    accounts. Ordinary self-service registration should go through
    ``/auth/sign-up``, which additionally gets enumeration safety and the
    verification/approval gates.

    .. warning::
       This performs **no step-up verification and no authorization check** —
       it cannot, having no request to authenticate. JAFAAL's own routes reach
       the same credential store through step-up-gated endpoints; this is the
       equivalent for code that has *already* established the caller is allowed
       to do it. Never wire it to an HTTP handler without that gate.

    Revokes everything the previous password could still reach — sessions, API
    keys, outstanding reset tokens, pending MFA tickets, step-up grants, live
    passkey-registration challenges — exactly as every other password path does.
    That is what makes an operator-driven reset evict an intruder rather than
    merely inconvenience them, so it is not optional. On an account created
    moments ago there is nothing to revoke and the sweep is a no-op.

    Participates in the caller's transaction and **does not commit** — wrap it in
    :func:`jafaal.unit_of_work` (or your own ``session.begin()``) so the
    credential lands atomically with whatever else you are writing::

        with jafaal.unit_of_work(db):
            user = repo.create_local_user("admin", "admin@example.com", db,
                                          is_active=True, is_verified=True)
            jafaal.set_password(user.id, os.environ["ADMIN_PASSWORD"], db)

    Args:
        user_id: The user to write the credential for. The account's tier
            selects the admin or regular minimum length.
        password: The plaintext password.
        db: Active database session.
        human_chosen: Whether a person picked this password (the default). Human
            passwords are held to the host's policy and screened against the
            installed breach corpus. Pass ``False`` **only** for a high-entropy
            secret your own code generated, where a composition policy is
            meaningless and a breach corpus cannot contain it — never to make a
            policy rejection go away for a password a person supplied.
        must_change: Require the account owner to replace this password before
            they can sign in. Recommended whenever *you* chose the password —
            seeding an administrator, or a support-desk reset — because such a
            password is known to whoever set it. Login then fails with
            :class:`~jafaal.PasswordChangeRequiredError` until a new password is
            written. **You must have a way for the user to set one**: JAFAAL
            ships no change-password endpoint, so wire one that calls this
            function, or leave the account's e-mail reachable so
            ``/auth/password-reset`` can serve as the escape hatch. Setting this
            with neither in place locks the account out.

    Raises:
        NotFoundError: If ``user_id`` does not resolve to a user.
        PasswordPolicyError: If the password is human-chosen and fails the
            configured policy or appears in the installed breach corpus.
    """
    # Imported here rather than at module scope: password_policy imports this
    # module for typing, so a top-level import would be circular at runtime.
    import jafaal.password_policy as jafaal_password_policy

    user = jafaal_ports.get_user_repository().get_by_id(user_id, db)
    if user is None:
        raise jafaal_exceptions.NotFoundError("User not found")

    service = DefaultIdentityService(
        db,
        jafaal_token_manager.get_token_manager(),
        jafaal_password_hasher.get_password_hasher(),
    )
    if human_chosen:
        password_hash = jafaal_password_policy.validate_and_hash_for_user(
            service,
            jafaal_ports.is_superuser(user),
            password,
        )
    else:
        password_hash = service.hash_password(password)
    service.set_local_password_hash(user_id, password_hash)
    if must_change:
        jafaal_credentials_crud.upsert_password_hash(user_id, password_hash, db, must_change=True)

    credential_sweep.revoke_derived_credentials(user_id, db, reason="password_set")

    jafaal_audit.record(
        jafaal_audit.Event.PASSWORD_CHANGED,
        level=logging.WARNING,
        user_id=user_id,
        actor="host",
    )

authenticate_user

authenticate_user(username, password, password_hasher, db)

Authenticates a user by verifying the provided username and password.

Parameters:

Name	Type	Description	Default
username	str	The username of the user attempting to authenticate.	required
password	str	The plaintext password provided by the user.	required
password_hasher	PasswordHasher	An instance of the password hasher for verifying and updating password hashes.	required
db	Session	The database session used for querying and updating user data.	required

Returns:

Type	Description
UserProtocol	jafaal_ports.UserProtocol: The authenticated user object if authentication is successful.

Raises:

Type	Description
JafaalError	If the username does not exist, the password is invalid, or the password exceeds AuthSettings.password_max_length.

Source code in jafaal/utils.py

def authenticate_user(
    username: str,
    password: str,
    password_hasher: jafaal_password_hasher.PasswordHasher,
    db: Session,
) -> jafaal_ports.UserProtocol:
    """
    Authenticates a user by verifying the provided username and password.

    Args:
        username (str): The username of the user attempting to authenticate.
        password (str): The plaintext password provided by the user.
        password_hasher (jafaal_password_hasher.PasswordHasher): An instance of the password hasher for verifying and updating password hashes.
        db (Session): The database session used for querying and updating user data.

    Returns:
        jafaal_ports.UserProtocol: The authenticated user object if authentication is successful.

    Raises:
        JafaalError: If the username does not exist, the password is invalid, or
            the password exceeds ``AuthSettings.password_max_length``.
    """
    # Get the user from the database
    user = jafaal_ports.get_user_repository().get_by_username(username, db)

    # Bound the input before any (deliberately slow) hashing work. Argon2 is
    # tuned to hundreds of milliseconds and hashes the whole input, so an
    # unauthenticated caller could otherwise post a multi-megabyte "password" on
    # every request. Checked before the user lookup result is used so the
    # rejection costs the same whether or not the account exists.
    if len(password) > jafaal_settings.get_settings().passwords.max_length:
        raise jafaal_exceptions.InvalidCredentialsError("Unable to authenticate with provided credentials")

    # Check if the user exists and if the password is correct
    if not user:
        # Run a dummy Argon2 verify so that the wall-clock latency of
        # the "user not found" branch matches the "user found, wrong
        # password" branch. Without this, Argon2's deliberately-tuned
        # ~hundreds-of-milliseconds verify time is trivially observable
        # from the network and lets an attacker enumerate valid
        # usernames without ever tripping FailedLoginAttempts (lockout
        # is only recorded on 401, which the attacker does not care
        # about while probing existence).
        password_hasher.dummy_verify()
        raise jafaal_exceptions.InvalidCredentialsError("Unable to authenticate with provided credentials")

    # Load the user's local password hash from the auth-owned credential
    # table. A missing row means the account has no local password.
    credential = jafaal_credentials_crud.get_credential(user.id, db)

    # User has no local password (SSO-only account). Treat identically
    # to "wrong password" so neither the response body nor the timing
    # discloses the account's auth modality. The dummy verify keeps
    # the latency consistent with a normal Argon2 verify.
    if credential is None:
        password_hasher.dummy_verify()
        raise jafaal_exceptions.InvalidCredentialsError("Unable to authenticate with provided credentials")

    # Verify password and get updated hash if applicable
    is_password_valid, updated_hash = password_hasher.verify_and_update(password, credential.password_hash)
    if not is_password_valid:
        raise jafaal_exceptions.InvalidCredentialsError("Unable to authenticate with provided credentials")

    # Update user hash if applicable
    if updated_hash:
        jafaal_credentials_crud.upsert_password_hash(user.id, updated_hash, db)

    # An operator-set password is known to whoever set it, so it opens nothing
    # until the account owner replaces it. Checked after verification on
    # purpose: reaching here already required the correct password, so the
    # distinct error reveals nothing a failed attempt would not have.
    if credential.must_change_password:
        raise jafaal_exceptions.PasswordChangeRequiredError

    # Return the user if the password is correct
    return user

complete_login

complete_login(
    response,
    request,
    user,
    client,
    token_manager,
    db,
    requested_scope=None,
)

Mint a session and its token bundle for an authenticated user.

Shared by password login and the authorization-code exchange, so both flows produce byte-identical token semantics, auditing, and new-device detection.

Token delivery follows client.token_delivery. For a browser client (cookie) the response follows RFC 9700 §7.2 rather than RFC 6749 §5.1 literally: the access token is returned in the body for in-memory storage, while the refresh token is set as an HttpOnly, SameSite=Strict cookie instead of being handed to page script.

Parameters:

Name	Type	Description	Default
response	Response	The HTTP response object, used to set the refresh cookie.	required
request	Request	The HTTP request, for IP and device fingerprinting.	required
user	UserProtocol	The authenticated user.	required
client	OAuthClient	The registered client the tokens are issued to.	required
token_manager	TokenManager	Utility for token generation.	required
db	Session	Database session for storing session information.	required
requested_scope	Sequence[str] | None	The scope the client asked for, applied as a final narrowing bound (RFC 6749 §3.3).	None

Returns:

Name	Type	Description
dict	dict	The RFC 6749 §5.1 token response (see :func:build_token_response).

Source code in jafaal/utils.py

def complete_login(
    response: Response,
    request: Request,
    user: jafaal_ports.UserProtocol,
    client: jafaal_settings.OAuthClient,
    token_manager: jafaal_token_manager.TokenManager,
    db: Session,
    requested_scope: Sequence[str] | None = None,
) -> dict:
    """Mint a session and its token bundle for an authenticated user.

    Shared by password login and the authorization-code exchange, so both flows
    produce byte-identical token semantics, auditing, and new-device detection.

    Token delivery follows ``client.token_delivery``. For a browser client
    (``cookie``) the response follows RFC 9700 §7.2 rather than RFC 6749 §5.1
    literally: the access token is returned in the body for in-memory storage,
    while the refresh token is set as an ``HttpOnly``, ``SameSite=Strict``
    cookie instead of being handed to page script.

    Args:
        response: The HTTP response object, used to set the refresh cookie.
        request: The HTTP request, for IP and device fingerprinting.
        user: The authenticated user.
        client: The registered client the tokens are issued to.
        token_manager: Utility for token generation.
        db: Database session for storing session information.
        requested_scope: The ``scope`` the client asked for, applied as a final
            narrowing bound (RFC 6749 §3.3).

    Returns:
        dict: The RFC 6749 §5.1 token response (see :func:`build_token_response`).
    """
    # Create the tokens
    (
        session_id,
        access_token_exp,
        access_token,
        refresh_token_exp,
        refresh_token,
        csrf_token,
    ) = create_tokens(user, token_manager, client=client, requested_scope=requested_scope)

    # Decide whether this login is from a not-previously-seen device *before*
    # the new session is written. Only pay the lookup when a host sink actually
    # wants the event (the default null sink and older sinks skip it).
    sink = jafaal_ports.get_event_sink()
    emit_new_device = not isinstance(sink, jafaal_ports.NullAuthEventSink) and hasattr(sink, "on_new_device_login")
    known_device = True
    if emit_new_device:
        try:
            known_device = jafaal_sessions_utils.is_known_device(user.id, request, db)
        except Exception:
            # Never let new-device detection break login; treat as known.
            known_device = True

    # Create the session and store it in the database
    # Note: csrf_token is NOT stored on initial login (csrf_token_hash = None).
    # This enables the page-reload bootstrap where the first /refresh call
    # after page reload establishes the CSRF binding. The httpOnly cookie is
    # sufficient authentication for the bootstrap refresh.
    jafaal_sessions_utils.create_session(
        session_id,
        user,
        request,
        refresh_token,
        db,
    )

    # Token delivery (cookie vs body) is centralised in build_token_response so
    # login, /refresh, and the code exchange share one delivery contract.
    jafaal_audit.record(
        jafaal_audit.Event.LOGIN_SUCCESS,
        user_id=user.id,
        username=user.username,
        session_id=session_id,
        client_id=client.client_id,
        ip=network.get_ip_address(request),
    )

    # Best-effort security notification: a login from a device fingerprint not
    # seen on any prior session. Never blocks or fails the login.
    if emit_new_device and not known_device:
        _fingerprint, device_description = jafaal_sessions_utils.device_fingerprint(request)
        jafaal_ports.dispatch_event(
            "on_new_device_login",
            jafaal_ports.NewDeviceLogin(
                user_id=user.id,
                username=user.username,
                ip=network.get_ip_address(request),
                device_description=device_description,
                session_id=session_id,
            ),
        )
    return build_token_response(
        response,
        client,
        session_id,
        access_token,
        access_token_exp,
        refresh_token,
        refresh_token_exp,
        csrf_token,
        granted_scope(user, client, requested_scope),
    )

create_tokens

create_tokens(
    user,
    token_manager,
    session_id=None,
    client=None,
    requested_scope=None,
)

Generates session tokens for a user, including access token, refresh token, and CSRF token.

Parameters:

Name	Type	Description	Default
user	UserProtocol	The user object for whom the tokens are being created.	required
token_manager	TokenManager	The token manager responsible for token creation.	required
session_id	str | None	An optional session ID. If not provided, a new unique session ID is generated.	None
client	OAuthClient | None	The registered client the tokens are issued to. Its scope ceiling narrows what the tokens carry.	None
requested_scope	Sequence[str] | None	The scope the client asked for (RFC 6749 §3.3), applied as a final narrowing bound. None means "everything this client and user are entitled to".	None

Returns:

Type	Description
tuple[str, datetime, str, datetime, str, str]	tuple[str, datetime, str, datetime, str, str]: A tuple containing: - session_id (str): The session identifier. - access_token_exp (datetime): Expiration datetime of the access token. - access_token (str): The access token string. - refresh_token_exp (datetime): Expiration datetime of the refresh token. - refresh_token (str): The refresh token string. - csrf_token (str): The CSRF token string.

Source code in jafaal/utils.py

def create_tokens(
    user: jafaal_ports.UserProtocol,
    token_manager: jafaal_token_manager.TokenManager,
    session_id: str | None = None,
    client: jafaal_settings.OAuthClient | None = None,
    requested_scope: Sequence[str] | None = None,
) -> tuple[str, datetime, str, datetime, str, str]:
    """
    Generates session tokens for a user, including access token, refresh token, and CSRF token.

    Args:
        user (jafaal_ports.UserProtocol): The user object for whom the tokens are being created.
        token_manager (jafaal_token_manager.TokenManager): The token manager responsible for token creation.
        session_id (str | None, optional): An optional session ID. If not provided, a new unique session ID is generated.
        client: The registered client the tokens are issued to. Its scope
            ceiling narrows what the tokens carry.
        requested_scope: The ``scope`` the client asked for (RFC 6749 §3.3),
            applied as a final narrowing bound. ``None`` means "everything this
            client and user are entitled to".

    Returns:
        tuple[str, datetime, str, datetime, str, str]:
            A tuple containing:
                - session_id (str): The session identifier.
                - access_token_exp (datetime): Expiration datetime of the access token.
                - access_token (str): The access token string.
                - refresh_token_exp (datetime): Expiration datetime of the refresh token.
                - refresh_token (str): The refresh token string.
                - csrf_token (str): The CSRF token string.
    """
    if session_id is None:
        # Generate a unique session ID
        session_id = str(uuid4())

    # Create the access, refresh tokens and csrf token
    access_token_exp, access_token = token_manager.create_token(
        session_id, user, jafaal_token_manager.TokenType.ACCESS, client, requested_scope
    )

    refresh_token_exp, refresh_token = token_manager.create_token(
        session_id, user, jafaal_token_manager.TokenType.REFRESH, client, requested_scope
    )

    csrf_token = token_manager.create_csrf_token()

    return (
        session_id,
        access_token_exp,
        access_token,
        refresh_token_exp,
        refresh_token,
        csrf_token,
    )

__getattr__

__getattr__(name)

Lazily import the model-touching public API on first access (PEP 562).

Source code in jafaal/__init__.py

def __getattr__(name: str) -> object:
    """Lazily import the model-touching public API on first access (PEP 562)."""
    module_name = _LAZY_EXPORTS.get(name)
    if module_name is None:
        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
    value = getattr(_import_module(module_name), name)
    globals()[name] = value  # cache so later access skips __getattr__
    return value