Authx#

def load_config(self, config: AuthXConfig) -> None:
    """Load and store the configuration for the authentication system.

    Sets the internal configuration object with the provided authentication configuration.

    Args:
        config: The configuration settings for the AuthX authentication system.

    Returns:
        None
    """
    self._config = config

async def get_access_token_from_request(
    self, request: Request, locations: Optional[TokenLocations] = None
) -> RequestToken:
    """Dependency to retrieve access token from request.

    Args:
        request (Request): Request to retrieve access token from
        locations (Optional[TokenLocations], optional): Locations to retrieve token from. Defaults to None.

    Raises:
        MissingTokenError: When no `access` token is available in request

    Returns:
        RequestToken: Request Token instance for `access` token type
    """
    return await self._get_token_from_request(request, optional=False, locations=locations)

async def get_refresh_token_from_request(
    self, request: Request, locations: Optional[TokenLocations] = None
) -> RequestToken:
    """Dependency to retrieve refresh token from request.

    Args:
        request (Request): Request to retrieve refresh token from
        locations (Optional[TokenLocations], optional): Locations to retrieve token from. Defaults to None.

    Raises:
        MissingTokenError: When no `refresh` token is available in request

    Returns:
        RequestToken: Request Token instance for `refresh` token type
    """
    return await self._get_token_from_request(request, refresh=True, optional=False, locations=locations)

def verify_token(
    self,
    token: RequestToken,
    verify_type: bool = True,
    verify_fresh: bool = False,
    verify_csrf: bool = True,
) -> TokenPayload:
    """Verify a request token.

    Attempts verification with the current key first, then falls back
    to the previous key if key rotation is configured.

    Args:
        token (RequestToken): RequestToken instance
        verify_type (bool, optional): Apply token type verification. Defaults to True.
        verify_fresh (bool, optional): Apply token freshness verification. Defaults to False.
        verify_csrf (bool, optional): Apply token CSRF verification. Defaults to True.

    Returns:
        TokenPayload: Verified token payload
    """
    try:
        try:
            return token.verify(
                key=self.config.public_key,
                algorithms=[self.config.JWT_ALGORITHM],
                verify_fresh=verify_fresh,
                verify_type=verify_type,
                verify_csrf=verify_csrf,
                audience=self.config.JWT_DECODE_AUDIENCE,
                issuer=self.config.JWT_DECODE_ISSUER,
            )
        except JWTDecodeError:
            previous_key = self.config.previous_public_key
            if previous_key is None:
                raise
            return token.verify(
                key=previous_key,
                algorithms=[self.config.JWT_ALGORITHM],
                verify_fresh=verify_fresh,
                verify_type=verify_type,
                verify_csrf=verify_csrf,
                audience=self.config.JWT_DECODE_AUDIENCE,
                issuer=self.config.JWT_DECODE_ISSUER,
            )
    except AuthXException as e:
        if e.login_type is None:
            e.login_type = self.login_type
        raise

def create_access_token(
    self,
    uid: str,
    fresh: bool = False,
    headers: Optional[dict[str, Any]] = None,
    expiry: Optional[DateTimeExpression] = None,
    data: Optional[dict[str, Any]] = None,
    audience: Optional[StringOrSequence] = None,
    scopes: Optional[list[str]] = None,
    *args: Any,
    **kwargs: Any,
) -> str:
    """Generate an Access Token.

    Args:
        uid (str): Unique identifier to generate token for
        fresh (bool, optional): Generate fresh token. Defaults to False.
        headers (Optional[dict[str, Any]], optional): Custom JWT headers. Defaults to None.
        expiry (Optional[DateTimeExpression], optional): Use a user defined expiry claim. Defaults to None.
        data (Optional[dict[str, Any]], optional): Additional data to store in token. Defaults to None.
        audience (Optional[StringOrSequence], optional): Audience claim. Defaults to None.
        scopes (Optional[list[str]], optional): List of scopes to include in the token. Defaults to None.

    Returns:
        str: Access Token

    Example:
        ```python
        # Token with scopes
        token = auth.create_access_token(
            uid="user123",
            scopes=["users:read", "posts:write"]
        )
        ```
    """
    return self._create_token(
        uid=uid,
        type="access",
        fresh=fresh,
        headers=headers,
        expiry=expiry,
        data=data,
        audience=audience,
        scopes=scopes,
    )

