File size: 1,927 Bytes
c794b6b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | """In-memory sliding-window rate limiter for abuse-sensitive endpoints."""
from __future__ import annotations
import time
from collections import defaultdict, deque
from typing import Deque, DefaultDict
from fastapi import HTTPException, Request, status
class RateLimiter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self._hits: DefaultDict[str, Deque[float]] = defaultdict(deque)
def _key(self, request: Request, bucket: str) -> str:
client = request.client.host if request.client else "unknown"
return f"{bucket}:{client}"
def check(self, request: Request, bucket: str) -> None:
now = time.time()
key = self._key(request, bucket)
q = self._hits[key]
while q and now - q[0] > self.window_seconds:
q.popleft()
if len(q) >= self.max_requests:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Too many requests. Try again shortly.",
)
q.append(now)
def check_ws(self, websocket, bucket: str) -> None:
"""Rate limit WebSocket actions by client host."""
now = time.time()
client = websocket.client.host if websocket.client else "ws-unknown"
key = f"{bucket}:ws:{client}"
q = self._hits[key]
while q and now - q[0] > self.window_seconds:
q.popleft()
if len(q) >= self.max_requests:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Too many requests. Try again shortly.",
)
q.append(now)
login_limiter = RateLimiter(max_requests=20, window_seconds=60)
refresh_limiter = RateLimiter(max_requests=30, window_seconds=60)
sos_limiter = RateLimiter(max_requests=10, window_seconds=60)
|