# app/middleware/auth.py from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials import jwt from datetime import datetime, timedelta import os security = HTTPBearer() JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY') JWT_ACCESS_TOKEN_EXPIRES = int(os.getenv('JWT_ACCESS_TOKEN_EXPIRES')) def create_access_token(data: dict): to_encode = data.copy() expire = datetime.utcnow() + timedelta(seconds=JWT_ACCESS_TOKEN_EXPIRES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, JWT_SECRET_KEY, algorithm="HS256") return encoded_jwt def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)): try: payload = jwt.decode(credentials.credentials, JWT_SECRET_KEY, algorithms=["HS256"]) username: str = payload.get("sub") if username is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials", headers={"WWW-Authenticate": "Bearer"}, ) return username except jwt.PyJWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials", headers={"WWW-Authenticate": "Bearer"}, ) def get_current_user(username: str = Depends(verify_token)): return username # For optional JWT authentication (some endpoints allow unauthenticated access) def get_optional_user(authorization: HTTPAuthorizationCredentials = Depends(security)): try: payload = jwt.decode(authorization.credentials, JWT_SECRET_KEY, algorithms=["HS256"]) username: str = payload.get("sub") return username except: return None