feat: per-IP rate limit and write bans

Apply sliding-window 2/s limits and BannedIP checks on write routes.
Also force SVG attachment disposition and claim burn-after-read in DB
before streaming, deleting the file via BackgroundTask after the response.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
肉末
2026-07-14 12:52:45 +08:00
parent f2a1f3a1d6
commit 175a4ad0d1
6 changed files with 197 additions and 19 deletions
+40
View File
@@ -0,0 +1,40 @@
"""In-memory sliding-window rate limit (per IP)."""
from __future__ import annotations
import threading
import time
from collections import defaultdict, deque
from app import config
_lock = threading.Lock()
_hits: dict[str, deque[float]] = defaultdict(deque)
def reset() -> None:
"""Clear all tracked hits (tests)."""
with _lock:
_hits.clear()
def allow(ip: str, *, now: float | None = None) -> bool:
"""Return True if ``ip`` may proceed; record the hit when allowed.
Allows when the number of hits in the last 1.0s is strictly less than
``settings.rate_limit_per_second``.
"""
if not ip:
ip = ""
limit = config.settings.rate_limit_per_second
ts = time.monotonic() if now is None else now
window_start = ts - 1.0
with _lock:
q = _hits[ip]
while q and q[0] <= window_start:
q.popleft()
if len(q) >= limit:
return False
q.append(ts)
return True