72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
"""Хэширование паролей (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])
|