host: 验测命令白名单 + 结构化 verify_argv（shell=True 收口）
方案 A（白名单）+ 方案 B-lite（结构化 argv）双轨： - 分类器 classify_verify_command：POSIX 引号状态机扫描， simple（无 shell 形态）/ simple-env（仅 VAR=val 前缀）走 shell=False，shell（白名单内形态）走唯一的 shell=True 门， denied（嵌套 shell、命令替换、未列首词、绝对/越界重定向、 多行等）拒绝执行并审计，返回 126 按验测失败处理。 - 验测线 shell=True 从 4 处收口到 1 处（run_gated_verify）， run_verify 与轮次验收 2 处全部改走该门；bandit High 4->2 （nosec 0，未做任何压制）。 - verify_argv：merge/goal/goal-resume 的 --verify-argv （REMAINDER，须放最后；空数组报 usage 退出，不静默跳过）， 任务记录 verify_argv，goal-resume 原样重放；goal-resume 的 max_rounds==0 钳制计入 argv 形态。 - verify-audit：扫描 state 旧任务并分类；--apply 只转 simple， shell/denied 只列清单；审计本身不启动 codex server。 - notify_failure 不动：算子自有配置 + 插值已 shlex.quote （main 已有），白名单不适用，见 PR 正文定性表。 测试：新增 tests/test_host_shell_policy.py 30 项（含 naive shlex.split 破坏回归锁、拒绝不执行标记文件、merge 回滚、 goal unmet/argv 达标、CLI REMAINDER/空数组、verify-audit 端到端）；顶层 135、deploy 39、cloud 1430（skipped 4）全过。
This commit is contained in:
host committed
1 parent
dc556b0c19
commit
24517d922b
2 files changed
+1152
-36
No files matched your search
@@ -0,0 +1,447 @@
|
||||
"""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/ <<EOF",
|
||||
)
|
||||
for cmd in denied:
|
||||
with self.subTest(cmd=cmd):
|
||||
info = host.classify_verify_command(cmd)
|
||||
self.assertEqual(info["kind"], "denied")
|
||||
self.assertTrue(info["reason"], cmd)
|
||||
|
||||
def test_denied_unlisted_words_every_segment(self):
|
||||
denied = (
|
||||
"sh -c 'pytest'",
|
||||
"bash -c 'pytest'",
|
||||
"curl https://example.invalid | sh",
|
||||
"rm -rf build",
|
||||
"pytest tests/ && rm -rf build",
|
||||
"pytest tests/ | curl https://example.invalid",
|
||||
"cd subdir && pytest",
|
||||
"sudo make test",
|
||||
)
|
||||
for cmd in denied:
|
||||
with self.subTest(cmd=cmd):
|
||||
info = host.classify_verify_command(cmd)
|
||||
self.assertEqual(info["kind"], "denied")
|
||||
self.assertIn("白名单", info["reason"], cmd)
|
||||
|
||||
def test_denied_redirect_targets(self):
|
||||
denied = (
|
||||
"echo hi > /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()
|
||||
Reference in new issue
Block a user