from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from fastapi import HTTPException, Security, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials import os from dotenv import load_dotenv from pathlib import Path # Load environment variables ROOT_DIR = Path(__file__).parent load_dotenv(ROOT_DIR / '.env') # JWT Configuration SECRET_KEY = os.environ['JWT_SECRET_KEY'] ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 1440 # 24 hours # Password hashing pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # Security security = HTTPBearer() def hash_password(password: str) -> str: """Hash a password""" return pwd_context.hash(password) def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify a password against a hash""" return pwd_context.verify(plain_password, hashed_password) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: """Create a JWT access token""" to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def decode_access_token(token: str) -> dict: """Decode and verify a JWT token""" try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) return payload except JWTError: raise HTTPException(status_code=401, detail="Could not validate credentials") async def get_current_admin(credentials: HTTPAuthorizationCredentials = Security(security)) -> dict: """Dependency to get current admin from JWT token""" token = credentials.credentials payload = decode_access_token(token) email: str = payload.get("sub") if email is None: raise HTTPException(status_code=401, detail="Invalid authentication credentials") return {"email": email}