""" JWT Authentication Module for HH Custom Tailor API. Handles: - JWT secret key management (env → file → ephemeral fallback) - Access token generation and verification (1 hour lifetime) - Refresh token generation and verification (30 days lifetime) - Refresh token DB storage (hashed) and revocation - FastAPI dependency for protected endpoints """ import os import logging import secrets from datetime import datetime, timedelta, timezone from hashlib import sha256 from pathlib import Path import jwt from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from db import sql_exe logger = logging.getLogger(__name__) # --- Configuration --- # Algorithm is hardcoded to prevent algorithm confusion attacks. # Never derive the algorithm from an unverified token. JWT_ALGORITHM = "HS256" # Token lifetimes (overridable via environment) ACCESS_TOKEN_EXPIRE_MINUTES = int( os.getenv("JWT_ACCESS_TOKEN_EXPIRE_MINUTES", "60") ) REFRESH_TOKEN_EXPIRE_DAYS = int( os.getenv("JWT_REFRESH_TOKEN_EXPIRE_DAYS", "30") ) # OAuth2 scheme — tells Swagger UI to show the "Authorize" button # and use Bearer token auth. The tokenUrl points to the login endpoint. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/token/") # --- Secret Key Management --- def _get_jwt_secret() -> str: """ Resolve JWT secret using a multi-tiered fallback: 1. Environment variable (preferred for production) 2. Local file (for persistent non-env deployments) 3. Ephemeral random secret (dev/testing only — logs a warning) WARNING: Changing the secret invalidates ALL existing tokens. All logged-in users will need to re-authenticate. """ # Tier 1: Environment variable env_secret = os.getenv("JWT_SECRET_KEY") if env_secret: return env_secret # Tier 2: Secret file secret_file = Path(__file__).parent / "jwt_secret.key" if secret_file.exists(): secret = secret_file.read_text().strip() if secret: return secret # Tier 3: Ephemeral (dev/testing only) logger.warning( "JWT_SECRET_KEY not found in environment or file. " "Generating ephemeral secret. This instance is ISOLATED — " "tokens will NOT be valid across restarts or multiple instances!" ) return secrets.token_hex(32) # Cache the secret at module load time for consistency within a process JWT_SECRET = _get_jwt_secret() # --- Token Creation --- def create_access_token(entity_id: int, email: str) -> str: """ Create a short-lived access token containing user identity claims. Args: entity_id: The customer entity ID (primary key) email: The customer email address Returns: Encoded JWT string """ now = datetime.now(timezone.utc) payload = { "sub": str(entity_id), "email": email, "type": "access", "iat": now, "exp": now + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES), } return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) def create_refresh_token(entity_id: int) -> str: """ Create a long-lived refresh token with minimal claims. The raw token is returned to the client; only its SHA-256 hash is stored in the database. Args: entity_id: The customer entity ID Returns: Encoded JWT string """ now = datetime.now(timezone.utc) # Include a unique jti (JWT ID) for revocation tracking jti = secrets.token_hex(16) payload = { "sub": str(entity_id), "type": "refresh", "jti": jti, "iat": now, "exp": now + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS), } return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) # --- Token Verification --- def _decode_token(token: str) -> dict: """ Decode and verify a JWT token. Enforces: - Hardcoded HS256 algorithm (rejects 'none' and others) - Expiry validation (exp claim) - Required claims presence Raises HTTPException on any validation failure. """ try: payload = jwt.decode( token, JWT_SECRET, algorithms=[JWT_ALGORITHM], # Hardcoded — rejects 'none' options={ "require": ["sub", "type", "exp", "iat"], "verify_exp": True, }, ) return payload except jwt.ExpiredSignatureError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Token has expired", headers={"WWW-Authenticate": "Bearer"}, ) except jwt.InvalidTokenError as e: logger.warning("Invalid token attempt: %s", type(e).__name__) raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token", headers={"WWW-Authenticate": "Bearer"}, ) async def get_current_user(token: str = Depends(oauth2_scheme)) -> int: """ FastAPI dependency that extracts and validates the current user from the Authorization Bearer token. Returns the entity_id (int) of the authenticated user. Usage: @app.get("/protected/") async def protected_endpoint(entity_id: int = Depends(get_current_user)): ... """ payload = _decode_token(token) if payload.get("type") != "access": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type. Use an access token.", headers={"WWW-Authenticate": "Bearer"}, ) try: entity_id = int(payload["sub"]) except (ValueError, KeyError): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token claims", headers={"WWW-Authenticate": "Bearer"}, ) # JWTs are stateless. By relying on the cryptographically signed token and # its short expiry time (1 hour), we avoid querying the database on every request. # If a user is deleted/deactivated, their refresh token is revoked, and their # current access token will expire naturally. return entity_id def verify_refresh_token(token: str) -> dict: """ Verify a refresh token and return its payload. Returns: dict with 'sub' (entity_id as string), 'jti', 'exp', etc. """ payload = _decode_token(token) if payload.get("type") != "refresh": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type. Use a refresh token.", headers={"WWW-Authenticate": "Bearer"}, ) return payload # --- Refresh Token DB Operations --- def hash_token(token: str) -> str: """SHA-256 hash of a token for secure DB storage.""" return sha256(token.encode("utf-8")).hexdigest() async def store_refresh_token(entity_id: int, token: str, expires_at: datetime): """Store the hash of a refresh token in the database.""" token_hash = hash_token(token) sql = ( "INSERT INTO refresh_tokens (entity_id, token_hash, expires_at, created_at) " "VALUES (:entity_id, :token_hash, :expires_at, NOW())" ) await sql_exe(sql, { "entity_id": entity_id, "token_hash": token_hash, "expires_at": expires_at.strftime("%Y-%m-%d %H:%M:%S"), }) async def is_refresh_token_valid(token: str) -> bool: """Check if a refresh token hash exists in DB and is not revoked.""" token_hash = hash_token(token) sql = ( "SELECT id FROM refresh_tokens " "WHERE token_hash = :token_hash AND is_revoked = 0 " "AND expires_at > NOW()" ) result = await sql_exe(sql, {"token_hash": token_hash}, single=True) return result is not None async def revoke_refresh_token(token: str): """Revoke a specific refresh token.""" token_hash = hash_token(token) sql = "UPDATE refresh_tokens SET is_revoked = 1 WHERE token_hash = :token_hash" await sql_exe(sql, {"token_hash": token_hash}) async def revoke_all_user_tokens(entity_id: int): """ Revoke ALL refresh tokens for a user. Called on password change, account deletion, etc. """ sql = "UPDATE refresh_tokens SET is_revoked = 1 WHERE entity_id = :entity_id" await sql_exe(sql, {"entity_id": entity_id}) async def cleanup_expired_tokens(): """Remove expired tokens from the database (housekeeping).""" sql = "DELETE FROM refresh_tokens WHERE expires_at < NOW()" await sql_exe(sql)