API reference

Public package

Public package exports for Litestar Security.

class litestar_security.AssuranceRequirement(methods: frozenset[str] = frozenset({}), traits: frozenset[AssuranceTrait] = frozenset({}), max_age: timedelta | None = None, purpose: str | None = None)

Bases: object

Method, trait, freshness, and purpose observations required by a route.

__init__(methods: frozenset[str] = frozenset({}), traits: frozenset[AssuranceTrait] = frozenset({}), max_age: timedelta | None = None, purpose: str | None = None) → None

class litestar_security.AssuranceTrait(*values)

Bases: str, Enum

Normalized assurance properties established by verified evidence.

class litestar_security.Authenticated(claims: ClaimsT, evidence: AuthenticationEvidence, grants: AuthorizationSnapshot = <factory>, restrictions: CredentialRestrictions = <factory>)

Bases: Generic[ClaimsT]

Carry the typed result of successful credential verification.

__init__(claims: ClaimsT, evidence: AuthenticationEvidence, grants: AuthorizationSnapshot = <factory>, restrictions: CredentialRestrictions = <factory>) → None

class litestar_security.AuthenticationEvidence(mechanism: str, slot: str, authenticated_at: datetime, expires_at: datetime | None = None, methods: frozenset[str] = frozenset({}), traits: frozenset[str] = frozenset({}), acr: str | None = None, amr: tuple[str, ...] = ())

Bases: object

Normalized evidence emitted by one successful authenticator.

__init__(mechanism: str, slot: str, authenticated_at: datetime, expires_at: datetime | None = None, methods: frozenset[str] = frozenset({}), traits: frozenset[str] = frozenset({}), acr: str | None = None, amr: tuple[str, ...] = ()) → None

class litestar_security.AuthenticationMechanism(authenticator: CredentialVerifier[CredentialT, ClaimsT], resolver: IdentityResolver[ClaimsT, UserT], scheme_name: str | None = None, security_scheme: SecurityScheme | None = None, session_capable: bool = False)

Bases: Generic[CredentialT, ClaimsT, UserT]

Pair one slot authenticator with its identity resolver.

__init__(authenticator: CredentialVerifier[CredentialT, ClaimsT], resolver: IdentityResolver[ClaimsT, UserT], scheme_name: str | None = None, security_scheme: SecurityScheme | None = None, session_capable: bool = False) → None

class litestar_security.AuthenticationPolicy(*_args: object, **_kwargs: object)

Bases: object

Immutable closed request-authentication expression.

class litestar_security.AuthenticationRegistry(slots: Sequence[CredentialSlot[Any]] = (), mechanisms: Sequence[AuthenticationMechanism[Any, Any, UserT]] = (), authorization_resolver: AuthorizationResolver[UserT] | None = None, require_default: bool = False)

Bases: Generic[UserT]

Validate and compile deterministic credential-slot ownership.

property slot_names: tuple[str, ...]

Return normalized slot names in configuration order.

property mechanism_names: tuple[str, ...]

Return normalized mechanism names in configuration order.

property default_mechanism_names: tuple[str, ...]

Return default-participating mechanism names in configuration order.

get_slot(name: str) → CredentialSlot[Any]

Look up an owned slot by normalized name.

Parameters:

name – The slot name, normalized before lookup.

Returns:

The registered slot.

get_mechanism(name: str) → AuthenticationMechanism[Any, Any, UserT]

Look up a mechanism by normalized name.

Parameters:

name – The mechanism name, normalized before lookup.

Returns:

The registered mechanism.

get_mechanism_for_slot(name: str) → AuthenticationMechanism[Any, Any, UserT] | None

Look up the sole mechanism owning a normalized slot.

Parameters:

name – The slot name, normalized before lookup.

Returns:

The owning mechanism, or None when no mechanism claims the slot.

evaluator() → _AuthenticationEvaluator[UserT]

Create a stateless evaluator bound to this compiled registry.

Returns:

An evaluator that may be shared across requests.

__init__(slots: Sequence[CredentialSlot[Any]] = (), mechanisms: Sequence[AuthenticationMechanism[Any, Any, UserT]] = (), authorization_resolver: AuthorizationResolver[UserT] | None = None, require_default: bool = False) → None

class litestar_security.AuthorizationDecision(granted: bool, code: str = 'allowed', path: tuple[str, ...] = (), authentication_required: bool = False)

Bases: object

One predicate’s verdict, carrying why it was reached.

Parameters:

• granted – Whether the predicate allows the connection.

• code – Stable machine-readable reason, reported when access is denied.

• path – Predicate names from the outermost composite inward, for diagnostics.

• authentication_required – Deny because no principal is authenticated, which translates to 401 instead of 403.

prefixed(*parts: str) → AuthorizationDecision

Return this decision with parts prepended to its diagnostic path.

Parameters:

*parts – Names to record ahead of the existing path.

Returns:

An equivalent decision reached through the named enclosing predicates.

__init__(granted: bool, code: str = 'allowed', path: tuple[str, ...] = (), authentication_required: bool = False) → None

class litestar_security.AuthorizationPredicate

Bases: object

Immutable authorization decision exposed as a native Litestar guard.

decide(connection: ASGIConnection[Any, Any, Any, Any]) → AuthorizationDecision

Decide whether this predicate allows the connection.

Override this to add a predicate of your own; the built-in composites accept any subclass. Return a decision rather than raising, so a composite can report which branch denied access.

Parameters:

connection – The connection being authorized.

Returns:

This predicate’s verdict.

Raises:

NotImplementedError – If a subclass does not override this method.

class litestar_security.AuthorizationResolver(*args, **kwargs)

Bases: Protocol[_UserT]

Application-owned resolution of authorization for one verified principal.

async resolve(principal: Principal[_UserT]) → AuthorizationSnapshot | InvalidCredentials | VerificationUnavailable

Load one immutable application authorization snapshot.

Parameters:

principal – The same-subject principal established by authentication.

Returns:

The immutable application authorization snapshot; InvalidCredentials for an expected authorization denial; or VerificationUnavailable for expected dependency trouble.

Raises:

Exception – For an unexpected resolver error or outage. The evaluator catches this boundary in _resolve_authorization() and maps it to one sanitized 503 response.

__init__(*args, **kwargs)

class litestar_security.AuthorizationSnapshot(scopes: frozenset[str] = frozenset({}), roles: frozenset[str] = frozenset({}), capabilities: frozenset[str] = frozenset({}), team_roles: Mapping[str, frozenset[str]]=<factory>, tenant_ids: frozenset[str] = frozenset({}), resources: frozenset[ResourcePermission] = frozenset({}), attributes: Mapping[str, object]=<factory>)

Bases: object

Immutable application authorization data.

__init__(scopes: frozenset[str] = frozenset({}), roles: frozenset[str] = frozenset({}), capabilities: frozenset[str] = frozenset({}), team_roles: Mapping[str, frozenset[str]]=<factory>, tenant_ids: frozenset[str] = frozenset({}), resources: frozenset[ResourcePermission] = frozenset({}), attributes: Mapping[str, object]=<factory>) → None

class litestar_security.AuthorizationSnapshotRefresher(*args, **kwargs)

Bases: Protocol[UserT]

Application hook returning one detached immutable authorization snapshot.

async refresh(*, principal: Principal[UserT], previous: AuthorizationSnapshot, route_name: str) → AuthorizationSnapshot

Resolve and return a new detached authorization snapshot.

Parameters:

• principal – The authenticated principal for the connection.

• previous – The prior immutable snapshot, which is never mutated.

• route_name – The bound application route name.

Returns:

A new detached AuthorizationSnapshot; any other runtime type is treated as unavailable by connection lifetime supervision.

Raises:

Exception – When refresh fails. Connection lifetime supervision treats this as unavailable and closes the connection.

__init__(*args, **kwargs)

class litestar_security.BlockingIntegration(implementation: SyncT)

Bases: Generic[SyncT]

Mark one explicitly synchronous application integration for startup normalization.

Parameters:

implementation – The complete synchronous feature protocol.

__init__(implementation: SyncT) → None

class litestar_security.CSPMode(*values)

Bases: str, Enum

Select the browser CSP response-header mode.

class litestar_security.ContentSecurityPolicy(directives: Mapping[str, Sequence[str]], mode: CSPMode = CSPMode.ENFORCE, nonce_directives: Sequence[str] = ())

Bases: object

Define one explicit Content Security Policy.

Parameters:

• mode – Whether the policy enforces or only reports violations.

• directives – Explicit directive names and their ordered source values.

• nonce_directives – Directives that receive the response-local nonce.

Raises:

ImproperlyConfiguredException – If a directive or source is unsafe, or a nonce directive is absent from directives.

property header_name: str

Return the standard header name for this policy.

Returns:

The enforcing or report-only CSP header name.

serialize(*, nonce: str | None = None) → str

Serialize the policy deterministically.

Parameters:

nonce – Response-local nonce to append to configured directives.

Returns:

A CSP header value.

Raises:

ImproperlyConfiguredException – If nonce directives exist but no response nonce was supplied.

__init__(directives: Mapping[str, Sequence[str]], mode: CSPMode = CSPMode.ENFORCE, nonce_directives: Sequence[str] = ()) → None

class litestar_security.CredentialRestrictions(scopes: frozenset[str] | None = None, roles: frozenset[str] | None = None, capabilities: frozenset[str] | None = None, team_ids: frozenset[str] | None = None, tenant_ids: frozenset[str] | None = None, resources: frozenset[ResourcePermission] | None = None)

Bases: object

Authorization bounds imposed by one credential.

__init__(scopes: frozenset[str] | None = None, roles: frozenset[str] | None = None, capabilities: frozenset[str] | None = None, team_ids: frozenset[str] | None = None, tenant_ids: frozenset[str] | None = None, resources: frozenset[ResourcePermission] | None = None) → None

class litestar_security.CredentialSlot(*args, **kwargs)

Bases: Protocol[_CredentialT]

Synchronous, non-blocking credential extraction boundary.

extract(connection: ASGIConnection[Any, Any, Any, Any]) → NoCredentials | PresentedCredential[_CredentialT] | InvalidCredentials

Extract at most one credential from the connection.

Runs synchronously on every request, so it must not block or perform I/O.

Parameters:

connection – The incoming connection.

Returns:

The presented credential, NoCredentials when this slot is empty, or InvalidCredentials when the slot is malformed.

__init__(*args, **kwargs)

class litestar_security.CredentialVerifier(*args, **kwargs)

Bases: Protocol[_RequestCredentialT_contra, _ClaimsT]

Async credential verification boundary.

async authenticate(credential: _RequestCredentialT_contra, connection: ASGIConnection[Any, Any, Any, Any]) → NoCredentials | Authenticated[_ClaimsT] | InvalidCredentials | VerificationUnavailable

Verify a credential without resolving application identity.

Parameters:

• credential – The value produced by this authenticator’s slot.

• connection – The incoming connection.

Returns:

The verified claims, or a sanitized outcome describing why verification did not succeed.

__init__(*args, **kwargs)

class litestar_security.ExternalCSRF(name: str, validate: Callable[[str, str, AuthenticationPolicy], bool])

Bases: object

Declare a named application-owned CSRF coverage validator.

__init__(name: str, validate: Callable[[str, str, AuthenticationPolicy], bool]) → None

class litestar_security.IdentityResolver(*args, **kwargs)

Bases: Protocol[_ResolverClaimsT_contra, _UserT]

Async mapping from verified claims to one application principal.

async resolve(claims: _ResolverClaimsT_contra) → Principal[_UserT] | InvalidCredentials | VerificationUnavailable

Resolve verified claims into a principal or sanitized resolution outcome.

Parameters:

claims – The claims produced by the paired authenticator.

Returns:

The application principal; InvalidCredentials for an expected unknown or inactive identity; or VerificationUnavailable for expected dependency trouble.

Raises:

Exception – For an unexpected resolver error or outage. The evaluator catches this boundary in _resolve() and maps it to one sanitized 503 response.

__init__(*args, **kwargs)

class litestar_security.InMemoryWebSocketConnectTokenStore

Bases: object

Deterministic concurrency-safe connect token store for tests and examples.

property records: tuple[WebSocketConnectAuthorization, ...]

Return a stable snapshot of digest-only records.

async create(record: WebSocketConnectAuthorization) → None

Persist one record while rejecting duplicate public IDs.

async consume(*, connect_token_id: str, digest: bytes, now: datetime) → WebSocketConnectAuthorization | None

Atomically return and delete one matching unexpired record.

__init__() → None

class litestar_security.InvalidCredentials(code: str = 'invalid_credentials')

Bases: object

Indicate that a presented credential cannot authenticate.

__init__(code: str = 'invalid_credentials') → None

class litestar_security.IssuedWebSocketConnectToken(value: str, expires_at: datetime)

Bases: object

Reveal-once WebSocket connect token value.

__init__(value: str, expires_at: datetime) → None

class litestar_security.LitestarSessionHandle(scope: HTTPScope | WebSocketScope)

Bases: object

Live view over Litestar’s native session scope value.

property is_available: bool

Return whether native session middleware attached state.

property can_persist: bool

Return whether this connection permits session mutation.

get(key: str, default: object = None) → object

Read the current native session mapping.

Parameters:

• key – The session key to read.

• default – The value to return when the key is absent.

Returns:

The stored value, or default.

set(key: str, value: object) → None

Store a value when the native session can persist.

Parameters:

• key – The session key to write.

• value – The value to store.

pop(key: str, default: object = None) → object

Remove a value when the native session can persist.

Parameters:

• key – The session key to remove.

• default – The value to return when the key is absent.

Returns:

The removed value, or default.

clear() → None

Clear the native session when it can persist.

__init__(scope: HTTPScope | WebSocketScope) → None

class litestar_security.MFAConfig(store: object, secret_protector: object, policy: TOTPPolicy | None = None, recovery_peppers: Sequence[RecoveryCodePepper] = (), login_methods: LoginMethodStore | None = None, events: SecurityEventSink | None = None, step_up_store: object | None = None, require_at_login: bool = False, login_challenge_store: object | None = None, route_prefix: str = '/auth', issuer: str = 'Litestar Security', register_routes: bool = True, docs: RouteDocs = <factory>)

Bases: object

Configure MFA capabilities without selecting a persistence technology.

__init__(store: object, secret_protector: object, policy: TOTPPolicy | None = None, recovery_peppers: Sequence[RecoveryCodePepper] = (), login_methods: LoginMethodStore | None = None, events: SecurityEventSink | None = None, step_up_store: object | None = None, require_at_login: bool = False, login_challenge_store: object | None = None, route_prefix: str = '/auth', issuer: str = 'Litestar Security', register_routes: bool = True, docs: RouteDocs = <factory>) → None

class litestar_security.MechanismRequirement(name: str, scopes: tuple[str, ...] = ())

Bases: object

Select one configured mechanism and optional provider scopes.

__init__(name: str, scopes: tuple[str, ...] = ()) → None

class litestar_security.NoCredentials

Bases: object

Indicate that an owned slot contains no credential.

__init__() → None

class litestar_security.NullSessionHandle

Bases: object

Stateless session capability for applications without sessions.

property is_available: bool

Return that no native session is attached.

property can_persist: bool

Return that no session mutations can persist.

get(key: str, default: object = None) → object

Return the caller’s default.

Parameters:

• key – The session key to read.

• default – The value to return when the key is absent.

Returns:

The stored value, or default.

set(key: str, value: object) → None

Reject writes when session storage is unavailable.

Parameters:

• key – Ignored; no session is attached.

• value – Ignored; no session is attached.

Raises:

SessionUnavailableError – Always, because no session is attached.

pop(key: str, default: object = None) → object

Return the caller’s default without retaining state.

Parameters:

• key – The session key to remove.

• default – The value to return when the key is absent.

Returns:

The removed value, or default.

clear() → None

Reject clearing when session storage is unavailable.

__init__() → None

class litestar_security.PasskeyConfig(store: object, challenge_store: object, rp_id: str, origins: ~collections.abc.Sequence[str], rp_name: str = 'Litestar Security', algorithms: ~collections.abc.Sequence[int] = (-8, -7, -257), challenge_ttl: ~datetime.timedelta = datetime.timedelta(seconds=300), allow_insecure_localhost: bool = False, worker_timeout: float = 10.0, attestation_trust: AttestationTrustMapper | None = None, login_methods: LoginMethodStore | None = None, events: SecurityEventSink | None = None, step_up_store: object | None = None, route_prefix: str = '/auth', register_routes: bool = True, docs: ~litestar_security._docs.RouteDocs = <factory>)

Bases: object

Configure exact WebAuthn relying-party and persistence boundaries.

__init__(store: object, challenge_store: object, rp_id: str, origins: ~collections.abc.Sequence[str], rp_name: str = 'Litestar Security', algorithms: ~collections.abc.Sequence[int] = (-8, -7, -257), challenge_ttl: ~datetime.timedelta = datetime.timedelta(seconds=300), allow_insecure_localhost: bool = False, worker_timeout: float = 10.0, attestation_trust: AttestationTrustMapper | None = None, login_methods: LoginMethodStore | None = None, events: SecurityEventSink | None = None, step_up_store: object | None = None, route_prefix: str = '/auth', register_routes: bool = True, docs: ~litestar_security._docs.RouteDocs = <factory>) → None

class litestar_security.PresentedCredential(value: CredentialT)

Bases: Generic[CredentialT]

A credential extracted from one owned request slot.

__init__(value: CredentialT) → None

class litestar_security.Principal(id: str | None, display_name: str | None = None, user: UserT | None = None)

Bases: Generic[UserT]

Stable identity envelope for anonymous, user, and service actors.

classmethod anonymous() → Principal[UserT]

Create an anonymous principal.

Returns:

A principal with no identity, used before authentication runs.

property is_authenticated: bool

Return whether this principal has an authenticated identity.

property has_user: bool

Return whether an application user is attached.

require_user() → UserT

Return the application user or fail without revealing actor state.

Returns:

The attached application user.

Raises:

NotAuthorizedException – If no user is attached. The message never distinguishes an anonymous caller from an authenticated one whose user could not be loaded.

__init__(id: str | None, display_name: str | None = None, user: UserT | None = None) → None

class litestar_security.ProblemDetail(status: int, title: str, detail: str, extra: dict[str, Any] | list[Any] | None = None)

Bases: WireStruct

The body a denial takes when the application converts every HTTP exception.

An application installing Litestar’s problem-details plugin with enable_for_all_http_exceptions=True replaces RouteError on every raised status, and the response is served as application/problem+json.

These are the members Litestar’s conversion actually emits, which is not the RFC 9457 five-member shape: the raised detail is moved onto title and detail falls back to the HTTP reason phrase, while type and instance are never produced. Unknown members are tolerated both because RFC 9457 permits extension members and because the sender is Litestar rather than this library.

status: int

The HTTP status, repeated in the body.

title: str

The raised explanation, which the conversion moves here from detail.

detail: str

The HTTP reason phrase, which the conversion leaves as the default.

extra: dict[str, Any] | list[Any] | None

Structured context the raised exception carried, carried through unchanged.

class litestar_security.PublicController(owner: Router)

Bases: SecureController

SecureController whose default policy skips authentication.

after_request

A sync or async function executed before a Request is passed to any route handler.

If this function returns a value, the request will not reach the route handler, and instead this value will be used.

after_response

A sync or async function called after the response has been awaited.

It receives the Request instance and should not return any values.

before_request

A sync or async function called immediately before calling the route handler.

It receives the Request instance and any non-None return value is used for the response, bypassing the route handler.

cache_control

A CacheControlHeader header to add to route handlers of this controller.

Can be overridden by route handlers.

dependencies

A string keyed dictionary of dependency Provider instances.

dto

AbstractDTO to use for (de)serializing and validation of request data.

etag

An etag header of type ETag to add to route handlers of this controller.

Can be overridden by route handlers.

exception_handlers

A map of handler functions to status codes and/or exception types.

guards

A sequence of Guard callables.

include_in_schema

A boolean flag dictating whether the route handler should be documented in the OpenAPI schema

middleware

A sequence of Middleware.

opt

A string key mapping of arbitrary values that can be accessed in Guards or wherever you have access to Request or ASGI Scope.

owner

The Router or Litestar app that owns the controller.

This value is set internally by Litestar and it should not be set when subclassing the controller.

parameters

A mapping of Parameter definitions available to all application paths.

path

A path fragment for the controller.

All route handlers under the controller will have the fragment appended to them. If not set it defaults to /.

request_class

A custom subclass of Request to be used as the default request for all route handlers under the controller.

request_max_body_size

Maximum allowed size of the request body in bytes. If this size is exceeded, a ‘413 - Request Entity Too Large’ error response is returned.

response_class

A custom subclass of Response to be used as the default response for all route handlers under the controller.

response_cookies

A list of Cookie instances.

response_headers

A string keyed dictionary mapping ResponseHeader instances.

return_dto

AbstractDTO to use for serializing outbound response data.

security

A sequence of dictionaries that to the schema of all route handlers under the controller.

signature_namespace

A mapping of names to types for use in forward reference resolution during signature modelling.

signature_types

A sequence of types for use in forward reference resolution during signature modelling.

These types will be added to the signature namespace using their __name__ attribute.

tags

A sequence of string tags that will be appended to the schema of all route handlers under the controller.

type_decoders

A sequence of tuples, each composed of a predicate testing for type identity and a msgspec hook for deserialization.

type_encoders

A mapping of types to callables that transform them into types supported for serialization.

websocket_class

A custom subclass of WebSocket to be used as the default websocket for all route handlers under the controller.

class litestar_security.RaisedErrorSchema(schema: type[object], media_type: str)

Bases: object

Declare how application exception handling renders raised route errors.

Parameters:

• schema – The body type application exception handlers serialize.

• media_type – The response content type used for that serialized body.

Raises:

ImproperlyConfiguredException – If schema is not a type or media_type is not a non-blank string.

__init__(schema: type[object], media_type: str) → None

class litestar_security.ResourcePermission(resource: str, scopes: frozenset[str] = frozenset({}))

Bases: object

Credential or application permission scoped to one resource.

__init__(resource: str, scopes: frozenset[str] = frozenset({})) → None

class litestar_security.RouteDocs(tags: ~collections.abc.Mapping[str, str] = <factory>, tag_descriptions: ~collections.abc.Mapping[str, str] = <factory>, operation_id: ~collections.abc.Callable[[str], str] | None = None, route_name: ~collections.abc.Callable[[str], str] | None = None)

Bases: object

Application-owned OpenAPI documentation for generated routes.

Tag groups are addressed by the stable keys of ROUTE_TAGS, never by their display names, so an application that renames a group keeps addressing it by the key it already used. None of this is security policy: an application cannot change what a route requires by documenting it differently.

tags: Mapping[str, str]

Replacement display name per stable tag-group key.

tag_descriptions: Mapping[str, str]

Replacement description per stable tag-group key.

operation_id: Callable[[str], str] | None

Rewrite every generated operation_id, which names the generated client function.

route_name: Callable[[str], str] | None

Rewrite every generated route name, which route_reverse resolves.

__init__(tags: ~collections.abc.Mapping[str, str] = <factory>, tag_descriptions: ~collections.abc.Mapping[str, str] = <factory>, operation_id: ~collections.abc.Callable[[str], str] | None = None, route_name: ~collections.abc.Callable[[str], str] | None = None) → None

class litestar_security.RouteError(status_code: int, detail: str, extra: dict[str, Any] | list[Any] | None = None)

Bases: WireStruct

The body a generated route sends when it raises rather than returns.

A denial - 400, 401, 429, 503, and the OAuth 409 - reaches the wire through Litestar’s exception handling, not through the handler’s return value, so the body is ExceptionResponseContent: the status repeated inside the payload, a human-readable detail, and extra when the raised exception carries structured context. A request-validation failure always carries one, as a list of {message, key, source} entries, so extra is a member clients see in practice rather than a theoretical one.

Distinguish this from OperationMessage, which is the body a handler returns - the 200 confirmations and the 409 conflict. The two are separate schemas because the distinction that decides the shape is raised-versus-returned, not error-versus-success.

Unknown members are tolerated here, against the base policy, because the sender is Litestar: an application handler may add its own members and a future Litestar may too, and neither is a reason for a client of this library to fail decoding.

status_code: int

The HTTP status, repeated in the body by Litestar’s exception handling.

detail: str

A human-readable explanation that never names an account.

extra: dict[str, Any] | list[Any] | None

Structured context the raised exception carried, when it carried any.

class litestar_security.SecureController(owner: Router)

Bases: Controller

Controller base whose typed auth attribute compiles into opt.

after_request

A sync or async function executed before a Request is passed to any route handler.

If this function returns a value, the request will not reach the route handler, and instead this value will be used.

after_response

A sync or async function called after the response has been awaited.

It receives the Request instance and should not return any values.

before_request

A sync or async function called immediately before calling the route handler.

It receives the Request instance and any non-None return value is used for the response, bypassing the route handler.

cache_control

A CacheControlHeader header to add to route handlers of this controller.

Can be overridden by route handlers.

dependencies

A string keyed dictionary of dependency Provider instances.

dto

AbstractDTO to use for (de)serializing and validation of request data.

etag

An etag header of type ETag to add to route handlers of this controller.

Can be overridden by route handlers.

exception_handlers

A map of handler functions to status codes and/or exception types.

guards

A sequence of Guard callables.

include_in_schema

A boolean flag dictating whether the route handler should be documented in the OpenAPI schema

middleware

A sequence of Middleware.

opt

A string key mapping of arbitrary values that can be accessed in Guards or wherever you have access to Request or ASGI Scope.

owner

The Router or Litestar app that owns the controller.

This value is set internally by Litestar and it should not be set when subclassing the controller.

parameters

A mapping of Parameter definitions available to all application paths.

path

A path fragment for the controller.

All route handlers under the controller will have the fragment appended to them. If not set it defaults to /.

request_class

A custom subclass of Request to be used as the default request for all route handlers under the controller.

request_max_body_size

Maximum allowed size of the request body in bytes. If this size is exceeded, a ‘413 - Request Entity Too Large’ error response is returned.

response_class

A custom subclass of Response to be used as the default response for all route handlers under the controller.

response_cookies

A list of Cookie instances.

response_headers

A string keyed dictionary mapping ResponseHeader instances.

return_dto

AbstractDTO to use for serializing outbound response data.

security

A sequence of dictionaries that to the schema of all route handlers under the controller.

signature_namespace

A mapping of names to types for use in forward reference resolution during signature modelling.

signature_types

A sequence of types for use in forward reference resolution during signature modelling.

These types will be added to the signature namespace using their __name__ attribute.

tags

A sequence of string tags that will be appended to the schema of all route handlers under the controller.

type_decoders

A sequence of tuples, each composed of a predicate testing for type identity and a msgspec hook for deserialization.

type_encoders

A mapping of types to callables that transform them into types supported for serialization.

websocket_class

A custom subclass of WebSocket to be used as the default websocket for all route handlers under the controller.

class litestar_security.SecurityConfig(slots: Sequence[CredentialSlot[Any]] = (), mechanisms: Any, ~typing.Any, ~litestar_security.config.UserT]]=(), max_openapi_combinations: int = 32, external_csrf: ExternalCSRF | None = None, exclude: Sequence[str] | str | None = None, require_default: bool = False, local_auth: LocalAuthConfig[UserT] | None = None, local_jwks: LocalJWKSConfig | None = None, oauth: OAuthConfig | None = None, protected_resource: ProtectedResourceConfig | None = None, mfa: MFAConfig | None = None, passkeys: PasskeyConfig | None = None, api_key: APIKeyConfig | None = None, iap: GoogleIAPConfig[UserT] | None = None, service_token: ServiceTokenConfig | None = None, headers: SecurityHeadersConfig | None = None, websocket: WebSocketSecurityConfig = <factory>, authorization_resolver: AuthorizationResolver[UserT] | None = None, jwks_providers: Sequence[JWKSProvider] = (), jwks_warmup_failure: Literal['fail_startup', 'lazy']='fail_startup', wire_rename: RenameStrategy | None = None, wire_forbid_unknown_fields: bool = True, raised_error_schema: RaisedErrorSchema | None = None)

Bases: Generic[UserT]

Configure the per-application security runtime.

exclude: Sequence[str] | str | None

Regular expressions matched against a route path to exclude it from security.

Mirrors JWTAuth.exclude: a single pattern or a sequence of patterns, joined into one expression and compiled with re. A pattern is anchored at the start of the route path, so "^/static" and "/static" both exclude /static/{file_path:path} while a bare "static" does not.

Exclusion is total and applies when the route is compiled, not per request: an excluded route is never authenticated, carries no principal, and contributes an anonymous security requirement to OpenAPI rather than the configured schemes. A route that declares its own auth= and also matches a pattern is a contradiction and is rejected at startup.

protected_resource: ProtectedResourceConfig | None

Describe this application as an OAuth 2.1 protected resource.

When set, the plugin publishes the RFC 9728 metadata document at /.well-known/oauth-protected-resource so an authorization server or a client can discover which issuers this resource trusts, which scopes it understands, and how a bearer token may be presented to it. The route is unauthenticated, as the specification requires.

wire_rename: RenameStrategy | None

How generated request and response members are spelled on the wire.

None keeps the field names as Python spells them, which is snake_case. Any of "lower", "upper", "camel", "pascal", and "kebab" selects a named strategy, and a Callable[[str], str] covers a house convention outside those five. The choice reaches the OpenAPI document as well as the wire, so a generated client follows it without further work.

A handful of generated schemas opt out because their member names belong to a specification rather than to this library - the RFC 6749 token response, the OIDC back-channel logout form, and the bodies Litestar’s own exception handling renders.

wire_forbid_unknown_fields: bool

Whether an unrecognized member in a request body is a decoding error.

Strictness applies to decoding, so it constrains request schemas only. Rejecting the unknown member is what keeps a stale or misspelled optional field from resolving to its default and producing a wrong but successful request.

__init__(slots: Sequence[CredentialSlot[Any]] = (), mechanisms: Any, ~typing.Any, ~litestar_security.config.UserT]]=(), max_openapi_combinations: int = 32, external_csrf: ExternalCSRF | None = None, exclude: Sequence[str] | str | None = None, require_default: bool = False, local_auth: LocalAuthConfig[UserT] | None = None, local_jwks: LocalJWKSConfig | None = None, oauth: OAuthConfig | None = None, protected_resource: ProtectedResourceConfig | None = None, mfa: MFAConfig | None = None, passkeys: PasskeyConfig | None = None, api_key: APIKeyConfig | None = None, iap: GoogleIAPConfig[UserT] | None = None, service_token: ServiceTokenConfig | None = None, headers: SecurityHeadersConfig | None = None, websocket: WebSocketSecurityConfig = <factory>, authorization_resolver: AuthorizationResolver[UserT] | None = None, jwks_providers: Sequence[JWKSProvider] = (), jwks_warmup_failure: Literal['fail_startup', 'lazy']='fail_startup', wire_rename: RenameStrategy | None = None, wire_forbid_unknown_fields: bool = True, raised_error_schema: RaisedErrorSchema | None = None) → None

raised_error_schema: RaisedErrorSchema | None

The body type and media type application handlers use for raised errors.

Generated routes raise their denial statuses through the application’s exception handlers. Set this when those handlers render a body other than Litestar’s default RouteError. The declaration changes only generated-route OpenAPI response specifications; it does not install an exception handler or alter runtime responses.

wire_policy() → WirePolicy

Return the wire convention every generated route body is built with.

Returns:

The casing strategy and unknown-field policy as one hashable value.

Raises:

ImproperlyConfiguredException – If the strategy is neither one of the named strategies nor a callable, or the unknown-field policy is not boolean.

class litestar_security.SecurityContext(session: SessionHandle, evidence: tuple[~litestar_security.context.AuthenticationEvidence, ...]=(), authorization: AuthorizationSnapshot = <factory>, restrictions: tuple[~litestar_security.context.CredentialRestrictions, ...]=())

Bases: object

Authentication evidence, authorization, and optional session capability.

property expires_at: datetime | None

Return the earliest bounded evidence expiry.

__init__(session: SessionHandle, evidence: tuple[~litestar_security.context.AuthenticationEvidence, ...]=(), authorization: AuthorizationSnapshot = <factory>, restrictions: tuple[~litestar_security.context.CredentialRestrictions, ...]=()) → None

class litestar_security.SecurityHeadersConfig(static: Mapping[str, str]=<factory>, csp: ContentSecurityPolicy | None = None)

Bases: object

Configure native static response headers and optional CSP.

Parameters:

• static – Application-owned static security header values.

• csp – Optional explicit Content Security Policy.

Raises:

ImproperlyConfiguredException – If names or values are unsafe, or a configured CSP header conflicts with csp.

classmethod hardened() → SecurityHeadersConfig

Create the recommended opt-in static browser-security baseline.

Returns:

A configuration with HSTS, frame, content-type, and referrer protections. Callers may supply more restrictive per-route headers.

__init__(static: Mapping[str, str]=<factory>, csp: ContentSecurityPolicy | None = None) → None

class litestar_security.SecurityPlugin(config: SecurityConfig[UserT] | None = None)

Bases: InitPlugin, ReceiveRoutePlugin, CLIPlugin, Generic[UserT]

Expose the Litestar Security configuration and CLI integration points.

__init__(config: SecurityConfig[UserT] | None = None) → None

Initialize the plugin.

on_app_init(app_config: AppConfig) → AppConfig

Validate ownership and install one typed security runtime.

Parameters:

app_config – The application configuration to extend.

Returns:

The same configuration, with the security middleware, dependencies, generated routes, and OpenAPI contributions installed.

Raises:

ImproperlyConfiguredException – If the application already owns something this plugin must own, such as a competing session middleware, CSRF config, or reserved dependency name.

receive_route(route: BaseRoute) → None

Compile every initial or dynamically registered route.

Parameters:

route – The route Litestar just registered.

Raises:

ImproperlyConfiguredException – If called before application initialization, or if the route’s declared policy cannot compile.

on_cli_init(cli: Group) → None

Attach the security command group to the Litestar CLI.

Parameters:

cli – The root Litestar CLI group.

class litestar_security.SessionHandle(*args, **kwargs)

Bases: Protocol

Uniform access to an optional native Litestar session.

property is_available: bool

Return whether a session is attached.

property can_persist: bool

Return whether session mutations can persist.

get(key: str, default: object = None) → object

Read a session value.

Parameters:

• key – The session key to read.

• default – The value to return when the key is absent.

Returns:

The stored value, or default.

set(key: str, value: object) → None

Store a session value.

Parameters:

• key – The session key to write.

• value – The value to store.

pop(key: str, default: object = None) → object

Remove and return a session value.

Parameters:

• key – The session key to remove.

• default – The value to return when the key is absent.

Returns:

The removed value, or default.

clear() → None

Remove all session values.

__init__(*args, **kwargs)

exception litestar_security.SessionPersistenceUnavailableError

Bases: SessionUnavailableError

Raised when attached session state is read-only.

__init__() → None

Initialize the stable public error.

exception litestar_security.SessionUnavailableError

Bases: RuntimeError

Raised when no native session storage is attached.

__init__() → None

Initialize the stable public error.

class litestar_security.VerificationUnavailable(code: str = 'verification_unavailable', retry_after: int | None = None)

Bases: object

Indicate that a verifier cannot make a trustworthy decision.

__init__(code: str = 'verification_unavailable', retry_after: int | None = None) → None

class litestar_security.WebSocketBinding(connection_id: str, subject_id: str, credential_ids: frozenset[str], session_id: str | None, route_name: str)

Bases: object

Secret-free identity and route binding supplied to revocation hooks.

__init__(connection_id: str, subject_id: str, credential_ids: frozenset[str], session_id: str | None, route_name: str) → None

class litestar_security.WebSocketCloseCodes(unauthenticated: int = 4401, unauthorized: int = 4403, verification_unavailable: int = 1013)

Bases: object

Map stable security outcomes to WebSocket close codes.

__init__(unauthenticated: int = 4401, unauthorized: int = 4403, verification_unavailable: int = 1013) → None

class litestar_security.WebSocketConnectAuthorization(connect_token_id: str, digest: bytes, subject_id: str, security_epoch: int, route_name: str, origin: str, restrictions: CredentialRestrictions, policy_fingerprint: str, issued_at: datetime, expires_at: datetime)

Bases: object

Storage-safe one-time connect token binding containing no recoverable value.

__init__(connect_token_id: str, digest: bytes, subject_id: str, security_epoch: int, route_name: str, origin: str, restrictions: CredentialRestrictions, policy_fingerprint: str, issued_at: datetime, expires_at: datetime) → None

class litestar_security.WebSocketConnectTokenIssuer(app: Litestar, store: WebSocketConnectTokenStore, clock: Callable[[], ~datetime.datetime]=<function WebSocketConnectTokenIssuer.<lambda>>, ttl: timedelta = datetime.timedelta(seconds=30))

Bases: object

Mint one-time WebSocket connect tokens by route name.

async issue(route_name: str, *, principal: Principal[Any], context: SecurityContext, origin: str, security_epoch: int, restrictions: CredentialRestrictions | None = None, ttl: timedelta | None = None) → IssuedWebSocketConnectToken

Resolve one route name to its compiled plan and mint a connect token.

Parameters:

• route_name – The registered Litestar route handler name.

• principal – The authenticated principal minting the connect token.

• context – The current request’s security context.

• origin – The exact canonical Origin the connect token is bound to.

• security_epoch – The authoritative non-negative epoch bound to the token.

• restrictions – Optional narrowed authorization restrictions.

• ttl – Optional override for the configured connect token lifetime.

Returns:

The reveal-once issued connect token.

Raises:

ImproperlyConfiguredException – If the route name does not resolve to a registered WebSocket handler with a compiled runtime plan.

__init__(app: Litestar, store: WebSocketConnectTokenStore, clock: Callable[[], ~datetime.datetime]=<function WebSocketConnectTokenIssuer.<lambda>>, ttl: timedelta = datetime.timedelta(seconds=30)) → None

class litestar_security.WebSocketConnectTokenService(store: ~litestar_security.websocket._connect_tokens.WebSocketConnectTokenStore, ttl: ~datetime.timedelta = datetime.timedelta(seconds=30), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function WebSocketConnectTokenService.<lambda>>, entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>)

