Authx#

def handle_errors(self, app: FastAPI) -> None:
    """Add the `FastAPI.exception_handlers` relative to AuthX exceptions.

    Args:
        app (FastAPI): the FastAPI application to handle errors for
    """
    self._set_app_exception_handler(app, exception=exceptions.JWTDecodeError, status_code=422, message=None)
    self._set_app_exception_handler(
        app,
        exception=exceptions.MissingTokenError,
        status_code=401,
        message=self.MSG_TokenError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.MissingCSRFTokenError,
        status_code=401,
        message=self.MSG_MissingCSRFTokenError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.TokenTypeError,
        status_code=401,
        message=self.MSG_TokenTypeError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.RevokedTokenError,
        status_code=401,
        message=self.MSG_RevokedTokenError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.TokenRequiredError,
        status_code=401,
        message=self.MSG_TokenRequiredError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.FreshTokenRequiredError,
        status_code=401,
        message=self.MSG_FreshTokenRequiredError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.AccessTokenRequiredError,
        status_code=401,
        message=self.MSG_AccessTokenRequiredError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.RefreshTokenRequiredError,
        status_code=401,
        message=self.MSG_RefreshTokenRequiredError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.CSRFError,
        status_code=401,
        message=self.MSG_CSRFError,
    )

def set_callback_get_model_instance(self, callback: ModelCallback[T]) -> None:
    """Set callback for model instance."""
    self.callback_get_model_instance = callback

def set_callback_token_blocklist(self, callback: TokenCallback) -> None:
    """Set callback for token."""
    self.callback_is_token_in_blocklist = callback

def set_subject_getter(self, callback: ModelCallback[T]) -> None:
    """Set the callback to run for subject retrieval and serialization."""
    self.set_callback_get_model_instance(callback)

def set_token_blocklist(self, callback: TokenCallback) -> None:
    """Set the callback to run for validation of revoked tokens."""
    self.set_callback_token_blocklist(callback)

def is_token_in_blocklist(self, token: Optional[str], **kwargs: ParamSpecKwargs) -> bool:
    """Check if token is in blocklist."""
    if self._check_token_callback_is_set(ignore_errors=True):
        callback: Optional[TokenCallback] = self.callback_is_token_in_blocklist
        if callback is not None and token is not None:
            return callback(token, **kwargs)  # type: ignore
    return False

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.

    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: _description_
    """
    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,
    )

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,
    *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): TODO. 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.

    Returns:
        str: Access Token
    """
    return self._create_token(
        uid=uid,
        type="access",
        fresh=fresh,
        headers=headers,
        expiry=expiry,
        data=data,
        audience=audience,
    )

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,
    *args: Any,
    **kwargs: Any,
) -> str:
    """Generate a Refresh Token.

    Args:
        uid (str): Unique identifier to generate token for
        headers (Optional[dict[str, Any]], optional): TODO. 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.

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

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
    """

    async def _auth_required(request: Request) -> Any:
        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

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.
    """
    token: TokenPayload = await self._auth_required(request=request)
    uid = token.sub
    return self._get_current_subject(uid=uid)

def get_token_from_request(
    self, type: TokenType = "access", optional: bool = True
) -> Callable[[Request], Awaitable[Optional[RequestToken]]]:
    """Return token from response if available.

    Args:
        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.

    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
    """

    async def _token_getter(request: Request) -> Optional[RequestToken]:
        # Specify locations as None since it's not provided in this context
        # Convert `optional` to a literal True or False based on its value
        optional_literal: Literal[True, False] = optional
        return await self._get_token_from_request(
            request,
            None,  # Explicitly passing None for locations
            refresh=(type == "refresh"),
            optional=optional_literal,
        )

    return _token_getter

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)
            if (
                datetime.timedelta(payload.time_until_expiry)  # type: ignore
                < 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