Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion scripts/host_timing_dogfood.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
# Operator-supplied cohort and comparison JSON must also be bounded before
# decoding; a valid but huge input cannot exhaust memory during parsing.
MAX_INPUT_JSON_BYTES = 64 * 1024
# Bound nesting independently of the JSON decoder's recursion strategy.
MAX_INPUT_JSON_DEPTH = 128
STRUCTURED = frozenset({
"run_process", "run_script", "run_shell", "run_skill_resource",
"cargo_check", "cargo_test", "cargo_fmt", "go_test",
Expand Down Expand Up @@ -73,10 +75,20 @@ def _read_bounded_json(path: Path) -> object:
if len(raw) > MAX_INPUT_JSON_BYTES:
raise ReportError("input JSON exceeds bounded size")
try:
return json.loads(raw.decode("utf-8"))
value = json.loads(raw.decode("utf-8"))
except (UnicodeError, ValueError, RecursionError) as exc:
# Deeply nested JSON can exhaust the parser stack below the byte cap.
raise ReportError("invalid input JSON") from exc
pending = [(value, 0)]
while pending:
item, depth = pending.pop()
if depth > MAX_INPUT_JSON_DEPTH:
raise ReportError("invalid input JSON")
if isinstance(item, dict):
pending.extend((child, depth + 1) for child in item.values())
elif isinstance(item, list):
pending.extend((child, depth + 1) for child in item)
return value


def _load_policy(path: Path) -> dict:
Expand Down
18 changes: 18 additions & 0 deletions scripts/tests/test_host_timing_dogfood.py
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,24 @@ def test_cli_malformed_json_is_rejected_with_static_error(self):
self.assertNotIn("private-secret", process.stderr)
self.assertNotIn(str(base), process.stderr)

def test_input_json_depth_has_an_explicit_inclusive_limit(self):
with tempfile.TemporaryDirectory() as root:
source = Path(root) / "input.json"
for kind in ("array", "object"):
for depth in (128, 129):
with self.subTest(kind=kind, depth=depth):
value = "private-marker"
for _ in range(depth - 1):
value = [value] if kind == "array" else {"value": value}
value = {"extra": value}
source.write_text(json.dumps(value), encoding="utf-8")
self.assertLess(source.stat().st_size, report.MAX_INPUT_JSON_BYTES)
if depth == 128:
self.assertEqual(report._read_bounded_json(source), value)
else:
with self.assertRaisesRegex(report.ReportError, "^invalid input JSON$"):
report._read_bounded_json(source)

def test_deeply_nested_json_reports_static_error_under_byte_limit(self):
# A bounded raw input can still exhaust json.loads parser recursion.
with tempfile.TemporaryDirectory() as root:
Expand Down
Loading