Bases: object

Issue and atomically consume exact one-handshake connect token bindings.

async issue(*, principal: Principal[Any], context: SecurityContext, route_name: str, origin: str, policy_fingerprint: str, security_epoch: int, restrictions: CredentialRestrictions | None = None) → IssuedWebSocketConnectToken

Issue one digest-only, exact-route connect token for an authenticated context.

async consume(value: object, *, route_name: str, origin: str, policy_fingerprint: str, current_security_epoch: Callable[[str], Awaitable[int | None]]) → WebSocketConnectAuthorization | None

Atomically consume a connect token before authoritative epoch and route checks.

__init__(store: ~litestar_security.websocket._connect_tokens.WebSocketConnectTokenStore, ttl: ~datetime.timedelta = datetime.timedelta(seconds=30), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function WebSocketConnectTokenService.<lambda>>, entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>) → None

class litestar_security.WebSocketConnectTokenStore(*args, **kwargs)

Bases: Protocol

Application-owned atomic persistence port for one-time connect tokens.

async create(record: WebSocketConnectAuthorization) → None

Persist one new digest-only record, rejecting duplicate IDs.

async consume(*, connect_token_id: str, digest: bytes, now: datetime) → WebSocketConnectAuthorization | None

Atomically return and delete one matching unexpired record.

__init__(*args, **kwargs)

class litestar_security.WebSocketRevocationSource(*args, **kwargs)

Bases: Protocol

Event-driven, secret-free application hook for one binding’s revocation.

async wait(binding: WebSocketBinding) → None

Block without polling until the supplied connection binding is revoked.

Parameters:

binding – The secret-free identity and route binding to supervise.

Returns:

None only after a genuine revocation of binding.

Raises:

Exception – When supervision fails. The connection lifetime treats this as unavailable and closes the connection.

__init__(*args, **kwargs)

class litestar_security.WebSocketSecurityConfig(allowed_origins: frozenset[str] = frozenset({}), connect_token_store: ~litestar_security.websocket._connect_tokens.WebSocketConnectTokenStore | None = None, connect_token_ttl: ~datetime.timedelta = datetime.timedelta(seconds=30), maximum_connect_token_ttl: ~datetime.timedelta = datetime.timedelta(seconds=120), connect_token_query_parameter: str = 'connect_token', current_security_epoch: ~collections.abc.Callable[[str], ~collections.abc.Awaitable[int | None]] | None = None, refresh_interval: ~datetime.timedelta | None = None, snapshot_refresher: ~litestar_security.websocket._bindings.AuthorizationSnapshotRefresher[~typing.Any] | None = None, revocation_source: ~litestar_security.websocket._bindings.WebSocketRevocationSource | None = None, close_codes: ~litestar_security.websocket._config.WebSocketCloseCodes = WebSocketCloseCodes(unauthenticated=4401, unauthorized=4403, verification_unavailable=1013), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function WebSocketSecurityConfig.<lambda>>, sleeper: ~collections.abc.Callable[[float], ~collections.abc.Awaitable[None]] = <function sleep>)

Bases: object

Configure WebSocket transport validation and optional lifetime hooks.

__init__(allowed_origins: frozenset[str] = frozenset({}), connect_token_store: ~litestar_security.websocket._connect_tokens.WebSocketConnectTokenStore | None = None, connect_token_ttl: ~datetime.timedelta = datetime.timedelta(seconds=30), maximum_connect_token_ttl: ~datetime.timedelta = datetime.timedelta(seconds=120), connect_token_query_parameter: str = 'connect_token', current_security_epoch: ~collections.abc.Callable[[str], ~collections.abc.Awaitable[int | None]] | None = None, refresh_interval: ~datetime.timedelta | None = None, snapshot_refresher: ~litestar_security.websocket._bindings.AuthorizationSnapshotRefresher[~typing.Any] | None = None, revocation_source: ~litestar_security.websocket._bindings.WebSocketRevocationSource | None = None, close_codes: ~litestar_security.websocket._config.WebSocketCloseCodes = WebSocketCloseCodes(unauthenticated=4401, unauthorized=4403, verification_unavailable=1013), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function WebSocketSecurityConfig.<lambda>>, sleeper: ~collections.abc.Callable[[float], ~collections.abc.Awaitable[None]] = <function sleep>) → None

class litestar_security.WireStruct

Bases: Struct

Base for every generated-route wire schema, and the default it is spelled in.

Field names reach the wire exactly as they are spelled in Python, and an unrecognized member is a decoding error rather than a silently discarded key. Rejecting the unknown member is what keeps a stale or misspelled optional field from resolving to its default and producing a wrong but successful request.

That is the default rather than a fixed policy. An application chooses the convention through SecurityConfig.wire_rename and wire_forbid_unknown_fields, and the generated routes carry the choice into the request body, the response body, and the OpenAPI schema together. A schema declares what it is called here; the configuration decides how it is spelled.

Subclasses restate frozen=True:

class LocalCredentials(WireStruct, frozen=True):
    '''Password credentials accepted by generated login handlers.'''

That keyword is redundant at runtime, because msgspec inherits the struct configuration, and required by the type checkers, which read immutability from the class keywords rather than from the base. Keeping the base frozen anyway means a subclass that omits the keyword is still immutable in fact, which is the safer direction for the mistake to fall.

Strictness applies to decoding, so it constrains request schemas only; response schemas inherit it inertly. A schema that must tolerate members it does not model - a specification-defined body whose sender may legitimately add them - overrides the policy for itself and records why:

class BackchannelLogout(WireStruct, frozen=True, forbid_unknown_fields=False):
    '''The specification permits unrecognized members.'''

Prefer that per-schema override to relaxing this base: it keeps the safe default intact and leaves the reason beside the schema that needs it.

litestar_security.all_of(*requirements: str | MechanismRequirement) → AuthenticationPolicy

Require every named authentication mechanism.

Parameters:

*requirements – Mechanism names or requirements that must all succeed.

Returns:

A policy satisfied only when every participant succeeds.

litestar_security.any_of(*requirements: str | MechanismRequirement) → AuthenticationPolicy

Require at least one named authentication mechanism.

Parameters:

*requirements – Mechanism names or requirements to accept.

Returns:

A policy satisfied by any one participant.

litestar_security.at_least(count: int, *requirements: str | MechanismRequirement) → AuthenticationPolicy

Require a positive threshold of named authentication mechanisms.

Parameters:

• count – How many participants must succeed.

• *requirements – Mechanism names or requirements to draw from.

Returns:

A policy satisfied by any count of the participants.

Raises:

ImproperlyConfiguredException – If the count is not between one and the number of participants.

litestar_security.exclude() → AuthenticationPolicy

Bypass request authentication while preserving the default CSRF policy.

Returns:

A policy that skips credential extraction and authentication.

async litestar_security.issue_websocket_connect_token(*, principal: Principal[Any], context: SecurityContext, route_name: str, origin: str, policy_fingerprint: str, security_epoch: int, restrictions: CredentialRestrictions, store: WebSocketConnectTokenStore, clock: Callable[[], datetime], ttl: timedelta = datetime.timedelta(seconds=30)) → IssuedWebSocketConnectToken

Issue one reveal-once WebSocket connect token through an application store.

Parameters:

• principal – The authenticated principal the connect token speaks for.

• context – The security context the connect token is bound to.

• route_name – The single route the connect token authorizes; it is valid nowhere else.

• origin – The exact origin the handshake must present.

• policy_fingerprint – The compiled policy binding the handshake revalidates.

• security_epoch – The authoritative non-negative epoch bound to the token.

• restrictions – The credential restrictions carried into the connection.

• store – The application store that persists the digest-only record.

• clock – The timezone-aware clock used for issuance and expiry.

• ttl – How long the connect token stays valid, bounded by the two-minute maximum.

Returns:

The issued connect token, whose reveal-once value is not recoverable from the stored record.

Raises:

ValueError – If the principal is unauthenticated, the context is not a SecurityContext, or any binding fails validation.

litestar_security.mechanism(name: str, *scopes: str) → MechanismRequirement

Select a named mechanism and its requested OAuth or OIDC scopes.

Parameters:

• name – The configured mechanism name.

• *scopes – Provider scopes to request. Only OAuth and OIDC schemes accept these.

Returns:

The requirement, for use inside a policy expression.

litestar_security.optional(policy: AuthenticationPolicy) → AuthenticationPolicy

Allow anonymous access only when a positive policy sees no credential.

A presented-but-invalid credential is still rejected: optional means the route tolerates absence, not failure.

Parameters:

policy – The positive policy to apply when a credential is present.

Returns:

A policy that admits anonymous callers alongside authenticated ones.

Raises:

ImproperlyConfiguredException – If the policy is public or already optional.

litestar_security.public() → AuthenticationPolicy

Deliberately skip request credential verification.

Returns:

A policy that authenticates nothing, leaving the anonymous principal in place.

litestar_security.required(*requirements: str | MechanismRequirement) → AuthenticationPolicy

Require an explicit OR expression or the implicit default participants.

Parameters:

*requirements – Mechanism names or requirements. Passing none requires any mechanism that participates by default.

Returns:

A policy that rejects a request presenting no accepted credential.

litestar_security.requires_all_of(*children: AuthorizationPredicate) → AuthorizationPredicate

Require every child predicate.

Parameters:

*children – The predicates that must all be satisfied.

Returns:

The composed predicate.

litestar_security.requires_any_of(*children: AuthorizationPredicate) → AuthorizationPredicate

Require at least one child predicate.

Parameters:

*children – The predicates to draw from.

Returns:

The composed predicate.

litestar_security.requires_assurance(*, methods: Collection[str] = (), traits: Collection[AssuranceTrait] = (), max_age: timedelta | None = None, purpose: str | None = None, clock: Callable[[], ~datetime.datetime]=<function <lambda>>) → AuthorizationPredicate

Require normalized authentication observations from immutable evidence.

Raw provider acr and amr values remain inert. Applications that understand them must map them to project-owned methods or traits before constructing evidence.

Parameters:

• methods – Verified authentication methods that must all be represented.

• traits – Verified assurance traits that must all be represented.

• max_age – Maximum age of every item of evidence used by the requirement.

• purpose – Optional action to which a step-up observation must be bound.

• clock – Injected UTC clock used for deterministic freshness decisions.

Returns:

A synchronous native Litestar authorization predicate.

litestar_security.requires_at_least(count: int, *children: AuthorizationPredicate) → AuthorizationPredicate

Require at least count child predicates.

Parameters:

• count – How many children must be satisfied.

• *children – The predicates to draw from.

Returns:

The composed predicate.

Raises:

ImproperlyConfiguredException – If the count is not between one and the number of children.

litestar_security.requires_authenticated() → AuthorizationPredicate

Require any authenticated principal.

Returns:

A predicate satisfied by any non-anonymous principal.

litestar_security.requires_capability(capability: str) → AuthorizationPredicate

Require one capability from the immutable authorization snapshot.

Parameters:

capability – The capability name, normalized before comparison.

Returns:

A predicate satisfied when the principal holds the capability.

litestar_security.requires_one_of(*children: AuthorizationPredicate) → AuthorizationPredicate

Require exactly one child predicate.

Parameters:

*children – The predicates to draw from.

Returns:

The composed predicate, denying when more than one child is satisfied.

litestar_security.requires_role(role: str) → AuthorizationPredicate

Require one role from the immutable authorization snapshot.

Parameters:

role – The role name, normalized before comparison.

Returns:

A predicate satisfied when the principal holds the role.

litestar_security.requires_scope(scope: str) → AuthorizationPredicate

Require one scope from the immutable authorization snapshot.

Parameters:

scope – The scope name, normalized before comparison.

Returns:

A predicate satisfied when the principal holds the scope.

litestar_security.requires_team_role(*, team_parameter: str = 'team_id', roles: Collection[str]) → AuthorizationPredicate

Require one allowed role for the team selected by a parsed path parameter.

The team is read from the parsed path parameter rather than the request body, so the value the guard checks is the one the route will act on.

Parameters:

• team_parameter – The path parameter naming the team.

• roles – The roles that satisfy the guard, normalized before comparison.

Returns:

A predicate satisfied when the principal holds one of the roles in that team.

Raises:

ImproperlyConfiguredException – If no roles are supplied.

litestar_security.requires_tenant(*, tenant_parameter: str = 'tenant_id') → AuthorizationPredicate

Require membership in the tenant selected by a parsed path parameter.

Parameters:

tenant_parameter – The path parameter naming the tenant.

Returns:

A predicate satisfied when the principal belongs to that tenant.

litestar_security.resolve_authorization(snapshot: AuthorizationSnapshot, restrictions: Sequence[CredentialRestrictions]) → AuthorizationSnapshot

Narrow application authorization by every credential-carried bound.

Parameters:

• snapshot – The application-resolved authorization source of truth.

• restrictions – Bounds from successful same-subject credentials.

Returns:

One immutable effective snapshot that never expands snapshot.

Notes

A credential never expands or restates attributes; they remain application-authoritative, and guards must not read them as a credential-granted authorization axis.

litestar_security.websocket_policy_fingerprint(plan: object) → str

Return a stable process-independent fingerprint for one compiled plan.

Parameters:

plan – The frozen compiled security plan.

Returns:

A hexadecimal SHA-256 fingerprint.

Configuration

Configuration for the Litestar Security plugin.

class litestar_security.config.ExternalCSRF(name: str, validate: Callable[[str, str, AuthenticationPolicy], bool])

Bases: object

Declare a named application-owned CSRF coverage validator.

__init__(name: str, validate: Callable[[str, str, AuthenticationPolicy], bool]) → None

class litestar_security.config.MFAConfig(store: object, secret_protector: object, policy: TOTPPolicy | None = None, recovery_peppers: Sequence[RecoveryCodePepper] = (), login_methods: LoginMethodStore | None = None, events: SecurityEventSink | None = None, step_up_store: object | None = None, require_at_login: bool = False, login_challenge_store: object | None = None, route_prefix: str = '/auth', issuer: str = 'Litestar Security', register_routes: bool = True, docs: RouteDocs = <factory>)

Bases: object

Configure MFA capabilities without selecting a persistence technology.

__init__(store: object, secret_protector: object, policy: TOTPPolicy | None = None, recovery_peppers: Sequence[RecoveryCodePepper] = (), login_methods: LoginMethodStore | None = None, events: SecurityEventSink | None = None, step_up_store: object | None = None, require_at_login: bool = False, login_challenge_store: object | None = None, route_prefix: str = '/auth', issuer: str = 'Litestar Security', register_routes: bool = True, docs: RouteDocs = <factory>) → None

class litestar_security.config.PasskeyConfig(store: object, challenge_store: object, rp_id: str, origins: ~collections.abc.Sequence[str], rp_name: str = 'Litestar Security', algorithms: ~collections.abc.Sequence[int] = (-8, -7, -257), challenge_ttl: ~datetime.timedelta = datetime.timedelta(seconds=300), allow_insecure_localhost: bool = False, worker_timeout: float = 10.0, attestation_trust: AttestationTrustMapper | None = None, login_methods: LoginMethodStore | None = None, events: SecurityEventSink | None = None, step_up_store: object | None = None, route_prefix: str = '/auth', register_routes: bool = True, docs: ~litestar_security._docs.RouteDocs = <factory>)

Bases: object

Configure exact WebAuthn relying-party and persistence boundaries.

__init__(store: object, challenge_store: object, rp_id: str, origins: ~collections.abc.Sequence[str], rp_name: str = 'Litestar Security', algorithms: ~collections.abc.Sequence[int] = (-8, -7, -257), challenge_ttl: ~datetime.timedelta = datetime.timedelta(seconds=300), allow_insecure_localhost: bool = False, worker_timeout: float = 10.0, attestation_trust: AttestationTrustMapper | None = None, login_methods: LoginMethodStore | None = None, events: SecurityEventSink | None = None, step_up_store: object | None = None, route_prefix: str = '/auth', register_routes: bool = True, docs: ~litestar_security._docs.RouteDocs = <factory>) → None

class litestar_security.config.RaisedErrorSchema(schema: type[object], media_type: str)

Bases: object

Declare how application exception handling renders raised route errors.

Parameters:

• schema – The body type application exception handlers serialize.

• media_type – The response content type used for that serialized body.

Raises:

ImproperlyConfiguredException – If schema is not a type or media_type is not a non-blank string.

__init__(schema: type[object], media_type: str) → None

class litestar_security.config.SecurityConfig(slots: Sequence[CredentialSlot[Any]] = (), mechanisms: Any, ~typing.Any, ~litestar_security.config.UserT]]=(), max_openapi_combinations: int = 32, external_csrf: ExternalCSRF | None = None, exclude: Sequence[str] | str | None = None, require_default: bool = False, local_auth: LocalAuthConfig[UserT] | None = None, local_jwks: LocalJWKSConfig | None = None, oauth: OAuthConfig | None = None, protected_resource: ProtectedResourceConfig | None = None, mfa: MFAConfig | None = None, passkeys: PasskeyConfig | None = None, api_key: APIKeyConfig | None = None, iap: GoogleIAPConfig[UserT] | None = None, service_token: ServiceTokenConfig | None = None, headers: SecurityHeadersConfig | None = None, websocket: WebSocketSecurityConfig = <factory>, authorization_resolver: AuthorizationResolver[UserT] | None = None, jwks_providers: Sequence[JWKSProvider] = (), jwks_warmup_failure: Literal['fail_startup', 'lazy']='fail_startup', wire_rename: RenameStrategy | None = None, wire_forbid_unknown_fields: bool = True, raised_error_schema: RaisedErrorSchema | None = None)

Bases: Generic[UserT]

Configure the per-application security runtime.

exclude: Sequence[str] | str | None

Regular expressions matched against a route path to exclude it from security.

Mirrors JWTAuth.exclude: a single pattern or a sequence of patterns, joined into one expression and compiled with re. A pattern is anchored at the start of the route path, so "^/static" and "/static" both exclude /static/{file_path:path} while a bare "static" does not.

Exclusion is total and applies when the route is compiled, not per request: an excluded route is never authenticated, carries no principal, and contributes an anonymous security requirement to OpenAPI rather than the configured schemes. A route that declares its own auth= and also matches a pattern is a contradiction and is rejected at startup.

protected_resource: ProtectedResourceConfig | None

Describe this application as an OAuth 2.1 protected resource.

When set, the plugin publishes the RFC 9728 metadata document at /.well-known/oauth-protected-resource so an authorization server or a client can discover which issuers this resource trusts, which scopes it understands, and how a bearer token may be presented to it. The route is unauthenticated, as the specification requires.

wire_rename: RenameStrategy | None

How generated request and response members are spelled on the wire.

None keeps the field names as Python spells them, which is snake_case. Any of "lower", "upper", "camel", "pascal", and "kebab" selects a named strategy, and a Callable[[str], str] covers a house convention outside those five. The choice reaches the OpenAPI document as well as the wire, so a generated client follows it without further work.

A handful of generated schemas opt out because their member names belong to a specification rather than to this library - the RFC 6749 token response, the OIDC back-channel logout form, and the bodies Litestar’s own exception handling renders.

wire_forbid_unknown_fields: bool

Whether an unrecognized member in a request body is a decoding error.

Strictness applies to decoding, so it constrains request schemas only. Rejecting the unknown member is what keeps a stale or misspelled optional field from resolving to its default and producing a wrong but successful request.

__init__(slots: Sequence[CredentialSlot[Any]] = (), mechanisms: Any, ~typing.Any, ~litestar_security.config.UserT]]=(), max_openapi_combinations: int = 32, external_csrf: ExternalCSRF | None = None, exclude: Sequence[str] | str | None = None, require_default: bool = False, local_auth: LocalAuthConfig[UserT] | None = None, local_jwks: LocalJWKSConfig | None = None, oauth: OAuthConfig | None = None, protected_resource: ProtectedResourceConfig | None = None, mfa: MFAConfig | None = None, passkeys: PasskeyConfig | None = None, api_key: APIKeyConfig | None = None, iap: GoogleIAPConfig[UserT] | None = None, service_token: ServiceTokenConfig | None = None, headers: SecurityHeadersConfig | None = None, websocket: WebSocketSecurityConfig = <factory>, authorization_resolver: AuthorizationResolver[UserT] | None = None, jwks_providers: Sequence[JWKSProvider] = (), jwks_warmup_failure: Literal['fail_startup', 'lazy']='fail_startup', wire_rename: RenameStrategy | None = None, wire_forbid_unknown_fields: bool = True, raised_error_schema: RaisedErrorSchema | None = None) → None

raised_error_schema: RaisedErrorSchema | None

The body type and media type application handlers use for raised errors.

Generated routes raise their denial statuses through the application’s exception handlers. Set this when those handlers render a body other than Litestar’s default RouteError. The declaration changes only generated-route OpenAPI response specifications; it does not install an exception handler or alter runtime responses.

wire_policy() → WirePolicy

Return the wire convention every generated route body is built with.

Returns:

The casing strategy and unknown-field policy as one hashable value.

Raises:

ImproperlyConfiguredException – If the strategy is neither one of the named strategies nor a callable, or the unknown-field policy is not boolean.

Workers

Runtime execution ports shared by the security components.

These are service ports rather than configuration: a worker budget, a metrics sink, and the bridge that runs an application’s blocking implementation off the event loop. They live below config so that the account services and the token providers can depend on them without depending on configuration, which in turn lets configuration reach the account services without a cycle.

litestar_security.config re-exports every name here, which is the documented import path.

class litestar_security.workers.BlockingCallRunner(limiter: CapacityLimiter = <factory>)

Bases: object

Submit explicit blocking feature operations through one finite worker budget.

async run(function: Callable[[...], ResultT], /, *args: object, **kwargs: object) → ResultT

Run one complete blocking operation without abandoning an in-flight mutation.

Parameters:

• function – The synchronous atomic operation.

• *args – Positional arguments forwarded to the operation.

• **kwargs – Keyword arguments forwarded to the operation.

Returns:

The operation result after its worker job completes.

__init__(limiter: CapacityLimiter = <factory>) → None

class litestar_security.workers.BlockingIntegration(implementation: SyncT)

Bases: Generic[SyncT]

Mark one explicitly synchronous application integration for startup normalization.

Parameters:

implementation – The complete synchronous feature protocol.

__init__(implementation: SyncT) → None

class litestar_security.workers.NoOpSecurityMetrics

Bases: object

Default metric sink with zero vendor or runtime overhead.

increment(name: str, *, attributes: Mapping[str, str] = mappingproxy({})) → None

Ignore a counter.

Parameters:

• name – The counter name.

• attributes – Dimensions to record with the increment.

observe(name: str, value: float, *, attributes: Mapping[str, str] = mappingproxy({})) → None

Ignore an observation.

Parameters:

• name – The measurement name.

• value – The observed value.

• attributes – Dimensions to record with the observation.

__init__() → None

class litestar_security.workers.SecurityMetrics(*args, **kwargs)

Bases: Protocol

Vendor-neutral synchronous metric sink that must not block.

increment(name: str, *, attributes: Mapping[str, str] = mappingproxy({})) → None

Increment one security counter.

Parameters:

• name – The counter name.

• attributes – Dimensions to record with the increment.

observe(name: str, value: float, *, attributes: Mapping[str, str] = mappingproxy({})) → None

Observe one security duration or size.

Parameters:

• name – The measurement name.

• value – The observed value.

• attributes – Dimensions to record with the observation.

__init__(*args, **kwargs)

class litestar_security.workers.WorkerLimits(network_tokens: int = 8, crypto_tokens: int = 32, timeout: float = 10.0)

Bases: object

Paired dedicated limiters that components may share as one worker budget.

__init__(network_tokens: int = 8, crypto_tokens: int = 32, timeout: float = 10.0) → None

Plugin

Litestar Security plugin integration.

class litestar_security.plugin.SecurityPlugin(config: SecurityConfig[UserT] | None = None)

Bases: InitPlugin, ReceiveRoutePlugin, CLIPlugin, Generic[UserT]

Expose the Litestar Security configuration and CLI integration points.

__init__(config: SecurityConfig[UserT] | None = None) → None

Initialize the plugin.

on_app_init(app_config: AppConfig) → AppConfig

Validate ownership and install one typed security runtime.

Parameters:

app_config – The application configuration to extend.

Returns:

The same configuration, with the security middleware, dependencies, generated routes, and OpenAPI contributions installed.

Raises:

ImproperlyConfiguredException – If the application already owns something this plugin must own, such as a competing session middleware, CSRF config, or reserved dependency name.

receive_route(route: BaseRoute) → None

Compile every initial or dynamically registered route.

Parameters:

route – The route Litestar just registered.

Raises:

ImproperlyConfiguredException – If called before application initialization, or if the route’s declared policy cannot compile.

on_cli_init(cli: Group) → None

Attach the security command group to the Litestar CLI.

Parameters:

cli – The root Litestar CLI group.

Browser response headers

Opt-in browser response security headers for Litestar applications.

class litestar_security.headers.CSPMode(*values)

Bases: str, Enum

Select the browser CSP response-header mode.

class litestar_security.headers.ContentSecurityPolicy(directives: Mapping[str, Sequence[str]], mode: CSPMode = CSPMode.ENFORCE, nonce_directives: Sequence[str] = ())

Bases: object

Define one explicit Content Security Policy.

Parameters:

• mode – Whether the policy enforces or only reports violations.

• directives – Explicit directive names and their ordered source values.

• nonce_directives – Directives that receive the response-local nonce.

Raises:

ImproperlyConfiguredException – If a directive or source is unsafe, or a nonce directive is absent from directives.

property header_name: str

Return the standard header name for this policy.

Returns:

The enforcing or report-only CSP header name.

serialize(*, nonce: str | None = None) → str

Serialize the policy deterministically.

Parameters:

nonce – Response-local nonce to append to configured directives.

Returns:

A CSP header value.

Raises:

ImproperlyConfiguredException – If nonce directives exist but no response nonce was supplied.

__init__(directives: Mapping[str, Sequence[str]], mode: CSPMode = CSPMode.ENFORCE, nonce_directives: Sequence[str] = ()) → None

class litestar_security.headers.SecurityHeadersConfig(static: Mapping[str, str]=<factory>, csp: ContentSecurityPolicy | None = None)

Bases: object

Configure native static response headers and optional CSP.

Parameters:

• static – Application-owned static security header values.

• csp – Optional explicit Content Security Policy.

Raises:

ImproperlyConfiguredException – If names or values are unsafe, or a configured CSP header conflicts with csp.

classmethod hardened() → SecurityHeadersConfig

Create the recommended opt-in static browser-security baseline.

Returns:

A configuration with HSTS, frame, content-type, and referrer protections. Callers may supply more restrictive per-route headers.

__init__(static: Mapping[str, str]=<factory>, csp: ContentSecurityPolicy | None = None) → None

Authentication

Typed authentication contracts and deterministic mechanism registration.

class litestar_security.authentication.Authenticated(claims: ClaimsT, evidence: AuthenticationEvidence, grants: AuthorizationSnapshot = <factory>, restrictions: CredentialRestrictions = <factory>)

Bases: Generic[ClaimsT]

Carry the typed result of successful credential verification.

__init__(claims: ClaimsT, evidence: AuthenticationEvidence, grants: AuthorizationSnapshot = <factory>, restrictions: CredentialRestrictions = <factory>) → None

class litestar_security.authentication.AuthenticationMechanism(authenticator: CredentialVerifier[CredentialT, ClaimsT], resolver: IdentityResolver[ClaimsT, UserT], scheme_name: str | None = None, security_scheme: SecurityScheme | None = None, session_capable: bool = False)

Bases: Generic[CredentialT, ClaimsT, UserT]

Pair one slot authenticator with its identity resolver.

__init__(authenticator: CredentialVerifier[CredentialT, ClaimsT], resolver: IdentityResolver[ClaimsT, UserT], scheme_name: str | None = None, security_scheme: SecurityScheme | None = None, session_capable: bool = False) → None

class litestar_security.authentication.AuthenticationPolicy(*_args: object, **_kwargs: object)

Bases: object

Immutable closed request-authentication expression.

class litestar_security.authentication.AuthenticationRegistry(slots: Sequence[CredentialSlot[Any]] = (), mechanisms: Sequence[AuthenticationMechanism[Any, Any, UserT]] = (), authorization_resolver: AuthorizationResolver[UserT] | None = None, require_default: bool = False)

Bases: Generic[UserT]

Validate and compile deterministic credential-slot ownership.

property slot_names: tuple[str, ...]

Return normalized slot names in configuration order.

property mechanism_names: tuple[str, ...]

Return normalized mechanism names in configuration order.

property default_mechanism_names: tuple[str, ...]

Return default-participating mechanism names in configuration order.

get_slot(name: str) → CredentialSlot[Any]

Look up an owned slot by normalized name.

Parameters:

name – The slot name, normalized before lookup.

Returns:

The registered slot.

get_mechanism(name: str) → AuthenticationMechanism[Any, Any, UserT]

Look up a mechanism by normalized name.

Parameters:

name – The mechanism name, normalized before lookup.

Returns:

The registered mechanism.

get_mechanism_for_slot(name: str) → AuthenticationMechanism[Any, Any, UserT] | None

Look up the sole mechanism owning a normalized slot.

Parameters:

name – The slot name, normalized before lookup.

Returns:

The owning mechanism, or None when no mechanism claims the slot.

evaluator() → _AuthenticationEvaluator[UserT]

Create a stateless evaluator bound to this compiled registry.

Returns:

An evaluator that may be shared across requests.

__init__(slots: Sequence[CredentialSlot[Any]] = (), mechanisms: Sequence[AuthenticationMechanism[Any, Any, UserT]] = (), authorization_resolver: AuthorizationResolver[UserT] | None = None, require_default: bool = False) → None

class litestar_security.authentication.AuthorizationResolver(*args, **kwargs)

Bases: Protocol[_UserT]

Application-owned resolution of authorization for one verified principal.

async resolve(principal: Principal[_UserT]) → AuthorizationSnapshot | InvalidCredentials | VerificationUnavailable

Load one immutable application authorization snapshot.

Parameters:

principal – The same-subject principal established by authentication.

Returns:

The immutable application authorization snapshot; InvalidCredentials for an expected authorization denial; or VerificationUnavailable for expected dependency trouble.

Raises:

Exception – For an unexpected resolver error or outage. The evaluator catches this boundary in _resolve_authorization() and maps it to one sanitized 503 response.

__init__(*args, **kwargs)

class litestar_security.authentication.CredentialSlot(*args, **kwargs)

Bases: Protocol[_CredentialT]

Synchronous, non-blocking credential extraction boundary.

extract(connection: ASGIConnection[Any, Any, Any, Any]) → NoCredentials | PresentedCredential[_CredentialT] | InvalidCredentials

Extract at most one credential from the connection.

Runs synchronously on every request, so it must not block or perform I/O.

Parameters:

connection – The incoming connection.

Returns:

The presented credential, NoCredentials when this slot is empty, or InvalidCredentials when the slot is malformed.

__init__(*args, **kwargs)

class litestar_security.authentication.CredentialVerifier(*args, **kwargs)

Bases: Protocol[_RequestCredentialT_contra, _ClaimsT]

Async credential verification boundary.

async authenticate(credential: _RequestCredentialT_contra, connection: ASGIConnection[Any, Any, Any, Any]) → NoCredentials | Authenticated[_ClaimsT] | InvalidCredentials | VerificationUnavailable

Verify a credential without resolving application identity.

Parameters:

• credential – The value produced by this authenticator’s slot.

• connection – The incoming connection.

Returns:

The verified claims, or a sanitized outcome describing why verification did not succeed.

__init__(*args, **kwargs)

class litestar_security.authentication.IdentityResolver(*args, **kwargs)

Bases: Protocol[_ResolverClaimsT_contra, _UserT]

Async mapping from verified claims to one application principal.

async resolve(claims: _ResolverClaimsT_contra) → Principal[_UserT] | InvalidCredentials | VerificationUnavailable

Resolve verified claims into a principal or sanitized resolution outcome.

Parameters:

claims – The claims produced by the paired authenticator.

Returns:

The application principal; InvalidCredentials for an expected unknown or inactive identity; or VerificationUnavailable for expected dependency trouble.

Raises:

Exception – For an unexpected resolver error or outage. The evaluator catches this boundary in _resolve() and maps it to one sanitized 503 response.

__init__(*args, **kwargs)

class litestar_security.authentication.InvalidCredentials(code: str = 'invalid_credentials')

Bases: object

Indicate that a presented credential cannot authenticate.

__init__(code: str = 'invalid_credentials') → None

class litestar_security.authentication.MechanismRequirement(name: str, scopes: tuple[str, ...] = ())

Bases: object

Select one configured mechanism and optional provider scopes.

__init__(name: str, scopes: tuple[str, ...] = ()) → None

class litestar_security.authentication.NoCredentials

Bases: object

Indicate that an owned slot contains no credential.

__init__() → None

class litestar_security.authentication.PresentedCredential(value: CredentialT)

Bases: Generic[CredentialT]

A credential extracted from one owned request slot.

__init__(value: CredentialT) → None

class litestar_security.authentication.VerificationUnavailable(code: str = 'verification_unavailable', retry_after: int | None = None)

Bases: object

Indicate that a verifier cannot make a trustworthy decision.

__init__(code: str = 'verification_unavailable', retry_after: int | None = None) → None

litestar_security.authentication.all_of(*requirements: str | MechanismRequirement) → AuthenticationPolicy

Require every named authentication mechanism.

Parameters:

*requirements – Mechanism names or requirements that must all succeed.

Returns:

A policy satisfied only when every participant succeeds.

litestar_security.authentication.any_of(*requirements: str | MechanismRequirement) → AuthenticationPolicy

Require at least one named authentication mechanism.

Parameters:

*requirements – Mechanism names or requirements to accept.

Returns:

A policy satisfied by any one participant.

litestar_security.authentication.at_least(count: int, *requirements: str | MechanismRequirement) → AuthenticationPolicy

Require a positive threshold of named authentication mechanisms.

Parameters:

• count – How many participants must succeed.

• *requirements – Mechanism names or requirements to draw from.

Returns:

A policy satisfied by any count of the participants.

Raises:

ImproperlyConfiguredException – If the count is not between one and the number of participants.

litestar_security.authentication.exclude() → AuthenticationPolicy

Bypass request authentication while preserving the default CSRF policy.

Returns:

A policy that skips credential extraction and authentication.

litestar_security.authentication.mechanism(name: str, *scopes: str) → MechanismRequirement

Select a named mechanism and its requested OAuth or OIDC scopes.

Parameters:

• name – The configured mechanism name.

• *scopes – Provider scopes to request. Only OAuth and OIDC schemes accept these.

Returns:

The requirement, for use inside a policy expression.

litestar_security.authentication.optional(policy: AuthenticationPolicy) → AuthenticationPolicy

Allow anonymous access only when a positive policy sees no credential.

A presented-but-invalid credential is still rejected: optional means the route tolerates absence, not failure.

Parameters:

policy – The positive policy to apply when a credential is present.

Returns:

A policy that admits anonymous callers alongside authenticated ones.

Raises:

ImproperlyConfiguredException – If the policy is public or already optional.

litestar_security.authentication.public() → AuthenticationPolicy

Deliberately skip request credential verification.

Returns:

A policy that authenticates nothing, leaving the anonymous principal in place.

litestar_security.authentication.required(*requirements: str | MechanismRequirement) → AuthenticationPolicy

Require an explicit OR expression or the implicit default participants.

Parameters:

*requirements – Mechanism names or requirements. Passing none requires any mechanism that participates by default.

Returns:

A policy that rejects a request presenting no accepted credential.

Security context

Immutable request security context contracts.

class litestar_security.context.AuthenticationEvidence(mechanism: str, slot: str, authenticated_at: datetime, expires_at: datetime | None = None, methods: frozenset[str] = frozenset({}), traits: frozenset[str] = frozenset({}), acr: str | None = None, amr: tuple[str, ...] = ())

Bases: object

Normalized evidence emitted by one successful authenticator.

__init__(mechanism: str, slot: str, authenticated_at: datetime, expires_at: datetime | None = None, methods: frozenset[str] = frozenset({}), traits: frozenset[str] = frozenset({}), acr: str | None = None, amr: tuple[str, ...] = ()) → None

class litestar_security.context.AuthorizationSnapshot(scopes: frozenset[str] = frozenset({}), roles: frozenset[str] = frozenset({}), capabilities: frozenset[str] = frozenset({}), team_roles: Mapping[str, frozenset[str]]=<factory>, tenant_ids: frozenset[str] = frozenset({}), resources: frozenset[ResourcePermission] = frozenset({}), attributes: Mapping[str, object]=<factory>)

Bases: object

Immutable application authorization data.

__init__(scopes: frozenset[str] = frozenset({}), roles: frozenset[str] = frozenset({}), capabilities: frozenset[str] = frozenset({}), team_roles: Mapping[str, frozenset[str]]=<factory>, tenant_ids: frozenset[str] = frozenset({}), resources: frozenset[ResourcePermission] = frozenset({}), attributes: Mapping[str, object]=<factory>) → None

class litestar_security.context.CredentialRestrictions(scopes: frozenset[str] | None = None, roles: frozenset[str] | None = None, capabilities: frozenset[str] | None = None, team_ids: frozenset[str] | None = None, tenant_ids: frozenset[str] | None = None, resources: frozenset[ResourcePermission] | None = None)

Bases: object

Authorization bounds imposed by one credential.

__init__(scopes: frozenset[str] | None = None, roles: frozenset[str] | None = None, capabilities: frozenset[str] | None = None, team_ids: frozenset[str] | None = None, tenant_ids: frozenset[str] | None = None, resources: frozenset[ResourcePermission] | None = None) → None

class litestar_security.context.LitestarSessionHandle(scope: HTTPScope | WebSocketScope)

Bases: object

Live view over Litestar’s native session scope value.

property is_available: bool

Return whether native session middleware attached state.

property can_persist: bool

Return whether this connection permits session mutation.

get(key: str, default: object = None) → object

Read the current native session mapping.

Parameters:

• key – The session key to read.

• default – The value to return when the key is absent.

Returns:

The stored value, or default.

set(key: str, value: object) → None

