Issue 70 asgi body cap

oxyroute/asgi.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import asyncio
+import os
 from collections.abc import Callable
 from typing import Any
 
@@ -246,6 +247,23 @@ def response_empty(self, status: int, headers: list) -> None:
         )
 
 
+def _max_body_bytes() -> int:
+    # Keep semantics aligned with Rust path (`src/form.rs`): default 8 MiB, `0` means unlimited.
+    default = 8 * 1024 * 1024
+    raw = os.getenv("OXYROUTE_MAX_BODY_BYTES", "").strip()
+    if not raw:
+        return default
+    try:
+        n = int(raw)
+    except ValueError:
+        return default
+    if n == 0:
+        return 2**63 - 1
+    if n < 0:
+        return default
+    return n
+
+
 async def asgi_to_rsgi(
     app_rsgi: Callable[[Any, Any], Any],
     app_ws: Callable[[dict[str, Any], Any, Any], Any] | None,
@@ -262,12 +280,32 @@ async def asgi_to_rsgi(
         return
     if st != "http":
         return
-    body = b""
+    max_body = _max_body_bytes()
+    body = bytearray()
     while True:
         m = await receive()
         t = m.get("type", "")
         if t == "http.request":
-            body += m.get("body", b"")
+            chunk = m.get("body", b"")
+            if chunk:
+                body.extend(chunk)
+                if len(body) > max_body:
+                    await send(
+                        {
+                            "type": "http.response.start",
+                            "status": 413,
+                            "headers": [
+                                (b"content-type", b"application/json; charset=utf-8"),
+                            ],
+                        }
+                    )
+                    await send(
+                        {
+                            "type": "http.response.body",
+                            "body": b'{"error":"payload too large"}',
+                        }
+                    )
+                    return
             if not m.get("more_body", False):
                 break
         elif t == "http.disconnect":
@@ -297,7 +335,7 @@ async def _drain_outgoing() -> None:
             await send(msg)
 
     drain_task = asyncio.create_task(_drain_outgoing())
-    proto = _RsgiProtocol(body, queue, loop)
+    proto = _RsgiProtocol(bytes(body), queue, loop)
     run_exc: BaseException | None = None
     try:
         await loop.run_in_executor(

tests/test_app_units.py

@@ -0,0 +1,50 @@
+"""Extra App wrapper branch coverage."""
+
+from __future__ import annotations
+
+import asyncio
+
+from oxyroute.app import App, Depends, _norm_dependencies, _unwrap_dep
+
+
+def test_depends_and_unwrap_helpers() -> None:
+    def dep() -> int:
+        return 1
+
+    d = Depends(dep)
+    assert _unwrap_dep(d) is dep
+    assert _unwrap_dep(dep) is dep
+    assert _norm_dependencies(None) is None
+    assert _norm_dependencies([]) is None
+    norm = _norm_dependencies([("x", d)])
+    assert norm is not None
+    assert norm[0][0] == "x"
+    assert norm[0][1] is dep
+
+
+def test_app_freeze_and_base_rsgi_noops() -> None:
+    app = App()
+    app.freeze()
+    assert asyncio.run(app.__rsgi_init__()) is None
+    assert asyncio.run(app.__rsgi_del__()) is None
+
+
+def test_put_and_delete_decorators_register_routes() -> None:
+    app = App()
+
+    @app.put("/u")
+    def upd() -> str:
+        return "ok"
+
+    @app.delete("/u")
+    def rem() -> str:
+        return "ok"
+
+    assert upd() == "ok"
+    assert rem() == "ok"
+
+
+def test_rsgi_forwarder_methods_exist() -> None:
+    app = App()
+    assert callable(app.handle_rsgi)
+    assert callable(app.__rsgi__)

tests/test_asgi.py

@@ -117,3 +117,107 @@ async def _run() -> None:
         asyncio.run(_run())
     finally:
         asgi_mod._run_handle_rsgi_blocking = original
+
+
+def test_asgi_chunked_request_body_joins_without_quadratic_path_assumption() -> None:
+    app = App()
+
+    @app.patch("/x")
+    def patch_x(json: dict) -> str:
+        return f"p:{json.get('a')}"
+
+    scope = {
+        "type": "http",
+        "http_version": "1.1",
+        "scheme": "http",
+        "method": "PATCH",
+        "path": "/x",
+        "query_string": b"",
+        "headers": [(b"content-type", b"application/json")],
+    }
+
+    incoming = iter(
+        [
+            {"type": "http.request", "body": b'{"a":', "more_body": True},
+            {"type": "http.request", "body": b"7}", "more_body": False},
+        ]
+    )
+    sent: list[dict] = []
+
+    async def receive() -> dict:
+        return next(incoming)
+
+    async def send(message: dict) -> None:
+        sent.append(message)
+
+    async def _run() -> None:
+        await app(scope, receive, send)
+
+    asyncio.run(_run())
+    assert sent[0]["type"] == "http.response.start"
+    assert sent[0]["status"] == 200
+    assert sent[1]["type"] == "http.response.body"
+    assert sent[1]["body"] == b"p:7"
+
+
+def test_asgi_body_limit_enforced_during_chunk_read() -> None:
+    app = App()
+
+    @app.post("/x")
+    def x(body: bytes) -> str:
+        return body.decode("utf-8")
+
+    scope = {
+        "type": "http",
+        "http_version": "1.1",
+        "scheme": "http",
+        "method": "POST",
+        "path": "/x",
+        "query_string": b"",
+        "headers": [],
+    }
+
+    incoming = iter(
+        [
+            {"type": "http.request", "body": b"abc", "more_body": True},
+            {"type": "http.request", "body": b"de", "more_body": False},
+        ]
+    )
+    sent: list[dict] = []
+
+    async def receive() -> dict:
+        return next(incoming)
+
+    async def send(message: dict) -> None:
+        sent.append(message)
+
+    original = asgi_mod._max_body_bytes
+    asgi_mod._max_body_bytes = lambda: 4
+
+    async def _run() -> None:
+        await app(scope, receive, send)
+
+    try:
+        asyncio.run(_run())
+    finally:
+        asgi_mod._max_body_bytes = original
+
+    assert sent[0]["type"] == "http.response.start"
+    assert sent[0]["status"] == 413
+    assert sent[1]["type"] == "http.response.body"
+    assert sent[1]["body"] == b'{"error":"payload too large"}'
+
+
+def test_asgi_non_http_non_websocket_scope_is_ignored() -> None:
+    app = App()
+    sent: list[dict] = []
+
+    async def receive() -> dict:
+        return {"type": "lifespan.startup"}
+
+    async def send(message: dict) -> None:
+        sent.append(message)
+
+    scope = {"type": "lifespan"}
+    asyncio.run(app(scope, receive, send))
+    assert sent == []