API Reference

Protocols

`IdentityResolverProtocol`

Protocol for resolving external OAuth identities to internal user IDs.

This protocol is used to map OAuth external IDs (like Google’s ‘sub’ claim) to internal UUIDs. This is necessary because OAuth providers use their own ID format which may not be valid UUIDs.

resolve_user_id

async def resolve_user_id(
    external_id: str,
    provider: str = 'google'
) -> str | None

source

Resolve an external OAuth ID to an internal user UUID.

Parameters

Parameter	Type	Description
`external_id`	str	The external ID from the OAuth provider (e.g., Google's sub claim).
`provider`	str	The OAuth provider name (e.g., "google", "apple").

Returns

Type	Description
str \| None	The internal user UUID if found, None otherwise.

resolve_user_id_sync

def resolve_user_id_sync(
    external_id: str,
    provider: str = 'google'
) -> str | None

source

Synchronous version of resolve_user_id.

Parameters

Parameter	Type	Description
`external_id`	str	The external ID from the OAuth provider.
`provider`	str	The OAuth provider name.

Returns

Type	Description
str \| None	The internal user UUID if found, None otherwise.

`OAuthIdentityStore`

source

Protocol for OAuth identity storage.

OAuthIdentityStore manages the linking between local user accounts and external OAuth2 identity providers. This enables users to authenticate using social login while maintaining a consistent local user identity.

Example

Using OAuth identity store

store = await container.resolve(OAuthIdentityStore)

# Link OAuth identity to user
identity = await store.create_oauth_identity(
    user_id="user-123",
    provider="google",
    provider_user_id="google-456"
)

# Find user by OAuth identity
identity = await store.get_oauth_identity("google", "google-456")
if identity:
    user = await store.get_user_by_oauth_identity(identity.id)

store = await container.resolve(OAuthIdentityStore)

# Link OAuth identity to user
identity = await store.create_oauth_identity(
    user_id="user-123",
    provider="google",
    provider_user_id="google-456"
)

# Find user by OAuth identity
identity = await store.get_oauth_identity("google", "google-456")
if identity:
    user = await store.get_user_by_oauth_identity(identity.id)

create_oauth_identity

async def create_oauth_identity(
    user_id: str,
    provider: str,
    provider_user_id: str
) -> OAuthIdentity

source

Create an OAuth identity link for a user.

Parameters

Parameter	Type	Description
`user_id`	str	The local user ID to link the OAuth identity to.
`provider`	str	The OAuth provider name (e.g., "google", "github").
`provider_user_id`	str	The user's ID at the OAuth provider.

Returns

Type	Description
OAuthIdentity	The created OAuthIdentity instance.

get_oauth_identity

async def get_oauth_identity(
    provider: str,
    provider_user_id: str
) -> OAuthIdentity | None

source

Get OAuth identity by provider and provider user ID.

Parameters

Parameter	Type	Description
`provider`	str	The OAuth provider name.
`provider_user_id`	str	The user's ID at the OAuth provider.

Returns

Type	Description
OAuthIdentity \| None	The OAuthIdentity if found, None otherwise.

get_oauth_identities_for_user

async def get_oauth_identities_for_user(user_id: str) -> list[OAuthIdentity]

source

Get all OAuth identities for a user

delete_oauth_identity

async def delete_oauth_identity(
    provider: str,
    provider_user_id: str
) -> bool

source

Delete OAuth identity

delete_oauth_identities_for_user

async def delete_oauth_identities_for_user(user_id: str) -> int

source

Delete all OAuth identities for a user

get_user_by_oauth_identity

async def get_user_by_oauth_identity(
    provider: str,
    provider_user_id: str
) -> str | None

source

Get local user_id by OAuth provider and external user ID.

This is the key method for resolving OAuth external IDs to local user IDs. Used when OAuth tokens contain non-UUID user identifiers (like Google’s sub claim).

Parameters

Parameter	Type	Description
`provider`	str	The OAuth provider name (e.g., "google", "github", "apple").
`provider_user_id`	str	The user's ID at the OAuth provider.

Returns

Type	Description
str \| None	The local user_id if found, None otherwise.

resolve_user_id

async def resolve_user_id(
    user_id_or_oauth_id: str,
    provider: str = 'google'
) -> str | None

source

Resolve a user_id that may be either a UUID or an OAuth external ID.

This method handles the common issue where OAuth providers (like Google) use non-UUID identifiers (e.g., “101158382316025899191”) which cannot be used directly in database queries expecting UUIDs.

Resolution logic:

If user_id_or_oauth_id is a valid UUID format, check if user exists
If not a valid UUID, treat it as an OAuth provider_user_id and look up

Parameters

Parameter	Type	Description
`user_id_or_oauth_id`	str	Either a local user UUID or an OAuth provider's external ID.
`provider`	str	The OAuth provider to search in (default: "google").

Returns

Type	Description
str \| None	The resolved local user_id if found, None otherwise.

`TokenValidatorProtocol`

source

Validates and decodes authentication tokens.

validate

async def validate(token: str) -> TokenPayload | None

source

Validate a token and return its payload, or None if invalid.

revoke

async def revoke(token: str) -> bool

source

Revoke a token, preventing future validation.

Classes

`APIKeyAuthenticator`

source

Authenticates requests by validating an API key.

Key extraction order:

request_context["headers"][config.header_name]
request_context["query_params"][config.query_param]

If the key is absent or the lookup returns None an Err(AuthenticationError) is returned. On success Ok(User) is returned.

Example

config = APIKeyConfig(
    lookup={"sk_live_abc123": admin_user},
)
authenticator = APIKeyAuthenticator(config)

result = await authenticator.authenticate({
    "headers": {"X-API-Key": "sk_live_abc123"},
    "query_params": {},
})

if result.is_ok():
    user = result.unwrap()

config = APIKeyConfig(
    lookup={"sk_live_abc123": admin_user},
)
authenticator = APIKeyAuthenticator(config)

result = await authenticator.authenticate({
    "headers": {"X-API-Key": "sk_live_abc123"},
    "query_params": {},
})

if result.is_ok():
    user = result.unwrap()

__init__

def __init__(config: APIKeyConfig) -> None

source

Initialise the authenticator.

Parameters

Parameter	Type	Description
`config`	APIKeyConfig	Key-extraction and lookup configuration.

authenticate

async def authenticate(request_context: dict[str, Any]) -> Result[User, AuthenticationError]

source

Authenticate a request using its API key.

Parameters

Parameter	Type	Description
`request_context`	dict[str, Any]	A mapping that must contain at least one of: * ``"headers":` `dict[str, str]` `of HTTP request headers. *` `"query_params":` `dict[str, str]`` of URL query params.

Returns

Type	Description
Result[User, AuthenticationError]	``Ok(User)` `when a valid API key is found.` `Err(AuthenticationError)` `when the key is absent, invalid, or the lookup returns` `None``.

`APIKeyConfig`

source

Configuration for APIKeyAuthenticator.

Attributes: header_name: HTTP header to inspect for the API key. Defaults to "X-API-Key". query_param: URL query parameter name to fall back to when the header is absent. Defaults to "api_key". lookup: Either a dict[str, User] mapping raw key strings to their owners, or an async/sync callable that accepts the raw key and returns User | None. This field is required.

`AuthAuthenticationFailedHook`

source

Payload fired when an authentication attempt fails.

Attributes: method: Authentication method that was attempted. reason: Short description of why authentication failed.

`AuthBundleProvider`

source

Composite provider that wires the full Quadkit auth stack.

Composes AuthenticationProvider, TokenProvider, SessionProvider, and AuthorizationProvider so that callers only need to register a single provider:

.. code-block:: python

container.add_provider(AuthBundleProvider(config=auth_config))

Dependencies registered by each sub-provider are available in the container after register completes.

Parameters

Parameter	Type	Description
`config`		Shared AuthConfig forwarded to every sub-provider. When ``None``, each sub-provider uses its own defaults.
`initial_roles`		Optional initial RBAC roles forwarded to AuthorizationProvider.
`enable_passkeys`		When ``True``, append PasskeyProvider to the sub-provider list (requires the WebAuthn extra to be installed).
`session_store`		Explicit host-owned session backend, retained through eager and deferred sub-provider composition.
`user_store`		Explicit host-owned identity store, or a store kind (``"memory"/"sql"``), forwarded to the authentication sub-provider.
`user_table`		SQL identity table name, ignored unless the resolved store kind is ``"sql"``.
`cache_service`		Explicit host-owned token blacklist backend.
`kwargs`		Extra keyword arguments forwarded to AuthenticationProvider.

__init__

def __init__(
    config: AuthConfig | None = None,
    initial_roles: dict[str, Any] | None = None,
    enable_passkeys: bool = False,
    session_store: SessionStore | None = None,
    *,
    user_store: UserStoreProtocol | UserStoreKind | None = None,
    user_table: str | None = None,
    cache_service: Any = None,
    **kwargs: Any
) -> None

source

from_config

def from_config(
    cls,
    config: AuthConfig,
    **context: Any
) -> Self

source

Create provider from config object.

async def register(container: ContainerRegistrarProtocol) -> None

source

Late config binding: the orchestrator injects the typed auth section (via config_key) after construction and before this call. If configure() ran with no explicit config, recompose now so the automatic path behaves identically to the explicit one.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	The DI container registrar.

boot

async def boot(container: BootContainerProtocol) -> None

source

Boot all auth sub-providers in registration order.

Parameters

Parameter	Type	Description
`container`	BootContainerProtocol	The DI container resolver.

shutdown

async def shutdown() -> None

source

Shut down all auth sub-providers in reverse registration order.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Aggregate health check across all sub-providers.

Returns DEGRADED if any sub-provider is unhealthy.

Parameters

Parameter	Type	Description
`timeout`	float	Per-provider timeout budget in seconds.

Returns

Type	Description
HealthCheckResult	An aggregated HealthCheckResult.

`AuthConfig`

source

Hierarchical root configuration for Quadkit Auth.

Attributes: name: Configuration name (default: “auth”) enabled: Whether the auth module is enabled users: Initial users to create user_store: Identity store selection (memory or sql) user_table: Table the SQL identity store owns (default users) roles: Role definitions for RBAC rbac: RBAC system configuration token: JWT configuration middleware: Authentication middleware configuration secret_key: Secret key for signing tokens admin_email: Initial admin email admin_password: Initial admin password login_rate_limit: Rate limit for login endpoints oauth2_providers: OAuth2 provider configurations

validate_security

def validate_security() -> AuthConfig

source

Ensure secure settings in production.

`AuthModule`

source

Full authentication and authorization stack (JWT, OAuth2/OIDC, RBAC, policies).

Call configure to configure the auth bundle with an AuthConfig.

Usage

from quadkit.auth.config import AuthConfig

@module(
    imports=[AuthModule.configure(AuthConfig(secret_key="..."))]
)
class AppModule(Module):
    pass

from quadkit.auth.config import AuthConfig

@module(
    imports=[AuthModule.configure(AuthConfig(secret_key="..."))]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    config: Any | None = None,
    initial_roles: dict[str, Any] | None = None,
    is_global: bool = True,
    *,
    user_store: UserStoreProtocol | UserStoreKind | None = None,
    user_table: str | None = None,
    cache_service: Any = None,
    session_store: SessionStore | None = None
) -> DynamicModule

source

Create an AuthModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	Any \| None	AuthConfig or ``None`` for framework defaults (development-only ephemeral secrets).
`initial_roles`	dict[str, Any] \| None	Optional RBAC role seed forwarded to the authorization sub-provider.
`user_store`	UserStoreProtocol \| UserStoreKind \| None	Host-owned store (or ``"memory"/"sql"`` kind) used by authentication. Defaults to in-memory storage; durable hosts must supply a persistent store. The host owns migrations and the store connection lifecycle.
`user_table`	str \| None	SQL identity table name. Ignored unless the resolved store kind is ``"sql"``.
`session_store`	SessionStore \| None	Explicit host-owned session backend. No migrations or cookie routes are enabled by supplying it.
`cache_service`	Any	Explicit host-owned token blacklist backend forwarded to authentication; must provide durable set/exists operations when the host promises persistent JWT revocation.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: Any = None
) -> DynamicModule

source

Return an in-memory AuthModule suitable for unit and integration testing.

Uses ephemeral in-memory storage with a fixed test secret key. No external token services, databases, or OAuth providers are configured.

Parameters

Parameter	Type	Description
`config`	Any	Optional test configuration override.

Returns

Type	Description
DynamicModule	A DynamicModule backed by in-memory auth storage.

`AuthResult`

source

Result of an authentication attempt.

`AuthStatus`

source

Authentication status values.

`AuthToken`

source

Authentication token response.

`AuthTokenIssuedHook`

source

Payload fired when an access or refresh token is issued.

Attributes: user_id: Identifier of the user the token was issued for. token_type: Token kind (e.g. "access", "refresh").

`AuthTokenRefreshedHook`

source

Payload fired when an access token is refreshed.

Attributes: user_id: Identifier of the user whose token was refreshed. token_type: Token kind that was refreshed.

`AuthTokenRevokedHook`

source

Payload fired when a token is explicitly revoked.

Attributes: user_id: Identifier of the user whose token was revoked. token_type: Token kind that was revoked.

`AuthUserAuthenticatedHook`

source

Payload fired when a user successfully authenticates.

Attributes: user_id: Identifier of the authenticated user. method: Authentication method used (e.g. "password", "oauth2").

`AuthenticationFailed`

source

Emitted when an authentication attempt fails.

Attributes: email: Email that was used in the failed attempt. reason: Human-readable reason for the failure. ip: Remote IP address of the request, if available.

`AuthenticationProvider`

source

User authentication ONLY (login/logout/validation).

config is the primary parameter and drives all defaults. The optional keyword arguments are override points for callers that need to supply custom implementations (e.g., a production-grade UserStoreProtocol backed by a database, or a pre-built token manager).

Parameters