Store a value when the native session can persist.

Parameters:

• key – The session key to write.

• value – The value to store.

pop(key: str, default: object = None) → object

Remove a value when the native session can persist.

Parameters:

• key – The session key to remove.

• default – The value to return when the key is absent.

Returns:

The removed value, or default.

clear() → None

Clear the native session when it can persist.

__init__(scope: HTTPScope | WebSocketScope) → None

class litestar_security.context.NullSessionHandle

Bases: object

Stateless session capability for applications without sessions.

property is_available: bool

Return that no native session is attached.

property can_persist: bool

Return that no session mutations can persist.

get(key: str, default: object = None) → object

Return the caller’s default.

Parameters:

• key – The session key to read.

• default – The value to return when the key is absent.

Returns:

The stored value, or default.

set(key: str, value: object) → None

Reject writes when session storage is unavailable.

Parameters:

• key – Ignored; no session is attached.

• value – Ignored; no session is attached.

Raises:

SessionUnavailableError – Always, because no session is attached.

pop(key: str, default: object = None) → object

Return the caller’s default without retaining state.

Parameters:

• key – The session key to remove.

• default – The value to return when the key is absent.

Returns:

The removed value, or default.

clear() → None

Reject clearing when session storage is unavailable.

__init__() → None

class litestar_security.context.Principal(id: str | None, display_name: str | None = None, user: UserT | None = None)

Bases: Generic[UserT]

Stable identity envelope for anonymous, user, and service actors.

classmethod anonymous() → Principal[UserT]

Create an anonymous principal.

Returns:

A principal with no identity, used before authentication runs.

property is_authenticated: bool

Return whether this principal has an authenticated identity.

property has_user: bool

Return whether an application user is attached.

require_user() → UserT

Return the application user or fail without revealing actor state.

Returns:

The attached application user.

Raises:

NotAuthorizedException – If no user is attached. The message never distinguishes an anonymous caller from an authenticated one whose user could not be loaded.

__init__(id: str | None, display_name: str | None = None, user: UserT | None = None) → None

class litestar_security.context.ResourcePermission(resource: str, scopes: frozenset[str] = frozenset({}))

Bases: object

Credential or application permission scoped to one resource.

__init__(resource: str, scopes: frozenset[str] = frozenset({})) → None

class litestar_security.context.SecurityContext(session: SessionHandle, evidence: tuple[~litestar_security.context.AuthenticationEvidence, ...]=(), authorization: AuthorizationSnapshot = <factory>, restrictions: tuple[~litestar_security.context.CredentialRestrictions, ...]=())

Bases: object

Authentication evidence, authorization, and optional session capability.

property expires_at: datetime | None

Return the earliest bounded evidence expiry.

__init__(session: SessionHandle, evidence: tuple[~litestar_security.context.AuthenticationEvidence, ...]=(), authorization: AuthorizationSnapshot = <factory>, restrictions: tuple[~litestar_security.context.CredentialRestrictions, ...]=()) → None

class litestar_security.context.SessionHandle(*args, **kwargs)

Bases: Protocol

Uniform access to an optional native Litestar session.

property is_available: bool

Return whether a session is attached.

property can_persist: bool

Return whether session mutations can persist.

get(key: str, default: object = None) → object

Read a session value.

Parameters:

• key – The session key to read.

• default – The value to return when the key is absent.

Returns:

The stored value, or default.

set(key: str, value: object) → None

Store a session value.

Parameters:

• key – The session key to write.

• value – The value to store.

pop(key: str, default: object = None) → object

Remove and return a session value.

Parameters:

• key – The session key to remove.

• default – The value to return when the key is absent.

Returns:

The removed value, or default.

clear() → None

Remove all session values.

__init__(*args, **kwargs)

exception litestar_security.context.SessionPersistenceUnavailableError

Bases: SessionUnavailableError

Raised when attached session state is read-only.

__init__() → None

Initialize the stable public error.

exception litestar_security.context.SessionUnavailableError

Bases: RuntimeError

Raised when no native session storage is attached.

__init__() → None

Initialize the stable public error.

litestar_security.context.resolve_authorization(snapshot: AuthorizationSnapshot, restrictions: Sequence[CredentialRestrictions]) → AuthorizationSnapshot

Narrow application authorization by every credential-carried bound.

Parameters:

• snapshot – The application-resolved authorization source of truth.

• restrictions – Bounds from successful same-subject credentials.

Returns:

One immutable effective snapshot that never expands snapshot.

Notes

A credential never expands or restates attributes; they remain application-authoritative, and guards must not read them as a credential-granted authorization axis.

Wire schemas

Wire-schema conventions for generated route bodies.

Every request and response schema on the generated /auth route tree shares one casing convention and one unknown-field policy through WireStruct. Applications defining their own schemas alongside the generated routes may inherit the same base so a single convention holds across the whole tree.

RouteError lives here rather than beside either route family’s own schemas because both need it and neither owns it: providers/oauth/ does not import from accounts/, and the body it describes is Litestar’s rather than this library’s.

class litestar_security.schema.ProblemDetail(status: int, title: str, detail: str, extra: dict[str, Any] | list[Any] | None = None)

Bases: WireStruct

The body a denial takes when the application converts every HTTP exception.

An application installing Litestar’s problem-details plugin with enable_for_all_http_exceptions=True replaces RouteError on every raised status, and the response is served as application/problem+json.

These are the members Litestar’s conversion actually emits, which is not the RFC 9457 five-member shape: the raised detail is moved onto title and detail falls back to the HTTP reason phrase, while type and instance are never produced. Unknown members are tolerated both because RFC 9457 permits extension members and because the sender is Litestar rather than this library.

status: int

The HTTP status, repeated in the body.

title: str

The raised explanation, which the conversion moves here from detail.

detail: str

The HTTP reason phrase, which the conversion leaves as the default.

extra: dict[str, Any] | list[Any] | None

Structured context the raised exception carried, carried through unchanged.

class litestar_security.schema.RouteError(status_code: int, detail: str, extra: dict[str, Any] | list[Any] | None = None)

Bases: WireStruct

The body a generated route sends when it raises rather than returns.

A denial - 400, 401, 429, 503, and the OAuth 409 - reaches the wire through Litestar’s exception handling, not through the handler’s return value, so the body is ExceptionResponseContent: the status repeated inside the payload, a human-readable detail, and extra when the raised exception carries structured context. A request-validation failure always carries one, as a list of {message, key, source} entries, so extra is a member clients see in practice rather than a theoretical one.

Distinguish this from OperationMessage, which is the body a handler returns - the 200 confirmations and the 409 conflict. The two are separate schemas because the distinction that decides the shape is raised-versus-returned, not error-versus-success.

Unknown members are tolerated here, against the base policy, because the sender is Litestar: an application handler may add its own members and a future Litestar may too, and neither is a reason for a client of this library to fail decoding.

status_code: int

The HTTP status, repeated in the body by Litestar’s exception handling.

detail: str

A human-readable explanation that never names an account.

extra: dict[str, Any] | list[Any] | None

Structured context the raised exception carried, when it carried any.

class litestar_security.schema.WirePolicy(rename: RenameStrategy | None = None, forbid_unknown_fields: bool = True)

Bases: object

How generated request and response bodies are spelled on the wire.

One value carries both halves of the convention, and it is hashable, so the generated routers a feature configuration caches stay one router per policy rather than one router that a second application can find already built for a casing it did not ask for.

rename: RenameStrategy | None

The casing strategy, or None for the field names as Python spells them.

forbid_unknown_fields: bool

Whether an unrecognized member is a decoding error rather than ignored.

__init__(rename: RenameStrategy | None = None, forbid_unknown_fields: bool = True) → None

class litestar_security.schema.WireStruct

Bases: Struct

Base for every generated-route wire schema, and the default it is spelled in.

Field names reach the wire exactly as they are spelled in Python, and an unrecognized member is a decoding error rather than a silently discarded key. Rejecting the unknown member is what keeps a stale or misspelled optional field from resolving to its default and producing a wrong but successful request.

That is the default rather than a fixed policy. An application chooses the convention through SecurityConfig.wire_rename and wire_forbid_unknown_fields, and the generated routes carry the choice into the request body, the response body, and the OpenAPI schema together. A schema declares what it is called here; the configuration decides how it is spelled.

Subclasses restate frozen=True:

class LocalCredentials(WireStruct, frozen=True):
    '''Password credentials accepted by generated login handlers.'''

That keyword is redundant at runtime, because msgspec inherits the struct configuration, and required by the type checkers, which read immutability from the class keywords rather than from the base. Keeping the base frozen anyway means a subclass that omits the keyword is still immutable in fact, which is the safer direction for the mistake to fall.

Strictness applies to decoding, so it constrains request schemas only; response schemas inherit it inertly. A schema that must tolerate members it does not model - a specification-defined body whose sender may legitimately add them - overrides the policy for itself and records why:

class BackchannelLogout(WireStruct, frozen=True, forbid_unknown_fields=False):
    '''The specification permits unrecognized members.'''

Prefer that per-schema override to relaxing this base: it keeps the safe default intact and leaves the reason beside the schema that needs it.

Authorization guards

Pure authorization predicates exposed as native Litestar guards.

class litestar_security.guards.AssuranceRequirement(methods: frozenset[str] = frozenset({}), traits: frozenset[AssuranceTrait] = frozenset({}), max_age: timedelta | None = None, purpose: str | None = None)

Bases: object

Method, trait, freshness, and purpose observations required by a route.

__init__(methods: frozenset[str] = frozenset({}), traits: frozenset[AssuranceTrait] = frozenset({}), max_age: timedelta | None = None, purpose: str | None = None) → None

class litestar_security.guards.AssuranceTrait(*values)

Bases: str, Enum

Normalized assurance properties established by verified evidence.

class litestar_security.guards.AuthorizationDecision(granted: bool, code: str = 'allowed', path: tuple[str, ...] = (), authentication_required: bool = False)

Bases: object

One predicate’s verdict, carrying why it was reached.

Parameters:

• granted – Whether the predicate allows the connection.

• code – Stable machine-readable reason, reported when access is denied.

• path – Predicate names from the outermost composite inward, for diagnostics.

• authentication_required – Deny because no principal is authenticated, which translates to 401 instead of 403.

prefixed(*parts: str) → AuthorizationDecision

Return this decision with parts prepended to its diagnostic path.

Parameters:

*parts – Names to record ahead of the existing path.

Returns:

An equivalent decision reached through the named enclosing predicates.

__init__(granted: bool, code: str = 'allowed', path: tuple[str, ...] = (), authentication_required: bool = False) → None

class litestar_security.guards.AuthorizationPredicate

Bases: object

Immutable authorization decision exposed as a native Litestar guard.

decide(connection: ASGIConnection[Any, Any, Any, Any]) → AuthorizationDecision

Decide whether this predicate allows the connection.

Override this to add a predicate of your own; the built-in composites accept any subclass. Return a decision rather than raising, so a composite can report which branch denied access.

Parameters:

connection – The connection being authorized.

Returns:

This predicate’s verdict.

Raises:

NotImplementedError – If a subclass does not override this method.

litestar_security.guards.requires_all_of(*children: AuthorizationPredicate) → AuthorizationPredicate

Require every child predicate.

Parameters:

*children – The predicates that must all be satisfied.

Returns:

The composed predicate.

litestar_security.guards.requires_any_of(*children: AuthorizationPredicate) → AuthorizationPredicate

Require at least one child predicate.

Parameters:

*children – The predicates to draw from.

Returns:

The composed predicate.

litestar_security.guards.requires_assurance(*, methods: Collection[str] = (), traits: Collection[AssuranceTrait] = (), max_age: timedelta | None = None, purpose: str | None = None, clock: Callable[[], ~datetime.datetime]=<function <lambda>>) → AuthorizationPredicate

Require normalized authentication observations from immutable evidence.

Raw provider acr and amr values remain inert. Applications that understand them must map them to project-owned methods or traits before constructing evidence.

Parameters:

• methods – Verified authentication methods that must all be represented.

• traits – Verified assurance traits that must all be represented.

• max_age – Maximum age of every item of evidence used by the requirement.

• purpose – Optional action to which a step-up observation must be bound.

• clock – Injected UTC clock used for deterministic freshness decisions.

Returns:

A synchronous native Litestar authorization predicate.

litestar_security.guards.requires_at_least(count: int, *children: AuthorizationPredicate) → AuthorizationPredicate

Require at least count child predicates.

Parameters:

• count – How many children must be satisfied.

• *children – The predicates to draw from.

Returns:

The composed predicate.

Raises:

ImproperlyConfiguredException – If the count is not between one and the number of children.

litestar_security.guards.requires_authenticated() → AuthorizationPredicate

Require any authenticated principal.

Returns:

A predicate satisfied by any non-anonymous principal.

litestar_security.guards.requires_capability(capability: str) → AuthorizationPredicate

Require one capability from the immutable authorization snapshot.

Parameters:

capability – The capability name, normalized before comparison.

Returns:

A predicate satisfied when the principal holds the capability.

litestar_security.guards.requires_one_of(*children: AuthorizationPredicate) → AuthorizationPredicate

Require exactly one child predicate.

Parameters:

*children – The predicates to draw from.

Returns:

The composed predicate, denying when more than one child is satisfied.

litestar_security.guards.requires_role(role: str) → AuthorizationPredicate

Require one role from the immutable authorization snapshot.

Parameters:

role – The role name, normalized before comparison.

Returns:

A predicate satisfied when the principal holds the role.

litestar_security.guards.requires_scope(scope: str) → AuthorizationPredicate

Require one scope from the immutable authorization snapshot.

Parameters:

scope – The scope name, normalized before comparison.

Returns:

A predicate satisfied when the principal holds the scope.

litestar_security.guards.requires_team_role(*, team_parameter: str = 'team_id', roles: Collection[str]) → AuthorizationPredicate

Require one allowed role for the team selected by a parsed path parameter.

The team is read from the parsed path parameter rather than the request body, so the value the guard checks is the one the route will act on.

Parameters:

• team_parameter – The path parameter naming the team.

• roles – The roles that satisfy the guard, normalized before comparison.

Returns:

A predicate satisfied when the principal holds one of the roles in that team.

Raises:

ImproperlyConfiguredException – If no roles are supplied.

litestar_security.guards.requires_tenant(*, tenant_parameter: str = 'tenant_id') → AuthorizationPredicate

Require membership in the tenant selected by a parsed path parameter.

Parameters:

tenant_parameter – The path parameter naming the tenant.

Returns:

A predicate satisfied when the principal belongs to that tenant.

Local accounts

Curated local-account, session, and refresh-token contracts.

class litestar_security.accounts.AccountLookup(*args, **kwargs)

Bases: Protocol[UserT]

Resolve the minimal application account projection.

async find_for_login(normalized_identifier: str) → LocalAccountState[UserT] | None

Find an account through an already-normalized identifier.

The caller normalizes before calling, so match the stored value exactly rather than normalizing again.

Parameters:

normalized_identifier – The identifier as normalized by the configured normalizer.

Returns:

The account projection, or None when no account matches.

async get_by_id(account_id: str) → LocalAccountState[UserT] | None

Resolve an account by its stable security identifier.

Parameters:

account_id – The stable account identifier carried on credentials.

Returns:

The account projection, or None when the account no longer exists.

__init__(*args, **kwargs)

class litestar_security.accounts.AssuranceRequirement(methods: frozenset[str] = frozenset({}), traits: frozenset[AssuranceTrait] = frozenset({}), max_age: timedelta | None = None, purpose: str | None = None)

Bases: object

Method, trait, freshness, and purpose observations required by a route.

__init__(methods: frozenset[str] = frozenset({}), traits: frozenset[AssuranceTrait] = frozenset({}), max_age: timedelta | None = None, purpose: str | None = None) → None

class litestar_security.accounts.AssuranceTrait(*values)

Bases: str, Enum

Normalized assurance properties established by verified evidence.

class litestar_security.accounts.CreateRefreshFamilyCommand(token_id: str, token_digest: bytes, account_id: str, family_id: str, security_epoch: int, created_at: datetime, token_expires_at: datetime, family_expires_at: datetime, scopes: frozenset[str] = frozenset({}), evidence: AuthenticationEvidence | None = None)

Bases: object

Initial opaque refresh token committed atomically with its family.

__init__(token_id: str, token_digest: bytes, account_id: str, family_id: str, security_epoch: int, created_at: datetime, token_expires_at: datetime, family_expires_at: datetime, scopes: frozenset[str] = frozenset({}), evidence: AuthenticationEvidence | None = None) → None

class litestar_security.accounts.CreateSessionCommand(session_id: str, binding_id: str, binding_digest: bytes, account_id: str, security_epoch: int, created_at: datetime, authenticated_at: datetime, expires_at: datetime, display_metadata: Mapping[str, str]=<factory>)

Bases: object

Candidate authenticated-session record for one atomic creation.

__init__(session_id: str, binding_id: str, binding_digest: bytes, account_id: str, security_epoch: int, created_at: datetime, authenticated_at: datetime, expires_at: datetime, display_metadata: Mapping[str, str]=<factory>) → None

class litestar_security.accounts.InvalidInvitation(detail: str = 'Invitation is invalid or unavailable.')

Bases: object

Generic invalid-invitation response without expiry or replay details.

__init__(detail: str = 'Invitation is invalid or unavailable.') → None

class litestar_security.accounts.LifecycleAccepted(detail: str = 'If eligible, the request will be processed.')

Bases: WireStruct

Shared enumeration-resistant response body for lifecycle requests.

class litestar_security.accounts.LifecycleRejected(detail: str = 'The request is invalid.')

Bases: object

Generic malformed lifecycle request response.

__init__(detail: str = 'The request is invalid.') → None

class litestar_security.accounts.LocalAccessToken(access_token: str, expires_in: int)

Bases: object

Secret-safe response from one local access-token issuance.

__init__(access_token: str, expires_in: int) → None

class litestar_security.accounts.LocalAccessTokenIssuer(signer: TokenSigner, issuer: str, audience: str, client_id: str = 'local', lifetime: timedelta = datetime.timedelta(seconds=600), clock: Callable[[], datetime]=<function utc_now>, token_ids: Callable[[], str]=<function new_event_id>)

Bases: Generic[UserT]

Issue local access tokens by signing a minimal server-owned claim set.

Every claim is chosen here rather than taken from the caller, so a token can only describe the account it was issued for.

async issue(account: LocalAccountState[UserT], *, scopes: AbstractSet[str] = frozenset({}), evidence: AuthenticationEvidence | None = None, now: datetime | None = None) → LocalAccessToken | InvalidCredentials | VerificationUnavailable

Issue one short-lived epoch-bound token without serializing application data.

The token carries only server-owned claims. Application user data stays out of it, so a leaked token reveals nothing beyond the account binding.

Parameters:

• account – The authenticated account to issue for.

• scopes – The scopes to record on the token.

• evidence – Verified authentication assurance to preserve in the access token.

• now – Override the clock, for tests and replayable issuance.

Returns:

The signed token and its lifetime, InvalidCredentials when the account may not be issued for, or VerificationUnavailable when signing or an epoch read failed.

__init__(signer: TokenSigner, issuer: str, audience: str, client_id: str = 'local', lifetime: timedelta = datetime.timedelta(seconds=600), clock: Callable[[], datetime]=<function utc_now>, token_ids: Callable[[], str]=<function new_event_id>) → None

class litestar_security.accounts.LocalAccount(account_id: Annotated[str, msgspec.Meta(description='The stable application-owned account identifier.')], display_name: Annotated[str | None, msgspec.Meta(description='An optional human-readable name to store with the account.')] = None)

Bases: WireStruct

Minimal account projection returned after session login.

class litestar_security.accounts.LocalAccountCapabilities(*args, **kwargs)

Bases: AccountLookup[UserT], PasswordCredentialStore, LoginMethodStore, VerificationTokenStore, RecoveryTokenStore, SecurityEpochStore, Protocol[UserT]

Structural account capabilities required by every local-auth profile.

class litestar_security.accounts.LocalAccountState(account_id: str, normalized_identifier: str, display_name: str | None, active: bool, verified: bool, security_epoch: int, user: UserT | None = None)

Bases: Generic[UserT]

Application-owned account projection needed by local authentication.

__init__(account_id: str, normalized_identifier: str, display_name: str | None, active: bool, verified: bool, security_epoch: int, user: UserT | None = None) → None

class litestar_security.accounts.LocalAuthMode(*values)

Bases: str, Enum

Visible local-authentication transport selection.

class litestar_security.accounts.LocalBearerIdentityResolver(accounts: AccountLookup[UserT])

Bases: Generic[UserT]

Resolve verified local JWT claims through exact account and epoch state.

async resolve(claims: JWTClaims) → Principal[UserT] | InvalidCredentials | VerificationUnavailable

Return a principal only for an active account at the exact current epoch.

Parameters:

claims – The verified claims from the local bearer token.

Returns:

The principal, InvalidCredentials when the account is inactive or the epoch has moved on, or VerificationUnavailable when a lookup failed.

__init__(accounts: AccountLookup[UserT]) → None

class litestar_security.accounts.LocalCredentials(identifier: Annotated[str, msgspec.Meta(description='The account identifier, normally an email address.')], password: Annotated[str, msgspec.Meta(description='The account password.')])

Bases: WireStruct

Password credentials accepted by generated login handlers.

class litestar_security.accounts.LocalIdentifier(identifier: Annotated[str, msgspec.Meta(description='The account identifier, normally an email address.')])

Bases: WireStruct

Typed enumeration-resistant identifier request.

class litestar_security.accounts.LocalInvitationRegistration(identifier: Annotated[str, msgspec.Meta(description='The account identifier, normally an email address.')], password: Annotated[str, msgspec.Meta(description='The replacement password, checked against the configured password policy.')], invitation_token: Annotated[str, msgspec.Meta(description='The single-use invitation token.')], display_name: Annotated[str | None, msgspec.Meta(description='An optional human-readable name to store with the account.')] = None)

Bases: WireStruct

Typed invite-only self-service registration input.

class litestar_security.accounts.LocalMFAChallenge(challenge: Annotated[str, msgspec.Meta(description='The opaque one-time challenge to present at completion.')], account_id: Annotated[str, msgspec.Meta(description='The account bound to this challenge.')], expires_at: Annotated[datetime, msgspec.Meta(description='When the opaque challenge can no longer be used.')], methods: Annotated[tuple[str, ...], msgspec.Meta(description='The permitted second-factor methods.')], code: Annotated[str, msgspec.Meta(description='The stable machine-readable MFA challenge outcome.')] = 'mfa_required', detail: Annotated[str, msgspec.Meta(description='The human-readable MFA challenge outcome.')] = 'Multi-factor authentication is required.')

Bases: WireStruct

A one-time challenge returned when password login requires MFA completion.

class litestar_security.accounts.LocalMFACompletion(challenge: Annotated[str, msgspec.Meta(description='The opaque one-time challenge from password login.')], account_id: Annotated[str, msgspec.Meta(description='The account identifier returned with the challenge.')], method: Annotated[str, msgspec.Meta(description='The selected second-factor method.')], code: Annotated[str, msgspec.Meta(description='The proof for the selected second-factor method.')], method_id: Annotated[str | None, msgspec.Meta(description='The selected TOTP method identifier, when required.')] = None)

Bases: WireStruct

Typed input that completes a pending password-login MFA challenge.

class litestar_security.accounts.LocalPasswordChange(current_password: Annotated[str, msgspec.Meta(description="The caller's current password.")], password: Annotated[str, msgspec.Meta(description='The replacement password, checked against the configured password policy.')], compromise: Annotated[bool, msgspec.Meta(description="Set when the current password is believed compromised. The caller's own session is revoked with the others rather than rebound.")] = False)

Bases: WireStruct

Typed authenticated password-change input.

class litestar_security.accounts.LocalPasswordReset(token: Annotated[str, msgspec.Meta(description='The single-use recovery token.')], password: Annotated[str, msgspec.Meta(description='The replacement password, checked against the configured password policy.')])

Bases: WireStruct

Typed password recovery completion input.

class litestar_security.accounts.LocalRegistration(identifier: Annotated[str, msgspec.Meta(description='The account identifier, normally an email address.')], password: Annotated[str, msgspec.Meta(description='The replacement password, checked against the configured password policy.')], display_name: Annotated[str | None, msgspec.Meta(description='An optional human-readable name to store with the account.')] = None)

Bases: WireStruct

Typed public self-service registration input.

class litestar_security.accounts.LocalSession(session_id: Annotated[str, msgspec.Meta(description='The session identifier, accepted by the revoke route.')], current: Annotated[bool, msgspec.Meta(description='Whether this is the session that made the request.')], created_at: Annotated[datetime, msgspec.Meta(description='When the session was established.')], last_seen_at: Annotated[datetime, msgspec.Meta(description='When the session was last used.')], expires_at: Annotated[datetime, msgspec.Meta(description='When the session expires without further use.')], display_metadata: Annotated[dict[str, str], msgspec.Meta(description='Application-supplied display fields, such as a device label or coarse location.')])

Bases: WireStruct

JSON-safe generated-route session projection.

class litestar_security.accounts.LocalSessionList(sessions: Annotated[tuple[LocalSession, ...], msgspec.Meta(description="The caller's own active sessions.")])

Bases: WireStruct

Safe caller-owned session inventory.

class litestar_security.accounts.LocalToken(token: Annotated[str, msgspec.Meta(description='The opaque token issued by a previous request.')])

Bases: WireStruct

Typed one-time or refresh-token request.

class litestar_security.accounts.LoginMethod(method_id: str, kind: str, created_at: datetime, display_name: str | None = None)

Bases: object

One application-owned viable login method.

__init__(method_id: str, kind: str, created_at: datetime, display_name: str | None = None) → None

class litestar_security.accounts.LoginMethodStore(*args, **kwargs)

Bases: Protocol

Maintain viable login methods through guarded atomic operations.

async register_login_method(account_id: str, method: LoginMethod, *, event: SecurityEvent) → None

Register one login method and its durable event.

Parameters:

• account_id – The account gaining the method.

• method – The method to record.

• event – The audit event to commit with the registration. Rejecting it must fail the registration.

async revoke_login_method(account_id: str, method_id: str, *, require_remaining: bool = True, event: SecurityEvent) → RevokeLoginMethodOutcome

Revoke a method without removing the final viable method by default.

Parameters:

• account_id – The account owning the method.

• method_id – The method to revoke.

• require_remaining – Refuse the revocation when it would leave the account with no way to sign in.

• event – The audit event to commit with the revocation. Rejecting it must fail the revocation.

Returns:

The outcome, distinguishing an absent method from a refused final one.

__init__(*args, **kwargs)

class litestar_security.accounts.NativeSessionAuth(accounts: ~litestar_security.accounts._sessions.NativeSessionStore[~litestar_security.accounts._sessions.UserT], binding: ~litestar_security.accounts._sessions.SessionBindingConfig, resolver: ~litestar_security.accounts._sessions.UserAuthSessionResolver[~litestar_security.accounts._sessions.UserT] | None = None, clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function NativeSessionAuth.<lambda>>, entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>, event_ids: ~collections.abc.Callable[[], str] = <function NativeSessionAuth.<lambda>>)

Bases: Generic[UserT]

Native Litestar session mechanism and fixation-resistant lifecycle service.

extract(connection: ASGIConnection[Any, Any, Any, Any]) → NoCredentials | PresentedCredential[_SessionCredential] | InvalidCredentials

Extract the native authentication payload and independent binding proof once.

Parameters:

connection – The incoming connection.

Returns:

The presented session credential, NoCredentials when the connection carries no session, or InvalidCredentials when what it carries is malformed.

async authenticate(credential: _SessionCredential, connection: ASGIConnection[Any, Any, Any, Any]) → Authenticated[LocalAccountState[UserT]] | InvalidCredentials | VerificationUnavailable

Verify registry, binding, account, and exact epoch state.

Parameters:

• credential – The session identifier and binding proof taken from the connection.

• connection – The incoming connection.

Returns:

The authenticated account, InvalidCredentials when any check fails, or VerificationUnavailable when a dependency failed.

async resolve(claims: LocalAccountState[UserT]) → Principal[UserT]

Resolve an already validated local account without another store call.

Parameters:

claims – The account projection produced by authentication.

Returns:

The principal for the request.

async establish(connection: ASGIConnection[Any, Any, Any, Any], account: LocalAccountState[UserT], *, evidence: AuthenticationEvidence | None = None, display_metadata: Mapping[str, str] = mappingproxy({}), now: datetime | None = None) → SessionAuthentication | VerificationUnavailable

Create or atomically rebind authenticated state and reveal one binding cookie.

A caller that already holds a session gets a new identifier rather than keeping the one it arrived with, which is what defeats session fixation.

Parameters:

• connection – The connection whose session state to write.

• account – The authenticated account to bind the session to.

• evidence – Verified method and trait evidence used to create the session.

• display_metadata – Application-supplied fields to show in the session list.

• now – Override the clock, for tests and replayable establishment.

Returns:

The established session and its reveal-once binding token, or VerificationUnavailable when a dependency failed.

async logout(connection: ASGIConnection[Any, Any, Any, Any], *, now: datetime | None = None) → bool | VerificationUnavailable

Clear local browser state and atomically revoke the current account-owned record.

Parameters:

• connection – The connection whose session state to clear.

• now – Override the clock, for tests and replayable logout.

Returns:

Whether an active session was revoked, or VerificationUnavailable when a dependency failed.

async revoke_session(connection: ASGIConnection[Any, Any, Any, Any], account_id: str, session_id: str, *, now: datetime | None = None) → bool | VerificationUnavailable

Atomically revoke one caller-owned session and clear it when current.

Parameters:

• connection – The connection whose session state to clear if it is the target.

• account_id – The authenticated caller’s account.

• session_id – The session to revoke.

• now – Override the clock, for tests and replayable revocation.

Returns:

Whether an active session was revoked, or VerificationUnavailable when a dependency failed. A session owned by another account is reported as not revoked rather than as a distinct failure.

async list_sessions(account_id: str, *, current_session_id: str | None = None) → tuple[SessionSummary, ...]

Return only safe account-session inventory projections.

Parameters:

• account_id – The account whose sessions to list.

• current_session_id – The caller’s own session, flagged as current in the result.

Returns:

Summaries carrying no binding material, filtered to the named account.

current_authentication(connection: ASGIConnection[Any, Any, Any, Any]) → SessionAuthentication | None

Return the strictly decoded current local-session projection.

Parameters:

connection – The connection to read session state from.

Returns:

The current session projection, or None when the connection carries no session or a malformed one.

prepare_password_rebind(connection: ASGIConnection[Any, Any, Any, Any], account: LocalAccountState[UserT], *, now: datetime | None = None) → SessionRebindPlan | VerificationUnavailable

Prepare reveal-once browser material without mutating registry or session state.

Preparation is deliberately separate from activation: the replacement session must not exist until the password mutation it accompanies has committed.

Parameters:

• connection – The connection whose session is being replaced.

• account – The account the replacement session will bind to.

• now – Override the clock, for tests and replayable preparation.

Returns:

The plan to hand to activate_password_rebind(), or VerificationUnavailable when the caller has no usable session.

async activate_password_rebind(connection: ASGIConnection[Any, Any, Any, Any], plan: SessionRebindPlan, security_epoch: int) → bool

Activate only a replacement record already accepted by the atomic password mutation.

Parameters:

• connection – The connection whose session state to rewrite.

• plan – The plan returned by prepare_password_rebind().

• security_epoch – The epoch the password mutation committed at.

Returns:

True when the replacement session became the connection’s session.

__init__(accounts: ~litestar_security.accounts._sessions.NativeSessionStore[~litestar_security.accounts._sessions.UserT], binding: ~litestar_security.accounts._sessions.SessionBindingConfig, resolver: ~litestar_security.accounts._sessions.UserAuthSessionResolver[~litestar_security.accounts._sessions.UserT] | None = None, clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function NativeSessionAuth.<lambda>>, entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>, event_ids: ~collections.abc.Callable[[], str] = <function NativeSessionAuth.<lambda>>) → None

class litestar_security.accounts.NativeSessionStore(*args, **kwargs)

Bases: SessionRegistry, Protocol[UserT]

Combined account, epoch, and session capabilities for native authentication.

async get_by_id(account_id: str) → LocalAccountState[UserT] | None

Load one local account projection.

Parameters:

account_id – The account named by the session.

Returns:

The account projection, or None when the account no longer exists.

async current_epoch(account_id: str) → int | None

Load the authoritative account security epoch.

Parameters:

account_id – The account whose epoch to read.

Returns:

The current epoch, or None when the account does not exist.

class litestar_security.accounts.NoOpSecurityEventSink

Bases: object

Accept security events when an application has not configured a sink.

async emit(event: SecurityEvent) → None

Discard one already-sanitized observational event.

Parameters:

event – The event to discard.

__init__() → None

class litestar_security.accounts.NotificationCommand(template: str, destination: str, token: str, expires_at: datetime, return_url: str | None = None)

Bases: object

Delivery-neutral notification data with a one-time opaque token.

__init__(template: str, destination: str, token: str, expires_at: datetime, return_url: str | None = None) → None

class litestar_security.accounts.OperationMessage(detail: Annotated[str, msgspec.Meta(description='A human-readable outcome that never names an account.')])

Bases: WireStruct

Stable generated-route status body, shared by every route in the tree.

class litestar_security.accounts.PasskeyAuthenticationStart(account_id: str)

Bases: WireStruct

Request bound passkey authentication options.

class litestar_security.accounts.PasskeyOptions(options: str, expires_at: datetime, binding: str | None = None)

Bases: WireStruct

Carry dependency-independent WebAuthn JSON options.

class litestar_security.accounts.PasskeyRegistrationStart(user_name: str, step_up_grant: str)

Bases: WireStruct

Request bound passkey registration options.

class litestar_security.accounts.PasskeySummary(credential_id: str, display_name: str | None, created_at: datetime, last_used_at: datetime | None, backup_eligible: bool, backup_state: bool, suspect: bool)

Bases: WireStruct

Safe caller-owned credential metadata.

class litestar_security.accounts.PasskeyVerification(account_id: str, response: str, binding: str | None = None, transport: str | None = None)

Bases: WireStruct

Submit one browser WebAuthn JSON response.

class litestar_security.accounts.PasswordChangeOutcome(status: PasswordChangeStatus, security_epoch: int | None = None)

Bases: object

Atomic password replacement and security-epoch outcome.

__init__(status: PasswordChangeStatus, security_epoch: int | None = None) → None

class litestar_security.accounts.PasswordChangeStatus(*values)

Bases: str, Enum

Atomic password-change outcomes.

class litestar_security.accounts.PasswordCredentialState(password_hash: str, security_epoch: int, active: bool, verified: bool)

Bases: object

Atomic password hash, account-state, and epoch snapshot for reauthentication.

__init__(password_hash: str, security_epoch: int, active: bool, verified: bool) → None

class litestar_security.accounts.PasswordCredentialStore(*args, **kwargs)

Bases: Protocol

Store password credentials through atomic security operations.

async get_password_state(account_id: str) → PasswordCredentialState | None

Load one atomic password hash, account-state, and security-epoch snapshot.

Read the hash, active/verified projection, and epoch in one operation. Values read separately can describe a state that never existed during a concurrent deactivation or verification-state change.

Parameters:

account_id – The account whose credential state to read.

Returns:

The paired hash, account-state projection, and epoch, or None when the account has no password.

async compare_and_replace_password(account_id: str, expected_hash: str, password_hash: str, *, event: SecurityEvent) → bool

Atomically replace a hash only when its expected value is current.

The comparison is what makes concurrent changes safe, so it must happen inside the same operation as the write.

Parameters:

• account_id – The account whose password to replace.

• expected_hash – The hash the caller read and expects to still be stored.

• password_hash – The replacement hash.

• event – The audit event to commit with the replacement. Rejecting it must fail the replacement.

Returns:

True when the stored hash matched and was replaced, False when it had already changed.

async replace_password_and_bump_epoch(account_id: str, password_hash: str, *, expected_epoch: int, event: SecurityEvent) → PasswordChangeOutcome

Atomically replace a password and increment the security epoch.

Advancing the epoch is what invalidates credentials issued before the change, so it must commit with the new hash or not at all.

Parameters:

• account_id – The account whose password to replace.

• password_hash – The replacement hash.

• expected_epoch – The epoch the caller read; a different stored epoch is a conflict.

• event – The audit event to commit with the replacement. Rejecting it must fail the replacement.

Returns:

The outcome, carrying the new epoch only when the replacement committed.

__init__(*args, **kwargs)

class litestar_security.accounts.PasswordPolicyViolation(*values)

Bases: str, Enum

Secret-free reasons a candidate password does not satisfy policy.

class litestar_security.accounts.PasswordReauthenticationProof(account_id: str, security_epoch: int, authenticated_at: datetime, expires_at: datetime)

Bases: object

Account- and epoch-bound recent password proof for sensitive mutation.

__init__(account_id: str, security_epoch: int, authenticated_at: datetime, expires_at: datetime) → None

class litestar_security.accounts.PasswordResetOutcome(status: PasswordResetStatus, account_id: str | None = None, security_epoch: int | None = None)

Bases: object

Atomic recovery-token consumption and password-reset outcome.

__init__(status: PasswordResetStatus, account_id: str | None = None, security_epoch: int | None = None) → None

class litestar_security.accounts.PasswordResetStatus(*values)

Bases: str, Enum

Atomic recovery-token/password-reset outcomes.

class litestar_security.accounts.PasswordVerificationStatus(*values)

Bases: str, Enum

Sanitized outcomes from one constant-work password verification.

class litestar_security.accounts.PendingTokenIssue(token_id: str, digest: bytes, purpose: TokenPurpose, expires_at: datetime, maximum_attempts: int)

Bases: object

Account-unbound hashed token material for one atomic registration.

bind(account_id: str, *, security_epoch: int | None = None) → TokenIssue

Bind this material to an application-allocated account ID.

Parameters:

• account_id – The identifier the application allocated for the new account.

• security_epoch – The epoch the account was created at.

Returns:

The bound issue, ready for the atomic store call.

__init__(token_id: str, digest: bytes, purpose: TokenPurpose, expires_at: datetime, maximum_attempts: int) → None

