Обновление файла

This commit is contained in:
2026-07-24 22:06:13 +03:00
parent c807a510fd
commit 144f4dc0a8
+23 -6
View File
@@ -1,25 +1,41 @@
"""Утилиты для паролей и JWT-токенов."""
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Optional from typing import Optional
from jose import jwt, JWTError import bcrypt
from passlib.context import CryptContext from jose import JWTError, jwt
from .config import settings from .config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
ALGORITHM = "HS256" ALGORITHM = "HS256"
# bcrypt ограничивает длину входа 72 байтами. Длиннее — обрезаем.
_BCRYPT_MAX_BYTES = 72
def _normalize_for_bcrypt(password: str) -> bytes:
"""bcrypt принимает максимум 72 байта. Длинные пароли обрезаем,
чтобы не получать ValueError при хешировании."""
encoded = password.encode("utf-8")
return encoded[:_BCRYPT_MAX_BYTES]
def hash_password(password: str) -> str: def hash_password(password: str) -> str:
return pwd_context.hash(password) """Возвращает bcrypt-хеш пароля (ASCII-строка)."""
salt = bcrypt.gensalt(rounds=12)
return bcrypt.hashpw(_normalize_for_bcrypt(password), salt).decode("utf-8")
def verify_password(plain: str, hashed: str) -> bool: def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed) """Проверяет пароль против bcrypt-хеша."""
try:
return bcrypt.checkpw(_normalize_for_bcrypt(plain), hashed.encode("utf-8"))
except (ValueError, TypeError):
return False
def create_access_token(data: dict, expires_minutes: Optional[int] = None) -> str: def create_access_token(data: dict, expires_minutes: Optional[int] = None) -> str:
"""Создаёт JWT с подписью HS256 и заданным сроком жизни."""
to_encode = data.copy() to_encode = data.copy()
expire = datetime.utcnow() + timedelta( expire = datetime.utcnow() + timedelta(
minutes=expires_minutes or settings.access_token_expire_minutes minutes=expires_minutes or settings.access_token_expire_minutes
@@ -29,6 +45,7 @@ def create_access_token(data: dict, expires_minutes: Optional[int] = None) -> st
def decode_access_token(token: str) -> Optional[dict]: def decode_access_token(token: str) -> Optional[dict]:
"""Декодирует JWT. Возвращает None при любой ошибке (просрочен, подделан и т.п.)."""
try: try:
return jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM]) return jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])
except JWTError: except JWTError: