File size: 3,780 Bytes
5fd7795 | 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 | """
Production Security & Observability Middlewares for CropAI
Includes Request-ID tracing, latency tracking, rate limiting, and security headers.
"""
import time
import uuid
import logging
from collections import defaultdict
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response, JSONResponse
from .config import settings
logger = logging.getLogger("cropai.middleware")
class RequestTracingMiddleware(BaseHTTPMiddleware):
"""Generates unique X-Request-ID and tracks request latency."""
async def dispatch(self, request: Request, call_next):
req_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
start_time = time.perf_counter()
# Attach request ID to request state
request.state.request_id = req_id
try:
response: Response = await call_next(request)
except Exception as exc:
duration_ms = (time.perf_counter() - start_time) * 1000.0
logger.error(
f"[{req_id}] Unhandled exception in {request.method} {request.url.path} "
f"({duration_ms:.2f}ms): {exc}",
exc_info=True
)
return JSONResponse(
status_code=500,
content={
"success": False,
"error": "Internal Server Error",
"request_id": req_id,
"detail": "An unexpected error occurred. Please try again later."
},
headers={"X-Request-ID": req_id}
)
duration_ms = (time.perf_counter() - start_time) * 1000.0
response.headers["X-Request-ID"] = req_id
response.headers["X-Process-Time-Ms"] = f"{duration_ms:.2f}"
# Security Headers
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
logger.info(f"[{req_id}] {request.method} {request.url.path} -> {response.status_code} ({duration_ms:.2f}ms)")
return response
class InMemoryRateLimiterMiddleware(BaseHTTPMiddleware):
"""
Sliding window in-memory rate limiter per IP address to safeguard Cloud Run
against resource exhaustion and denial-of-service.
"""
def __init__(self, app, max_requests_per_minute: int = 120):
super().__init__(app)
self.max_requests = max_requests_per_minute
self.window_seconds = 60.0
self._requests = defaultdict(list)
async def dispatch(self, request: Request, call_next):
# Exempt health checks and documentation endpoints
path = request.url.path
if path in ["/health", "/docs", "/openapi.json", "/redoc", "/api/v1/status"]:
return await call_next(request)
client_ip = request.client.host if request.client else "unknown"
now = time.time()
# Clean old timestamps
history = self._requests[client_ip]
self._requests[client_ip] = [t for t in history if now - t < self.window_seconds]
if len(self._requests[client_ip]) >= self.max_requests:
logger.warning(f"Rate limit exceeded for IP: {client_ip} ({len(self._requests[client_ip])} reqs/min)")
return JSONResponse(
status_code=429,
content={
"success": False,
"error": "Too Many Requests",
"detail": f"Rate limit of {self.max_requests} requests per minute exceeded. Please slow down."
},
headers={"Retry-After": "30"}
)
self._requests[client_ip].append(now)
return await call_next(request)
|