""" Module to handle Auth0 API security settings and operations. """ import time from dataclasses import dataclass from functools import lru_cache from typing import Literal from urllib.parse import quote import httpx import jwt from fastapi import HTTPException, status from src.backend import logger from src.backend.constants import Auth0 as Auth0Constants from .types import Auth0Token logger = logger.new_logger(__name__) @dataclass class Auth0Settings: """Dataclass to hold Auth0 API security settings.""" domain: str audience: str issuer: str algorithms: str scope: str client_id: str # Client ID for the Management API client_id_spa: str # Client ID for the SPA frontend client_secret: str # Client Secret for the Management API grant_type: str redirect_uri: str class Auth0: """Auth0 client to handle authentication operations (e.g. fetching access tokens, verifying JWT tokens, etc.). as well as Management API operations (e.g. fetching user info). It uses token and verification caching for performance. """ _token_invalid_if_secs_left: int = 60 def __init__(self, settings: Auth0Settings | dict[str, str]): """Initialize the Auth0 client.""" logger.debug("Initializing Auth0 client.") logger.debug("Fetching Auth0 Client settings...") # Settings should not be mutated once set, so that we can # cache the results of the verify method. try: self._settings = ( Auth0Settings(**settings) if isinstance(settings, dict) else settings ) except TypeError as ex: raise ValueError("Invalid Auth0 settings provided.") from ex logger.debug("Auth0 client initialized.") self._jwks_client = None self._token: Auth0Token | None = None self._expires_at: int | None = None @property def settings(self) -> Auth0Settings: """Get the settings for the Auth0 client.""" return self._settings @property def token(self) -> Auth0Token | None: """Get the access token.""" return self._token @property def expires_at(self) -> int | None: """Get the expiration time of the access token.""" return self._expires_at @property def is_cached_token_valid(self) -> bool: """Check if the cached token is still valid.""" return self._is_cached_token_valid() @property def _current_timestamp(self) -> int: """Get the current timestamp in seconds.""" return int(time.time()) @property def _new_async_client(self) -> httpx.AsyncClient: """Create a new async HTTP client with retries.""" return httpx.AsyncClient( timeout=30, transport=httpx.AsyncHTTPTransport(retries=3), ) async def get_token(self) -> Auth0Token: """Get an access token from Auth0 using the client credentials. The returned token is cached for future use and will be used until it expires; then a new token will be fetched and cached. """ if self._is_cached_token_valid(): logger.debug("Using cached access token.") return self._token logger.debug("Fetching access token from Auth0...") settings = self._settings auth0_token_url = f"https://{settings.domain}/oauth/token" auth0_data = { Auth0Constants.CLIENT_ID: settings.client_id, Auth0Constants.CLIENT_SECRET: settings.client_secret, Auth0Constants.AUDIENCE: settings.audience, Auth0Constants.GRANT_TYPE: settings.grant_type, } async with self._new_async_client as client: response = await client.post( auth0_token_url, data=auth0_data, ) response.raise_for_status() payload = response.json() # Cache the token and its expiration time (minus 60 seconds to be safe) self._token = payload[Auth0Constants.ACCESS_TOKEN] self._expires_at = ( self._current_timestamp + payload[Auth0Constants.EXPIRES_IN] - 60 ) logger.debug("Access token fetched and cached.") return self._token @lru_cache(maxsize=128) def verify(self, token: Auth0Token) -> dict: """Verify the JWT token using Auth0 and return the payload, if valid. Otherwise, raise an exception. Args: token: The token to verify. """ logger.debug("Verifying new JWT token...") if not self._jwks_client: # Lazy load the JWKS client self._jwks_client = self._get_jwks_client() try: signing_key = self._jwks_client.get_signing_key_from_jwt(token).key except (jwt.exceptions.PyJWKClientError, jwt.exceptions.DecodeError) as ex: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN) from ex try: settings = self._settings payload = jwt.decode( token, signing_key, algorithms=settings.algorithms, audience=settings.audience, issuer=settings.issuer, ) except Exception as ex: logger.info(f"JWT token verification failed: {ex}") raise HTTPException(status_code=status.HTTP_403_FORBIDDEN) from ex logger.debug("New JWT token verified and is valid.") return payload async def get_users(self) -> list[dict]: """Get users information from Auth0 using the access token. https://auth0.com/docs/api/management/v2/users/get-users-by-id Returns: A list of dictionaries with user information. """ return await self._api_dispatch("users") async def get_user(self, user_id: str) -> dict: """Get user information from Auth0 using the access token. https://auth0.com/docs/api/management/v2/users/get-users-by-id Args: user_id: The user_id to fetch information for. The syntax is "auth0|", where is an alphanumeric string. This is also the sub claim in the JWT token. Returns: A dictionary with the user information. """ return await self._api_dispatch(f"users/{user_id}") def _get_jwks_client(self) -> jwt.PyJWKClient: """Get a client for the JSON Web Key Set (JWKS) endpoint.""" return jwt.PyJWKClient(f"https://{self._settings.domain}/.well-known/jwks.json") async def _api_dispatch( self, endpoint: str, method: Literal["get", "post"] = "get" ) -> dict | list[dict]: """Dispatch an API request to the Auth0 API. Args: endpoint: The endpoint to dispatch the request to. method: The HTTP method to use for the request. Returns: A dictionary with the response from the API. """ # Make sure we have a valid token before making the request if not self._is_cached_token_valid(): await self.get_token() async with self._new_async_client as client: method_func = getattr(client, method.lower()) response = await method_func( # Quote the endpoint to avoid issues with special characters # e.g. when fetching user info with a user_id that contains a pipe f"https://{self._settings.domain}/api/v2/{quote(endpoint)}", headers={"authorization": f"Bearer {self._token}"}, ) response.raise_for_status() return response.json() def _is_cached_token_valid(self) -> bool: """Check if the cached token is still valid.""" return bool( self._token and (self._expires_at or 0) > self._current_timestamp + self._token_invalid_if_secs_left )