from passlib.context import CryptContext from datetime import datetime, timedelta from typing import Optional, Union import uuid from jose import JWTError, jwt from ..config import settings # Password hashing context pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def hash_password(password: str) -> str: """Hash a password using bcrypt.""" return pwd_context.hash(password) def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify a plain password against its 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: # Default to 7 days if no expiration is provided expire = datetime.utcnow() + timedelta(days=settings.ACCESS_TOKEN_EXPIRE_DAYS) to_encode.update({"exp": expire, "iat": datetime.utcnow()}) encoded_jwt = jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM) return encoded_jwt def verify_token(token: str) -> Optional[dict]: """Verify a JWT token and return the payload if valid.""" try: payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) return payload except JWTError: return None def verify_user_id_from_token(token: str) -> Optional[uuid.UUID]: """Extract user_id from JWT token.""" payload = verify_token(token) if payload: user_id_str = payload.get("sub") if user_id_str: try: return uuid.UUID(user_id_str) except ValueError: return None return None