Dmitry
600496048f
get_current_user навешивается на роутер целиком, публичными остаются только health и auth (список закреплён tests/api/test_route_auth.py). API не стартует с плейсхолдером или коротким JWT_SECRET без ALLOW_DEV_SECRET=1. Лимитер логинов ограничен по числу ключей. Клиентский IP берётся из последнего хопа X-Forwarded-For.
44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
"""Tiny in-memory sliding-window limiter for the login endpoint.
|
|
|
|
Single-process API, single user: no need for Redis. Caddy adds nothing here by
|
|
default, so this is the only brute-force protection — keep it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import deque
|
|
|
|
|
|
class SlidingWindowLimiter:
|
|
def __init__(self, limit: int, window_seconds: float = 60.0, max_keys: int = 10_000) -> None:
|
|
self.limit = limit
|
|
self.window = window_seconds
|
|
self.max_keys = max_keys
|
|
self._hits: dict[str, deque[float]] = {}
|
|
|
|
def hit(self, key: str) -> float | None:
|
|
"""Register an attempt. Returns seconds to wait if over the limit, else None."""
|
|
now = time.monotonic()
|
|
q = self._hits.get(key)
|
|
if q is None:
|
|
if len(self._hits) >= self.max_keys:
|
|
self._evict(now)
|
|
q = self._hits[key] = deque()
|
|
while q and now - q[0] > self.window:
|
|
q.popleft()
|
|
if len(q) >= self.limit:
|
|
return self.window - (now - q[0])
|
|
q.append(now)
|
|
return None
|
|
|
|
def reset(self, key: str) -> None:
|
|
self._hits.pop(key, None)
|
|
|
|
def _evict(self, now: float) -> None:
|
|
"""Make room for one more key: drop the expired ones, then the oldest if still full."""
|
|
for k in [k for k, q in self._hits.items() if not q or now - q[-1] > self.window]:
|
|
del self._hits[k]
|
|
if len(self._hits) >= self.max_keys:
|
|
del self._hits[next(iter(self._hits))] # dicts keep insertion order
|