Parameter	Type	Description
`config`		The resolved AuthConfig. JWT credentials (``config.token.secret_key,` `config.token.algorithm) are used to build a JWTTokenManager automatically when no` `token_manager`` override is provided.
`password_policy`		Override the password policy derived from ``config.password. When None the policy is built from config; falls back to` `PasswordPolicy()`` when config is also absent.
`user_store`		Override the user store with a ready instance. When supplied it wins over ``user_store_kind``/config selection.
`user_store_kind`		Identity-store selection (``"memory"` `or` `"sql"). Overrides` `config.user_store;` `"sql"` `defers store construction to boot, which resolves the application's` `DatabaseProviderProtocol`` (registration never needs a live database) and initializes the store's table before the authentication service is built.
`user_table`		Table the SQL identity store owns. Overrides ``config.user_table; defaults to` `users``.
`token_manager`		Supply a fully constructed token manager. When None and ``config.token`` is present the manager is built automatically from the JWT configuration.
`cache_service`		Optional cache backend passed to the auto-built token manager for token blacklisting. Ignored when ``token_manager`` is provided directly.
`mfa_service`		Optional MFA manager for multi-factor authentication flows.

__init__

def __init__(
    config: Annotated[AuthConfig, Inject] | None = None,
    *,
    password_policy: PasswordPolicy | None = None,
    user_store: UserStoreProtocol | None = None,
    user_store_kind: UserStoreKind | None = None,
    user_table: str | None = None,
    token_manager: Any = None,
    cache_service: Any = None,
    mfa_service: MFAManager | None = None
) -> None

source

password_hasher

property password_hasher() -> PasswordHasherProtocol

source

Composed Argon2id-default hasher with a bcrypt legacy shim.

Builds the single composed instance (ODD-1 Option A): new hashes use Argon2id with OWASP-2024 parameters; stored bcrypt hashes keep verifying through the legacy shim and upgrade on the next login.

service

property service() -> AuthenticationService

source

Get or create the authentication service.

get_user

async def get_user(user_id: str) -> Any | None

source

Fetch a user by their ID.

Satisfies AuthProviderProtocol.

Parameters

Parameter	Type	Description
`user_id`	str	The unique identifier of the user to retrieve.

Returns

Type	Description
Any \| None	The user object, or ``None`` if not found.

has_any_role

def has_any_role(
    user: Any,
    roles: list[str]
) -> bool

source

Return True if user holds at least one of the given roles.

Satisfies AuthProviderProtocol. Delegates to the sibling AuthorizationService resolved during boot. When no authorization provider is registered in the same container (e.g. an app only wires AuthenticationProvider without AuthorizationProvider), conservatively returns False rather than raising.

Parameters

Parameter	Type	Description
`user`	Any	Authenticated user object.
`roles`	list[str]	Role names to check.

Returns

Type	Description
bool	True if the user has at least one of the supplied roles.

has_any_permission

def has_any_permission(
    user: Any,
    permissions: list[str]
) -> bool

source

Return True if user has at least one of the given permissions.

Satisfies AuthProviderProtocol. See has_any_role for delegation and fallback behavior.

Parameters

Parameter	Type	Description
`user`	Any	Authenticated user object.
`permissions`	list[str]	Permission names to check.

Returns

Type	Description
bool	True if the user has at least one of the supplied permissions.

async def register(container: ContainerRegistrarProtocol) -> None

source

Registers the protocols for dependency injection across extensions without direct imports. The single PasswordHasherProtocol binding is the composed hasher (Argon2id default, bcrypt legacy shim) — one registration per protocol per AGENTS.md §2.6.

boot

async def boot(container: BootContainerProtocol) -> None

source

Initialize authentication provider and register with kernel health registry.

shutdown

async def shutdown() -> None

source

Shutdown authentication provider.

verify_token

async def verify_token(token: str) -> Result[VerifiedToken, TokenError]

source

Verify a JWT token and return a Result with the decoded payload.

Satisfies AuthProviderProtocol.

Delegates to token_manager.verify_token() when a token manager is configured. Returns an Err result if no token manager is set or if the token is invalid for any expected domain reason.

Infrastructure failures (cache unavailable, network errors) are still raised as exceptions and must be handled by the caller.

Parameters

Parameter	Type	Description
`token`	str	The JWT token string to verify.

Returns

Type	Description
Result[VerifiedToken, TokenError]	``Ok(VerifiedToken)` `if the token is valid and not revoked, or` `Err(TokenError)`` for expected domain failures.

validate_session

async def validate_session(token: str) -> Any

source

Validate a session token and return user information.

Delegates to the underlying authentication service’s get_user_from_token method. Returns a Result[VerifiedToken, TokenError] or None when no token manager is configured.

Parameters

Parameter	Type	Description
`token`	str	The session or JWT token to validate.

Returns

Type	Description
Any	``Result[VerifiedToken, TokenError]` `on success/failure, or` `None``.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check authentication provider health.

A SQL identity store that was requested but not yet resolved (deferred past boot) reports DEGRADED rather than claiming healthy persistence it does not have; once materialised it is healthy only when its DDL has run.

`AuthenticationService`

source

Service for core authentication operations.

Handles user login, registration, token creation and validation.

__init__

def __init__(
    password_policy: PasswordPolicy,
    user_store: UserStoreProtocol,
    token_manager: JWTTokenManager,
    lockout_config: LockoutConfig | None = None,
    event_bus: EventBusProtocol | None = None,
    tracker: LoginAttemptTracker | None = None,
    hooks: HookRegistryProtocol | None = None,
    password_hasher: PasswordHasherProtocol | None = None,
    activity_tracker: AuthActivityTracker | None = None
) -> None

source

set_hook_registry

def set_hook_registry(hooks: HookRegistryProtocol | None) -> None

source

Attach an optional hook registry after provider boot wiring.

authenticate_user

async def authenticate_user(
    email: str,
    password: str
) -> Result[User, InvalidCredentialsError | AccountLockedError]

source

Authenticate a user with email and password.

Always performs a password hash verification regardless of whether the user exists. This constant-time behaviour prevents user-enumeration attacks via timing side-channels.

If the account has exceeded LockoutConfig.max_attempts within the observation window an AccountLockedError is returned immediately without performing credential verification.

Returns

Type	Description
Result[User, InvalidCredentialsError \| AccountLockedError]	``Ok(User)` `if authentication succeeds,` `Err(AccountLockedError)` `if the account is temporarily locked,` `Err(InvalidCredentialsError)`` otherwise.

change_user_password

async def change_user_password(
    user_id: str,
    current_password: str,
    new_password: str
) -> Result[None, InvalidCredentialsError | PasswordPolicyError]

source

Change a user’s password after verifying the current password.

This lives on AuthenticationService so applications that use the provider’s composed password hasher keep the same Argon2id-default plus bcrypt-legacy behaviour for both login and password changes.

register_user

async def register_user(request: RegisterRequest) -> Result[User, EmailExistsError | UsernameExistsError | PasswordPolicyError]

source

Parameters

