Spaces:
Runtime error
Runtime error
| """Bounded ASGI request handling and process-local submission throttling.""" | |
| from __future__ import annotations | |
| import logging | |
| import math | |
| import re | |
| import time | |
| from collections import OrderedDict, deque | |
| from ipaddress import ip_address, ip_network | |
| from threading import Lock | |
| from uuid import uuid4 | |
| from starlette.datastructures import Headers, MutableHeaders | |
| from starlette.responses import JSONResponse | |
| from starlette.types import ASGIApp, Message, Receive, Scope, Send | |
| from app.api.logging import request_id_context | |
| from app.config import Settings | |
| logger = logging.getLogger("vera.requests") | |
| REQUEST_ID_PATTERN = re.compile(r"[A-Za-z0-9._-]{1,64}\Z") | |
| class SlidingWindowLimiter: | |
| """Bound memory and requests per client; restart/multiple-process limits are documented.""" | |
| def __init__(self, limit: int, window: int, max_clients: int) -> None: | |
| self.limit, self.window, self.max_clients = limit, window, max_clients | |
| self.clients: OrderedDict[str, deque[float]] = OrderedDict() | |
| self.lock = Lock() | |
| def check(self, client: str, now: float | None = None) -> int: | |
| """Return zero when allowed, otherwise seconds until a retry may be accepted.""" | |
| now = time.monotonic() if now is None else now | |
| with self.lock: | |
| while self.clients and next(iter(self.clients.values()))[-1] <= now - self.window: | |
| self.clients.popitem(last=False) | |
| bucket = self.clients.get(client) | |
| if bucket is None: | |
| if len(self.clients) >= self.max_clients: | |
| return self.window | |
| bucket = self.clients[client] = deque() | |
| while bucket and bucket[0] <= now - self.window: | |
| bucket.popleft() | |
| if len(bucket) >= self.limit: | |
| return max(1, math.ceil(bucket[0] + self.window - now)) | |
| bucket.append(now) | |
| self.clients.move_to_end(client) | |
| return 0 | |
| class RequestGuard: | |
| """Apply correlation IDs, body bounds, security headers, and POST rate limits.""" | |
| def __init__(self, app: ASGIApp, settings: Settings) -> None: | |
| self.app = app | |
| self.settings = settings | |
| self.limiter = SlidingWindowLimiter( | |
| settings.REVIEW_RATE_LIMIT, | |
| settings.REVIEW_RATE_WINDOW_SECONDS, | |
| settings.RATE_LIMIT_MAX_CLIENTS, | |
| ) | |
| self.proxies = [ | |
| ip_network(value.strip(), strict=False) | |
| for value in settings.TRUSTED_PROXY_CIDRS.split(",") | |
| if value.strip() | |
| ] | |
| def client_key(self, scope: Scope, headers: Headers) -> str: | |
| """Strip only explicitly trusted proxy hops from the right of a forwarded chain.""" | |
| peer = (scope.get("client") or ("unknown", 0))[0] | |
| try: | |
| address = ip_address(peer) | |
| if not any(address in network for network in self.proxies): | |
| return str(address) | |
| hops = headers.get("x-forwarded-for", "").split(",") | |
| if len(hops) > 20: | |
| return peer | |
| for hop in reversed(hops): | |
| address = ip_address(hop.strip()) | |
| if not any(address in network for network in self.proxies): | |
| return str(address) | |
| return str(address) | |
| except ValueError: | |
| return peer | |
| async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: | |
| if scope["type"] != "http": | |
| await self.app(scope, receive, send) | |
| return | |
| headers = Headers(scope=scope) | |
| supplied = headers.get("x-request-id", "") | |
| request_id = supplied if REQUEST_ID_PATTERN.fullmatch(supplied) else uuid4().hex | |
| token = request_id_context.set(request_id) | |
| scope.setdefault("state", {})["request_id"] = request_id | |
| started = time.monotonic() | |
| status_code = 500 | |
| response_started = False | |
| async def guarded_send(message: Message) -> None: | |
| nonlocal status_code, response_started | |
| if message["type"] == "http.response.start": | |
| response_started = True | |
| status_code = message["status"] | |
| outgoing = MutableHeaders(scope=message) | |
| outgoing["X-Request-ID"] = request_id | |
| outgoing["X-Content-Type-Options"] = "nosniff" | |
| outgoing["X-Frame-Options"] = "DENY" | |
| outgoing["Referrer-Policy"] = "no-referrer" | |
| outgoing["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()" | |
| outgoing["Cache-Control"] = "no-store" | |
| origin = headers.get("origin") | |
| if origin and ( | |
| origin in self.settings.cors_origins or "*" in self.settings.cors_origins | |
| ): | |
| outgoing["Access-Control-Allow-Origin"] = origin | |
| outgoing["Access-Control-Expose-Headers"] = ( | |
| "X-Total-Count, X-Request-ID, Retry-After" | |
| ) | |
| outgoing.add_vary_header("Origin") | |
| if self.settings.is_production: | |
| outgoing["Strict-Transport-Security"] = "max-age=31536000" | |
| await send(message) | |
| async def reject(code: int, detail: str, extra: dict[str, str] | None = None) -> None: | |
| await JSONResponse( | |
| {"detail": detail, "request_id": request_id}, status_code=code, headers=extra | |
| )(scope, receive, guarded_send) | |
| try: | |
| length = headers.get("content-length") | |
| if length is not None: | |
| if not length.isascii() or not length.isdigit() or len(length) > 10: | |
| await reject(400, "Invalid Content-Length header.") | |
| return | |
| if int(length) > self.settings.MAX_REQUEST_BODY_BYTES: | |
| await reject(413, "Request body is too large.") | |
| return | |
| if headers.get("content-encoding", "identity").lower() != "identity": | |
| await reject(415, "Compressed request bodies are not supported.") | |
| return | |
| if scope["method"] == "POST" and scope["path"].rstrip("/") == "/api/reviews": | |
| retry = self.limiter.check(self.client_key(scope, headers)) | |
| if retry: | |
| await reject( | |
| 429, | |
| "Too many reviews submitted. Please wait before trying again.", | |
| {"Retry-After": str(retry)}, | |
| ) | |
| return | |
| # Buffer at most the configured limit before any handler can persist side effects. | |
| body = bytearray() | |
| while True: | |
| message = await receive() | |
| if message["type"] == "http.disconnect": | |
| status_code = 499 | |
| return | |
| chunk = message.get("body", b"") | |
| if len(body) + len(chunk) > self.settings.MAX_REQUEST_BODY_BYTES: | |
| await reject(413, "Request body is too large.") | |
| return | |
| body.extend(chunk) | |
| if not message.get("more_body", False): | |
| break | |
| delivered = False | |
| async def replay() -> Message: | |
| nonlocal delivered | |
| if not delivered: | |
| delivered = True | |
| return {"type": "http.request", "body": bytes(body), "more_body": False} | |
| return await receive() | |
| await self.app(scope, replay, guarded_send) | |
| except Exception as exc: | |
| logger.error("request_failed", extra={"exception_type": type(exc).__name__}) | |
| if response_started: | |
| raise | |
| await reject(500, "The service could not complete this request.") | |
| finally: | |
| route = getattr(scope.get("route"), "path", "unmatched") | |
| logger.info( | |
| "request_completed", | |
| extra={ | |
| "method": scope["method"], | |
| "route": route, | |
| "status_code": status_code, | |
| "duration_ms": round((time.monotonic() - started) * 1000, 2), | |
| }, | |
| ) | |
| request_id_context.reset(token) | |