feat: add auth0 api endpoints

Author: VictoriaTskhondia

backend/pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
     "fastapi-pagination>=0.12.34",
     "bcrypt==4.0.1",
     "google-genai>=1.5.0",
+    "itsdangerous (>=2.2.0,<3.0.0)",
 ]
 
 [tool.uv]

backend/src/auth/auth0_api.py

@@ -0,0 +1,85 @@
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from sqlmodel import Session
+from starlette.responses import RedirectResponse
+
+from src.core.db import get_db
+from src.dependencies.auth0 import (
+    get_auth0_service,
+    get_current_user,
+    get_current_user_claims
+)
+from src.services.auth0 import Auth0Service
+from src.users.auth0 import get_or_create_user_from_auth0
+from src.users.models import User
+from src.users.schemas import UserPublic
+
+router = APIRouter(prefix="/auth0", tags=["auth0"])
+
+
+@router.get("/login")
+async def login(
+    request: Request,
+    auth_service: Annotated[Auth0Service, Depends(get_auth0_service)]
+) -> RedirectResponse:
+    return await auth_service.login(request)
+
+
+@router.get("/callback")
+async def callback(
+    request: Request,
+    session: Annotated[Session, Depends(get_db)],
+    auth_service: Annotated[Auth0Service, Depends(get_auth0_service)]
+) -> RedirectResponse:
+    try:
+        # Exchange auth code for tokens
+        token_response = await auth_service.callback(request)
+        access_token = token_response.get("access_token")
+
+        # Store access token in session for later use
+        request.session["access_token"] = access_token
+
+        # Get user info from Auth0
+        user_info = await auth_service.get_user_info(access_token)
+
+        # Get or create user in our database
+        db_user = await get_or_create_user_from_auth0(session, user_info)
+
+        # Store user ID in session
+        request.session["user_id"] = str(db_user.id)
+
+        # Redirect to the frontend after successful authentication
+        return RedirectResponse(url="/")
+    except Exception as e:
+        # Log the error and redirect to error page
+        return RedirectResponse(url=f"/auth0/error?message={str(e)}")
+
+
+@router.get("/logout")
+async def logout(
+    auth_service: Annotated[Auth0Service, Depends(get_auth0_service)]
+) -> RedirectResponse:
+    return auth_service.logout()
+
+
+@router.get("/me", response_model=UserPublic)
+async def read_users_me(
+    current_user: Annotated[User, Depends(get_current_user)]
+) -> User:
+    return current_user
+
+
+@router.get("/validate")
+async def validate_token(
+    claims: Annotated[dict[str, Any], Depends(get_current_user_claims)]
+) -> dict[str, Any]:
+    return claims
+
+
+@router.get("/error")
+async def auth_error(message: str = "Authentication error"):
+    raise HTTPException(
+        status_code=401,
+        detail=message
+    )

backend/src/core/config.py

@@ -49,9 +49,10 @@ class Settings(BaseSettings):
     # Auth0 Configuration
     AUTH0_CLIENT_ID: str
     AUTH0_CLIENT_SECRET: str
-    AUTH0_ISSUER: str
+    AUTH0_DOMAIN: str
+    AUTH0_ISSUER: str = ""  # Default empty string to make it optional
     AUTH0_CALLBACK_URL: str
-    AUTH0_LOGOUT_URL: str
+    AUTH0_LOGOUT_URL: str = ""  # Default empty string to make it optional
     AUTH0_AUDIENCE: str
 
     # Session Configuration

backend/src/dependencies/auth0.py

@@ -0,0 +1,109 @@
+from typing import Annotated, Any
+
+from fastapi import Depends, HTTPException, Request
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlmodel import Session
+
+from src.core.db import get_db
+from src.services.auth0 import Auth0Service, UserInfo
+from src.users.auth0 import get_or_create_user_from_auth0, get_user_by_auth0_id
+from src.users.models import User
+
+from functools import lru_cache
+
+security = HTTPBearer()
+
+
+# Initialize Auth0Service
+# This is a singleton instance of Auth0Service
+# to be reused across requests.
+@lru_cache()
+def get_auth0_service() -> Auth0Service:
+    """
+    Provides a singleton Auth0Service instance using lru_cache.
+    This is the recommended way in FastAPI for services that should be reused.
+    """
+    return Auth0Service()
+
+
+async def get_token_from_header(
+    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)]
+) -> str:
+    """
+    Extract and return the JWT token from Authorization header.
+    """
+    return credentials.credentials
+
+
+async def get_current_user_claims(
+    token: Annotated[str, Depends(get_token_from_header)],
+    auth_service: Annotated[Auth0Service, Depends(get_auth0_service)]
+) -> dict[str, Any]:
+    """
+    Validate token and return the user claims.
+    """
+    claims = await auth_service.validate_token(token)
+
+    return claims
+
+
+async def get_current_user_info(
+    request: Request,
+    token: Annotated[str, Depends(get_token_from_header)],
+    claims: Annotated[dict[str, Any], Depends(get_current_user_claims)],
+    auth_service: Annotated[Auth0Service, Depends(get_auth0_service)]
+) -> UserInfo:
+    """
+    Get user information from Auth0.
+    """
+    # Access token should be available in the request's session 
+    # after the callback flow
+    access_token = request.session.get("access_token")
+    if not access_token:
+        # If not in session, use the token from header
+        access_token = token
+
+    return await auth_service.get_user_info(access_token)
+
+
+async def get_current_user(
+    request: Request,
+    session: Annotated[Session, Depends(get_db)],
+    user_info: Annotated[UserInfo, Depends(get_current_user_info)]
+) -> User:
+    """
+    Get the current user from the database.
+    """
+    try:
+        # Check if user_id is in session
+        user_id = request.session.get("user_id")
+        if user_id:
+            # Use the user ID from the session
+            user = session.get(User, user_id)
+            if user:
+                return user
+
+        # If no valid user in session, look up by Auth0 ID
+        auth0_id = user_info.get("sub")
+        if not auth0_id:
+            raise ValueError("Missing Auth0 user ID")
+
+        # Try to find by Auth0 ID
+        user = await get_user_by_auth0_id(session, auth0_id)
+        if user:
+            # Store user ID in session for future requests
+            request.session["user_id"] = str(user.id)
+            return user
+
+        # Create or link user if not found
+        user = await get_or_create_user_from_auth0(session, user_info)
+
+        # Store user ID in session for future requests
+        request.session["user_id"] = str(user.id)
+        return user
+
+    except Exception as e:
+        raise HTTPException(
+            status_code=401,
+            detail=f"User integration failed: {str(e)}"
+        )

backend/src/routers.py

@@ -1,13 +1,15 @@
 from fastapi import APIRouter
 
 from src.auth.api import router as auth_router
+from src.auth.auth0_api import router as auth0_router
 from src.flashcards.api import router as flashcards_router
 from src.stats.api import router as stats_router
 from src.users.api import router as user_router
 
 api_router = APIRouter()
 
 api_router.include_router(auth_router, tags=["login"])
+api_router.include_router(auth0_router)
 api_router.include_router(user_router, prefix="/users", tags=["users"])
 api_router.include_router(flashcards_router, tags=["flashcards"])
 api_router.include_router(stats_router, tags=["stats"])