fix(tradein/auth): доля слотов сверки пароля на адрес — потолок перестаёт бить по своим (#2714)

Потолок темпа из #2665 держал слоты общим котлом: флуд занимал все четыре, и
легитимный вход с ВЕРНЫМ паролем получал 429 столько раз, сколько пытался
(замер на стенде с настоящим bcrypt: 20 попыток → 0×200, 20×429).

Теперь ключ (у единственного вызывающего — IP клиента) не берёт больше половины
ёмкости: сколько бы один источник ни слал, вторая половина остаётся тем, кто
приходит впервые. Учёт по ключу живёт в самой `verify_password_bounded` и
отдаётся тем же `_release_verify_slot` — инвариант «одна точка выноса = одна
точка учёта» не делится надвое.

Граница применимости честная и записана в коде: ключом может быть только IP, а
он подделывается за вторым прокси, разделяется за NAT и ротируется ботнетом.
Это поднимает стоимость атаки, но не закрывает её; принципиально закрывают
только доказательство работы или второй фактор.

Замер на стенде после правки: 20 попыток → 5×200, 15×429, и НИ ОДНОГО 429 от
потолка сверок. Оставшиеся 15 — `_LOGIN_LIMITER` (5 попыток на пару имя+IP за
300с), отдельная и намеренная защита, срабатывающая независимо от флуда.

Тесты: вход с другого ключа во время флуда (API-уровень, мерит заявленное — не
латентность и не число попыток атакующего), доля слотов на ключ, возврат слота
при исключении в сверке, autouse-сторож «слотов занято 0» после каждого теста.
Формулировка «вход не заблокирован совсем» в тесте темпа поправлена: она про
попытки атакующего и ничего не говорит про живого пользователя.

Refs #2714, #2665, #2712

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

@@ -35,6 +35,11 @@ Security:
     потолок (проверок/с не больше workers/282мс). Убрать одно без другого
     нельзя: вынос без потолка ускорил бы перебор вчетверо, потолок без выноса
     оставил бы отказ в обслуживании. Сверх очереди — 429, не ожидание.
+    Слоты делятся ПО АДРЕСУ (#2714): один источник не занимает больше половины,
+    иначе потолок бил и по своим — легитимный вход с верным паролем во время
+    флуда получал 429 столько раз, сколько пытался. Ключ — IP, поэтому защита
+    поднимает стоимость атаки, но не закрывает её (подделка за вторым прокси,
+    общий адрес за NAT, ротация через ботнет) — см. docstring той же функции.
   - Поверх него — ГЛОБАЛЬНЫЙ счётчик неудач на ИМЯ, без IP в ключе (#2571):
     лимит по паре (username, IP) распределённый перебор обходит целиком, просто
     меняя адрес. Превышение порога не блокирует вход, а замедляет ответ
@@ -260,7 +265,13 @@ async def login(
     # ВСЕГДА вызывается — dummy-хеш при отсутствующем юзере/NULL password_hash
     # держит время ответа одинаковым независимо от существования аккаунта.
     try:
-        password_ok = await verify_password_bounded(body.password, hash_to_check)
+        # key=ip — доля слотов на адрес (#2714): один источник не занимает больше
+        # половины ёмкости, и вход остаётся открыт тем, кто приходит с других
+        # адресов. Ключ — ИМЕННО адрес, не имя: имя присылает клиент, и перебор
+        # менял бы его каждую попытку, получая полную долю на каждое. Границы
+        # применимости (IP подделывается за вторым прокси, разделяется за NAT,
+        # ротируется ботнетом) — в docstring `verify_password_bounded`.
+        password_ok = await verify_password_bounded(body.password, hash_to_check, key=ip)
     except PasswordVerifyOverloadedError:
         # Настоящий потолок темпа (#2665): слоты проверки заняты, ждать нельзя —
         # ждущий держит соединение к БД. Отказ ОДИНАКОВ для любого имени и

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

@@ -96,8 +96,34 @@ _VERIFY_POOL = ThreadPoolExecutor(
 # потому одинаково честен под несколькими event loop'ами в тестах.
 _verify_inflight = 0
 
+# То же самое, но в разрезе ключа (#2714). Запись живёт РОВНО пока ключ держит
+# хотя бы слот и удаляется на нуле: размер словаря ограничен числом слотов
+# (`login_password_verify_max_inflight`), а не числом когда-либо виденных
+# адресов — иначе перебор с ротацией IP растил бы его без границы.
+_verify_inflight_by_key: dict[str, int] = {}
 
-async def verify_password_bounded(plain: str, hashed: str) -> bool:
+
+def _per_key_slot_cap() -> int:
+    """Сколько слотов из общего лимита разрешено ОДНОМУ ключу.
+
+    Половина — минимальное деление, при котором один источник, сколько бы он ни
+    слал, физически не может занять всё: вторая половина остаётся тем, кто
+    приходит впервые. Настройкой не сделано сознательно — это доля, а не
+    величина, и подкручивать её нечем: 100% возвращает поведение, ради отказа
+    от которого правка написана.
+
+    Читается на каждом вызове, а не на импорте, — как `_throttle_delay_s`:
+    иначе тестовый monkeypatch лимита не влиял бы на долю.
+
+    `max(1, …)`: при `max_inflight=1` половина округлилась бы в 0, и КАЖДЫЙ вход
+    получал бы отказ молча (свободных слотов нет ни у кого). Молчаливый отказ
+    всем — ровно тот класс поломки, от которого страхует `ge=1` на самой
+    настройке; здесь тот же страховочный пол, но от деления.
+    """
+    return max(1, settings.login_password_verify_max_inflight // 2)
+
+
+async def verify_password_bounded(plain: str, hashed: str, *, key: str) -> bool:
     """`verify_password`, унесённая с событийного цикла И с сознательным потолком темпа (#2665).
 
     ДВЕ ПОЛОВИНЫ ОДНОЙ ПРАВКИ, И ЖИВУТ ОНИ ЗДЕСЬ ВМЕСТЕ НЕ ИЗ ЛЮБВИ К ПОРЯДКУ.
@@ -129,24 +155,52 @@ async def verify_password_bounded(plain: str, hashed: str) -> bool:
     app/api/v1/auth.py; тогда потолок надо переносить в общее хранилище,
     предварительно убедившись, что оно реально доступно.
 
+    ДОЛЯ НА КЛЮЧ (#2714). Слоты — общий котёл, и потолок исправно бил по своим:
+    пока флуд держал все четыре, легитимный вход с ВЕРНЫМ паролем получал 429
+    столько раз, сколько пытался. Поэтому *key* (у единственного вызывающего —
+    IP клиента) не берёт больше `_per_key_slot_cap()`: сколько бы один источник
+    ни слал, половина ёмкости остаётся тем, кто приходит впервые. Учёт по ключу
+    живёт ЗДЕСЬ ЖЕ и отдаётся тем же `_release_verify_slot` — инвариант «одна
+    точка выноса = одна точка учёта» не делится надвое.
+
+    Чего это НЕ делает, и это не оговорка ради приличия. Ключом может быть
+    только IP, а IP:
+      - подделывается, если между нами и клиентом окажется ещё один прокси
+        (сейчас доверенный хоп ровно один — Caddy, `ratelimit._client_ip` берёт
+        правый элемент XFF; появится второй — ключ станет клиентским вводом);
+      - разделяется: за NAT/корпоративным шлюзом вся организация приходит с
+        одного адреса и делит одну долю с чужим перебором;
+      - меняется: ботнет или ротация прокси дают злоумышленнику столько ключей,
+        сколько ему нужно, и доля на ключ перестаёт быть ограничением.
+    То есть это ПОДНИМАЕТ СТОИМОСТЬ атаки (одного адреса больше не хватает,
+    чтобы закрыть вход всем), но не закрывает её. Закрывают принципиально
+    только доказательство работы на входе или второй фактор — отдельный разговор
+    и отдельная цена.
+
     Raises:
         PasswordVerifyOverloadedError: очередь на проверку заполнена
-            (`login_password_verify_max_inflight`). Отказ мгновенный: ждать
-            нельзя, ждущий запрос держит соединение к БД.
+            (`login_password_verify_max_inflight`) ЛИБО *key* уже держит свою
+            долю (`_per_key_slot_cap`). Отказ мгновенный: ждать нельзя, ждущий
+            запрос держит соединение к БД. Оба случая неразличимы снаружи
+            намеренно — отказ приходит ДО сверки и потому ничего не сообщает о
+            том, существует ли учётка.
     """
     global _verify_inflight
 
     if _verify_inflight >= settings.login_password_verify_max_inflight:
         raise PasswordVerifyOverloadedError
+    if _verify_inflight_by_key.get(key, 0) >= _per_key_slot_cap():
+        raise PasswordVerifyOverloadedError
 
     loop = asyncio.get_running_loop()
     _verify_inflight += 1
+    _verify_inflight_by_key[key] = _verify_inflight_by_key.get(key, 0) + 1
     try:
         work = _VERIFY_POOL.submit(verify_password, plain, hashed)
     except BaseException:
         # Работа в пул НЕ встала — колбэка не будет, слот отдаём здесь. Иначе
         # утёкший слот навсегда отнимает у входа часть и без того малой ёмкости.
-        _verify_inflight -= 1
+        _release_verify_slot(key)
         raise
 
     # Слот освобождает ЗАВЕРШЕНИЕ РАБОТЫ, а не выход из этой корутины. Отмена
@@ -160,24 +214,35 @@ async def verify_password_bounded(plain: str, hashed: str) -> bool:
     # Именно поэтому колбэк висит на future ПУЛА, а не на обёртке из
     # `run_in_executor`: у обёртки «готово» наступает и при отмене — тест
     # `test_bounded_slot_freed_by_the_work_not_by_cancellation` ловит эту разницу.
-    work.add_done_callback(lambda _f: _schedule_verify_slot_release(loop))
+    work.add_done_callback(lambda _f: _schedule_verify_slot_release(loop, key))
     return await asyncio.wrap_future(work)
 
 
-def _schedule_verify_slot_release(loop: asyncio.AbstractEventLoop) -> None:
+def _schedule_verify_slot_release(loop: asyncio.AbstractEventLoop, key: str) -> None:
     """Возвращает слот по факту завершения работы в пуле (см. вызывающую).
 
-    Колбэк future пула исполняется В ПОТОКЕ ПУЛА, а счётчик — собственность
+    Колбэк future пула исполняется В ПОТОКЕ ПУЛА, а счётчики — собственность
     потока событийного цикла (на том и держится арифметика без лока), поэтому
     декремент переносим в цикл через `call_soon_threadsafe`.
     """
     try:
-        loop.call_soon_threadsafe(_release_verify_slot)
+        loop.call_soon_threadsafe(_release_verify_slot, key)
     except RuntimeError:
         # Цикл уже закрыт (остановка процесса) — освобождать нечего и некому.
         logger.debug("verify slot release skipped: event loop is closed")
 
 
-def _release_verify_slot() -> None:
+def _release_verify_slot(key: str) -> None:
+    """Единственное место, где слот отдают: и общий счётчик, и счётчик ключа.
+
+    Оба — одним движением и здесь же, а не по одному на каждом пути выхода:
+    разъедься они, и достаточно забыть одну строчку, чтобы ключ навсегда унёс
+    с собой долю ёмкости, которую никто уже не вернёт.
+    """
     global _verify_inflight
     _verify_inflight -= 1
+    left = _verify_inflight_by_key.get(key, 0) - 1
+    if left > 0:
+        _verify_inflight_by_key[key] = left
+    else:
+        _verify_inflight_by_key.pop(key, None)

tradein-mvp/backend/tests/conftest.py

@@ -1,13 +1,17 @@
 """Repo-wide test config for tradein-mvp/backend.
 
-Currently only registers custom pytest markers so they don't emit
-PytestUnknownMarkWarning when used (`--strict-markers` is not enabled in
-pyproject.toml, so an unregistered marker would only warn, not fail — this
-just keeps output clean and documents intent in one place).
+Регистрирует кастомные pytest-маркеры (иначе PytestUnknownMarkWarning:
+`--strict-markers` в pyproject.toml не включён, так что незарегистрированный
+маркер только предупреждал бы) и сторожит глобальное состояние, которое
+переживает отдельный тест, — см. `_no_leaked_password_verify_slots`.
 """
 
 from __future__ import annotations
 
+import sys
+
+import pytest
+
 
 def pytest_configure(config) -> None:
     config.addinivalue_line(
@@ -16,3 +20,40 @@ def pytest_configure(config) -> None:
         "Pango/cairo/GObject libs, self-skips where unavailable (see "
         "tests/test_pdf_real_render.py docstring for how to run it for real).",
     )
+
+
+@pytest.fixture(autouse=True)
+def _no_leaked_password_verify_slots():
+    """Тест не оставляет за собой занятых слотов проверки пароля (#2665, #2714).
+
+    Счётчики в `app.core.password` — состояние ПРОЦЕССА, а `pytest-asyncio` даёт
+    каждому тесту свой событийный цикл. Слот освобождает колбэк, посланный в
+    цикл через `call_soon_threadsafe`; если цикл к тому моменту закрыт,
+    `_schedule_verify_slot_release` ловит RuntimeError и слот не возвращается
+    никогда. На проде цикл живёт столько же, сколько процесс, и ветка
+    недостижима — а в тестах она копится молча и роняет НЕ ТОТ тест, который
+    её устроил: при пуле в 1 поток пары утечек хватает, чтобы всё дальнейшее
+    получало 429 «на ровном месте».
+
+    Поэтому проверка тут и общая: считаем слоты после каждого теста.
+
+    `sys.modules.get`, а не import: тестам, которые password.py не трогают
+    (большинство), незачем тянуть `Settings()` с его требованием DATABASE_URL.
+    """
+    yield
+
+    password_mod = sys.modules.get("app.core.password")
+    if password_mod is None:
+        return
+
+    inflight = password_mod._verify_inflight
+    by_key = dict(password_mod._verify_inflight_by_key)
+    # Сброс ДО assert: иначе одна утечка красит все последующие тесты и виновник
+    # теряется среди пострадавших.
+    password_mod._verify_inflight = 0
+    password_mod._verify_inflight_by_key.clear()
+
+    assert inflight == 0 and not by_key, (
+        f"тест оставил {inflight} занятых слотов проверки пароля (по ключам: {by_key}) — "
+        "утечка слота при пуле в 1 поток это вечный 429 всем на входе"
+    )

tradein-mvp/backend/tests/test_auth_api.py

@@ -847,10 +847,117 @@ async def test_login_flood_capped_by_rate_while_api_stays_responsive(
     # 3. Лишнее ОТКЛОНЯЕТСЯ, а не копится в очереди: очередь держала бы
     #    соединения к БД и выбрала бы пул (QueuePool 5+10).
     assert codes.count(429) > codes.count(401), "избыток должен получать 429, а не ждать"
-    # 4. Вход не заблокирован совсем: потолок — это темп, а не «ноль попыток».
+    # 4. Потолок режет ТЕМП, а не обнуляет попытки: до bcrypt доезжает хоть
+    #    что-то. Это утверждение ПРО АТАКУЮЩЕГО и ни слова не говорит о том,
+    #    войдёт ли в это время живой человек — он не войдёт, пока слоты заняты
+    #    флудом (#2714). Проверка «дверь открыта своим» — отдельным тестом ниже,
+    #    и мерит она вход С ДРУГОГО КЛЮЧА, а не число попыток атакующего.
     assert len(attempts) >= 2
 
 
+# ---------------------------------------------------------------------------
+# #2714 — потолок не должен бить по своим: доля слотов на один ключ
+# ---------------------------------------------------------------------------
+
+
+async def test_flood_from_one_ip_leaves_login_open_for_another_ip(
+    store: _Store, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Пока один адрес непрерывно флудит, человек с ДРУГОГО адреса входит (#2714).
+
+    Именно это ломал потолок #2665 в исходном виде: слоты — общий котёл, флуд
+    занимал все четыре, и легитимный вход с ВЕРНЫМ паролем получал 429 столько
+    раз, сколько пытался (замер в issue: 20 попыток → 20×429, 0×200).
+
+    Тест мерит ровно заявленное — ВХОД С ДРУГОГО КЛЮЧА, а не латентность одного
+    запроса и не число попыток атакующего: с общим котлом «попытки атакующего
+    доезжают» и «свой войдёт» — разные утверждения, и первое зелено, когда
+    второе ложно.
+
+    Число попыток пробы = `login_rate_limit`, и это не подгонка: столько входов
+    по паре (имя, IP) вообще разрешено за окно соседним `_LOGIN_LIMITER`.
+    Просить больше значило бы мерить ЕГО 429 вместо потолка сверок — то есть
+    получить красный тест на исправном коде.
+
+    Стенд БЕЗ `RateLimitMiddleware` (его в тестовом приложении нет), поэтому
+    флуд здесь плотнее, чем один адрес может выдать на проде (там его режут
+    300 запросов за 60с). Так и задумано: проверяем худший случай.
+    """
+    verify_s = 0.05
+    flood_ip = "203.0.113.66"
+    legit_ip = "198.51.100.10"
+    legit_password = "Secret123!"
+
+    patch_identity_sessions(monkeypatch, lambda: _FakeDB(store))
+    _capture_events(monkeypatch)
+    legit_hash = hash_password(legit_password)
+    store.add_user("realuser", legit_hash, role="employee")
+    app = _build_test_app(store)
+
+    def _slow_verify(plain: str, hashed: str) -> bool:
+        """Двойник bcrypt: столько же БЛОКИРУЮЩЕГО времени, только меньше.
+
+        Вердикт настоящий (а не всегда-False, как в тесте про темп выше) — без
+        него легитимный вход не дошёл бы до 200 и мерить было бы нечего.
+        """
+        time.sleep(verify_s)
+        return hashed == legit_hash and plain == legit_password
+
+    monkeypatch.setattr(password_mod, "verify_password", _slow_verify)
+
+    flood_codes: list[int] = []
+    flood_over = asyncio.Event()
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app), base_url="https://testserver"
+    ) as client:
+
+        async def flooder(n: int) -> None:
+            i = 0
+            while not flood_over.is_set():
+                i += 1
+                # Своё имя на каждую попытку — иначе флуд упрётся в
+                # `_LOGIN_LIMITER` (5 на пару имя+IP) и до потолка сверок не
+                # доедет вовсе: тест стал бы зелёным, ничего не проверив.
+                resp = await client.post(
+                    "/api/v1/auth/login",
+                    json={"username": f"nosuchuser{n}x{i}", "password": "guess"},
+                    headers={"x-forwarded-for": flood_ip},
+                )
+                flood_codes.append(resp.status_code)
+
+        floods = [asyncio.create_task(flooder(n)) for n in range(8)]
+        try:
+            # Ждём ДОКАЗАННОГО насыщения: 429 у атакующего = слоты кончились.
+            # Без этого условия проба могла бы пройти по пустой очереди и тест
+            # был бы зелёным на сломанном коде.
+            saturation_deadline = time.monotonic() + 10.0
+            while flood_codes.count(429) < 4:
+                assert time.monotonic() < saturation_deadline, (
+                    f"флуд не насытил слоты за 10с ({len(flood_codes)} ответов, "
+                    f"429: {flood_codes.count(429)}) — мерить справедливость не на чем"
+                )
+                await asyncio.sleep(0.01)
+
+            probe_codes: list[int] = []
+            for _ in range(config.settings.login_rate_limit):
+                resp = await client.post(
+                    "/api/v1/auth/login",
+                    json={"username": "realuser", "password": legit_password},
+                    headers={"x-forwarded-for": legit_ip},
+                )
+                probe_codes.append(resp.status_code)
+        finally:
+            flood_over.set()
+            await asyncio.gather(*floods)
+
+    assert probe_codes.count(200) == len(probe_codes), (
+        f"легитимный вход с {legit_ip} во время флуда с {flood_ip}: "
+        f"200={probe_codes.count(200)}, 429={probe_codes.count(429)}, "
+        f"401={probe_codes.count(401)} — потолок бьёт по своим (#2714)"
+    )
+
+
 # ---------------------------------------------------------------------------
 # POST /logout
 # ---------------------------------------------------------------------------

tradein-mvp/backend/tests/test_password.py

@@ -121,13 +121,27 @@ def test_verify_ceiling_defaults_stay_within_the_db_pool() -> None:
         "потолок перебора = workers/282мс. Подъём — осознанное решение "
         "«во сколько раз ускоряем перебор», а не рефакторинг: правь вместе с тестом"
     )
+    # ЛИТЕРАЛЫ, а не арифметика от настройки. Доля на ключ (#2714) считается как
+    # max_inflight // 2, и сторож вида `cap == max_inflight // 2` был бы
+    # тавтологией: подъём max_inflight до 64 он бы проспал, а вместе с ним —
+    # возврат к «один адрес занимает всё» (доля 32 при очереди в 4 живых слота
+    # ничего не делит). Поэтому здесь зафиксированы ОБА числа.
+    assert settings.login_password_verify_max_inflight == 4, (
+        "очередь 4 выбрана под QueuePool 5+10 и худшее ожидание 4/1×282мс ≈ 1.1с; "
+        "меняешь — пересчитывай и долю на ключ ниже"
+    )
+    assert password_mod._per_key_slot_cap() == 2, (
+        "один адрес держит не больше 2 слотов из 4: половина ёмкости обязана "
+        "оставаться тем, кто приходит впервые (#2714)"
+    )
 
 
 async def test_bounded_gives_same_answer_as_sync() -> None:
     """Обёртка не меняет вердикт — она меняет только ГДЕ он считается."""
     hashed = hash_password("correct horse battery staple")
-    assert await verify_password_bounded("correct horse battery staple", hashed) is True
-    assert await verify_password_bounded("wrong password", hashed) is False
+    key = "203.0.113.1"
+    assert await verify_password_bounded("correct horse battery staple", hashed, key=key) is True
+    assert await verify_password_bounded("wrong password", hashed, key=key) is False
 
 
 async def test_bounded_runs_off_the_event_loop_thread(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -145,7 +159,7 @@ async def test_bounded_runs_off_the_event_loop_thread(monkeypatch: pytest.Monkey
         return True
 
     monkeypatch.setattr(password_mod, "verify_password", _spy)
-    assert await verify_password_bounded("x", "y") is True
+    assert await verify_password_bounded("x", "y", key="203.0.113.1") is True
     assert seen and seen[0] != loop_thread
 
 
@@ -165,8 +179,13 @@ async def test_bounded_rejects_surplus_instead_of_queueing(monkeypatch: pytest.M
 
     monkeypatch.setattr(password_mod, "verify_password", _slow)
 
+    # У каждого запроса СВОЙ ключ: тест про ОБЩИЙ потолок, и отказывать здесь
+    # обязан именно он. С одним ключом на всех первым сработал бы лимит доли
+    # (#2714) — числа сошлись бы по другой причине, а поломка общего потолка
+    # осталась бы незамеченной.
     results = await asyncio.gather(
-        *(verify_password_bounded("x", "y") for _ in range(6)), return_exceptions=True
+        *(verify_password_bounded("x", "y", key=f"203.0.113.{i}") for i in range(6)),
+        return_exceptions=True,
     )
     rejected = [r for r in results if isinstance(r, PasswordVerifyOverloadedError)]
     admitted = [r for r in results if r is False]
@@ -174,7 +193,7 @@ async def test_bounded_rejects_surplus_instead_of_queueing(monkeypatch: pytest.M
     assert len(rejected) == 4, results
 
     # Слоты возвращаются: после отработки очереди вход снова доступен.
-    assert await verify_password_bounded("x", "y") is False
+    assert await verify_password_bounded("x", "y", key="203.0.113.9") is False
 
 
 async def test_bounded_slot_freed_by_the_work_not_by_cancellation(
@@ -198,7 +217,7 @@ async def test_bounded_slot_freed_by_the_work_not_by_cancellation(
 
     monkeypatch.setattr(password_mod, "verify_password", _blocked)
 
-    task = asyncio.create_task(verify_password_bounded("x", "y"))
+    task = asyncio.create_task(verify_password_bounded("x", "y", key="203.0.113.1"))
     await asyncio.to_thread(started.wait, 5)
 
     task.cancel()
@@ -206,12 +225,84 @@ async def test_bounded_slot_freed_by_the_work_not_by_cancellation(
         await task
 
     # Работа всё ещё занимает поток — слот занят, следующий получает отказ.
+    # Ключ ДРУГОЙ: отказ обязан прийти от общего потолка (max_inflight=1), а не
+    # от доли на ключ — иначе тест проверял бы не тот механизм.
     with pytest.raises(PasswordVerifyOverloadedError):
-        await verify_password_bounded("x", "y")
+        await verify_password_bounded("x", "y", key="203.0.113.2")
 
     finish.set()
     for _ in range(100):  # дать колбэку доехать до цикла
         await asyncio.sleep(0.01)
         if settings.login_password_verify_max_inflight > password_mod._verify_inflight:
             break
-    assert await verify_password_bounded("x", "y") is False
+    # Отменённая работа вернула И общий слот, И слот своего ключа: тот же адрес
+    # снова обслуживается (утечка по ключу при пуле в 1 поток была бы вечным
+    # отказом именно этому адресу и больше ничем себя не проявила).
+    assert await verify_password_bounded("x", "y", key="203.0.113.1") is False
+
+
+async def test_bounded_frees_slot_when_verify_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Исключение внутри сверки тоже возвращает слот — оба счётчика.
+
+    `verify_password` глотает ValueError/TypeError сама, так что сюда доезжает
+    только неожиданное (падение библиотеки, MemoryError). Пул из ОДНОГО потока
+    не прощает: один невозвращённый слот — вечный 429 всем на входе, и внешне
+    это выглядит не как ошибка bcrypt, а как «вход сломался неизвестно почему».
+    """
+
+    def _boom(plain: str, hashed: str) -> bool:
+        raise MemoryError("bcrypt died")
+
+    monkeypatch.setattr(password_mod, "verify_password", _boom)
+    with pytest.raises(MemoryError):
+        await verify_password_bounded("x", "y", key="10.0.0.3")
+
+    monkeypatch.setattr(password_mod, "verify_password", lambda plain, hashed: False)
+    assert await verify_password_bounded("x", "y", key="10.0.0.3") is False
+    assert password_mod._verify_inflight == 0
+    assert not password_mod._verify_inflight_by_key
+
+
+async def test_one_key_cannot_take_more_than_its_share(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Один ключ занимает не больше своей доли — остальным ёмкость остаётся (#2714).
+
+    Меряем именно ЭТО, а не латентность: с общим котлом слотов один источник
+    выбирал его целиком, и вход с другого адреса получал 429 бессрочно —
+    потолок темпа исправно работал против легитимных пользователей.
+    """
+    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 4)
+    assert password_mod._per_key_slot_cap() == 2  # 4 // 2 — исходные условия теста
+
+    finish = threading.Event()
+
+    def _blocked(plain: str, hashed: str) -> bool:
+        finish.wait(5)
+        return False
+
+    monkeypatch.setattr(password_mod, "verify_password", _blocked)
+
+    async def _wait_inflight(n: int) -> None:
+        deadline = time.monotonic() + 5
+        while password_mod._verify_inflight < n:
+            assert (
+                time.monotonic() < deadline
+            ), f"слотов занято {password_mod._verify_inflight} < {n}"
+            await asyncio.sleep(0.005)
+
+    flood = [
+        asyncio.create_task(verify_password_bounded("x", "y", key="10.0.0.1")) for _ in range(2)
+    ]
+    await _wait_inflight(2)
+
+    # Третий с ТОГО ЖЕ адреса — отказ, хотя два слота из четырёх свободны.
+    with pytest.raises(PasswordVerifyOverloadedError):
+        await verify_password_bounded("x", "y", key="10.0.0.1")
+
+    # А с другого — пускают. Задачу ставим ДО finish.set() и ждём, пока она
+    # займёт слот: иначе «пустили» означало бы только «флуд успел закончиться».
+    legit = asyncio.create_task(verify_password_bounded("x", "y", key="10.0.0.2"))
+    await _wait_inflight(3)
+
+    finish.set()
+    assert await legit is False, "вход с другого адреса обязан пройти во время флуда"
+    assert [await f for f in flood] == [False, False]