Parameter	Type	Description
`request`	RegisterRequest	Registration request with email, name, and password.

Returns

Type	Description
Result[User, EmailExistsError \| UsernameExistsError \| PasswordPolicyError]	``Ok(User)` `on success.` `Err(PasswordPolicyError)` `if passwords do not match or the password violates the policy.` `Err(EmailExistsError)` `if the email is already registered.` `Err(UsernameExistsError)`` if the name is already registered.

create_token

def create_token(user: User) -> AuthToken

source

Create an authentication token for a user.

verify_token

async def verify_token(token: str) -> Result[VerifiedToken, TokenError]

source

Verify and decode an authentication token.

Returns

Type	Description
Result[VerifiedToken, TokenError]	``Ok(VerifiedToken)` `if valid,` `Err(TokenError)`` otherwise.

refresh_token

async def refresh_token(refresh_token: str) -> Result[AuthToken, TokenError]

source

Refresh an access token using a refresh token.

Parameters

Parameter	Type	Description
`refresh_token`	str	The refresh token string.

Returns

Type	Description
Result[AuthToken, TokenError]	``Ok(AuthToken)` `if the refresh succeeds.` `Err(TokenError)`` if the refresh token is invalid or expired.

get_user_from_token

async def get_user_from_token(token: str) -> Result[VerifiedToken, TokenError]

source

Get user information from token.

Returns

Type	Description
Result[VerifiedToken, TokenError]	``Ok(VerifiedToken)` `if valid,` `Err(TokenError)`` otherwise.

shutdown

async def shutdown() -> None

source

Cancel and await all pending background event tasks.

`AuthorizationProvider`

source

Role-based access control and permission management.

__init__

def __init__(
    config: Annotated[AuthConfig, Inject] | None = None,
    initial_roles: dict[str, Any] | None = None,
    **kwargs: Any
) -> None

source

auth_config

property auth_config() -> AuthConfig | None

source

async def register(container: ContainerRegistrarProtocol) -> None

source

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Initialize authorization provider.

shutdown

async def shutdown() -> None

source

Shutdown authorization provider.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check authorization provider health.

`GoogleOAuthProvider`

source

Registers a first-class Google OAuth verification service.

__init__

def __init__(
    config: Annotated[AuthConfig, Inject] | None = None,
    google_oauth: dict[str, str] | None = None,
    http_client: Any | None = None,
    **kwargs: Any
) -> None

source

service

property service() -> GoogleOAuthService | None

source

Return the registered Google OAuth service, if any.

async def register(container: ContainerRegistrarProtocol) -> None

source

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Initialize Google OAuth support.

shutdown

async def shutdown() -> None

source

Shutdown Google OAuth support.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check Google OAuth provider health.

`GoogleOAuthService`

source

Verify Google OAuth tokens and normalize the verified claims.

__init__

def __init__(
    *,
    client_id: str,
    http_client: HTTPClientProtocol | None = None,
    jwks_url: str = GOOGLE_JWKS_URL,
    tokeninfo_url: str = GOOGLE_TOKENINFO_URL,
    userinfo_url: str = GOOGLE_USERINFO_URL,
    allowed_issuers: tuple[str, Ellipsis] = GOOGLE_ISSUERS,
    jwks_cache_ttl_seconds: int = 300
) -> None

source

verify_token

async def verify_token(token: str) -> VerifiedIdentityClaims

source

Verify a Google token, preferring ID-token JWKS validation.

Parameters

Parameter	Type	Description
`token`	str	Google-issued ID token or access token.

Returns

Type	Description
VerifiedIdentityClaims	Normalized, verified Google identity claims.

Raises

Exception	Description
OAuth2Error	If the token cannot be verified or normalized.

verify_id_token

async def verify_id_token(token: str) -> VerifiedIdentityClaims

source

Verify a Google ID token against Google’s JWKS.

verify_userinfo_token

async def verify_userinfo_token(token: str) -> VerifiedIdentityClaims

source

Verify a Google access token via the userinfo endpoint.

verify_tokeninfo

async def verify_tokeninfo(token: str) -> VerifiedIdentityClaims

source

Verify a Google token using the tokeninfo endpoint fallback.

`JWTConfig`

source

JWT Configuration

JWT verification policy

The framework enforces verified-only JWT decoding. Signature verification is never disabled.

PRODUCTION / STAGING: A secret is required and must meet strength checks (no known default value, >= 32 bytes for HS algorithms). An explicit required_audience is mandatory so tokens cannot cross service boundaries.
DEVELOPMENT: A missing secret falls back to a generated ephemeral secret so signature verification stays enabled; tokens are invalidated on restart. Set QK_AUTH__TOKEN__SECRET_KEY for a stable development secret.

validate_jwt_security

def validate_jwt_security() -> JWTConfig

source

Enforce verified-only JWT policy based on deployment environment.

`JWTTokenManager`

source

JWT token management with key rotation support.

The JWTTokenManager handles creation, validation, and renewal of JWT tokens for user authentication. It supports multiple signing keys for seamless key rotation without invalidating existing tokens.

Attributes: current_key_id: ID of the currently active signing key. keys: Dictionary of key_id -> key material. algorithm: JWT signing algorithm (HS256, RS256, etc.). access_expiration_hours: Expiration time for access tokens. refresh_expiration_days: Expiration time for refresh tokens. cache_service: Optional cache backend for token validation caching. rotation_interval: Interval in seconds between key rotations.

Example

Basic token operations

manager = JWTTokenManager(
    current_key_id="v1",
    keys={"v1": SecretStr("secure-secret-key")},
)

# Create access token
token = await manager.create_access_token(user_id="user123")

# Verify and decode
payload = await manager.verify_token(token)

manager = JWTTokenManager(
    current_key_id="v1",
    keys={"v1": SecretStr("secure-secret-key")},
)

# Create access token
token = await manager.create_access_token(user_id="user123")

# Verify and decode
payload = await manager.verify_token(token)

Note

In production, secrets should not be hardcoded. Use environment variables or a secrets management service.

__init__

def __init__(
    current_key_id: str,
    keys: dict[str, str | SecretStr | dict[str, str | SecretStr]] | None = None,
    algorithm: str = const.DEFAULT_TOKEN_ALGORITHM,
    access_expiration_hours: float = 24,
    refresh_expiration_days: float = 30,
    cache_service: CacheBackendProtocol | None = None,
    rotation_interval_days: int = 90,
    grace_period_seconds: float = const.DEFAULT_JWT_KEY_ROTATION_GRACE_PERIOD_SECONDS,
    logger: Logger | None = None,
    *,
    audit_logger: AuditLoggerProtocol | None = None,
    binding_config: TokenBindingConfig | None = None,
    required_audience: str | None = None,
    ids: IdGeneratorProtocol | None = None
) -> None

source

Initialize the JWT token manager.

Parameters

