"""Хэширование паролей (argon2) и выпуск/проверка JWT (access + refresh).""" import uuid from datetime import UTC, datetime, timedelta from typing import Any import jwt from argon2 import PasswordHasher from argon2.exceptions import VerifyMismatchError from core.config import get_settings JWT_ALGORITHM = "HS256" _hasher = PasswordHasher() def hash_password(password: str) -> str: """Захэшировать пароль алгоритмом argon2 для хранения в БД.""" return _hasher.hash(password) def verify_password(password: str, password_hash: str) -> bool: """Сверить пароль с сохранённым argon2-хэшем; пароль/хэш никогда не логируются.""" try: return _hasher.verify(password_hash, password) except VerifyMismatchError: return False def create_access_token(user_id: uuid.UUID, role: str) -> str: """Выпустить access-токен: `sub`=user_id, `role`=роль, TTL из настроек.""" settings = get_settings() now = datetime.now(UTC) payload = { "sub": str(user_id), "role": role, "type": "access", "iat": now, "exp": now + timedelta(minutes=settings.access_token_ttl_minutes), } return jwt.encode(payload, settings.jwt_secret, algorithm=JWT_ALGORITHM) def create_refresh_token(user_id: uuid.UUID) -> tuple[str, str]: """Выпустить refresh-токен с уникальным `jti`. Возвращает пару (token, jti); сохранение jti в Redis — ответственность вызывающего кода (`services.auth.AuthService`). """ settings = get_settings() jti = str(uuid.uuid4()) now = datetime.now(UTC) payload = { "sub": str(user_id), "jti": jti, "type": "refresh", "iat": now, "exp": now + timedelta(days=settings.refresh_token_ttl_days), } token = jwt.encode(payload, settings.jwt_secret, algorithm=JWT_ALGORITHM) return token, jti def decode_token(token: str) -> dict[str, Any]: """Декодировать и верифицировать JWT (сигнатура + срок действия). Бросает `jwt.PyJWTError` (или подкласс) при невалидном/просроченном токене. """ settings = get_settings() return jwt.decode(token, settings.jwt_secret, algorithms=[JWT_ALGORITHM])