29 lines
1.1 KiB
Python
29 lines
1.1 KiB
Python
"""Limit POST /api/v1/events body size (matches nginx client_max_body_size)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
|
|
class IngestBodySizeLimitMiddleware(BaseHTTPMiddleware):
|
|
def __init__(self, app, *, max_bytes: int = 2_097_152) -> None:
|
|
super().__init__(app)
|
|
self._max_bytes = max_bytes
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
if request.method == "POST" and request.url.path.rstrip("/") == "/api/v1/events":
|
|
content_length = request.headers.get("content-length")
|
|
if content_length:
|
|
try:
|
|
size = int(content_length)
|
|
except ValueError:
|
|
size = 0
|
|
if size > self._max_bytes:
|
|
return JSONResponse(
|
|
status_code=413,
|
|
content={"detail": f"Request body too large (max {self._max_bytes} bytes)"},
|
|
)
|
|
return await call_next(request)
|