backend/rate_limiter.py

"""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)