code_runner: contain hostile result.json, v0.1.1

Deep or oversized values from the sandbox now fail that example only; compare refuses nesting past 64 levels.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWYGW5NZL5uVpnqmNKo

code_runner/code_runner/_compare.py

@@ -6,6 +6,24 @@ import json
 import math
 
 TOLERANCE = 1e-5  # LeetCode accepts floating answers within 1e-5
+MAX_DEPTH = 64  # real answers nest a handful of levels; deeper is hostile
+
+
+class TooDeepError(ValueError):
+    """A compared value nests deeper than ``MAX_DEPTH``."""
+
+
+def too_deep(value: object, limit: int = MAX_DEPTH) -> bool:
+    """Whether ``value`` nests lists/dicts deeper than ``limit`` (iterative)."""
+    stack: list[tuple[object, int]] = [(value, 0)]
+    while stack:
+        node, depth = stack.pop()
+        if isinstance(node, list | dict):
+            if depth >= limit:
+                return True
+            children = node if isinstance(node, list) else node.values()
+            stack.extend((child, depth + 1) for child in children)
+    return False
 
 
 def _as_number(value: object) -> float | None:
@@ -55,7 +73,12 @@ def matches(expected: object, actual: object, order: str) -> bool:
 
     Raises:
         ValueError: ``order`` is none of those.
