import jwt import bcrypt import secrets from datetime import datetime, timedelta from typing import Optional, Dict, Any from fastapi import Header, HTTPException from config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES def hash_password(password: str) -> str: """Hash a password using bcrypt""" salt = bcrypt.gensalt() hashed = bcrypt.hashpw(password.encode('utf-8'), salt) return hashed.decode('utf-8') def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify a password against its hash""" return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) def create_access_token(data: Dict[Any, Any], 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) -> Optional[Dict[str, Any]]: """Decode and verify a JWT access token""" try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) return payload except jwt.PyJWTError: return None def generate_session_token() -> str: """Generate a secure random session token""" return secrets.token_urlsafe(32) async def authenticate_user(authorization: str = Header(None)) -> Dict[str, Any]: """ Authenticate user from JWT token in Authorization header This is a FastAPI dependency that can be used with Depends() Usage: @router.post("/endpoint") async def endpoint(current_user: dict = Depends(authenticate_user)): # current_user contains user data from JWT token """ # Import here to avoid circular dependency from auth_middleware import get_current_user_from_token return await get_current_user_from_token(authorization)