File size: 3,113 Bytes
de8c765 | 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 | from fastapi import Request, Response
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from collections import defaultdict
import time
import asyncio
from Backend.core.logging import get_logger
logger = get_logger(__name__)
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
response.headers["Permissions-Policy"] = "geolocation=(self), camera=(self)"
if request.url.scheme == "https":
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
return response
class RateLimitMiddleware(BaseHTTPMiddleware):
def __init__(self, app, requests_per_minute: int = 60, burst_limit: int = 10):
super().__init__(app)
self.requests_per_minute = requests_per_minute
self.burst_limit = burst_limit
self.requests = defaultdict(list)
self.lock = asyncio.Lock()
async def dispatch(self, request: Request, call_next):
client_ip = request.client.host if request.client else "unknown"
current_time = time.time()
async with self.lock:
self.requests[client_ip] = [
t for t in self.requests[client_ip]
if current_time - t < 60
]
if len(self.requests[client_ip]) >= self.requests_per_minute:
logger.warning(f"Rate limit exceeded for {client_ip}")
return JSONResponse(
status_code=429,
content={"detail": "Too many requests. Please slow down."},
headers={"Retry-After": "60"}
)
recent_requests = [t for t in self.requests[client_ip] if current_time - t < 1]
if len(recent_requests) >= self.burst_limit:
logger.warning(f"Burst limit exceeded for {client_ip}")
return JSONResponse(
status_code=429,
content={"detail": "Too many requests. Please slow down."},
headers={"Retry-After": "1"}
)
self.requests[client_ip].append(current_time)
return await call_next(request)
class RequestValidationMiddleware(BaseHTTPMiddleware):
MAX_CONTENT_LENGTH = 50 * 1024 * 1024
async def dispatch(self, request: Request, call_next):
content_length = request.headers.get("content-length")
if content_length and int(content_length) > self.MAX_CONTENT_LENGTH:
return JSONResponse(
status_code=413,
content={"detail": "Request entity too large"}
)
return await call_next(request)
|