Parameter	Type	Description
`current_key_id`	str	The ID of the currently active signing key.
`keys`	dict[str, str \| SecretStr \| dict[str, str \| SecretStr]] \| None	Dictionary mapping key IDs to key material. If None, current_key_id is treated as a single secret.
`algorithm`	str	JWT signing algorithm (e.g., "HS256", "RS256").
`access_expiration_hours`	float	Hours until access tokens expire.
`refresh_expiration_days`	float	Days until refresh tokens expire.
`cache_service`	CacheBackendProtocol \| None	Optional cache backend for token validation.
`rotation_interval_days`	int	Days between key rotations.
`grace_period_seconds`	float	Seconds after key rotation during which tokens signed by the outgoing key remain valid. Defaults to 3600 (1 hour). Set to 0 to invalidate all old-key tokens immediately on rotation.
`audit_logger`	AuditLoggerProtocol \| None	Optional AuditLoggerProtocol used to record token revocation events. When not provided, no audit entries are written.
`binding_config`	TokenBindingConfig \| None	Optional TokenBindingConfig for opt-in client binding. When set, tokens embed a ``bind` `claim containing a SHA-256 hash of active binding factors (IP, fingerprint). Tokens issued without a` `bind`` claim continue to verify successfully so that binding can be enabled incrementally.
`required_audience`	str \| None	When set, every call to ``verify_token` `will enforce that the token's` `aud` `claim matches this value. Pass` `allow_missing_audience=True` `to` `verify_token`` to bypass the check on a per-call basis for trusted internal paths.

keys

property keys() -> dict[str, Any]

source

Live view of the key material managed by the key store.

current_key_id

property current_key_id() -> str

source

The key ID currently used for signing new tokens.

current_key_id

def current_key_id(value: str) -> None

source

Allow external callers to update the active key ID on the store.

set_blacklist_resolver

def set_blacklist_resolver(resolver: Any) -> None

source

Attach a deferred blacklist-cache source.

Parameters

Parameter	Type	Description
`resolver`	Any	Zero-arg callable returning a ``CacheBackendProtocol` `or` `None``; invoked lazily on first revocation use.

set_blacklist_cache

def set_blacklist_cache(cache: Any) -> None

source

Attach a cache backend to the blacklist for durable revocation.

Lets DI providers supply the application cache during boot() when it was not available at construction time.

Parameters

Parameter	Type	Description
`cache`	Any	Cache backend used for token/user revocation entries.

set_hook_registry

def set_hook_registry(hooks: HookRegistryProtocol | None) -> None

source

Attach an optional hook registry after provider boot wiring.

rotate_key

async def rotate_key(
    new_key_id: str,
    new_secret: str | dict
) -> None

source

Rotate to a new signing key, delegating lifecycle to the key store.

Old keys are retained for grace_period_seconds (constructor parameter, default 3600 s) so tokens they signed remain verifiable during the overlap window.

Parameters

Parameter	Type	Description
`new_key_id`	str	ID for new key.
`new_secret`	str \| dict	New secret key (string for symmetric or dict for asymmetric).

list_keys

def list_keys() -> dict[str, dict[str, Any]]

source

Return current key metadata (for inspection/operations).

get_user_from_token

async def get_user_from_token(token: str) -> Result[VerifiedToken, ContractsTokenError]

source

Extract user information from access token.

Returns

Type	Description
Result[VerifiedToken, ContractsTokenError]	``Ok(VerifiedToken)` `if the token is a valid access token, or` `Err(TokenError)`` for expected domain failures.

`LockoutConfig`

source

Configuration for account lockout on repeated failed login attempts.

Attributes: max_failed_attempts: Number of failed attempts within the observation window that triggers a lockout. Defaults to 5. lockout_duration_seconds: Length of the rolling observation window in seconds. The lockout is lifted automatically once all recorded failures fall outside this window. Defaults to 300 (5 minutes). max_attempts: Canonical alias for max_failed_attempts. Both fields default to 5 and are kept in sync by __post_init__.

Note

For distributed deployments pass a CacheBackendProtocol to LoginAttemptTracker instead of relying on the in-process dict managed by AuthenticationService.

`LoginAttemptTracker`

source

Tracks failed login attempts and enforces account lockout.

Supports both in-process state (no external dependency) and distributed state via an injected CacheBackendProtocol. When a cache backend is provided all attempt records are stored there so that multiple application instances share the same view — preventing an attacker from bypassing lockout by rotating across instances.

Parameters

Parameter	Type	Description
`max_attempts`		Consecutive failures within lockout_duration_seconds that trigger a lockout. Defaults to 5.
`lockout_duration_seconds`		Rolling window size and cache TTL. Defaults to 900 seconds (15 minutes).
`cache`		Optional cache backend for distributed state. When ``None` `an in-process` `dict`` is used instead.

Example

tracker = LoginAttemptTracker(max_attempts=5, lockout_duration_seconds=900)

await tracker.record_failure("[email protected]")
locked = await tracker.is_locked("[email protected]")  # False until 5 failures

await tracker.clear("[email protected]")  # reset on successful login

tracker = LoginAttemptTracker(max_attempts=5, lockout_duration_seconds=900)

await tracker.record_failure("[email protected]")
locked = await tracker.is_locked("[email protected]")  # False until 5 failures

await tracker.clear("[email protected]")  # reset on successful login

__init__

def __init__(
    max_attempts: int = 5,
    lockout_duration_seconds: int = 900,
    cache: CacheBackendProtocol | None = None
) -> None

source

is_locked

async def is_locked(identifier: str) -> bool

source

Return True if identifier has exceeded the failure threshold.

Parameters

Parameter	Type	Description
`identifier`	str	Username, e-mail address, or IP used as the tracking key.

record_failure

async def record_failure(identifier: str) -> None

source

Record a failed authentication attempt for identifier.

Parameters

Parameter	Type	Description
`identifier`	str	Username, e-mail address, or IP used as the tracking key.

clear

async def clear(identifier: str) -> None

source

Remove all recorded failures for identifier (call on success).

Parameters

Parameter	Type	Description
`identifier`	str	Username, e-mail address, or IP used as the tracking key.

`MFAConfig`

source

Multi-factor authentication configuration.

Controls TOTP settings, backup code generation, and challenge attempt limits. MFA state is stored in the user profile (user.profile['mfa']) — no separate MFA table is required.

`MongoDBOAuthIdentityStore`

source

MongoDB-backed OAuth identity store

__init__

def __init__(
    db_provider: DatabaseProviderProtocol,
    collection_name: str = 'oauth_identities'
)

source

create_oauth_identity

async def create_oauth_identity(
    user_id: str,
    provider: str,
    provider_user_id: str
) -> OAuthIdentity

source

Create OAuth identity link

get_oauth_identity

async def get_oauth_identity(
    provider: str,
    provider_user_id: str
) -> OAuthIdentity | None

source

Get OAuth identity by provider and provider user ID

get_oauth_identities_for_user

async def get_oauth_identities_for_user(user_id: str) -> list[OAuthIdentity]

source

Get all OAuth identities for a user

delete_oauth_identity

async def delete_oauth_identity(
    provider: str,
    provider_user_id: str
) -> bool

source

Delete OAuth identity

delete_oauth_identities_for_user

async def delete_oauth_identities_for_user(user_id: str) -> int