+        TooDeepError: a value nests deeper than ``MAX_DEPTH`` (the recursive
+            walks below would blow the stack).
     """
+    if too_deep(expected) or too_deep(actual):
+        msg = f"value nested deeper than {MAX_DEPTH} levels"
+        raise TooDeepError(msg)
     if order == "exact":
         return _equal(expected, actual)
     if order == "any":

code_runner/code_runner/_run.py

@@ -8,7 +8,7 @@ from pathlib import Path
 import stat
 import tempfile
 
-from code_runner._compare import matches
+from code_runner._compare import TooDeepError, matches, too_deep
 from code_runner._languages import Language, get_language
 from code_runner._model import MAX_CAPTURE, CaseResult, Example, Limits
 from code_runner._sandbox import BoxResult, execute
@@ -16,6 +16,9 @@ from code_runner._starter import Entry, parse_starter
 
 Box = Callable[[list[str], Path, Limits, int, tuple[str, ...]], BoxResult]
 _OUT_OF_MEMORY = ("out of memory", "allocation failed")
+_MAX_ERROR = 200
+# Everything hostile JSON can raise on the host: deep nesting, huge numbers.
+_HOSTILE = (ValueError, TypeError, RecursionError, MemoryError, OverflowError)
 
 
 def _positional(example: Example, entry: Entry) -> list[object]:
@@ -41,9 +44,11 @@ def _read_result(scratch: Path) -> dict[str, object] | None:
         raw = handle.read(MAX_CAPTURE * 16)
     try:
         payload = json.loads(raw)
-    except ValueError:
+    except _HOSTILE:  # ValueError, or RecursionError on "[" * 200000
         return None
-    return payload if isinstance(payload, dict) and "ok" in payload else None
+    if not isinstance(payload, dict) or "ok" not in payload:
+        return None
+    return None if too_deep(payload) else payload
 
 
 def _crash(box: BoxResult) -> str:
@@ -53,17 +58,25 @@ def _crash(box: BoxResult) -> str:
     return f"Crashed (exit {box.returncode})"
 
 
+def _took(ms: object, fallback: float) -> float:
+    if isinstance(ms, bool) or not isinstance(ms, int | float):
+        return fallback
+    try:
+        return float(ms)
+    except OverflowError:  # an int too big for a float
+        return fallback
+
+
 def _outcome(scratch: Path, box: BoxResult) -> tuple[object, str | None, float]:
     if box.timed_out:
         return None, "TimeLimitExceeded", box.elapsed_ms
     payload = _read_result(scratch)
     if payload is None:
         return None, _crash(box), box.elapsed_ms
-    ms = payload.get("ms")
-    took = float(ms) if isinstance(ms, int | float) else box.elapsed_ms
+    took = _took(payload.get("ms"), box.elapsed_ms)
     if payload["ok"] is True:
         return payload.get("value"), None, took
-    return None, str(payload.get("error", "Error")), took
+    return None, str(payload.get("error", "Error"))[:_MAX_ERROR], took
 
 
 @dataclass(frozen=True)
@@ -95,7 +108,12 @@ def _case(job: _Job, example: Example) -> CaseResult:
             language.ro_binds(job.binary),
         )
         actual, error, ms = _outcome(scratch, done)
-    ok = error is None and matches(example.expected, actual, example.order)
+    ok = False
+    if error is None:
+        try:
+            ok = matches(example.expected, actual, example.order)
+        except (TooDeepError, RecursionError, MemoryError) as exc:
+            actual, error = None, f"InvalidResult: {type(exc).__name__}"
     return CaseResult(
         example.n, example.args, example.expected, actual, ok, done.stdout, error, ms
     )

code_runner/code_runner/tests/test_compare.py

@@ -1,6 +1,6 @@
 import pytest
 
-from code_runner._compare import matches
+from code_runner._compare import MAX_DEPTH, TooDeepError, matches, too_deep
 
 
 @pytest.mark.parametrize(
@@ -40,3 +40,26 @@ def test_matches(expected: object, actual: object, order: str, *, want: bool) ->
 def test_unknown_order_is_refused() -> None:
     with pytest.raises(ValueError, match="order"):
         matches(1, 1, "sideways")
+
+
+def _nest(depth: int) -> object:
+    value: object = 1
+    for _ in range(depth):
+        value = [value]
+    return value
+
+
+@pytest.mark.parametrize("order", ["exact", "any", "deep"])
+def test_deep_nesting_is_refused_not_recursed(order: str) -> None:
+    deep = _nest(100_000)  # would raise RecursionError in the recursive walks
+    with pytest.raises(TooDeepError):
+        matches(deep, deep, order)
+    with pytest.raises(TooDeepError):
+        matches(1, deep, order)
+
+
+def test_nesting_up_to_the_limit_still_compares() -> None:
+    assert matches(_nest(MAX_DEPTH), _nest(MAX_DEPTH), "deep") is True
+    assert too_deep(_nest(MAX_DEPTH + 1)) is True
+    assert too_deep({"a": [{"b": _nest(MAX_DEPTH)}]}) is True
+    assert too_deep({"a": [{"b": 1}]}) is False

code_runner/code_runner/tests/test_internals.py

@@ -99,6 +99,49 @@ def test_result_file_variants() -> None:
     assert named.error == "KeyError: x"
 
 
+def test_hostile_result_files_cost_one_example_not_the_run() -> None:
+    hostile = {
+        "bomb": "[" * 200_000,
+        "deep": '{"ok": true, "value": ' + "[" * 5000 + "]" * 5000 + "}",
+        "huge_ms": '{"ok": true, "value": 1, "ms": 1' + "0" * 400 + "}",
+        "bool_ms": '{"ok": true, "value": 1, "ms": true}',
+        "long_error": '{"ok": false, "error": "' + "x" * 5000 + '"}',
+    }
+    results = {
+        name: _run(_fake(_box(), {"result.json": text}))
+        for name, text in hostile.items()
+    }
+    assert results["bomb"].error == "Crashed (exit 0)"
+    assert results["deep"].error == "Crashed (exit 0)"
+    assert (results["huge_ms"].ok, results["huge_ms"].ms) == (True, 7.0)
+    assert results["bool_ms"].ms == 7.0
+    assert len(results["long_error"].error) == 200
+
+
+def test_too_deep_expected_or_actual_is_that_examples_error() -> None:
+    deep: object = 1
+    for _ in range(100):
+        deep = [deep]
+    files = {"result.json": '{"ok": true, "value": 1}'}
+    cases = [Example(1, {"a": 1}, deep), EXAMPLE]
+    results = run("python3", "", STARTER, cases, box=_fake(_box(), files))
+    assert (results[0].ok, results[0].actual) == (False, None)
+    assert results[0].error == "InvalidResult: TooDeepError"
+    assert results[1].ok is True  # the run went on
+
+
+def test_recursion_in_compare_is_that_examples_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def boom(*args: object) -> bool:
+        raise RecursionError
+
+    monkeypatch.setattr("code_runner._run.matches", boom)
+    files = {"result.json": '{"ok": true, "value": 1}'}
+    result = _run(_fake(_box(), files))
+    assert result.error == "InvalidResult: RecursionError"
+
+
 def test_unreadable_result_files_are_a_crash(tmp_path: Path) -> None:
     secret = tmp_path / "secret"
     secret.write_text('{"ok": true, "value": "stolen"}')

code_runner/code_runner/tests/test_run.py

@@ -205,3 +205,23 @@ def test_unsupported_language_and_empty_examples() -> None:
     with pytest.raises(RunnerError, match="language"):
         run("cobol", "", "", [])
     assert run("python3", "def f(a): pass\n", "def f(a):\n", []) == []
+
+
+@needs_sandbox
+def test_json_bomb_result_costs_that_example_only() -> None:
+    code = (
+        "import os\n"
+        "def f(a):\n"
+        "    if a:\n"
+        "        open('/work/result.json', 'w').write('[' * 200000)\n"
+        "        os._exit(0)\n"
+        "    return 1\n"
+    )
+    out = run(
+        "python3",
+        code,
+        "def f(a):\n",
+        [Example(1, {"a": 1}, 1), Example(2, {"a": 0}, 1)],
+    )
+    assert (out[0].ok, out[0].error) == (False, "Crashed (exit 0)")
+    assert out[1].ok is True

code_runner/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "code_runner"
-version = "0.1.0"
+version = "0.1.1"
 description = "Run a learner solution (Python, JavaScript, TypeScript) against a problem's examples in a bubblewrap sandbox"
 requires-python = ">=3.14"
 # Stdlib only on purpose: the tutor server imports it from the system python.