"""Пароли, 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)