fix(server): reject over-deep JSON request bodies before parsing

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")

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")