class litestar_security.accounts.PurposeTokenCodec(pepper: bytes, entropy: Callable[[int], bytes] = <function token_bytes>)

Bases: object

Generate and verify strict purpose-bound opaque one-time tokens.

issue(purpose: TokenPurpose, *, now: datetime, lifetime: timedelta, template: str, destination: str, return_url: str | None = None, maximum_attempts: int = 5) → PurposeTokenDelivery

Create one digest-bound issue whose raw token exists only in its notification.

The raw token appears only in the notification. Storage keeps its digest, so a leaked database cannot be replayed against these routes.

Parameters:

• purpose – The closed namespace the token is bound to.

• now – The issue timestamp.

• lifetime – How long the token stays valid.

• template – The notification template the application renders.

• destination – Where the notification is delivered.

• return_url – An approved callback to embed, or None.

• maximum_attempts – How many consume attempts the store should allow.

Returns:

The storage issue paired with the notification carrying the raw token.

proof(token: object, *, expected_purpose: TokenPurpose) → PurposeTokenProof | None

Return a storage proof after one HMAC work class, or generic invalid.

Every rejection costs the same HMAC work, so timing does not separate a malformed token from a well-formed one for another purpose.

Parameters:

• token – The presented token, of any type.

• expected_purpose – The namespace the token must be bound to.

Returns:

The storage-facing proof, or None for any rejection.

__init__(pepper: bytes, entropy: Callable[[int], bytes] = <function token_bytes>) → None

class litestar_security.accounts.PurposeTokenDelivery

Bases: object

Codec-created storage issue and durable notification outbox plan.

__init__() → None

Prevent callers from bypassing codec-owned digest binding.

bind(account_id: str, *, security_epoch: int | None = None) → tuple[TokenIssue, NotificationCommand]

Bind the storage material while preserving the codec-created notification.

Parameters:

• account_id – The identifier the application allocated for the new account.

• security_epoch – The epoch the account was created at.

Returns:

The bound issue and the notification to deliver.

exception litestar_security.accounts.PurposeTokenGenerationError

Bases: RuntimeError

Indicate that one-time token material could not be generated safely.

__init__() → None

Initialize a stable secret-free error.

class litestar_security.accounts.PurposeTokenProof(token_id: str, digest: bytes, purpose: TokenPurpose)

Bases: object

Secret-free parsed lookup and HMAC proof passed to an atomic store.

__init__(token_id: str, digest: bytes, purpose: TokenPurpose) → None

class litestar_security.accounts.RateLimitAttempt(operation: str, client_key: str | None = None, subject_digest: str | None = None, cost: int = 1)

Bases: object

One bucketed attempt presented to a limiter.

Parameters:

• operation – Canonical local.* name of the entry point being consumed.

• client_key – Application-supplied trusted client identity, or None to skip the client bucket.

• subject_digest – Peppered digest of the normalized identifier, or None when the operation carries no identifier. Never a raw identifier.

• cost – Units this attempt consumes from each bucket.

__init__(operation: str, client_key: str | None = None, subject_digest: str | None = None, cost: int = 1) → None

class litestar_security.accounts.RateLimitDecision(allowed: bool, retry_after: int | None = None)

Bases: object

One limiter verdict for a single attempt.

Parameters:

• allowed – Whether the attempt may proceed.

• retry_after – Whole seconds until the caller may retry, reported only when the attempt was denied.

__init__(allowed: bool, retry_after: int | None = None) → None

class litestar_security.accounts.RateLimitGuard(limiter: RateLimiter, pepper: bytes, events: SecurityEventSink = <factory>, clock: Callable[[], datetime]=<function utc_now>, event_ids: Callable[[], str]=<function new_event_id>)

Bases: object

Bucket one operation’s attempts without exposing the identifier.

Every service that limits an entry point shares one guard, so denial audit events are constructed the same way everywhere instead of once per service.

Parameters:

• limiter – The configured budget implementation.

• pepper – Secret used to derive subject digests; at least 32 bytes.

• events – Sink notified when an attempt is denied.

• clock – Source of the current time for denial events.

• event_ids – Factory for unique denial event identifiers.

async check(operation: str, *, client_key: str | None = None, identifier: str | None = None) → RateLimited | VerificationUnavailable | None

Consume one attempt, returning None when the caller may proceed.

Parameters:

• operation – The rate-limited operation name.

• client_key – The caller identity for the client bucket, or None to skip it.

• identifier – The submitted account identifier for the subject bucket, or None to skip it. It is digested before it reaches the limiter.

Returns:

None when the attempt may proceed, RateLimited when the budget is spent, or VerificationUnavailable when the limiter failed. A limiter outage fails closed, so it can never silently remove the limit.

subject_digest(identifier: str) → str

Derive the stable peppered bucket digest for one normalized identifier.

Parameters:

identifier – The normalized account identifier.

Returns:

The hex digest used as a bucket key, so a limiter backend never stores identifiers.

__init__(limiter: RateLimiter, pepper: bytes, events: SecurityEventSink = <factory>, clock: Callable[[], datetime]=<function utc_now>, event_ids: Callable[[], str]=<function new_event_id>) → None

class litestar_security.accounts.RateLimitPolicy(limit: int, window: timedelta)

Bases: object

One operation’s budget, applied to each bucket independently.

Parameters:

• limit – Attempts allowed per window in a single bucket.

• window – Length of the fixed window the limit applies to.

__init__(limit: int, window: timedelta) → None

class litestar_security.accounts.RateLimited(retry_after: int | None = None, code: str = 'rate_limited')

Bases: object

Sanitized outcome returned when an operation exhausted its budget.

Parameters:

• retry_after – Whole seconds until the caller may retry, when the limiter reported one.

• code – Stable machine-readable reason.

__init__(retry_after: int | None = None, code: str = 'rate_limited') → None

class litestar_security.accounts.RateLimiter(*args, **kwargs)

Bases: Protocol

Application-owned budget for one abuse-prone operation.

Implementations MUST consume atomically: N concurrent acquire() calls against the same bucket under a policy limit of k must admit exactly k, never more or fewer. A limiter that raises is treated as unavailable and fails closed, so raising is the correct response to a backend outage.

async acquire(request: RateLimitAttempt) → RateLimitDecision

Consume one attempt’s cost and report whether it may proceed.

Parameters:

request – The operation, buckets, and cost to charge.

Returns:

The decision, carrying retry_after seconds when it denies.

Raises:

Exception – Any failure signals an outage. The caller fails closed and answers 503 rather than letting the limit lapse.

__init__(*args, **kwargs)

class litestar_security.accounts.RecoveryCodes(codes: tuple[str, ...])

Bases: WireStruct

Reveal a replacement recovery-code set once.

class litestar_security.accounts.RecoveryTokenStore(*args, **kwargs)

Bases: Protocol

Issue and atomically consume password-recovery tokens.

async issue(issue: TokenIssue, notification: NotificationCommand, *, event: SecurityEvent) → None

Commit a recovery issue, notification, and durable event.

Store the digest the issue carries, never the token itself: the token is the secret sent to the account holder.

Parameters:

• issue – The token digest, account binding, and expiry to store.

• notification – The delivery the application should send.

• event – The audit event to commit with the issue. Rejecting it must fail the issue.

async issue_absent() → None

Perform one durable round trip that commits nothing.

Called instead of issue() when the identifier resolves to no eligible account. The durable step MUST cost the same whether or not the identifier resolves: an implementation that answers quickly for unknown accounts makes a present account measurably slower to probe, defeating the shared-response guarantee. Commit, notify, and mutate nothing.

async consume_and_reset(token_id: str, digest: bytes, new_password_hash: str, *, now: datetime, event: SecurityEvent) → PasswordResetOutcome

Consume only at its issued epoch, then reset password and advance epoch atomically.

The epoch check is what stops a stale recovery token from undoing a password change made after the token was issued.

Parameters:

• token_id – The identifier carried by the presented token.

• digest – The digest to compare against the stored one.

• new_password_hash – The encoded replacement hash.

• now – The timestamp to evaluate expiry against.

• event – The audit event to commit with the reset. Rejecting it must fail the reset.

Returns:

The outcome, carrying the account and its new epoch only when reset.

__init__(*args, **kwargs)

class litestar_security.accounts.RefreshFamilyContext(account_id: str, family_id: str, security_epoch: int, token_expires_at: datetime, family_expires_at: datetime, scopes: frozenset[str] = frozenset({}), evidence: AuthenticationEvidence | None = None)

Bases: object

Secret-free preflight state revalidated by the atomic rotation call.

__init__(account_id: str, family_id: str, security_epoch: int, token_expires_at: datetime, family_expires_at: datetime, scopes: frozenset[str] = frozenset({}), evidence: AuthenticationEvidence | None = None) → None

class litestar_security.accounts.RefreshPreflightOutcome(status: RefreshRotationStatus, family_revoked: bool = False)

Bases: object

Proof-checked negative preflight outcome with exact revocation evidence.

__init__(status: RefreshRotationStatus, family_revoked: bool = False) → None

class litestar_security.accounts.RefreshReceiptContext(token_id: str, family_id: str, account_id: str, security_epoch: int, idempotency_digest: bytes | None = None)

Bases: object

Public receipt binding values; no raw credential material is retained.

__init__(token_id: str, family_id: str, account_id: str, security_epoch: int, idempotency_digest: bytes | None = None) → None

class litestar_security.accounts.RefreshReceiptKey(key_id: str, key: bytes)

Bases: object

One AES-256-GCM receipt key selected by a non-secret key ID.

__init__(key_id: str, key: bytes) → None

class litestar_security.accounts.RefreshReceiptReplay(context: RefreshFamilyContext, sealed_receipt: bytes)

Bases: object

Proof-checked same-key replay recoverable without speculative crypto.

__init__(context: RefreshFamilyContext, sealed_receipt: bytes) → None

class litestar_security.accounts.RefreshReceiptSealer(active_key: ~litestar_security.accounts._receipts.RefreshReceiptKey, retained_keys: tuple[~litestar_security.accounts._receipts.RefreshReceiptKey, ...] = (), entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>)

Bases: object

Seal exact refresh responses with rotating AES-GCM keys and bound AAD.

seal(response: TokenPair, context: RefreshReceiptContext, *, expires_at: datetime) → bytes

Seal one exact response and authenticate all replay decision fields.

The replay decision fields are authenticated as associated data, so a receipt cannot be moved to a different family or token.

Parameters:

• response – The token pair to seal.

• context – The family, token, and idempotency binding to authenticate.

• expires_at – When the receipt stops being replayable.

Returns:

The sealed receipt bytes for the store to keep with the family.

unseal(sealed_receipt: bytes, context: RefreshReceiptContext, *, now: datetime) → TokenPair | InvalidCredentials

Recover one response only while its bound receipt and key remain valid.

Parameters:

• sealed_receipt – The stored receipt bytes.

• context – The binding the receipt must authenticate against.

• now – The timestamp to evaluate the receipt window against.

Returns:

The original token pair, or InvalidCredentials when the receipt is expired, bound elsewhere, or sealed under a key no longer retained.

__init__(active_key: ~litestar_security.accounts._receipts.RefreshReceiptKey, retained_keys: tuple[~litestar_security.accounts._receipts.RefreshReceiptKey, ...] = (), entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>) → None

class litestar_security.accounts.RefreshRotationOutcome(status: RefreshRotationStatus, sealed_receipt: bytes | None = None, family_revoked: bool = False)

Bases: object

Atomic strict rotation, idempotent receipt, or replay outcome.

__init__(status: RefreshRotationStatus, sealed_receipt: bytes | None = None, family_revoked: bool = False) → None

class litestar_security.accounts.RefreshRotationStatus(*values)

Bases: str, Enum

Atomic refresh-token rotation outcomes.

class litestar_security.accounts.RefreshTokenCodec(pepper: bytes, entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>)

Bases: object

Issue and verify opaque refresh tokens while storing only HMAC digests.

issue() → RefreshTokenIssue

Create one lookup/secret pair and its storage-safe digest.

Returns:

The reveal-once token alongside the digest to store. The secret half is never recoverable from what is stored.

verify(refresh_token: str) → RefreshTokenProof | InvalidCredentials

Parse one canonical token while keeping malformed work in the HMAC class.

Parameters:

refresh_token – The presented opaque token.

Returns:

The parsed lookup and digest, or InvalidCredentials. Every rejection costs the same work.

digest_idempotency_key(token_id: str, value: str) → bytes | InvalidCredentials

Hash one canonical key carrying at least 128 bits of caller entropy.

Parameters:

• token_id – The token the key is scoped to, so a key cannot be reused across tokens.

• value – The caller’s Idempotency-Key header.

Returns:

The digest to compare against the stored one, or InvalidCredentials when the key carries too little entropy to be safe.

__init__(pepper: bytes, entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>) → None

class litestar_security.accounts.RefreshTokenFamilyStore(*args, **kwargs)

Bases: Protocol

Atomic strict refresh-family rotation and revocation boundary.

async create_family(command: CreateRefreshFamilyCommand, *, event: SecurityEvent) → bool

Create one family only if its account epoch is still current, atomically.

Parameters:

• command – The family identifier, account binding, epoch, and first token digest.

• event – The audit event to commit with the family. Rejecting it must fail the creation.

Returns:

True when the family was created, False when the account epoch had already moved on.

async prepare_rotation(proof: RefreshTokenProof, idempotency_digest: bytes | None, *, now: datetime, event: SecurityEvent) → RefreshFamilyContext | RefreshReceiptReplay | RefreshPreflightOutcome

Atomically return active state, recover a receipt, or revoke and record consumed reuse.

This is where reuse detection lives. A token that was already consumed means the token leaked, so the whole family must be revoked in the same operation that observes the reuse.

Parameters:

• proof – The verified identifier and digest of the presented token.

• idempotency_digest – The digest of the caller’s Idempotency-Key, or None when the caller sent none.

• now – The timestamp to evaluate expiry against.

• event – The audit event to commit with the outcome. Rejecting it must fail the preparation.

Returns:

The active family context to rotate from, a stored receipt when the caller is retrying with a matching idempotency key, or a result describing why rotation cannot proceed.

async rotate(command: RotateRefreshCommand, *, now: datetime, event: SecurityEvent) → RefreshRotationOutcome

Atomically revalidate context/current epoch and rotate or revoke.

Revalidate rather than trusting the prepared context: the epoch can move between preparation and rotation.

Parameters:

• command – The family, expected prior token, replacement digest, and receipt to store.

• now – The commit timestamp.

• event – The audit event to commit with the rotation. Rejecting it must fail the rotation.

Returns:

The outcome, distinguishing a committed rotation from a revocation.

async revoke_family(family_id: str, *, event: SecurityEvent) → bool

Revoke one refresh-token family.

Parameters:

• family_id – The family to revoke.

• event – The audit event to commit with the revocation. Rejecting it must fail the revocation.

Returns:

True when an active family was revoked.

async revoke_token(token_id: str, token_digest: bytes, *, event: SecurityEvent) → bool

Revoke the family owning one exact presented token.

Parameters:

• token_id – The identifier carried by the presented token.

• token_digest – The digest that must match the stored one.

• event – The audit event to commit with the revocation. Rejecting it must fail the revocation.

Returns:

True when the digest matched and the family was revoked.

async revoke_token_for_account(account_id: str, token_id: str, token_digest: bytes, *, event: SecurityEvent) → bool

Revoke one exact token only when its family belongs to the caller account.

Check ownership inside this operation. A caller must not be able to revoke another account’s token by presenting its identifier.

Parameters:

• account_id – The authenticated caller’s account.

• token_id – The identifier carried by the presented token.

• token_digest – The digest that must match the stored one.

• event – The audit event to commit with the revocation. Rejecting it must fail the revocation.

Returns:

True when the caller owned the family and it was revoked.

async revoke_for_account(account_id: str, *, event: SecurityEvent) → int

Revoke every refresh family for an account.

Parameters:

• account_id – The account whose families to revoke.

• event – The audit event to commit with the revocations. Rejecting it must fail them.

Returns:

The number of active families revoked.

__init__(*args, **kwargs)

class litestar_security.accounts.RefreshTokenIssue(refresh_token: str, token_id: str, digest: bytes)

Bases: object

Reveal-once opaque refresh token plus storage-safe material.

__init__(refresh_token: str, token_id: str, digest: bytes) → None

class litestar_security.accounts.RefreshTokenProof(token_id: str, digest: bytes)

Bases: object

Parsed refresh-token lookup and fixed-size domain-separated digest.

__init__(token_id: str, digest: bytes) → None

class litestar_security.accounts.RefreshTokenService(accounts: object, store: ~litestar_security.accounts._refresh.RefreshTokenFamilyStore, codec: ~litestar_security.accounts._refresh_tokens.RefreshTokenCodec, receipts: ~litestar_security.accounts._receipts.RefreshReceiptSealer, access_tokens: LocalAccessTokenIssuer[UserT], idle_lifetime: ~datetime.timedelta = datetime.timedelta(days=7), absolute_lifetime: ~datetime.timedelta = datetime.timedelta(days=30), receipt_window: ~datetime.timedelta = datetime.timedelta(seconds=30), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function utc_now>, family_ids: ~collections.abc.Callable[[], str] = <function _new_refresh_family_id>, event_ids: ~collections.abc.Callable[[], str] = <function _new_refresh_event_id>, rate_limits: ~litestar_security.accounts._rate_limits.RateLimitGuard | None = None)

Bases: Generic[UserT]

Issue, strictly rotate, and revoke opaque local refresh families.

async issue(account: LocalAccountState[UserT], *, scopes: Set[str] = frozenset({}), evidence: AuthenticationEvidence | None = None, now: datetime | None = None) → TokenPair | InvalidCredentials | VerificationUnavailable

Create the initial family before revealing either credential.

Parameters:

• account – The authenticated account to issue for. It must be active and verified.

• scopes – The scopes to bind into the access token.

• evidence – Verified authentication assurance to preserve in the initial access token.

• now – Override the clock, for tests and replayable issuance.

Returns:

The token pair, InvalidCredentials when the account may not be issued for, or VerificationUnavailable when a dependency failed.

async rotate(refresh_token: str, *, idempotency_key: str | None = None, now: datetime | None = None, client_key: str | None = None) → TokenPair | RateLimited | InvalidCredentials | VerificationUnavailable

Return exactly the store-accepted sealed response or one safe failure.

Only the client bucket applies: the presented value is a refresh token, and digesting it into a bucket key would let a limiter backend become a record of which tokens were attempted.

Parameters:

• refresh_token – The opaque token presented by the client.

• idempotency_key – Replays a lost response instead of tripping reuse detection, when it matches the key sent with the original request.

• now – Override the clock, for tests and replayable rotation.

• client_key – The caller identity for the rate-limit bucket, or None to skip client-keyed limiting.

Returns:

The rotated pair, RateLimited when the budget is spent, InvalidCredentials when the token is rejected or was reused, or VerificationUnavailable when a dependency failed.

async revoke(refresh_token: str, *, now: datetime | None = None) → bool | InvalidCredentials | VerificationUnavailable

Revoke the family owning one exact presented opaque token.

Parameters:

• refresh_token – The opaque token whose family to revoke.

• now – Override the clock, for tests and replayable revocation.

Returns:

Whether an active family was revoked, InvalidCredentials when the token is rejected, or VerificationUnavailable when the store failed.

async revoke_for_account(account_id: str, refresh_token: str, *, now: datetime | None = None) → bool | InvalidCredentials | VerificationUnavailable

Revoke one caller-owned refresh family without exposing cross-account state.

Parameters:

• account_id – The authenticated caller’s account.

• refresh_token – The opaque token whose family to revoke.

• now – Override the clock, for tests and replayable revocation.

Returns:

Whether an active family was revoked, InvalidCredentials when the token is rejected, or VerificationUnavailable when the store failed. A token owned by another account is reported as not revoked rather than as a distinct failure.

__init__(accounts: object, store: ~litestar_security.accounts._refresh.RefreshTokenFamilyStore, codec: ~litestar_security.accounts._refresh_tokens.RefreshTokenCodec, receipts: ~litestar_security.accounts._receipts.RefreshReceiptSealer, access_tokens: LocalAccessTokenIssuer[UserT], idle_lifetime: ~datetime.timedelta = datetime.timedelta(days=7), absolute_lifetime: ~datetime.timedelta = datetime.timedelta(days=30), receipt_window: ~datetime.timedelta = datetime.timedelta(seconds=30), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function utc_now>, family_ids: ~collections.abc.Callable[[], str] = <function _new_refresh_family_id>, event_ids: ~collections.abc.Callable[[], str] = <function _new_refresh_event_id>, rate_limits: ~litestar_security.accounts._rate_limits.RateLimitGuard | None = None) → None

class litestar_security.accounts.RegistrationCommand(normalized_identifier: str, display_name: str | None = None)

Bases: object

Application-neutral local registration input.

__init__(normalized_identifier: str, display_name: str | None = None) → None

class litestar_security.accounts.RegistrationMode(*values)

Bases: str, Enum

Supported self-service registration policies.

class litestar_security.accounts.RegistrationOutcome(status: RegistrationStatus, account: LocalAccountState[UserT] | None = None)

Bases: Generic[UserT]

Atomic registration outcome.

__init__(status: RegistrationStatus, account: LocalAccountState[UserT] | None = None) → None

class litestar_security.accounts.RegistrationPolicy(mode: RegistrationMode, require_verification: bool = True)

Bases: object

Explicit self-service registration policy.

classmethod disabled() → RegistrationPolicy

Disable self-service registration.

Returns:

A policy that generates no registration route.

classmethod public(*, require_verification: bool = True) → RegistrationPolicy

Enable public self-service registration.

Parameters:

require_verification – Issue a verification token with the account and leave the account unverified until that token is consumed.

Returns:

A policy that generates an open registration route.

classmethod invite_only(*, require_verification: bool = True) → RegistrationPolicy

Require an atomic invitation consume during registration.

Parameters:

require_verification – Issue a verification token with the account and leave the account unverified until that token is consumed.

Returns:

A policy whose registration route additionally requires an invitation token.

__init__(mode: RegistrationMode, require_verification: bool = True) → None

class litestar_security.accounts.RegistrationStatus(*values)

Bases: str, Enum

Atomic registration outcomes.

class litestar_security.accounts.RegistrationStore(*args, **kwargs)

Bases: Protocol[UserT]

Create an account and consume any invitation atomically.

async register(command: RegistrationCommand, password_hash: str, *, invitation_digest: bytes | None, verification: PurposeTokenDelivery | None, now: datetime, event: SecurityEvent) → RegistrationOutcome[UserT]

Commit registration, invitation, verification, notification, and event.

Every part commits together. Creating the account but failing to consume the invitation would let one invitation create unlimited accounts.

Parameters:

• command – The normalized identifier and display name to register.

• password_hash – The encoded hash for the new account.

• invitation_digest – The digest of the presented invitation to consume, or None under a policy that requires no invitation.

• verification – The verification token and notification to store with the account, or None when the policy requires no verification.

• now – The commit timestamp.

• event – The audit event to commit with the registration. Rejecting it must fail the registration.

Returns:

The outcome, carrying the account projection only when it was created.

__init__(*args, **kwargs)

class litestar_security.accounts.ResolvedUserAuthSession(session: UserAuthSession, account: LocalAccountState[UserT])

Bases: Generic[UserT]

One consistent session and account read produced by an application store.

__init__(session: UserAuthSession, account: LocalAccountState[UserT]) → None

class litestar_security.accounts.RevokeLoginMethodOutcome(status: RevokeLoginMethodStatus)

Bases: object

Atomic login-method revocation outcome.

__init__(status: RevokeLoginMethodStatus) → None

class litestar_security.accounts.RevokeLoginMethodStatus(*values)

Bases: str, Enum

Atomic login-method revocation outcomes.

class litestar_security.accounts.RotateRefreshCommand(token_id: str, token_digest: bytes, account_id: str, family_id: str, security_epoch: int, successor_id: str, successor_digest: bytes, successor_expires_at: datetime, family_expires_at: datetime, sealed_receipt: bytes, receipt_expires_at: datetime, idempotency_digest: bytes | None = None, scopes: frozenset[str] = frozenset({}), evidence: AuthenticationEvidence | None = None)

Bases: object

Candidate one-time refresh rotation passed to an atomic store.

__init__(token_id: str, token_digest: bytes, account_id: str, family_id: str, security_epoch: int, successor_id: str, successor_digest: bytes, successor_expires_at: datetime, family_expires_at: datetime, sealed_receipt: bytes, receipt_expires_at: datetime, idempotency_digest: bytes | None = None, scopes: frozenset[str] = frozenset({}), evidence: AuthenticationEvidence | None = None) → None

class litestar_security.accounts.SecurityEpochStore(*args, **kwargs)

Bases: Protocol

Resolve the exact current account security epoch.

async current_epoch(account_id: str) → int | None

Return the current epoch or None for an absent account.

Read authoritative state rather than a cache: a stale epoch keeps revoked credentials working.

Parameters:

account_id – The account whose epoch to read.

Returns:

The current epoch, or None when the account does not exist.

__init__(*args, **kwargs)

class litestar_security.accounts.SecurityEpochValidator(store: SecurityEpochStore)

Bases: object

Validate one presented epoch against authoritative application state.

async validate(account_id: str, presented_epoch: int) → InvalidCredentials | VerificationUnavailable | None

Return None only when the exact current epoch matches.

Parameters:

• account_id – The account named by the presented credential.

• presented_epoch – The epoch the credential was issued at.

Returns:

None when the credential is still current, InvalidCredentials when the epoch has moved on, and VerificationUnavailable when the store could not be read.

__init__(store: SecurityEpochStore) → None

class litestar_security.accounts.SecurityEvent(event_id: str, occurred_at: datetime, operation: str, outcome: str, account_id: str | None = None, principal_id: str | None = None, mechanism: str | None = None, session_id: str | None = None, family_id: str | None = None, correlation: Mapping[str, str]=<factory>)

Bases: object

Secret-free event committed with a security decision or mutation.

__init__(event_id: str, occurred_at: datetime, operation: str, outcome: str, account_id: str | None = None, principal_id: str | None = None, mechanism: str | None = None, session_id: str | None = None, family_id: str | None = None, correlation: Mapping[str, str]=<factory>) → None

class litestar_security.accounts.SecurityEventSink(*args, **kwargs)

Bases: Protocol

Application-owned sink for secret-free, non-transactional decisions.

async emit(event: SecurityEvent) → None

Record one sanitized security decision.

Events are secret-free by construction, so a sink may forward them anywhere. Observational events cannot change the decision they describe; raising from one is logged and dropped.

Parameters:

event – The event to record.

__init__(*args, **kwargs)

class litestar_security.accounts.SessionAuthentication(session_id: str, binding_id: str, account_id: str, security_epoch: int, authenticated_at: datetime, expires_at: datetime, assurance_expires_at: datetime | None = None, methods: frozenset[str] = frozenset({'password'}), traits: frozenset[str] = frozenset({'session'}), amr: tuple[str, ...] = ('pwd',))

Bases: object

Authentication state stored inside the native Litestar session.

__init__(session_id: str, binding_id: str, account_id: str, security_epoch: int, authenticated_at: datetime, expires_at: datetime, assurance_expires_at: datetime | None = None, methods: frozenset[str] = frozenset({'password'}), traits: frozenset[str] = frozenset({'session'}), amr: tuple[str, ...] = ('pwd',)) → None

class litestar_security.accounts.SessionBindingConfig(pepper: bytes, cookie_name: str = '__Host-litestar-security-binding', secure: bool = True, same_site: Literal['lax', 'strict', 'none'] = 'lax', path: str = '/', domain: str | None = None, max_age: int = 1209600, touch_interval: timedelta = datetime.timedelta(seconds=300), preserve_session_keys: tuple[str, ...] = (), allow_insecure: bool = False)

Bases: object

Independent proof-of-possession cookie configuration.

__init__(pepper: bytes, cookie_name: str = '__Host-litestar-security-binding', secure: bool = True, same_site: Literal['lax', 'strict', 'none'] = 'lax', path: str = '/', domain: str | None = None, max_age: int = 1209600, touch_interval: timedelta = datetime.timedelta(seconds=300), preserve_session_keys: tuple[str, ...] = (), allow_insecure: bool = False) → None

class litestar_security.accounts.SessionBindingProof(binding_id: str, digest: bytes)

Bases: object

Parsed binding lookup and domain-separated digest without the raw secret.

__init__(binding_id: str, digest: bytes) → None

class litestar_security.accounts.SessionRebindPlan(prior_session_id: str, command: CreateSessionCommand, binding_token: str, authenticated_at: datetime)

Bases: object

Reveal-once browser state prepared for an atomic password-session rebind.

__init__(prior_session_id: str, command: CreateSessionCommand, binding_token: str, authenticated_at: datetime) → None

class litestar_security.accounts.SessionRegistry(*args, **kwargs)

Bases: Protocol

Atomic authenticated-session inventory and revocation boundary.

async create(command: CreateSessionCommand, *, event: SecurityEvent) → UserAuthSession

Create a registry record with its durable event.

Parameters:

• command – The session identifier, account binding, epoch, and lifetime to store.

• event – The audit event to commit with the record. Rejecting it must fail the creation.

Returns:

The stored record.

async get(session_id: str) → UserAuthSession | None

Load one current session record.

Parameters:

session_id – The session to load.

Returns:

The record, or None when the session is absent, expired, or revoked.

async list_for_account(account_id: str) → Sequence[UserAuthSession]

List safe session metadata for one account.

Parameters:

account_id – The account whose sessions to list.

Returns:

The account’s active session records, which may be empty.

async touch(session_id: str, *, now: datetime) → UserAuthSession | None

Apply the implementation’s bounded last-seen write policy.

Called on every authenticated request, so throttling the write is the implementation’s decision rather than the caller’s.

Parameters:

• session_id – The session that was just used.

• now – The observation timestamp.

Returns:

The current record, or None when the session is no longer valid.

async revoke_session_for_account(account_id: str, session_id: str, *, event: SecurityEvent) → bool

Revoke one session only when atomically owned by the account.

Check ownership inside this operation. A caller must not be able to revoke another account’s session by naming its identifier.

Parameters:

• account_id – The authenticated caller’s account.

• session_id – The session to revoke.

• event – The audit event to commit with the revocation. Rejecting it must fail the revocation.

Returns:

True when the caller owned an active session that was revoked.

async revoke_sessions_for_account(account_id: str, *, event: SecurityEvent) → int

Revoke every authenticated session for an account.

Parameters:

• account_id – The account whose sessions to revoke.

• event – The audit event to commit with the revocations. Rejecting it must fail them.

Returns:

The number of active sessions revoked.

async revoke_other_sessions(account_id: str, session_id: str, *, event: SecurityEvent) → int

Revoke all account sessions except the named current session.

Parameters:

• account_id – The account whose sessions to revoke.

• session_id – The one session to keep, normally the caller’s own.

• event – The audit event to commit with the revocations. Rejecting it must fail them.

Returns:

The number of other active sessions revoked.

async rebind(prior_session_id: str, command: CreateSessionCommand, *, event: SecurityEvent) → UserAuthSession | None

Revoke a prior record and create its replacement atomically.

Both halves commit together. A window in which neither or both sessions are valid is what session fixation exploits.

Parameters:

• prior_session_id – The session being replaced.

• command – The replacement session to create.

• event – The audit event to commit with the rebind. Rejecting it must fail the rebind.

Returns:

The replacement record, or None when the prior session was already gone.

__init__(*args, **kwargs)

class litestar_security.accounts.SessionSummary(session_id: str, current: bool, created_at: datetime, last_seen_at: datetime, expires_at: datetime, display_metadata: Mapping[str, str]=<factory>)

Bases: object

Safe authenticated-session inventory projection.

__init__(session_id: str, current: bool, created_at: datetime, last_seen_at: datetime, expires_at: datetime, display_metadata: Mapping[str, str]=<factory>) → None

class litestar_security.accounts.StepUpAuthorization(step_up_grant: str)

Bases: WireStruct

Carry the grant authorizing one sensitive factor operation.

class litestar_security.accounts.StepUpGrant(grant: str, purpose: str, expires_at: datetime)

Bases: WireStruct

Return one short-lived transport-bound grant.

class litestar_security.accounts.StepUpVerification(method: str, credential: str, method_id: str | None = None)

Bases: WireStruct

Present one configured factor for a purpose-bound grant.

class litestar_security.accounts.StoreRateLimiter(policies: Mapping[str, RateLimitPolicy]=<factory>, store_name: str = 'litestar_security.rate_limits', store: Store | None = None, clock: Callable[[], datetime]=<function utc_now>)

Bases: object

Fixed-window limiter over a native Litestar store.

The store is resolved by name from the application registry during startup, so an unconfigured name yields Litestar’s in-memory default. A process-wide lock serializes every bundled limiter instance’s read-modify-write operation, making counting exact within one process. Native stores expose no compare-and-increment, however, so a shared backend is not atomic across worker processes or machines. Multi-process deployments must supply a RateLimiter backed by an atomic primitive and verify it with litestar_security.testing.assert_rate_limiter_conformance().

Parameters:

• policies – Budget per operation. Operations absent from the mapping are not limited by this limiter.

• store_name – Registry name resolved during application startup.

• store – Pre-resolved store, bypassing registry resolution.

• clock – Source of the current time.

bind(store: Store) → None

Attach the store resolved from the application registry at startup.

Parameters:

store – The store to count in. A shared backend shares bucket values, but cannot make this read-modify-write implementation atomic across worker processes.

Raises:

ImproperlyConfiguredException – If the value is not a Litestar store.

async acquire(request: RateLimitAttempt) → RateLimitDecision

Consume one attempt from every configured bucket for the operation.

A process-wide lock makes the complete multi-bucket accounting operation exact across bundled limiter instances in this process. The underlying store has no compare-and-increment, so deployments spanning multiple processes or machines must provide an atomic RateLimiter and verify it with litestar_security.testing.assert_rate_limiter_conformance().

Parameters:

request – The operation, buckets, and cost to charge.

Returns:

The decision. An operation absent from the policy mapping is allowed. When several buckets are exhausted, the longest wait is reported.

Raises:

RuntimeError – If the store has not been resolved, or a stored counter cannot be read as an integer.

__init__(policies: Mapping[str, RateLimitPolicy]=<factory>, store_name: str = 'litestar_security.rate_limits', store: Store | None = None, clock: Callable[[], datetime]=<function utc_now>) → None

class litestar_security.accounts.TOTPEnrollment(label: str, step_up_grant: str)

Bases: WireStruct

Request a protected TOTP enrollment.

class litestar_security.accounts.TOTPProvisioning(enrollment_id: str, method_id: str, provisioning_uri: str, expires_at: datetime)

Bases: WireStruct

Reveal one TOTP provisioning URI.

class litestar_security.accounts.TOTPVerification(enrollment_id: str, code: str)

Bases: WireStruct

Activate one pending TOTP enrollment.

class litestar_security.accounts.TokenIssue(token_id: str, digest: bytes, purpose: TokenPurpose, expires_at: datetime, maximum_attempts: int, account_id: str, issued_security_epoch: int | None = None)

Bases: object

Hashed, purpose-bound token material accepted by an atomic store.

__init__(token_id: str, digest: bytes, purpose: TokenPurpose, expires_at: datetime, maximum_attempts: int, account_id: str, issued_security_epoch: int | None = None) → None

class litestar_security.accounts.TokenPair(access_token: str, refresh_token: str, expires_in: int, token_type: Literal['Bearer'] = 'Bearer')

Bases: WireStruct

Secret-safe token response recovered from a sealed rotation receipt.

class litestar_security.accounts.TokenPurpose(*values)

Bases: str, Enum

Closed namespaces for one-time local-account tokens.

class litestar_security.accounts.UnlimitedRateLimiter

Bases: object

Allow every attempt, for deployments that limit at the edge instead.

async acquire(request: RateLimitAttempt) → RateLimitDecision

Allow one attempt without consuming any budget.

Parameters:

request – Ignored; nothing is counted.

Returns:

An allowing decision.

__init__() → None

class litestar_security.accounts.UserAuthSession(session_id: str, binding_id: str, binding_digest: bytes, account_id: str, security_epoch: int, created_at: datetime, authenticated_at: datetime, last_seen_at: datetime, expires_at: datetime, display_metadata: Mapping[str, str]=<factory>)

Bases: object

Application-owned authenticated-session registry projection.

__init__(session_id: str, binding_id: str, binding_digest: bytes, account_id: str, security_epoch: int, created_at: datetime, authenticated_at: datetime, last_seen_at: datetime, expires_at: datetime, display_metadata: Mapping[str, str]=<factory>) → None

class litestar_security.accounts.UserAuthSessionResolver(*args, **kwargs)

Bases: Protocol[UserT]

Resolve the complete authoritative session state in one consistent read.

async resolve_user_auth_session(session_id: str, account_id: str, *, now: datetime) → ResolvedUserAuthSession[UserT] | None

Load a session and its account from one consistent snapshot.

Implementations must bind both returned values to the requested identifiers and must not return stale account epoch or activation state.

Parameters:

• session_id – Session identifier presented by the client.

• account_id – Account identifier embedded in the session payload.

• now – Authoritative UTC time for expiry-aware storage queries.

Returns:

The consistent session and account state, or None for invalid credentials.

Raises:

Exception – When authoritative storage is unavailable.

__init__(*args, **kwargs)

class litestar_security.accounts.VerificationOutcome(status: VerificationStatus, account_id: str | None = None, security_epoch: int | None = None)

Bases: object

Atomic verification-token consumption outcome.

__init__(status: VerificationStatus, account_id: str | None = None, security_epoch: int | None = None) → None

class litestar_security.accounts.VerificationStatus(*values)

Bases: str, Enum

Atomic purpose-token consumption outcomes.

class litestar_security.accounts.VerificationTokenStore(*args, **kwargs)

Bases: Protocol

Issue and atomically consume account-verification tokens.

async issue(issue: TokenIssue, notification: NotificationCommand, *, event: SecurityEvent) → None

Commit a verification issue, notification, and durable event.

Store the digest the issue carries, never the token itself: the token is the secret sent to the account holder.

