from datetime import datetime, timedelta, timezone from typing import Optional, Tuple from jose import jwt, JWTError from passlib.context import CryptContext from app.core.config import settings pwd_context = CryptContext(schemes=["argon2"], deprecated="auto") 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) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: to_encode = data.copy() if expires_delta: expire = datetime.now(timezone.utc) + expires_delta else: expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire, "type": "access"}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt def create_refresh_token(data: dict) -> str: to_encode = data.copy() expire = datetime.now(timezone.utc) + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS) to_encode.update({"exp": expire, "type": "refresh"}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt def decode_token(token: str) -> Optional[dict]: try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) return payload except JWTError: return None def create_token_pair(user_id: int, username: str) -> Tuple[str, str]: access = create_access_token({"sub": str(user_id), "username": username}) refresh = create_refresh_token({"sub": str(user_id), "username": username}) return access, refresh