lauadminandClaude Opus 5.5
b8f2fe6b6e
- All frontend pages, labels, notices and errors; html lang=uk, uk-UA money format - Brand "Assistant System" in the top bar and page title - Backend error details returned to the UI (auth, orders, receipts, cash registers, Checkbox/CRM/NP errors) and CLI output - Tests updated for the new messages Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
98 lines
3.0 KiB
Python
98 lines
3.0 KiB
Python
"""Пароли, 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)
|