Parameters:

• issue – The token digest, account binding, and expiry to store.

• notification – The delivery the application should send.

• event – The audit event to commit with the issue. Rejecting it must fail the issue.

async issue_absent() → None

Perform one durable round trip that commits nothing.

Called instead of issue() when the identifier resolves to no eligible account. The durable step MUST cost the same whether or not the identifier resolves: an implementation that answers quickly for unknown accounts makes a present account measurably slower to probe, defeating the shared-response guarantee. Commit, notify, and mutate nothing.

async consume_and_verify(token_id: str, digest: bytes, *, now: datetime, event: SecurityEvent) → VerificationOutcome

Consume a verification token and verify its account atomically.

Marking the token used and marking the account verified must commit together, so one token can never verify twice.

Parameters:

• token_id – The identifier carried by the presented token.

• digest – The digest to compare against the stored one.

• now – The timestamp to evaluate expiry against.

• event – The audit event to commit with the consumption. Rejecting it must fail the consumption.

Returns:

The outcome, carrying the account and its epoch only when consumed.

__init__(*args, **kwargs)

litestar_security.accounts.normalize_identifier(value: str) → str

Apply the default compatibility, whitespace, and case normalization.

Parameters:

value – The submitted identifier.

Returns:

The NFKC-normalized, stripped, case-folded identifier.

Raises:

ValueError – If the value is not text.

Testing helpers

Deterministic conformance helpers for security integration test suites.

class litestar_security.testing.BackendBarrier(reached: Event = <factory>, release: Event = <factory>)

Bases: object

Deterministically pause one named backend operation.

__init__(reached: Event = <factory>, release: Event = <factory>) → None

class litestar_security.testing.BackendEvent(sequence: int, operation: str, details: Mapping[str, str])

Bases: object

One secret-free deterministic reference-backend operation.

__init__(sequence: int, operation: str, details: Mapping[str, str]) → None

class litestar_security.testing.FakeClock(now: datetime)

Bases: object

Mutable UTC clock owned by one test.

__init__(now: datetime) → None

Initialize at one timezone-aware instant.

Parameters:

now – Initial time.

Raises:

ValueError – If now is naive.

advance(delta: timedelta) → datetime

Advance by a positive duration.

Parameters:

delta – Positive duration to add.

Returns:

The updated UTC datetime.

Raises:

ValueError – If delta is not positive.

class litestar_security.testing.FakeOAuthHTTPTransport(responses: list[Response])

Bases: AsyncBaseTransport

Deterministic queued HTTPX transport for provider conformance tests.

__init__(responses: list[Response]) → None

Initialize with responses consumed in order.

Parameters:

responses – Provider responses to return.

async handle_async_request(request: Request) → Response

Record one request and return the next response.

class litestar_security.testing.FakeOAuthProvider(*, name: str, tokens: ProviderTokenSet, identity: ProviderIdentity)

Bases: object

Deterministic async provider with public lifecycle call history.

__init__(*, name: str, tokens: ProviderTokenSet, identity: ProviderIdentity) → None

Initialize fixed provider results.

build_authorization_url(start: OAuthTransactionStart) → str

Return a deterministic URL.

async exchange_code(*, code: SecretStr, transaction: OAuthTransaction, now: datetime | None = None) → ProviderTokenSet

Return configured exchange tokens.

async resolve_identity(tokens: ProviderTokenSet, *, transaction: OAuthTransaction, now: datetime | None = None) → ProviderIdentity

Return the configured identity.

async refresh(refresh_token: SecretStr, *, current_scopes: frozenset[str] | None = None, now: datetime | None = None) → ProviderTokenSet

Return configured refresh tokens.

async revoke(token: SecretStr, *, token_type_hint: str | None) → None

Record deterministic revocation.

class litestar_security.testing.InMemoryAPIKeyStore(observe: Callable[[str, Mapping[str, str]], Awaitable[None]])

Bases: object

Atomic digest-only API-key store for tests and examples.

__init__(observe: Callable[[str, Mapping[str, str]], Awaitable[None]]) → None

Initialize isolated records and an aggregate diagnostic callback.

Parameters:

observe – Async operation callback owned by the aggregate backend.

property records: tuple[APIKeyState, ...]

Return a stable immutable record snapshot.

async get(key_id: str) → APIKeyState | None

Return one digest-only record.

async create(record: APIKeyState) → None

Atomically create one unique digest-only record.

async rotate(*, current_key_id: str, replacement: APIKeyState, overlap_until: datetime | None, now: datetime) → None

Atomically replace one current record with one successor.

async revoke(*, key_id: str, now: datetime) → None

Atomically revoke one existing key.

class litestar_security.testing.InMemoryLocalAccountStore(observe: Callable[[str, Mapping[str, str]], Awaitable[None]], *, clock: Callable[[], datetime], identifiers: Callable[[str], str], entropy: Callable[[int], bytes])

Bases: object

Atomic in-memory local-account, session, and refresh reference store.

__init__(observe: Callable[[str, Mapping[str, str]], Awaitable[None]], *, clock: Callable[[], datetime], identifiers: Callable[[str], str], entropy: Callable[[int], bytes]) → None

Initialize isolated state with aggregate deterministic sources.

async find_for_login(normalized_identifier: str) → LocalAccountState[object] | None

Find one account through its normalized identifier.

async get_by_id(account_id: str) → LocalAccountState[object] | None

Return one account by its stable identifier.

async current_epoch(account_id: str) → int | None

Return the authoritative account epoch.

async get_password_state(account_id: str) → PasswordCredentialState | None

Return one atomic password and account-state snapshot.

async compare_and_replace_password(account_id: str, expected_hash: str, password_hash: str, *, event: SecurityEvent) → bool

Replace one current password hash atomically.

async replace_password_and_bump_epoch(account_id: str, password_hash: str, *, expected_epoch: int, event: SecurityEvent) → PasswordChangeOutcome

Replace a password and advance its exact security epoch.

async register_login_method(account_id: str, method: LoginMethod, *, event: SecurityEvent) → None

Record a login method for an existing account.

async revoke_login_method(account_id: str, method_id: str, *, require_remaining: bool = True, event: SecurityEvent) → RevokeLoginMethodOutcome

Revoke one login method while preserving the requested invariant.

async register(command: RegistrationCommand, password_hash: str, *, invitation_digest: bytes | None, verification: PurposeTokenDelivery | None, now: datetime, event: SecurityEvent) → RegistrationOutcome[object]

Create one account and optional verification issue atomically.

async issue(issue: TokenIssue, notification: NotificationCommand, *, event: SecurityEvent) → None

Store one purpose-token issue without retaining its delivery secret.

async issue_absent() → None

Perform the deterministic no-op used for absent accounts.

async consume_and_verify(token_id: str, digest: bytes, *, now: datetime, event: SecurityEvent) → VerificationOutcome

Consume one verification token and mark its account verified.

async consume_and_reset(token_id: str, digest: bytes, new_password_hash: str, *, now: datetime, event: SecurityEvent) → PasswordResetOutcome

Consume one recovery token and reset its account password atomically.

async create(command: CreateSessionCommand, *, event: SecurityEvent) → UserAuthSession

Create one native session record.

async get(session_id: str) → UserAuthSession | None

Return one currently stored native session.

async list_for_account(account_id: str) → tuple[UserAuthSession, ...]

Return the account’s current native-session records.

async touch(session_id: str, *, now: datetime) → UserAuthSession | None

Advance one session’s last-seen time.

async revoke_session_for_account(account_id: str, session_id: str, *, event: SecurityEvent) → bool

Revoke one account-owned native session.

async revoke_sessions_for_account(account_id: str, *, event: SecurityEvent) → int

Revoke every native session owned by one account.

async revoke_other_sessions(account_id: str, session_id: str, *, event: SecurityEvent) → int

Revoke all native sessions except the named current one.

async rebind(prior_session_id: str, command: CreateSessionCommand, *, event: SecurityEvent) → UserAuthSession | None

Replace one existing session with a successor atomically.

async create_family(command: CreateRefreshFamilyCommand, *, event: SecurityEvent) → bool

Create a refresh family when its account epoch remains current.

async prepare_rotation(proof: RefreshTokenProof, idempotency_digest: bytes | None, *, now: datetime, event: SecurityEvent) → RefreshFamilyContext | RefreshReceiptReplay | RefreshPreflightOutcome

Resolve one exact refresh token for a later atomic rotation.

async rotate(command: RotateRefreshCommand, *, now: datetime, event: SecurityEvent) → RefreshRotationOutcome

Atomically rotate one prepared refresh token.

async revoke_family(family_id: str, *, event: SecurityEvent) → bool

Revoke every token in one refresh family.

async revoke_token(token_id: str, token_digest: bytes, *, event: SecurityEvent) → bool

Revoke the family owning one exact presented token.

async revoke_token_for_account(account_id: str, token_id: str, token_digest: bytes, *, event: SecurityEvent) → bool

Revoke one exact refresh token only for its owning account.

async revoke_for_account(account_id: str, *, event: SecurityEvent) → int

Revoke every refresh family owned by one account.

class litestar_security.testing.InMemoryMFALoginChallengeStore

Bases: object

Atomic in-memory digest-only MFA login challenge store.

__init__() → None

Initialize isolated mutable state.

async put(challenge: MFALoginChallenge) → None

Store one pending digest-only challenge.

Parameters:

challenge – Pending second-factor state.

async consume(challenge_digest: bytes, *, account_id: str, security_epoch: int, now: datetime) → MFALoginChallenge | None

Atomically burn and return one exact, current challenge.

Parameters:

• challenge_digest – Presented challenge digest.

• account_id – Expected local account.

• security_epoch – Expected current account epoch.

• now – Consumption time.

Returns:

The record only for the winning exact, unexpired match.

class litestar_security.testing.InMemoryMFAStore

Bases: object

Atomic in-memory implementation of the MFA store contract.

__init__() → None

Initialize isolated mutable state.

async create_totp_enrollment(enrollment: PendingTOTPEnrollment) → None

Store one enrollment.

Parameters:

enrollment – Protected pending enrollment.

async get_totp_enrollment(enrollment_id: str) → PendingTOTPEnrollment | None

Load one enrollment.

Parameters:

enrollment_id – Enrollment identifier.

Returns:

The pending enrollment, if present.

async activate_totp(account_id: str, enrollment_id: str, *, accepted_counter: int, login_method: LoginMethod, event: SecurityEvent, now: datetime) → TOTPMethod | None

Atomically consume and activate one enrollment.

Parameters:

• account_id – Expected owner.

• enrollment_id – Enrollment to consume.

• accepted_counter – Verified initial counter.

• login_method – Viable method committed with activation.

• event – Durable creation event.

• now – Commit timestamp.

Returns:

The active method only for the winning call.

async activate_totp_with_recovery_codes(account_id: str, enrollment_id: str, *, accepted_counter: int, codes: tuple[RecoveryCodeDigest, ...], login_method: LoginMethod, event: SecurityEvent, now: datetime) → TOTPMethod | None

Atomically activate one enrollment and replace recovery codes.

async get_totp_method(account_id: str, method_id: str) → TOTPMethod | None

Load an owner-checked active method.

Parameters:

• account_id – Expected owner.

• method_id – Method identifier.

Returns:

The active method only for its owner.

async advance_totp_counter(method_id: str, *, accepted_counter: int, now: datetime) → bool

Atomically advance a strictly monotonic TOTP counter.

Parameters:

• method_id – Method identifier.

• accepted_counter – Verified counter.

• now – Commit timestamp.

Returns:

Whether this call won the monotonic update.

async replace_recovery_codes(account_id: str, codes: tuple[RecoveryCodeDigest, ...], *, now: datetime) → None

Atomically replace an account’s complete digest set.

Parameters:

• account_id – Owning account.

• codes – Complete replacement set.

• now – Commit timestamp, accepted for protocol parity.

async consume_recovery_code(account_id: str, digest: bytes, *, now: datetime) → bool

Atomically compare and consume one recovery digest.

Parameters:

• account_id – Expected owner.

• digest – Presented HMAC digest.

• now – Commit timestamp, accepted for protocol parity.

Returns:

Whether this call consumed one matching digest.

class litestar_security.testing.InMemoryOIDCSessionLogoutStore(*, session_mappings: tuple[tuple[str, str, str | None, str | None], ...], frontchannel_bindings: Mapping[tuple[str, str, str], str], clock: Callable[[], datetime] | None = None)

Bases: object

Lock-protected OIDC mapped-session logout reference for deterministic tests.

__init__(*, session_mappings: tuple[tuple[str, str, str | None, str | None], ...], frontchannel_bindings: Mapping[tuple[str, str, str], str], clock: Callable[[], datetime] | None = None) → None

Initialize fixed secret-free mappings and browser bindings.

Parameters:

• session_mappings – (provider, issuer, subject, session_id) rows, one per local session.

• frontchannel_bindings – Browser binding by exact provider, issuer, and provider-session tuple.

• clock – Time source used to reject expired logout identities.

async consume_backchannel(identity: OIDCLogoutIdentity, *, now: datetime) → int | None

Consume one token id and revoke every active exact identity mapping.

async revoke_frontchannel(provider: str, issuer: str, session_id: str, *, binding: str, now: datetime) → int | None

Consume one exact browser-bound mapping and revoke its active local sessions.

class litestar_security.testing.InMemoryPasskeyStore

Bases: object

Atomic in-memory passkey credential store.

__init__() → None

Initialize isolated mutable state.

async add_credential(credential: PasskeyCredential, *, login_method: LoginMethod, event: SecurityEvent) → bool

Atomically register a credential, login method, and event.

Parameters:

• credential – Verified credential.

• login_method – Viable method committed with the credential.

• event – Durable creation event.

Returns:

Whether it was absent and added.

async get_credential(credential_id: bytes) → PasskeyCredential | None

Load one credential.

Parameters:

credential_id – Binary credential identifier.

Returns:

The credential, if present.

async record_assertion(credential_id: bytes, *, expected_version: int, sign_count: int, backup_eligible: bool, backup_state: bool, clone_risk: bool, now: datetime) → PasskeyAssertionStatus

Atomically record one verified assertion.

Parameters:

• credential_id – Credential to update.

• expected_version – Optimistic version.

• sign_count – Verified new signature counter.

• backup_eligible – Immutable BE flag.

• backup_state – Current BS flag.

• clone_risk – Whether the counter signaled possible cloning.

• now – Commit timestamp.

Returns:

Structured record, conflict, or clone-risk status.

async list_credentials(account_id: str) → tuple[PasskeyCredential, ...]

List an account’s credentials.

Parameters:

account_id – Owning account.

Returns:

Stable credential snapshot.

async rename_credential(account_id: str, credential_id: bytes, display_name: str) → PasskeyCredential | None

Atomically rename one owner-checked credential.

Parameters:

• account_id – Expected owner.

• credential_id – Credential identifier.

• display_name – Replacement metadata.

Returns:

Updated credential, or None.

class litestar_security.testing.InMemorySecurityBackend(*, clock: Callable[[], datetime] | None = None, identifiers: Callable[[str, int], str] | None = None, entropy: Callable[[int], bytes] | None = None, password_hash: str = '$litestar-security$deterministic-test-hash', protector: OAuthTransactionProtector | None = None)

Bases: object

Deterministic aggregate backend intended only for tests and examples.

__init__(*, clock: Callable[[], datetime] | None = None, identifiers: Callable[[str, int], str] | None = None, entropy: Callable[[int], bytes] | None = None, password_hash: str = '$litestar-security$deterministic-test-hash', protector: OAuthTransactionProtector | None = None) → None

Create isolated deterministic stores and value sources.

Parameters:

• clock – Injected timezone-aware clock.

• identifiers – Deterministic namespace and sequence formatter.

• entropy – Exact-length byte factory.

• password_hash – Precomputed test hash; plaintext passwords are never accepted.

• protector – Test protector for recoverable OAuth transaction secrets.

Raises:

ValueError – If a deterministic source is malformed.

property call_counts: Mapping[str, int]

Return an immutable copy of operation counts.

property events: tuple[BackendEvent, ...]

Return the ordered secret-free diagnostic snapshot.

clock() → datetime

Return one timezone-aware UTC instant.

next_identifier(namespace: str) → str

Return the next deterministic identifier in one aggregate sequence.

entropy(length: int) → bytes

Return exact-length deterministic bytes.

install_barrier(operation: str) → BackendBarrier

Install and return a deterministic operation barrier.

set_failpoint(operation: str, error: Exception) → None

Raise one injected error whenever the named operation is reached.

clear_controls() → None

Remove every barrier and failpoint without changing stored state.

class litestar_security.testing.InMemoryStepUpStore

Bases: object

Atomic in-memory digest-only step-up store.

__init__() → None

Initialize isolated mutable state.

async put(record: StepUpGrantState) → None

Store one grant record.

Parameters:

record – Digest-only grant.

async consume(grant_digest: bytes, *, principal_id: str, security_epoch: int, purpose: str, transport_digest: bytes, now: datetime) → StepUpGrantState | None

Atomically burn and return one exact current grant.

Parameters:

• grant_digest – Presented grant digest.

• principal_id – Expected principal.

• security_epoch – Expected current epoch.

• purpose – Expected protected action.

• transport_digest – Expected transport binding digest.

• now – Consumption time.

Returns:

The record only for the winning exact match.

class litestar_security.testing.InMemoryWebAuthnChallengeStore

Bases: object

Atomic in-memory digest-only WebAuthn challenge store.

__init__() → None

Initialize isolated mutable state.

async put(challenge: WebAuthnChallenge) → None

Store one digest-only challenge.

Parameters:

challenge – Bound challenge state.

async consume(challenge_digest: bytes, *, binding_digest: bytes, purpose: str, now: datetime) → WebAuthnChallenge | None

Atomically burn and return one exact challenge.

Parameters:

• challenge_digest – Presented challenge digest.

• binding_digest – Current transport binding digest.

• purpose – Expected ceremony.

• now – Consumption time.

Returns:

The record only for the winning exact match.

class litestar_security.testing.InMemoryWebSocketConnectTokenStore

Bases: object

Deterministic concurrency-safe connect token store for tests and examples.

property records: tuple[WebSocketConnectAuthorization, ...]

Return a stable snapshot of digest-only records.

async create(record: WebSocketConnectAuthorization) → None

Persist one record while rejecting duplicate public IDs.

async consume(*, connect_token_id: str, digest: bytes, now: datetime) → WebSocketConnectAuthorization | None

Atomically return and delete one matching unexpired record.

__init__() → None

class litestar_security.testing.InMemoryWebSocketRevocationSource

Bases: object

Deterministic per-binding WebSocket revocation source for tests.

__init__() → None

Initialize an isolated set of per-binding revocation events.

Parameters:

None.

Returns:

None.

Raises:

None. –

async wait(binding: WebSocketBinding) → None

Block until the exact binding has been revoked.

Parameters:

binding – The complete secret-free binding to supervise.

Returns:

None after revoke() releases this exact binding.

Raises:

None. –

revoke(binding: WebSocketBinding) → None

Release waiters for one exact binding.

Parameters:

binding – The complete secret-free binding to revoke.

Returns:

None.

Raises:

None. –

class litestar_security.testing.MemoryOAuthAccountStore(*, login_method_counts: Mapping[str, int] | None = None, provider: str = 'example', client_id: str = 'client', protector: OAuthTransactionProtector | None = None)

Bases: object

Atomic in-memory reference store for provider account behavior.

__init__(*, login_method_counts: Mapping[str, int] | None = None, provider: str = 'example', client_id: str = 'client', protector: OAuthTransactionProtector | None = None) → None

Create a store with authoritative total login-method counts.

Parameters:

• login_method_counts – Existing local and provider methods per account.

• provider – Provider namespace used in token associated data.

• client_id – OAuth client identifier used in token associated data.

• protector – Optional encryption port enabling retained tokens.

async login(identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, provision_unknown: bool, retain_tokens: bool = False, now: datetime) → OAuthLoginOutcome

Atomically resolve or provision one exact identity.

Parameters:

• identity – Exact provider identity.

• grant – Provider-observed grant.

• tokens – Exchanged provider tokens.

• provision_unknown – Whether an unknown identity may create an account.

• retain_tokens – Whether the aggregate should retain the token set.

• now – Aware mutation time.

Returns:

The linked account and whether this call provisioned it.

Raises:

OAuthAccountError – If the identity is unknown and provisioning is disabled.

async get_tokens(provider_account_id: str, *, now: datetime) → StoredProviderTokens | None

Return retained tokens when configured.

async replace_tokens(provider_account_id: str, *, expected_version: int, tokens: ProviderTokenSet, now: datetime) → bool

Compare and replace retained tokens.

async discard_tokens(provider_account_id: str, *, expected_version: int | None = None) → bool

Discard retained tokens at an optional observed version.

async stage_revocation_retry(failure: OAuthRevocationFailure, tokens: ProviderTokenSet, *, expected_version: int) → bool

Discard active tokens only when their observed version still matches.

async resolve_provider_account(account_id: str, provider: str) → LinkedProviderAccount | None

Resolve one exact account-owned provider link.

async link(account_id: str, identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, retain_tokens: bool = False, now: datetime) → LinkedProviderAccount

Commit an exact identity and its token policy under one aggregate lock.

async upgrade(account_id: str, provider_account_id: str, identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, retain_tokens: bool, now: datetime) → LinkedProviderAccount

Commit one grant and the exchanged token policy together.

async unlink(account_id: str, provider: str, provider_account_id: str, *, require_remaining: bool, now: datetime) → UnlinkOutcome

Remove one owned link and all retained credentials under the aggregate lock.

class litestar_security.testing.MemoryOAuthTransactionStore(*, protector: OAuthTransactionProtector, capacity: int = 1024, clock: Callable[[], datetime] | None = None)

Bases: object

Atomic in-memory reference store with protected recoverable secrets.

__init__(*, protector: OAuthTransactionProtector, capacity: int = 1024, clock: Callable[[], datetime] | None = None) → None

Initialize the reference store.

Parameters:

• protector – Application-owned transaction secret protection.

• capacity – Maximum number of live transactions retained.

• clock – Aware time source used for bounded expiry cleanup.

Raises:

ImproperlyConfiguredException – If the protector contract is absent.

async create(transaction: OAuthTransaction) → None

Protect and persist one new transaction.

Parameters:

transaction – The validated server-side transaction.

Raises:

ValueError – If an identical transaction lookup already exists.

async consume(*, state_digest: bytes, binding_digest: bytes, provider: str, now: datetime) → OAuthTransaction | None

Atomically return and remove one exact, unexpired match.

Parameters:

• state_digest – The state lookup digest.

• binding_digest – The dedicated browser-cookie digest.

• provider – The provider route receiving the callback.

• now – The authoritative callback time.

Returns:

The one consumed transaction, or None for every lookup miss.

class litestar_security.testing.OAuthRequestObservation(method: str, url: str, header_names: frozenset[str], form_fields: frozenset[str])

Bases: object

Secret-free projection of one provider HTTP request.

__init__(method: str, url: str, header_names: frozenset[str], form_fields: frozenset[str]) → None

class litestar_security.testing.StaticAuthorizationResolver(resolution: AuthorizationSnapshot | InvalidCredentials | VerificationUnavailable)

Bases: Generic[UserT]

Authorization resolver that returns one configured detached outcome.

async resolve(principal: Principal[UserT]) → AuthorizationSnapshot | InvalidCredentials | VerificationUnavailable

Return the configured authorization-resolution outcome.

Parameters:

principal – Ignored authenticated principal, which is never mutated.

Returns:

The configured immutable snapshot or sanitized authorization-resolution outcome.

Raises:

None. –

__init__(resolution: AuthorizationSnapshot | InvalidCredentials | VerificationUnavailable) → None

class litestar_security.testing.StaticAuthorizationSnapshotRefresher(snapshot: AuthorizationSnapshot)

Bases: Generic[UserT]

WebSocket snapshot refresher that returns one configured immutable snapshot.

async refresh(*, principal: Principal[UserT], previous: AuthorizationSnapshot, route_name: str) → AuthorizationSnapshot

Return the configured immutable authorization snapshot.

Parameters:

• principal – Ignored authenticated principal, which is never mutated.

• previous – Ignored prior snapshot, which is never mutated or returned.

• route_name – Ignored bound route name.

Returns:

The configured immutable authorization snapshot.

Raises:

None. –

__init__(snapshot: AuthorizationSnapshot) → None

class litestar_security.testing.StaticIdentityResolver(resolution: Principal[UserT] | InvalidCredentials | VerificationUnavailable)

Bases: Generic[ClaimsT, UserT]

Identity resolver that returns one configured outcome without retaining claims.

async resolve(claims: ClaimsT) → Principal[UserT] | InvalidCredentials | VerificationUnavailable

Return the configured identity-resolution outcome.

Parameters:

claims – Ignored verified claims, which are never retained.

Returns:

The configured principal or sanitized identity-resolution outcome.

Raises:

None. –

__init__(resolution: Principal[UserT] | InvalidCredentials | VerificationUnavailable) → None

class litestar_security.testing.StoreConformanceFactories(api_key_store: Callable[[], APIKeyStore] | None = None, local_account_store: Callable[[], _ConformanceLocalAccountStore] | None = None, mfa_login_challenge_store: Callable[[], MFALoginChallengeStore] | None = None, mfa_store: Callable[[], MFAStore] | None = None, oidc_session_logout_store: Callable[[], OIDCSessionLogoutStore] | None = None, oauth_account_store: Callable[[], OAuthAccountStore] | None = None, oauth_transaction_protector: Callable[[], OAuthTransactionProtector] | None = None, oauth_transaction_store: Callable[[], OAuthTransactionStore] | None = None, passkey_store: Callable[[], PasskeyStore] | None = None, refresh_family_store: Callable[[], _ConformanceRefreshFamilyStore] | None = None, secret_protector: Callable[[], SecretProtector] | None = None, session_registry: Callable[[], SessionRegistry] | None = None, step_up_store: Callable[[], StepUpStore] | None = None, webauthn_challenge_store: Callable[[], WebAuthnChallengeStore] | None = None, websocket_connect_token_store: Callable[[], WebSocketConnectTokenStore] | None = None)

Bases: object

Isolated zero-argument factories for explicitly enabled capabilities.

__init__(api_key_store: Callable[[], APIKeyStore] | None = None, local_account_store: Callable[[], _ConformanceLocalAccountStore] | None = None, mfa_login_challenge_store: Callable[[], MFALoginChallengeStore] | None = None, mfa_store: Callable[[], MFAStore] | None = None, oidc_session_logout_store: Callable[[], OIDCSessionLogoutStore] | None = None, oauth_account_store: Callable[[], OAuthAccountStore] | None = None, oauth_transaction_protector: Callable[[], OAuthTransactionProtector] | None = None, oauth_transaction_store: Callable[[], OAuthTransactionStore] | None = None, passkey_store: Callable[[], PasskeyStore] | None = None, refresh_family_store: Callable[[], _ConformanceRefreshFamilyStore] | None = None, secret_protector: Callable[[], SecretProtector] | None = None, session_registry: Callable[[], SessionRegistry] | None = None, step_up_store: Callable[[], StepUpStore] | None = None, webauthn_challenge_store: Callable[[], WebAuthnChallengeStore] | None = None, websocket_connect_token_store: Callable[[], WebSocketConnectTokenStore] | None = None) → None

async litestar_security.testing.assert_api_key_store_conformance(factory: Callable[[], APIKeyStore]) → None

Assert API-key isolation and atomic rotation behavior.

Parameters:

factory – Isolated zero-argument store factory.

Returns:

None when every invariant holds.

Raises:

AssertionError – If APIKeyStore isolation, lookup, or atomic rotation is violated.

async litestar_security.testing.assert_local_account_store_conformance(factory: Callable[[], _ConformanceLocalAccountStore]) → None

Assert local-account isolation and atomic security transitions.

Parameters:

factory – Isolated zero-argument local-account store factory.

Returns:

None when every local-account capability invariant holds.

Raises:

AssertionError – If a local-account capability violates an atomicity, replay, or final-method invariant.

async litestar_security.testing.assert_mfa_login_challenge_store_conformance(factory: Callable[[], MFALoginChallengeStore]) → None

Assert MFA-login challenges are bound, one-shot, and expiry-safe.

Parameters:

factory – Isolated zero-argument MFA login challenge-store factory.

Returns:

None when every MFA login challenge invariant holds.

Raises:

AssertionError – If a challenge can be replayed or survives a rejected binding or expiry.

async litestar_security.testing.assert_mfa_store_conformance(factory: Callable[[], MFAStore]) → None

Assert atomic TOTP counter and recovery-code consumption.

Parameters:

factory – Isolated zero-argument MFA-store factory.

Returns:

None when every MFA-store invariant holds.

Raises:

AssertionError – If a counter update is non-atomic or a recovery code can be reused.

async litestar_security.testing.assert_oauth_account_store_conformance(factory: Callable[[], OAuthAccountStore]) → None

Assert final-method protection and atomic OAuth identity unlinking.

Parameters:

factory – Isolated zero-argument OAuth account-store factory.

Returns:

None when every OAuth account-store invariant holds.

Raises:

AssertionError – If a final method can be removed or concurrent unlinking has two winners.

async litestar_security.testing.assert_oauth_transaction_protector_conformance(factory: Callable[[], OAuthTransactionProtector]) → None

Assert OAuth transaction protection preserves all required AEAD properties.

Parameters:

factory – Isolated zero-argument protector factory.

Returns:

None when the protector authenticates associated data and uses a fresh ciphertext.

Raises:

AssertionError – If the protector violates a round-trip, key-version, associated-data, or non-determinism invariant.

async litestar_security.testing.assert_oauth_transaction_store_conformance(factory: Callable[[], OAuthTransactionStore]) → None

Assert OAuth transactions preserve matching, expiry, and one-shot consumption.

Parameters:

factory – Isolated zero-argument OAuth transaction-store factory.

Returns:

None when every OAuth transaction invariant holds.

Raises:

AssertionError – If callback state can be replayed or a mismatched callback is accepted.

async litestar_security.testing.assert_oidc_session_logout_store_conformance(factory: Callable[[], OIDCSessionLogoutStore]) → None

Assert atomic OIDC logout against a fixed seeded mapped-session scenario.

The factory must return a fresh store seeded with two active mappings for ("conformance-provider", "https://issuer.example", "conformance-subject", "conformance-session"), one unrelated mapping, and the exact "conformance-browser-binding" front-channel binding for that tuple.

Parameters:

factory – Isolated zero-argument factory that returns the required seeded store.

Returns:

None when the store preserves exact OIDC ownership and one-shot semantics.

Raises:

AssertionError – If the seeded mappings can be incorrectly revoked or replayed.

async litestar_security.testing.assert_passkey_store_conformance(factory: Callable[[], PasskeyStore]) → None

Assert optimistic assertion recording and clone-risk results.

Parameters:

factory – Isolated zero-argument passkey-store factory.

Returns:

None when every passkey-store invariant holds.

Raises:

AssertionError – If only one optimistic writer is not recorded or clone risk is lost.

async litestar_security.testing.assert_rate_limiter_conformance(factory: Callable[[int], RateLimiter], *, limit: int = 5, concurrency: int = 20) → None

Assert exact atomic admission for concurrent acquires against one bucket.

Parameters:

• factory – Factory receiving the budget that the returned limiter must enforce.

• limit – Positive number of attempts that the limiter must admit.

• concurrency – Number of concurrent attempts; it must be at least limit.

Returns:

None when the limiter admits exactly limit concurrent attempts.

Raises:

• ValueError – If limit or concurrency is not a valid conformance scenario.

• AssertionError – If concurrent acquires over-admit or under-admit the configured budget.

async litestar_security.testing.assert_refresh_family_store_conformance(factory: Callable[[], _ConformanceRefreshFamilyStore]) → None

Assert strict refresh-family creation, rotation, replay, and ownership behavior.

Parameters:

factory – Isolated zero-argument combined local-account and refresh-family store factory frozen at the conformance clock.

Returns:

None when every refresh-family invariant holds.

Raises:

AssertionError – If a refresh-family transition is not exact, atomic, or account-owned.

async litestar_security.testing.assert_secret_protector_conformance(factory: Callable[[], SecretProtector]) → None

Assert MFA-secret protection preserves all required AEAD properties.

Parameters:

factory – Isolated zero-argument protector factory.

Returns:

None when the protector authenticates associated data and uses a fresh ciphertext.

Raises:

AssertionError – If the protector violates a round-trip, key-version, associated-data, or non-determinism invariant.

async litestar_security.testing.assert_security_backend_conformance(factories: StoreConformanceFactories) → None

Run only the conformance scenarios whose factories were supplied.

Parameters:

factories – Explicit feature factories to exercise.

Returns:

None when every enabled feature passes.

Raises:

AssertionError – If any enabled feature violates its public protocol.

async litestar_security.testing.assert_session_registry_conformance(factory: Callable[[], SessionRegistry], *, now: datetime = datetime.datetime(2026, 1, 1, 0, 0, tzinfo=datetime.timezone.utc)) → None

Assert session-registry state, atomic replacement, and ownership behavior.

Parameters:

• factory – Isolated zero-argument session-registry factory initialized so get() evaluates expiry against now.

• now – Time used for every created record and expiry assertion.

Returns:

None when every session-registry invariant holds.

Raises:

AssertionError – If session creation, expiry, replacement, or revocation violates its public contract.

async litestar_security.testing.assert_step_up_store_conformance(factory: Callable[[], StepUpStore]) → None

Assert one-time exact-binding step-up grant consumption.

Parameters:

factory – Isolated zero-argument step-up store factory.

Returns:

None when the store has one winner and rejects every distinct binding mismatch.

Raises:

AssertionError – If a grant can be replayed, double-consumed, or accepted with an altered binding.

async litestar_security.testing.assert_webauthn_challenge_store_conformance(factory: Callable[[], WebAuthnChallengeStore]) → None

Assert WebAuthn challenges burn once and enforce every binding.

Parameters:

factory – Isolated zero-argument WebAuthn challenge-store factory.

Returns:

None when every WebAuthn challenge invariant holds.

Raises:

AssertionError – If consume-once, binding, purpose, or expiry behavior is violated.

async litestar_security.testing.assert_websocket_connect_token_store_conformance(factory: Callable[[], WebSocketConnectTokenStore]) → None

Assert WebSocket connect tokens are exact, one-shot, and expiry-safe.

Parameters:

factory – Isolated zero-argument WebSocket connect-token store factory.

Returns:

None when every WebSocket connect-token invariant holds.

Raises:

AssertionError – If a wrong digest burns a token, or a token can be reused or outlive expiry.

Providers

Opaque API-key contracts and reveal-once key material.

class litestar_security.providers.api_key.APIKeyClaims(key_id: str, subject_id: str)

Bases: object

Verified digest-free identity carried from authentication to resolution.

__init__(key_id: str, subject_id: str) → None

class litestar_security.providers.api_key.APIKeyCodec(pepper: bytes, prefix: str = 'lsk', entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>, comparator: ~collections.abc.Callable[[bytes, bytes], bool] = <built-in function compare_digest>)

Bases: object

Issue and parse strict opaque API keys without persistence access.

Parameters:

• pepper – Secret mixed into every stored digest; at least 32 bytes.

• prefix – Version-carrying key prefix accepted and issued by this codec.

• entropy – Source of key-id and secret bytes.

• comparator – Digest equality used by matches(). A supplied comparator must compare in constant time over equal-length digests, as the default hmac.compare_digest() does; a variable-time comparator reintroduces a timing side channel on the stored digest. This contract is documented rather than runtime-enforceable, so construction only verifies callability.

issue(*, subject_id: str, restrictions: CredentialRestrictions | None = None, expires_at: datetime | None = None) → tuple[IssuedAPIKey, APIKeyState]

Create reveal-once key material paired with a digest-only record.

Parameters:

• subject_id – The application identity this key authenticates.

• restrictions – Optional authorization bounds carried by the key.

• expires_at – Optional exclusive expiry timestamp.

Returns:

The reveal-once value and the storage-safe record.

Raises:

• APIKeyGenerationError – If the entropy source raises or returns an invalid value.

• ValueError – If record metadata is invalid.

proof(value: object) → APIKeyProof | None

Parse a canonical key into storage-safe lookup and digest material.

Parameters:

value – The presented API-key value of any runtime type.

Returns:

A digest-only proof, or None when parsing fails.

matches(proof: APIKeyProof, record: APIKeyState) → bool

Compare one computed digest with a record through the configured comparator.

The comparator receives two equal-length digests and must compare them in constant time; see APIKeyCodec for the contract an override honors.

Parameters:

• proof – The storage-safe proof derived from a presented key.

• record – The looked-up digest-only record.

Returns:

True only when both public lookup and digest match.

__init__(pepper: bytes, prefix: str = 'lsk', entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>, comparator: ~collections.abc.Callable[[bytes, bytes], bool] = <built-in function compare_digest>) → None

class litestar_security.providers.api_key.APIKeyConfig(store: APIKeyStore | BlockingIntegration[_SyncAPIKeyStore], pepper: bytes, identity_resolver: object | None = None, usage_sink: APIKeyUsageSink | None = None, usage_write_interval: timedelta = datetime.timedelta(seconds=300), usage_buffer_capacity: int = 1024, prefix: str = 'lsk', header_name: str = 'X-API-Key')

Bases: object

API-key persistence, digest, usage, and namespace configuration.

as_dict() → dict[str, object]

Return a secret-redacted representation for explicit serialization.

Returns:

Public configuration values with the pepper redacted.

build(resolver: IdentityResolver[APIKeyClaims, UserT] | None = None, *, clock: Callable[[], datetime] = <function _utc_now>, entropy: Callable[[int], bytes] = <function token_bytes>, metrics: SecurityMetrics | None = None, participates_by_default: bool = True) → tuple[CredentialSlot[str], AuthenticationMechanism[str, APIKeyClaims, UserT], APIKeyService]

Build one physical slot, mechanism, and lifecycle service.

Parameters:

• resolver – Application identity resolver for verified API-key claims.

• clock – Time source for authentication and mutations.

• entropy – Random-byte source used only for issuance.

• metrics – Optional vendor-neutral usage metrics.

• participates_by_default – Include API keys in implicit protection.

Returns:

The slot, authentication mechanism, and lifecycle service.

__init__(store: APIKeyStore | BlockingIntegration[_SyncAPIKeyStore], pepper: bytes, identity_resolver: object | None = None, usage_sink: APIKeyUsageSink | None = None, usage_write_interval: timedelta = datetime.timedelta(seconds=300), usage_buffer_capacity: int = 1024, prefix: str = 'lsk', header_name: str = 'X-API-Key') → None

exception litestar_security.providers.api_key.APIKeyGenerationError

Bases: RuntimeError

Indicate that a configured entropy source failed closed.

__init__() → None

Initialize a stable error without exposing entropy-source detail.

class litestar_security.providers.api_key.APIKeyProof(key_id: str, digest: bytes)

Bases: object

Canonical public lookup and digest passed toward storage verification.

__init__(key_id: str, digest: bytes) → None

class litestar_security.providers.api_key.APIKeyService(config: APIKeyConfig, codec: APIKeyCodec, clock: Callable[[], datetime], usage: BufferedAPIKeyUsage | None = None)

Bases: object

Issue, rotate, revoke, and flush API keys through atomic application ports.

async issue(*, subject_id: str, restrictions: CredentialRestrictions | None = None, expires_at: datetime | None = None) → IssuedAPIKey

Issue one reveal-once key and persist only its digest record.

Parameters:

• subject_id – The application identity the key authenticates.

• restrictions – Optional credential authorization bounds.

• expires_at – Optional exclusive expiry.

Returns:

The reveal-once key.

async rotate(*, current_key_id: str, subject_id: str, restrictions: CredentialRestrictions | None = None, expires_at: datetime | None = None, overlap: timedelta = datetime.timedelta(0)) → IssuedAPIKey

Atomically replace one key with an optional bounded overlap.

Parameters:

• current_key_id – The public lookup being replaced.

• subject_id – The replacement identity binding.

• restrictions – Replacement authorization bounds.

• expires_at – Replacement exclusive expiry.

• overlap – How long the current key may remain valid.

Returns:

The reveal-once replacement.

Raises:

ValueError – If overlap is negative.

async revoke(key_id: str) → None

Revoke one key immediately through the atomic store operation.

Parameters:

key_id – The public lookup to revoke.

async flush_usage() → None

Flush eligible buffered usage observations.

async close() → None

Flush all pending usage observations during application shutdown.

__init__(config: APIKeyConfig, codec: APIKeyCodec, clock: Callable[[], datetime], usage: BufferedAPIKeyUsage | None = None) → None

class litestar_security.providers.api_key.APIKeyState(key_id: str, subject_id: str, digest: bytes, restrictions: CredentialRestrictions = <factory>, expires_at: datetime | None = None, revoked_at: datetime | None = None, overlap_until: datetime | None = None)

Bases: object

Storage-safe API-key state containing only a keyed digest.

The application store must persist this record without adding the raw key or its secret component.

is_valid_at(now: datetime) → bool

Return whether expiry and revocation permit use at one instant.

Parameters:

now – The timezone-aware instant to evaluate.

Returns:

True while the record is active or inside its explicit overlap.

Raises:

ValueError – If now is not timezone-aware.

as_dict() → dict[str, object]

Return a secret-redacted representation for explicit serialization.

Returns:

A dictionary containing public metadata and a redacted digest.

__init__(key_id: str, subject_id: str, digest: bytes, restrictions: CredentialRestrictions = <factory>, expires_at: datetime | None = None, revoked_at: datetime | None = None, overlap_until: datetime | None = None) → None

class litestar_security.providers.api_key.APIKeyStore(*args, **kwargs)

Bases: Protocol

Application-owned atomic persistence port for digest-only API keys.

Implementations must reject duplicate IDs. rotate() must create the replacement and transition the current record in one atomic operation, bounding overlap by the current record’s original expiry and rejecting an already-revoked current record so concurrent rotations have one winner. No method may accept or persist a raw key or secret component.

async get(key_id: str) → APIKeyState | None

Return one record by its indexed public lookup.

Parameters:

key_id – The canonical 16-character public lookup.

Returns:

The digest-only record, or None when it does not exist.

async create(record: APIKeyState) → None

Persist one new record and reject a duplicate ID atomically.

Parameters:

record – The digest-only record to create.

Raises:

Exception – When the ID exists or persistence cannot commit.

async rotate(*, current_key_id: str, replacement: APIKeyState, overlap_until: datetime | None, now: datetime) → None

Atomically create a successor and revoke the current record.

Implementations must reject a missing or already-revoked current record and set a live current record’s revocation to now. When overlap is requested, they must cap it at the current record’s original expiry; None means the current key stops immediately.

Parameters:

• current_key_id – The public lookup being replaced.

• replacement – The digest-only successor record.

• overlap_until – The requested inclusive end of old-key overlap.

• now – The transition timestamp.

Raises:

Exception – When either transition cannot commit as one unit.

async revoke(*, key_id: str, now: datetime) → None

Atomically revoke one key with no remaining overlap.

Parameters:

• key_id – The public lookup to revoke.

• now – The revocation timestamp.

Raises:

Exception – When revocation cannot commit.

__init__(*args, **kwargs)

class litestar_security.providers.api_key.APIKeyUsageSink(*args, **kwargs)

Bases: Protocol

Application-owned sink for coalesced, secret-free API-key usage.

async record(*, key_id: str, used_at: datetime) → None

Persist one coalesced usage observation.

Parameters:

• key_id – The public lookup only, never raw key material.

• used_at – The timezone-aware observation time.

Returns:

None after accepting the best-effort observation.

Raises:

Exception – When the sink cannot persist the observation.

Notes

BufferedAPIKeyUsage.flush() catches every sink exception, increments security.api_key.usage_failure, drops the pending observation, and never changes authentication or API-key validity.

__init__(*args, **kwargs)

class litestar_security.providers.api_key.BufferedAPIKeyUsage(sink: object, interval: timedelta, capacity: int = 1024, metrics: SecurityMetrics = <factory>)

Bases: object

Bound and coalesce best-effort usage observations away from requests.

observe(key_id: str, used_at: datetime) → None

Retain one latest secret-free observation without performing I/O.

Parameters:

• key_id – The public key lookup.

• used_at – The timezone-aware usage timestamp.

async flush(*, force: bool = False) → None

Write eligible coalesced observations without raising sink failures.

Parameters:

force – Ignore the interval during shutdown.

async close() → None

Flush every pending observation during shutdown.

__init__(sink: object, interval: timedelta, capacity: int = 1024, metrics: SecurityMetrics = <factory>) → None

class litestar_security.providers.api_key.IssuedAPIKey(key_id: str, value: str)

Bases: object

Reveal-once raw API key returned only to the issuing caller.

as_dict() → dict[str, str]

Return a secret-redacted representation for explicit serialization.

Returns:

A dictionary containing the lookup ID and a redaction marker.

__init__(key_id: str, value: str) → None

Google Identity-Aware Proxy assertion verification.

class litestar_security.providers.iap.GoogleIAPClaims(subject: str, email: str | None = None, authorized_party: str | None = None, hosted_domain: str | None = None, access_levels: tuple[str, ...] = (), device_id: str | None = None, external_identity: GoogleIAPExternalIdentity | None = None)

Bases: object

Verified IAP identity fields offered to the application resolver.

__init__(subject: str, email: str | None = None, authorized_party: str | None = None, hosted_domain: str | None = None, access_levels: tuple[str, ...] = (), device_id: str | None = None, external_identity: GoogleIAPExternalIdentity | None = None) → None

class litestar_security.providers.iap.GoogleIAPConfig(audience: str | frozenset[str], identity_resolver: ~litestar_security.authentication.IdentityResolver[~litestar_security.providers.iap._iap.GoogleIAPClaims, ~litestar_security.providers.iap._iap.UserT], jwks: ~litestar_security.providers.jwks._provider.JWKSProvider, issuer: str = 'https://cloud.google.com/iap', header_name: str = 'X-Goog-IAP-JWT-Assertion', clock_skew: ~datetime.timedelta = datetime.timedelta(seconds=30), worker_limits: ~litestar_security.workers.WorkerLimits = <factory>)

Bases: Generic[UserT]

Pinned trust configuration for Google IAP signed assertions.

build(*, clock: Callable[[], ~datetime.datetime]=<function GoogleIAPConfig.<lambda>>) → tuple[CredentialSlot[str], AuthenticationMechanism[str, GoogleIAPClaims, UserT]]

Build the sole IAP assertion slot and authoritative mechanism.

Parameters:

clock – Time source used for key freshness and claim validation.

Returns:

The signed-header slot and paired IAP mechanism.

__init__(audience: str | frozenset[str], identity_resolver: ~litestar_security.authentication.IdentityResolver[~litestar_security.providers.iap._iap.GoogleIAPClaims, ~litestar_security.providers.iap._iap.UserT], jwks: ~litestar_security.providers.jwks._provider.JWKSProvider, issuer: str = 'https://cloud.google.com/iap', header_name: str = 'X-Goog-IAP-JWT-Assertion', clock_skew: ~datetime.timedelta = datetime.timedelta(seconds=30), worker_limits: ~litestar_security.workers.WorkerLimits = <factory>) → None

class litestar_security.providers.iap.GoogleIAPExternalIdentity(subject: str, email: str | None = None, email_verified: bool | None = None, sign_in_provider: str | None = None, tenant: str | None = None, sign_in_attributes: Mapping[str, str]=<factory>)

Bases: object

Validated external Identity Platform identity nested in an IAP assertion.

__init__(subject: str, email: str | None = None, email_verified: bool | None = None, sign_in_provider: str | None = None, tenant: str | None = None, sign_in_attributes: Mapping[str, str]=<factory>) → None

JSON Web Token signing, verification, and bearer slot composition.

class litestar_security.providers.jwt.BearerSlotSelector(issuers: frozenset[str], audiences: frozenset[str] = frozenset({}), token_types: frozenset[str] = frozenset({'application/at+jwt', 'at+jwt'}))

Bases: object

Route unverified bearer metadata only to a configured trust domain.

__init__(issuers: frozenset[str], audiences: frozenset[str] = frozenset({}), token_types: frozenset[str] = frozenset({'application/at+jwt', 'at+jwt'})) → None

class litestar_security.providers.jwt.BearerTokenSlot(name: str, selector: BearerSlotSelector, verifier: JWTVerifier[JWTClaims])

Bases: object

Bind one logical bearer routing selector to one verifier.

__init__(name: str, selector: BearerSlotSelector, verifier: JWTVerifier[JWTClaims]) → None

class litestar_security.providers.jwt.CompositeBearerConfig(mechanism_name: str, slots: tuple[BearerTokenSlot, ...], maximum_token_bytes: int = 16384)

Bases: object

Own one bearer namespace and dispatch it to exactly one JWT verifier.

build(resolver: ~litestar_security.authentication.IdentityResolver[~litestar_security.providers.jwt._claims.JWTClaims, ~litestar_security.providers.jwt._bearer.UserT], *, clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function _utc_now>, participates_by_default: bool = True, scheme_name: str | None = None) → tuple[CredentialSlot[str], AuthenticationMechanism[str, JWTClaims, UserT]]

Build one physical slot and one native bearer mechanism.

Parameters:

• resolver – The identity resolver for verified claims.

• clock – The clock used for expiry decisions.

• participates_by_default – Include this mechanism in implicit required().

• scheme_name – The OpenAPI security scheme name to publish under.

Returns:

The credential slot paired with its mechanism.

__init__(mechanism_name: str, slots: tuple[BearerTokenSlot, ...], maximum_token_bytes: int = 16384) → None

class litestar_security.providers.jwt.JWTClaims(issuer: str, subject: str | None, audiences: frozenset[str], expires_at: datetime, issued_at: datetime, not_before: datetime | None, token_id: str | None, client_id: str | None, scopes: frozenset[str], raw: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None], bearer_slot: str | None = None)

Bases: object

Verified, normalized JWT claims without the compact credential.

__init__(issuer: str, subject: str | None, audiences: frozenset[str], expires_at: datetime, issued_at: datetime, not_before: datetime | None, token_id: str | None, client_id: str | None, scopes: frozenset[str], raw: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None], bearer_slot: str | None = None) → None

class litestar_security.providers.jwt.JWTValidationConfig(issuer: str, audiences: frozenset[str], algorithms: frozenset[str], required_claims: frozenset[str] = frozenset({'aud', 'exp', 'iat', 'iss', 'sub'}), access_token_profile: bool = True, subject_required: bool = True, clock_skew: timedelta = datetime.timedelta(seconds=30), maximum_lifetime: timedelta | None = datetime.timedelta(seconds=3600), token_types: frozenset[str] = frozenset({'application/at+jwt', 'at+jwt'}))

Bases: object

Pin one issuer’s accepted JWT verification profile.

__init__(issuer: str, audiences: frozenset[str], algorithms: frozenset[str], required_claims: frozenset[str] = frozenset({'aud', 'exp', 'iat', 'iss', 'sub'}), access_token_profile: bool = True, subject_required: bool = True, clock_skew: timedelta = datetime.timedelta(seconds=30), maximum_lifetime: timedelta | None = datetime.timedelta(seconds=3600), token_types: frozenset[str] = frozenset({'application/at+jwt', 'at+jwt'})) → None

class litestar_security.providers.jwt.JWTVerifier(*args, **kwargs)

Bases: Protocol, Generic[ClaimsT]

Verify one compact JWT against a configured trust domain.

property config: JWTValidationConfig

Return the verifier’s pinned trust profile.

async verify(token: str, *, now: datetime) → NoCredentials | Authenticated[ClaimsT] | InvalidCredentials | VerificationUnavailable

Return a structured authentication outcome.

Parameters:

• token – The compact JWT to verify.

• now – The verification timestamp, used for expiry and not-before checks.

Returns:

The verified claims, or a sanitized outcome. A rejected signature and a rejected claim are not distinguished.

__init__(*args, **kwargs)

class litestar_security.providers.jwt.LocalJWKSConfig(key_set: VerificationKeySet, route_prefix: str = '/auth', cache_max_age: int = 300)

Bases: object

Immutable public representation of one local verification-key generation.

__init__(key_set: VerificationKeySet, route_prefix: str = '/auth', cache_max_age: int = 300) → None

class litestar_security.providers.jwt.LocalKeyRing(issuer: str, active_signing_key: SigningKey, verification_keys: tuple[~litestar_security.providers.jwt._keys.VerificationKey, ...]=(), worker_limits: WorkerLimits = <factory>, metrics: SecurityMetrics = <factory>)

Bases: object

Immutable active and retained local key configuration.

property all_verification_keys: tuple[VerificationKey, ...]

Return the active key followed by retained verification-only keys.

property verification_key_set: VerificationKeySet

Return the public verification-only view used by local or custom signers.

build_signer() → TokenSigner

Build the local signer without generating or discovering key material.

Returns:

A signer bound to the configured active signing key.

build_verifier(config: JWTValidationConfig, *, mechanism_name: str = 'jwt', slot_name: str = 'authorization.bearer') → JWTVerifier[JWTClaims]

Build one exact-kid verifier across the active and retained keys.

Retained keys stay accepted so tokens signed before a rotation keep verifying until they expire.

Parameters:

• config – The pinned trust profile. Its issuer must match this key ring.

• mechanism_name – The mechanism the verifier belongs to.

• slot_name – The credential slot the verifier reads.

Returns:

A verifier that selects keys by exact key identifier.

async mint_capability(*, purpose: str, subject: str, audience: str, lifetime: timedelta, claims: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None] | None = None) → str

Mint one bounded, single-purpose capability JWT.

Parameters:

• purpose – The application-defined capability purpose.

• subject – The principal this capability represents.

• audience – The exact service or resource that may accept it.

• lifetime – The positive capability lifetime, no longer than 24 hours.

• claims – Optional JSON application claims, excluding reserved names.

Returns:

A compact capability JWT with a hard-pinned capability+jwt type.

Raises:

• ValueError – If a capability input or lifetime is invalid.

• RuntimeError – If capability signing is unavailable.

async verify_capability(raw: str, *, purpose: str, audience: str, now: datetime) → VerifiedCapability | InvalidCredentials | VerificationUnavailable

Verify one capability JWT against this key ring.

Parameters:

• raw – The untrusted compact JWT.

• purpose – The exact application-defined capability purpose to accept.

• audience – The exact service or resource that may accept the capability.

• now – The timezone-aware verification timestamp.

Returns:

The verified capability, a sanitized invalid-credential outcome, or an unavailable-verification outcome for unexpected worker failures.

Raises:

• Never for untrusted credential input; failures are returned as –

• sanitized invalid-credential or unavailable-verification outcomes. –

__init__(issuer: str, active_signing_key: SigningKey, verification_keys: tuple[~litestar_security.providers.jwt._keys.VerificationKey, ...]=(), worker_limits: WorkerLimits = <factory>, metrics: SecurityMetrics = <factory>) → None

class litestar_security.providers.jwt.SigningKey(key_id: str, algorithm: Literal['EdDSA', 'ES256', 'RS256', 'HS256'], private_key: bytes, public_jwk: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None] | None = None)

Bases: object

One explicit local signing key and its public verification metadata.

as_verification_key() → VerificationKey

Return the active key’s verification-only representation.

Returns:

The public half, carrying no private key material.

__init__(key_id: str, algorithm: Literal['EdDSA', 'ES256', 'RS256', 'HS256'], private_key: bytes, public_jwk: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None] | None = None) → None

class litestar_security.providers.jwt.SyncJWTVerifier(*args, **kwargs)

Bases: Protocol, Generic[ClaimsT]

Blocking custom verifier normalized once into the crypto worker.

property config: JWTValidationConfig

Return the verifier’s pinned trust profile.

verify(token: str, *, now: datetime) → NoCredentials | Authenticated[ClaimsT] | InvalidCredentials | VerificationUnavailable

Return a structured authentication outcome.

Parameters:

• token – The compact JWT to verify.

• now – The verification timestamp, used for expiry and not-before checks.

Returns:

The verified claims, or a sanitized outcome. A rejected signature and a rejected claim are not distinguished.

__init__(*args, **kwargs)

class litestar_security.providers.jwt.SyncTokenSigner(*args, **kwargs)

Bases: Protocol

Blocking custom access-JWT signer normalized once into the crypto worker.

Implementations emit access JWTs whose protected header has a non-empty kid, a supported non-none alg, and typ="at+jwt". Untrusted caller claims must not choose those headers.

sign(claims: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None], *, now: datetime) → str

Return one compact signed access token.

Parameters:

• claims – The claim set to sign.

• now – The signing timestamp.

Returns:

A compact access JWT with the required protected-header profile.

Raises:

Exception – When signing cannot produce that access JWT.

__init__(*args, **kwargs)

class litestar_security.providers.jwt.TokenSigner(*args, **kwargs)

Bases: Protocol

Sign caller-built local claims without owning application persistence.

Implementations emit access JWTs whose protected header has a non-empty kid, a supported non-none alg, and typ="at+jwt". Untrusted caller claims must not choose those headers.

async sign(claims: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None], *, now: datetime) → str

Return one compact signed access token.

Parameters:

• claims – The claim set to sign.

• now – The signing timestamp.

Returns:

A compact access JWT with the required protected-header profile.

Raises:

Exception – When signing cannot produce that access JWT.

__init__(*args, **kwargs)

class litestar_security.providers.jwt.VerificationKey(key_id: str, algorithm: Literal['EdDSA', 'ES256', 'RS256', 'HS256'], key: bytes, public_jwk: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None] | None = None)

Bases: object

One explicit verification-only key retained for local rotation.

__init__(key_id: str, algorithm: Literal['EdDSA', 'ES256', 'RS256', 'HS256'], key: bytes, public_jwk: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None] | None = None) → None

class litestar_security.providers.jwt.VerificationKeySet(issuer: str, keys: tuple[VerificationKey, ...])

Bases: object

One issuer’s immutable verification-only keys for local or custom signers.

build_verifier(config: JWTValidationConfig, *, mechanism_name: str = 'jwt', slot_name: str = 'authorization.bearer', worker_limits: WorkerLimits | None = None, metrics: SecurityMetrics | None = None) → JWTVerifier[JWTClaims]

Build one exact-kid verifier across this trusted key set.

Parameters:

• config – The pinned trust profile. Its issuer must match this key set.

• mechanism_name – The mechanism the verifier belongs to.

• slot_name – The credential slot the verifier reads.

• worker_limits – The shared crypto-worker budget verification runs inside.

• metrics – The sink offered verification measurements.

Returns:

A verifier that selects keys by exact key identifier, never by a claim read from the unverified token.

__init__(issuer: str, keys: tuple[VerificationKey, ...]) → None

class litestar_security.providers.jwt.VerifiedCapability(purpose: str, subject: str, audience: str, issued_at: datetime, expires_at: datetime, token_id: str, claims: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None])

Bases: object

Verified application capability claims without the compact credential.

Parameters:

• purpose – The exact application-defined capability purpose.

• subject – The principal the capability represents.

• audience – The exact service or resource allowed to accept the capability.

• issued_at – The timezone-aware timestamp at which the capability was issued.

• expires_at – The timezone-aware timestamp at which the capability expires.

• token_id – The unique capability identifier used for optional application-level consumption.

• claims – The immutable application claims with reserved credential claims removed.

Returns:

A frozen capability projection that contains no compact credential.

Raises:

Never directly raises; invalid credentials are rejected before this value is created. –

__init__(purpose: str, subject: str, audience: str, issued_at: datetime, expires_at: datetime, token_id: str, claims: Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None]) → None

