fix(tradein/auth): отказ по насыщению — до выборки из БД и с агрегированным следом (#2715)

Два пункта хвоста #2712/#2717.

1. След инцидента. Отказ при насыщении сознательно не пишет login_failed и не
   тратит бюджет неудач по имени (иначе насыщением блокируют чужую учётку) —
   значит инцидент был виден только строкой logger.warning НА КАЖДЫЙ отказ.
   Лог у бэкенда общий и ограниченный (docker json-file, 20m × 3): при флуде в
   сотни запросов в секунду 60 МБ прокручиваются за минуты и выселяют все
   прочие логи ровно во время атаки. Теперь на окно (1с) — одна строка и одно
   событие login_verify_saturated в user_events, оба с числом отказов,
   накопленных с прошлой записи. Событие важнее строки: аудит переживает и
   ротацию логов, и редеплой. Имя в событии пустое намеренно — отказ случился
   до того, как мы на имя посмотрели, а запись присланного дала бы атакующему
   строки аудита с любым именем на выбор.

2. Гейт насыщения переехал ПЕРЕД выборкой пользователя. Раньше заведомо
   отклоняемый запрос всё равно брал соединение из пула и делал SELECT по имени
   — и это была единственная работа на пути отказа, чьё время зависит от
   существования учётки (bcrypt, который эту разницу ровняет, до отказанного
   запроса не доходит). Добавлен verify_slots_saturated(key) — тот же предикат,
   что решает отказ, но без взятия слота; verify_password_bounded зовёт его же,
   так что двум условиям разъехаться нечем и инвариант «одна точка выноса =
   одна точка учёта» цел. Предчек учитывает и общий потолок, и долю на ключ
   (#2714): при флуде с одного адреса первой упирается именно доля.

Замер (20 заведомо отклоняемых запросов): до — 20 выборок из БД, 20 строк лога;
после — 0 выборок, 1 строка. Все три ветки проверены мутацией кода: тесты
краснеют.

Refs #2715

tradein-mvp/backend/app/api/v1/auth.py

@@ -40,6 +40,10 @@ Security:
     флуда получал 429 столько раз, сколько пытался. Ключ — IP, поэтому защита
     поднимает стоимость атаки, но не закрывает её (подделка за вторым прокси,
     общий адрес за NAT, ротация через ботнет) — см. docstring той же функции.
+    Отказ по насыщению выдаётся ДО выборки из реестра (#2715): иначе на этом
+    пути оставалась бы единственная работа, время которой зависит от того,
+    существует ли имя, — а bcrypt, который эту разницу ровняет, до него уже не
+    доходит. След инцидента — агрегированный, `_saturated_429`.
   - Поверх него — ГЛОБАЛЬНЫЙ счётчик неудач на ИМЯ, без IP в ключе (#2571):
     лимит по паре (username, IP) распределённый перебор обходит целиком, просто
     меняя адрес. Превышение порога не блокирует вход, а замедляет ответ
@@ -54,6 +58,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import secrets
+import time
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
@@ -61,7 +66,12 @@ from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
-from app.core.password import PasswordVerifyOverloadedError, hash_password, verify_password_bounded
+from app.core.password import (
+    PasswordVerifyOverloadedError,
+    hash_password,
+    verify_password_bounded,
+    verify_slots_saturated,
+)
 from app.core.ratelimit import SlidingWindowLimiter, _client_ip
 from app.services.auth_session import create_session, get_user_by_username, revoke_session
 from app.services.identity_store import AccessState, get_identity_db
@@ -174,6 +184,85 @@ def _throttle_delay_s(fails_in_window: int) -> float:
     return min(settings.login_username_throttle_max_delay_s, 2.0 ** min(excess - 1, 16))
 
 
+# Не чаще одной записи в это окно на ВСЕ отказы по насыщению (#2715). Окно, а не
+# запись на запрос, потому что лог у бэкенда общий и ограниченный (docker
+# json-file, max-size 20m × max-file 3): при флуде в сотни запросов в секунду
+# строка на каждый отказ прокручивает 60 МБ за минуты и выселяет ВСЕ остальные
+# логи ровно во время атаки — то есть в момент, когда они нужнее всего.
+# Значение не в настройках намеренно: это не тюнинг, а «человек читает лог», и
+# крутить его нечем — меньше секунды возвращает исходную проблему, больше
+# ухудшает разрешение по времени, не давая взамен ничего.
+_SATURATION_REPORT_WINDOW_S = 1.0
+
+# Отказов с прошлой записи и когда была прошлая запись (monotonic). Обычные
+# глобалы без лока — по той же причине, что и счётчик слотов в
+# `app.core.password`: обе строчки исполняются в потоке событийного цикла и
+# между чтением и записью нет `await`.
+_saturation_rejected = 0
+_saturation_reported_at = 0.0
+
+
+def _saturated_429(ip: str) -> HTTPException:
+    """429 «слоты сверки заняты» + АГРЕГИРОВАННЫЙ след инцидента.
+
+    Событие неудачного входа тут не пишется и бюджет неудач по имени не
+    тратится сознательно (#2712): пароль не проверялся, это не попытка входа, а
+    трата бюджета означала бы, что насыщением можно заблокировать чужую учётку.
+    Но тогда весь инцидент виден ровно здесь, и до #2715 — только строкой в
+    логе на каждый отклонённый запрос (см. `_SATURATION_REPORT_WINDOW_S`).
+
+    Поэтому на окно приходится одна строка в лог И одно событие
+    `login_verify_saturated` в `user_events` — с числом отказов, накопленных с
+    прошлой записи. Событие важнее строки: аудит переживает и ротацию логов, и
+    редеплой, а по `created_at` соседних строк восстанавливается темп атаки
+    (сколько прошло между записями), поэтому длительность окна в payload не
+    дублируется. Первый отказ отчитывается сразу, а не в конце окна: одиночная
+    аномалия обязана быть видна, даже если продолжения не будет.
+
+    Чего это НЕ делает: у GlitchTip-проекта нет ни правил, ни получателей
+    (#2673), так что уведомление никому не уйдёт — след появляется в аудите и
+    в логе, а не в чьём-то телефоне. Проверить доставку поведенчески сейчас
+    нельзя, и утверждать её здесь было бы враньём.
+
+    `username=""` — не заглушка: имя не пишем ПОТОМУ, что отказ случился до
+    того, как мы на него посмотрели. Записывай мы присланное, атакующий
+    наполнял бы аудит строками с любым именем на выбор. `ip` — адрес последнего
+    отклонённого запроса, то есть ОБРАЗЕЦ: при распределённом флуде адресов
+    много, и по одной записи их не восстановить (счётчик — восстановит).
+    """
+    global _saturation_rejected, _saturation_reported_at
+
+    _saturation_rejected += 1
+    now = time.monotonic()
+    if now - _saturation_reported_at >= _SATURATION_REPORT_WINDOW_S:
+        rejected, _saturation_rejected = _saturation_rejected, 0
+        _saturation_reported_at = now
+        logger.warning(
+            "login rejected: password verify saturated — %d отказов с прошлой записи "
+            "(не чаще раза в %.0fс), последний ip=%s",
+            rejected,
+            _SATURATION_REPORT_WINDOW_S,
+            ip,
+        )
+        schedule_event(
+            event_type="login_verify_saturated",
+            username="",
+            ip=ip,
+            path="/api/v1/auth/login",
+            method="POST",
+            payload={"rejected": rejected},
+        )
+
+    # Retry-After 1с — порядок времени одной сверки, не окно соседнего
+    # `_LOGIN_LIMITER`. Ответ ОДИН И ТОТ ЖЕ для любого имени: отказ приходит до
+    # сверки и потому ничего не сообщает о том, существует ли учётка.
+    return HTTPException(
+        status_code=429,
+        detail="слишком много попыток входа, попробуйте позже",
+        headers={"Retry-After": "1"},
+    )
+
+
 async def _reject_invalid_credentials(
     db: Session, username: str, ip: str, user_agent: str | None
 ) -> HTTPException:
@@ -256,6 +345,16 @@ async def login(
             headers={"Retry-After": str(int(retry_after) + 1)},
         )
 
+    # Гейт насыщения — ДО выборки из реестра (#2715). Заведомо отклоняемый
+    # запрос не берёт соединение из пула и не делает SELECT по имени: под
+    # насыщением это была бы единственная работа на пути отказа, а значит и
+    # единственное, чьё время зависит от существования учётки — bcrypt, который
+    # эту разницу ровняет, до отказанного запроса не доходит вовсе. Решение
+    # всё равно остаётся за `verify_password_bounded` ниже (тот же предикат,
+    # `except` под ним никуда не делся) — здесь только экономия похода в базу.
+    if verify_slots_saturated(ip):
+        raise _saturated_429(ip)
+
     user = get_user_by_username(db, body.username)
     hash_to_check = (
         user["password_hash"]
@@ -274,17 +373,11 @@ async def login(
         password_ok = await verify_password_bounded(body.password, hash_to_check, key=ip)
     except PasswordVerifyOverloadedError:
         # Настоящий потолок темпа (#2665): слоты проверки заняты, ждать нельзя —
-        # ждущий держит соединение к БД. Отказ ОДИНАКОВ для любого имени и
+        # ждущий держит соединение к БД. Предчек выше сюда почти всё и отсекает,
-        # случается ДО сверки, поэтому оракулом существования учётки не служит и
+        # но авторитетен ИМЕННО ЭТОТ отказ, поэтому ветка остаётся. Ответ —
-        # бюджет неудач по имени не тратит (это не попытка входа: пароль не
+        # тот же самый и с той же аргументацией, что у предчека: один helper,
-        # проверялся). Retry-After 1с — порядок времени одной проверки, не окно
+        # чтобы две ветки не разъехались (одинаковость 429 — часть защиты).
-        # соседнего `_LOGIN_LIMITER`.
+        raise _saturated_429(ip) from None
-        logger.warning("login rejected: password verify saturated ip=%s", ip)
-        raise HTTPException(
-            status_code=429,
-            detail="слишком много попыток входа, попробуйте позже",
-            headers={"Retry-After": "1"},
-        ) from None
 
     # Пароль проверен ВЫШЕ и безусловно — только теперь смотрим на состояние
     # доступа. Порядок несущий, а не стилистический: см. модульный docstring.

tradein-mvp/backend/app/core/password.py

@@ -123,6 +123,32 @@ def _per_key_slot_cap() -> int:
     return max(1, settings.login_password_verify_max_inflight // 2)
 
 
+def verify_slots_saturated(key: str) -> bool:
+    """Тот же предикат, по которому отказывает `verify_password_bounded`, но БЕЗ взятия слота.
+
+    Нужен вызывающему ровно затем, чтобы отказать ДО похода в БД (#2715). Гейт
+    стоял ПОСЛЕ выборки пользователя, и каждый заведомо отклоняемый запрос всё
+    равно брал соединение из пула и делал SELECT по имени — тогда, когда система
+    уже перегружена. Хуже того, под насыщением эта выборка оставалась
+    ЕДИНСТВЕННОЙ работой на пути отказа: bcrypt, который ровняет время ответа
+    для существующего и несуществующего имени, ниже по течению и до него не
+    доходит, так что разницу «строка найдена / не найдена» ничто не маскировало.
+
+    Предчек, а не решение: авторитетная проверка остаётся внутри
+    `verify_password_bounded` — она зовёт ЭТУ ЖЕ функцию, так что разъехаться
+    двум условиям нечем, и инвариант «одна точка выноса = одна точка учёта»
+    цел (слот здесь не резервируется и не отдаётся).
+
+    Учитывает и общий потолок, и долю на ключ (#2714) — иначе предчек не
+    покрывал бы главный случай: при флуде с ОДНОГО адреса первым упирается
+    именно доля, и большинство отказов снова ходило бы в базу.
+    """
+    return (
+        _verify_inflight >= settings.login_password_verify_max_inflight
+        or _verify_inflight_by_key.get(key, 0) >= _per_key_slot_cap()
+    )
+
+
 async def verify_password_bounded(plain: str, hashed: str, *, key: str) -> bool:
     """`verify_password`, унесённая с событийного цикла И с сознательным потолком темпа (#2665).
 
@@ -193,9 +219,10 @@ async def verify_password_bounded(plain: str, hashed: str, *, key: str) -> bool:
     """
     global _verify_inflight
 
-    if _verify_inflight >= settings.login_password_verify_max_inflight:
+    # АВТОРИТЕТНАЯ проверка. Вызывающий может спросить то же самое заранее
-        raise PasswordVerifyOverloadedError
+    # (`verify_slots_saturated`, #2715), но решение принимается здесь и только
-    if _verify_inflight_by_key.get(key, 0) >= _per_key_slot_cap():
+    # здесь — предчек экономит поход в БД, а не заменяет этот отказ.
+    if verify_slots_saturated(key):
         raise PasswordVerifyOverloadedError
 
     loop = asyncio.get_running_loop()

tradein-mvp/backend/tests/test_auth_api.py

@@ -32,6 +32,7 @@ in-memory fake DB standing in for the identity registry:
 from __future__ import annotations
 
 import asyncio
+import logging
 import os
 import re
 import time
@@ -259,6 +260,10 @@ def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
     auth_mod.reset_cache_for_tests()
     auth_router._LOGIN_LIMITER._hits.clear()
     auth_router._USERNAME_FAIL_LIMITER._hits.clear()
+    # Агрегатор отказов по насыщению (#2715) — тоже глобал процесса: без сброса
+    # недосчитанные отказы одного теста всплывают в записи другого.
+    monkeypatch.setattr(auth_router, "_saturation_rejected", 0)
+    monkeypatch.setattr(auth_router, "_saturation_reported_at", 0.0)
     monkeypatch.setattr(config.settings, "auth_mode", "dual")
     # Каждый тест стартует в ДЕФОЛТНОМ режиме реестра (сегодняшний прод), даже
     # если предыдущий переключался на `auth`.
@@ -958,6 +963,128 @@ async def test_flood_from_one_ip_leaves_login_open_for_another_ip(
     )
 
 
+# ---------------------------------------------------------------------------
+# #2715 — отказ по насыщению: до похода в БД и со следом, который не выселяет лог
+# ---------------------------------------------------------------------------
+
+
+def _saturate_verify_slots(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Слоты сверки заняты — снаружи ровно то же, что живой флуд, но без гонок.
+
+    Именно счётчик, а не мок `verify_password_bounded`: проверяем настоящий
+    предикат отказа (`verify_slots_saturated` читает этот же глобал), а не
+    собственную заглушку.
+    """
+    monkeypatch.setattr(password_mod, "_verify_inflight", 999)
+
+
+def test_saturated_login_answers_before_touching_the_registry(
+    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Под насыщением отказ приходит ДО выборки пользователя (#2715).
+
+    Гейт стоял после SELECT'а, и каждый заведомо отклоняемый запрос всё равно
+    брал соединение из пула — тогда, когда система уже перегружена. Хуже того,
+    эта выборка оставалась ЕДИНСТВЕННОЙ работой на пути отказа: bcrypt, ровняющий
+    время ответа для существующего и несуществующего имени, до отказанного
+    запроса не доходит вовсе, так что разницу маскировать было нечем.
+
+    Мерим не тайминг (в CI флейкует), а сам факт похода в реестр — и заодно
+    побайтовую одинаковость ответа для живого и выдуманного имени.
+    """
+    store.add_user("alice", hash_password("Secret123!"), role="employee")
+    _capture_events(monkeypatch)
+
+    lookups: list[str] = []
+    real_lookup = auth_router.get_user_by_username
+
+    def _spy(db: Any, username: str) -> Any:
+        lookups.append(username)
+        return real_lookup(db, username)
+
+    monkeypatch.setattr(auth_router, "get_user_by_username", _spy)
+    _saturate_verify_slots(monkeypatch)
+
+    bodies = []
+    for name in ("alice", "ghost"):
+        resp = client.post(
+            "/api/v1/auth/login",
+            json={"username": name, "password": "x"},
+            headers={"x-forwarded-for": "203.0.113.5"},
+        )
+        assert resp.status_code == 429, resp.text
+        assert resp.headers["Retry-After"] == "1"
+        bodies.append(resp.text)
+
+    assert lookups == [], f"под насыщением всё-таки сходили в реестр: {lookups}"
+    # Существующее и несуществующее имя — неразличимы (#2571 на этом пути тоже).
+    assert bodies[0] == bodies[1]
+
+
+def test_saturation_is_reported_once_per_window_and_lands_in_audit(
+    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch, caplog: Any
+) -> None:
+    """Двадцать отказов — одна запись в лог и одно событие в аудит, со счётчиком.
+
+    Строка на каждый отказ делила с остальным бэкендом `json-file max-size 20m ×
+    max-file 3`: при флуде в сотни запросов в секунду 60 МБ прокручиваются за
+    минуты и выселяют ВСЕ остальные логи ровно во время атаки. Поэтому окно.
+
+    А событие в `user_events` — потому что до #2715 инцидент не оставлял в
+    аудите ни строчки: событие неудачного входа тут не пишется намеренно
+    (пароль не проверялся, и трата бюджета неудач дала бы блокировку чужой
+    учётки насыщением) — значит нужен отдельный тип события, и он обязан
+    появляться независимо от того, ротировался лог или нет.
+    """
+    # ЛИТЕРАЛ, а не арифметика от настройки: окно — компромисс «видно вовремя»
+    # против «не выселяет лог», и подъём его до минут прячет атаку целиком.
+    assert auth_router._SATURATION_REPORT_WINDOW_S == 1.0
+
+    events = _capture_events(monkeypatch)
+    _saturate_verify_slots(monkeypatch)
+    # Окно на весь тест — иначе медленный CI разбил бы 20 запросов на два окна
+    # и число записей стало бы функцией скорости раннера.
+    monkeypatch.setattr(auth_router, "_SATURATION_REPORT_WINDOW_S", 60.0)
+    caplog.set_level(logging.WARNING, logger="app.api.v1.auth")
+
+    for i in range(20):
+        resp = client.post(
+            "/api/v1/auth/login",
+            json={"username": f"ghost{i}", "password": "x"},
+            headers={"x-forwarded-for": "203.0.113.5"},
+        )
+        assert resp.status_code == 429, resp.text
+
+    lines = [r for r in caplog.records if "saturated" in r.getMessage()]
+    assert len(lines) == 1, f"20 отказов дали {len(lines)} строк в логе — агрегации нет"
+
+    saturated = [e for e in events if e["event_type"] == "login_verify_saturated"]
+    assert len(saturated) == 1, saturated
+    # Первый отказ отчитывается сразу (одиночная аномалия обязана быть видна),
+    # поэтому в первой записи он один — накопленное придёт следующей.
+    assert saturated[0]["payload"] == {"rejected": 1}
+    assert saturated[0]["ip"] == "203.0.113.5"
+    # Имя не пишем: отказ случился ДО того, как мы на него посмотрели, а запись
+    # присланного дала бы атакующему аудит-строки с любым именем на выбор.
+    assert saturated[0]["username"] == ""
+
+    # Бюджет неудач по имени не тронут — иначе насыщением блокируют чужой вход.
+    assert [e for e in events if e["event_type"] == "login_failed"] == []
+    assert not auth_router._USERNAME_FAIL_LIMITER._hits
+
+    # Окно прошло — следующий отказ приносит НАКОПЛЕННОЕ, а не единицу.
+    monkeypatch.setattr(auth_router, "_saturation_reported_at", time.monotonic() - 61.0)
+    resp = client.post(
+        "/api/v1/auth/login",
+        json={"username": "ghost-last", "password": "x"},
+        headers={"x-forwarded-for": "203.0.113.5"},
+    )
+    assert resp.status_code == 429
+    saturated = [e for e in events if e["event_type"] == "login_verify_saturated"]
+    assert len(saturated) == 2
+    assert saturated[1]["payload"] == {"rejected": 20}, "счётчик за окно потерян"
+
+
 # ---------------------------------------------------------------------------
 # POST /logout
 # ---------------------------------------------------------------------------