"""Пароли, JWT и refresh-токены.""" from __future__ import annotations import hashlib import secrets import uuid from datetime import UTC, datetime, timedelta from typing import Any, Literal import jwt from argon2 import PasswordHasher from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError from app.core.config import settings ALGORITHM = "HS256" TokenType = Literal["access", "refresh"] _hasher = PasswordHasher() class TokenError(Exception): """Токен отсутствует, просрочен, повреждён или имеет неверный тип.""" # --- Пароли ----------------------------------------------------------------- def hash_password(password: str) -> str: return _hasher.hash(password) def verify_password(password: str, password_hash: str) -> bool: try: _hasher.verify(password_hash, password) except (VerifyMismatchError, VerificationError, InvalidHashError): return False return True def password_needs_rehash(password_hash: str) -> bool: """True, если хеш создан старыми параметрами argon2 и его стоит обновить.""" try: return _hasher.check_needs_rehash(password_hash) except InvalidHashError: return True # --- Access-токены ---------------------------------------------------------- def create_access_token(user_id: uuid.UUID, role: str) -> str: now = datetime.now(UTC) payload = { "sub": str(user_id), "role": role, "type": "access", "iat": now, "exp": now + timedelta(minutes=settings.access_token_expire_minutes), "jti": secrets.token_urlsafe(16), } return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM) def decode_access_token(token: str) -> dict[str, Any]: try: payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM]) except jwt.ExpiredSignatureError as exc: raise TokenError("Срок действия токена истёк") from exc except jwt.InvalidTokenError as exc: raise TokenError("Некорректный токен") from exc # Без этой проверки refresh-токен можно было бы предъявить как access. if payload.get("type") != "access": raise TokenError("Ожидался access-токен") return payload # --- Refresh-токены --------------------------------------------------------- def generate_refresh_token() -> tuple[str, str]: """Возвращает (сырой токен для клиента, sha256-хеш для хранения в БД). Сырое значение не сохраняется нигде: в БД лежит только хеш. """ raw = secrets.token_urlsafe(48) return raw, hash_refresh_token(raw) def hash_refresh_token(raw: str) -> str: return hashlib.sha256(raw.encode()).hexdigest() def refresh_token_expiry() -> datetime: return datetime.now(UTC) + timedelta(days=settings.refresh_token_expire_days)