"""host.py 验测命令白名单（shell=True 收口）的测试。 覆盖三件事，缺一不可： 1. 分类：simple / simple-env / shell / denied 的判定与拒绝原因； 2. 执行：拒绝绝不执行（第二条命令/命令替换的标记文件必须不 出现）；simple/simple-env 走 shell=False；shell 形态（复合、 管道、重定向、环境变量、glob、引用）与旧 shell 语义等价； 3. 迁移：verify-audit 报告分类、--apply 只转 simple、不碰 shell/denied；naive 的 shlex.split 转换会破坏复合/管道/ 重定向/环境变量语义（回归锁：以后不许有人把“改成 shell=False”理解成“直接 shlex.split 就行”）。 """ import json import os import subprocess import sys import tempfile import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT)) import host def kind_of(cmd, cwd=None): return host.classify_verify_command(cmd, cwd=cwd)["kind"] class ClassifyTest(unittest.TestCase): def test_simple_commands(self): for cmd in ("python3 -m unittest discover -s tests -v", "test -f result.txt", "true", "[ -f result.txt ]", "echo hello", "make test"): with self.subTest(cmd=cmd): self.assertEqual(kind_of(cmd), "simple") def test_simple_argv_keeps_quoted_words(self): info = host.classify_verify_command( "python3 -m pytest \"tests/a b.py\"") self.assertEqual(info["kind"], "simple") self.assertEqual(info["argv"], ["python3", "-m", "pytest", "tests/a b.py"]) info = host.classify_verify_command("echo 'a b'") self.assertEqual(info["argv"], ["echo", "a b"]) def test_single_quoted_dollar_is_literal_not_expansion(self): # 单引号内 $ 是字面量：shell 与 argv 语义一致，走 simple。 info = host.classify_verify_command("echo '$HOME'") self.assertEqual(info["kind"], "simple") self.assertEqual(info["argv"], ["echo", "$HOME"]) def test_shell_features(self): cases = { "python3 -m pytest tests/ && echo done": "compound", "python3 -m pytest tests/; echo done": "compound", "python3 -m pytest tests/ | tail -20": "pipe", "python3 -m pytest tests/ > out.log 2>&1": "redirect", "echo hi > /dev/null": "redirect", "echo $HOME": "env", "echo ${HOME}/x": "env", "echo ~/x": "env", "python3 -m pytest tests/test_*.py": "glob", } for cmd, feature in cases.items(): with self.subTest(cmd=cmd): info = host.classify_verify_command(cmd) self.assertEqual(info["kind"], "shell") self.assertIn(feature, info["features"]) def test_assign_only_is_simple_env(self): info = host.classify_verify_command( "FOO=bar python3 -m pytest tests/") self.assertEqual(info["kind"], "simple-env") self.assertEqual(info["argv"], ["python3", "-m", "pytest", "tests/"]) self.assertEqual(info["env"], {"FOO": "bar"}) # 赋值里带展开就不再是 simple-env：须走 shell 形态。 info = host.classify_verify_command("FOO=~/x python3 -m pytest") self.assertEqual(info["kind"], "shell") def test_denied_escape_shapes(self): denied = ( "pytest $(cat list.txt)", "echo `id`", "echo \"$(id)\"", "(pytest tests/)", "pytest tests/\nrm -rf /", "pytest tests/ # comment", "echo 'unclosed", "pytest tests/ &&", "FOO=bar", "> out.log", "echo hi > $OUT.log", "python3 -m pytest tests/ < /tmp/x.log", "echo hi > ../x.log", "echo hi > /etc/passwd", "echo hi >&/tmp/x.log", ) for cmd in denied: with self.subTest(cmd=cmd): self.assertEqual(kind_of(cmd), "denied") def test_empty_is_empty_not_denied(self): self.assertEqual(kind_of(""), "empty") self.assertEqual(kind_of(" "), "empty") def test_runner_args_are_not_policed_by_first_word(self): # 诚实边界：首词白名单拦不住运行器自身的参数（-c、 # Makefile、npm script 本来就能跑代码），这里只锁定 # “当前实现确实放行”这一事实，防以后有人误以为它是沙箱。 self.assertEqual(kind_of("python3 -c 'print(1)'"), "simple") def test_repo_script_path_needs_executable_under_cwd(self): with tempfile.TemporaryDirectory() as tmp: script = Path(tmp, "verify.sh") script.write_text("#!/bin/sh\necho ok\n", encoding="utf-8") os.chmod(script, 0o755) self.assertEqual(kind_of("./verify.sh", cwd=tmp), "simple") self.assertEqual(kind_of("./missing.sh", cwd=tmp), "denied") self.assertEqual(kind_of("../escape.sh", cwd=tmp), "denied") self.assertEqual(kind_of("/bin/true", cwd=tmp), "denied") script.chmod(0o644) info = host.classify_verify_command("./verify.sh", cwd=tmp) self.assertEqual(info["kind"], "denied") self.assertIn("不可执行", info["reason"]) class GatedRunTest(unittest.TestCase): def test_denied_never_executes_second_command(self): with tempfile.TemporaryDirectory() as tmp: marker = os.path.join(tmp, "pwned") rc, out = host.run_gated_verify( f"echo hi; touch {marker}", tmp) self.assertEqual(rc, 126) self.assertFalse(os.path.exists(marker)) self.assertIn("白名单拒绝", out) def test_denied_never_executes_substitution(self): with tempfile.TemporaryDirectory() as tmp: marker = os.path.join(tmp, "pwned") rc, _ = host.run_gated_verify(f"echo $(touch {marker})", tmp) self.assertEqual(rc, 126) self.assertFalse(os.path.exists(marker)) def test_denied_is_audited(self): with tempfile.TemporaryDirectory() as tmp: state_path = os.path.join(tmp, "host-state.json") host.run_gated_verify("rm -rf build", tmp, audit_state_path=state_path) log = Path(tmp, "host-run.log").read_text() self.assertIn("verify denied rc=126", log) self.assertIn("rm -rf build", log) def test_simple_runs_without_shell_expansion(self): with tempfile.TemporaryDirectory() as tmp: # shell=False 时 $HOME 不展开：输出就是字面量（经 # simple 分类？$HOME 是 env 形态，走 shell；这里用 # 引号字面量证明 simple 路径确实不经过 shell）。 rc, out = host.run_gated_verify("echo '$HOME'", tmp) self.assertEqual((rc, out), (0, "$HOME")) def test_simple_env_applies_assignments(self): with tempfile.TemporaryDirectory() as tmp: rc, out = host.run_gated_verify( "FOO=bar python3 -c \"import os; print(os.environ['FOO'])\"", tmp) self.assertEqual((rc, out), (0, "bar")) def test_shell_compound_pipe_redirect_env_glob_equivalence(self): # 与旧 shell=True 语义逐项等价：同一命令在新旧两路下 # 输出一致（旧路直接 subprocess shell=True 对照）。 with tempfile.TemporaryDirectory() as tmp: Path(tmp, "a.txt").write_text("hello\n", encoding="utf-8") Path(tmp, "b.txt").write_text("world\n", encoding="utf-8") cases = ( "echo a && echo b", "cat a.txt | grep -q hello && echo found", "echo data > out.log", "FOO=hi python3 -c \"import os; print(os.environ['FOO'])\"", "cat *.txt", ) for cmd in cases: with self.subTest(cmd=cmd): old = subprocess.run(cmd, shell=True, cwd=tmp, capture_output=True, text=True) rc, out = host.run_gated_verify(cmd, tmp) self.assertEqual(rc, old.returncode) self.assertEqual(out, (old.stdout + old.stderr).strip()) def test_argv_is_literal_no_shell_parsing(self): with tempfile.TemporaryDirectory() as tmp: rc, out = host.run_verify_argv(["echo", "a;b", "$(x)", "*"], tmp) self.assertEqual((rc, out), (0, "a;b $(x) *")) with self.assertRaises(TypeError): host.run_verify_argv("echo hi", tmp) def test_run_verify_empty_and_default_unittest(self): with tempfile.TemporaryDirectory() as tmp: self.assertEqual(host.run_verify("", tmp), (0, "")) self.assertEqual(host.run_verify("true", tmp), (0, "")) class NaiveShlexSplitBreakageTest(unittest.TestCase): """锁死一条教训：把 shell 字符串直接 shlex.split 后走 shell=False 不是等价转换——复合、管道、重定向、环境变量、 glob 的语义全变。白名单的 simple 路径只对无形态命令用 shlex 结果，shell 形态必须走 shell=True（白名单内）或 argv。""" def _naive(self, cmd, cwd): import shlex v = subprocess.run(shlex.split(cmd), shell=False, cwd=cwd, capture_output=True, text=True) return v.returncode, (v.stdout + v.stderr).strip() def test_naive_split_breaks_compound(self): with tempfile.TemporaryDirectory() as tmp: rc, out = self._naive("echo a && echo b", tmp) # naive 把 && 当成 echo 的参数，输出错误且 rc 仍 0： # 这正是最危险的“静默语义变化”。 self.assertEqual(out, "a && echo b") rc2, out2 = host.run_gated_verify("echo a && echo b", tmp) self.assertEqual((rc2, out2), (0, "a\nb")) def test_naive_split_breaks_pipe_and_redirect(self): with tempfile.TemporaryDirectory() as tmp: rc, out = self._naive("echo hello | grep hello", tmp) self.assertNotEqual(out, "hello") rc, out = self._naive("echo data > out.log", tmp) self.assertFalse(os.path.exists(os.path.join(tmp, "out.log"))) def test_naive_split_breaks_env_assignment(self): with tempfile.TemporaryDirectory() as tmp: with self.assertRaises(FileNotFoundError): self._naive("FOO=bar echo hi", tmp) class MergeAndGoalIntegrationTest(unittest.TestCase): def _make_repo(self, tmp): subprocess.run(["git", "init", "-q"], cwd=tmp, check=True) subprocess.run(["git", "config", "user.name", "Test"], cwd=tmp, check=True) subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=tmp, check=True) Path(tmp, "mod.py").write_text("def f():\n return 1\n", encoding="utf-8") subprocess.run(["git", "add", "-A"], cwd=tmp, check=True) subprocess.run(["git", "commit", "-qm", "initial"], cwd=tmp, check=True) subprocess.run(["git", "worktree", "add", os.path.join(tmp, "wt"), "-b", "host/t1", "HEAD"], cwd=tmp, check=True) return os.path.join(tmp, "wt") def test_merge_denied_verify_rolls_back_and_keeps_worktree(self): with tempfile.TemporaryDirectory() as tmp: wt = self._make_repo(tmp) pre = subprocess.run(["git", "rev-parse", "HEAD"], cwd=tmp, capture_output=True, text=True).stdout.strip() ok, msg = host.merge_task(tmp, wt, "host/t1", "t1", verify_command="rm -rf build") self.assertFalse(ok) self.assertIn("白名单拒绝", msg) now = subprocess.run(["git", "rev-parse", "HEAD"], cwd=tmp, capture_output=True, text=True).stdout.strip() self.assertEqual(now, pre) self.assertTrue(os.path.isdir(wt)) def test_merge_verify_argv_wins_over_string(self): with tempfile.TemporaryDirectory() as tmp: wt = self._make_repo(tmp) ok, msg = host.merge_task(tmp, wt, "host/t1", "t1", verify_command="rm -rf build", verify_argv=["true"]) self.assertTrue(ok, msg) self.assertFalse(os.path.isdir(wt)) def test_goal_denied_verify_is_unmet_not_crash(self): class FakeServer: def run_turn(self, thread_id, prompt, model=None, effort=None, sandbox=None, timeout=None): return {"status": "completed", "text": "done", "turn_id": "u1"} with tempfile.TemporaryDirectory() as tmp: state_path = os.path.join(tmp, "host-state.json") state = {"tasks": {}, "automations": []} task = {"name": "g", "thread_id": "x", "cwd": tmp, "sandbox": None, "max_retries": 0, "retry_count": 0, "turns": []} state["tasks"]["g"] = task result = host.run_goal_loop(FakeServer(), state, state_path, task, "make it", "rm -rf build", 1, "m", "low", None, cfg={}, timeout=600) self.assertFalse(result["ok"]) self.assertEqual(task["status"], "goal-unmet") self.assertIn("白名单拒绝", task["goal_last_verify"]) def test_goal_verify_argv_runs_shell_false(self): class FakeServer: def run_turn(self, thread_id, prompt, model=None, effort=None, sandbox=None, timeout=None): return {"status": "completed", "text": "done", "turn_id": "u1"} with tempfile.TemporaryDirectory() as tmp: state_path = os.path.join(tmp, "host-state.json") state = {"tasks": {}, "automations": []} task = {"name": "g", "thread_id": "x", "cwd": tmp, "sandbox": None, "max_retries": 0, "retry_count": 0, "turns": []} state["tasks"]["g"] = task result = host.run_goal_loop(FakeServer(), state, state_path, task, "make it", None, 1, "m", "low", None, cfg={}, timeout=600, verify_argv=["echo", "a;b"]) self.assertTrue(result["ok"]) self.assertEqual(task["status"], "goal-met") self.assertEqual(task["verify_argv"], ["echo", "a;b"]) saved = host.load_state(state_path) self.assertEqual(saved["tasks"]["g"]["verify_argv"], ["echo", "a;b"]) class VerifyAuditTest(unittest.TestCase): def test_report_classifies_and_apply_converts_only_simple(self): with tempfile.TemporaryDirectory() as tmp: state = {"tasks": { "simple-task": {"verify_command": "true", "cwd": tmp}, "shell-task": {"verify_command": "echo a && echo b", "cwd": tmp}, "denied-task": {"verify_command": "rm -rf build", "cwd": tmp}, "argv-task": {"verify_command": None, "verify_argv": ["true"], "cwd": tmp}, "newform-task": {"verify_command": None, "cwd": tmp}, }, "automations": []} report = host.build_verify_audit_report(state) by_name = {r["task"]: r for r in report} self.assertEqual(set(by_name), {"simple-task", "shell-task", "denied-task", "argv-task"}) self.assertEqual(by_name["simple-task"]["kind"], "simple") self.assertEqual(by_name["simple-task"]["argv"], ["true"]) self.assertEqual(by_name["shell-task"]["kind"], "shell") self.assertIsNone(by_name["shell-task"]["argv"]) self.assertEqual(by_name["denied-task"]["kind"], "denied") self.assertEqual(by_name["argv-task"]["kind"], "argv") class Args: apply = True json = False state = os.path.join(tmp, "host-state.json") with open(Args.state, "w", encoding="utf-8") as f: json.dump(state, f) # _cmd_verify_audit 内部用 load/save 语义：这里直接 # 传已加载的 state（main 里 state 正是 load_state 得来）。 loaded = host.load_state(Args.state) host._cmd_verify_audit(Args(), loaded) saved = host.load_state(Args.state) self.assertEqual(saved["tasks"]["simple-task"]["verify_argv"], ["true"]) self.assertIsNone(saved["tasks"]["shell-task"].get("verify_argv")) self.assertIsNone(saved["tasks"]["denied-task"].get("verify_argv")) def test_cli_verify_argv_remainder_eats_dash_flags(self): # --verify-argv 必须是 REMAINDER：nargs="+" 会把 -m 当成 # 未识别选项（实测回归）。这里用 merge 分支验证：任务无 # repo 时在 server 初始化后打印提示并返回，不跑验测。 with tempfile.TemporaryDirectory() as tmp: state_path = os.path.join(tmp, "host-state.json") with open(state_path, "w", encoding="utf-8") as f: json.dump({"tasks": {"t": {}}, "automations": []}, f) out = subprocess.run( [sys.executable, str(REPO_ROOT / "host.py"), "--state", state_path, "merge", "--task", "t", "--verify-argv", "python3", "-m", "pytest", "tests/"], capture_output=True, text=True, timeout=60) self.assertEqual(out.returncode, 0, out.stderr[-2000:]) self.assertNotIn("unrecognized arguments", out.stderr) self.assertIn("非 git worktree 任务", out.stdout) def test_cli_verify_argv_empty_is_usage_error_not_skip(self): with tempfile.TemporaryDirectory() as tmp: state_path = os.path.join(tmp, "host-state.json") with open(state_path, "w", encoding="utf-8") as f: json.dump({"tasks": {"t": {}}, "automations": []}, f) out = subprocess.run( [sys.executable, str(REPO_ROOT / "host.py"), "--state", state_path, "merge", "--task", "t", "--verify-argv"], capture_output=True, text=True, timeout=60) self.assertEqual(out.returncode, 2) self.assertIn("--verify-argv 后须至少给一个参数", out.stderr) def test_audit_cli_does_not_boot_server(self): # verify-audit 必须在 srv 构造前返回：没有 codex 二进制 # 也能跑，且不产生 host-steps/server 相关副作用。 with tempfile.TemporaryDirectory() as tmp: state_path = os.path.join(tmp, "host-state.json") with open(state_path, "w", encoding="utf-8") as f: json.dump({"tasks": { "t1": {"verify_command": "echo a && echo b", "cwd": tmp}}, "automations": []}, f) out = subprocess.run( [sys.executable, str(REPO_ROOT / "host.py"), "--state", state_path, "verify-audit"], capture_output=True, text=True, timeout=60) self.assertEqual(out.returncode, 0, out.stderr[-2000:]) self.assertIn("shell", out.stdout) self.assertIn("t1", out.stdout) if __name__ == "__main__": unittest.main()