#!/usr/bin/env python3 """Codex PreToolUse guard: block high-risk ops without crude path false-positives. Design: - Prefer content / precise path checks over basename/suffix blanket bans. - Read protection focuses on secrets entering model context. - Write protection focuses on credential vaults & shell startup hijack. - Delete protection covers vaults, shell rc, credential files, and OS/home roots — not a blanket ban on `rm` / `sudo rm -rf`. - Allowed `rm`/`rmdir` (not sudo) are rewritten to a trash command so deletes go to Trash instead of unlinking: macOS `/usr/bin/trash`, Linux `trash-put` / `gio trash`, otherwise `codex_trash.py` (XDG). zsh aliases are not used: Codex snapshots run `unalias -a`. - .env: only deny when sensitive keys have real values (not placeholders). """ from __future__ import annotations import json import hashlib import sqlite3 import time import uuid import os import re import shlex import shutil import sys from pathlib import Path HOME = str(Path.home()) # --------------------------------------------------------------------------- # Credential vaults under $HOME — write always denied; read with allow-list # --------------------------------------------------------------------------- VAULT_PREFIXES = ( f"{HOME}/.ssh", f"{HOME}/.gnupg", f"{HOME}/.aws", f"{HOME}/.kube", ) # Non-secret (or low-risk) files that agents often need to inspect VAULT_READ_ALLOW: dict[str, frozenset[str]] = { f"{HOME}/.ssh": frozenset( { "config", "known_hosts", "known_hosts.old", "authorized_keys", "authorized_keys2", } ), f"{HOME}/.aws": frozenset({"config"}), # kube/config almost always embeds tokens — do not allow by default f"{HOME}/.gnupg": frozenset({"gpg.conf", "dirmngr.conf", "gpg-agent.conf"}), } READ_EXACT = frozenset( { f"{HOME}/.netrc", f"{HOME}/.codex/auth.json", } ) # Shell startup files: block shell redirects / tee / sed -i, not all agent edits # (Write/Edit tools are allowed — user often asks agent to tweak aliases) WRITE_SHELL_EXACT = frozenset( { f"{HOME}/.zshrc", f"{HOME}/.bashrc", f"{HOME}/.bash_profile", f"{HOME}/.profile", f"{HOME}/.zprofile", f"{HOME}/.zshenv", } ) WRITE_COMMANDS = frozenset( { "tee", "cp", "mv", "install", "touch", "mkdir", "rmdir", "rm", "trash", "chmod", "chown", "chgrp", "ln", "truncate", } ) DELETE_COMMANDS = frozenset({"rm", "rmdir", "trash"}) REWRITE_TO_TRASH = frozenset({"rm", "rmdir"}) SHELL_SEPARATORS = frozenset({"|", "||", "&&", ";"}) _CODEX_TRASH = Path(__file__).with_name("codex_trash.py") # sudo/env/nice 等包装命令：剥掉后再判断真实操作，避免 sudo rm 被一刀切 COMMAND_WRAPPERS = frozenset( { "sudo", "command", "nice", "nohup", "time", "env", "stdbuf", "chronic", "ionice", "unbuffer", } ) SUDO_VALUE_FLAGS = frozenset( { "-u", "--user", "-g", "--group", "-h", "--host", "-C", "--close-from", "-D", "--chdir", "-R", "--chroot", "-T", "--command-timeout", "-U", "--other-user", "-p", "--prompt", } ) # Commands that typically pull file contents into context READISH_COMMANDS = frozenset( { "cat", "head", "tail", "less", "more", "bat", "batcat", "nl", "od", "hexdump", "xxd", "strings", "source", ".", "type", # sometimes used oddly } ) # Grep-family: only treat path-like operands as read targets GREP_COMMANDS = frozenset({"rg", "grep", "egrep", "fgrep", "ag", "ack", "git"}) ENV_SAFE_BASENAMES = frozenset( { ".env.example", ".env.sample", ".env.template", ".env.example.local", } ) # Project development environments under this trusted workspace tree may be # read by agents, including through symlinks created in temporary worktrees. ENV_READ_ALLOW_PREFIXES = (f"{HOME}/workspaces",) # Key-material extensions — content-scanned, not path-banned KEY_MATERIAL_SUFFIXES = (".pem", ".p12", ".pfx", ".key") KEY_MATERIAL_BASENAMES = frozenset( { "id_rsa", "id_dsa", "id_ecdsa", "id_ed25519", "id_ed25519_sk", "id_ecdsa_sk", } ) PRIVATE_KEY_MARKERS = ( "-----BEGIN PRIVATE KEY-----", "-----BEGIN RSA PRIVATE KEY-----", "-----BEGIN EC PRIVATE KEY-----", "-----BEGIN DSA PRIVATE KEY-----", "-----BEGIN OPENSSH PRIVATE KEY-----", "-----BEGIN ENCRYPTED PRIVATE KEY-----", ) # Public / non-secret PEM material PUBLIC_PEM_MARKERS = ( "-----BEGIN CERTIFICATE-----", "-----BEGIN PUBLIC KEY-----", "-----BEGIN RSA PUBLIC KEY-----", "-----BEGIN SSH2 PUBLIC KEY-----", "ssh-rsa ", "ssh-ed25519 ", "ecdsa-sha2-", "sk-ssh-ed25519@", "sk-ecdsa-sha2-", ) ENV_SCAN_MAX_BYTES = 256 * 1024 KEY_SCAN_MAX_BYTES = 64 * 1024 MIN_SECRET_VALUE_LEN = 8 SENSITIVE_KEY_RE = re.compile( r"(?:" r"(?:^|_)(?:" r"ACCESS_?KEY(?:_ID|_SECRET|_ID_SECRET)?" r"|SECRET_?ACCESS_?KEY" r"|SECRET_?KEY" r"|API[_-]?KEY" r"|APP[_-]?KEY" r"|PRIVATE[_-]?KEY" r"|CLIENT[_-]?(?:SECRET|TOKEN|KEY)" r"|AUTH[_-]?TOKEN" r"|ACCESS[_-]?TOKEN" r"|REFRESH[_-]?TOKEN" r"|BEARER[_-]?TOKEN" r"|SESSION[_-]?TOKEN" r"|SESSION[_-]?SECRET" r"|JWT[_-]?SECRET" r"|SIGNING[_-]?SECRET" r"|WEBHOOK[_-]?SECRET" r"|ENCRYPTION[_-]?KEY" r"|MASTER[_-]?KEY" r"|SERVICE[_-]?ACCOUNT[_-]?KEY" r"|PASSWORD" r"|PASSWD" r"|CREDENTIALS?" r"|CONNECTION[_-]?STRING" r"|DATABASE[_-]?URL" r"|DB[_-]?URL" r"|MONGO(?:DB)?[_-]?URI" r"|REDIS[_-]?URL" r"|DSN\b" r")" r"|(?:^|_)(?:" r"AWS|ALIYUN|ALIBABA|OSS|ACS|TENCENT|QCLOUD|COS|" r"HUAWEI|HWC|AZURE|GCP|GOOGLE|MINIO|S3|R2|" r"CLOUDFLARE|STRIPE|OPENAI|ANTHROPIC|CLERK|" r"SUPABASE|FIREBASE|GITHUB|GITLAB|SLACK|DISCORD|" r"TWILIO|SENDGRID|MAILGUN|SMTP|NPM|DOCKER|HF|" r"HUGGINGFACE|VERCEL|NETLIFY|DIGITALOCEAN|LINODE|" r"DO_SPACES|WASABI|BACKBLAZE|B2" r")[_-]?(?:" r"AK|SK|ACCESS_?KEY(?:_ID|_SECRET)?" r"|SECRET(?:_?KEY|_?ACCESS_?KEY)?" r"|API[_-]?KEY" r"|TOKEN" r"|PASSWORD" r"|PRIVATE[_-]?KEY" r"|CLIENT[_-]?SECRET" r")" r")", re.IGNORECASE, ) SENSITIVE_VALUE_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = ( ("AWS Access Key ID", re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b")), ("Aliyun AccessKey ID", re.compile(r"\bLTAI[0-9A-Za-z]{12,20}\b")), ("Tencent SecretId", re.compile(r"\bAKID[0-9A-Za-z]{13,32}\b")), ("GitHub PAT", re.compile(r"\b(?:ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9]{36,}\b")), ("GitHub fine-grained PAT", re.compile(r"\bgithub_pat_[A-Za-z0-9_]{20,}\b")), ("GitLab PAT", re.compile(r"\bglpat-[A-Za-z0-9\-_]{20,}\b")), ("Slack token", re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b")), ("Stripe secret key", re.compile(r"\bsk_(?:live|test)_[A-Za-z0-9]{16,}\b")), ("OpenAI-style key", re.compile(r"\bsk-[A-Za-z0-9]{20,}\b")), ("Anthropic key", re.compile(r"\bsk-ant-[A-Za-z0-9\-_]{20,}\b")), ("Google API key", re.compile(r"\bAIza[0-9A-Za-z\-_]{35}\b")), ("JWT", re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b")), ("Private key block", re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----")), ("Bearer-like high-entropy secret", re.compile(r"\b[A-Za-z0-9+/_-]{40,}={0,2}\b")), ) PLACEHOLDER_VALUES = { "", "''", '""', "null", "none", "nil", "undefined", "todo", "tbd", "fixme", "changeme", "change-me", "change_me", "placeholder", "your-key", "your_key", "your-api-key", "your_api_key", "your-secret", "your_secret", "your-token", "your_token", "xxx", "xxxx", "xxxxx", "xxxxxx", "xxxxxxxx", "example", "sample", "test", "testing", "dummy", "fake", "secret", "password", "passwd", "token", "apikey", "api_key", "api-key", "access_key", "access-key", "secret_key", "secret-key", "redacted", "", "[redacted]", "replace_me", "replace-me", "insert_here", "insert-here", "n/a", "na", "-", "--", "...", "****", "*****", "******", } PLACEHOLDER_PREFIXES = ( "your-", "your_", "my-", "my_", "example-", "example_", "sample-", "sample_", "changeme", "change-me", "replace-", "replace_", "insert-", "insert_", "xxx", "<", "[", "${", "$(", ) # Local/dev connection strings without credentials are not treated as secrets LOCAL_HOSTS = frozenset( { "localhost", "127.0.0.1", "0.0.0.0", "::1", "host.docker.internal", "db", "redis", "postgres", "mysql", "mongo", "mongodb", } ) def _expand_home_refs(value: str) -> str: value = value.replace("${HOME}", HOME).replace("$HOME", HOME) if value.startswith("~"): value = os.path.expanduser(value) return value def _deny(reason: str) -> int: payload = { "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": reason, } } json.dump(payload, sys.stdout, ensure_ascii=False) sys.stdout.write("\n") return 0 def _allow_updated_command(payload: dict, new_command: str, reason: str) -> int: tool_input = dict(payload.get("tool_input") or {}) tool_input["cmd" if "cmd" in tool_input and "command" not in tool_input else "command"] = new_command json.dump( { "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "allow", "permissionDecisionReason": reason, "updatedInput": tool_input, } }, sys.stdout, ensure_ascii=False, ) sys.stdout.write("\n") return 0 def _normalize_path(token: str, cwd: str | None = None) -> str: target = _expand_home_refs(token.strip("\"'")) if not target: return target if not os.path.isabs(target): base = cwd if cwd and os.path.isabs(cwd) else os.getcwd() target = os.path.join(base, target) return os.path.abspath(target) def _path_under(prefix: str, path: str) -> bool: return path == prefix or path.startswith(prefix + os.sep) def _vault_prefix_for(path: str) -> str | None: for prefix in VAULT_PREFIXES: if _path_under(prefix, path): return prefix return None def _is_vault_read_allowed(path: str) -> bool: """Allow low-risk files inside credential vaults (config, *.pub, …).""" vault = _vault_prefix_for(path) if vault is None: return False # Directory itself (e.g. ls ~/.ssh) — allow; listing names ≠ reading secrets if path.rstrip("/") == vault: return True base = os.path.basename(path) if base.endswith(".pub"): return True allow = VAULT_READ_ALLOW.get(vault, frozenset()) if base in allow: return True # ~/.aws/config-* style if vault == f"{HOME}/.aws" and base.startswith("config"): return True return False def _looks_like_path_token(token: str) -> bool: if not token or token.startswith("-"): return False # shell operators if token in {"|", "||", "&", "&&", ";", ">", ">>", "<", "<<", "(", ")", "{", "}"}: return False if token.startswith("1>") or token.startswith("2>"): return False t = token.strip("\"'") if not t: return False if t.startswith("~") or t.startswith("/") or t.startswith("./") or t.startswith("../"): return True if "/" in t or t.startswith("."): return True # bare sensitive basenames base = os.path.basename(t) if base in KEY_MATERIAL_BASENAMES or base in READ_EXACT or _is_env_basename(base): return True if any(base.endswith(s) for s in KEY_MATERIAL_SUFFIXES): return True return False def _is_env_basename(basename: str) -> bool: if basename in ENV_SAFE_BASENAMES: return False return basename == ".env" or basename.startswith(".env.") def _looks_like_env_file(token: str) -> bool: raw = _expand_home_refs(token.strip("\"'")) return _is_env_basename(os.path.basename(raw.rstrip("/"))) def _is_env_read_allowed(path: str) -> bool: """Allow env files whose actual target lives in a trusted workspace tree.""" real_path = os.path.realpath(path) return any(_path_under(os.path.realpath(prefix), real_path) for prefix in ENV_READ_ALLOW_PREFIXES) def _looks_like_key_material_path(token: str) -> bool: raw = _expand_home_refs(token.strip("\"'")) base = os.path.basename(raw.rstrip("/")) if base in KEY_MATERIAL_BASENAMES: return True # id_rsa.pub is public if base.endswith(".pub"): return False lower = base.lower() return any(lower.endswith(s) for s in KEY_MATERIAL_SUFFIXES) def _read_text_head(path: str, limit: int) -> str | None: try: if not os.path.isfile(path): return None size = os.path.getsize(path) if size <= 0: return "" with open(path, "rb") as fh: data = fh.read(min(size, limit)) if b"\x00" in data: return None return data.decode("utf-8", errors="replace") except OSError: return None def _private_key_content_reason(content: str) -> str | None: upper = content for marker in PRIVATE_KEY_MARKERS: if marker in upper: return f"文件内容含私钥材料 ({marker})" # PKCS#12 is binary — handled by null-byte path; if base64 armor appears: if "-----BEGIN PKCS12-----" in upper: return "文件内容含 PKCS12 材料" return None def _is_public_only_key_content(content: str) -> bool: if _private_key_content_reason(content): return False if not content.strip(): return True # Only public markers / whitespace / comments if any(m in content for m in PUBLIC_PEM_MARKERS): return True # Short cert-like without private markers: allow return False def _key_material_block_reason(token: str, cwd: str | None = None) -> str | None: """Block reading key material only when content is private (or vault private).""" if not _looks_like_key_material_path(token): return None path = _normalize_path(token, cwd) # Inside vault: allow only allow-listed / *.pub if _vault_prefix_for(path): if _is_vault_read_allowed(path): return None return "已拦截敏感读取: 凭据目录中的密钥材料" content = _read_text_head(path, KEY_SCAN_MAX_BYTES) if content is None: # Missing / binary (e.g. .p12): if suffix is high-risk container, deny lower = path.lower() if lower.endswith(".p12") or lower.endswith(".pfx"): if os.path.isfile(path): return "已拦截敏感读取: PKCS12/PFX 密钥容器" return None if content == "": return None if _is_public_only_key_content(content): return None hit = _private_key_content_reason(content) if hit: return f"已拦截敏感读取: {hit}" # Unknown content under .key/.pem — do not blanket deny (fixtures, empty placeholders) return None def _is_placeholder_value(value: str) -> bool: v = value.strip() if not v: return True if len(v) >= 2 and v[0] == v[-1] and v[0] in {"'", '"'}: v = v[1:-1].strip() if not v: return True lower = v.lower() if lower in PLACEHOLDER_VALUES: return True if any(lower.startswith(p) for p in PLACEHOLDER_PREFIXES): return True if re.fullmatch(r"[xX*]{3,}", v): return True if re.fullmatch(r"\$\{[^}]+\}", v) or re.fullmatch(r"\$\([^\)]+\)", v): return True if re.fullmatch(r"<[^>]+>", v) or re.fullmatch(r"\[[^\]]+\]", v): return True if lower in {"true", "false", "0", "1", "yes", "no", "on", "off"}: return True return False def _strip_value_quotes(value: str) -> str: v = value.strip() if len(v) >= 2 and v[0] == v[-1] and v[0] in {"'", '"'}: return v[1:-1].strip() return v def _has_real_secret_value(value: str) -> bool: if _is_placeholder_value(value): return False v = _strip_value_quotes(value) if len(v) < MIN_SECRET_VALUE_LEN: return False if len(set(v)) == 1: return False return True def _url_has_embedded_credentials(value: str) -> bool: """True if URL embeds userinfo that looks like a real password/token.""" v = _strip_value_quotes(value) # scheme://user:pass@host m = re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*://([^/@]+@)", v) if not m: return False userinfo = m.group(1)[:-1] # drop trailing @ if ":" not in userinfo: # user only, no password — treat as non-secret unless token-like user return _has_real_secret_value(userinfo) and len(userinfo) >= 16 user, password = userinfo.split(":", 1) if not password: return False return _has_real_secret_value(password) def _is_local_connection_without_secret(key: str, value: str) -> bool: """Allow local docker-compose style URLs without real passwords.""" if not re.search(r"(URL|URI|DSN|CONNECTION)", key, re.I): return False v = _strip_value_quotes(value) if _url_has_embedded_credentials(v): return False # redis://localhost:6379 or postgres://postgres@db:5432/app host_m = re.search(r"://(?:[^/@]+@)?([^/:]+)", v) if not host_m: return False host = host_m.group(1).lower() return host in LOCAL_HOSTS or host.endswith(".local") def _parse_env_assignment(line: str) -> tuple[str, str] | None: s = line.strip() if not s or s.startswith("#"): return None if s.lower().startswith("export "): s = s[7:].strip() if "=" not in s: return None key, value = s.split("=", 1) key = key.strip() value = value.strip() if not key or not re.match(r"^[A-Za-z_][A-Za-z0-9_.-]*$", key): return None if value and value[0] not in {"'", '"'}: value = re.split(r"\s+#", value, maxsplit=1)[0].strip() return key, value def _scan_env_content_for_secrets(content: str) -> str | None: for label, pattern in SENSITIVE_VALUE_PATTERNS: if label == "Bearer-like high-entropy secret": continue if not pattern.search(content): continue for line in content.splitlines(): parsed = _parse_env_assignment(line) if not parsed: continue _, value = parsed if pattern.search(value) and _has_real_secret_value(value): return f"检测到疑似真实密钥值 ({label})" for line in content.splitlines(): parsed = _parse_env_assignment(line) if not parsed: continue key, value = parsed if not _has_real_secret_value(value): continue if _is_local_connection_without_secret(key, value): continue # URL vars: only block when credentials embedded or clearly remote secret DSN if re.search(r"(URL|URI|DSN|CONNECTION_STRING)$", key, re.I): if not _url_has_embedded_credentials(value): # remote URL without userinfo — usually not a secret by itself continue if SENSITIVE_KEY_RE.search(key): return f"检测到敏感环境变量已赋值 ({key})" for label, pattern in SENSITIVE_VALUE_PATTERNS: if not pattern.search(value): continue if label == "Bearer-like high-entropy secret": raw = _strip_value_quotes(value) if len(raw) < 40: continue classes = sum( [ bool(re.search(r"[a-z]", raw)), bool(re.search(r"[A-Z]", raw)), bool(re.search(r"[0-9]", raw)), bool(re.search(r"[+/=_-]", raw)), ] ) if classes < 3: continue return f"检测到疑似真实密钥值 ({label}, key={key})" return None def _env_secret_block_reason(token: str, cwd: str | None = None) -> str | None: if not _looks_like_env_file(token): return None path = _normalize_path(token, cwd) if _is_env_read_allowed(path): return None content = _read_text_head(path, ENV_SCAN_MAX_BYTES) if content is None or content == "": return None hit = _scan_env_content_for_secrets(content) if hit: return f"已拦截敏感读取: .env 含真实密钥 — {hit}" return None def _vault_or_exact_read_reason(token: str, cwd: str | None = None) -> str | None: raw = token.strip("\"'") path = _normalize_path(raw, cwd) base = os.path.basename(path) if path in READ_EXACT or _expand_home_refs(raw) in READ_EXACT: return "已拦截敏感读取: 命中受保护凭据文件" vault = _vault_prefix_for(path) if vault is not None: if _is_vault_read_allowed(path): return None return "已拦截敏感读取: 命中凭据目录中的敏感路径" # Bare private key basenames outside vault — content scan only if base in KEY_MATERIAL_BASENAMES: return _key_material_block_reason(token, cwd=cwd) return None def _sensitive_read_reason(token: str, cwd: str | None = None) -> str | None: """Unified read deny reason, or None if allowed.""" if not _looks_like_path_token(token) and not _looks_like_env_file(token): # Still check exact expanded home paths that are bare basenames expanded = _expand_home_refs(token.strip("\"'")) if expanded not in READ_EXACT and os.path.basename(expanded) not in KEY_MATERIAL_BASENAMES: return None env_reason = _env_secret_block_reason(token, cwd=cwd) if env_reason: return env_reason key_reason = _key_material_block_reason(token, cwd=cwd) if key_reason: return key_reason return _vault_or_exact_read_reason(token, cwd=cwd) def _sensitive_write_reason( token: str, cwd: str | None = None, *, protect_shell_rc: bool = True, ) -> str | None: path = _normalize_path(token, cwd) if _vault_prefix_for(path) is not None: return "已拦截高风险写入: 目标位于凭据目录" if protect_shell_rc and ( path in WRITE_SHELL_EXACT or _expand_home_refs(token.strip("\"'")) in WRITE_SHELL_EXACT ): return "已拦截高风险写入: 目标为 shell 启动文件（防劫持）" return None # Exact directories whose deletion is catastrophic; children stay allowed # (e.g. rm -rf /var/tmp/foo is fine, rm -rf /var is not). CATASTROPHIC_EXACT = frozenset( { "/", HOME, "/Users", "/home", "/root", "/System", "/private", "/etc", "/bin", "/sbin", "/usr", "/var", "/opt", "/Library", "/Applications", "/boot", "/lib", "/lib64", "/srv", "/mnt", "/proc", "/sys", "/dev", "/run", "/snap", "/mnt/c", "/mnt/wsl", "/mnt/wslg", } ) def _is_home_or_root_wipe(path: str) -> bool: """True for `/`, `$HOME`, OS roots, and wiping all children (`/*`, `~/*`).""" stripped = path.rstrip("/") or "/" if stripped in CATASTROPHIC_EXACT: return True if path.endswith("/*"): parent = path[:-2].rstrip("/") or "/" return parent in CATASTROPHIC_EXACT parent = os.path.dirname(path).rstrip("/") or "/" return os.path.basename(path) == "*" and parent in CATASTROPHIC_EXACT def _sensitive_delete_reason(token: str, cwd: str | None = None) -> str | None: write_reason = _sensitive_write_reason(token, cwd=cwd, protect_shell_rc=True) if write_reason: return write_reason.replace("写入", "删除").replace("（防劫持）", "（防误删启动文件）") path = _normalize_path(token, cwd) raw = _expand_home_refs(token.strip("\"'")) if _is_home_or_root_wipe(path) or _is_home_or_root_wipe(raw): return "已拦截敏感删除: 目标为系统根或用户主目录" if path in READ_EXACT or raw in READ_EXACT: return "已拦截敏感删除: 命中受保护凭据文件" return None def _unwrap_segment(segment: list[str]) -> list[str]: """Strip sudo/env/nice wrappers so `sudo rm` is judged as `rm`.""" tokens = list(segment) while tokens: name = os.path.basename(tokens[0]).lower() if name == "sudo": tokens = tokens[1:] while tokens: tok = tokens[0] if tok == "--": tokens = tokens[1:] break if tok.startswith("-"): flag = tok.split("=", 1)[0] if flag in SUDO_VALUE_FLAGS and "=" not in tok and len(tokens) > 1: tokens = tokens[2:] else: tokens = tokens[1:] continue break continue if name == "env": tokens = tokens[1:] while tokens and (tokens[0].startswith("-") or "=" in tokens[0]): tokens = tokens[1:] continue if name in COMMAND_WRAPPERS: tokens = tokens[1:] while tokens and tokens[0].startswith("-"): tokens = tokens[1:] continue break return tokens def _join_shell_tokens(tokens: list[str]) -> str: parts: list[str] = [] for token in tokens: if token in SHELL_SEPARATORS: parts.append(token) else: parts.append(shlex.quote(token)) return " ".join(parts) def _trash_operand(token: str) -> str: if token.startswith("-"): return f"./{token}" return token def _trash_argv() -> list[str] | None: """Platform trash frontend: macOS trash(1), Linux trash-cli/gio, else XDG helper.""" if sys.platform == "darwin": darwin = "/usr/bin/trash" if os.path.isfile(darwin) and os.access(darwin, os.X_OK): return [darwin] else: for name in ("trash-put", "trash"): found = shutil.which(name) if not found: continue real = os.path.basename(os.path.realpath(found)) if real in {"rm", "rm.coreutils"}: continue return [found] gio = shutil.which("gio") if gio: return [gio, "trash"] if _CODEX_TRASH.is_file(): return [sys.executable, str(_CODEX_TRASH)] return None def _rewrite_rm_segment(segment: list[str]) -> list[str] | None: """Map a simple rm/rmdir segment to trash. Leave sudo rm alone.""" if not segment: return None if os.path.basename(segment[0]).lower() == "sudo": return None unwrapped = _unwrap_segment(segment) if not unwrapped: return None name = os.path.basename(unwrapped[0]).lower() if name not in REWRITE_TO_TRASH: return None paths = [t for t in unwrapped[1:] if t != "--" and not t.startswith("-")] if not paths: return None trash = _trash_argv() if not trash: return None return [*trash, *(_trash_operand(p) for p in paths)] def _extract_cwd(payload: dict) -> str: for obj in (payload["tool_input"], payload): for key in ("workdir", "cwd", "working_directory"): value = obj.get(key) if isinstance(value, str) and value.strip(): return _normalize_path(value, payload.get("cwd")) return os.getcwd() def _approval_database(): directory = Path(HOME) / ".local/state/agent-guard" directory.mkdir(parents=True, exist_ok=True, mode=0o700) directory.chmod(0o700) path = directory / "approvals.sqlite3" fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o600) os.close(fd) path.chmod(0o600) db = sqlite3.connect(path, timeout=2, isolation_level=None) db.execute("CREATE TABLE IF NOT EXISTS requests (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, state TEXT NOT NULL, created REAL NOT NULL, approved REAL, consumed REAL)") return db def approve_request(request_id: str): """Operator-only action after explicit consent; never called by evaluate.""" if not re.fullmatch(r"[a-f0-9]{32}", request_id): raise ValueError("无效 request_id") db = _approval_database() try: db.execute("BEGIN IMMEDIATE") now = time.time() result = db.execute("UPDATE requests SET state='approved', approved=? WHERE id=? AND state='pending' AND created>?", (now, request_id, now - 900)) if result.rowcount != 1: raise ValueError("请求不存在、已批准、已消费或已过期；不能重新批准") db.commit() finally: db.close() def _read_permission(payload, cwd, reason, host): if host == "claude": print(json.dumps({"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "ask", "permissionDecisionReason": reason}}, ensure_ascii=False)) return 0 if not isinstance(payload.get("session_id"), str) or not payload["session_id"].strip(): return _deny("敏感读取需要确认；缺少 session_id，无法创建绑定本次会话的请求") fingerprint = hashlib.sha256(json.dumps([payload["session_id"], cwd, payload["tool_name"], payload["tool_input"]], sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() db = _approval_database() try: db.execute("BEGIN IMMEDIATE") now = time.time() row = db.execute("SELECT id FROM requests WHERE fingerprint=? AND state='approved' AND created>? AND approved>? ORDER BY created LIMIT 1", (fingerprint, now - 900, now - 300)).fetchone() if row: result = db.execute("UPDATE requests SET state='consumed', consumed=? WHERE id=? AND state='approved'", (now, row[0])) if result.rowcount != 1: raise ValueError("批准状态发生变化") db.commit() return 0 row = db.execute("SELECT id FROM requests WHERE fingerprint=? AND state='pending' AND created>? ORDER BY created LIMIT 1", (fingerprint, now - 900)).fetchone() request_id = row[0] if row else uuid.uuid4().hex if not row: db.execute("INSERT INTO requests(id,fingerprint,state,created) VALUES(?,?,'pending',?)", (request_id, fingerprint, now)) db.commit() finally: db.close() return _deny(f"{reason}；需要用户手动确认，request_id={request_id}。说明本次工具、命令、目录与读取目的；不得自动批准或换工具绕过。仅在用户明确同意此请求后执行 pre_tool_use_guard.py approve {request_id}，再原样重试一次。请求15分钟有效，批准后最多5分钟且不超过请求期限；一次放行即消费，执行失败也不恢复。") READ_COMMANDS = frozenset({"cat", "head", "tail", "less", "more", "bat", "tac", "nl", "wc", "sort", "uniq", "cut", "strings", "base64", "xxd", "od", "hexdump", "grep", "egrep", "fgrep", "rg", "sed", "awk", "gawk", "jq", "yq", "diff", "cmp"}) # Options that consume values are command-specific. In particular cat -n/-e, # tail -f and grep -n are switches, unlike head -n, grep -f and grep -e. READER_VALUE_FLAGS = { "cat": "", "head": "-c -n --bytes --lines", "tail": "-c -n -s --bytes --lines --pid --sleep-interval --max-unchanged-stats", "less": "-b -h -j -p -P -x -y -z -# --buffers --max-back-scroll --jump-target --pattern --prompt --tabs --max-forw-scroll --window", "more": "-n --lines", "bat": "-l -H -r --language --highlight-line --line-range --theme --style --pager --terminal-width --wrap --tabs --file-name --color --paging", "tac": "-s --separator", "nl": "-b -d -f -h -i -l -n -s -v -w --body-numbering --section-delimiter --footer-numbering --header-numbering --line-increment --join-blank-lines --number-format --number-separator --starting-line-number --number-width", "wc": "", "sort": "-k -t -S --key --field-separator --buffer-size --parallel --batch-size --compress-program", "uniq": "-f -s -w --skip-fields --skip-chars --check-chars", "cut": "-b -c -d -f --bytes --characters --delimiter --fields --output-delimiter", "strings": "-n -t -T -e -s --bytes --radix --target --encoding --output-separator", "base64": "-w -b --wrap --break", "xxd": "-c -g -l -o -s", "od": "-A -j -N -t --address-radix --skip-bytes --read-bytes --format", "hexdump": "-n -s --length --skip", "grep": "-A -B -C -m -d -D --after-context --before-context --context --max-count --directories --devices --include --exclude --exclude-dir --label", "rg": "-A -B -C -m -M -g -t -T -j -r --after-context --before-context --context --max-count --max-columns --glob --iglob --type --type-not --threads --replace --encoding --max-depth --sort --sortr --type-add --type-clear --path-separator --pre --pre-glob", "sed": "", "awk": "-F -v --field-separator --assign", "jq": "-L --indent", "yq": "--indent -I --input-format --output-format -p -o", "diff": "-I -L -F -W -U -C --ignore-matching-lines --label --show-function-line --width --unified --context --horizon-lines --starting-file --exclude --tabsize", "cmp": "-i -n --ignore-initial --bytes", } READER_SWITCHES = { "cat": "--show-all --number-nonblank --show-ends --number --squeeze-blank --show-tabs --show-nonprinting", "head": "--quiet --silent --verbose --zero-terminated", "tail": "--quiet --silent --verbose --zero-terminated --retry", "less": "--quit-if-one-screen --RAW-CONTROL-CHARS --chop-long-lines --no-init --LINE-NUMBERS --line-numbers --ignore-case --IGNORE-CASE --raw-control-chars --force --squeeze-blank-lines --QUIET --quiet", "more": "--silent --logical --print-over --clean-print --squeeze --plain --no-pause", "bat": "--number --plain --no-paging --show-all --force-colorization --unbuffered --list-languages --list-themes", "tac": "--before --regex", "nl": "--no-renumber", "wc": "--bytes --chars --lines --max-line-length --words", "sort": "--ignore-leading-blanks --dictionary-order --ignore-case --general-numeric-sort --ignore-nonprinting --month-sort --human-numeric-sort --numeric-sort --random-sort --reverse --version-sort --merge --stable --unique --zero-terminated", "uniq": "--count --repeated --ignore-case --unique --zero-terminated", "cut": "--complement --only-delimited --zero-terminated", "strings": "--all --data --print-file-name --include-all-whitespace", "base64": "--decode --ignore-garbage", "xxd": "", "od": "--output-duplicates --traditional", "hexdump": "--one-byte-octal --one-byte-char --canonical --two-bytes-decimal --two-bytes-octal --two-bytes-hex --no-squeezing", "grep": "--extended-regexp --fixed-strings --basic-regexp --perl-regexp --ignore-case --no-ignore-case --invert-match --word-regexp --line-regexp --null-data --no-messages --no-filename --with-filename --line-number --byte-offset --only-matching --quiet --silent --text --recursive --dereference-recursive --files-with-matches --files-without-match --count --null --binary --unix-byte-offsets --line-buffered", "rg": "--files --hidden --no-ignore --no-ignore-vcs --no-ignore-parent --no-ignore-dot --no-ignore-exclude --no-ignore-global --no-ignore-messages --no-messages --line-number --no-line-number --ignore-case --case-sensitive --smart-case --invert-match --word-regexp --line-regexp --fixed-strings --pcre2 --pretty --passthru --files-with-matches --files-without-match --count --count-matches --text --binary --search-zip --null --null-data --multiline --multiline-dotall --crlf --no-heading --heading --column --no-column --no-config --stats --json --type-list --debug --trace --pcre2-version --follow --only-matching --quiet --one-file-system --no-unicode --unicode --no-require-git", "sed": "--quiet --silent --regexp-extended --separate --unbuffered --null-data --posix --sandbox --follow-symlinks", "awk": "--posix --traditional --bignum --non-decimal-data", "jq": "--raw-output --raw-input --slurp --compact-output --null-input --exit-status --join-output --ascii-output --sort-keys --color-output --monochrome-output --unbuffered --stream --stream-errors --seq --tab --args --jsonargs", "yq": "--null-input --exit-status --no-colors --colors --prettyPrint --unwrapScalar", "diff": "--text --ignore-space-change --ignore-blank-lines --ignore-case --ignore-all-space --ignore-trailing-space --ignore-tab-expansion --brief --report-identical-files --recursive --new-file --unidirectional-new-file --side-by-side --expand-tabs --initial-tab --show-c-function --left-column --suppress-common-lines --strip-trailing-cr", "cmp": "--print-bytes --verbose --silent --quiet", } READER_OPTIONAL = { "tail": "--follow", "sort": "--check", "uniq": "--all-repeated --group", "grep": "--color --colour", "diff": "--color", "hexdump": "--color", "awk": "--lint", } READER_SHORT_SWITCHES = { "cat": "AbeEnstTuv", "head": "qvz", "tail": "fFqvz", "less": "FRSXNniIrfsQqmMeEgGJKwWuUV", "more": "dlfpcesur", "bat": "npPAfu", "tac": "br", "nl": "p", "wc": "cmlLw", "sort": "bdfgiMhnRrVcCmsuz", "uniq": "cdDiuz", "cut": "nsz", "strings": "adfwoVv", "base64": "dD", "xxd": "abCEehipruvd", "od": "abcdfilosxv", "hexdump": "bcCdoxvL", "grep": "EFGPiyvwxzshHnboqaIrRlLcZUyu", "rg": "hVnisSvw xFPp lLa z0UuNoq".replace(" ", ""), "sed": "nErsuz", "awk": "bcPOS", "jq": "rRscnej aSCM".replace(" ", ""), "yq": "nrjePCM", "diff": "abBiwZEqs rNPytTpcu".replace(" ", ""), "cmp": "bls", } def _reader_operands(name, args): name = {"egrep": "grep", "fgrep": "grep", "gawk": "awk"}.get(name, name) pattern_needed = name in {"grep", "rg", "sed", "awk", "jq", "yq"} expressions = {"-e", "--regexp", "--expression", "--source"} if name in {"grep", "rg", "sed", "awk", "hexdump"} else set() files = {"-f", "--file"} if name in {"grep", "rg", "sed", "awk", "jq", "hexdump"} else set() files |= {"wc": {"--files0-from"}, "sort": {"--files0-from", "--random-source"}, "grep": {"--exclude-from"}, "diff": {"--exclude-from", "--from-file", "--to-file"}, "bat": {"--config-file"}, "base64": {"-i", "--input"}}.get(name, set()) outputs = {"sort": {"-o", "--output", "-T", "--temporary-directory"}, "less": {"-o", "-O", "--log-file", "--LOG-FILE"}, "base64": {"-o", "--output"}}.get(name, set()) values = set(READER_VALUE_FLAGS[name].split()) named_values = {"--arg", "--argjson", "--slurpfile", "--rawfile", "--argfile"} if name == "jq" else set() switches = set(READER_SWITCHES[name].split()) | {"--help", "--version"} optional = set(READER_OPTIONAL.get(name, "").split()) in_place, enumerate_files, reads, writes, positional, index, options = False, False, [], [], [], 0, True executes_code = name in {"sed", "awk", "yq"} def take(flag, attached=None): nonlocal index, pattern_needed, executes_code if attached is None: if index >= len(args): raise ValueError("选项缺少值") value = args[index] index += 1 else: value = attached if flag in named_values: if index >= len(args): raise ValueError("命名参数缺少值") if flag in {"--slurpfile", "--rawfile", "--argfile"}: reads.append(args[index]) index += 1 elif flag in outputs: writes.append(value) elif flag in files: reads.append(value) if flag in {"-f", "--file"}: pattern_needed = False elif flag in expressions: pattern_needed = False if name == "rg" and flag == "--pre": executes_code = True taking = values | files | outputs | expressions | named_values while index < len(args): arg = args[index] index += 1 if options and arg == "--": options = False continue if options and arg.startswith("--"): flag, equals, value = arg.partition("=") if flag in {"--in-place", "--inplace"} and name in {"sed", "yq"}: in_place = True elif flag in taking: take(flag, value if equals else None) elif flag in switches and not equals: if name == "rg" and flag == "--files": enumerate_files = True elif flag not in optional: raise ValueError("未支持的长选项；请使用已支持的完整拼写") continue if options and arg.startswith("-") and arg != "-": if name == "xxd" and arg == "-ps": arg = "-p" for offset, letter in enumerate(arg[1:], 2): flag = "-" + letter if flag == "-i" and name in {"sed", "yq"}: in_place = True break # optional attached backup suffix; never consume next operand if flag in taking: take(flag, arg[offset:] or None) break if name == "od" and flag in {"-w", "-S"}: break # optional attached value; the next token is still a file if letter not in READER_SHORT_SWITCHES[name]: raise ValueError("未支持的短选项；不能确定其参数角色") continue positional.append(arg) if pattern_needed and positional: positional = positional[1:] if name in {"uniq", "xxd"} and len(positional) > 1: writes.extend(positional[1:]) positional = positional[:1] reads.extend(p for p in positional if p != "-") if in_place: writes.extend(positional) if enumerate_files: reads = [] # filename enumeration, not file contents return reads, writes, executes_code def _transfer_operands(name, args, cwd): """cp/mv/install/ln options; option values are never positional targets.""" values = {"-t", "--target-directory", "-S", "--suffix"} if name == "install": values |= {"-m", "--mode", "-o", "--owner", "-g", "--group", "--strip-program"} switches = { "cp": "--archive --attributes-only --copy-contents --dereference --force --interactive --link --no-clobber --no-dereference --no-target-directory --one-file-system --parents --recursive --remove-destination --strip-trailing-slashes --symbolic-link --verbose --keep-directory-symlink --debug", "mv": "--force --interactive --no-clobber --no-target-directory --strip-trailing-slashes --verbose --exchange --no-copy --debug", "install": "--compare --directory --preserve-timestamps --strip --verbose --no-target-directory", "ln": "--directory --force --interactive --logical --no-dereference --physical --relative --symbolic --verbose --no-target-directory", } switches = set(switches[name].split()) | {"--help", "--version"} short_switches = {"cp": "abdfHilLnPpRrsTuvx", "mv": "bfinTuv", "install": "bCcDdpsv", "ln": "bdfiLnPrsTv"}[name] optional = {"--backup", "--update"} if name == "cp": optional |= {"--reflink", "--sparse", "--preserve"} values |= {"--no-preserve"} positional, target, index, options, directories = [], None, 0, True, False def take(flag, attached=None): nonlocal index, target if attached is None: if index >= len(args): raise ValueError("文件操作选项缺少值") value = args[index] index += 1 else: value = attached if flag in {"-t", "--target-directory"}: target = value elif flag in {"-S", "--suffix"} and ("/" in value or "\\" in value): raise ValueError("不支持含路径分隔符的备份后缀") for_no_args = False while index < len(args): arg = args[index] index += 1 if options and arg == "--": options = False continue if options and arg.startswith("--"): flag, equals, value = arg.partition("=") if flag in values: take(flag, value if equals else None) elif flag in switches and not equals: directories |= name == "install" and flag == "--directory" for_no_args |= flag in {"--help", "--version"} elif flag not in optional: raise ValueError("未支持的文件操作长选项") continue if options and arg.startswith("-") and arg != "-": for offset, letter in enumerate(arg[1:], 2): flag = "-" + letter if flag in values: take(flag, arg[offset:] or None) break directories |= name == "install" and flag == "-d" if letter not in short_switches: raise ValueError("未支持的文件操作短选项") continue positional.append(arg) if for_no_args: return [], [] if directories: return [], positional if target is not None: return positional, [target] if len(positional) >= 2: return positional[:-1], positional[-1:] if name == "ln" and len(positional) == 1: return positional, [os.path.join(cwd, os.path.basename(positional[0]))] raise ValueError("无法确定文件操作的源和目标") def _shell_effects(command, cwd): """Parse shell syntax without execution. Unknown programs remain outside scope.""" sys.path.insert(0, str(Path(__file__).with_name("guard-libs"))) import bashlex # bashlex 0.18 leaves heredoc delimiter quotes intact (upstream TODO). # Apply quote removal to that delimiter only, never to the body or command. if not getattr(bashlex.heredoc.makeheredoc, "guard_quote_removal", False): original = bashlex.heredoc.makeheredoc def makeheredoc(tokenizer, node, lineno, killleading): node.output.word = shlex.split(node.output.word)[0] return original(tokenizer, node, lineno, killleading) makeheredoc.guard_quote_removal = True bashlex.heredoc.makeheredoc = makeheredoc hard, reads, rewrites = [], [], [] def record(path, operation, directory, tool_edit=False): checker = {"read": _sensitive_read_reason, "write": _sensitive_write_reason, "delete": _sensitive_delete_reason}[operation] reason = checker(path, cwd=directory, **({"protect_shell_rc": False} if tool_edit and operation == "write" else {})) if operation == "read" and _normalize_path(path, directory) in VAULT_PREFIXES: reason = "敏感读取：读取凭据目录内容" if reason: (reads if operation == "read" else hard).append(reason) def redirect(p, directory, shell_name=""): if p.type in {"<<", "<<-"}: delimiter = command[p.output.pos[0]:p.output.pos[1]] body = getattr(p, "heredoc", None) if body and not any(c in delimiter for c in "'\"\\") and any(c in body.value for c in "$`"): hard.append("无法可靠解析含展开的 here-document；请使用不展开的引用定界符") if body and shell_name in {"sh", "bash", "zsh"}: nested_hard, nested_reads, _ = _shell_effects(body.value.rsplit("\n", 1)[0], directory) hard.extend(nested_hard) reads.extend(nested_reads) elif hasattr(p.output, "word") and p.type != "<<<": if p.type in {"<", "<>"}: record(p.output.word, "read", directory) if p.type in {">", ">>", ">|", "&>", "&>>", "<>", ">&"}: record(p.output.word, "write", directory) for child in getattr(p.output, "parts", []): visit(child, directory) def visit(node, directory): if node.kind == "redirect": redirect(node, directory) return directory if node.kind == "command": words = [p.word for p in node.parts if p.kind == "word"] words = _unwrap_segment(words) name = os.path.basename(words[0]) if words else "" args = words[1:] for p in node.parts: if p.kind == "redirect": redirect(p, directory, name) # Substitutions are real commands even inside echo/printf. for child in getattr(p, "parts", []): visit(child, directory) paths = [a for a in args if not a.startswith("-")] if name.startswith("mkfs") or (name == "diskutil" and "eraseDisk" in args) or (name == "dd" and any(a.startswith("of=/dev/") for a in args)): hard.append("已拦截危险磁盘操作") if name in {"cp", "mv", "install", "ln"}: sources, destinations = _transfer_operands(name, args, directory) for destination in destinations: record(destination, "write", directory) for src in sources: if name == "mv": record(src, "delete", directory) reason = _sensitive_read_reason(src, directory) if reason: hard.append("已拦截敏感文件移动；读取批准不能授权移走源文件") record(src, "read", directory) elif name in WRITE_COMMANDS: for path in paths: record(path, "delete" if name in DELETE_COMMANDS else "write", directory) elif name in READ_COMMANDS: input_paths, output_paths, _ = _reader_operands(name, args) for path in output_paths: record(path, "write", directory) for path in input_paths: record(path, "read", directory) elif name in {"sh", "bash", "zsh"} and "-c" in args: index = args.index("-c") + 1 if index < len(args): nested_hard, nested_reads, _ = _shell_effects(args[index], directory) hard.extend(nested_hard) reads.extend(nested_reads) if name in REWRITE_TO_TRASH and not any(p.kind != "word" for p in node.parts) and not any(getattr(p, "parts", []) for p in node.parts) and not any(c in command[node.pos[0]:node.pos[1]] for c in "*?["): replacement = _rewrite_rm_segment([p.word for p in node.parts]) if replacement: rewrites.append((node.pos, _join_shell_tokens(replacement))) if name == "cd": if len(args) != 1 or args[0].startswith("-"): hard.append("无法确定 cd 后的工作目录") else: directory = _normalize_path(args[0], directory) return directory if node.kind == "pipeline": commands = [_unwrap_segment([w.word for w in p.parts if w.kind == "word"]) for p in node.parts if p.kind == "command"] names = [os.path.basename(words[0]) for words in commands if words] if any(n in {"curl", "wget"} for n in names) and any(n in {"sh", "bash", "zsh"} for n in names): hard.append("已拦截下载内容直接交给 shell 执行") if hasattr(node, "command"): visit(node.command, directory) for child in getattr(node, "parts", []): directory = visit(child, directory) for child in getattr(node, "list", []): visit(child, directory) for child in getattr(node, "redirects", []): visit(child, directory) return directory for tree in bashlex.parse(command + "\n"): cwd = visit(tree, cwd) rewritten = command for (start, end), replacement in sorted(rewrites, reverse=True): rewritten = rewritten[:start] + replacement + rewritten[end:] return hard, reads, rewritten def _absolute_read_call(command): """Older Codex drops workdir. Approve only cwd-independent simple reads then.""" import bashlex trees = bashlex.parse(command + "\n") if len(trees) != 1 or trees[0].kind != "command": return False parts = trees[0].parts if any(p.kind != "word" or getattr(p, "parts", []) for p in parts): return False words = [p.word for p in parts] if not words or words[0] not in READ_COMMANDS: return False input_paths, output_paths, executes_code = _reader_operands(words[0], words[1:]) return not output_paths and not executes_code and all(os.path.isabs(_expand_home_refs(p)) and not any(c in p for c in "*?[") for p in input_paths) def evaluate(payload, host="codex"): try: if not isinstance(payload, dict) or not isinstance(payload.get("tool_input"), dict) or not isinstance(payload.get("tool_name"), str): return _deny("无效 hook 输入；未执行操作") if host not in {"codex", "claude"}: return _deny("未知 hook 宿主") cwd, data = _extract_cwd(payload), payload["tool_input"] tool = payload["tool_name"].lower() hard, reads, rewritten = [], [], None if tool in {"apply_patch", "applypatch"}: patch_text = data.get("input", data.get("patch", data.get("command", ""))) if not isinstance(patch_text, str) or not patch_text.startswith("*** Begin Patch"): return _deny("无法解析补丁操作") original = None for line in patch_text.splitlines(): match = re.fullmatch(r"\*\*\* (Add File|Update File|Delete File|Move to): (.+)", line) if not match: continue op, path = match.groups() if op == "Move to": if original is None: return _deny("补丁移动缺少源路径") reason = _sensitive_delete_reason(original, cwd) if reason: hard.append(reason) checker = _sensitive_delete_reason if op == "Delete File" else _sensitive_write_reason reason = checker(path, cwd, **({"protect_shell_rc": False} if op != "Delete File" else {})) if reason: hard.append(reason) if op == "Update File": original = path elif tool in {"bash", "exec_command", "shell", "shell_command"}: command = data.get("command", data.get("cmd")) if not isinstance(command, str): return _deny("shell 输入缺少命令") hard, reads, rewritten = _shell_effects(command, cwd) else: for key in ("file_path", "path", "target_file", "filename"): path = data.get(key) if not isinstance(path, str): continue if tool in {"edit", "write", "multiedit", "create"}: reason = _sensitive_write_reason(path, cwd, protect_shell_rc=False) if reason: hard.append(reason) elif tool == "read": reason = _sensitive_read_reason(path, cwd) if reason: reads.append(reason) if hard: return _deny(hard[0]) if reads: if host == "codex" and tool in {"bash", "exec_command", "shell", "shell_command"} and not any(data.get(k) for k in ("workdir", "cwd", "working_directory")) and not _absolute_read_call(command): return _deny("宿主未提供实际工具工作目录；本调用无法建立精确单次读取授权。请将敏感读取拆成使用绝对文件路径的单个读取命令，再向用户请求确认；不得绕过此检查。") result = _read_permission(payload, cwd, reads[0], host) # A consumed approval grants this exact call; do not rewrite it. return result if rewritten is not None and rewritten != data.get("command", data.get("cmd")): return _allow_updated_command(payload, rewritten, "普通删除已改写到废纸篓") return 0 except Exception as exc: # Never expose exception data (may include command text or secrets). return _deny(f"护栏检查失败（{type(exc).__name__}），未放行；请检查解析依赖、语法或授权状态库") def main(): if len(sys.argv) == 3 and sys.argv[1] == "approve": try: approve_request(sys.argv[2]) print("已批准该请求一次；请原样重试。") return 0 except Exception as exc: print(f"批准失败（{type(exc).__name__}）；请求未放行", file=sys.stderr) return 2 host = "claude" if sys.argv[1:] == ["--host", "claude"] else "codex" if sys.argv[1:] and host == "codex": return _deny("未知护栏参数") try: payload = json.load(sys.stdin) except Exception: return _deny("hook 输入不是有效 JSON；未执行操作") return evaluate(payload, host) if __name__ == "__main__": raise SystemExit(main())