diff --git a/src/trinity/orchestration/reward.py b/src/trinity/orchestration/reward.py index b1da624..7e0e95e 100644 --- a/src/trinity/orchestration/reward.py +++ b/src/trinity/orchestration/reward.py @@ -45,6 +45,7 @@ import json import os import re +import secrets import subprocess import sys import tempfile @@ -1189,6 +1190,36 @@ def _functional_harness(fn_name: str, stdin_data: str, expected_output: str) -> return "\n".join(lines) + "\n" +def _run_checked_script(candidate_code: str, checker: str, *, timeout_s: int) -> bool: + """Run ``candidate_code`` then ``checker``, requiring the checker to run to + completion — not merely a zero exit status. + + The assert- and call-based flavors append verification code *after* the + untrusted candidate. Judging a pass by exit code alone let a candidate that + terminates the process early — ``sys.exit(0)``, ``exit()``/``quit()``, + ``raise SystemExit(0)``, or even a hard ``os._exit(0)`` — short-circuit the + appended checks and score correct regardless of correctness. + + We append a write of a per-run random sentinel *after* the checker. A pass + requires exit code 0 AND the sentinel present in stdout, so any early + termination before the checker's asserts run fails: the sentinel is never + emitted. The token is unguessable per run, so a candidate cannot forge it by + printing a fixed string. + """ + sentinel = "__TRINITY_PASS_" + secrets.token_hex(16) + "__" + script = ( + candidate_code + + "\n\n" + + checker + + "\n" + + "import sys as _trinity_sys\n" + + f"_trinity_sys.stdout.write({sentinel!r})\n" + + "_trinity_sys.stdout.flush()\n" + ) + ok, stdout = _exec_script_capture(script, stdin_data="", timeout_s=timeout_s) + return ok and sentinel in stdout + + def _run_one_test( code: str, test: object, timeout_s: int, fn_name: str | None = None ) -> bool: @@ -1211,8 +1242,8 @@ def _run_one_test( if ttype == "functional" and has_io: stdin_data = str(test.get("stdin", test.get("input", ""))) expected = str(test.get("expected_stdout", test.get("output", ""))) - script = code + "\n\n" + _functional_harness(fn_name or "", stdin_data, expected) - return _exec_script(script, stdin_data="", timeout_s=timeout_s) + checker = _functional_harness(fn_name or "", stdin_data, expected) + return _run_checked_script(code, checker, timeout_s=timeout_s) stdin_data: str | None = None expected_stdout: str | None = None @@ -1245,8 +1276,7 @@ def _run_one_test( return False if assert_block is not None: - script = code + "\n\n" + assert_block + "\n" - return _exec_script(script, stdin_data="", timeout_s=timeout_s) + return _run_checked_script(code, assert_block, timeout_s=timeout_s) # stdin/stdout test. ok, stdout = _exec_script_capture( @@ -1312,12 +1342,6 @@ def _apply() -> None: # pragma: no cover - runs only in the child process. return _apply -def _exec_script(script: str, *, stdin_data: str, timeout_s: int) -> bool: - """Run a script; pass iff it exits 0 within the timeout. No output check.""" - ok, _ = _exec_script_capture(script, stdin_data=stdin_data, timeout_s=timeout_s) - return ok - - def _exec_script_capture( script: str, *, stdin_data: str, timeout_s: int ) -> tuple[bool, str]: diff --git a/tests/test_reward_sandbox.py b/tests/test_reward_sandbox.py index 5ffe35e..6e7fcb2 100644 --- a/tests/test_reward_sandbox.py +++ b/tests/test_reward_sandbox.py @@ -69,3 +69,59 @@ def test_rlimits_do_not_break_ordinary_grading(): code = "import sys\nprint(int(sys.stdin.read()) + 1)" tests = [{"input": "41\n", "output": "42"}] assert R.run_pass_at_1(code, tests, timeout_s=5) is True + + +# --------------------------------------------------------------------------- +# Early-exit reward hack: a candidate must not short-circuit the appended +# checker by terminating the process with a zero exit status before the +# asserts run. Pass now requires the checker to run to completion. +# --------------------------------------------------------------------------- +_WRONG_ADD = "def add(a, b):\n return 999\n" +_ADD_ASSERT = ["assert add(2, 3) == 5"] + + +def test_assert_rejects_sys_exit_before_checker(): + hack = _WRONG_ADD + "import sys\nsys.exit(0)\n" + assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False + + +def test_assert_rejects_exit_builtin_before_checker(): + hack = _WRONG_ADD + "exit()\n" + assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False + + +def test_assert_rejects_os_exit_before_checker(): + # os._exit is a hard exit that bypasses SystemExit handling; the missing + # success sentinel still fails it. + hack = _WRONG_ADD + "import os\nos._exit(0)\n" + assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False + + +def test_assert_rejects_raise_systemexit_before_checker(): + hack = _WRONG_ADD + "raise SystemExit(0)\n" + assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False + + +def test_assert_correct_code_still_passes(): + good = "def add(a, b):\n return a + b\n" + assert R.run_pass_at_1(good, _ADD_ASSERT, timeout_s=5) is True + + +def test_assert_rejects_forged_success_string(): + # A candidate cannot fake a pass by printing a plausible marker; the real + # sentinel is random per run. + hack = _WRONG_ADD + "print('__TRINITY_PASS__')\nimport sys\nsys.exit(0)\n" + assert R.run_pass_at_1(hack, _ADD_ASSERT, timeout_s=5) is False + + +def test_functional_rejects_sys_exit_before_checker(): + # LiveCodeBench call-based path: wrong Solution that exits before the harness. + bad = ( + "class Solution:\n" + " def twoSum(self, nums, target):\n" + " return [9, 9]\n" + "import sys\n" + "sys.exit(0)\n" + ) + spec = {"tests": [{"input": "[2,7,11,15]\n9", "output": "[0,1]", "testtype": "functional"}], "fn_name": "twoSum"} + assert R.score_text("livecodebench", "```python\n" + bad + "```", spec) == 0.0