mrleo1nidandClaude Sonnet 5.5
bbbcc6a5d8
- Rate-limit failed logins (8 per 15 min per IP or username, 429 + Retry-After) - Add ruff/mypy config and GitHub Actions workflow (ruff, pytest, tsc, tests, build) - Apply ruff autofixes (import order, pyupgrade) - Extract rpy_helpers.py, engine-util.ts, TranslationParts.tsx and translation-types.ts from oversized modules - Add dependency-free React UI tests - Refresh README and ARCHITECTURE (Naninovel/Tomie, status and plans) Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
import time
|
|
from collections import defaultdict, deque
|
|
|
|
from fastapi import Request
|
|
|
|
|
|
class FailureLimiter:
|
|
"""Sliding-window counter of failed attempts, kept in process memory.
|
|
|
|
Several uvicorn workers would each count separately, which only loosens the
|
|
limit; the service runs a single backend process.
|
|
"""
|
|
|
|
def __init__(self, max_failures: int, window_seconds: float) -> None:
|
|
self.max_failures = max_failures
|
|
self.window = window_seconds
|
|
self._hits: dict[str, deque[float]] = defaultdict(deque)
|
|
|
|
def _prune(self, key: str, now: float) -> deque[float]:
|
|
hits = self._hits[key]
|
|
while hits and now - hits[0] > self.window:
|
|
hits.popleft()
|
|
if not hits:
|
|
self._hits.pop(key, None)
|
|
return hits
|
|
|
|
def retry_after(self, *keys: str) -> int:
|
|
"""Seconds until a blocked key is allowed again, 0 if none is blocked."""
|
|
now = time.monotonic()
|
|
wait = 0.0
|
|
for key in keys:
|
|
hits = self._prune(key, now)
|
|
if len(hits) >= self.max_failures:
|
|
wait = max(wait, hits[0] + self.window - now)
|
|
return int(wait) + 1 if wait > 0 else 0
|
|
|
|
def fail(self, *keys: str) -> None:
|
|
now = time.monotonic()
|
|
for key in keys:
|
|
self._prune(key, now)
|
|
self._hits[key].append(now)
|
|
|
|
def reset(self, *keys: str) -> None:
|
|
for key in keys:
|
|
self._hits.pop(key, None)
|
|
|
|
|
|
def client_ip(request: Request) -> str:
|
|
forwarded = request.headers.get("x-forwarded-for")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
login_limiter = FailureLimiter(max_failures=8, window_seconds=15 * 60)
|