From 3e31f1d78b87b1e202e8a4c4cb839a378eafda16 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 26 Sep 2026 18:34:07 +0530 Subject: [PATCH] test(code-execution): trim #121936 tests to the two shared-host invariants Keep one invariant per spawn path (persistent remote kernel and per-call sandbox): token never on argv, dirs owner-only. Drop the command-string duplicates, the malformed-seq replay test (the guard itself stays in tools/code_execution_rpc.py) and the real-fs E2E class. The dropped st_mode == 0o600 assertions were the Windows-failing ones flagged in review, so no POSIX guard is needed on what remains. --- tests/tools/test_code_execution.py | 91 --------------- tests/tools/test_code_kernel_remote.py | 148 ------------------------- 2 files changed, 239 deletions(-) diff --git a/tests/tools/test_code_execution.py b/tests/tools/test_code_execution.py index 9bf1572cd4..a8a753e06b 100644 --- a/tests/tools/test_code_execution.py +++ b/tests/tools/test_code_execution.py @@ -277,97 +277,6 @@ class TestRemoteSharedHostLockdown(unittest.TestCase): self.assertIn("umask 077", mkdir_cmd) self.assertIn("chmod 700", mkdir_cmd) - def test_rpc_response_write_is_owner_only(self): - """_rpc_poll_loop's res file write must be owner-only and must not put - the payload in argv on a stdin-capable backend: it carries tool results - on a shared-host backend.""" - from tools.code_execution_rpc import _rpc_poll_loop - - commands, payloads = [], [] - - class Env: - def execute(self, command, cwd=None, timeout=None, stdin_data=None): - commands.append(command) - payloads.append(stdin_data) - if command.startswith("ls "): - return {"output": "/rpc/req_000001\n", "returncode": 0} - if command.startswith("cat "): - return {"output": json.dumps( - {"tool": "not_allowed", "args": {}, "seq": 1, - "token": "T"})} - return {"output": "", "returncode": 0} - - stop = threading.Event() - t = threading.Thread( - target=_rpc_poll_loop, - args=(Env(), "/rpc", "t1", [], [0], 5, frozenset(), stop, "T"), - daemon=True) - t.start() - deadline = time.monotonic() + 10 - while not any("base64 -d" in c for c in commands): - if time.monotonic() > deadline: - stop.set() - t.join(5) - self.fail("rpc poll loop never wrote a response file") - time.sleep(0.01) - stop.set() - t.join(5) - res_write = next(c for c in commands if "base64 -d" in c) - self.assertIn("umask 077", res_write) - self.assertIn("res_000001", res_write) - # On a pipe-capable backend the result rides stdin, not argv. - res_idx = commands.index(res_write) - self.assertIsNotNone(payloads[res_idx], - "res payload should travel via stdin_data") - - def test_rpc_malformed_seq_request_removed_without_replay(self): - """An authenticated req with a non-int seq cannot form the res name the - caller polls. It must be removed without dispatch: formatting res_ - after dispatch raises, leaves the req in place, and replays the tool - call every poll cycle.""" - from tools.code_execution_rpc import _rpc_poll_loop - - commands = [] - dispatched = [] - - class Env: - def execute(self, command, cwd=None, timeout=None, stdin_data=None): - commands.append(command) - if command.startswith("ls "): - return {"output": "/rpc/req_000001\n", "returncode": 0} - if command.startswith("cat "): - return {"output": json.dumps( - {"tool": "read_file", "args": {"path": "/x"}, - "seq": "not-an-int", "token": "T"})} - return {"output": "", "returncode": 0} - - def fake_dispatch(name, args): - dispatched.append(name) - return "ok" - - stop = threading.Event() - with patch("tools.code_execution_rpc._default_dispatch", - return_value=fake_dispatch): - t = threading.Thread( - target=_rpc_poll_loop, - args=(Env(), "/rpc", "t1", [], [0], 5, - frozenset({"read_file"}), stop, "T"), - daemon=True) - t.start() - deadline = time.monotonic() + 10 - while not any("rm -f" in c and "req_000001" in c for c in commands): - if time.monotonic() > deadline: - stop.set() - t.join(5) - self.fail("malformed-seq request was never removed") - time.sleep(0.01) - stop.set() - t.join(5) - self.assertEqual(dispatched, [], - "malformed-seq request must not dispatch a tool call") - self.assertFalse(any("base64 -d" in c for c in commands), - "no res write for a request that cannot be answered") - @unittest.skipIf(sys.platform == "win32", "UDS not available on Windows") class TestExecuteCode(unittest.TestCase): diff --git a/tests/tools/test_code_kernel_remote.py b/tests/tools/test_code_kernel_remote.py index ada22b9299..360ad28e6b 100644 --- a/tests/tools/test_code_kernel_remote.py +++ b/tests/tools/test_code_kernel_remote.py @@ -10,7 +10,6 @@ state_lost/state_reset reporting, fail-open, and owner isolation. """ import json import os -import shutil import sys import time import unittest @@ -392,153 +391,6 @@ class TestSharedHostLockdown(RemoteKernelBase): env_content = base64.b64decode(env_ship).decode() self.assertIn(f"HERMES_RPC_TOKEN={kernel.rpc_token}", env_content) - def test_remote_write_transport_follows_stdin_mode(self): - """Pipe-mode backends carry the payload on stdin (never argv); heredoc - backends embed stdin in the command anyway, so they keep the echo pipe.""" - import base64 - from tools.code_execution_rpc import _remote_write_cmd - - class HeredocEnv: - _stdin_mode = "heredoc" - - cmd, stdin = _remote_write_cmd(HeredocEnv(), "/x/f", "data") - self.assertIsNone(stdin) - self.assertIn("echo '", cmd) - cmd2, stdin2 = _remote_write_cmd(ScriptedEnv([]), "/x/f", "data") - self.assertIsNotNone(stdin2) - self.assertNotIn("echo '", cmd2) - self.assertEqual(base64.b64decode(stdin2).decode(), "data") - - def test_stub_req_files_are_owner_only(self): - """The generated file-RPC stub writes req files mode 600: they carry - the token + tool args and sit in a dir a same-uid kernel shares.""" - import tempfile - import types - from tools.code_execution_tool import generate_hermes_tools_module - src = generate_hermes_tools_module(["read_file"], transport="file") - rpc_dir = tempfile.mkdtemp() - self.addCleanup(shutil.rmtree, rpc_dir, True) - with patch.dict(os.environ, {"HERMES_RPC_DIR": rpc_dir, - "HERMES_RPC_TOKEN": "test-token"}): - mod = types.ModuleType("hermes_tools") - exec(compile(src, "hermes_tools.py", "exec"), mod.__dict__) - with open(os.path.join(rpc_dir, "res_000001"), "w") as f: - f.write(json.dumps("ok")) - mod.read_file("/etc/hostname") - req = os.path.join(rpc_dir, "req_000001") - self.assertTrue(os.path.exists(req)) - self.assertEqual(os.stat(req).st_mode & 0o777, 0o600) - - def test_runner_cell_res_files_are_owner_only(self): - """The remote runner writes cell_res files mode 600: they carry the - cell's output in a dir under shared temp.""" - import tempfile - import threading - import types - from tools.code_kernel import RUNNER_CELL_SOURCE - from tools.code_kernel_remote import REMOTE_KERNEL_RUNNER_SOURCE - kdir = tempfile.mkdtemp() - self.addCleanup(shutil.rmtree, kdir, True) - cells = os.path.join(kdir, "cells") - os.makedirs(cells) - src = REMOTE_KERNEL_RUNNER_SOURCE.format( - cell_source=RUNNER_CELL_SOURCE, capture_limit=10000, idle_exit=2) - with patch.dict(os.environ, {"HERMES_KERNEL_DIR": kdir, - "HERMES_RPC_DIR": f"{kdir}/rpc", - "HERMES_RPC_TOKEN": "t"}): - mod = types.ModuleType("kernel_runner") - exec(compile(src, "kernel_runner.py", "exec"), mod.__dict__) - with open(os.path.join(cells, "cell_req_000001.json"), "w") as f: - json.dump({"code": "print('hi')", "id": "000001"}, f) - runner = threading.Thread(target=mod.main, daemon=True) - runner.start() - res = os.path.join(cells, "cell_res_000001.json") - deadline = time.monotonic() + 10 - while not os.path.exists(res): - self.assertLess(time.monotonic(), deadline, - "runner never wrote the cell result") - time.sleep(0.02) - self.assertEqual(os.stat(res).st_mode & 0o777, 0o600) - runner.join(5) - - -@unittest.skipUnless(os.name == "posix" and shutil.which("bash"), - "needs a POSIX bash transport") -class TestRemoteKernelLocalEnvE2E(RemoteKernelBase): - """Real end-to-end: spawn the remote kernel through LocalEnvironment's real - bash transport, run a cell, and stat the modes on disk. This is the same - code path an ssh backend drives, pointed at this host.""" - - def test_kernel_tree_is_owner_only_on_real_fs(self): - self._ship.stop() - self._poll.stop() - from tools.environments.local import LocalEnvironment - env = LocalEnvironment(cwd="/", timeout=60) - try: - result = _run(env, code=( - "import os\n" - "print('KDIR_MODE=%o' % (os.stat(os.environ['HERMES_KERNEL_DIR']).st_mode & 0o777))\n" - "print('RPC_MODE=%o' % (os.stat(os.environ['HERMES_RPC_DIR']).st_mode & 0o777))\n" - "print('PP_DELIVERED=%s' % (os.environ.get('PYTHONPATH') == os.environ['HERMES_KERNEL_DIR']))\n" - "print('TOKEN_DELIVERED=%s' % bool(os.environ.get('HERMES_RPC_TOKEN')))\n" - ), timeout=60) - self.assertEqual(result["status"], "success", result) - blob = json.dumps(result) - self.assertIn("KDIR_MODE=700", blob) - self.assertIn("RPC_MODE=700", blob) - # The env file must actually deliver its vars to the runner — a - # broken source would silently degrade cells to no-RPC. - self.assertIn("PP_DELIVERED=True", blob) - self.assertIn("TOKEN_DELIVERED=True", blob) - kernel = next(iter(_REMOTE_KERNELS.values())) - # kernel.env is consumed by the subshell source: the token file is - # gone after launch while the runner keeps the values in its env. - self.assertFalse(os.path.exists( - os.path.join(kernel.kernel_dir, "kernel.env"))) - for name in ("kernel_runner.py", "hermes_tools.py", "runner.log"): - p = os.path.join(kernel.kernel_dir, name) - self.assertTrue(os.path.exists(p), p) - self.assertEqual(os.stat(p).st_mode & 0o777, 0o600, p) - # The subshell confinement keeps the token and execution-scoped vars - # out of the backend's session snapshot: they must not leak into the - # snapshot file or the environ of a later command on the same env - # (issue #71296 snapshot-leak class). - probe = env.execute( - "printenv HERMES_RPC_TOKEN; printenv HERMES_KERNEL_DIR; " - "printenv HERMES_RPC_DIR; printenv PYTHONPATH", - cwd="/", timeout=15) - self.assertNotIn(kernel.rpc_token, probe.get("output", "")) - self.assertNotIn(kernel.kernel_dir, probe.get("output", "")) - self.assertNotIn( - kernel.rpc_token, - open(env._snapshot_path).read() if os.path.exists(env._snapshot_path) else "") - finally: - shutdown_all_remote_kernels() - - def test_per_call_sandbox_is_owner_only_on_real_fs(self): - self._ship.stop() - self._poll.stop() - from tools.environments.local import LocalEnvironment - from tools.code_execution_tool import _run_remote_per_call - env = LocalEnvironment(cwd="/", timeout=60) - code = ( - "import os\n" - "print('SANDBOX_MODE=%o' % (os.stat(os.path.dirname(os.environ['HERMES_RPC_DIR'])).st_mode & 0o777))\n" - "print('RPC_MODE=%o' % (os.stat(os.environ['HERMES_RPC_DIR']).st_mode & 0o777))\n" - "print('SCRIPT_MODE=%o' % (os.stat('script.py').st_mode & 0o777))\n" - "print('TOOLS_MODE=%o' % (os.stat('hermes_tools.py').st_mode & 0o777))\n" - "print('ENVFILE_MODE=%o' % (os.stat('sandbox.env').st_mode & 0o777))\n" - ) - out = json.loads(_run_remote_per_call( - env, "local", code, "t-e2e", frozenset({"read_file"}), - timeout=60, max_tool_calls=5, exec_start=time.monotonic())) - self.assertEqual(out["status"], "success", out) - self.assertIn("SANDBOX_MODE=700", out["output"]) - self.assertIn("RPC_MODE=700", out["output"]) - self.assertIn("SCRIPT_MODE=600", out["output"]) - self.assertIn("TOOLS_MODE=600", out["output"]) - self.assertIn("ENVFILE_MODE=600", out["output"]) - if __name__ == "__main__": unittest.main()