import secrets from datetime import datetime, timedelta from typing import Any, Dict, Optional, Tuple, Union from jose import jwt from passlib.context import CryptContext from app.core.config import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # Define token types ACCESS_TOKEN_TYPE = "access" REFRESH_TOKEN_TYPE = "refresh" # Expiration times ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES REFRESH_TOKEN_EXPIRE_DAYS = 30 def create_access_token( subject: Union[str, Any], expires_delta: Optional[timedelta] = None ) -> Tuple[str, datetime]: if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta( minutes=ACCESS_TOKEN_EXPIRE_MINUTES ) to_encode = {"exp": expire, "sub": str(subject), "type": ACCESS_TOKEN_TYPE} encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256") return encoded_jwt, expire def create_refresh_token( subject: Union[str, Any], expires_delta: Optional[timedelta] = None ) -> Tuple[str, datetime]: if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS) # Generate a secure random token refresh_token = secrets.token_urlsafe(64) # No need to encode anything as we'll store this in the database return refresh_token, expire def verify_token(token: str) -> Optional[Dict[str, Any]]: try: payload = jwt.decode( token, settings.SECRET_KEY, algorithms=["HS256"] ) return payload except jwt.JWTError: return None def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: return pwd_context.hash(password)