split-trainer/server/splitapi/security.py

"""Примитивы безопасности: проверка почты и паролей, хеширование (scrypt), токены."""
from __future__ import annotations

import base64
import hashlib
import hmac
import re
import secrets
import threading
import unicodedata

# Параметры scrypt (OWASP допускает N=2^15, r=8, p=1 как разумный минимум для небольшого сервера).
SCRYPT_N = 2**15
SCRYPT_R = 8
SCRYPT_P = 1
_DKLEN = 32
_SALT_BYTES = 16
_MAX_MEM = 256 * 1024 * 1024

PASSWORD_MIN = 8
PASSWORD_MAX = 128           # верхняя граница ограничивает стоимость хеширования для атакующего
_EMAIL_MAX = 254

# Не больше нескольких одновременных хешей: каждый занимает десятки МБ памяти.
SLOT_WAIT = 5.0                       # сколько секунд ждать свободный слот, прежде чем ответить «сервер занят»
_slots = threading.BoundedSemaphore(4)


class HashBusy(RuntimeError):
    """Все слоты хеширования заняты: лучше быстро отказать, чем копить зависшие запросы."""

# fullmatch, а не match с $: $ срабатывает и перед завершающим переводом строки.
# В адресе допустимы только безопасные символы: % | / ! # $ & * = ? ^ ` { } ~ формально валидны по RFC,
# но опасны для старых почтовых систем (переадресация через %, выполнение через |).
_LOCAL_RE = re.compile(r"[a-z0-9._+'-]{1,64}")
_LABEL_RE = re.compile(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?")
_TOKEN_RE = re.compile(r"[A-Za-z0-9_-]{16,100}")

_COMMON_PASSWORDS = frozenset(
    {
        "password", "password1", "password123", "12345678", "123456789", "1234567890", "12341234",
        "00000000", "11111111", "qwertyui", "qwertyuiop", "qwerty123", "iloveyou", "admin123",
        "letmein1", "abc12345", "пароль123", "йцукенгш", "qazwsxedc", "1q2w3e4r",
    }
)


def normalize_email(raw: object) -> str | None:
    """Приводит адрес к каноническому виду (нижний регистр, домен в punycode) или возвращает None."""
    if not isinstance(raw, str):
        return None
    email = raw.strip().lower()
    if not email or len(email) > _EMAIL_MAX or email.count("@") != 1:
        return None
    local, domain = email.split("@")
    if not _LOCAL_RE.fullmatch(local) or local.startswith(".") or local.endswith(".") or ".." in local:
        return None
    try:
        ascii_domain = domain.encode("idna").decode("ascii")
    except UnicodeError:
        return None
    labels = ascii_domain.split(".")
    if len(ascii_domain) > 253 or len(labels) < 2 or len(labels[-1]) < 2:
        return None
    if not all(_LABEL_RE.fullmatch(label) for label in labels):
        return None
    return f"{local}@{ascii_domain}"


def validate_password(password: object, email: str) -> str | None:
    """Возвращает код ошибки или None, если пароль допустим."""
    if not isinstance(password, str) or len(password) < PASSWORD_MIN:
        return "password_too_short"
    if len(password) > PASSWORD_MAX:
        return "password_too_long"
    try:
        password.encode("utf-8")
    except UnicodeEncodeError:
        return "password_invalid"         # одиночный суррогат: json его пропускает, а хешер упал бы
    lowered = password.lower()
    local = email.split("@")[0].lower()
    if lowered in (email.lower(), local) or lowered in _COMMON_PASSWORDS:
        return "password_weak"
    return None


def _prepare(password: str) -> bytes:
    return unicodedata.normalize("NFKC", password).encode("utf-8")


def _derive(password: str, salt: bytes, n: int, r: int, p: int) -> bytes:
    if not _slots.acquire(timeout=SLOT_WAIT):
        raise HashBusy("слишком много одновременных проверок пароля")
    try:
        return hashlib.scrypt(_prepare(password), salt=salt, n=n, r=r, p=p, dklen=_DKLEN, maxmem=_MAX_MEM)
    finally:
        _slots.release()


def hash_password(password: str, n: int | None = None) -> str:
    """Формат: scrypt$N$r$p$соль$хеш (параметры внутри позволяют позже усилить хеширование)."""
    n = n or SCRYPT_N
    salt = secrets.token_bytes(_SALT_BYTES)
    digest = _derive(password, salt, n, SCRYPT_R, SCRYPT_P)
    return "$".join(
        ["scrypt", str(n), str(SCRYPT_R), str(SCRYPT_P), base64.b64encode(salt).decode(), base64.b64encode(digest).decode()]
    )


def _parse(stored: str) -> tuple[int, int, int, bytes, bytes] | None:
    try:
        scheme, n, r, p, salt, digest = stored.split("$")
        if scheme != "scrypt":
            return None
        n_i, r_i, p_i = int(n), int(r), int(p)
        # Защита от подменённых параметров: не даём заставить сервер считать гигабайтный хеш
        if not (2**10 <= n_i <= 2**20 and n_i & (n_i - 1) == 0 and 1 <= r_i <= 16 and 1 <= p_i <= 4):
            return None
        return n_i, r_i, p_i, base64.b64decode(salt, validate=True), base64.b64decode(digest, validate=True)
    except (ValueError, TypeError):
        return None


def verify_password(stored: str, password: str) -> bool:
    parsed = _parse(stored) if isinstance(stored, str) else None
    if parsed is None:
        return False
    n, r, p, salt, expected = parsed
    try:
        actual = _derive(password, salt, n, r, p)
    except (ValueError, MemoryError):
        return False
    return hmac.compare_digest(actual, expected)


def needs_rehash(stored: str) -> bool:
    parsed = _parse(stored)
    return parsed is None or parsed[:3] != (SCRYPT_N, SCRYPT_R, SCRYPT_P)


_dummy_lock = threading.Lock()
_dummy_hash: str | None = None


def verify_dummy(password: str) -> bool:
    """Тратит столько же времени, как проверка настоящего пароля: ответ на вход не выдаёт, есть ли такая почта."""
    global _dummy_hash
    with _dummy_lock:
        if _dummy_hash is None:
            _dummy_hash = hash_password(secrets.token_urlsafe(16))
    verify_password(_dummy_hash, password)
    return False


def new_token() -> tuple[str, str]:
    """Возвращает (токен для пользователя, его SHA-256 для базы): в базе токены в открытом виде не хранятся."""
    raw = secrets.token_urlsafe(32)
    return raw, hashlib.sha256(raw.encode("ascii")).hexdigest()


def hash_token(raw: object) -> str | None:
    if not isinstance(raw, str) or not _TOKEN_RE.fullmatch(raw):
        return None
    return hashlib.sha256(raw.encode("ascii")).hexdigest()