Skip to content

Authentication (middleware, authenticator, backend)

litestar_auth.authentication

Authentication package.

AuthenticationBackend(*, name, transport, strategy)

Compose a transport and strategy into a reusable auth backend.

Store backend components used for auth flows.

Source code in litestar_auth/authentication/backend.py 
25
26
27
28
29
def __init__(self, *, name: str, transport: TransportProtocol, strategy: StrategyProtocol[UP, ID]) -> None:
    """Store backend components used for auth flows."""
    self.name = name
    self.transport = transport
    self.strategy = strategy

authenticate(connection, user_manager) async

Resolve a user from the current request via transport and strategy.

Returns:

Type Description
UP | None

Authenticated user or None when no valid token is present.
Source code in litestar_auth/authentication/backend.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def authenticate(
    self,
    connection: ASGIConnection[Any, Any, Any, Any],
    user_manager: UserManagerProtocol[UP, ID],
) -> UP | None:
    """Resolve a user from the current request via transport and strategy.

    Returns:
        Authenticated user or ``None`` when no valid token is present.
    """
    token = await self.transport.read_token(connection)
    return await self.strategy.read_token(token, user_manager)

login(user) async

Issue a token through the configured strategy and transport.

Returns:

Type Description
Response[Any]

Response mutated by the configured transport for login.
Source code in litestar_auth/authentication/backend.py 
39
40
41
42
43
44
45
46
async def login(self, user: UP) -> Response[Any]:
    """Issue a token through the configured strategy and transport.

    Returns:
        Response mutated by the configured transport for login.
    """
    token = await self.strategy.write_token(user)
    return self.transport.set_login_token(Response(content=None), token)

logout(user, token) async

Invalidate a token and clear transport-managed state.

When the transport is a :class:CookieTransport, the refresh-token cookie is also expired so the browser does not retain it after logout.

Returns:

Type Description
Response[Any]

Response mutated by the configured transport for logout.
Source code in litestar_auth/authentication/backend.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
async def logout(self, user: UP, token: str) -> Response[Any]:
    """Invalidate a token and clear transport-managed state.

    When the transport is a :class:`CookieTransport`, the refresh-token
    cookie is also expired so the browser does not retain it after logout.

    Returns:
        Response mutated by the configured transport for logout.
    """
    await self.strategy.destroy_token(token, user)
    response = self.transport.set_logout(Response(content=None))
    if isinstance(self.transport, CookieTransport):
        self.transport.clear_refresh_token(response)
    return response

terminate_session(connection, user) async

Terminate the current authenticated session for a connection.

This method orchestrates logout in one explicit place by reading the current transport token and delegating token invalidation plus transport cleanup to logout.

Returns:

Type Description
Response[Any]

Response mutated by the configured transport for logout.

Raises:

Type Description
NotAuthorizedException

If the current transport token is unavailable.
Source code in litestar_auth/authentication/backend.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def terminate_session(
    self,
    connection: ASGIConnection[Any, Any, Any, Any],
    user: UP,
) -> Response[Any]:
    """Terminate the current authenticated session for a connection.

    This method orchestrates logout in one explicit place by reading the
    current transport token and delegating token invalidation plus transport
    cleanup to ``logout``.

    Returns:
        Response mutated by the configured transport for logout.

    Raises:
        NotAuthorizedException: If the current transport token is unavailable.
    """
    read_token = self.transport.read_token
    if isinstance(self.transport, LogoutTokenReadable):
        read_token = self.transport.read_logout_token
    token = await read_token(connection)
    if token is None:
        msg = "Authentication credentials were not provided."
        raise NotAuthorizedException(detail=msg)
    await _invalidate_refresh_artifacts(self.strategy, user)
    return await self.logout(user, token)

with_session(session)

Return a backend whose strategy is rebound to the provided session when supported.

Source code in litestar_auth/authentication/backend.py 
31
32
33
34
35
36
37
def with_session[S](self, session: S) -> AuthenticationBackend[UP, ID]:
    """Return a backend whose strategy is rebound to the provided session when supported."""
    bound_strategy = _bind_strategy_session(self.strategy, session)
    if bound_strategy is self.strategy:
        return self

    return type(self)(name=self.name, transport=self.transport, strategy=bound_strategy)

Authenticator(backends, user_manager)

Try configured authentication backends in order.

Store backends and the user manager used for token resolution.

