Spaces:
Sleeping
Sleeping
File size: 4,456 Bytes
b84ea83 | 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 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """
Rate Limiter Middleware — IP-based, in-memory
Protects API endpoints without authentication.
Uses a sliding-window approach per client IP.
Configuration via .env:
RATE_LIMIT_ENABLED=True
RATE_LIMIT_REQUESTS_PER_MINUTE=60
RATE_LIMIT_REQUESTS_PER_HOUR=1000
"""
import os
import time
import logging
from collections import defaultdict
from fastapi import Request
from fastapi.responses import JSONResponse
logger = logging.getLogger(__name__)
# ── Configuration ──
RATE_LIMIT_ENABLED = os.getenv("RATE_LIMIT_ENABLED", "True").lower() == "true"
REQUESTS_PER_MINUTE = int(os.getenv("RATE_LIMIT_REQUESTS_PER_MINUTE", "60"))
REQUESTS_PER_HOUR = int(os.getenv("RATE_LIMIT_REQUESTS_PER_HOUR", "1000"))
# ── In-Memory Store ──
# Structure: { ip: [timestamp1, timestamp2, ...] }
_request_log: dict[str, list[float]] = defaultdict(list)
# Paths excluded from rate limiting
EXCLUDED_PATHS = {"/", "/docs", "/redoc", "/openapi.json", "/api/v1/health"}
def _get_client_ip(request: Request) -> str:
"""Extract client IP, respecting X-Forwarded-For header."""
forwarded = request.headers.get("X-Forwarded-For")
if forwarded:
return forwarded.split(",")[0].strip()
return request.client.host if request.client else "unknown"
def _cleanup_old_entries(ip: str, window_seconds: int) -> None:
"""Remove timestamps older than the given window."""
cutoff = time.time() - window_seconds
_request_log[ip] = [t for t in _request_log[ip] if t > cutoff]
async def rate_limit_middleware(request: Request, call_next):
"""
FastAPI middleware that enforces per-IP rate limits.
Returns 429 Too Many Requests when limits are exceeded.
Adds X-RateLimit-* headers to every response.
"""
if not RATE_LIMIT_ENABLED:
return await call_next(request)
# Skip rate limiting for docs and health
if request.url.path in EXCLUDED_PATHS:
return await call_next(request)
client_ip = _get_client_ip(request)
now = time.time()
# Record this request
_request_log[client_ip].append(now)
# ── Check per-minute limit ──
_cleanup_old_entries(client_ip, 60)
requests_last_minute = len(_request_log[client_ip])
if requests_last_minute > REQUESTS_PER_MINUTE:
logger.warning(
f"Rate limit exceeded (per-minute) for IP: {client_ip} "
f"({requests_last_minute}/{REQUESTS_PER_MINUTE})"
)
return JSONResponse(
status_code=429,
content={
"status": "error",
"error": {
"code": "RATE_LIMIT_EXCEEDED",
"message": "Too many requests. Please slow down.",
"details": {
"limit": REQUESTS_PER_MINUTE,
"window": "1 minute",
"retry_after_seconds": 60,
},
},
},
headers={"Retry-After": "60"},
)
# ── Check per-hour limit ──
_cleanup_old_entries(client_ip, 3600)
requests_last_hour = len(_request_log[client_ip])
if requests_last_hour > REQUESTS_PER_HOUR:
logger.warning(
f"Rate limit exceeded (per-hour) for IP: {client_ip} "
f"({requests_last_hour}/{REQUESTS_PER_HOUR})"
)
return JSONResponse(
status_code=429,
content={
"status": "error",
"error": {
"code": "RATE_LIMIT_EXCEEDED",
"message": "Hourly request limit reached. Please try again later.",
"details": {
"limit": REQUESTS_PER_HOUR,
"window": "1 hour",
"retry_after_seconds": 3600,
},
},
},
headers={"Retry-After": "3600"},
)
# ── Proceed with request ──
response = await call_next(request)
# Add rate limit headers to response
response.headers["X-RateLimit-Limit-Minute"] = str(REQUESTS_PER_MINUTE)
response.headers["X-RateLimit-Remaining-Minute"] = str(
max(0, REQUESTS_PER_MINUTE - requests_last_minute)
)
response.headers["X-RateLimit-Limit-Hour"] = str(REQUESTS_PER_HOUR)
response.headers["X-RateLimit-Remaining-Hour"] = str(
max(0, REQUESTS_PER_HOUR - requests_last_hour)
)
return response
|