source

Delete all OAuth identities for a user

get_user_by_oauth_identity

async def get_user_by_oauth_identity(
    provider: str,
    provider_user_id: str
) -> str | None

source

Get local user_id by OAuth provider and external user ID.

resolve_user_id

async def resolve_user_id(
    user_id_or_oauth_id: str,
    provider: str = 'google'
) -> str | None

source

Resolve user_id from either UUID or OAuth external ID.

Resolution logic:

If user_id_or_oauth_id is a valid UUID format, check if user exists
If not a valid UUID, treat it as an OAuth provider_user_id and look up

resolve_user_id_sync

def resolve_user_id_sync(
    external_id: str,
    provider: str = 'google'
) -> str | None

source

Synchronous resolution is not supported for database-backed store.

`OAuthIdentity`

source

OAuth identity linking user to provider

__init__

def __init__(
    user_id: str,
    provider: str,
    provider_user_id: str,
    created_at: datetime | None = None,
    updated_at: datetime | None = None
)

source

`PasswordChanged`

source

Emitted when a user successfully changes their password.

Attributes: user_id: ID of the user whose password was changed.

`PasswordHasher`

source

Bcrypt password hasher implementing the PasswordHasherProtocol.

Provides secure password hashing using bcrypt with UTF-8 aware truncation. The bcrypt cost factor is per-instance (rounds) so the configured value is honored both at construction and by the container singleton.

__init__

def __init__(rounds: int = _DEFAULT_BCRYPT_ROUNDS) -> None

source

hash

async def hash(password: str) -> str

source

Hash a password using bcrypt with UTF-8 aware truncation.

Parameters

Parameter	Type	Description
`password`	str	Plain text password.

Returns

Type	Description
str	Bcrypt hash string (``$2b$$...``) using the configured cost factor.

Raises

Exception	Description
RuntimeError	If bcrypt is not available.

verify

async def verify(
    password: str,
    hashed_password: str | bytes
) -> bool

source

Verify a password against its hash asynchronously.

Parameters

Parameter	Type	Description
`password`	str	Plain text password.
`hashed_password`	str \| bytes	Stored hash to compare against.

Returns

Type	Description
bool	True if the password matches the hash, False otherwise.

Raises

Exception	Description
RuntimeError	If bcrypt is not available.

needs_rehash

def needs_rehash(hashed_password: str) -> bool

source

Compare the stored hash’s cost to the configured target.

Parses the self-describing bcrypt prefix ($2b$<cost>$...); returns True when the stored cost is below the configured rounds. Unparseable or unknown formats return True (fail-closed) — safe because rehashing only ever runs after a successful verify().

Parameters

Parameter	Type	Description
`hashed_password`	str	Stored bcrypt hash string.

Returns

Type	Description
bool	True when the hash should be re-computed at the current cost.

rehash_if_needed

async def rehash_if_needed(
    password: str,
    hashed_password: str | None
) -> str | None

source

Rehash the password if needed.

`PasswordPolicy`

source

Password policy configuration.

Lazy-loads common passwords file only when needed. Implements PasswordPolicyProtocol for dependency injection compatibility.

__init__

def __init__(
    min_length: int = 8,
    max_length: int = 128,
    require_uppercase: bool = True,
    require_lowercase: bool = True,
    require_digits: bool = False,
    require_special: bool = False,
    prevent_common: bool = True,
    prevent_reuse: bool = False,
    history_size: int = 5,
    common_passwords_file: str | None = None,
    banned_patterns: list[str] | None = None
)

source

from_config

def from_config(
    cls,
    config: PasswordConfig
) -> PasswordPolicy

source

Build a PasswordPolicy from a PasswordConfig dataclass.

Parameters

Parameter	Type	Description
`config`	PasswordConfig	Password complexity configuration from ``AuthConfig.password``.

Returns

Type	Description
PasswordPolicy	A configured PasswordPolicy instance.

validate

def validate(password: str) -> None

source

Validate password against policy.

Parameters

Parameter	Type	Description
`password`	str	Plain text password to validate.

Raises

Exception	Description
ValueError	If the password violates the policy.

is_valid

def is_valid(password: str) -> bool

source

Return True if the password satisfies the policy without raising.

Parameters

Parameter	Type	Description
`password`	str	Plain text password.

Returns

Type	Description
bool	True if valid, False otherwise.

`PersistentTokenRevocationStore`

source

Persistent token revocation list backed by a CacheBackendProtocol.

Stores revoked token IDs in the cache so that revocations survive process restarts and are consistent across all nodes in a distributed deployment.

A sentinel value ("1") is written under quadkit:auth:revoked:{token_id} with the configured TTL. When expires_at is provided to revoke, the TTL is capped to the token’s remaining lifetime so stale entries are not kept longer than needed.

Parameters

Parameter	Type	Description
`cache`		Cache backend used to persist revoked token identifiers.
`ttl`		Default expiry in seconds for revocation records. Defaults to ``86400`` (24 h).

__init__

def __init__(
    cache: CacheBackendProtocol,
    ttl: int = _DEFAULT_TTL
) -> None

source

revoke

async def revoke(
    token_id: str,
    expires_at: datetime | None = None
) -> None

source

Mark token_id as revoked.

If the token is already expired (expires_at is in the past) the call is a no-op — there is no point persisting an entry for a token that the signature validation layer will reject on its own.

Parameters

Parameter	Type	Description
`token_id`	str	Unique identifier of the JWT/opaque token to revoke.
`expires_at`	datetime \| None	Optional token expiry time. Used to compute a shorter TTL so the cache entry is not held longer than the token's own lifetime.

is_revoked

async def is_revoked(token_id: str) -> bool

source

Return True if token_id is present in the revocation list.

Parameters

Parameter	Type	Description
`token_id`	str	The token identifier to check.

Returns

Type	Description
bool	``True` `when the token has been explicitly revoked and the revocation entry has not yet expired,` `False`` otherwise.

`RBACConfig`

source

RBAC system configuration.

`RoleDefinition`

source

A named collection of permissions.

The one role model in the framework (spec D2). Permission entries are strings in "resource.action" form, optionally scoped as "resource.action:scope" (scope in {“self”, “team”, “all”}, carried forward from the former rbac.Permission.scope field — evaluation of scope by the PDP is deferred, spec §7).

`SQLAlchemyOAuthIdentityStore`

source

Database-backed OAuth identity store

__init__

def __init__(db_provider: DatabaseProviderProtocol)

source

create_oauth_identity

async def create_oauth_identity(
    user_id: str,
    provider: str,
    provider_user_id: str
) -> OAuthIdentity

source

Create OAuth identity link

get_oauth_identity

async def get_oauth_identity(
    provider: str,
    provider_user_id: str
) -> OAuthIdentity | None

source

Get OAuth identity by provider and provider user ID

get_oauth_identities_for_user

async def get_oauth_identities_for_user(user_id: str) -> list[OAuthIdentity]

source

Get all OAuth identities for a user

delete_oauth_identity

async def delete_oauth_identity(
    provider: str,
    provider_user_id: str
) -> bool

