from datetime import datetime, timedelta, timezone from typing import Any import jwt from jwt import PyJWTError import bcrypt from app.config import get_settings settings = get_settings() def _encode(plain: str) -> bytes: # bcrypt only uses the first 72 bytes; passlib truncated silently, so do the # same to keep existing hashes verifiable (bcrypt>=5 raises instead). return plain.encode("utf-8")[:72] def hash_password(plain: str) -> str: return bcrypt.hashpw(_encode(plain), bcrypt.gensalt(rounds=12)).decode("ascii") def verify_password(plain: str, hashed: str) -> bool: try: return bcrypt.checkpw(_encode(plain), hashed.encode("ascii")) except ValueError: # malformed hash return False def create_access_token(data: dict[str, Any]) -> str: payload = data.copy() expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes) payload.update({"exp": expire, "type": "access"}) return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm) def create_admin_token(data: dict[str, Any]) -> str: payload = data.copy() expire = datetime.now(timezone.utc) + timedelta(hours=8) payload.update({"exp": expire, "type": "access", "role": "admin"}) return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm) def create_refresh_token(data: dict[str, Any]) -> str: payload = data.copy() expire = datetime.now(timezone.utc) + timedelta(days=settings.refresh_token_expire_days) payload.update({"exp": expire, "type": "refresh"}) return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm) def decode_token(token: str) -> dict[str, Any]: try: return jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm]) except PyJWTError: raise ValueError("Invalid or expired token")