from datetime import datetime, timedelta from typing import Any, Dict, Optional, Union from jose import jwt from passlib.context import CryptContext from app.core.config import settings # Password hashing pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") 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 get_password_hash(password: str) -> str: """Generate password hash.""" return pwd_context.hash(password) # JWT token functions def create_access_token(subject: Union[str, Any], expires_delta: Optional[timedelta] = None) -> str: """ Create a JWT access token. """ if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode = {"exp": expire, "sub": str(subject)} encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt def create_refresh_token(subject: Union[str, Any], expires_delta: Optional[timedelta] = None) -> str: """ Create a JWT refresh token. """ if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.REFRESH_TOKEN_EXPIRE_MINUTES) to_encode = {"exp": expire, "sub": str(subject), "type": "refresh"} encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt def create_password_reset_token(subject: Union[str, Any]) -> str: """ Create a password reset token. """ expire = datetime.utcnow() + timedelta(hours=settings.PASSWORD_RESET_TOKEN_EXPIRE_HOURS) to_encode = {"exp": expire, "sub": str(subject), "type": "password_reset"} encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt