Function get_current_user

api/auth.py:38–106

A FastAPI dependency that validates the Authorization token, and returns an AuthenticatedUser object with the user's ID and pro status. Raises HTTPException if validation fails.

(request: Request)

# --- The Updated FastAPI Dependency ---
async def get_current_user(request: Request) -> AuthenticatedUser:
    """
    A FastAPI dependency that validates the Authorization token,
    and returns an AuthenticatedUser object with the user's ID and pro status.
    Raises HTTPException if validation fails.
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    auth_header = request.headers.get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        logger.warning("Authorization header is missing or does not start with Bearer")
        raise credentials_exception

    token = auth_header.split(" ")[1]

    try:
        signing_key = jwks_client.get_signing_key_from_jwt(token)

        payload = jwt.decode(
            token,
            signing_key.key,
            algorithms=ALGORITHMS,
            audience=API_AUDIENCE,
            issuer=ISSUER,
            options={"verify_exp": True}
        )
        
        user_id = payload.get("sub")
        if user_id is None:
            logger.warning("Token is valid but 'sub' (user ID) is missing.")
            raise credentials_exception

        # 1. Define the claim name for the whole metadata object
        app_metadata_claim = f"{CUSTOM_CLAIM_NAMESPACE}app_metadata"
        
        # 2. Get the metadata dictionary from the token (it will be None if not present)
        app_metadata = payload.get(app_metadata_claim) or {} # Use empty dict as fallback

        # 3. Get the specific values from the metadata dictionary
        is_pro_status = app_metadata.get("is_pro", False)
        is_max_status = app_metadata.get("is_max", False)
        is_plus_status = app_metadata.get("is_plus", False)

        # 4. Get email from namespaced claim (set by Auth0 Action)
        email_claim = f"{CUSTOM_CLAIM_NAMESPACE}email"
        user_email = payload.get(email_claim)

        logger.info(f"User is pro: {is_pro_status}, is max: {is_max_status}, is plus: {is_plus_status}")

        # Return the structured user data
        return AuthenticatedUser(
            id=user_id,
            email=user_email,
            is_pro=is_pro_status,