| """ |
| 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() |
|
|
| |
| 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}" |
|
|
| |
| 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): |
| |
| 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() |
|
|
| |
| 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) |
|
|
