NotBigGhostandClaude Opus 4.8
038a788f99
Вход игрока и админа получают лимит login-user/admin-login-user, не зависящий от IP: ротация X-Forwarded-For (#58) больше не снимает защиту полностью. Успешный вход сбрасывает счётчики аккаунта. В docstring ratelimit — про сброс при рестарте и необходимость внешнего стора при нескольких воркерах. #60 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XfTsytzT6TojfmprRDKiV6
80 lines
3.9 KiB
Python
80 lines
3.9 KiB
Python
"""Ограничение неудачных попыток ввода пароля (защита от перебора).
|
|
|
|
Счётчики живут в памяти процесса — как и SSE-шина, это рассчитано на один воркер uvicorn
|
|
(`--workers 1`, см. entrypoint.sh/run.*). Перезапуск их обнуляет; для окна в 15 минут это
|
|
приемлемо. Ограничения устойчивости (#60):
|
|
* рестарт (в т.ч. деплой) сбрасывает окно — злоумышленник получает новую квоту после
|
|
перезапуска, но окно короткое, а рестарты редки;
|
|
* при уходе от одного воркера лимит делится между процессами (каждый считает своё) —
|
|
тогда счётчики нужно вынести во внешний стор (Redis pub/sub, как отмечено в CLAUDE.md
|
|
про SSE-шину), общий для всех воркеров.
|
|
Помимо пары «IP + логин» и лимита по IP есть IP-независимый лимит на аккаунт
|
|
(`login-user:*` / `admin-login-user:*`), чтобы ротация X-Forwarded-For / многих адресов
|
|
(#58) не снимала защиту полностью.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from collections import deque
|
|
|
|
from app.core.errors import TooManyAttemptsError
|
|
|
|
_WINDOW_SECONDS = 15 * 60
|
|
# Выше этого числа ключей при записи неудачи вычищаем протухшие, чтобы поток попыток
|
|
# с разных адресов не копил память бесконечно.
|
|
_PRUNE_ABOVE = 10_000
|
|
|
|
|
|
class LoginThrottle:
|
|
"""Скользящее окно неудач по произвольным ключам (IP, пара «IP + логин», id игрока).
|
|
|
|
Роуты синхронные и выполняются в пуле потоков, поэтому доступ под замком."""
|
|
|
|
def __init__(self, window_seconds: int = _WINDOW_SECONDS) -> None:
|
|
self.window = window_seconds
|
|
self._fails: dict[str, deque[float]] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def _recent(self, key: str, now: float) -> deque[float]:
|
|
attempts = self._fails.get(key)
|
|
if attempts is None:
|
|
return deque()
|
|
while attempts and now - attempts[0] >= self.window:
|
|
attempts.popleft()
|
|
if not attempts:
|
|
del self._fails[key]
|
|
return attempts
|
|
|
|
def check(self, limits: dict[str, int]) -> None:
|
|
"""Бросает TooManyAttemptsError, если хотя бы по одному ключу лимит исчерпан."""
|
|
now = time.monotonic()
|
|
with self._lock:
|
|
for key, limit in limits.items():
|
|
attempts = self._recent(key, now)
|
|
if len(attempts) >= limit:
|
|
# Попытка снова примется, когда из окна выпадет неудача, после которой
|
|
# в нём остаётся limit-1 записей.
|
|
frees_at = attempts[len(attempts) - limit] + self.window
|
|
raise TooManyAttemptsError(retry_after=int(frees_at - now) + 1)
|
|
|
|
def fail(self, keys: dict[str, int]) -> None:
|
|
now = time.monotonic()
|
|
with self._lock:
|
|
if len(self._fails) > _PRUNE_ABOVE:
|
|
for key in list(self._fails):
|
|
self._recent(key, now)
|
|
for key in keys:
|
|
self._fails.setdefault(key, deque()).append(now)
|
|
|
|
def reset(self, key: str) -> None:
|
|
with self._lock:
|
|
self._fails.pop(key, None)
|
|
|
|
def clear(self) -> None:
|
|
with self._lock:
|
|
self._fails.clear()
|
|
|
|
|
|
login_throttle = LoginThrottle()
|