Source code in litestar_auth/authentication/authenticator.py 
19
20
21
22
23
24
25
26
def __init__(
    self,
    backends: list[AuthenticationBackend[UP, ID]],
    user_manager: UserManagerProtocol[UP, ID],
) -> None:
    """Store backends and the user manager used for token resolution."""
    self.backends = backends
    self.user_manager = user_manager

authenticate(connection) async

Return the first authenticated user and backend name.

Returns:

Type Description
UP | None

Tuple of authenticated user and backend name, or (None, None)
str | None

when no backend resolves the request.
Source code in litestar_auth/authentication/authenticator.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
async def authenticate(
    self,
    connection: ASGIConnection[Any, Any, Any, Any],
) -> tuple[UP | None, str | None]:
    """Return the first authenticated user and backend name.

    Returns:
        Tuple of authenticated user and backend name, or ``(None, None)``
        when no backend resolves the request.
    """
    for backend in self.backends:
        user = await backend.authenticate(connection, self.user_manager)
        if user is not None:
            return user, backend.name

    return None, None

LitestarAuthMiddleware(app, *, get_request_session, authenticator_factory, auth_cookie_names=frozenset(), exclude=None, exclude_from_auth_key='exclude_from_auth', exclude_http_methods=None, scopes=None)

Bases: AbstractAuthenticationMiddleware

Resolve request users through an authenticator built with the request-scoped DB session.

Initialize the middleware.

Parameters:

Name Type Description Default
app ASGIApp

ASGI app to wrap.

 required
get_request_session RequestSessionProvider

Returns the shared request AsyncSession (Advanced Alchemy provide_session semantics); must not close the session.

 required
authenticator_factory AuthenticatorFactory[UP, ID]

Factory that binds the request-local session into an authenticator.

 required
auth_cookie_names frozenset[bytes]

Cookie names that should count as auth credentials when present.

 frozenset()
exclude str | list[str] | None

Optional route patterns excluded from middleware processing.

 None
exclude_from_auth_key str

Route opt key used to bypass auth.

 'exclude_from_auth'
exclude_http_methods Sequence[Method] | None

Optional HTTP methods excluded from auth.

 None
scopes Scopes | None

Optional ASGI scope types handled by the middleware.

 None
Source code in litestar_auth/authentication/middleware.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(  # noqa: PLR0913
    self,
    app: ASGIApp,
    *,
    get_request_session: RequestSessionProvider,
    authenticator_factory: AuthenticatorFactory[UP, ID],
    auth_cookie_names: frozenset[bytes] = frozenset(),
    exclude: str | list[str] | None = None,
    exclude_from_auth_key: str = "exclude_from_auth",
    exclude_http_methods: Sequence[Method] | None = None,
    scopes: Scopes | None = None,
) -> None:
    """Initialize the middleware.

    Args:
        app: ASGI app to wrap.
        get_request_session: Returns the shared request ``AsyncSession`` (Advanced Alchemy
            ``provide_session`` semantics); must not close the session.
        authenticator_factory: Factory that binds the request-local session into an authenticator.
        auth_cookie_names: Cookie names that should count as auth credentials when present.
        exclude: Optional route patterns excluded from middleware processing.
        exclude_from_auth_key: Route opt key used to bypass auth.
        exclude_http_methods: Optional HTTP methods excluded from auth.
        scopes: Optional ASGI scope types handled by the middleware.
    """
    super().__init__(
        app=app,
        exclude=exclude,
        exclude_from_auth_key=exclude_from_auth_key,
        exclude_http_methods=exclude_http_methods,
        scopes=scopes,
    )
    self.get_request_session = get_request_session
    self.authenticator_factory = authenticator_factory
    self.auth_cookie_names = auth_cookie_names

authenticate_request(connection) async

Authenticate the request and return the resolved user or None.

Returns:

Type Description
AuthenticationResult

Authentication result containing the resolved user and backend name.
Source code in litestar_auth/authentication/middleware.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@override
async def authenticate_request(
    self,
    connection: ASGIConnection[Any, Any, Any, Any],
) -> AuthenticationResult:
    """Authenticate the request and return the resolved user or ``None``.

    Returns:
        Authentication result containing the resolved user and backend name.
    """
    session = self.get_request_session(connection.app.state, connection.scope)
    authenticator = self.authenticator_factory(session)
    user, backend_name = await authenticator.authenticate(connection)

    if user is None and _request_supplied_auth_credentials(connection, auth_cookie_names=self.auth_cookie_names):
        logger.warning("Authentication token validation failed", extra={"event": "token_validation_failed"})
    return AuthenticationResult(user=user, auth=backend_name)