itsme00/full-stack-todo

1"""
2JWT Authentication Middleware for Midnight Genesis
3
4Implements JWT validation using jose library with BETTER_AUTH_SECRET.
5Enforces Constitution Principle II: Zero Trust Security Model.
6
7Owner: @fastapi-jwt-guardian
8"""
9
10from fastapi import Depends, HTTPException, status
11from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
12from jose import JWTError, jwt
13import os
14from dotenv import load_dotenv
15from pathlib import Path
16from typing import Dict
17
18# Load environment variables from backend/.env
19load_dotenv(Path(__file__).parent / ".env")
20
21# Security scheme setup
22security = HTTPBearer()
23
24# Load BETTER_AUTH_SECRET from environment
25# CRITICAL: This MUST match the secret in frontend/.env.local
26BETTER_AUTH_SECRET = os.getenv("BETTER_AUTH_SECRET")
27
28if not BETTER_AUTH_SECRET:
29    raise ValueError(
30        "BETTER_AUTH_SECRET not set in environment variables. "
31        "Please add it to backend/.env and ensure it matches frontend/.env.local"
32    )
33
34
35async def get_current_user(
36    credentials: HTTPAuthorizationCredentials = Depends(security),
37) -> Dict[str, int]:
38    """
39    JWT validation dependency for protected routes.
40
41    Validates JWT token from Authorization header and extracts user_id.
42
43    Args:
44        credentials: HTTPAuthorizationCredentials from request header
45
46    Returns:
47        Dict containing user_id extracted from JWT token
48
49    Raises:
50        HTTPException 401: If token is missing, invalid, or expired
51        HTTPException 401: If user_id (sub claim) is missing from token
52
53    Usage:
54        @router.get("/protected")
55        async def protected_route(current_user: dict = Depends(get_current_user)):
56            user_id = current_user["user_id"]
57            # ... use user_id for data isolation
58    """
59    token = credentials.credentials
60
61    try:
62        # Decode and validate JWT token
63        # Better Auth uses HS256 algorithm by default
64        payload = jwt.decode(
65            token,
66            BETTER_AUTH_SECRET,
67            algorithms=["HS256"],
68        )
69
70        # Extract user_id from 'sub' claim (subject)
71        # Better Auth stores user ID in the 'sub' claim
72        user_id = payload.get("sub")
73
74        if user_id is None:
75            raise HTTPException(
76                status_code=status.HTTP_401_UNAUTHORIZED,
77                detail="Invalid token: missing user ID",
78                headers={"WWW-Authenticate": "Bearer"},
79            )
80
81        # Return user_id as integer
82        # CRITICAL: This is the ONLY source of user_id for data isolation
83        # Never accept user_id from client input
84        return {"user_id": int(user_id)}
85
86    except JWTError as e:
87        # Token validation failed (invalid signature, expired, malformed, etc.)
88        raise HTTPException(
89            status_code=status.HTTP_401_UNAUTHORIZED,
90            detail=f"Could not validate credentials: {str(e)}",
91            headers={"WWW-Authenticate": "Bearer"},
92        )
93    except ValueError:
94        # user_id is not a valid integer
95        raise HTTPException(
96            status_code=status.HTTP_401_UNAUTHORIZED,
97            detail="Invalid token: user ID must be an integer",
98            headers={"WWW-Authenticate": "Bearer"},
99        )
100
101
102async def get_current_user_optional(
103    credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False)),
104) -> Dict[str, int] | None:
105    """
106    Optional JWT validation dependency for routes that work with or without auth.
107
108    Returns user data if token is valid, None if token is missing or invalid.
109
110    Returns:
111        Dict containing user_id if authenticated, None otherwise
112    """
113    if credentials is None:
114        return None
115
116    try:
117        return await get_current_user(credentials)
118    except HTTPException:
119        return None
120