import hashlib import logging import secrets from datetime import datetime, timedelta, timezone from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from app.core.config import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") logger = logging.getLogger(__name__) def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def generate_session_id() -> str: """Unique session ID stored in DB — invalidates old sessions.""" return secrets.token_hex(32) def generate_password_reset_token() -> str: return secrets.token_urlsafe(48) def hash_password_reset_token(token: str) -> str: return hashlib.sha256(token.encode("utf-8")).hexdigest() def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: to_encode = data.copy() expire = datetime.now(timezone.utc) + ( expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) ) to_encode["exp"] = expire if "sub" in to_encode: to_encode["sub"] = str(to_encode["sub"]) return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) def decode_token(token: str) -> Optional[dict]: try: payload = jwt.decode( token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM] ) return payload except JWTError as e: logger.warning("JWT decode error: %s", e) return None