litestar_security.providers.jwt.build_access_token_claims(*, issuer: str, audience: str, subject: str, client_id: str, security_epoch: int, now: datetime, lifetime: timedelta, scopes: Set[str] = frozenset({}), methods: Set[str] = frozenset({}), traits: Set[str] = frozenset({}), amr: Sequence[str] = (), authenticated_at: datetime | None = None, jti: str | None = None, not_before: datetime | None = None) → Mapping[str, bool | int | float | str | list[bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | dict[str, bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None] | None]

Build minimal deterministic RFC 9068-style local access-token claims.

The claim set is server-owned and minimal. Application data stays out of it, so a leaked token reveals nothing beyond the account binding.

Parameters:

• issuer – The iss claim.

• audience – The aud claim.

• subject – The sub claim, normally the account identifier.

• client_id – The client_id claim.

• security_epoch – The epoch the token is bound to, so a later change invalidates it.

• now – The issue timestamp.

• lifetime – How long the token stays valid.

• scopes – The scopes to record.

• methods – Normalized authentication methods to preserve.

• traits – Normalized assurance traits to preserve.

• amr – Ordered authentication-method references to preserve.

• authenticated_at – Original authentication time for freshness checks.

• jti – The token identifier, or None to omit it.

• not_before – When the token becomes valid, or None to omit it.

Returns:

The claim set, ready to sign.

litestar_security.providers.jwt.build_local_jwks_handler(config: LocalJWKSConfig) → HTTPRouteHandler

Build one native public Litestar handler for immutable local JWKS bytes.

Parameters:

config – The publication settings and precomputed canonical response.

Returns:

A public handler serving the key set with a stable ETag and cache headers.

litestar_security.providers.jwt.extend_composite_bearer(mechanism: AuthenticationMechanism[str, JWTClaims, UserT], slot: BearerTokenSlot, resolver: IdentityResolver[JWTClaims, UserT]) → AuthenticationMechanism[str, JWTClaims, UserT]

Extend one library-built composite while preserving one physical bearer owner.

Only one mechanism may own the physical bearer slot, so an additional issuer extends the existing composite rather than registering a second reader.

Parameters:

• mechanism – The mechanism built by CompositeBearerConfig.

• slot – The additional trust slot to accept.

• resolver – The identity resolver for that slot’s claims.

Returns:

The extended mechanism, still owning exactly one bearer slot.

litestar_security.providers.jwt.normalize_signer(signer: TokenSigner | SyncTokenSigner, *, worker_limits: WorkerLimits | None = None, metrics: SecurityMetrics | None = None) → TokenSigner

Normalize one custom signer once without blocking the event loop.

Parameters:

• signer – The application’s signer, blocking or async.

• worker_limits – The shared crypto-worker budget a blocking signer runs inside.

• metrics – The sink offered signing measurements.

Returns:

An async signer.

litestar_security.providers.jwt.normalize_verifier(verifier: JWTVerifier[ClaimsT] | SyncJWTVerifier[ClaimsT], *, worker_limits: WorkerLimits | None = None, metrics: SecurityMetrics | None = None) → JWTVerifier[ClaimsT]

Normalize one custom verifier once without blocking the event loop.

Parameters:

• verifier – The application’s verifier, blocking or async.

• worker_limits – The shared crypto-worker budget a blocking verifier runs inside.

• metrics – The sink offered verification measurements.

Returns:

An async verifier.

OpenID Connect discovery with pinned, validated issuer metadata.

class litestar_security.providers.oidc.DiscoveryPolicy(allowed_issuers: frozenset[str], allowed_jwks_origins: frozenset[str] = frozenset({}), allowed_oauth_origins: frozenset[str] = frozenset({}), require_https: bool = True, allow_private_hosts: bool = False, allowed_ports: frozenset[int] = frozenset({443}), connect_timeout: float = 2.0, read_timeout: float = 3.0, maximum_document_bytes: int = 65536)

Bases: object

Exact operator-controlled OIDC discovery network boundary.

__init__(allowed_issuers: frozenset[str], allowed_jwks_origins: frozenset[str] = frozenset({}), allowed_oauth_origins: frozenset[str] = frozenset({}), require_https: bool = True, allow_private_hosts: bool = False, allowed_ports: frozenset[int] = frozenset({443}), connect_timeout: float = 2.0, read_timeout: float = 3.0, maximum_document_bytes: int = 65536) → None

class litestar_security.providers.oidc.KeycloakClaims(realm_roles: frozenset[str] = frozenset({}), client_roles: Mapping[str, frozenset[str]]=<factory>, scopes: frozenset[str] = frozenset({}), permissions: frozenset[ResourcePermission] = frozenset({}))

Bases: object

Deterministic authorization fields mapped from a verified Keycloak JWT.

__init__(realm_roles: frozenset[str] = frozenset({}), client_roles: Mapping[str, frozenset[str]]=<factory>, scopes: frozenset[str] = frozenset({}), permissions: frozenset[ResourcePermission] = frozenset({})) → None

class litestar_security.providers.oidc.OIDCDiscoveryClient(policy: DiscoveryPolicy, algorithms: frozenset[str], *, transport: AsyncBaseTransport | None = None, resolver: Callable[[str, int], Awaitable[Sequence[str]]] | None = None)

Bases: object

Async-native bounded discovery client for an exact issuer allowlist.

__init__(policy: DiscoveryPolicy, algorithms: frozenset[str], *, transport: AsyncBaseTransport | None = None, resolver: Callable[[str, int], Awaitable[Sequence[str]]] | None = None) → None

Create one owned client with redirects and proxy environment disabled.

async discover(issuer: str, *, discovery_url: str | None = None) → OIDCMetadata

Fetch and validate metadata for one configured issuer.

Parameters:

• issuer – The issuer to discover, matched against configured trust anchors.

• discovery_url – Where the metadata document lives, for a provider that does not publish it at {issuer}/.well-known/openid-configuration. It must share the issuer’s exact origin, so an override changes the path this client requests and never the host it reaches.

Returns:

The validated metadata.

Raises:

OIDCDiscoveryError – If the client is closed, the issuer is not a configured anchor, or the document fails validation.

async aclose() → None

Close the owned HTTP client idempotently.

exception litestar_security.providers.oidc.OIDCDiscoveryError

Bases: RuntimeError

Sanitized operational or remote-metadata discovery failure.

class litestar_security.providers.oidc.OIDCJWTLogoutTokenConsumer(verifiers: Mapping[str, JWTVerifier[JWTClaims]])

Bases: object

Verify logout-token JWTs for atomic application-side consumption.

async consume(provider: str, logout_token: str, *, now: datetime) → OIDCLogoutIdentity

Verify signature and logout claims, then atomically consume jti.

__init__(verifiers: Mapping[str, JWTVerifier[JWTClaims]]) → None

class litestar_security.providers.oidc.OIDCMetadata(issuer: str, jwks_uri: str, authorization_endpoint: str | None, token_endpoint: str | None, end_session_endpoint: str | None, algorithms: frozenset[str], revocation_endpoint: str | None = None)

Bases: object

Validated provider metadata needed by authentication integrations.

__init__(issuer: str, jwks_uri: str, authorization_endpoint: str | None, token_endpoint: str | None, end_session_endpoint: str | None, algorithms: frozenset[str], revocation_endpoint: str | None = None) → None

class litestar_security.providers.oidc.OIDCProvider(oauth: OAuthProviderClient, metadata: OIDCMetadata, verifier: JWTVerifier[JWTClaims], retain_tokens_by_default: bool = True)

Bases: object

OIDC code-flow provider with an independently configured ID-token verifier.

property name: str

Return the configured provider name.

property issuer: str

Return the exact discovered issuer.

property end_session_endpoint: str | None

Return the validated provider logout endpoint when advertised.

build_authorization_url(start: OAuthTransactionStart) → str

Build the OAuth authorization URL.

Parameters:

start – The bound transaction start values.

Returns:

The fixed provider authorization URL.

build_reauthentication_url(start: OAuthTransactionStart, *, max_age: int) → str

Build a forced-authentication request using OIDC max_age.

Parameters:

• start – The bound authorization transaction.

• max_age – Maximum accepted signed authentication age, including zero.

Returns:

The authorization URL with the reserved max_age parameter.

Raises:

OAuthProviderError – If the age is invalid or provider state is inconsistent.

async exchange_code(*, code: SecretStr, transaction: OAuthTransaction, now: datetime | None = None) → ProviderTokenSet

Exchange a code through the composed OAuth client.

Parameters:

• code – The provider authorization code.

• transaction – The consumed transaction.

• now – The authoritative response time.

Returns:

The validated provider token set.

async resolve_identity(tokens: ProviderTokenSet, *, transaction: OAuthTransaction, now: datetime | None = None) → ProviderIdentity

Verify the ID token and bind its claims to the consumed transaction.

Parameters:

• tokens – The provider token response containing an ID token.

• transaction – The consumed transaction containing issuer and nonce bindings.

• now – The authoritative verification time.

Returns:

An immutable normalized provider identity.

Raises:

OAuthProviderError – If verification or any OIDC binding fails.

async refresh(refresh_token: SecretStr, *, current_scopes: frozenset[str] | None = None, now: datetime | None = None) → ProviderTokenSet

Refresh provider credentials.

Parameters:

• refresh_token – The protected refresh credential.

• current_scopes – Existing granted scopes when the provider omits them.

• now – The authoritative response time.

Returns:

The rotated token set.

async revoke(token: SecretStr, *, token_type_hint: str | None) → None

Revoke a provider credential.

Parameters:

• token – The credential to revoke.

• token_type_hint – Its optional standardized kind.

async aclose() → None

Close the owned OAuth transport.

__init__(oauth: OAuthProviderClient, metadata: OIDCMetadata, verifier: JWTVerifier[JWTClaims], retain_tokens_by_default: bool = True) → None

class litestar_security.providers.oidc.ServiceTokenConfig(issuer: str, audiences: frozenset[str], allowed_algorithms: frozenset[str], jwks: ~litestar_security.providers.jwks._provider.JWKSProvider, jwks_uri: str, scopes_claim: str = 'scope', actor_id_claim: str = 'sub', clock_skew: ~datetime.timedelta = datetime.timedelta(seconds=30), worker_limits: ~litestar_security.workers.WorkerLimits = <factory>)

Bases: object

Pinned external workload-token trust and claim profile.

build(*, clock: Callable[[], ~datetime.datetime]=<function ServiceTokenConfig.<lambda>>) → tuple[CredentialSlot[str], AuthenticationMechanism[str, JWTClaims, object]]

Build one native bearer slot and external service mechanism.

Parameters:

clock – Time source used by the composite bearer verifier.

Returns:

The physical bearer slot and service authentication mechanism.

__init__(issuer: str, audiences: frozenset[str], allowed_algorithms: frozenset[str], jwks: ~litestar_security.providers.jwks._provider.JWKSProvider, jwks_uri: str, scopes_claim: str = 'scope', actor_id_claim: str = 'sub', clock_skew: ~datetime.timedelta = datetime.timedelta(seconds=30), worker_limits: ~litestar_security.workers.WorkerLimits = <factory>) → None

async litestar_security.providers.oidc.discover_google_oidc_provider(*, client_id: str, client_secret: SecretStr, discovery: OIDCDiscoveryClient, jwks: JWKSProvider, scopes: frozenset[str] = frozenset({'email', 'openid', 'profile'}), offline_access: bool = False, worker_limits: WorkerLimits | None = None, http_policy: OAuthHTTPPolicy | None = None) → OIDCProvider

Discover Google’s exact issuer using caller-owned shared resources.

async litestar_security.providers.oidc.discover_oidc_provider(*, name: str, issuer: str, client_id: str, client_secret: SecretStr | None, discovery: OIDCDiscoveryClient, jwks: JWKSProvider, scopes: frozenset[str] = frozenset({'email', 'openid', 'profile'}), client_auth: OAuthClientAuth = OAuthClientAuth.CLIENT_SECRET_BASIC, worker_limits: WorkerLimits | None = None, http_policy: OAuthHTTPPolicy | None = None) → OIDCProvider

Discover and construct OIDC over application-owned shared resources.

Parameters:

• name – Stable local provider name.

• issuer – Exact issuer allowed by the discovery client.

• client_id – Registered OIDC client identifier.

• client_secret – Protected client secret when required.

• discovery – Shared application-owned discovery client.

• jwks – Shared application-owned JWKS provider.

• scopes – Allowlisted provider scopes.

• client_auth – Token endpoint client authentication method.

• worker_limits – Shared bounded crypto-worker budget.

• http_policy – Bounded OAuth transport policy.

Returns:

A discovered provider whose OAuth client is owned by the result.

Raises:

• ImproperlyConfiguredException – If shared resources or trust inputs are invalid.

• OIDCDiscoveryError – If discovery fails.

litestar_security.providers.oidc.google_oidc_provider(*, client_id: str, client_secret: SecretStr, metadata: OIDCMetadata, verifier: JWTVerifier[JWTClaims], scopes: frozenset[str] = frozenset({'email', 'openid', 'profile'}), offline_access: bool = False, http_policy: OAuthHTTPPolicy | None = None) → OIDCProvider

Construct Google’s pinned OIDC profile.

Parameters:

• client_id – Registered Google client identifier.

• client_secret – Protected Google client secret.

• metadata – Validated Google discovery result.

• verifier – ID-token verifier pinned to Google and the client.

• scopes – Allowed Google scopes.

• offline_access – Request refresh access while the user is absent.

• http_policy – Bounded OAuth transport policy.

Returns:

The Google OIDC provider.

Raises:

ImproperlyConfiguredException – If discovery does not name Google’s exact issuer.

litestar_security.providers.oidc.keycloak_oidc_provider(*, base_url: str, realm: str, client_id: str, client_secret: SecretStr | None, metadata: OIDCMetadata, verifier: JWTVerifier[JWTClaims], scopes: frozenset[str] = frozenset({'email', 'openid', 'profile'}), client_auth: OAuthClientAuth = OAuthClientAuth.CLIENT_SECRET_BASIC, http_policy: OAuthHTTPPolicy | None = None) → OIDCProvider

Construct a Keycloak provider pinned to one exact realm issuer.

Parameters:

• base_url – Exact HTTPS Keycloak base URL.

• realm – Exact realm path segment.

• client_id – Registered Keycloak client identifier.

• client_secret – Protected client secret, if required.

• metadata – Validated realm discovery result.

• verifier – ID-token verifier pinned to the realm and client.

• scopes – Allowed Keycloak scopes.

• client_auth – Token-endpoint client authentication method.

• http_policy – Bounded OAuth transport policy.

Returns:

The realm-pinned Keycloak OIDC provider.

Raises:

ImproperlyConfiguredException – If the realm issuer is not exact.

litestar_security.providers.oidc.map_keycloak_claims(claims: JWTClaims) → KeycloakClaims | InvalidCredentials

Map verified Keycloak claims without discovery, HTTP, or token exchange.

Parameters:

claims – Claims returned by an already-successful JWT verifier.

Returns:

Validated Keycloak authorization fields or InvalidCredentials.

litestar_security.providers.oidc.oidc_provider(*, name: str, client_id: str, client_secret: SecretStr | None, metadata: OIDCMetadata, verifier: JWTVerifier[JWTClaims], scopes: frozenset[str] = frozenset({'email', 'openid', 'profile'}), client_auth: OAuthClientAuth = OAuthClientAuth.CLIENT_SECRET_BASIC, revocation_endpoint: str | None = None, extra_authorization_parameters: Mapping[str, str] | None = None, http_policy: OAuthHTTPPolicy | None = None) → OIDCProvider

Construct a generic OIDC provider from validated discovery metadata.

Parameters:

• name – Stable local provider name.

• client_id – Registered OIDC client identifier.

• client_secret – Protected client secret, if required.

• metadata – Validated discovery result.

• verifier – Distinct verifier pinned to the issuer and client.

• scopes – Allowed request scopes; openid is always required.

• client_auth – Token-endpoint client authentication method.

• revocation_endpoint – Optional fixed revocation endpoint.

• extra_authorization_parameters – Optional fixed authorization parameters.

• http_policy – Bounded OAuth transport policy.

Returns:

A configured OIDC lifecycle provider.

Raises:

ImproperlyConfiguredException – If required discovery endpoints are absent.

Remote JWKS discovery with immutable per-issuer cache snapshots.

class litestar_security.providers.jwks.AsyncJWKSFetcher(*args, **kwargs)

Bases: Protocol

Async transport boundary for one exact configured JWKS source.

async fetch(request: JWKSFetchTarget) → JWKSFetchOutcome

Return one finite-byte response without following redirects.

Redirects are not followed, because a redirect could move key fetching to a host the operator never configured.

Parameters:

request – The issuer, configured JWKS URI, and optional ETag only; it carries no byte ceiling.

Returns:

A response with a finite bytes body.

Raises:

Exception – When the configured source cannot be fetched.

Notes

The cache parser enforces JWKSCachePolicy.maximum_document_bytes after fetch. The optional HTTPX fetcher separately enforces its configured transport response ceiling.

async aclose() → None

Close resources owned by the fetcher.

__init__(*args, **kwargs)

class litestar_security.providers.jwks.CachedJWKSProvider(entries: Sequence[JWKSSource], fetcher: AsyncJWKSFetcher | SyncJWKSFetcher, *, policy: JWKSCachePolicy | None = None, cache: JWKSCache | None = None, metrics: SecurityMetrics | None = None, fetcher_owned: bool = False, worker_limits: WorkerLimits | None = None)

Bases: object

Configured remote-key cache with a lock-free immutable fresh path.

__init__(entries: Sequence[JWKSSource], fetcher: AsyncJWKSFetcher | SyncJWKSFetcher, *, policy: JWKSCachePolicy | None = None, cache: JWKSCache | None = None, metrics: SecurityMetrics | None = None, fetcher_owned: bool = False, worker_limits: WorkerLimits | None = None) → None

Allocate every exact cache entry at startup.

async select_key(issuer: str, jwks_uri: str, kid: str, algorithm: str, *, now: datetime) → VerificationKey | InvalidCredentials | VerificationUnavailable

Read a fresh snapshot directly or refresh one exact entry.

A fresh snapshot is read without locking. Only a refresh coordinates, and the provider collapses concurrent refreshes of one entry into a single fetch.

Parameters:

• issuer – The token issuer, matched against configured trust anchors.

• jwks_uri – The key set to select from.

• kid – The exact key identifier named by the token header.

• algorithm – The algorithm named by the token header.

• now – The selection timestamp, used for freshness decisions.

Returns:

The verification key, InvalidCredentials when no configured key matches, or VerificationUnavailable when keys could not be reached.

async warmup(*, now: datetime) → VerificationUnavailable | None

Eagerly populate configured entries when startup warming is enabled.

Parameters:

now – The warm-up timestamp.

Returns:

None when warming succeeded or is disabled, otherwise VerificationUnavailable.

async aclose() → None

Close this provider idempotently without closing its caller-owned fetcher.

class litestar_security.providers.jwks.HttpxJWKSFetcher(timeout: float = 5.0, maximum_response_bytes: int = 1048576, allow_private_hosts: bool = False, transport: AsyncBaseTransport | None = None, resolver: Callable[[str, int], Awaitable[Sequence[str]]] | None = None)

Bases: object

HTTPX-backed async fetcher for operator-configured JWKS endpoints.

async fetch(request: JWKSFetchTarget) → JWKSFetchOutcome

Return one bounded response without following redirects.

Parameters:

request – The exact configured JWKS URI and optional ETag condition.

Returns:

The bounded transport response, including un-followed redirect status codes for the provider to reject.

Raises:

• _FetchGuardError – If the configured URI or host fails its network boundary, the response is encoded, or it exceeds its byte ceiling.

• httpx.HTTPError – If the outbound request fails. Any exception raised here becomes VerificationUnavailable at the JWKS provider.

async aclose() → None

Close the owned HTTP client idempotently.

Returns:

None.

__init__(timeout: float = 5.0, maximum_response_bytes: int = 1048576, allow_private_hosts: bool = False, transport: AsyncBaseTransport | None = None, resolver: Callable[[str, int], Awaitable[Sequence[str]]] | None = None) → None

class litestar_security.providers.jwks.InMemoryJWKSCache

Bases: object

Hold key snapshots for the lifetime of one process.

This is the default. Construct one explicitly and hand it to several providers to give them a shared key set and a shared fetch schedule.

__init__() → None

Start with no stored snapshots.

get(issuer: str, jwks_uri: str) → JWKSSnapshot | None

Return the stored snapshot for one configured source.

Parameters:

• issuer – The configured issuer the snapshot belongs to.

• jwks_uri – The key set the snapshot was parsed from.

Returns:

The stored snapshot, or None when nothing is stored.

set(issuer: str, jwks_uri: str, snapshot: JWKSSnapshot) → None

Store the newest snapshot for one configured source.

Parameters:

• issuer – The configured issuer the snapshot belongs to.

• jwks_uri – The key set the snapshot was parsed from.

• snapshot – The immutable snapshot to store.

invalidate(issuer: str, jwks_uri: str) → None

Drop any snapshot stored for one configured source.

Parameters:

• issuer – The configured issuer the snapshot belongs to.

• jwks_uri – The key set the snapshot was parsed from.

coordinator(issuer: str, jwks_uri: str) → JWKSCacheCoordinator

Return stable coordination state for one configured source.

Parameters:

• issuer – The configured issuer the coordination belongs to.

• jwks_uri – The configured key-set URI.

Returns:

Stable coordination state for the exact source pair.

class litestar_security.providers.jwks.JWKSCache(*args, **kwargs)

Bases: Protocol

Store remote key snapshots so components can share one fetch schedule.

An implementer must honor three invariants:

• Snapshots are immutable. Store and return the value as given; never mutate one in place, and never hand back a partially populated key set.

• ``set`` is last-write-wins. The most recent write for a key is the one a later get returns. No merging, no ordering by generation.

• A miss is indistinguishable from an expired entry. Returning None is always safe: the caller refetches. An implementation may therefore evict, expire, or bound itself however it likes, and must never fabricate or extend a snapshot to avoid a miss.

Methods are synchronous because they sit on the token-verification hot path, where the fresh read must not await.

get(issuer: str, jwks_uri: str) → JWKSSnapshot | None

Return the stored snapshot for one configured source.

Parameters:

• issuer – The configured issuer the snapshot belongs to.

• jwks_uri – The key set the snapshot was parsed from.

Returns:

The stored snapshot, or None when nothing is stored.

set(issuer: str, jwks_uri: str, snapshot: JWKSSnapshot) → None

Store the newest snapshot for one configured source.

Parameters:

• issuer – The configured issuer the snapshot belongs to.

• jwks_uri – The key set the snapshot was parsed from.

• snapshot – The immutable snapshot to store.

invalidate(issuer: str, jwks_uri: str) → None

Drop any snapshot stored for one configured source.

Dropping an absent entry is not an error.

Parameters:

• issuer – The configured issuer the snapshot belongs to.

• jwks_uri – The key set the snapshot was parsed from.

coordinator(issuer: str, jwks_uri: str) → JWKSCacheCoordinator

Return stable coordination state for one configured source.

Calls for the same exact pair must return the same object so providers sharing this cache also share refresh and unknown-key coordination.

Parameters:

• issuer – The configured issuer the coordination belongs to.

• jwks_uri – The configured key-set URI.

Returns:

Stable coordination state for the exact source pair.

__init__(*args, **kwargs)

class litestar_security.providers.jwks.JWKSCacheCoordinator(lock: ~anyio.Lock = <factory>, refresh: object | None = None, forced_generation: int | None = None, negative: ~collections.OrderedDict[tuple[int, str, str], ~datetime.datetime] = <factory>, users: int = 0)

Bases: object

Share refresh and negative-key state for one exact cache entry.

Cache implementations return the same coordinator for repeated requests for one exact (issuer, jwks_uri) pair. Applications normally only construct this value while implementing JWKSCache; providers manage its contents.

Parameters:

• lock – Lock serializing refresh and negative-key changes.

• refresh – Opaque in-flight refresh state owned by a provider.

• forced_generation – The generation whose unknown-key refresh was used.

• negative – Bounded generation-scoped unknown-key expirations.

• users – Number of providers attached to this coordination state.

__init__(lock: ~anyio.Lock = <factory>, refresh: object | None = None, forced_generation: int | None = None, negative: ~collections.OrderedDict[tuple[int, str, str], ~datetime.datetime] = <factory>, users: int = 0) → None

class litestar_security.providers.jwks.JWKSCachePolicy(default_ttl: timedelta = datetime.timedelta(seconds=900), minimum_ttl: timedelta = datetime.timedelta(seconds=30), maximum_ttl: timedelta = datetime.timedelta(days=1), unknown_kid_cooldown: timedelta = datetime.timedelta(seconds=30), stale_if_error: timedelta = datetime.timedelta(0), warm_on_startup: bool = False, maximum_document_bytes: int = 1048576, maximum_keys: int = 128, maximum_unknown_keys: int = 1024)

Bases: object

Local freshness and bounded-document policy for remote JWKS entries.

__init__(default_ttl: timedelta = datetime.timedelta(seconds=900), minimum_ttl: timedelta = datetime.timedelta(seconds=30), maximum_ttl: timedelta = datetime.timedelta(days=1), unknown_kid_cooldown: timedelta = datetime.timedelta(seconds=30), stale_if_error: timedelta = datetime.timedelta(0), warm_on_startup: bool = False, maximum_document_bytes: int = 1048576, maximum_keys: int = 128, maximum_unknown_keys: int = 1024) → None

class litestar_security.providers.jwks.JWKSFetchOutcome(status_code: int, body: bytes = b'', headers: Mapping[str, str]=<factory>)

Bases: object

Transport-neutral bounded response returned by a custom fetcher.

__init__(status_code: int, body: bytes = b'', headers: Mapping[str, str]=<factory>) → None

class litestar_security.providers.jwks.JWKSFetchTarget(issuer: str, jwks_uri: str, etag: str | None = None)

Bases: object

One conditional request for an exact configured JWKS source.

__init__(issuer: str, jwks_uri: str, etag: str | None = None) → None

class litestar_security.providers.jwks.JWKSProvider(*args, **kwargs)

Bases: Protocol

Select remote verification keys without exposing cache internals.

async select_key(issuer: str, jwks_uri: str, kid: str, algorithm: str, *, now: datetime) → VerificationKey | InvalidCredentials | VerificationUnavailable

Return a key or one stable authentication outcome.

Parameters:

• issuer – The token issuer, matched against configured trust anchors.

• jwks_uri – The key set to select from.

• kid – The exact key identifier named by the token header.

• algorithm – The algorithm named by the token header.

• now – The selection timestamp, used for freshness decisions.

Returns:

The verification key, InvalidCredentials when no configured key matches, or VerificationUnavailable when keys could not be reached.

async warmup(*, now: datetime) → VerificationUnavailable | None

Warm configured entries when enabled.

Parameters:

now – The warm-up timestamp.

Returns:

None when warming succeeded or is disabled, otherwise VerificationUnavailable.

async aclose() → None

Close owned runtime resources.

__init__(*args, **kwargs)

class litestar_security.providers.jwks.JWKSSnapshot(keys: Mapping[tuple[str, str], VerificationKey], etag: str | None, fresh_until: datetime, stale_until: datetime, generation: int, source_uri: str)

Bases: object

One immutable parsed key set together with its freshness bounds.

Parameters:

• keys – Verification keys indexed by the exact (kid, algorithm) pair a token header names.

• etag – The entity tag the source returned, used for conditional refresh.

• fresh_until – When the snapshot stops being served without a refresh.

• stale_until – How long the snapshot may still answer while the source is unreachable.

• generation – Increases on every parsed replacement, so a consumer can tell a rotation from a revalidation.

• source_uri – The key set this snapshot was parsed from.

__init__(keys: Mapping[tuple[str, str], VerificationKey], etag: str | None, fresh_until: datetime, stale_until: datetime, generation: int, source_uri: str) → None

class litestar_security.providers.jwks.JWKSSource(issuer: str, jwks_uri: str, algorithms: frozenset[str])

Bases: object

One exact configured issuer and JWKS source.

__init__(issuer: str, jwks_uri: str, algorithms: frozenset[str]) → None

class litestar_security.providers.jwks.SyncJWKSFetcher(*args, **kwargs)

Bases: Protocol

Blocking transport boundary normalized once into a bounded worker.

fetch(request: JWKSFetchTarget) → JWKSFetchOutcome

Return one finite-byte response without following redirects.

Redirects are not followed, because a redirect could move key fetching to a host the operator never configured.

Parameters:

request – The issuer, configured JWKS URI, and optional ETag only; it carries no byte ceiling.

Returns:

A response with a finite bytes body.

Raises:

Exception – When the configured source cannot be fetched.

Notes

The cache parser enforces JWKSCachePolicy.maximum_document_bytes after fetch. The optional HTTPX fetcher separately enforces its configured transport response ceiling.

__init__(*args, **kwargs)

litestar_security.providers.jwks.normalize_fetcher(fetcher: AsyncJWKSFetcher | SyncJWKSFetcher, *, limiter: CapacityLimiter, timeout: float = 10.0, metrics: SecurityMetrics | None = None) → AsyncJWKSFetcher

Normalize one custom transport once at configuration time.

Parameters:

• fetcher – The application’s transport, blocking or async.

• limiter – The capacity limiter a blocking transport runs inside.

• timeout – How long one blocking fetch may occupy a worker.

• metrics – The sink offered fetch measurements.

Returns:

An async fetcher. A blocking transport is wrapped so it never occupies the event loop.

OAuth authorization transaction and provider lifecycle contracts.

class litestar_security.providers.oauth.AESGCMOAuthTransactionProtector(active_key: ~litestar_security.providers.oauth._transactions.OAuthTransactionProtectorKey, retained_keys: tuple[~litestar_security.providers.oauth._transactions.OAuthTransactionProtectorKey, ...] = (), entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>)

Bases: object

Protect OAuth transaction secrets with AES-256-GCM application-owned keys.

property active_key_version: str

Return the version used by the next protection operation.

async protect(secret: bytes, *, associated_data: bytes) → ProtectedOAuthSecret

Encrypt one transaction secret under exact associated data.

Parameters:

• secret – Plaintext transaction secret bytes.

• associated_data – Unencrypted transaction and purpose binding.

Returns:

A versioned, nonce-prefixed ciphertext envelope.

Raises:

ValueError – If the entropy source does not return a 12-byte nonce.

async unprotect(protected: ProtectedOAuthSecret, *, associated_data: bytes) → bytes

Decrypt one envelope only under its original associated data.

Parameters:

• protected – Versioned ciphertext envelope.

• associated_data – Exact unencrypted transaction and purpose binding.

Returns:

The authenticated plaintext bytes.

Raises:

• ValueError – If the key version or ciphertext envelope is invalid.

• cryptography.exceptions.InvalidTag – If authentication fails.

__init__(active_key: ~litestar_security.providers.oauth._transactions.OAuthTransactionProtectorKey, retained_keys: tuple[~litestar_security.providers.oauth._transactions.OAuthTransactionProtectorKey, ...] = (), entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>) → None

exception litestar_security.providers.oauth.AccountLinkError(code: str = 'oauth_account_denied')

Bases: OAuthAccountError

Reject a duplicate cross-account provider identity.

class litestar_security.providers.oauth.GitHubOAuthProvider(*, client_id: str, client_secret: SecretStr, scopes: frozenset[str] = frozenset({'read:user', 'user:email'}), policy: OAuthHTTPPolicy | None = None, transport: AsyncBaseTransport | None = None)

Bases: object

GitHub OAuth specialization using current profile and verified-email APIs.

__init__(*, client_id: str, client_secret: SecretStr, scopes: frozenset[str] = frozenset({'read:user', 'user:email'}), policy: OAuthHTTPPolicy | None = None, transport: AsyncBaseTransport | None = None) → None

Create a GitHub provider with fixed authorization and API endpoints.

Parameters:

• client_id – Registered GitHub OAuth application identifier.

• client_secret – Protected GitHub OAuth application secret.

• scopes – Allowlisted request scopes.

• policy – Bounded shared HTTP policy.

• transport – Optional application or test transport.

property name: str

Return GitHub’s stable local provider name.

build_authorization_url(start: OAuthTransactionStart) → str

Build the GitHub Authorization Code plus PKCE URL.

Parameters:

start – The bound transaction start values.

Returns:

The fixed GitHub authorization URL.

async exchange_code(*, code: SecretStr, transaction: OAuthTransaction, now: datetime | None = None) → ProviderTokenSet

Exchange a GitHub authorization code.

Parameters:

• code – The callback code.

• transaction – The consumed transaction.

• now – The authoritative response time.

Returns:

The validated GitHub token set.

async resolve_identity(tokens: ProviderTokenSet, *, transaction: OAuthTransaction, now: datetime | None = None) → ProviderIdentity

Re-fetch GitHub profile and verified email for this login.

Parameters:

• tokens – The validated GitHub token set.

• transaction – The consumed transaction.

• now – The authoritative identity resolution time.

Returns:

Identity keyed only by GitHub’s stable numeric user ID.

Raises:

OAuthProviderError – If the transaction, scopes, or API responses are invalid.

async refresh(refresh_token: SecretStr, *, current_scopes: frozenset[str] | None = None, now: datetime | None = None) → ProviderTokenSet

Refresh an expiring GitHub user token.

Parameters:

• refresh_token – The protected GitHub refresh credential.

• current_scopes – Current grant used when the response omits scope.

• now – The authoritative response time.

Returns:

The rotated GitHub token set.

async revoke(token: SecretStr, *, token_type_hint: str | None) → None

Delete one GitHub OAuth application token grant.

Parameters:

• token – The access token to delete.

• token_type_hint – Must be access_token when supplied.

Raises:

OAuthProviderError – If deletion fails.

async aclose() → None

Close the shared owned GitHub HTTP client.

exception litestar_security.providers.oauth.InvalidOAuthCallback

Bases: RuntimeError

Reject every invalid OAuth callback with one stable public outcome.

__init__() → None

Initialize a generic secret-free failure.

exception litestar_security.providers.oauth.InvalidProviderGrantError(*, closed: bool = False, retry_after: int | None = None)

Bases: OAuthProviderError

Indicate that refresh requires provider reauthorization.

class litestar_security.providers.oauth.LinkedProviderAccount(provider_account_id: str, account_id: str, provider: str, issuer: str, subject: str, grant: ProviderGrant, linked_at: datetime)

Bases: object

One exact provider identity linked to one application account.

__init__(provider_account_id: str, account_id: str, provider: str, issuer: str, subject: str, grant: ProviderGrant, linked_at: datetime) → None

class litestar_security.providers.oauth.MemoryOAuthAccountStore(*, login_method_counts: Mapping[str, int] | None = None, provider: str = 'example', client_id: str = 'client', protector: OAuthTransactionProtector | None = None)

Bases: object

Atomic in-memory reference store for provider account behavior.

__init__(*, login_method_counts: Mapping[str, int] | None = None, provider: str = 'example', client_id: str = 'client', protector: OAuthTransactionProtector | None = None) → None

Create a store with authoritative total login-method counts.

Parameters:

• login_method_counts – Existing local and provider methods per account.

• provider – Provider namespace used in token associated data.

• client_id – OAuth client identifier used in token associated data.

• protector – Optional encryption port enabling retained tokens.

async login(identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, provision_unknown: bool, retain_tokens: bool = False, now: datetime) → OAuthLoginOutcome

Atomically resolve or provision one exact identity.

Parameters:

• identity – Exact provider identity.

• grant – Provider-observed grant.

• tokens – Exchanged provider tokens.

• provision_unknown – Whether an unknown identity may create an account.

• retain_tokens – Whether the aggregate should retain the token set.

• now – Aware mutation time.

Returns:

The linked account and whether this call provisioned it.

Raises:

OAuthAccountError – If the identity is unknown and provisioning is disabled.

async get_tokens(provider_account_id: str, *, now: datetime) → StoredProviderTokens | None

Return retained tokens when configured.

async replace_tokens(provider_account_id: str, *, expected_version: int, tokens: ProviderTokenSet, now: datetime) → bool

Compare and replace retained tokens.

async discard_tokens(provider_account_id: str, *, expected_version: int | None = None) → bool

Discard retained tokens at an optional observed version.

async stage_revocation_retry(failure: OAuthRevocationFailure, tokens: ProviderTokenSet, *, expected_version: int) → bool

Discard active tokens only when their observed version still matches.

async resolve_provider_account(account_id: str, provider: str) → LinkedProviderAccount | None

Resolve one exact account-owned provider link.

async link(account_id: str, identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, retain_tokens: bool = False, now: datetime) → LinkedProviderAccount

Commit an exact identity and its token policy under one aggregate lock.

async upgrade(account_id: str, provider_account_id: str, identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, retain_tokens: bool, now: datetime) → LinkedProviderAccount

Commit one grant and the exchanged token policy together.

async unlink(account_id: str, provider: str, provider_account_id: str, *, require_remaining: bool, now: datetime) → UnlinkOutcome

Remove one owned link and all retained credentials under the aggregate lock.

class litestar_security.providers.oauth.MemoryOAuthTransactionStore(*, protector: OAuthTransactionProtector, capacity: int = 1024, clock: Callable[[], datetime] | None = None)

Bases: object

Atomic in-memory reference store with protected recoverable secrets.

__init__(*, protector: OAuthTransactionProtector, capacity: int = 1024, clock: Callable[[], datetime] | None = None) → None

Initialize the reference store.

Parameters:

• protector – Application-owned transaction secret protection.

• capacity – Maximum number of live transactions retained.

• clock – Aware time source used for bounded expiry cleanup.

Raises:

ImproperlyConfiguredException – If the protector contract is absent.

async create(transaction: OAuthTransaction) → None

Protect and persist one new transaction.

Parameters:

transaction – The validated server-side transaction.

Raises:

ValueError – If an identical transaction lookup already exists.

async consume(*, state_digest: bytes, binding_digest: bytes, provider: str, now: datetime) → OAuthTransaction | None

Atomically return and remove one exact, unexpired match.

Parameters:

• state_digest – The state lookup digest.

• binding_digest – The dedicated browser-cookie digest.

• provider – The provider route receiving the callback.

• now – The authoritative callback time.

Returns:

The one consumed transaction, or None for every lookup miss.

exception litestar_security.providers.oauth.OAuthAccountError(code: str = 'oauth_account_denied')

Bases: RuntimeError

Stable secret-free account lifecycle failure.

__init__(code: str = 'oauth_account_denied') → None

Initialize one stable application-facing code.

class litestar_security.providers.oauth.OAuthAccountService(*, store: OAuthAccountStore, provision_unknown: bool = False)

Bases: object

Coordinate exact login/link/scope/vault behavior over atomic ports.

__init__(*, store: OAuthAccountStore, provision_unknown: bool = False) → None

Create the account lifecycle service.

Parameters:

• store – Atomic provider-account store.

• provision_unknown – Whether the aggregate store may provision an unknown identity.

async login(identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, retain_tokens: bool = False, now: datetime) → OAuthLoginOutcome

Delegate the complete login mutation to the aggregate store.

async link(proof: OAuthLinkProof, identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, retain_tokens: bool = False, now: datetime) → LinkedProviderAccount

Link after exact fresh purpose and epoch validation.

async unlink(proof: OAuthLinkProof, provider: str, provider_account_id: str, *, now: datetime) → UnlinkOutcome

Atomically preserve a remaining login method, then discard tokens.

static missing_scopes(*, current: frozenset[str], requested: frozenset[str], allowed: frozenset[str]) → frozenset[str]

Return only allowlisted scopes absent from the current grant.

async apply_scope_upgrade(proof: OAuthLinkProof, provider_account_id: str, identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, required_scopes: frozenset[str], retain_tokens: bool = False, now: datetime) → LinkedProviderAccount

Record only the provider’s actual grant after step-up.

async refresh(provider_account_id: str, provider: OAuthProvider, *, now: datetime) → ProviderTokenSet

Single-flight refresh and optimistic rotation for one provider account.

async revoke(provider_account_id: str, provider: OAuthProvider, *, now: datetime) → None

Revoke retained credentials without losing material needed for retry.

class litestar_security.providers.oauth.OAuthAccountStore(*args, **kwargs)

Bases: Protocol

Atomic behavior-oriented provider account persistence boundary.

async login(identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, provision_unknown: bool, retain_tokens: bool, now: datetime) → OAuthLoginOutcome

Atomically resolve or provision, link, observe, and retain or discard tokens.

async get_tokens(provider_account_id: str, *, now: datetime) → StoredProviderTokens | None

Return decrypted provider tokens for the owning coordinator.

async link(account_id: str, identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, retain_tokens: bool, now: datetime) → LinkedProviderAccount

Atomically link an identity, grant, and optional retained tokens.

async upgrade(account_id: str, provider_account_id: str, identity: ProviderIdentity, grant: ProviderGrant, tokens: ProviderTokenSet, *, retain_tokens: bool, now: datetime) → LinkedProviderAccount

Atomically replace a grant and its newly exchanged tokens.

async unlink(account_id: str, provider: str, provider_account_id: str, *, require_remaining: bool, now: datetime) → UnlinkOutcome

Atomically remove the link, grant, and retained tokens while preserving another method.

async replace_tokens(provider_account_id: str, *, expected_version: int, tokens: ProviderTokenSet, now: datetime) → bool

Compare and replace retained tokens atomically.

async discard_tokens(provider_account_id: str, *, expected_version: int | None = None) → bool

Discard tokens, optionally only at one observed version.

async stage_revocation_retry(failure: OAuthRevocationFailure, tokens: ProviderTokenSet, *, expected_version: int) → bool

Atomically move active tokens into durable revocation-retry state.

async resolve_provider_account(account_id: str, provider: str) → LinkedProviderAccount | None

Resolve one account-owned provider link without crossing ownership.

__init__(*args, **kwargs)

class litestar_security.providers.oauth.OAuthAuthorization(*, url: str, binding_cookie: Cookie)

Bases: Struct

Authorization redirect and dedicated browser-binding cookie.

class litestar_security.providers.oauth.OAuthCallbackOutcome(operation: OAuthOperation, return_to: str, identity: ProviderIdentity, linked: LinkedProviderAccount, authenticated_at: datetime, provisioned: bool)

Bases: object

Presentation-neutral result of one consumed OAuth callback.

__init__(operation: OAuthOperation, return_to: str, identity: ProviderIdentity, linked: LinkedProviderAccount, authenticated_at: datetime, provisioned: bool) → None

class litestar_security.providers.oauth.OAuthClientAuth(*values)

Bases: str, Enum

Supported OAuth token-endpoint client authentication methods.

class litestar_security.providers.oauth.OAuthConfig(*, oauth_service: OAuthLifecycle, oidc_service: OIDCLogoutLifecycleService | None = None, route_prefix: str = '/auth', register_routes: bool = True, docs: RouteDocs | None = None)

Bases: object

Interactive provider route configuration and service graph.

__init__(*, oauth_service: OAuthLifecycle, oidc_service: OIDCLogoutLifecycleService | None = None, route_prefix: str = '/auth', register_routes: bool = True, docs: RouteDocs | None = None) → None

Validate provider uniqueness and generated-route ownership.

Parameters:

• oauth_service – Shared route and custom-controller service.

• oidc_service – Optional verified OIDC logout workflow.

• route_prefix – Absolute non-root mount path.

• register_routes – Whether the plugin installs generated routes.

• docs – Application-owned OpenAPI documentation for the generated routes: tag renames, tag descriptions, and optional operation-id and route-name transforms.

Raises:

ImproperlyConfiguredException – If any input is invalid.

build_route_handlers(*, wire: WirePolicy | None = None) → tuple[Router, ...]

Build and cache generated OAuth routes.

One router is cached per wire policy rather than one overall, so a router stays a pure function of the configuration that caches it and two applications sharing this configuration with different casing each get their own.

Parameters:

wire – How the generated bodies are spelled on the wire. Defaults to the field names as Python spells them, with unknown members rejected.

Returns:

One router, or an empty tuple when register_routes is False. The same object is returned for every call naming the same policy.

class litestar_security.providers.oauth.OAuthConfirmation(*, return_to: str = '/')

Bases: WireStruct

Provider confirmation redirect request.

class litestar_security.providers.oauth.OAuthEndpointConfig(name: str, client_id: str, client_secret: ~litestar_security.providers.oauth._transactions.SecretStr | None, client_auth: ~litestar_security.providers.oauth._provider.OAuthClientAuth, authorization_endpoint: str, token_endpoint: str, revocation_endpoint: str | None, allowed_scopes: frozenset[str], required_scopes: frozenset[str], extra_authorization_parameters: ~collections.abc.Mapping[str, str] = <factory>)

Bases: object

Normalized operator-controlled endpoints and provider client policy.

__init__(name: str, client_id: str, client_secret: ~litestar_security.providers.oauth._transactions.SecretStr | None, client_auth: ~litestar_security.providers.oauth._provider.OAuthClientAuth, authorization_endpoint: str, token_endpoint: str, revocation_endpoint: str | None, allowed_scopes: frozenset[str], required_scopes: frozenset[str], extra_authorization_parameters: ~collections.abc.Mapping[str, str] = <factory>) → None

class litestar_security.providers.oauth.OAuthHTTPPolicy(connect_timeout: float = 2.0, read_timeout: float = 5.0, write_timeout: float = 5.0, pool_timeout: float = 2.0, maximum_connections: int = 20, maximum_response_bytes: int = 65536)

Bases: object

Bounded HTTP resource policy shared by one provider client.

__init__(connect_timeout: float = 2.0, read_timeout: float = 5.0, write_timeout: float = 5.0, pool_timeout: float = 2.0, maximum_connections: int = 20, maximum_response_bytes: int = 65536) → None

class litestar_security.providers.oauth.OAuthLifecycle(*args, **kwargs)

Bases: Protocol

Application boundary used identically by generated or custom controllers.

property provider_names: frozenset[str]

Return the exact configured interactive provider names.

async begin(*, provider: str, operation: OAuthOperation, account_id: str | None, provider_account_id: str | None, return_to: str, scopes: frozenset[str] | None, step_up_grant: str | None, request: Request[Any, Any, Any]) → OAuthAuthorization

Create one transaction and return its safe redirect.

async complete_callback(*, provider: str, code: str, state: str, request: Request[Any, Any, Any]) → OAuthCallbackOutcome | OAuthReauthenticationOutcome | OAuthRevalidationOutcome

Consume a callback and commit provider-account state without presentation adaptation.

async establish_login(outcome: OAuthCallbackOutcome, *, request: Request[Any, Any, Any]) → OAuthOperationSummary | OAuthReauthenticationOutcome | OAuthRevalidationOutcome | Response[Any]

Establish the configured local transport for a completed login.

async revalidate(*, provider: str, account_id: str, return_to: str, request: Request[Any, Any, Any]) → OAuthAuthorization

Begin exact linked-provider possession confirmation.

async reauthenticate(*, provider: str, purpose: str, account_id: str, return_to: str, request: Request[Any, Any, Any]) → OAuthAuthorization

Begin capability-gated OIDC freshness verification.

async unlink(*, provider: str, provider_account_id: str, account_id: str, step_up_grant: str, request: Request[Any, Any, Any]) → OAuthOperationSummary

Atomically unlink a provider account.

async revoke(*, provider: str, account_id: str, step_up_grant: str, request: Request[Any, Any, Any]) → OAuthOperationSummary

Locally delete and attempt upstream revocation.

async logout(*, provider: str, account_id: str, request: Request[Any, Any, Any]) → OAuthLogout

Complete local logout independently of provider availability.

__init__(*args, **kwargs)

class litestar_security.providers.oauth.OAuthLifecycleService(*, registrations: tuple[~litestar_security.providers.oauth._routes.OAuthProviderRegistration, ...], transactions: ~litestar_security.providers.oauth._transactions.OAuthTransactionService, accounts: ~litestar_security.providers.oauth._accounts.OAuthAccountService, local: ~litestar_security.providers.oauth._routes.OAuthLocalTransport, step_up: ~litestar_security.providers.oauth._routes.OAuthStepUpAuthorizer | None = None, clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function OAuthLifecycleService.<lambda>>)

Bases: object

Concrete OAuth transaction, provider, account, and local-login workflow.

__init__(*, registrations: tuple[~litestar_security.providers.oauth._routes.OAuthProviderRegistration, ...], transactions: ~litestar_security.providers.oauth._transactions.OAuthTransactionService, accounts: ~litestar_security.providers.oauth._accounts.OAuthAccountService, local: ~litestar_security.providers.oauth._routes.OAuthLocalTransport, step_up: ~litestar_security.providers.oauth._routes.OAuthStepUpAuthorizer | None = None, clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function OAuthLifecycleService.<lambda>>) → None

Build one application-lifecycle-owned OAuth service graph.

property provider_names: frozenset[str]

Return configured provider names.

async begin(*, provider: str, operation: OAuthOperation, account_id: str | None, provider_account_id: str | None, return_to: str, scopes: frozenset[str] | None, step_up_grant: str | None, request: Request[Any, Any, Any]) → OAuthAuthorization

Consume required step-up and create one bound authorization transaction.

async callback(*, provider: str, code: str, state: str, request: Request[Any, Any, Any]) → OAuthOperationSummary | OAuthReauthenticationOutcome | OAuthRevalidationOutcome | Response[Any]

Adapt a neutral callback outcome to the generated route response.

async complete_callback(*, provider: str, code: str, state: str, request: Request[Any, Any, Any]) → OAuthCallbackOutcome | OAuthReauthenticationOutcome | OAuthRevalidationOutcome

Consume a callback and commit account state without presenting HTTP or establishing a session.

async revalidate(*, provider: str, account_id: str, return_to: str, request: Request[Any, Any, Any]) → OAuthAuthorization

Begin exact linked-provider possession confirmation without freshness semantics.

async reauthenticate(*, provider: str, purpose: str, account_id: str, return_to: str, request: Request[Any, Any, Any]) → OAuthAuthorization

Begin configured OIDC reauthentication for one exact purpose.

async establish_login(outcome: OAuthCallbackOutcome, *, request: Request[Any, Any, Any]) → OAuthOperationSummary | OAuthReauthenticationOutcome | OAuthRevalidationOutcome | Response[Any]

Establish the configured local transport for a completed login only.

async aclose() → None

Close each lifecycle-owned provider exactly once.

async unlink(*, provider: str, provider_account_id: str, account_id: str, step_up_grant: str, request: Request[Any, Any, Any]) → OAuthOperationSummary

Consume step-up and atomically unlink one account-owned provider identity.

async revoke(*, provider: str, account_id: str, step_up_grant: str, request: Request[Any, Any, Any]) → OAuthOperationSummary

Consume step-up and revoke the exact account-owned provider grant.

async logout(*, provider: str, account_id: str, request: Request[Any, Any, Any]) → OAuthLogout

Complete local logout before returning an optional fixed RP redirect.

class litestar_security.providers.oauth.OAuthLink(*, step_up_grant: str, return_to: str = '/')

Bases: WireStruct

Purpose-bound link request.

class litestar_security.providers.oauth.OAuthLinkProof(account_id: str, purpose: str, security_epoch: int, transaction_account_id: str, transaction_security_epoch: int, consumed: bool)

Bases: object

Consumed purpose-bound proof tied to account and security epoch.

valid_for(purpose: str) → bool

Return whether every callback binding remains current.

Parameters:

purpose – Required operation purpose.

Returns:

Whether account, epoch, purpose, and consumption all match.

__init__(account_id: str, purpose: str, security_epoch: int, transaction_account_id: str, transaction_security_epoch: int, consumed: bool) → None

class litestar_security.providers.oauth.OAuthLocalTransport(*args, **kwargs)

Bases: Protocol

Establish and revoke the configured local authentication transport.

async establish(*, account_id: str, identity: ProviderIdentity, request: Request[Any, Any, Any], authenticated_at: datetime) → OAuthOperationSummary | Response[Any]

Establish a session, token pair, or explicit hybrid transport.

async logout(*, account_id: str, request: Request[Any, Any, Any]) → None

Invalidate the configured local transport.

__init__(*args, **kwargs)

class litestar_security.providers.oauth.OAuthLoginOutcome(linked: LinkedProviderAccount, provisioned: bool)

Bases: object

Atomic login result and first-provisioning signal.

__init__(linked: LinkedProviderAccount, provisioned: bool) → None

class litestar_security.providers.oauth.OAuthLogout(*, detail: str = 'Logged out.', redirect_url: str | None = None)

Bases: Struct

Local logout confirmation plus optional validated provider redirect.

class litestar_security.providers.oauth.OAuthOperation(*values)

Bases: str, Enum

Purpose bound to one OAuth authorization transaction.

class litestar_security.providers.oauth.OAuthOperationSummary(*, detail: str, provider_account_id: str | None = None, account_id: str | None = None, revoked_sessions: int | None = None)

Bases: WireStruct

Secret-free provider lifecycle response.

Each identifier has its own member, and a response carries only the members its operation actually resolved. Linking reports the provider account it bound, establishing a local session reports the local account, and a logout reports how many sessions it revoked.

class litestar_security.providers.oauth.OAuthProvider(*args, **kwargs)

Bases: Protocol

Interactive OAuth provider lifecycle implemented by configured providers.

build_authorization_url(start: OAuthTransactionStart) → str

Build the provider URL from one ephemeral transaction start.

Parameters:

start – The transaction plus its redacted raw browser values.

Returns:

The fixed provider authorization endpoint and bound query.

async exchange_code(*, code: SecretStr, transaction: OAuthTransaction, now: datetime | None = None) → ProviderTokenSet

Exchange one callback code.

Parameters:

• code – The provider callback code.

• transaction – The atomically consumed transaction.

• now – The authoritative response time.

Returns:

Validated provider tokens.

async resolve_identity(tokens: ProviderTokenSet, *, transaction: OAuthTransaction, now: datetime | None = None) → ProviderIdentity

Resolve a verified identity from validated tokens.

Parameters:

• tokens – Validated provider tokens.

• transaction – The consumed transaction binding this identity lookup.

• now – The authoritative verification time.

Returns:

The verified provider identity.

async refresh(refresh_token: SecretStr, *, current_scopes: frozenset[str] | None = None, now: datetime | None = None) → ProviderTokenSet

Refresh provider credentials.

Parameters:

• refresh_token – The protected stored refresh credential.

• current_scopes – Current grant used when the response omits scope.

• now – The authoritative response time.

Returns:

The rotated validated provider token set.

async revoke(token: SecretStr, *, token_type_hint: str | None) → None

Revoke a provider credential.

Parameters:

• token – The credential to revoke.

• token_type_hint – The optional standardized token type hint.

__init__(*args, **kwargs)

class litestar_security.providers.oauth.OAuthProviderClient(config: OAuthEndpointConfig, *, policy: OAuthHTTPPolicy | None = None, transport: AsyncBaseTransport | None = None, resolver: Callable[[str, int], Awaitable[Sequence[str]]] | None = None)

Bases: object

Lifecycle-owned fixed-endpoint OAuth HTTP client.

__init__(config: OAuthEndpointConfig, *, policy: OAuthHTTPPolicy | None = None, transport: AsyncBaseTransport | None = None, resolver: Callable[[str, int], Awaitable[Sequence[str]]] | None = None) → None

Create one bounded async client.

Parameters:

• config – Normalized static provider configuration.

• policy – Explicit resource limits.

• transport – Optional test or application transport.

• resolver – Optional asynchronous endpoint address resolver.

property name: str

Return the configured provider name.

property closed: bool

Return whether the owned HTTP client has closed.

build_authorization_url(start: OAuthTransactionStart) → str

Build an Authorization Code plus S256 URL.

Parameters:

start – The ephemeral transaction start.

Returns:

The fixed authorization endpoint and encoded transaction bindings.

Raises:

OAuthProviderError – If the client is closed or the transaction does not match the provider configuration.

async exchange_code(*, code: SecretStr, transaction: OAuthTransaction, now: datetime | None = None) → ProviderTokenSet

Exchange one code using the exact redirect URI and PKCE verifier.

Parameters:

• code – The callback authorization code.

• transaction – The atomically consumed transaction.

• now – The authoritative response time.

Returns:

Validated provider tokens.

Raises:

OAuthProviderError – If the request or response fails validation.

async refresh(refresh_token: SecretStr, *, current_scopes: frozenset[str] | None = None, now: datetime | None = None) → ProviderTokenSet

Refresh provider credentials through the fixed token endpoint.

Parameters:

• refresh_token – The protected stored refresh credential.

• current_scopes – Current grant used when the response omits scope.

• now – The authoritative response time.

Returns:

Validated rotated tokens.

Raises:

OAuthProviderError – If the request or response fails validation.

async revoke(token: SecretStr, *, token_type_hint: str | None) → None

Revoke a provider credential at the configured fixed endpoint.

Parameters:

• token – The access or refresh credential.

• token_type_hint – Optional RFC 7009 token type hint.

Raises:

OAuthProviderError – If revocation is unsupported or fails.

async aclose() → None

Close the owned client idempotently.

exception litestar_security.providers.oauth.OAuthProviderError(*, closed: bool = False, retry_after: int | None = None)

Bases: RuntimeError

Stable secret-free provider request failure.

__init__(*, closed: bool = False, retry_after: int | None = None) → None

Initialize a closed-client or generic request failure.

class litestar_security.providers.oauth.OAuthProviderRegistration(provider: ~litestar_security.providers.oauth._provider.OAuthProvider, redirect_uri: str, default_scopes: frozenset[str], expected_issuer: str | None = None, include_nonce: bool = False, end_session_endpoint: str | None = None, post_logout_redirect_uri: str | None = None, retain_tokens: bool = False, reauthentication: ~collections.abc.Mapping[str, ~litestar_security.providers.oauth._routes.OIDCReauthenticationPolicy] = <factory>)

Bases: object

Static routing and protocol metadata for one interactive provider.

classmethod oidc(*, provider: OAuthProvider, redirect_uri: str, post_logout_redirect_uri: str | None = None) → OAuthProviderRegistration

Derive an OIDC registration from one validated provider.

Parameters:

• provider – Configured OIDC provider exposing validated metadata.

• redirect_uri – Exact application callback URI.

• post_logout_redirect_uri – Fixed return URI for provider logout.

Returns:

An immutable registration with nonce, issuer, scopes, logout, and retention derived.

Raises:

ImproperlyConfiguredException – If the provider is not a configured OIDC provider.

__init__(provider: ~litestar_security.providers.oauth._provider.OAuthProvider, redirect_uri: str, default_scopes: frozenset[str], expected_issuer: str | None = None, include_nonce: bool = False, end_session_endpoint: str | None = None, post_logout_redirect_uri: str | None = None, retain_tokens: bool = False, reauthentication: ~collections.abc.Mapping[str, ~litestar_security.providers.oauth._routes.OIDCReauthenticationPolicy] = <factory>) → None

class litestar_security.providers.oauth.OAuthReauthenticationOutcome(*, account_id: str, provider: str, provider_account_id: str, credential: StepUpCredential)

Bases: WireStruct

Exact OIDC freshness result containing one purpose-bound credential.

class litestar_security.providers.oauth.OAuthReauthenticationProvider(*args, **kwargs)

Bases: OAuthProvider, Protocol

Provider capable of forcing fresh, signed authentication evidence.

build_reauthentication_url(start: OAuthTransactionStart, *, max_age: int) → str

Build an authorization URL that requires signed provider freshness.

Parameters:

• start – The bound authorization transaction.

• max_age – Maximum accepted provider authentication age in seconds.

Returns:

The provider authorization URL with its freshness request.

class litestar_security.providers.oauth.OAuthRedirectPolicy(callback_uris: Mapping[str, frozenset[str]], return_to: frozenset[str] = frozenset({'/'}), allow_insecure_localhost: bool = False)

Bases: object

Configured exact callback and same-origin return destinations.

validate(*, provider: str, redirect_uri: str, return_to: str) → None

Require exact configured callback and return destinations.

Parameters:

• provider – The statically configured provider name.

• redirect_uri – The exact callback URI sent to the provider.

• return_to – The server-side post-login destination.

Raises:

InvalidOAuthCallback – If any value is absent or not an exact match.

__init__(callback_uris: Mapping[str, frozenset[str]], return_to: frozenset[str] = frozenset({'/'}), allow_insecure_localhost: bool = False) → None

class litestar_security.providers.oauth.OAuthRevalidationOutcome(*, account_id: str, provider: str, provider_account_id: str)

Bases: WireStruct

Exact linked-provider possession result without a freshness claim.

class litestar_security.providers.oauth.OAuthRevocationFailure(provider_account_id: str, failed_token_types: frozenset[str], occurred_at: datetime)

Bases: object

Secret-free upstream revocation retry classification.

__init__(provider_account_id: str, failed_token_types: frozenset[str], occurred_at: datetime) → None

class litestar_security.providers.oauth.OAuthScopeUpgrade(*, provider_account_id: str, scopes: frozenset[str], step_up_grant: str, return_to: str = '/')

Bases: WireStruct

Incremental provider-scope request.

class litestar_security.providers.oauth.OAuthStepUp(*, step_up_grant: str)

Bases: WireStruct

Provider-account action requiring fresh step-up.

class litestar_security.providers.oauth.OAuthStepUpAuthorization(security_epoch: int, session_binding: str | None)

Bases: object

Authoritative account epoch and transport binding from consumed step-up.

__init__(security_epoch: int, session_binding: str | None) → None

class litestar_security.providers.oauth.OAuthStepUpAuthorizer(*args, **kwargs)

Bases: Protocol

Consume purpose-bound grants and expose current authoritative epochs.

async authorize(*, grant: str, account_id: str, purpose: str, request: Request[Any, Any, Any]) → OAuthStepUpAuthorization

Consume one exact step-up grant for the current transport.

async current_security_epoch(account_id: str) → int

Return the current authoritative account security epoch.

session_binding(request: Request[Any, Any, Any]) → str | None

Return the current transport binding used by callback validation.

async issue(*, account_id: str, purpose: str, authenticated_at: datetime, acr: str | None, amr: tuple[str, ...], request: Request[Any, Any, Any]) → StepUpCredential

Issue one purpose-bound credential from provider freshness evidence.

__init__(*args, **kwargs)

class litestar_security.providers.oauth.OAuthTransaction(state_digest: bytes, binding_digest: bytes, operation: ~litestar_security.providers.oauth._transactions.OAuthOperation, provider: str, expected_issuer: str | None, redirect_uri: str, return_to: str, requested_scopes: frozenset[str], pkce_verifier: ~litestar_security.providers.oauth._transactions.SecretStr, nonce: ~litestar_security.providers.oauth._transactions.SecretStr | None = None, account_id: str | None = None, session_binding: str | None = None, security_epoch: int | None = None, provider_account_id: str | None = None, step_up_purpose: str | None = None, maximum_authentication_age: int | None = None, expires_at: ~datetime.datetime = <factory>)

Bases: object

Server-side state for one purpose-bound OAuth authorization request.

__init__(state_digest: bytes, binding_digest: bytes, operation: ~litestar_security.providers.oauth._transactions.OAuthOperation, provider: str, expected_issuer: str | None, redirect_uri: str, return_to: str, requested_scopes: frozenset[str], pkce_verifier: ~litestar_security.providers.oauth._transactions.SecretStr, nonce: ~litestar_security.providers.oauth._transactions.SecretStr | None = None, account_id: str | None = None, session_binding: str | None = None, security_epoch: int | None = None, provider_account_id: str | None = None, step_up_purpose: str | None = None, maximum_authentication_age: int | None = None, expires_at: ~datetime.datetime = <factory>) → None

class litestar_security.providers.oauth.OAuthTransactionProtector(*args, **kwargs)

Bases: Protocol

Protect recoverable transaction secrets with application-owned keys.

property active_key_version: str

Return the stable version used by the next protection operation.

async protect(secret: bytes, *, associated_data: bytes) → ProtectedOAuthSecret

Protect one secret under exact transaction-associated data.

Parameters:

• secret – The plaintext secret to protect.

• associated_data – The transaction identity and secret purpose.

Returns:

An opaque ciphertext envelope.

async unprotect(protected: ProtectedOAuthSecret, *, associated_data: bytes) → bytes

Recover one secret under its original transaction-associated data.

Parameters:

• protected – The stored opaque envelope.

• associated_data – The transaction identity and secret purpose.

Returns:

The recovered plaintext for immediate protocol use.

__init__(*args, **kwargs)

class litestar_security.providers.oauth.OAuthTransactionProtectorKey(key_version: str, key: bytes)

Bases: object

One AES-256-GCM OAuth transaction key selected by a non-secret version.

__init__(key_version: str, key: bytes) → None

class litestar_security.providers.oauth.OAuthTransactionService(store: OAuthTransactionStore, pepper: bytes, redirects: OAuthRedirectPolicy, lifetime: timedelta = datetime.timedelta(seconds=600), entropy: Callable[[int], bytes] | None = None)

Bases: object

Generate, persist, and atomically consume OAuth transactions.

async start(*, operation: OAuthOperation, provider: str, redirect_uri: str, return_to: str, requested_scopes: frozenset[str], now: datetime, include_nonce: bool, expected_issuer: str | None = None, account_id: str | None = None, session_binding: str | None = None, browser_binding: SecretStr | None = None, security_epoch: int | None = None, provider_account_id: str | None = None, step_up_purpose: str | None = None, maximum_authentication_age: int | None = None) → OAuthTransactionStart

Create and persist one independent browser transaction.

Parameters:

• operation – The exact login, link, or scope-upgrade purpose.

• provider – The configured provider receiving the authorization request.

• redirect_uri – The configured exact callback URI.

• return_to – The configured server-side post-login destination.

• requested_scopes – The immutable provider scope request.

• now – The authoritative creation time.

• include_nonce – Whether the provider uses an OIDC nonce.

• expected_issuer – The fixed issuer expected on callback.

• account_id – The account bound to a link or scope upgrade.

• session_binding – The optional Litestar session binding.

• browser_binding – An existing dedicated browser binding to reuse across concurrent transactions.

• security_epoch – Authoritative epoch bound by consumed step-up.

• provider_account_id – Provider link targeted by scope upgrade.

• step_up_purpose – Purpose a successful provider reauthentication may issue.

• maximum_authentication_age – Maximum signed provider authentication age in seconds.

Returns:

Browser-facing state, binding, challenge, nonce, and stored transaction.

Raises:

• InvalidOAuthCallback – If redirect or transaction inputs are invalid.

• OAuthTransactionUnavailable – If entropy, protection, or persistence fails.

async consume(*, state: SecretStr | str, browser_binding: SecretStr | str, provider: str, operation: OAuthOperation | None, session_binding: str | None, now: datetime) → OAuthTransaction

Atomically consume one exact callback transaction.

Parameters:

• state – The provider-returned opaque state.

• browser_binding – The dedicated cookie value.

• provider – The provider route receiving the callback.

• operation – The operation expected by that route, or None when a shared callback dispatches from the consumed transaction.

• session_binding – The optional current Litestar session binding.

• now – The authoritative callback time.

Returns:

The consumed, recovered transaction.

Raises:

• InvalidOAuthCallback – For every absent, expired, replayed, or mismatched callback.

• OAuthTransactionUnavailable – If persistence or protection fails.

__init__(store: OAuthTransactionStore, pepper: bytes, redirects: OAuthRedirectPolicy, lifetime: timedelta = datetime.timedelta(seconds=600), entropy: Callable[[int], bytes] | None = None) → None

class litestar_security.providers.oauth.OAuthTransactionStart(state: SecretStr, browser_binding: SecretStr, pkce_challenge: str, nonce: SecretStr | None, transaction: OAuthTransaction)

Bases: object

Fresh browser-facing material plus its server-side transaction.

__init__(state: SecretStr, browser_binding: SecretStr, pkce_challenge: str, nonce: SecretStr | None, transaction: OAuthTransaction) → None

class litestar_security.providers.oauth.OAuthTransactionStore(*args, **kwargs)

Bases: Protocol

Persist transactions and consume a matching transaction atomically.

Implementations must protect the recoverable PKCE verifier and nonce at rest. consume() must perform matching and deletion as one atomic operation so no two callbacks can receive the same transaction.

async create(transaction: OAuthTransaction) → None

Persist one new transaction.

Parameters:

transaction – The validated server-side transaction.

async consume(*, state_digest: bytes, binding_digest: bytes, provider: str, now: datetime) → OAuthTransaction | None

Atomically return and remove one exact, unexpired match.

Parameters:

• state_digest – The state lookup digest.

• binding_digest – The dedicated browser-cookie digest.

• provider – The provider route receiving the callback.

• now – The authoritative callback time.

Returns:

The one consumed transaction, or None for every lookup miss.

__init__(*args, **kwargs)

exception litestar_security.providers.oauth.OAuthTransactionUnavailable

Bases: RuntimeError

Indicate that transaction persistence or protection is unavailable.

__init__() → None

Initialize a stable secret-free failure.

class litestar_security.providers.oauth.OIDCBackchannelLogout(*, logout_token: str)

Bases: WireStruct

OIDC back-channel logout token form, decoded from a form-encoded body.

class litestar_security.providers.oauth.OIDCLogoutIdentity(provider: str, issuer: str, subject: str | None, session_id: str | None, token_id: str, expires_at: datetime)

Bases: object

Verified logout-token identity whose jti awaits store consumption.

__init__(provider: str, issuer: str, subject: str | None, session_id: str | None, token_id: str, expires_at: datetime) → None

class litestar_security.providers.oauth.OIDCLogoutLifecycleService(*, provider_issuers: ~collections.abc.Mapping[str, str], consumer: ~litestar_security.providers.oauth._routes.OIDCLogoutTokenConsumer, sessions: ~litestar_security.providers.oauth._routes.OIDCSessionLogoutStore, rate_limits: RateLimitGuard | None = None, client_key: ~collections.abc.Callable[[~litestar.connection.request.Request[~typing.Any, ~typing.Any, ~typing.Any]], str | None] | None = None, clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function OIDCLogoutLifecycleService.<lambda>>)

Bases: object

Concrete verified OIDC front- and back-channel local logout workflow.

__init__(*, provider_issuers: ~collections.abc.Mapping[str, str], consumer: ~litestar_security.providers.oauth._routes.OIDCLogoutTokenConsumer, sessions: ~litestar_security.providers.oauth._routes.OIDCSessionLogoutStore, rate_limits: RateLimitGuard | None = None, client_key: ~collections.abc.Callable[[~litestar.connection.request.Request[~typing.Any, ~typing.Any, ~typing.Any]], str | None] | None = None, clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function OIDCLogoutLifecycleService.<lambda>>) → None

Build one fixed-issuer logout service.

Parameters:

• provider_issuers – Exact configured issuer per provider.

• consumer – Logout-token verifier that yields a verified identity and jti.

• sessions – Store that atomically consumes that jti and revokes mapped sessions.

• rate_limits – Optional budget consumed by each front-channel attempt.

• client_key – Trusted client identity extractor for the rate-limit client bucket, defaulting to the peer address without trusting any forwarding header.

• clock – Source of the current aware time.

property provider_names: frozenset[str]

Return providers supporting OIDC logout.

async backchannel(provider: str, logout_token: str) → OAuthOperationSummary

Verify a logout token, check its issuer, then consume and revoke through the store.

async frontchannel(provider: str, issuer: str, session_id: str, *, request: Request[Any, Any, Any]) → OAuthOperationSummary

Revoke one exact provider-session mapping the caller’s binding owns.

Parameters:

• provider – Configured provider route segment.

• issuer – The iss query value, which must equal the configured issuer.

• session_id – The sid query value naming the provider session.

• request – The request whose browser-binding cookie proves ownership.

Returns:

The revoked-session count response.

Raises:

• NotAuthorizedException – If the issuer, session id, binding, ownership, or replay marker is rejected. Every refusal shares one shape.

• TooManyRequestsException – If the attempt exhausted its budget. The budget is consumed before any validation, so a rejected sid pays exactly as much as a revoking one.

• ServiceUnavailableException – If the session store or the limiter is unavailable. An outage never removes the limit or the binding.

class litestar_security.providers.oauth.OIDCLogoutTokenConsumer(*args, **kwargs)

Bases: Protocol

Verify logout-token signature, claims, and events, yielding its jti.

async consume(provider: str, logout_token: str, *, now: datetime) → OIDCLogoutIdentity

Return one verified logout identity without consuming its jti.

__init__(*args, **kwargs)

class litestar_security.providers.oauth.OIDCReauthenticationPolicy(max_age: int = 0, acr_values: frozenset[str] = frozenset({}), amr_values: frozenset[str] = frozenset({}))

Bases: object

Provider freshness and assurance requirements for one local purpose.

__init__(max_age: int = 0, acr_values: frozenset[str] = frozenset({}), amr_values: frozenset[str] = frozenset({})) → None

class litestar_security.providers.oauth.OIDCSessionLogoutStore(*args, **kwargs)

Bases: Protocol

Atomically consume a verified logout jti and revoke mapped sessions.

async consume_backchannel(identity: OIDCLogoutIdentity, *, now: datetime) → int | None

Consume jti and revoke sessions atomically, returning none on replay.

async revoke_frontchannel(provider: str, issuer: str, session_id: str, *, binding: str, now: datetime) → int | None

Atomically consume the one-shot front-channel marker and revoke owned sessions.

An implementation must revoke only the sessions that the presented browser binding owns for the exact (provider, issuer, session_id) tuple, and must consume the replay marker in the same operation, so a repeated or unowned request observes None instead of a second revocation.

Parameters:

• provider – Configured provider name.

• issuer – The already-validated configured issuer.

• session_id – The provider session identifier being revoked.

• binding – The browser-binding value presented by the caller.

• now – The aware revocation time.

Returns:

The revoked-session count, or None for a replayed or unowned request.

__init__(*args, **kwargs)

class litestar_security.providers.oauth.ProtectedOAuthSecret(ciphertext: bytes, key_version: str)

Bases: object

Opaque application-protected OAuth transaction secret.

__init__(ciphertext: bytes, key_version: str) → None

class litestar_security.providers.oauth.ProtectedResourceConfig(resource: str, authorization_servers: Sequence[str] = (), scopes_supported: Sequence[str] = (), bearer_methods_supported: Sequence[str] = ('header',), resource_documentation: str | None = None, route_prefix: str = '', advertise_resource_metadata: bool = True, cache_max_age: int = 300)

Bases: object

Describe this application as an OAuth 2.1 protected resource.

The values become the RFC 9728 metadata document served from /.well-known/oauth-protected-resource. Every member is validated at construction so an invalid advertisement fails at startup rather than reaching a client that trusts it.

Parameters:

• resource – The resource identifier, an absolute URI carrying no query and no fragment.

• authorization_servers – Issuer identifiers of the authorization servers able to issue tokens for this resource.

• scopes_supported – Scope tokens this resource understands.

• bearer_methods_supported – How a bearer token may be presented; one or more of header, body, and query.

• resource_documentation – An absolute URI where developer documentation for this resource is published.

• route_prefix – Where the metadata route is mounted. The default empty value mounts it at the application root, which is where RFC 9728 requires it to be reachable; a non-empty value must be an absolute non-root path.

• advertise_resource_metadata – Whether bearer authentication failures advertise this document through the RFC 9728 resource_metadata challenge parameter.

• cache_max_age – Seconds a client may cache the document, at most a day.

__init__(resource: str, authorization_servers: Sequence[str] = (), scopes_supported: Sequence[str] = (), bearer_methods_supported: Sequence[str] = ('header',), resource_documentation: str | None = None, route_prefix: str = '', advertise_resource_metadata: bool = True, cache_max_age: int = 300) → None

class litestar_security.providers.oauth.ProtectedResourceMetadata

Bases: _RequiredProtectedResourceMetadata

The RFC 9728 protected-resource metadata document.

Member names are defined by the specification, not by this application’s wire conventions, and an absent optional member is omitted rather than emitted empty.

class litestar_security.providers.oauth.ProviderGrant(scopes: frozenset[str], expires_at: datetime)

Bases: object

Secret-free provider scope grant projection.

__init__(scopes: frozenset[str], expires_at: datetime) → None

class litestar_security.providers.oauth.ProviderIdentity(provider: str, issuer: str, subject: str, display_name: str | None, email: str | None, email_verified: bool, raw_claims: Mapping[str, object], acr: str | None = None, amr: tuple[str, ...] = (), authenticated_at: datetime | None = None)

Bases: object

Verified immutable provider identity.

__init__(provider: str, issuer: str, subject: str, display_name: str | None, email: str | None, email_verified: bool, raw_claims: Mapping[str, object], acr: str | None = None, amr: tuple[str, ...] = (), authenticated_at: datetime | None = None) → None

class litestar_security.providers.oauth.ProviderTokenReference(provider_account_id: str, version: int, scopes: frozenset[str], expires_at: datetime)

Bases: object

Secret-free optimistic token-vault reference.

__init__(provider_account_id: str, version: int, scopes: frozenset[str], expires_at: datetime) → None

class litestar_security.providers.oauth.ProviderTokenSet(access_token: SecretStr, token_type: str, scopes: frozenset[str], expires_at: datetime, refresh_token: SecretStr | None = None, id_token: SecretStr | None = None)

Bases: object

Validated provider credentials kept out of normal representations.

__init__(access_token: SecretStr, token_type: str, scopes: frozenset[str], expires_at: datetime, refresh_token: SecretStr | None = None, id_token: SecretStr | None = None) → None

class litestar_security.providers.oauth.SecretStr(_value: str)

Bases: object

A string whose normal representations never reveal its value.

get_secret_value() → str

Return the secret to the narrow protocol boundary that needs it.

Returns:

The original secret string.

__init__(_value: str) → None

class litestar_security.providers.oauth.StepUpOAuthAuthorizer(service: StepUpService, current_epoch: Callable[[str], Awaitable[int | None]], transport_binding: Callable[[Request[Any, Any, Any]], bytes | None], session_binding: Callable[[Request[Any, Any, Any]], str | None])

Bases: object

Adapt StepUpService grants to OAuth lifecycle authorization.

async authorize(*, grant: str, account_id: str, purpose: str, request: Request[Any, Any, Any]) → OAuthStepUpAuthorization

Consume one exact step-up grant for the current OAuth operation.

Parameters:

• grant – One-time step-up grant presented by the authenticated account.

• account_id – Account the grant must belong to.

• purpose – Exact OAuth operation the grant authorizes.

• request – Request from which application callbacks derive bindings.

Returns:

The current epoch and optional callback-session binding.

Raises:

• NotAuthorizedException – If the grant or transport binding is absent or invalid.

• ServiceUnavailableException – If the epoch or step-up service is unavailable.

async current_security_epoch(account_id: str) → int

Return the application callback’s current valid epoch.

Parameters:

account_id – Account whose security epoch must be read.

Returns:

The current valid non-negative security epoch.

Raises:

ServiceUnavailableException – If the callback fails or returns no valid epoch.

async issue(*, account_id: str, purpose: str, authenticated_at: datetime, acr: str | None, amr: tuple[str, ...], request: Request[Any, Any, Any]) → StepUpCredential

Issue one transport-bound grant from verified OIDC freshness evidence.

__init__(service: StepUpService, current_epoch: Callable[[str], Awaitable[int | None]], transport_binding: Callable[[Request[Any, Any, Any]], bytes | None], session_binding: Callable[[Request[Any, Any, Any]], str | None]) → None

class litestar_security.providers.oauth.StoredProviderTokens(reference: ProviderTokenReference, tokens: ProviderTokenSet)

Bases: object

Versioned decrypted tokens returned only to the refresh service.

__init__(reference: ProviderTokenReference, tokens: ProviderTokenSet) → None

class litestar_security.providers.oauth.UnlinkOutcome(status: UnlinkStatus, provider_account_id: str | None = None)

Bases: object

Outcome of one atomic identity, login-method, and grant removal.

__init__(status: UnlinkStatus, provider_account_id: str | None = None) → None

class litestar_security.providers.oauth.UnlinkStatus(*values)

Bases: str, Enum

Atomic provider unlink outcomes.

litestar_security.providers.oauth.build_oauth_routes(config: OAuthConfig, wire: WirePolicy | None = None) → Router

Build native generated OAuth lifecycle routes.

Parameters:

• config – Validated provider route configuration.

• wire – How the request and response bodies are spelled. Defaults to the field names as Python spells them, with unknown members rejected.

Returns:

One no-store router.

litestar_security.providers.oauth.build_protected_resource_handler(config: ProtectedResourceConfig) → HTTPRouteHandler

Build one native public Litestar handler for the RFC 9728 metadata document.

Parameters:

config – The advertised values and their precomputed canonical response.

Returns:

A public handler serving the metadata with a stable ETag and cache headers.

litestar_security.providers.oauth.oauth_binding_cookie(binding: SecretStr | str, *, max_age: int = 600) → Cookie

Build the dedicated host-only OAuth browser-binding cookie.

Parameters:

• binding – The fresh browser-binding value.

• max_age – The bounded cookie lifetime in seconds.

Returns:

A secure native Litestar cookie value.

Raises:

ValueError – If the cookie value or lifetime is invalid.

litestar_security.providers.oauth.pkce_s256(verifier: SecretStr | str) → str

Build an RFC 7636 S256 challenge from one strict verifier.

Parameters:

verifier – A 43-128 character PKCE verifier.

Returns:

The unpadded base64url SHA-256 challenge.

Raises:

ValueError – If the verifier is not canonical PKCE material.

WebSocket-specific transport policy.

Content Security Policy connect-src is complementary browser hardening. It does not replace exact server-side Origin validation or credential policy.

class litestar_security.websocket.AuthorizationSnapshotRefresher(*args, **kwargs)

Bases: Protocol[UserT]

Application hook returning one detached immutable authorization snapshot.

async refresh(*, principal: Principal[UserT], previous: AuthorizationSnapshot, route_name: str) → AuthorizationSnapshot

Resolve and return a new detached authorization snapshot.

Parameters:

• principal – The authenticated principal for the connection.

• previous – The prior immutable snapshot, which is never mutated.

• route_name – The bound application route name.

Returns:

A new detached AuthorizationSnapshot; any other runtime type is treated as unavailable by connection lifetime supervision.

Raises:

Exception – When refresh fails. Connection lifetime supervision treats this as unavailable and closes the connection.

__init__(*args, **kwargs)

class litestar_security.websocket.InMemoryWebSocketConnectTokenStore

Bases: object

Deterministic concurrency-safe connect token store for tests and examples.

property records: tuple[WebSocketConnectAuthorization, ...]

Return a stable snapshot of digest-only records.

async create(record: WebSocketConnectAuthorization) → None

Persist one record while rejecting duplicate public IDs.

async consume(*, connect_token_id: str, digest: bytes, now: datetime) → WebSocketConnectAuthorization | None

Atomically return and delete one matching unexpired record.

__init__() → None

class litestar_security.websocket.IssuedWebSocketConnectToken(value: str, expires_at: datetime)

Bases: object

Reveal-once WebSocket connect token value.

__init__(value: str, expires_at: datetime) → None

class litestar_security.websocket.WebSocketBinding(connection_id: str, subject_id: str, credential_ids: frozenset[str], session_id: str | None, route_name: str)

Bases: object

Secret-free identity and route binding supplied to revocation hooks.

__init__(connection_id: str, subject_id: str, credential_ids: frozenset[str], session_id: str | None, route_name: str) → None

class litestar_security.websocket.WebSocketCloseCodes(unauthenticated: int = 4401, unauthorized: int = 4403, verification_unavailable: int = 1013)

Bases: object

Map stable security outcomes to WebSocket close codes.

__init__(unauthenticated: int = 4401, unauthorized: int = 4403, verification_unavailable: int = 1013) → None

class litestar_security.websocket.WebSocketConnectAuthorization(connect_token_id: str, digest: bytes, subject_id: str, security_epoch: int, route_name: str, origin: str, restrictions: CredentialRestrictions, policy_fingerprint: str, issued_at: datetime, expires_at: datetime)

Bases: object

Storage-safe one-time connect token binding containing no recoverable value.

__init__(connect_token_id: str, digest: bytes, subject_id: str, security_epoch: int, route_name: str, origin: str, restrictions: CredentialRestrictions, policy_fingerprint: str, issued_at: datetime, expires_at: datetime) → None

class litestar_security.websocket.WebSocketConnectTokenIssuer(app: Litestar, store: WebSocketConnectTokenStore, clock: Callable[[], ~datetime.datetime]=<function WebSocketConnectTokenIssuer.<lambda>>, ttl: timedelta = datetime.timedelta(seconds=30))

Bases: object

Mint one-time WebSocket connect tokens by route name.

async issue(route_name: str, *, principal: Principal[Any], context: SecurityContext, origin: str, security_epoch: int, restrictions: CredentialRestrictions | None = None, ttl: timedelta | None = None) → IssuedWebSocketConnectToken

Resolve one route name to its compiled plan and mint a connect token.

Parameters:

• route_name – The registered Litestar route handler name.

• principal – The authenticated principal minting the connect token.

• context – The current request’s security context.

• origin – The exact canonical Origin the connect token is bound to.

• security_epoch – The authoritative non-negative epoch bound to the token.

• restrictions – Optional narrowed authorization restrictions.

• ttl – Optional override for the configured connect token lifetime.

Returns:

The reveal-once issued connect token.

Raises:

ImproperlyConfiguredException – If the route name does not resolve to a registered WebSocket handler with a compiled runtime plan.

__init__(app: Litestar, store: WebSocketConnectTokenStore, clock: Callable[[], ~datetime.datetime]=<function WebSocketConnectTokenIssuer.<lambda>>, ttl: timedelta = datetime.timedelta(seconds=30)) → None

class litestar_security.websocket.WebSocketConnectTokenService(store: ~litestar_security.websocket._connect_tokens.WebSocketConnectTokenStore, ttl: ~datetime.timedelta = datetime.timedelta(seconds=30), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function WebSocketConnectTokenService.<lambda>>, entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>)

Bases: object

Issue and atomically consume exact one-handshake connect token bindings.

async issue(*, principal: Principal[Any], context: SecurityContext, route_name: str, origin: str, policy_fingerprint: str, security_epoch: int, restrictions: CredentialRestrictions | None = None) → IssuedWebSocketConnectToken

Issue one digest-only, exact-route connect token for an authenticated context.

async consume(value: object, *, route_name: str, origin: str, policy_fingerprint: str, current_security_epoch: Callable[[str], Awaitable[int | None]]) → WebSocketConnectAuthorization | None

Atomically consume a connect token before authoritative epoch and route checks.

__init__(store: ~litestar_security.websocket._connect_tokens.WebSocketConnectTokenStore, ttl: ~datetime.timedelta = datetime.timedelta(seconds=30), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function WebSocketConnectTokenService.<lambda>>, entropy: ~collections.abc.Callable[[int], bytes] = <function token_bytes>) → None

class litestar_security.websocket.WebSocketConnectTokenStore(*args, **kwargs)

Bases: Protocol

Application-owned atomic persistence port for one-time connect tokens.

async create(record: WebSocketConnectAuthorization) → None

Persist one new digest-only record, rejecting duplicate IDs.

async consume(*, connect_token_id: str, digest: bytes, now: datetime) → WebSocketConnectAuthorization | None

Atomically return and delete one matching unexpired record.

__init__(*args, **kwargs)

class litestar_security.websocket.WebSocketHandshake(origin: str | None, uses_cookie_credentials: bool, uses_authorization_header: bool, connect_token: str | None)

Bases: object

Describe credential transports presented by one WebSocket handshake.

__init__(origin: str | None, uses_cookie_credentials: bool, uses_authorization_header: bool, connect_token: str | None) → None

class litestar_security.websocket.WebSocketRevocationSource(*args, **kwargs)

Bases: Protocol

Event-driven, secret-free application hook for one binding’s revocation.

async wait(binding: WebSocketBinding) → None

Block without polling until the supplied connection binding is revoked.

Parameters:

binding – The secret-free identity and route binding to supervise.

Returns:

None only after a genuine revocation of binding.

Raises:

Exception – When supervision fails. The connection lifetime treats this as unavailable and closes the connection.

__init__(*args, **kwargs)

class litestar_security.websocket.WebSocketSecurityConfig(allowed_origins: frozenset[str] = frozenset({}), connect_token_store: ~litestar_security.websocket._connect_tokens.WebSocketConnectTokenStore | None = None, connect_token_ttl: ~datetime.timedelta = datetime.timedelta(seconds=30), maximum_connect_token_ttl: ~datetime.timedelta = datetime.timedelta(seconds=120), connect_token_query_parameter: str = 'connect_token', current_security_epoch: ~collections.abc.Callable[[str], ~collections.abc.Awaitable[int | None]] | None = None, refresh_interval: ~datetime.timedelta | None = None, snapshot_refresher: ~litestar_security.websocket._bindings.AuthorizationSnapshotRefresher[~typing.Any] | None = None, revocation_source: ~litestar_security.websocket._bindings.WebSocketRevocationSource | None = None, close_codes: ~litestar_security.websocket._config.WebSocketCloseCodes = WebSocketCloseCodes(unauthenticated=4401, unauthorized=4403, verification_unavailable=1013), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function WebSocketSecurityConfig.<lambda>>, sleeper: ~collections.abc.Callable[[float], ~collections.abc.Awaitable[None]] = <function sleep>)

Bases: object

Configure WebSocket transport validation and optional lifetime hooks.

__init__(allowed_origins: frozenset[str] = frozenset({}), connect_token_store: ~litestar_security.websocket._connect_tokens.WebSocketConnectTokenStore | None = None, connect_token_ttl: ~datetime.timedelta = datetime.timedelta(seconds=30), maximum_connect_token_ttl: ~datetime.timedelta = datetime.timedelta(seconds=120), connect_token_query_parameter: str = 'connect_token', current_security_epoch: ~collections.abc.Callable[[str], ~collections.abc.Awaitable[int | None]] | None = None, refresh_interval: ~datetime.timedelta | None = None, snapshot_refresher: ~litestar_security.websocket._bindings.AuthorizationSnapshotRefresher[~typing.Any] | None = None, revocation_source: ~litestar_security.websocket._bindings.WebSocketRevocationSource | None = None, close_codes: ~litestar_security.websocket._config.WebSocketCloseCodes = WebSocketCloseCodes(unauthenticated=4401, unauthorized=4403, verification_unavailable=1013), clock: ~collections.abc.Callable[[], ~datetime.datetime] = <function WebSocketSecurityConfig.<lambda>>, sleeper: ~collections.abc.Callable[[float], ~collections.abc.Awaitable[None]] = <function sleep>) → None

litestar_security.websocket.extract_websocket_handshake(connection: ASGIConnection[Any, Any, Any, Any], *, config: WebSocketSecurityConfig, uses_cookie_credentials: bool) → WebSocketHandshake

Extract and validate one WebSocket handshake without verifying credentials.

The caller derives uses_cookie_credentials from the existing common credential-slot extraction. Reusable header and cookie credentials remain owned by those common parsers; this function only applies WebSocket Origin and URL constraints.

Parameters:

• connection – The incoming Litestar WebSocket connection.

• config – Validated WebSocket security configuration.

• uses_cookie_credentials – Whether a common credential slot found a cookie- or session-backed credential.

Returns:

A redacted description of the presented WebSocket transports.

Raises:

WebSocketException – If Origin policy fails or a reusable URL credential is presented.

async litestar_security.websocket.issue_websocket_connect_token(*, principal: Principal[Any], context: SecurityContext, route_name: str, origin: str, policy_fingerprint: str, security_epoch: int, restrictions: CredentialRestrictions, store: WebSocketConnectTokenStore, clock: Callable[[], datetime], ttl: timedelta = datetime.timedelta(seconds=30)) → IssuedWebSocketConnectToken

Issue one reveal-once WebSocket connect token through an application store.

Parameters:

• principal – The authenticated principal the connect token speaks for.

• context – The security context the connect token is bound to.

• route_name – The single route the connect token authorizes; it is valid nowhere else.

• origin – The exact origin the handshake must present.

• policy_fingerprint – The compiled policy binding the handshake revalidates.

• security_epoch – The authoritative non-negative epoch bound to the token.

• restrictions – The credential restrictions carried into the connection.

• store – The application store that persists the digest-only record.

• clock – The timezone-aware clock used for issuance and expiry.

• ttl – How long the connect token stays valid, bounded by the two-minute maximum.

Returns:

The issued connect token, whose reveal-once value is not recoverable from the stored record.

Raises:

ValueError – If the principal is unauthenticated, the context is not a SecurityContext, or any binding fails validation.

litestar_security.websocket.websocket_policy_fingerprint(plan: object) → str

Return a stable process-independent fingerprint for one compiled plan.

Parameters:

plan – The frozen compiled security plan.

Returns:

A hexadecimal SHA-256 fingerprint.