source

Delete OAuth identity

delete_oauth_identities_for_user

async def delete_oauth_identities_for_user(user_id: str) -> int

source

Delete all OAuth identities for a user

get_user_by_oauth_identity

async def get_user_by_oauth_identity(
    provider: str,
    provider_user_id: str
) -> str | None

source

Get local user_id by OAuth provider and external user ID.

resolve_user_id

async def resolve_user_id(
    user_id_or_oauth_id: str,
    provider: str = 'google'
) -> str | None

source

Resolve user_id from either UUID or OAuth external ID.

Resolution logic:

If user_id_or_oauth_id is a valid UUID format, check if user exists
If not a valid UUID, treat it as an OAuth provider_user_id and look up

resolve_user_id_sync

def resolve_user_id_sync(
    external_id: str,
    provider: str = 'google'
) -> str | None

source

Synchronous resolution is not supported for database-backed store.

This method is required by IdentityResolverProtocol but cannot be implemented safely for an async database provider.

`SessionCookieBackend`

source

Session-cookie authentication backend for SSR flows.

Reads session ID from cookie, resolves user from SessionRepositoryProtocol.

Because session records contain only a user_id, resolving the full AuthenticatedUserProtocol requires a user_fetcher callable that looks up the concrete user object from whatever store the application uses (database, cache, etc.). This keeps the backend decoupled from any specific user-store implementation.

Constructor args: session_repository: Persistence backend for session records. user_fetcher: Async callable that accepts a user_id string and returns the matching AuthenticatedUserProtocol, or None when no user exists for that ID. cookie_name: Name of the session cookie (default "session_id"). secure: Whether to set the Secure flag (default True). http_only: Whether to set the HttpOnly flag (default True). same_site: The SameSite policy (default "lax").

Example

backend = SessionCookieBackend(
    session_repository=sql_session_repo,
    user_fetcher=user_service.get_authenticated_user,
)

# In an SSR handler:
user = await backend.authenticate(request)
if user is None:
    # redirect to /login
    ...

backend = SessionCookieBackend(
    session_repository=sql_session_repo,
    user_fetcher=user_service.get_authenticated_user,
)

# In an SSR handler:
user = await backend.authenticate(request)
if user is None:
    # redirect to /login
    ...

__init__

def __init__(
    session_repository: SessionRepositoryProtocol,
    user_fetcher: Callable[[str], Awaitable[AuthenticatedUserProtocol | None]],
    ids: IdGeneratorProtocol | None = None,
    cookie_name: str = 'session_id',
    secure: bool = True,
    http_only: bool = True,
    same_site: str = 'lax'
) -> None

source

authenticate

async def authenticate(request: Any) -> AuthenticatedUserProtocol | None

source

Extract session cookie, validate, return user or None.

Reads cookie_name from request.cookies, fetches the corresponding active session record, resolves the user via user_fetcher, and refreshes the last_active_at timestamp.

Parameters

Parameter	Type	Description
`request`	Any	The incoming request object. Must expose a ``cookies`` mapping attribute (Starlette / ASGI-compatible).

Returns

Type	Description
AuthenticatedUserProtocol \| None	The authenticated user if the session is valid, ``None`` otherwise.

async def login(
    response: Any,
    user_id: str,
    expires_in: int = 86400
) -> str

source

Create a session record and set the session cookie on response.

Parameters

Parameter	Type	Description
`response`	Any	The outgoing response object. Must expose a ``set_cookie` `method with` `key,` `value,` `max_age,` `secure,` `httponly, and` `samesite`` kwargs (Starlette / ASGI-compatible).
`user_id`	str	Identifier of the user to create a session for.
`expires_in`	int	Session lifetime in seconds (default 86 400 = 1 day).

Returns

Type	Description
str	The newly created ``session_id``.

logout

async def logout(
    request: Any,
    response: Any
) -> None

source

Invalidate the session and clear the cookie.

Parameters

Parameter	Type	Description
`request`	Any	The incoming request. ``request.cookies`` is read to find the current session ID.
`response`	Any	The outgoing response. ``delete_cookie`` is called to remove the session cookie from the browser.

`SessionCreated`

source

Emitted when a new session/token is issued.

Attributes: session_id: Identifier for the new session. user_id: ID of the user who owns the session.

`SessionRevoked`

source

Emitted when a session/token is invalidated.

Attributes: session_id: Identifier of the revoked session. user_id: ID of the user who owned the session.

`TokenRevoked`

source

Emitted when a specific token is revoked.

Attributes: token_id: JTI or opaque identifier of the revoked token. user_id: ID of the user who owned the token. reason: Human-readable reason for revocation.

`User`

source

User model representing an authenticated user.

This model serves as the core user representation within the auth package. It implements the AuthenticatedUserProtocol protocol.

has_role

def has_role(role: str) -> bool

source

Check if user has a specific role.

has_permission

def has_permission(permission: str) -> bool

source

Check if user has a specific permission.

with_role

def with_role(role: str) -> User

source

Return a new User with the given role added.

without_role

def without_role(role: str) -> User

source

Return a new User with the given role removed.

with_permission

def with_permission(permission: str) -> User

source

Return a new User with the given permission added.

without_permission

def without_permission(permission: str) -> User

source

Return a new User with the given permission removed.

record_login

def record_login() -> User

source

Return a new User with the current login recorded.

to_public_dict

def to_public_dict() -> dict[str, Any]

source

Return a safe public representation of the user.

from_dict

def from_dict(
    cls,
    data: dict[str, Any]
) -> User

source

Create from dictionary.

`UserAuthenticated`

source

Emitted when a user successfully authenticates.

Attributes: user_id: ID of the authenticated user. method: Authentication method used (e.g. "password"). ip: Remote IP address of the request, if available.

`UserLockedOut`

source

Emitted when a login attempt is rejected because the account is locked.

Attributes: user_id: ID of the locked-out user. email: Email address of the locked-out user.

`UserLoggedIn`

source

Emitted after a successful password-based login.

Attributes: user_id: ID of the authenticated user. email: Email address used to log in.

`UserLoggedOut`

source

Emitted when a user explicitly logs out.

Attributes: user_id: ID of the user who logged out.

`UserLoginFailed`

source

Emitted when a login attempt fails due to bad credentials.

Attributes: email: Email address used in the failed attempt. reason: Human-readable reason for the failure.

`UserRegistered`

source

Emitted when a new user account is created.

Attributes: user_id: ID of the newly created user. email: Email address of the newly created user.

`UserService`

source

Service for user management operations.

Handles user CRUD, password management, and administrative operations.

__init__

def __init__(
    password_policy: PasswordPolicy,
    user_store: Any,
    event_bus: EventBusProtocol | None = None
) -> None

source

create_user

async def create_user(
    name: str,
    email: str,
    password: str,
    roles: list[str] | None = None
) -> Result[User, EmailExistsError | UsernameExistsError | PasswordPolicyError]

source

Create a new user.

Returns

