Cap request body size (HTTP 413)

LimitRequestBodyMiddleware (pure ASGI, registered outermost) rejects request
bodies larger than PIRATS_MAX_BODY_BYTES (default 1 MiB) before they're buffered
into memory: it checks the declared Content-Length first, then counts the bytes
actually streamed so a chunked/length-omitting client can't bypass the header
check. Exposed as services.pirats.maxBodyBytes and documented in the README.

Tested in isolation (Content-Length fast path + streamed path) and through the
real app.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-15 15:59:48 -07:00
parent 122e66b817
commit 692c6d26c1
5 changed files with 140 additions and 1 deletions

View File

@@ -2525,3 +2525,47 @@ def test_create_and_join_sanitize_names():
assert "\x01" not in player.name
finally:
app.dependency_overrides.clear()
def test_request_body_size_limit_middleware():
from fastapi import FastAPI, Request
from fastapi.testclient import TestClient
from pirats.main import LimitRequestBodyMiddleware
app = FastAPI()
app.add_middleware(LimitRequestBodyMiddleware, max_bytes=1000)
@app.post("/echo")
async def echo(request: Request):
body = await request.body()
return {"len": len(body)}
client = TestClient(app)
# Under the limit: passes through and the handler sees the full body.
r = client.post("/echo", content=b"x" * 500)
assert r.status_code == 200
assert r.json()["len"] == 500
# Over the limit with a declared Content-Length: rejected up front.
r = client.post("/echo", content=b"x" * 5000)
assert r.status_code == 413
# Over the limit via a streamed (length-omitting) body: rejected once the
# counted bytes exceed the cap.
def oversized_chunks():
for _ in range(10):
yield b"x" * 500 # 5000 bytes total
r = client.post("/echo", content=oversized_chunks())
assert r.status_code == 413
def test_request_body_size_limit_on_real_app():
from fastapi.testclient import TestClient
from pirats.main import app, DEFAULT_MAX_BODY_BYTES
# No DB override needed: the middleware rejects before routing/dependencies.
client = TestClient(app)
r = client.post("/api/game", content=b"x" * (DEFAULT_MAX_BODY_BYTES + 1))
assert r.status_code == 413