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.
This commit is contained in:
kshitijk4poor
@@ -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):
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user