def create_refresh_token(
    self,
    uid: str,
    headers: Optional[dict[str, Any]] = None,
    expiry: Optional[DateTimeExpression] = None,
    data: Optional[dict[str, Any]] = None,
    audience: Optional[StringOrSequence] = None,
    scopes: Optional[list[str]] = None,
    *args: Any,
    **kwargs: Any,
) -> str:
    """Generate a Refresh Token.

    Args:
        uid (str): Unique identifier to generate token for
        headers (Optional[dict[str, Any]], optional): Custom JWT headers. Defaults to None.
        expiry (Optional[DateTimeExpression], optional): Use a user defined expiry claim. Defaults to None.
        data (Optional[dict[str, Any]], optional): Additional data to store in token. Defaults to None.
        audience (Optional[StringOrSequence], optional): Audience claim. Defaults to None.
        scopes (Optional[list[str]], optional): List of scopes to include in the token. Defaults to None.

    Returns:
        str: Refresh Token
    """
    return self._create_token(
        uid=uid,
        type="refresh",
        headers=headers,
        expiry=expiry,
        data=data,
        audience=audience,
        scopes=scopes,
    )

def create_token_pair(
    self,
    uid: str,
    fresh: bool = False,
    headers: Optional[dict[str, Any]] = None,
    access_expiry: Optional[DateTimeExpression] = None,
    refresh_expiry: Optional[DateTimeExpression] = None,
    data: Optional[dict[str, Any]] = None,
    audience: Optional[StringOrSequence] = None,
    access_scopes: Optional[list[str]] = None,
    refresh_scopes: Optional[list[str]] = None,
) -> TokenResponse:
    """Generate an access and refresh token pair.

    Convenience method that creates both tokens at once and returns them
    in a standardized ``TokenResponse`` model.

    Args:
        uid: Unique identifier of the user.
        fresh: Whether the access token should be marked as fresh. Defaults to False.
        headers: Optional custom JWT headers applied to both tokens.
        access_expiry: Optional expiry override for the access token.
        refresh_expiry: Optional expiry override for the refresh token.
        data: Optional additional data stored in both tokens.
        audience: Optional audience claim for both tokens.
        access_scopes: Optional scopes for the access token.
        refresh_scopes: Optional scopes for the refresh token.

    Returns:
        TokenResponse: A model containing ``access_token``, ``refresh_token``, and ``token_type``.

    Example:
        ```python
        tokens = auth.create_token_pair(uid="user123", fresh=True)
        return tokens  # {"access_token": "...", "refresh_token": "...", "token_type": "bearer"}
        ```
    """
    access_token = self.create_access_token(
        uid=uid,
        fresh=fresh,
        headers=headers,
        expiry=access_expiry,
        data=data,
        audience=audience,
        scopes=access_scopes,
    )
    refresh_token = self.create_refresh_token(
        uid=uid,
        headers=headers,
        expiry=refresh_expiry,
        data=data,
        audience=audience,
        scopes=refresh_scopes,
    )
    return TokenResponse(access_token=access_token, refresh_token=refresh_token)

def set_access_cookies(
    self,
    token: str,
    response: Response,
    max_age: Optional[int] = None,
) -> None:
    """Add 'Set-Cookie' for access token in response header.

    Args:
        token (str): Access token
        response (Response): response to set cookie on
        max_age (Optional[int], optional): Max Age cookie parameter. Defaults to None.
    """
    self._set_cookies(token=token, type="access", response=response, max_age=max_age)

def set_refresh_cookies(
    self,
    token: str,
    response: Response,
    max_age: Optional[int] = None,
) -> None:
    """Add 'Set-Cookie' for refresh token in response header.

    Args:
        token (str): Refresh token
        response (Response): response to set cookie on
        max_age (Optional[int], optional): Max Age cookie parameter. Defaults to None.
    """
    self._set_cookies(token=token, type="refresh", response=response, max_age=max_age)

def unset_access_cookies(
    self,
    response: Response,
) -> None:
    """Remove 'Set-Cookie' for access token in response header.

    Args:
        response (Response): response to remove cooke from
    """
    self._unset_cookies("access", response=response)

def unset_refresh_cookies(
    self,
    response: Response,
) -> None:
    """Remove 'Set-Cookie' for refresh token in response header.

    Args:
        response (Response): response to remove cooke from
    """
    self._unset_cookies("refresh", response=response)

def unset_cookies(
    self,
    response: Response,
) -> None:
    """Remove 'Set-Cookie' for tokens from response headers.

    Args:
        response (Response): response to remove token cookies from
    """
    self.unset_access_cookies(response)
    self.unset_refresh_cookies(response)

def get_dependency(self, request: Request, response: Response) -> AuthXDependency[Any]:
    """FastAPI Dependency to return a AuthX sub-object within the route context.

    Args:
        request (Request): Request context managed by FastAPI
        response (Response): Response context managed by FastAPI

    Note:
        The AuthXDeps is a utility class, to enable quick token operations
        within the route logic. It provides methods to avoid additional code
        in your route that would be outside of the route logic

        Such methods includes setting and unsetting cookies without the need
        to generate a response object beforehand

    Returns:
        AuthXDeps: The contextful AuthX object
    """
    return AuthXDependency(self, request=request, response=response)

def token_required(
    self,
    type: str = "access",
    verify_type: bool = True,
    verify_fresh: bool = False,
    verify_csrf: Optional[bool] = None,
    locations: Optional[TokenLocations] = None,
) -> Callable[[Request], Awaitable[TokenPayload]]:
    """Dependency to enforce valid token availability in request.

    Args:
        type (str, optional): Require a given token type. Defaults to "access".
        verify_type (bool, optional): Apply type verification. Defaults to True.
        verify_fresh (bool, optional): Require token freshness. Defaults to False.
        verify_csrf (Optional[bool], optional): Enable CSRF verification. Defaults to None.
        locations (Optional[TokenLocations], optional): Locations to retrieve token from. Defaults to None.

    Returns:
        Callable[[Request], TokenPayload]: Dependency for Valid token Payload retrieval
    """
    header_scheme, cookie_scheme, query_scheme = self._openapi_security_dependencies(
        type=type,
        locations=locations,
    )
    header_dependency = Depends(header_scheme)
    cookie_dependency = Depends(cookie_scheme)
    query_dependency = Depends(query_scheme)

    async def _auth_required(
        request: Request,
        _authx_openapi_header: Any = header_dependency,
        _authx_openapi_cookie: Any = cookie_dependency,
        _authx_openapi_query: Any = query_dependency,
    ) -> Any:
        self.ensure_request_exception_handlers(request)
        return await self._auth_required(
            request=request,
            type=type,
            verify_csrf=verify_csrf,
            verify_type=verify_type,
            verify_fresh=verify_fresh,
            locations=locations,
        )

    return _auth_required

def scopes_required(
    self,
    *scopes: str,
    all_required: bool = True,
    verify_type: bool = True,
    verify_fresh: bool = False,
    verify_csrf: Optional[bool] = None,
    locations: Optional[TokenLocations] = None,
) -> Callable[[Request], Awaitable[TokenPayload]]:
    """Dependency to enforce required scopes in token.

    Creates a FastAPI dependency that validates that the token contains
    the required scopes. Supports both simple and hierarchical scopes
    with wildcard matching (e.g., "admin:*" matches "admin:users").

    Args:
        *scopes: Variable number of scope strings required.
        all_required: If True (default), ALL scopes must be present (AND logic).
                     If False, at least ONE scope must be present (OR logic).
        verify_type: Apply token type verification. Defaults to True.
        verify_fresh: Require token freshness. Defaults to False.
        verify_csrf: Enable CSRF verification. Defaults to None (uses config).
        locations: Locations to retrieve token from. Defaults to None.

    Returns:
        Callable[[Request], Awaitable[TokenPayload]]: Dependency for scope validation.

    Raises:
        InsufficientScopeError: When token lacks required scopes.

    Example:
        ```python
        # Require single scope
        @app.get("/users", dependencies=[Depends(auth.scopes_required("users:read"))])
        async def list_users(): ...

        # Require multiple scopes (AND)
        @app.delete("/users/{id}", dependencies=[Depends(auth.scopes_required("users:read", "users:delete"))])
        async def delete_user(id: int): ...

        # Require any of the scopes (OR)
        @app.get("/admin", dependencies=[Depends(auth.scopes_required("admin", "superuser", all_required=False))])
        async def admin_panel(): ...

        # Wildcard scope
        @app.get("/admin/users", dependencies=[Depends(auth.scopes_required("admin:*"))])
        async def admin_users(): ...
        ```
    """
    required_scopes = list(scopes)
    header_scheme, cookie_scheme, query_scheme = self._openapi_security_dependencies(
        type="access",
        locations=locations,
    )
    header_dependency = Depends(header_scheme)
    cookie_dependency = Depends(cookie_scheme)
    query_dependency = Depends(query_scheme)

    async def _scopes_required(
        request: Request,
        _authx_openapi_header: Any = header_dependency,
        _authx_openapi_cookie: Any = cookie_dependency,
        _authx_openapi_query: Any = query_dependency,
    ) -> TokenPayload:
        self.ensure_request_exception_handlers(request)
        payload = await self._auth_required(
            request=request,
            type="access",
            verify_type=verify_type,
            verify_fresh=verify_fresh,
            verify_csrf=verify_csrf,
            locations=locations,
        )

        if not has_required_scopes(required_scopes, payload.scopes, all_required=all_required):
            raise InsufficientScopeError(
                required=required_scopes,
                provided=payload.scopes,
                login_type=self.login_type,
            )

        return payload

    return _scopes_required

async def get_current_subject(self, request: Request) -> Optional[T]:
    """Retrieve the currently authenticated subject from the request.

    Validates the request token and fetches the corresponding subject based on the user identifier.

    Args:
        request: The HTTP request containing authentication credentials.

    Returns:
        The authenticated subject if present, otherwise None.
    """
    self.ensure_request_exception_handlers(request)
    token: TokenPayload = await self._auth_required(request=request)
    uid = token.sub
    return await self._get_current_subject(uid=uid)

