From f2b993bfddbfd5cb82bf251065709dba596f90a6 Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Tue, 18 Aug 2026 17:40:44 +0900
Subject: [PATCH] fix(server): reject over-deep JSON request bodies before
 parsing

Strix found that _coerce_json had no nesting-depth cap: a payload well
under max_body_bytes but deeply nested (JSON bomb) burns disproportionate
CPU during json.loads. Reject nesting past MAX_JSON_NESTING_DEPTH with a
cheap byte scan before parsing, closing the DoS path the fuzz target's
RecursionError handling only papered over.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 contextual_orchestrator/server.py     | 36 +++++++++++++++++++++
 tests/test_request_body_json_depth.py | 46 +++++++++++++++++++++++++++
 2 files changed, 82 insertions(+)
 create mode 100644 tests/test_request_body_json_depth.py

diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py
index c58d4cb79..ae1d82e8b 100644
--- a/contextual_orchestrator/server.py
+++ b/contextual_orchestrator/server.py
@@ -164,7 +164,43 @@ def _error_payload(error_code: str, error_message: str, error_detail: dict[str,
     }
 
 
+MAX_JSON_NESTING_DEPTH = 32
+
+
+def _reject_excessive_json_nesting(payload: bytes, max_depth: int = MAX_JSON_NESTING_DEPTH) -> None:
+    """Reject JSON with object/array nesting deeper than max_depth before parsing.
+
+    json.loads() has no built-in depth cap, so a deeply nested payload well
+    under max_body_bytes can still burn disproportionate CPU/stack during
+    parsing (JSON-bomb DoS). Structural brackets are always single ASCII
+    bytes and UTF-8 continuation/lead bytes are always >= 0x80, so a raw
+    byte scan that only toggles on an unescaped '"' is safe without decoding.
+    """
+    depth = 0
+    in_string = False
+    escaped = False
+    for byte in payload:
+        char = chr(byte)
+        if in_string:
+            if escaped:
+                escaped = False
+            elif char == "\\":
+                escaped = True
+            elif char == '"':
+                in_string = False
+            continue
+        if char == '"':
+            in_string = True
+        elif char in "{[":
+            depth += 1
+            if depth > max_depth:
+                raise RequestError(400, "invalid_json", "request body JSON nesting exceeds the allowed depth")
+        elif char in "}]":
+            depth -= 1
+
+
 def _coerce_json(payload: bytes) -> dict[str, Any]:
+    _reject_excessive_json_nesting(payload)
     value = json.loads(payload.decode("utf-8"))
     if not isinstance(value, dict):
         raise RequestError(400, "invalid_json", "request body must be a JSON object")
diff --git a/tests/test_request_body_json_depth.py b/tests/test_request_body_json_depth.py
new file mode 100644
index 000000000..ec49ce19c
--- /dev/null
+++ b/tests/test_request_body_json_depth.py
@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from contextual_orchestrator.server import MAX_JSON_NESTING_DEPTH, RequestError, _coerce_json  # noqa: E402
+
+
+def _nested_json_bomb(depth: int) -> bytes:
+    return (b'{"a":' * depth) + b"1" + (b"}" * depth)
+
+
+def test_shallow_json_is_accepted() -> None:
+    body = _coerce_json(b'{"a": {"b": {"c": 1}}}')
+    assert body == {"a": {"b": {"c": 1}}}
+
+
+def test_excessive_nesting_is_rejected_before_parsing() -> None:
+    bomb = _nested_json_bomb(MAX_JSON_NESTING_DEPTH + 1)
+    try:
+        _coerce_json(bomb)
+    except RequestError as exc:
+        assert exc.code == "invalid_json"
+    else:
+        raise AssertionError("expected RequestError for over-deep JSON nesting")
+
+
+def test_nesting_at_the_limit_is_accepted() -> None:
+    payload = _nested_json_bomb(MAX_JSON_NESTING_DEPTH)
+    _coerce_json(payload)  # must not raise
+
+
+def test_braces_inside_strings_do_not_count_toward_depth() -> None:
+    value = "{" * (MAX_JSON_NESTING_DEPTH + 5)
+    body = _coerce_json(('{"a": "%s"}' % value).encode("utf-8"))
+    assert body["a"] == value
+
+
+if __name__ == "__main__":  # pragma: no cover
+    test_shallow_json_is_accepted()
+    test_excessive_nesting_is_rejected_before_parsing()
+    test_nesting_at_the_limit_is_accepted()
+    test_braces_inside_strings_do_not_count_toward_depth()
+    print("ok")