肉末
175a4ad0d1
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>
41 lines
977 B
Python
41 lines
977 B
Python
"""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
|