Type	Description
Result[User, EmailExistsError \| UsernameExistsError \| PasswordPolicyError]	``Ok(User)` `on success.` `Err(PasswordPolicyError)` `if the password does not satisfy policy.` `Err(EmailExistsError)` `if the email address is already registered.` `Err(UsernameExistsError)`` if the name is already registered.

Infrastructure exceptions (e.g. DB connectivity) propagate unchanged.

get_user

async def get_user(user_id: str) -> User | None

source

Get user by ID.

update_user

async def update_user(user: User) -> Result[User, UserNotFoundError | ValidationError]

source

Update user information.

Returns

Type	Description
Result[User, UserNotFoundError \| ValidationError]	``Ok(User)` `on success.` `Err(UserNotFoundError)` `if the user does not exist.` `Err(ValidationError)`` if the supplied user data is invalid (e.g. empty email).

Infrastructure exceptions (e.g. DB connectivity) propagate unchanged.

delete_user

async def delete_user(user_id: str) -> Result[None, UserNotFoundError]

source

Delete a user.

Returns

Type	Description
Result[None, UserNotFoundError]	``Ok(None)` `on success.` `Err(UserNotFoundError)`` if the user does not exist.

Raises

Exception	Description
AuthorizationError	If attempting to delete the protected ``admin`` account (security boundary — not a recoverable Result path).

Infrastructure exceptions (e.g. DB connectivity) propagate unchanged.

lock_user

async def lock_user(user_id: str) -> Result[None, UserNotFoundError]

source

Deactivate (lock) a user account.

Sets is_active = False so the user cannot log in until unlocked.

Returns

Type	Description
Result[None, UserNotFoundError]	``Ok(None)` `on success.` `Err(UserNotFoundError)`` if the user does not exist.

Infrastructure exceptions propagate unchanged.

unlock_user

async def unlock_user(user_id: str) -> Result[None, UserNotFoundError]

source

Reactivate (unlock) a previously locked user account.

Sets is_active = True so the user may log in again.

Returns

Type	Description
Result[None, UserNotFoundError]	``Ok(None)` `on success.` `Err(UserNotFoundError)`` if the user does not exist.

Infrastructure exceptions propagate unchanged.

change_user_password

async def change_user_password(
    user_id: str,
    current_password: str,
    new_password: str
) -> Result[None, InvalidCredentialsError | PasswordPolicyError]

source

Change a user’s password (requires current password).

Returns

Type	Description
Result[None, InvalidCredentialsError \| PasswordPolicyError]	``Ok(None)` `on success.` `Err(InvalidCredentialsError)` `if the current password is wrong or the user does not exist.` `Err(PasswordPolicyError)`` if the new password violates policy or has been used recently.

Infrastructure exceptions (e.g. DB connectivity) propagate unchanged.

set_user_password

async def set_user_password(
    user_id: str,
    new_password: str,
    force: bool = False
) -> None

source

Set a user’s password (admin operation).

list_users

async def list_users(
    skip: int = 0,
    limit: int = 100
) -> list[User]

source

List users with pagination.

count_users

async def count_users() -> int

source

Count total users.

shutdown

async def shutdown() -> None

source

Cancel and await all pending background event tasks.

`UserStatus`

source

User account status values.

Functions

`optional_auth`

optional_auth

def optional_auth(func: Callable) -> Callable

source

Decorator that attaches optional auth: user is populated if present, never blocked.

Can be used directly without invocation (unlike require_auth which always requires parentheses)

@optional_auth
async def public_endpoint(request):
    user = getattr(request.state, "user", None)
    ...

@optional_auth
async def public_endpoint(request):
    user = getattr(request.state, "user", None)
    ...

Parameters

Parameter	Type	Description
`func`	Callable	The async route handler to wrap.

Returns

Type	Description
Callable	The wrapped handler which always passes through.

`require_auth`

require_auth

def require_auth(
    roles: list[str] | None = None,
    permissions: list[str] | None = None,
    optional: bool = False
) -> Callable[[Callable], Callable]

source

Decorator that protects a route handler with authentication and RBAC/ABAC checks.

Parameters

Parameter	Type	Description
`roles`	list[str] \| None	Required role names. Any single matching role is sufficient.
`permissions`	list[str] \| None	Required permission strings. All must be satisfied.
`optional`	bool	When ``True, allow unauthenticated requests to proceed (user will simply be` `None` `in` `request.state``).

Returns

Type	Description
Callable[[Callable], Callable]	A decorator that wraps the route handler.

Raises

Exception	Description
ValueError	If no request object can be found in the handler's arguments.

`require_permissions`

require_permissions

def require_permissions(*permissions: str) -> Callable[[Callable], Callable]

source

Shorthand decorator requiring the user to hold all of *permissions*.

Parameters

Parameter	Type	Description

Returns

Type	Description
Callable[[Callable], Callable]	A decorator equivalent to ``require_auth(permissions=list(permissions))``.

`require_roles`

require_roles

def require_roles(*roles: str) -> Callable[[Callable], Callable]

source

Shorthand decorator requiring the user to have at least one of *roles*.

Parameters

Parameter	Type	Description

Returns

Type	Description
Callable[[Callable], Callable]	A decorator equivalent to ``require_auth(roles=list(roles))``.

Exceptions

`AlreadyVerifiedError`

source

Account is already verified.

__init__

def __init__(
    message: str = 'Account is already verified',
    user_id: str | None = None,
    **kwargs: Any
) -> None

source

`AuthError`

source

Base exception for all auth errors.

`AuthorizationError`

source

Raised when user lacks required permissions.

`InvalidTokenError`

source

Raised when a token is malformed or invalid.

`TokenAudienceError`

source

Token audience claim does not match expected.

__init__

def __init__(
    message: str = 'Token audience mismatch',
    expected: str | None = None,
    actual: str | None = None,
    **kwargs: Any
) -> None

source

`TokenBlacklistedError`

source

Token has been explicitly revoked.

__init__

def __init__(
    message: str = 'Token has been revoked',
    **kwargs: Any
) -> None

source

`TokenError`

source

Base exception for token-related errors.

`TokenExpiredError`

source

Raised when a token has expired.

__init__

def __init__(
    message: str = 'Token has expired',
    expiration_time: str | None = None,
    **kwargs: Any
) -> None

source

`TokenExpiredVerificationError`

source

Account verification has expired.

__init__

def __init__(
    message: str = 'Verification has expired',
    user_id: str | None = None,
    **kwargs: Any
) -> None

source

`TokenInvalidError`

source

Token is structurally invalid or has wrong type.

__init__

def __init__(
    message: str = 'Token is invalid',
    reason: str | None = None,
    **kwargs: Any
) -> None

source

`TokenNotFoundError`

source

Token record does not exist.

__init__

def __init__(
    message: str = 'Token not found',
    token_id: str | None = None,
    **kwargs: Any
) -> None

source

`VerificationError`

source

Base class for expected, recoverable account-verification failures.

All subtypes signal situations the caller should handle gracefully (e.g. redirect to a re-verification page or surface a user-facing error).

__init__

def __init__(
    message: str = 'Account verification error',
    **kwargs: Any
) -> None

source