Transports¶

@override
async def read_token(self, connection: ASGIConnection[Any, Any, Any, Any]) -> str | None:
    """Return an API-key token from ``Authorization`` or ``X-API-Key`` headers."""
    signed_request_scheme = read_signed_api_key_request(connection)
    if signed_request_scheme is not None:
        return signed_request_scheme

    bearer_token = self._read_bearer_token(connection)
    if bearer_token is not None:
        return bearer_token

    header_token = connection.headers.get(API_KEY_HEADER_NAME)
    if header_token is None:
        return None
    token = header_token.strip()
    if not token.startswith(f"{self.prefix}_"):
        return None
    return token

@override
def set_login_token(self, response: Response[Any], token: str) -> Response[Any]:
    """Reject login issuance because API keys are not login-flow tokens.

    Raises:
        TokenError: Always, because API keys are not issued by login.
    """
    del response, token
    msg = "API-key transport cannot issue login tokens."
    raise TokenError(msg)

@override
def set_logout(self, response: Response[Any]) -> Response[Any]:
    """Leave logout responses unchanged because API keys keep no client state.

    Returns:
        The original response.
    """
    return response

@override
async def read_token(self, connection: ASGIConnection[Any, Any, Any, Any]) -> str | None:
    """Return the bearer token from the Authorization header when present."""
    authorization = connection.headers.get("Authorization")
    if authorization is None:
        return None

    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != self.scheme:
        return None
    token = token.strip()
    if not token:
        return None
    return token

@override
def set_login_token(self, response: Response[Any], token: str) -> Response[Any]:
    """Store the issued bearer token in the response body.

    Returns:
        The mutated response.
    """
    response.content = {"access_token": token, "token_type": self.scheme}
    response.media_type = MediaType.JSON
    return response

@override
def set_logout(self, response: Response[Any]) -> Response[Any]:
    """Clear the response body because bearer transport keeps no client state.

    Returns:
        The mutated response.
    """
    response.content = None
    return response

def clear_refresh_token(self, response: Response[Any]) -> Response[Any]:
    """Expire the refresh-token cookie immediately.

    Returns:
        The mutated response.
    """
    return self._set_cookie(
        response,
        key=self.refresh_cookie_name,
        value="",
        max_age=0,
        httponly=True,
    )

async def read_logout_token(self, connection: ASGIConnection[Any, Any, Any, Any]) -> str | None:
    """Return the access-token cookie value to invalidate during logout.

    Logout token sourcing is explicit here: cookie logout invalidates the
    access-token cookie and does not read refresh-token cookies.
    """
    return await self.read_token(connection)

async def read_refresh_token(self, connection: ASGIConnection[Any, Any, Any, Any]) -> str | None:
    """Return the refresh token from the dedicated refresh-token cookie."""
    return connection.cookies.get(self.refresh_cookie_name)

@override
async def read_token(self, connection: ASGIConnection[Any, Any, Any, Any]) -> str | None:
    """Return the authentication token from the configured cookie."""
    return connection.cookies.get(self.cookie_name)

@override
def set_login_token(self, response: Response[Any], token: str) -> Response[Any]:
    r"""Persist the issued token in the configured cookie.

    Security:
        When this transport is used for browser-based authentication, you MUST
        pair it with an explicit CSRF protection mechanism (for example, a
        separate CSRF cookie and a required X-CSRF-Token header on state-changing
        requests). This is especially important when ``samesite=\"none\"`` is used
        for cross-site scenarios, because browsers will attach cookies
        automatically to cross-origin requests.

    Returns:
        The mutated response.
    """
    return self._set_cookie(
        response,
        key=self.cookie_name,
        value=token,
        max_age=self.max_age,
        httponly=self.httponly,
    )

@override
def set_logout(self, response: Response[Any]) -> Response[Any]:
    """Remove the access-token cookie by expiring it immediately.

    Note:
        This transport-level method clears only the access-token cookie.
        The refresh-token cookie is cleared by
        :meth:`AuthenticationBackend.logout`, which calls
        :meth:`clear_refresh_token` after this method.

    Returns:
        The mutated response.
    """
    return self._set_cookie(
        response,
        key=self.cookie_name,
        value="",
        max_age=0,
        httponly=self.httponly,
    )

def set_refresh_token(self, response: Response[Any], refresh_token: str) -> Response[Any]:
    """Persist a refresh token in a dedicated HttpOnly cookie.

    Note:
        This library intentionally treats refresh tokens as a separate artifact from the
        access-token cookie used for request authentication.

    Returns:
        The mutated response.
    """
    # Security: use dedicated refresh_max_age so the cookie outlives the access-token cookie.
    effective_max_age = self.refresh_max_age if self.refresh_max_age is not None else self.max_age
    return self._set_cookie(
        response,
        key=self.refresh_cookie_name,
        value=refresh_token,
        max_age=effective_max_age,
        httponly=True,
    )

@abstractmethod
async def read_token(self, connection: ASGIConnection[Any, Any, Any, Any]) -> str | None:
    """Extract an authentication token from the incoming connection."""

@abstractmethod
def set_login_token(self, response: Response[Any], token: str) -> Response[Any]:
    """Attach an authentication token to the login response."""

@abstractmethod
def set_logout(self, response: Response[Any]) -> Response[Any]:
    """Clear any transport-managed authentication state on logout."""