Agent-System-Dev/scripts/zcode/write_guard.py

#!/usr/bin/env python3
"""Check every write target; a local branch guard, not an OS sandbox."""
import json
import os
from pathlib import Path
import re
import subprocess
import sys


def deny(reason):
    raise ValueError(reason)


def path_value(value):
    if not isinstance(value, str) or not value.strip() or any(c in value for c in "\x00\r\n"):
        deny("写入路径缺失或无法解析")
    return value


def patch_paths(patch):
    if not isinstance(patch, str):
        deny("ApplyPatch 缺少 patch_text")
    lines = patch.strip().splitlines()
    if len(lines) < 3 or lines[0] != "*** Begin Patch" or lines[-1] != "*** End Patch":
        deny("无法解析 ApplyPatch 边界")
    paths = []
    operation = None
    moved = False
    for line in lines[1:-1]:
        match = re.fullmatch(r"\*\*\* (Add File|Update File|Delete File): (.+)", line)
        if match:
            operation = match[1]
            moved = False
            paths.append(path_value(match[2]))
        elif line.startswith("*** Move to: ") and operation == "Update File" and not moved:
            paths.append(path_value(line[len("*** Move to: "):]))
            moved = True
        elif line == "*** End of File" and operation == "Update File":
            continue
        elif line.startswith("***") or operation is None:
            deny("ApplyPatch 包含未知路径指令")
        elif not (line.startswith(("+", "-", " ", "@@")) or line == ""):
            deny("ApplyPatch 包含无法解析的内容")
    if not paths:
        deny("ApplyPatch 没有可检查的路径")
    return paths


def check_path(raw, cwd):
    target = Path(raw)
    if not target.is_absolute():
        target = cwd / target
    # A tool can follow a symlink OR replace/delete its directory entry. Check
    # both locations, including a dangling final symlink on a protected branch.
    location = target.parent.resolve() / target.name
    check_location(location, entry_only=location.is_symlink())
    resolved = target.resolve()
    if resolved != location:
        check_location(resolved)


def check_location(target, entry_only=False):
    ancestor = target if target.is_dir() and not entry_only else target.parent
    while not ancestor.exists():
        ancestor = ancestor.parent
    if not ancestor.is_dir():
        deny(f"目标祖先不是目录：{ancestor}")
    # GIT_DIR/GIT_WORK_TREE must not redirect inspection to another checkout.
    env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")}
    env["LC_ALL"] = "C"
    result = subprocess.run(["git", "-C", str(ancestor), "rev-parse", "--show-toplevel"],
                            env=env, capture_output=True, text=True)
    if result.returncode:
        if any((p / ".git").exists() or (p / ".git").is_symlink()
               for p in (ancestor, *ancestor.parents)):
            deny(f"无法读取目标仓库状态：{target}")
        if "not a git repository" not in result.stderr:
            deny(f"无法确认目标仓库：{target}")
        print(f"write-guard: outside-git（未受 Git 分支规则覆盖）：{target}", file=sys.stderr)
        return
    repo = result.stdout.strip()
    branch_result = subprocess.run(["git", "-C", repo, "symbolic-ref", "--quiet", "--short", "HEAD"],
                                   env=env, capture_output=True, text=True)
    if branch_result.returncode:
        deny(f"仓库处于 detached HEAD 或状态不可读：{repo}；先建立任务分支（A1）")
    branch = branch_result.stdout.strip()
    if not re.fullmatch(r"(?:feature|fix|hotfix|release)/[A-Za-z0-9._-]+", branch):
        deny(f"分支 {branch} 不允许写入（目标 {target}，仓库 {repo}）；先 make task-start NAME=feature/<slug>（A1/A3）")


def main():
    data = json.load(sys.stdin)
    if not isinstance(data, dict):
        deny("钩子输入必须为对象")
    tool = data.get("tool_name")
    if tool in {"Read", "Glob", "Grep", "WebFetch", "WebSearch"}:
        return
    if tool not in {None, "Edit", "Write", "ApplyPatch", "apply_patch"}:
        deny("未知写入工具，必须先补充适配")
    payload = data.get("tool_input", data)
    if isinstance(payload, str) and tool in {"ApplyPatch", "apply_patch"}:
        paths = patch_paths(payload)
    elif isinstance(payload, dict):
        # Installed zcode uses patch_text; aliases use the same structured format.
        patches = [payload[k] for k in ("patch_text", "patch", "input") if k in payload]
        paths = [path_value(payload[k]) for k in ("file_path", "path", "filename") if k in payload]
        for patch in patches:
            paths.extend(patch_paths(patch))
    else:
        deny("无法解析写入工具参数")
    if not paths:
        deny("写入请求未提供可检查路径")
    cwd = Path(path_value(data.get("cwd", os.getcwd()))).resolve()
    if not cwd.is_dir():
        deny("钩子 cwd 不存在")
    for path in paths:
        check_path(path, cwd)


if __name__ == "__main__":
    try:
        main()
    except (ValueError, OSError, RuntimeError) as error:
        print(f"write-guard: 拒绝——{error}", file=sys.stderr)
        sys.exit(2)