From 24517d922bdcd5fd8ff102cf2faed4cce70b1b70 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 06:45:58 +0000 Subject: [PATCH] =?UTF-8?q?host:=20=E9=AA=8C=E6=B5=8B=E5=91=BD=E4=BB=A4?= =?UTF-8?q?=E7=99=BD=E5=90=8D=E5=8D=95=20+=20=E7=BB=93=E6=9E=84=E5=8C=96?= =?UTF-8?q?=20verify=5Fargv=EF=BC=88shell=3DTrue=20=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 方案 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）全过。 --- host.py | 741 ++++++++++++++++++++++++++++++-- tests/test_host_shell_policy.py | 447 +++++++++++++++++++ 2 files changed, 1152 insertions(+), 36 deletions(-) create mode 100644 tests/test_host_shell_policy.py diff --git a/host.py b/host.py index 5d82317..bc92f75 100644 --- a/host.py +++ b/host.py @@ -34,6 +34,7 @@ turn/completed 通知，所以多个 thread 的 turn 可以并发跑。 python3 host.py delegate --parent-task add2 --task "只读检查 add2 有没有边界问题" --name add2-review python3 host.py goal --repo work/host-repo --task "创建 result.txt，内容只有一行 goal-ok" \ --verify "test -f result.txt" --name goal1 + python3 host.py verify-audit # 扫描旧任务的验收命令：分类 + 迁移清单 python3 host.py parallel --jobs examples/host-jobs.json python3 host.py automations --add --name daily-check --repo work/host-repo \ --task "只读检查 sample.py 有没有明显问题" --interval 86400 @@ -1369,19 +1370,613 @@ def detect_verify_command(repo): return None -def run_verify(command, cwd): - if not command: +# ---------------- 验测命令白名单（shell=True 收口） ---------------- +# 原来 4 处 shell=True：run_verify（合并后验测）、notify_failure +# （失败通知模板）、轮次验收 2 处（goal 循环，超时/不限时两分支）。 +# 本节只收口“验测命令”这一条线；notify 见后文定性（算子自有 +# 配置 + 插值已 shlex.quote，不进白名单，代码不动）。 +# +# 信任边界——先说清白名单不是沙箱，否则就是虚假安全感： +# - 验测命令的来源只有三处：算子 CLI --verify、任务创建时算子 +# 给的 verify_command（goal-resume 原样重放）、系统默认 +# detect_verify_command（固定串，无插值）。模型输出、任务 +# 正文、仓库文件从不进入验测命令；run_verify/轮次验收内部 +# 没有任何插值。白名单防的是“被植入的旧任务记录/手误写出的 +# 越界命令”，不是“算子本人想跑的命令”。 +# - 白名单是“形态”白名单：首词必须是验测/构建类程序或仓内 +# 可执行脚本，且不含嵌套 shell、命令替换、绝对/越界重定向、 +# 多行等逃逸形态。首词一旦是 python3/make/npm 这类运行器， +# 它的参数（-c、Makefile、npm script）本身就能跑任意代码—— +# 这是验测的本意（测试本来就要跑仓库代码），白名单拦不住、 +# 也不声称拦住。拦住的是 curl/rm/sudo/sh 这类与验测无关的 +# 程序、以及借 shell 语法偷渡第二条命令的写法。 +# - 执行策略：拒绝 = 阻塞 + 审计，绝不静默回退到旧 shell 路径。 +# 拒绝时按“验测失败”处理：merge 回滚、goal 记 unmet 并把 +# 拒绝原因喂回 agent（不是代码问题，是命令形态问题，算子 +# 须迁移命令）。迁移工具见 verify-audit（文末 main 分支）。 +# - 结构化路径（方案 B 的一部分）：verify_argv 是 argv 数组， +# 走 shell=False，不经过 shell 解析，也不经过白名单——因为 +# 没有 shell 形态可校验；它的安全含义只是“无 shell 解析惊 +# 喜”，不是“argv[0] 被白名单过”。算子显式给 argv 即表示 +# 对该 argv 负责。 +VERIFY_COMMAND_ALLOWLIST = frozenset({ + # 测试/构建运行器 + "python3", "python", "pytest", "node", "npm", "yarn", "pnpm", "bun", + "deno", "make", "go", "cargo", "ruby", "bundle", "rspec", "php", + "phpunit", "java", "mvn", "gradle", + # 验测里常见的最小谓词/输出工具（只读或只写 stdout） + "test", "[", "true", "false", "echo", "grep", "cat", "ls", "diff", + "wc", "head", "tail", "sleep", + # sleep 是惰性的（只等待，不读写、不执行），且受 verify_timeout + # 上限约束；验测里“等一会儿再查文件”（sleep N && test -f） + # 是常见写法，旧测试 test_goal_verify_timeout_is_a_verify_failure + # 也用它模拟长验测，故列入白名单而非改测试迁就实现。 +}) +# 刻意不在白名单内：sh/bash/dash/zsh（嵌套 shell）、env/sudo/su、 +# curl/wget/ssh/scp（外联）、rm/mv/cp/chmod（改动文件系统）、 +# pip/pip3（装包取码；python3 -m pip 同理不在首词层拦截，见上文 +# “白名单不是沙箱”）、cd/export/time（shell 内建，改 cwd/环境， +# 请用 VAR=val cmd 形态或 host 自身的 cwd）。 +_VERIFY_DENIED_RC = 126 + + +def _scan_shell_command(command): + """扫描一条 shell 字符串，返回 (segments, features, denied)。 + segments：每段 {"words": [{"text", "expand", "glob"}], + "redirects": [{"target", "expand", "glob", "fd"}]}；features 是 + {"compound", "pipe", "redirect", "env", "glob", "assign"} 的子集 + （assign 由 classify 在分段校验时补）；denied 非 None 时整条 + 拒绝，原因是给算子看的迁移提示。 + 扫描按 POSIX 引号状态机：单引号内一切字面；双引号内 $ 仍 + 展开、反引号仍执行；反斜杠转义下一字符（双引号内只对 + $ ` " \\ 换行起转义，其余保留反斜杠——本扫描只关心元字符 + 是否被转义，取保守：反斜杠后的字符一律视为字面）。 + """ + if "\x00" in command: + return [], set(), "命令含 NUL 字节" + if "\n" in command or "\r" in command: + return [], set(), "命令含换行：多行命令请改写成 && 连接的单行" + segments = [] + features = set() + words = [] + redirects = [] + cur = [] + cur_expand = False + cur_glob = False + cur_started = False + pending_redirect = None # None | "file" | "fd"（>& / <& 后） + i = 0 + n = len(command) + state = "normal" # normal | squote | dquote + + def flush_word(): + nonlocal cur, cur_expand, cur_glob, cur_started, pending_redirect + if not cur_started: + return + word = {"text": "".join(cur), "expand": cur_expand, + "glob": cur_glob} + if pending_redirect is not None: + redirects.append({"target": word["text"], + "expand": word["expand"], + "glob": word["glob"], + "fd": pending_redirect == "fd"}) + pending_redirect = None + else: + words.append(word) + cur = [] + cur_expand = False + cur_glob = False + cur_started = False + + def end_segment(): + nonlocal words, redirects + flush_word() + if pending_redirect is not None: + return "重定向后缺目标" + segments.append({"words": words, "redirects": redirects}) + words = [] + redirects = [] + return None + + def dollar(i): + """处理 $ 后的展开：返回 (新下标, 拒绝原因或 None)。 + 调用方保证 command[i] == "$"。""" + nonlocal cur_started, cur_expand + nxt = command[i + 1] if i + 1 < n else "" + if nxt == "(": + return i, "命令替换 $(...) 不在白名单形态内" + if nxt == "{": + j = command.find("}", i + 2) + if j < 0: + return i, "${ 未闭合" + cur.append(command[i:j + 1]) + cur_started = True + cur_expand = True + features.add("env") + return j + 1, None + if nxt and (nxt.isalnum() or nxt in "@*#?$!-"): + cur.append("$") + cur_started = True + cur_expand = True + features.add("env") + return i + 1, None + cur.append("$") + cur_started = True + return i + 1, None + + while i < n: + c = command[i] + if state == "squote": + if c == "'": + state = "normal" + else: + cur.append(c) + i += 1 + continue + if state == "dquote": + if c == '"': + state = "normal" + i += 1 + continue + if c == "\\" and i + 1 < n and command[i + 1] in "$`\"\\\n": + cur.append(command[i + 1]) + i += 2 + continue + if c == "`": + return [], set(), "反引号命令替换不在白名单形态内" + if c == "$": + i, err = dollar(i) + if err: + return [], set(), err + continue + cur.append(c) + i += 1 + continue + # state == "normal" + if c in " \t": + flush_word() + i += 1 + continue + if c == "\\" and i + 1 < n: + cur.append(command[i + 1]) + cur_started = True + i += 2 + continue + if c == "'": + state = "squote" + cur_started = True + i += 1 + continue + if c == '"': + state = "dquote" + cur_started = True + i += 1 + continue + if c == "`": + return [], set(), "反引号命令替换不在白名单形态内" + if c == "$": + i, err = dollar(i) + if err: + return [], set(), err + continue + if c == "(" or c == ")": + return [], set(), "子 shell/分组括号不在白名单形态内" + if c == "{" or c == "}": + return [], set(), "花括号分组/展开不在白名单形态内" + if c == "#": + if not cur_started: + return [], set(), "行内注释 # 会截断可见命令，请去掉注释再写" + cur.append(c) + cur_started = True + i += 1 + continue + if c == "~": + if not cur_started or (cur and cur[-1] == "="): + cur_expand = True + features.add("env") + cur.append(c) + cur_started = True + i += 1 + continue + if c in "*?": + cur_glob = True + features.add("glob") + cur.append(c) + cur_started = True + i += 1 + continue + if c == "[": + # 段首（或仅跟赋值前缀后）且后跟空白的 [ 是 test 命令 + # 本身，不是 glob；其它位置的 [ 是 glob 形态。 + prefix = [w["text"] for w in words] + at_command = (not cur_started + and (i + 1 >= n or command[i + 1] in " \t") + and all(re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", t) + for t in prefix)) + if at_command: + cur.append(c) + cur_started = True + i += 1 + continue + cur_glob = True + features.add("glob") + cur.append(c) + cur_started = True + i += 1 + continue + if c == "!": + if not words and not cur_started: + return [], set(), "取反 ! 不在白名单形态内" + cur.append(c) + cur_started = True + i += 1 + continue + if c == "&": + if command[i + 1:i + 2] == ">": + flush_word() + features.add("redirect") + pending_redirect = "file" + i += 2 + if command[i:i + 1] == ">": + i += 1 + continue + if command[i + 1:i + 2] == "&": + err = end_segment() + if err: + return [], set(), err + features.add("compound") + i += 2 + continue + return [], set(), "后台 & 不在白名单形态内（验收须前台跑完才有 rc）" + if c == "|": + if command[i + 1:i + 2] == "|": + err = end_segment() + if err: + return [], set(), err + features.add("compound") + i += 2 + continue + err = end_segment() + if err: + return [], set(), err + features.add("pipe") + i += 1 + continue + if c == ";": + err = end_segment() + if err: + return [], set(), err + features.add("compound") + i += 1 + continue + if c in "><": + # 数字前缀是 fd（如 2>、2>&1），不是参数。 + if cur_started and "".join(cur).isdigit(): + cur = [] + cur_expand = False + cur_glob = False + cur_started = False + else: + flush_word() + if c == "<" and command[i + 1:i + 2] == "<": + return [], set(), "here-doc << 不在白名单形态内" + if command[i + 1:i + 2] == "&": + pending_redirect = "fd" + i += 2 + else: + pending_redirect = "file" + if c == ">": + i += 1 + if command[i:i + 1] == ">": + i += 1 + if command[i:i + 1] == "|": + i += 1 + else: + i += 1 + features.add("redirect") + continue + cur.append(c) + cur_started = True + i += 1 + if state != "normal": + return [], set(), "引号未闭合" + err = end_segment() + if err: + return [], set(), err + return segments, features, None + + +def _is_assign_word(text): + return re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", text) is not None + + +def classify_verify_command(command, cwd=None): + """验测字符串分类，返回 dict： + kind: "empty"（空/空白串，等价无验收）| "simple"（无 shell + 形态，走 shell=False）| "simple-env"（只有 VAR=val 前缀， + 走 shell=False + env）| "shell"（白名单内 shell 形态，走 + shell=True）| "denied"（拒绝执行，reason 给迁移提示）； + argv: simple/simple-env 时的 argv 数组（shell/denied 时 None）； + env: simple-env 时的赋值 dict（其它 kind 时 None）； + reason: denied 时的拒绝原因，其它 kind 时 ""； + features: 形态特征集合；notes: 审计用备注（如脚本路径因无 + cwd 未校验——执行时 cwd 一定有，以执行时结论为准）。 + """ + if command is None or not command.strip(): + return {"kind": "empty", "argv": None, "env": None, "reason": "", + "features": frozenset(), "notes": []} + segments, features, denied = _scan_shell_command(command) + if denied is not None: + return {"kind": "denied", "argv": None, "env": None, + "reason": denied, "features": features, "notes": []} + notes = [] + argv = None + env = None + for seg in segments: + words = seg["words"] + redirects = seg["redirects"] + if not words and not redirects: + return {"kind": "denied", "argv": None, "env": None, + "reason": "空段：结尾操作符后缺命令", + "features": features, "notes": notes} + if not words: + return {"kind": "denied", "argv": None, "env": None, + "reason": "只有重定向、没有命令", + "features": features, "notes": notes} + assigns = [] + idx = 0 + while idx < len(words) and _is_assign_word(words[idx]["text"]): + assigns.append(words[idx]["text"]) + idx += 1 + if assigns: + features = features | {"assign"} + if idx >= len(words): + return {"kind": "denied", "argv": None, "env": None, + "reason": "只有环境变量赋值、没有命令", + "features": features, "notes": notes} + cmd_word = words[idx]["text"] + if "/" in cmd_word: + if cmd_word.startswith("/"): + return {"kind": "denied", "argv": None, "env": None, + "reason": f"绝对路径命令不在白名单内：{cmd_word}", + "features": features, "notes": notes} + if ".." in cmd_word.split("/"): + return {"kind": "denied", "argv": None, "env": None, + "reason": f"越界路径命令不在白名单内：{cmd_word}", + "features": features, "notes": notes} + if cwd is not None: + if not os.path.isdir(cwd): + return {"kind": "denied", "argv": None, "env": None, + "reason": "脚本路径无法校验：cwd 不存在", + "features": features, "notes": notes} + base = os.path.realpath(cwd) + target = os.path.realpath(os.path.join(cwd, cmd_word)) + if os.path.commonpath([base, target]) != base: + return {"kind": "denied", "argv": None, "env": None, + "reason": f"脚本路径越出 cwd：{cmd_word}", + "features": features, "notes": notes} + if not os.path.isfile(target) or not os.access(target, os.X_OK): + return {"kind": "denied", "argv": None, "env": None, + "reason": f"仓内脚本不存在或不可执行：{cmd_word}", + "features": features, "notes": notes} + else: + notes.append(f"脚本路径未校验（无 cwd）：{cmd_word}") + elif cmd_word not in VERIFY_COMMAND_ALLOWLIST: + return {"kind": "denied", "argv": None, "env": None, + "reason": f"首词不在验测白名单内：{cmd_word}", + "features": features, "notes": notes} + for red in redirects: + target = red["target"] + is_fd = red["fd"] and (target.isdigit() or target == "-") + if is_fd: + continue + # >&file 这种写法在 shell 里是写文件，不是 fd 跳转： + # fd 标记但目标不是数字/- 时按文件目标校验。 + if red["expand"] or red["glob"]: + return {"kind": "denied", "argv": None, "env": None, + "reason": "重定向目标含变量展开/glob，无法静态校验", + "features": features, "notes": notes} + if target == "/dev/null": + continue + if target.startswith("/") or ".." in target.split("/"): + return {"kind": "denied", "argv": None, "env": None, + "reason": f"重定向目标不在 cwd 内：{target}", + "features": features, "notes": notes} + if not features: + argv = [w["text"] for w in segments[0]["words"]] + return {"kind": "simple", "argv": argv, "env": None, "reason": "", + "features": features, "notes": notes} + if features == {"assign"}: + seg = segments[0] + assigns = [] + idx = 0 + while idx < len(seg["words"]) and _is_assign_word( + seg["words"][idx]["text"]): + assigns.append(seg["words"][idx]["text"]) + idx += 1 + env = {} + for a in assigns: + name, value = a.split("=", 1) + env[name] = value + argv = [w["text"] for w in seg["words"][idx:]] + return {"kind": "simple-env", "argv": argv, "env": env, "reason": "", + "features": features, "notes": notes} + return {"kind": "shell", "argv": None, "env": None, "reason": "", + "features": features, "notes": notes} + + +def _audit_verify(state_path, command, kind, rc): + """验测执行审计：只写 host-run.log（state 同目录），写失败 + 静默——审计不能反过来搞垮验测本身。""" + if not state_path: + return + try: + log_event(state_path, + f"verify {kind} rc={rc} cmd={command[:200]}") + except Exception: + pass + + +def run_verify_argv(argv, cwd, timeout=600, audit_state_path=None): + """结构化验测：argv 数组走 shell=False，无 shell 解析。 + 空数组等价无验收（0, ""）；非字符串元素是调用方 bug，抛错 + 不静默——调用方（merge/goal）只传 CLI 解析出的字符串数组。""" + if not argv: return 0, "" - v = subprocess.run(command, shell=True, cwd=cwd, - capture_output=True, text=True, timeout=600) + if not isinstance(argv, (list, tuple)): + raise TypeError("verify_argv 须是字符串数组") + if not all(isinstance(a, str) for a in argv): + raise TypeError("verify_argv 须是字符串数组") + v = subprocess.run(list(argv), shell=False, cwd=cwd, + capture_output=True, text=True, timeout=timeout) + _audit_verify(audit_state_path, shlex.join(list(argv)), "argv", + v.returncode) return v.returncode, (v.stdout + v.stderr).strip()[:2000] -def merge_task(repo, worktree, branch, task_name, verify_command=None): +def run_gated_verify(command, cwd, timeout=None, audit_state_path=None): + """验测字符串的唯一执行口：run_verify 与轮次验收都走这里。 + simple/simple-env → shell=False；白名单内 shell 形态 → + shell=True（这是验测线唯一的 shell=True）；拒绝 → 不执行、 + 记审计、返回 126，调用方按验测失败处理（merge 回滚、goal + 记 unmet）。TimeoutExpired/OSError 原样抛出，调用方原有 + 的超时/中断处理不变。""" + if not command or not command.strip(): + return 0, "" + info = classify_verify_command(command, cwd=cwd) + kind = info["kind"] + if kind == "denied": + _audit_verify(audit_state_path, command, "denied", + _VERIFY_DENIED_RC) + return _VERIFY_DENIED_RC, ( + f"验收命令被白名单拒绝，未执行：{info['reason']}；" + f"请改用白名单内命令、verify_argv，或跑 " + f"host.py verify-audit 看迁移清单") + if kind in ("simple", "simple-env"): + env = None + if info["env"] is not None: + env = dict(os.environ) + env.update(info["env"]) + v = subprocess.run(info["argv"], shell=False, cwd=cwd, + capture_output=True, text=True, timeout=timeout, + env=env) + _audit_verify(audit_state_path, command, kind, v.returncode) + return v.returncode, (v.stdout + v.stderr).strip()[:2000] + # shell：白名单内形态，仍走 shell=True——这是本文件验测线 + # 唯一剩下的 shell=True（notify 的那处是另一条线，见 + # notify_failure 的定性，代码不动）。 + v = subprocess.run(command, shell=True, cwd=cwd, + capture_output=True, text=True, timeout=timeout) + _audit_verify(audit_state_path, command, "shell", v.returncode) + return v.returncode, (v.stdout + v.stderr).strip()[:2000] + + +def run_verify(command, cwd, audit_state_path=None): + if not command: + return 0, "" + return run_gated_verify(command, cwd, timeout=600, + audit_state_path=audit_state_path) + + +def build_verify_audit_report(state): + """纯函数：扫描 state 里带验收命令的任务，返回报告行列表。 + 每行：task（任务名）、kind（classify 结果；已有 argv 的记 + "argv"）、command（字符串原值截断 200）、features（排序后 + 的形态特征）、reason（denied 时）、argv（simple 时可迁移的 + 数组；simple-env 已走 shell=False 无需迁移，记 None）、 + notes、applied（调用方 --apply 后填 True）。 + 没有验收命令的任务（新形态 goal）不进报告——报告只回答 + “哪些旧任务带有 shell 验收命令”。""" + report = [] + for name, t in (state.get("tasks") or {}).items(): + if not isinstance(t, dict): + continue + argv = t.get("verify_argv") + if argv is not None: + report.append({"task": name, "kind": "argv", + "command": None, + "features": [], "reason": "", + "argv": (list(argv) if isinstance(argv, list) + else None), + "notes": [], "applied": False}) + continue + cmd = t.get("verify_command") + if not cmd: + continue + cwd = t.get("cwd") + info = classify_verify_command( + cmd, cwd=cwd if (cwd and os.path.isdir(cwd)) else None) + kind = info["kind"] + migratable = info["argv"] if kind == "simple" else None + report.append({"task": name, "kind": kind, + "command": cmd[:200], + "features": sorted(info["features"]), + "reason": info["reason"], + "argv": migratable, + "notes": list(info["notes"]), + "applied": False}) + return report + + +def _cmd_verify_audit(args, state): + """host.py verify-audit：只读扫描 + 可选 --apply。 + --apply 只转换 kind == "simple" 的任务（argv 与原字符串在 + 无 shell 形态时严格等价）；simple-env 已走 shell=False 无需 + 转；shell/denied 一律只列清单、不自动改——转换 shell 形态 + 必须算子逐条确认，绝不静默丢弃语义。""" + report = build_verify_audit_report(state) + applied = 0 + if args.apply: + tasks = state.get("tasks") or {} + for row in report: + if row["kind"] == "simple" and row["argv"] is not None: + t = tasks.get(row["task"]) + if isinstance(t, dict) and t.get("verify_argv") is None: + t["verify_argv"] = row["argv"] + row["applied"] = True + applied += 1 + if applied: + save_state(args.state, state) + if args.json: + print(json.dumps({"state": args.state, "applied": applied, + "tasks": report}, + indent=2, ensure_ascii=False)) + else: + print(f"verify-audit: state={args.state} 任务 {len(report)} 条") + for row in report: + feats = ",".join(row["features"]) or "-" + extra = "" + if row["reason"]: + extra = f" reason={row['reason']}" + if row["notes"]: + extra += " notes=" + ";".join(row["notes"]) + if row["applied"]: + extra += " [已转 verify_argv]" + elif row["kind"] == "simple": + extra += f" [可转 argv={row['argv']}]" + print(f" {row['task']}: {row['kind']} features={feats}{extra}") + counts = {} + for row in report: + counts[row["kind"]] = counts.get(row["kind"], 0) + 1 + print(f"汇总：{counts}；--apply 已转 {applied} 条") + denied = counts.get("denied", 0) + if denied: + print(f"注意：denied {denied} 条下次执行时将被阻塞（不执行），" + f"须先改写命令或改用 verify_argv") + if counts.get("shell", 0): + print("shell 形态的任务仍走 shell=True（白名单内形态），" + "迁移到 verify_argv 须算子逐条确认语义等价") + + +def merge_task(repo, worktree, branch, task_name, verify_command=None, + verify_argv=None, audit_state_path=None): """合并 + 验测 + 回滚（纯 git 操作，不依赖 server，可单测）。 返回 (ok, message)：验测失败时 reset 回合并前 HEAD，分支和 - worktree 保留给修复；验测通过时才删 worktree/分支。""" - if verify_command is None: + worktree 保留给修复；验测通过时才删 worktree/分支。 + verify_argv（白名单轨）：非 None 时优先于 verify_command 字符串， + 走 shell=False；空数组表示跳过验测（等价 --no-verify）。 + verify_command 为 None 且无 argv 时才用系统默认。""" + if verify_argv is None and verify_command is None: verify_command = detect_verify_command(repo) pre_head = git(["rev-parse", "HEAD"], repo).stdout.strip() # 合并前先清掉 worktree 里的 __pycache__ / *.pyc：agent 跑 @@ -1449,12 +2044,23 @@ def merge_task(repo, worktree, branch, task_name, verify_command=None): # 自动解成功，继续走下面的验测（不 return，落到验测分支） print(f"注意：合并冲突用 -X ours 自动解（重复加同一段），" f"新增行 {len(added)} 行已核对都在主仓里。") - if verify_command: - rc, out = run_verify(verify_command, repo) - if rc != 0: - git(["reset", "--hard", pre_head], repo) - return False, (f"合并后验测失败，已回滚到 {pre_head[:8]}，" - f"分支 {branch} 和 worktree 保留：\n{out[:1000]}") + verify_rc = None + verify_out = "" + verify_ran = False + if verify_argv is not None: + if verify_argv: + verify_rc, verify_out = run_verify_argv( + verify_argv, repo, timeout=600, + audit_state_path=audit_state_path) + verify_ran = True + elif verify_command: + verify_rc, verify_out = run_verify( + verify_command, repo, audit_state_path=audit_state_path) + verify_ran = True + if verify_ran and verify_rc != 0: + git(["reset", "--hard", pre_head], repo) + return False, (f"合并后验测失败，已回滚到 {pre_head[:8]}，" + f"分支 {branch} 和 worktree 保留：\n{verify_out[:1000]}") git(["worktree", "remove", "--force", worktree], repo) git(["branch", "-d", branch], repo) return True, f"merged {branch} into {repo}" @@ -2004,7 +2610,8 @@ def _llm_declared_done(text): def run_goal_loop(server, state, state_path, task, goal_text, verify, max_rounds, model, effort, sandbox, cfg=None, timeout=None, verify_timeout=600, start_round=1, - last_verify="", last_turn_status=None, last_turn_text=""): + last_verify="", last_turn_status=None, last_turn_text="", + verify_argv=None): """目标循环（长任务版，对齐 Codex App 的可续跑语义）。 两种验收形态（2026-10-27 起，console goal 模式走新形态）： @@ -2015,6 +2622,10 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, 一直跑到模型宣告、预算耗尽或用户取消为止（与 turn_timeout=0 不限时同约定）。轮次提示里会把标记约定写明，模型才知道怎么 宣告；同一 thread 保留历史，续轮提示可以很短。 + verify_argv（白名单轨）：argv 数组时优先于 verify 字符串， + 走 shell=False；两者都没有才走新形态。verify_argv 由 CLI + --verify-argv 或任务记录的 verify_argv 来，goal-resume 原样 + 重放；新开目标循环时 verify_argv=None 会清掉旧记录。 每轮结束立即落盘（检查点），进程被杀 / VM 回收后可用 goal-resume 从 goal_rounds_done 的下一轮续跑，而不是从头重来。 与云端 worker 的解析契约（cloud/sandbox.py）： @@ -2035,8 +2646,18 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, goal-unmet 穷尽轮次后不允许 goal-resume（按既有规则须新建 任务），goal-resume 只接受 goal-running / interrupted / running。 """ + # 验测形态（白名单轨）：verify_argv 优先于 verify 字符串；两者 + # 都没有时走模型自决新形态。verify_command 存字符串原值（可 + # 能为 None），verify_argv 存数组（None 表示未用，顺带清掉 + # 旧任务残留的 argv，避免 goal-resume 重放过期 argv）。 + # verify_text 是喂给 agent 看的验收命令：argv 时用 shlex.join + # 还原成可读的一行（执行仍走 argv，不经过 shell 解析）。 + verify_text = (shlex.join(verify_argv) + if verify_argv is not None else verify) + has_verify = verify_text is not None task["goal"] = goal_text task["verify_command"] = verify + task["verify_argv"] = verify_argv task["goal_max_rounds"] = int(max_rounds) task["goal_rounds_done"] = int(start_round) - 1 task["goal_last_verify"] = last_verify @@ -2090,7 +2711,7 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, break if round_no == 1 and not last_verify and not last_turn_status: prompt = goal_text - if verify is None: + if not has_verify: # 新形态首轮必须把标记约定写进提示，否则模型不知 # 道怎么宣告完成，循环永远等不到结束信号。 prompt = (f"{goal_text}\n\n" @@ -2103,14 +2724,14 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, prompt = (f"上一轮回合未完成（状态：{last_turn_status}），" f"请继续完成目标，不要重复已完成的部分。\n" f"目标：{goal_text}\n") - if verify is not None: - prompt += (f"验收命令：{verify}\n" + if has_verify: + prompt += (f"验收命令：{verify_text}\n" f"上一轮验收输出：\n{last_verify}\n") else: prompt += (f"确认目标已完全达成时，在回复末尾单独一行输出：" f"{GOAL_DONE_MARKER}。\n") prompt += f"上一轮部分文本：\n{last_turn_text[:1000]}" - elif verify is None: + elif not has_verify: # 新形态续轮：同一 thread 已有历史，只需说明“还没 # 宣告达成、继续做”，并重申标记约定。 prompt = (f"上一轮未宣告目标达成，请继续完成目标，" @@ -2124,7 +2745,7 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, # 必须把验收命令和失败输出喂回去，才是可用的目标循环。 prompt = (f"上一轮验收未通过，请针对性修复，不要重复已完成的部分。\n" f"目标：{goal_text}\n" - f"验收命令：{verify}\n" + f"验收命令：{verify_text}\n" f"上一轮验收输出：\n{last_verify}") try: result = run_task_turn(server, state, task, prompt, @@ -2144,23 +2765,26 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, Exception("连续 3 轮回合失败，停止空转")) else: consecutive_failed = 0 - if verify is None: + if not has_verify: # 新形态没有验收命令可跑，v_rc 记 -（轮次行格式不变， # 云端解析只读 verify_out 做 unmet 详情，不受影响）。 v_rc, v_out = None, "" else: try: v_rc, v_out = None, None - if verify_timeout and verify_timeout > 0: - v = subprocess.run(verify, shell=True, cwd=task["cwd"], - capture_output=True, text=True, - timeout=verify_timeout) + vt = verify_timeout if (verify_timeout + and verify_timeout > 0) else None + if verify_argv is not None: + v_rc, v_out = run_verify_argv( + verify_argv, task["cwd"], timeout=vt, + audit_state_path=state_path) else: - v = subprocess.run(verify, shell=True, cwd=task["cwd"], - capture_output=True, text=True) - v_rc = v.returncode - v_out = (v.stdout + v.stderr).strip()[:1000] or \ - f"exit code {v.returncode}" + v_rc, v_out = run_gated_verify( + verify, task["cwd"], timeout=vt, + audit_state_path=state_path) + # 轮次行/goal_last_verify 保持旧口径：截断 1000， + # 空输出时记 exit code（旧代码同款，不改契约）。 + v_out = v_out[:1000] or f"exit code {v_rc}" except subprocess.TimeoutExpired as exc: # 验测本身也可能很长（测试套件）：超时是验测失败， # 不是进程挂死，喂回下一轮让 agent 缩小验测范围。 @@ -2208,7 +2832,7 @@ def run_goal_loop(server, state, state_path, task, goal_text, verify, ensure_ascii=False) + "\n") except Exception: pass - if verify is None: + if not has_verify: # 模型自决：只有完整跑完的回合里的标记行才算数。 if (result["status"] == "completed" and _llm_declared_done(last_turn_text)): @@ -2348,6 +2972,9 @@ def main(): p.add_argument("--verify", default=None, help="合并后验测命令；不传时仓库有 tests/ 就默认跑 unittest") p.add_argument("--no-verify", action="store_true") + p.add_argument("--verify-argv", nargs=argparse.REMAINDER, default=None, + help="合并后验测 argv（优先于 --verify，走 shell=False；" + "其后全部参数都作为 argv，须放在命令最后）") p = sub.add_parser("review-comment") p.add_argument("--task", required=True) p.add_argument("--file", required=True) @@ -2430,6 +3057,9 @@ def main(): p.add_argument("--verify", default=None, help="验收命令；不传时走模型自决：模型在回合末尾" "单独一行输出“目标已达成”即达标") + p.add_argument("--verify-argv", nargs=argparse.REMAINDER, default=None, + help="验收 argv（优先于 --verify，走 shell=False；" + "其后全部参数都作为 argv，须放在命令最后）") p.add_argument("--max-rounds", type=int, default=3, help="轮次上限；0 表示不限轮次（模型宣告完成才结束），" "默认 3") @@ -2447,8 +3077,17 @@ def main(): help="不传时沿用原任务的 goal_max_rounds；传了则覆盖总轮次") p.add_argument("--verify-timeout", type=int, default=600, help="验测命令上限（秒）；0 表示不限时，默认 600") + p.add_argument("--verify-argv", nargs=argparse.REMAINDER, default=None, + help="断点续跑时覆盖任务的验收 argv（走 shell=False；" + "其后全部参数都作为 argv，须放在命令最后）") p = sub.add_parser("parallel") p.add_argument("--jobs", required=True) + p = sub.add_parser("verify-audit") + p.add_argument("--apply", action="store_true", + help="只转换 simple 形态的任务为 verify_argv；" + "shell/denied 只列清单、不自动改") + p.add_argument("--json", action="store_true", + help="以 JSON 输出报告（默认文本报告）") p = sub.add_parser("automations") p.add_argument("--add", action="store_true") p.add_argument("--name") @@ -2852,6 +3491,18 @@ def main(): pass print("host daemon stopped", flush=True) return + # verify-audit 是只读扫描工具（+ 可选 --apply 写回），绝不 + # 启动 codex server：放在 srv 构造之前早返回，否则每次审计 + # 都要拉起 app-server 进程。 + # --verify-argv 用 REMAINDER：空数组表示算子写了标志却没给 + # 参数，那是笔误不是“跳过验测”，直接报 usage 退出（exit 2）， + # 绝不能静默当成 --no-verify。 + if (args.cmd in ("merge", "goal", "goal-resume") + and getattr(args, "verify_argv", None) == []): + ap.error("--verify-argv 后须至少给一个参数") + if args.cmd == "verify-audit": + _cmd_verify_audit(args, state) + return srv = Server(codex_bin=args.codex_bin, codex_home=args.codex_home, approval=args.approval, progress_interval=args.progress_interval, @@ -2940,9 +3591,15 @@ def main(): # unittest，--no-verify 跳过，--verify 可指定命令。 # 验测失败时 merge_task 已 reset 回合并前 HEAD，分支和 # worktree 保留，任务标 merge-failed，不删改动。 + # --no-verify 时 argv 一并跳过；两者都给时 argv 优先。 verify_cmd = "" if args.no_verify else args.verify + verify_argv = None if args.no_verify else args.verify_argv + if verify_cmd and verify_argv: + print("verify_argv 已指定，--verify 字符串被忽略") ok, msg = merge_task(t["repo"], t["cwd"], t["branch"], - args.task, verify_command=verify_cmd) + args.task, verify_command=verify_cmd, + verify_argv=verify_argv, + audit_state_path=args.state) print(msg) if ok: t["status"] = "merged" @@ -3342,11 +3999,14 @@ def main(): args.approval_policy, seed_dir=args.seed_dir) save_state(args.state, state) + if args.verify and args.verify_argv: + print("verify_argv 已指定，--verify 字符串被忽略") run_goal_loop(srv, state, args.state, task, args.task, args.verify, args.max_rounds, args.model, args.effort, args.sandbox, cfg=cfg, timeout=args.turn_timeout, - verify_timeout=args.verify_timeout) + verify_timeout=args.verify_timeout, + verify_argv=args.verify_argv) elif args.cmd == "goal-resume": # 断点丢失守卫（2026-10-29 长任务实弹）：host-state 里没有 # 该任务时直接拒绝、正常退出，绝不能让 KeyError traceback @@ -3391,7 +4051,8 @@ def main(): start_round=1, last_verify="", last_turn_status=None, - last_turn_text="") + last_turn_text="", + verify_argv=args.verify_argv) return if t.get("status") not in ("goal-running", "interrupted", "running"): @@ -3413,12 +4074,19 @@ def main(): # P1：断点续跑时旧验收形态的 0 轮次（脏数据，0 在旧形态 # 里表示不限轮次）不按不限轮次跑，钳制到 100 轮（与云端 # GOAL_MAX_ROUNDS_CAP 同值）；新形态（verify 为空）的 0 - # 保持不限轮次，上界是云端配额。 - if max_rounds == 0 and t.get("verify_command"): + # 保持不限轮次，上界是云端配额。argv 形态同样是旧形态， + # 钳制条件须把 verify_argv 也算上，否则 argv 任务会被 + # 当成新形态不限轮次跑。 + if max_rounds == 0 and (t.get("verify_command") + or t.get("verify_argv") + or args.verify_argv): max_rounds = 100 srv.resume_thread(t["thread_id"]) last_turn = (t.get("turns") or [{}])[-1] last_result = last_turn.get("result") or {} + resume_argv = (args.verify_argv + if args.verify_argv is not None + else t.get("verify_argv")) run_goal_loop(srv, state, args.state, t, t["goal"], t["verify_command"], max_rounds, args.model, args.effort, args.sandbox, cfg=cfg, @@ -3427,7 +4095,8 @@ def main(): start_round=int(t.get("goal_rounds_done") or 0) + 1, last_verify=t.get("goal_last_verify") or "", last_turn_status=t.get("goal_last_turn_status"), - last_turn_text=last_result.get("text", "") or "") + last_turn_text=last_result.get("text", "") or "", + verify_argv=resume_argv) elif args.cmd == "parallel": with open(args.jobs) as f: jobs = json.load(f) diff --git a/tests/test_host_shell_policy.py b/tests/test_host_shell_policy.py new file mode 100644 index 0000000..32821d8 --- /dev/null +++ b/tests/test_host_shell_policy.py @@ -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/ < /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()