Authentication (middleware, authenticator, backend)¶

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.
    """
    result = await self.authenticate_with_context(connection, user_manager)
    return None if result is None else result[0]

async def authenticate_with_context(
    self,
    connection: ASGIConnection[Any, Any, Any, Any],
    user_manager: UserManagerProtocol[UP, ID],
) -> tuple[UP, object] | None:
    """Resolve a user and request auth context from the current request.

    Returns:
        Authenticated user plus request auth context, or ``None`` when no valid token is present.
    """
    token = await self.transport.read_token(connection)
    if isinstance(self.strategy, ContextualStrategy):
        contextual_strategy = cast(
            "ContextualStrategy[UP, ID, _AuthenticationResultWithContext[UP]]",
            self.strategy,
        )
        result = await contextual_strategy.read_token_with_context(token, user_manager)
        if result is None:
            return None
        return result.user, result.context

    user = await self.strategy.read_token(token, user_manager)
    if user is None:
        return None
    return user, self.name

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)

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.

    Raises:
        ClientException: When token revocation cannot be recorded (for example, an
            in-memory denylist at capacity), surfaced as HTTP 503 with the library
            error ``code`` in the JSON body.
    """
    try:
        await self.strategy.destroy_token(token, user)
    except TokenError as exc:
        raise ClientException(
            status_code=503,
            detail=str(exc),
            extra={"code": exc.code},
        ) from exc
    response = self.transport.set_logout(Response(content=None))
    if isinstance(self.transport, CookieTransport):
        self.transport.clear_refresh_token(response)
    return response

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)

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)

async def authenticate(
    self,
    connection: ASGIConnection[Any, Any, Any, Any],
) -> tuple[UP | None, object | None]:
    """Return the first authenticated user and request auth context.

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

    return None, None

@override
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    """Buffer signed request bodies before authentication so signatures cover raw bytes."""
    buffered_signed_body = False
    if self._api_key_backend_present and _has_signed_api_key_authorization_header(scope.get("headers", [])):
        body, receive = await _buffer_body_for_signature(
            receive,
            max_body_bytes=self.api_key_signed_body_max_bytes,
            max_messages=self.api_key_signed_body_max_messages,
        )
        cast("dict[str, Any]", scope)[API_KEY_SIGNED_BODY_SCOPE_KEY] = body
        buffered_signed_body = True
    try:
        await super().__call__(scope, receive, send)
    finally:
        if buffered_signed_body:
            cast("dict[str, Any]", scope).pop(API_KEY_SIGNED_BODY_SCOPE_KEY, None)

@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.
    """
    set_scope_superuser_role_name(connection.scope, self)
    session = self.get_request_session(connection.app.state, connection.scope)
    authenticator = self.authenticator_factory(session)
    user, auth_context = await authenticator.authenticate(connection)

    if user is not None:
        await _record_successful_api_key_use_if_applicable(
            connection,
            authenticator=authenticator,
            auth_context=auth_context,
        )

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