from fastapi import Depends, HTTPException, status from fastapi.security import APIKeyHeader, OAuth2PasswordBearer from jose import jwt, JWTError from app.config import get_settings api_key_header = APIKeyHeader(name=get_settings().API_KEY_NAME, auto_error=False) oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") async def validate_api_key(api_key_header: str = Depends(api_key_header)): if api_key_header != get_settings().API_KEY: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Could not validate credentials" ) return api_key_header def get_user(user_id: str): # Define here how to fetch a user from your database # using the user_id. Here's a simple placeholder implementation: # TODO: Update this function to fetch a user from your actual database if get_settings().ENVIRONMENT == "local": return {"id": user_id, "name": "Dummy User"} else: user = None if user_id == "known_id": user = {"id": user_id, "name": "Known User"} return user def validate_jwt_token(token: str = Depends(oauth2_scheme)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: # The SECRET_KEY should match the NEXTAUTH_SECRET in the front end payload = jwt.decode(token, get_settings().SECRET_KEY, algorithms=[get_settings().ALGORITHM]) user_id: str = payload.get("sub") if user_id is None: raise credentials_exception user = get_user(user_id=user_id) if user is None: raise credentials_exception return user except JWTError: raise credentials_exception async def validate_token(token: str = Depends(oauth2_scheme)): token_data = validate_jwt_token(token) if not token_data: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Could not validate credentials" ) return token