from datetime import datetime, timedelta from typing import Any, Optional from jose import jwt from passlib.context import CryptContext from app.core.config import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def hash_password(password: str) -> str: # bcrypt limits to 72 bytes; truncate defensively to avoid runtime errors if len(password.encode("utf-8")) > 72: password = password.encode("utf-8")[:72].decode("utf-8", errors="ignore") return pwd_context.hash(password) def verify_password(password: str, password_hash: str) -> bool: return pwd_context.verify(password, password_hash) def create_access_token(subject: str, expires_minutes: Optional[int] = None) -> str: expire = datetime.utcnow() + timedelta(minutes=expires_minutes or settings.access_token_expires_minutes) payload = {"sub": subject, "exp": expire, "type": "access"} return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm) def create_refresh_token(subject: str, expires_days: Optional[int] = None) -> str: expire = datetime.utcnow() + timedelta(days=expires_days or settings.refresh_token_expires_days) payload = {"sub": subject, "exp": expire, "type": "refresh"} return jwt.encode(payload, settings.jwt_refresh_secret_key, algorithm=settings.jwt_algorithm) def decode_token(token: str, token_type: str) -> dict[str, Any]: secret = settings.jwt_secret_key if token_type == "access" else settings.jwt_refresh_secret_key return jwt.decode(token, secret, algorithms=[settings.jwt_algorithm])