async def get_token_from_request(
    self,
    request: Request,
    type: TokenType = "access",
    optional: bool = True,
    locations: Optional[TokenLocations] = None,
) -> Optional[RequestToken]:
    """Retrieve token from request.

    Args:
        request (Request): The FastAPI request object.
        type (TokenType, optional): The type of token to retrieve from request.
            Defaults to "access".
        optional (bool, optional): Whether or not to enforce token presence in request.
            Defaults to True.
        locations (Optional[TokenLocations], optional): Locations to retrieve token from.
            Defaults to None (uses configured JWT_TOKEN_LOCATION).

    Note:
        When `optional=True`, the return value might be `None`
        if no token is available in request.

        When `optional=False`, raises a MissingTokenError.

    Returns:
        Optional[RequestToken]: The RequestToken if available, None if optional and not found.

    Example:
        ```python
        token = await auth.get_token_from_request(request)
        token = await auth.get_token_from_request(request, type="refresh")
        token = await auth.get_token_from_request(request, optional=False)
        ```
    """
    if optional:
        return await self._get_token_from_request(
            request,
            locations=locations,
            refresh=(type == "refresh"),
            optional=True,
        )
    else:
        return await self._get_token_from_request(
            request,
            locations=locations,
            refresh=(type == "refresh"),
            optional=False,
        )

async def implicit_refresh_middleware(
    self,
    request: Request,
    call_next: Callable[[Request], Coroutine[Any, Any, Response]],
) -> Response:
    """FastAPI Middleware to enable token refresh for an APIRouter.

    Args:
        request (Request): Incoming request
        call_next (Coroutine): Endpoint logic to be called

    Note:
        This middleware is only based on `access` tokens.
        Using implicit refresh mechanism makes use of `refresh`
        tokens unnecessary.

    Note:
        The refreshed `access` token will not be considered as
        `fresh`

    Note:
        The implicit refresh mechanism is only enabled
        for authorization through cookies.

    Returns:
        Response: Response with update access token cookie if relevant
    """
    response = await call_next(request)

    if self.config.has_location("cookies") and self._implicit_refresh_enabled_for_request(request):
        with contextlib.suppress(AuthXException):
            # Refresh mechanism
            token = await self._get_token_from_request(
                request=request,
                locations=["cookies"],
                refresh=False,
                optional=False,
            )
            payload = self.verify_token(token, verify_fresh=False, verify_csrf=False)
            if payload.time_until_expiry < self.config.JWT_IMPLICIT_REFRESH_DELTATIME:
                new_token = self.create_access_token(uid=payload.sub, fresh=False, data=payload.extra_dict)
                self.set_access_cookies(new_token, response=response)
    return response

def rate_limited(
    self,
    max_requests: int = 10,
    window: int = 60,
    key_func: Optional[Callable[[Request], str]] = None,
) -> Callable[[Request], Awaitable[TokenPayload]]:
    """Dependency combining rate limiting with access token verification.

    Args:
        max_requests: Maximum requests allowed within the window.
        window: Time window in seconds.
        key_func: Callable to extract rate limit key from request. Defaults to client IP.

    Returns:
        A FastAPI dependency that enforces both rate limiting and token auth.

    Example:
        ```python
        @app.get("/api", dependencies=[Depends(auth.rate_limited(max_requests=5, window=60))])
        async def api_route(): ...
        ```
    """
    limiter = RateLimiter(max_requests=max_requests, window=window, key_func=key_func)

    async def _rate_limited_auth(request: Request) -> TokenPayload:
        await limiter(request)
        return await self._auth_required(request=request)

    return _rate_limited_auth

def set_session_store(self, store: Any) -> None:
    """Register a session storage backend.

    Args:
        store: An object implementing the ``SessionStoreProtocol``.
    """
    self._session_store = store

async def create_session(
    self,
    uid: str,
    request: Optional[Request] = None,
    device_info: Optional[dict[str, Any]] = None,
) -> SessionInfo:
    """Create a new session and persist it via the session store.

    Args:
        uid: User identifier.
        request: Optional HTTP request for IP/User-Agent extraction.
        device_info: Optional additional device metadata.

    Returns:
        The created ``SessionInfo`` instance.
    """
    ip_address: Optional[str] = None
    user_agent: Optional[str] = None
    if request is not None:
        if request.client is not None:
            ip_address = request.client.host
        user_agent = request.headers.get("user-agent")

    session = SessionInfo(
        uid=uid,
        ip_address=ip_address,
        user_agent=user_agent,
        device_info=device_info,
    )

    if self._session_store is not None:
        await self._session_store.create(session)

    return session