SkillsHooksPromptsAgentsPersonasModelsPoliciesToolsTemplatesBundlesCategoriesStart here
← Hooks
Hk hookgovernanceblockingstable

Branch Guard

Consolidated PreToolUse guard for Bash, Edit, Write, and NotebookEdit. Prevents protected-branch mutations, direct work in primary clones, non-canonical worktree creation, commit verification bypasses, and blocking secret-pattern use; it also requests confirmation for destructive GitHub CLI operations.

id hook/branch-guardv2.0.0by convergent-systems-key
Event
PreToolUse
Trigger
tool-nameBash|Edit|Write|NotebookEdit
Language
python
Side effects
  • blocks protected-branch mutations, primary-clone changes, non-canonical worktree creation, commit verification bypasses, and blocking secret-pattern matches
  • requests confirmation before destructive GitHub CLI operations
  • emits warn-level secret-pattern matches as a system message and diagnostics to stderr
  • writes violation audit records under $AI_ROOT/audit/violations/
Platforms
linuxmacoswindows
Notes
Standalone stdlib Python hook. Reads Git metadata directly and needs no sibling scripts or external commands; installed as ~/.ai/hooks/branch-guard.py.

Script · branch-guard.py

#!/usr/bin/env python3
"""hooks/guard-dispatch.py — consolidated Claude Code PreToolUse guard.

One interpreter spawn per tool call runs every applicable check. This is
the only enforcement surface for these checks — the command-wrapper
system (~/.ai/bin/git, command-wrappers.toml) that some of this logic
was originally written for was never installed and has been removed;
worktree-guard.py, destructive-gh-guard.py, and the other wrapper-only
scripts were deleted along with it. The Branch Guard policy helpers are embedded here so this file can be
installed as a standalone hook.

Checks, by tool:

  Bash
    secret-guard    deny commands containing blocking secret-shaped strings
                    (patterns.json block_level; warn-level → systemMessage)
    repo-guard      deny mutating work inside a PRIMARY repo clone — work
                    happens only in linked worktrees (Principal directive
                    2026-08-16; Constitution §3.2.10 territory)
    branch-guard    deny git mutation of protected branches (Common.md §2.2)
    worktree-guard  deny `git worktree add` outside
                    $AI_ROOT/worktrees/<org>/<repo>/<name>
    no-verify       deny `git commit --no-verify` (SPEC.md §10.3)
    gh-guard        escalate destructive gh operations to a permission ask

  Edit | Write | NotebookEdit
    repo-guard      deny file mutation inside a primary clone
    secret-guard    deny content containing blocking secret-shaped strings

Design constraints:
  - stdlib only; NO subprocess calls. Repo topology (primary clone vs
    linked worktree, current branch, origin URL) is probed by reading
    .git/HEAD and .git/config directly, so the hook adds ~30 ms of
    interpreter start-up and nothing else per tool call.
  - Fail open on internal errors: a broken guard must degrade to the
    permission system, never brick the session (top-level try/except,
    always exit 0). Denials are the ONLY intentional output.
  - Repo-guard policy: ~/.ai/governance/policy/repo-guard.json
      {"exempt_paths": []}                 # primary clones where direct
                                           # work stays allowed; empty by
                                           # default — no repo, including
                                           # ~/.ai itself, is exempt
    Session bypass (logged): AI_REPO_GUARD_BYPASS=1.

Output schema (PreToolUse): permissionDecision deny/ask JSON on stdout,
exit 0. Silent exit 0 on allow.

Self-check:
  --self-check
"""
from __future__ import annotations

import fnmatch
import json
import os
import re
import shlex
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, NamedTuple, Optional

HOOKS_DIR = Path(__file__).resolve().parent


def log(*parts: object) -> None:
    """Print structured diagnostics to stderr without polluting hook output."""
    print("[ai/branch-guard]", *parts, file=sys.stderr, flush=True)


def patterns_path() -> Path:
    root = Path(os.environ.get("AI_ROOT", str(Path.home() / ".ai")))
    candidates = [HOOKS_DIR / "patterns.json", root / "hooks" / "patterns.json"]
    return next((path for path in candidates if path.is_file()), candidates[0])


def load_patterns() -> list[dict]:
    """Load and compile the canonical and optional local secret patterns."""
    entries: dict[str, dict] = {}
    for path in (patterns_path(), patterns_path().with_name("patterns.local.json")):
        if not path.is_file():
            continue
        for entry in json.loads(path.read_text(encoding="utf-8")).get("patterns", []):
            entries[entry["id"]] = entry
    patterns = list(entries.values())
    for entry in patterns:
        entry["_compiled"] = re.compile(entry["regex"])
    return patterns


def redact_snippet(line: str, start: int, end: int, redaction: str,
                   context: int = 20) -> str:
    return (line[max(0, start - context):start] + redaction
            + line[end:min(len(line), end + context)])


def scan_lines(lines: Iterable[str], patterns: list[dict]) -> list[dict]:
    hits = []
    for line_number, line in enumerate(lines, start=1):
        for entry in patterns:
            for match in entry["_compiled"].finditer(line):
                hits.append({
                    "pattern_id": entry["id"],
                    "severity": entry.get("severity", "medium"),
                    "line": line_number,
                    "col": match.start() + 1,
                    "snippet": redact_snippet(
                        line, match.start(), match.end(),
                        entry.get("redaction", "[REDACTED]")),
                })
    return hits


# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

FILE_TOOLS = {"Edit", "Write", "NotebookEdit"}

# git subcommands that mutate the working tree, index, or HEAD of the
# checkout they run in. These are what "doing work here" means.
MUTATING_GIT = {
    "add", "am", "apply", "checkout", "switch", "restore", "reset", "clean",
    "commit", "merge", "rebase", "cherry-pick", "revert", "mv", "rm",
    "stash", "pull",
}

# git subcommands that mutate a protected branch when HEAD is on it.
BRANCH_GUARDED = {"commit", "merge", "rebase", "cherry-pick", "revert", "am", "pull"}

# git global options that consume a value.
GIT_VALUE_OPTS = {"-C", "-c", "--git-dir", "--work-tree", "--namespace", "--exec-path"}

# `git worktree add` flags that consume a value.
WORKTREE_VALUE_FLAGS = {"-b", "-B", "--reason", "--orphan"}

# Shell programs whose non-flag arguments are file-mutation targets.
# Value = which positional args to check: "all", "dest" (last only),
# "skip1" (all but the first non-flag arg), "existing" (only args that
# exist as files — for sed, whose first positional is the script).
SHELL_MUTATORS = {
    "rm": "all", "mv": "all", "touch": "all", "mkdir": "all", "tee": "all",
    "truncate": "all", "cp": "dest", "rsync": "dest", "ln": "dest",
    "chmod": "skip1", "chown": "skip1", "sed": "existing",
}

# Command prefixes that wrap another command.
WRAPPER_PROGS = {"sudo", "command", "env", "nohup", "nice", "time", "builtin"}

GH_GUARDED = {("repo", "delete"), ("release", "delete"),
              ("secret", "delete"), ("auth", "logout")}

REPO_BYPASS_ENV = "AI_REPO_GUARD_BYPASS"
GH_BYPASS_ENV = "AI_ALLOW_DESTRUCTIVE_GH"

_ENV_ASSIGN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")
_SEG_BOUNDARY = re.compile(r"^[;|&()]+$")
_REDIR_OP = re.compile(r"^(&?>{1,2}|<{1,3}|[<>]&)$")


def ai_root() -> Path:
    return Path(os.environ.get("AI_ROOT", str(Path.home() / ".ai")))


def worktrees_root() -> Path:
    return ai_root() / "worktrees"


# ---------------------------------------------------------------------------
# Repo topology probe — pure Python, no subprocess
# ---------------------------------------------------------------------------

class RepoProbe(NamedTuple):
    kind: Optional[str]        # "primary" | "worktree" | None
    top: Optional[Path]        # working-tree root
    head_dir: Optional[Path]   # directory containing this checkout's HEAD
    common_dir: Optional[Path] # shared .git directory


def probe(start: Path) -> RepoProbe:
    """Walk up from `start` (a directory) to classify the enclosing checkout.

    A `.git` DIRECTORY marks the primary clone. A `.git` FILE marks a
    linked worktree (gitdir under .git/worktrees/) or a submodule checkout
    (gitdir under .git/modules/ — treated as primary: it lives inside the
    superproject's primary clone, so the same no-work rule applies)."""
    try:
        start = start.resolve()
    except OSError:
        return RepoProbe(None, None, None, None)
    for d in (start, *start.parents):
        g = d / ".git"
        if g.is_dir():
            return RepoProbe("primary", d, g, g)
        if g.is_file():
            try:
                text = g.read_text(encoding="utf-8", errors="replace").strip()
            except OSError:
                return RepoProbe(None, None, None, None)
            if not text.startswith("gitdir:"):
                return RepoProbe(None, None, None, None)
            gd = Path(text[len("gitdir:"):].strip())
            if not gd.is_absolute():
                gd = (d / gd).resolve()
            parts = gd.parts
            if "worktrees" in parts:
                i = len(parts) - 1 - parts[::-1].index("worktrees")
                return RepoProbe("worktree", d, gd, Path(*parts[:i]))
            if "modules" in parts:
                return RepoProbe("primary", d, gd, gd)
            return RepoProbe("worktree", d, gd, gd)
    return RepoProbe(None, None, None, None)


def current_branch(pr: RepoProbe) -> str:
    """Branch name from HEAD, '' when detached/unreadable."""
    if pr.head_dir is None:
        return ""
    try:
        head = (pr.head_dir / "HEAD").read_text(encoding="utf-8").strip()
    except OSError:
        return ""
    if head.startswith("ref: refs/heads/"):
        return head[len("ref: refs/heads/"):]
    return ""


def origin_owner_repo(pr: RepoProbe) -> str:
    """'owner/repo' parsed from [remote "origin"] in the repo config, or ''."""
    if pr.common_dir is None:
        return ""
    try:
        lines = (pr.common_dir / "config").read_text(
            encoding="utf-8", errors="replace").splitlines()
    except OSError:
        return ""
    in_origin = False
    for line in lines:
        s = line.strip()
        if s.startswith("["):
            in_origin = s.replace("'", '"') in ('[remote "origin"]',)
        elif in_origin and s.startswith("url"):
            _, _, url = s.partition("=")
            return canonical_owner_repo(url.strip())
    return ""


# ---------------------------------------------------------------------------
# Policy
# ---------------------------------------------------------------------------

def branch_guard_policy() -> dict:
    """Load the protected-branch policy, using its safe defaults if absent."""
    p = ai_root() / "governance" / "policy" / "branch-guard.json"
    if p.is_file():
        try:
            return json.loads(p.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError) as e:
            log("branch-guard policy parse error (using defaults):", e)
    return {
        "names": ["main", "master"],
        "patterns": ["release/*"],
        "exempt_remotes": [],
    }


def canonical_owner_repo(url: str) -> str:
    """Return owner/repo from a supported HTTPS or SSH remote URL."""
    url = url.strip()
    ssh_match = re.match(r"^[^@]+@[^:]+:(.+?)(?:\.git)?$", url)
    if ssh_match:
        return ssh_match.group(1).strip("/")
    from urllib.parse import urlparse
    parsed = urlparse(url)
    if parsed.scheme in ("https", "http", "git") and parsed.netloc:
        return parsed.path.strip("/").removesuffix(".git")
    return ""


def is_protected(branch: str, policy: dict) -> bool:
    if not branch:
        return False
    if branch in (policy.get("names") or []):
        return True
    return any(fnmatch.fnmatch(branch, pattern)
               for pattern in policy.get("patterns") or [])


def repo_guard_policy() -> dict:
    p = ai_root() / "governance" / "policy" / "repo-guard.json"
    policy = {"exempt_paths": []}
    if p.is_file():
        try:
            policy.update(json.loads(p.read_text(encoding="utf-8")))
        except (json.JSONDecodeError, OSError) as e:
            log("repo-guard policy parse error (using defaults):", e)
    return policy


def exempt_roots(policy: dict) -> list[Path]:
    out = []
    for raw in policy.get("exempt_paths") or []:
        try:
            out.append(Path(raw).expanduser().resolve())
        except OSError:
            continue
    return out


def is_exempt(top: Optional[Path], exempts: list[Path]) -> bool:
    if top is None:
        return False
    return any(top == e or top.is_relative_to(e) for e in exempts)


# ---------------------------------------------------------------------------
# Decisions
# ---------------------------------------------------------------------------

class Decision(NamedTuple):
    action: str   # "deny" | "ask"
    reason: str


def emit(decision: Optional[Decision], warnings: list[str]) -> None:
    if decision is not None:
        print(json.dumps({"hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": decision.action,
            "permissionDecisionReason": decision.reason,
        }}))
    elif warnings:
        print(json.dumps({"systemMessage": " | ".join(warnings)}))


def write_violation(guard: str, what: str, remediation: str) -> None:
    """Best-effort violation record per Common.md §5.3; never raises."""
    try:
        vdir = ai_root() / "audit" / "violations"
        vdir.mkdir(parents=True, exist_ok=True)
        ts = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H%M%S-%fZ")
        (vdir / f"{ts}-{guard}.md").write_text(
            f"# Violation — {ts}\n\n"
            f"- **File / Rule violated:** {guard}\n"
            f"- **What happened:** {what}\n"
            f"- **How noticed:** hook-detected (guard-dispatch.py PreToolUse)\n"
            f"- **Remediation:** {remediation}\n",
            encoding="utf-8")
    except OSError as e:
        log("could not write violation record:", e)


# ---------------------------------------------------------------------------
# Secret scanning
# ---------------------------------------------------------------------------

def _pattern_blocks(entry: dict) -> bool:
    """blocking unless explicitly warn-level; absent block_level defaults
    by severity (high → blocking) so patterns.json stays backward-compatible."""
    level = entry.get("block_level")
    if level is not None:
        return level == "blocking"
    return entry.get("severity") == "high"


def scan_secrets(text: str, patterns: list[dict],
                 where: str) -> tuple[Optional[Decision], list[str]]:
    hits = scan_lines(text.splitlines() or [text], patterns)
    if not hits:
        return None, []
    by_id = {e["id"]: e for e in patterns}
    blocking = [h for h in hits if _pattern_blocks(by_id.get(h["pattern_id"], {}))]
    warnings = []
    if blocking:
        h = blocking[0]
        extra = f" (and {len(blocking) - 1} more)" if len(blocking) > 1 else ""
        return Decision("deny",
            f"Possible secret in {where}: pattern={h['pattern_id']} "
            f"severity={h['severity']}{extra}.\n"
            f"Snippet (redacted): {h['snippet']}\n"
            "Per Constitution §3.5 (no secrets in artifacts; non-overridable). "
            "Keep secrets in env vars/keychain; transfer via clipboard."), []
    ids = sorted({h["pattern_id"] for h in hits})
    warnings.append(f"[ai/guard] warn-level secret pattern match in {where}: "
                    f"{', '.join(ids)} — verify this is not a real credential.")
    return None, warnings


# ---------------------------------------------------------------------------
# Bash command parsing
# ---------------------------------------------------------------------------

def tokenize(command: str) -> Optional[list[str]]:
    lex = shlex.shlex(command, posix=True, punctuation_chars=True)
    lex.whitespace_split = True
    try:
        return list(lex)
    except ValueError:
        return None  # unparseable (heredoc/unbalanced quotes) — fail open


def split_segments(tokens: list[str]) -> list[list[str]]:
    segs: list[list[str]] = []
    cur: list[str] = []
    for t in tokens:
        if _SEG_BOUNDARY.match(t) and ">" not in t and "<" not in t:
            if cur:
                segs.append(cur)
            cur = []
        else:
            cur.append(t)
    if cur:
        segs.append(cur)
    return segs


def strip_prefixes(seg: list[str]) -> list[str]:
    """Drop leading env assignments and wrapper programs (sudo/env/…)."""
    i = 0
    while i < len(seg):
        t = seg[i]
        if _ENV_ASSIGN.match(t):
            i += 1
        elif os.path.basename(t) in WRAPPER_PROGS:
            i += 1
        else:
            break
    return seg[i:]


class Segment(NamedTuple):
    argv: list[str]        # command tokens, redirections removed
    redirects: list[str]   # write-redirection targets


def extract_redirects(seg: list[str]) -> Segment:
    argv: list[str] = []
    redirects: list[str] = []
    i = 0
    while i < len(seg):
        t = seg[i]
        if _REDIR_OP.match(t):
            is_write = ">" in t
            target = seg[i + 1] if i + 1 < len(seg) else None
            # a pure-digit token just before a redirect is an fd, not an arg
            if argv and argv[-1].isdigit():
                argv.pop()
            if (is_write and target and not target.isdigit()
                    and not target.startswith("&")
                    and not target.startswith("/dev/")):
                redirects.append(target)
            i += 2 if target is not None else 1
            continue
        argv.append(t)
        i += 1
    return Segment(argv, redirects)


def resolve_path(raw: str, cwd: Path) -> Path:
    p = Path(raw).expanduser()
    if not p.is_absolute():
        p = cwd / p
    try:
        return p.resolve()
    except OSError:
        return Path(os.path.normpath(str(p)))


def nearest_existing(p: Path) -> Path:
    while not p.exists() and p != p.parent:
        p = p.parent
    return p


def path_in_primary(raw: str, cwd: Path, exempts: list[Path]) -> Optional[RepoProbe]:
    """Probe of the primary clone a path lands in, else None."""
    anchor = nearest_existing(resolve_path(raw, cwd))
    if not anchor.is_dir():
        anchor = anchor.parent
    pr = probe(anchor)
    if pr.kind == "primary" and not is_exempt(pr.top, exempts):
        return pr
    return None


# ---------------------------------------------------------------------------
# git invocation analysis
# ---------------------------------------------------------------------------

class GitCall(NamedTuple):
    subcmd: str
    args: list[str]
    chdir: Optional[str]   # last -C value, if any


def parse_git(argv: list[str]) -> Optional[GitCall]:
    """argv[0] is 'git'. Returns None when no subcommand is found."""
    chdir: Optional[str] = None
    i = 1
    while i < len(argv):
        t = argv[i]
        if not t.startswith("-"):
            return GitCall(t, argv[i + 1:], chdir)
        if t == "-C" and i + 1 < len(argv):
            nxt = argv[i + 1]
            chdir = nxt if chdir is None else os.path.join(chdir, nxt)
            i += 2
        elif t in GIT_VALUE_OPTS and "=" not in t and i + 1 < len(argv):
            i += 2
        else:
            i += 1
    return None


def worktree_add_target(args: list[str]) -> Optional[str]:
    """Target path of `git worktree add`, given args after 'worktree'."""
    if not args or args[0] != "add":
        return None
    i = 1
    while i < len(args):
        t = args[i]
        if t == "--":
            return args[i + 1] if i + 1 < len(args) else None
        if not t.startswith("-"):
            return t
        if t in WORKTREE_VALUE_FLAGS and i + 1 < len(args):
            i += 2
        else:
            i += 1
    return None


def push_targets(args: list[str], branch: str) -> list[str]:
    pos = [a for a in args if not a.startswith("-")]
    refspecs = pos[1:] if len(pos) >= 2 else [branch]
    return [r.split(":", 1)[1] if ":" in r else r for r in refspecs if r]


def suggest_worktree(pr: RepoProbe) -> str:
    owner_repo = origin_owner_repo(pr) or "<org>/<repo>"
    return (f"git worktree add {worktrees_root()}/{owner_repo}/<branch-slug> "
            f"-b <branch>")


# ---------------------------------------------------------------------------
# Per-tool checks
# ---------------------------------------------------------------------------

def check_file_tool(tool_input: dict, cwd: Path, patterns: list[dict],
                    exempts: list[Path]) -> tuple[Optional[Decision], list[str]]:
    raw = tool_input.get("file_path") or tool_input.get("notebook_path") or ""
    if raw:
        pr = path_in_primary(raw, cwd, exempts)
        if pr is not None and os.environ.get(REPO_BYPASS_ENV) != "1":
            write_violation("repo-guard",
                            f"File mutation attempted in primary clone {pr.top}: {raw}",
                            "Denied; directed to a linked worktree.")
            return Decision("deny",
                f"{raw} is inside the PRIMARY clone at {pr.top}. The primary "
                "clone is read-only; all work happens in a linked worktree.\n"
                f"Create/use one: {suggest_worktree(pr)}\n"
                f"(Session bypass, logged: {REPO_BYPASS_ENV}=1)"), []
    content = (tool_input.get("content") or tool_input.get("new_string")
               or tool_input.get("new_source") or "")
    if content:
        return scan_secrets(str(content), patterns, "file content")
    return None, []


def check_git_segment(call: GitCall, cwd: Path,
                      exempts: list[Path]) -> Optional[Decision]:
    gdir = resolve_path(call.chdir, cwd) if call.chdir else cwd
    pr = probe(nearest_existing(gdir))
    branch = current_branch(pr)
    sub, args = call.subcmd, call.args

    # worktree-guard: canonical placement for new worktrees
    if sub == "worktree":
        target = worktree_add_target(args)
        if target is not None and not is_exempt(pr.top, exempts):
            resolved = resolve_path(target, cwd)
            ok = False
            try:
                rel = resolved.relative_to(worktrees_root().resolve())
                parts = rel.parts
                ok = len(parts) == 3
                owner_repo = origin_owner_repo(pr)
                if ok and owner_repo:
                    owner, repo = owner_repo.split("/", 1)
                    ok = (parts[0].lower() == owner.lower()
                          and parts[1].lower() == repo.lower())
            except ValueError:
                ok = False
            if not ok:
                return Decision("deny",
                    f"Worktree target {resolved} is not canonical. Worktrees "
                    f"live at {worktrees_root()}/<org>/<repo>/<name>.\n"
                    f"Use: {suggest_worktree(pr)}")
        return None

    # branch-guard: protected-branch mutation (applies in ANY checkout)
    bpolicy = branch_guard_policy()
    exempt_repo = False
    owner_repo = origin_owner_repo(pr)
    if owner_repo:
        for ex in bpolicy.get("exempt_remotes") or []:
            ex_norm = ex.strip("/").removesuffix(".git")
            if "/" in ex_norm and owner_repo.lower() == ex_norm.lower():
                exempt_repo = True
                break
    if not exempt_repo:
        ff_only = "--ff-only" in args
        if (sub in BRANCH_GUARDED and not (sub == "pull" and ff_only)
                and is_protected(branch, bpolicy)):
            write_violation("branch-guard",
                            f"`git {sub}` attempted on protected branch '{branch}'.",
                            "Denied. Branch off in a worktree and open a PR.")
            return Decision("deny",
                f"`git {sub}` would mutate protected branch '{branch}' "
                "(Constitution §3.2.10). Work on a feature branch in a "
                "worktree and open a PR.")
        if sub == "push":
            for t in push_targets(args, branch):
                if is_protected(t, bpolicy):
                    write_violation("branch-guard",
                                    f"`git push` targeting protected branch '{t}'.",
                                    "Denied. Push a feature branch and open a PR.")
                    return Decision("deny",
                        f"`git push` targets protected branch '{t}' "
                        "(Constitution §3.2.10). Push a feature branch and "
                        "open a PR instead.")

    # no-verify guard
    if sub == "commit" and ("--no-verify" in args or "-n" in args):
        return Decision("deny",
            "`git commit --no-verify`/`-n` bypasses the secret pre-commit "
            "scan (SPEC.md §10.3). Re-run without it.")

    # repo-guard: no work in a primary clone
    if (pr.kind == "primary" and not is_exempt(pr.top, exempts)
            and sub in MUTATING_GIT
            and os.environ.get(REPO_BYPASS_ENV) != "1"):
        if sub == "pull" and "--ff-only" in args:
            return None  # keeping the parked clone fresh is maintenance
        if sub in ("checkout", "switch"):
            pos = [a for a in args if not a.startswith("-")]
            if len(pos) == 1 and not any(a.startswith("-") for a in args) \
                    and is_protected(pos[0], bpolicy):
                return None  # parking the clone back on its default branch
        write_violation("repo-guard",
                        f"`git {sub}` attempted in primary clone {pr.top} "
                        f"(branch '{branch or 'detached'}').",
                        "Denied; directed to a linked worktree.")
        return Decision("deny",
            f"`git {sub}` targets the PRIMARY clone at {pr.top}. The primary "
            "clone stays parked (read-only) on its default branch; all work "
            "happens in a linked worktree.\n"
            f"Create/use one: {suggest_worktree(pr)}\n"
            f"Allowed here: status/log/diff/fetch, `git pull --ff-only`, "
            f"worktree/branch management. (Session bypass, logged: "
            f"{REPO_BYPASS_ENV}=1)")
    return None


def check_bash(command: str, cwd: Path, patterns: list[dict],
               exempts: list[Path]) -> tuple[Optional[Decision], list[str]]:
    decision, warnings = scan_secrets(command, patterns, "Bash command")
    if decision is not None:
        return decision, warnings

    tokens = tokenize(command)
    if tokens is None:
        return None, warnings  # fail open on unparseable input

    eff_cwd = cwd
    for raw_seg in split_segments(tokens):
        seg = strip_prefixes(raw_seg)
        if not seg:
            continue
        argv, redirects = extract_redirects(seg)

        # write-redirections into a primary clone are file mutations
        for target in redirects:
            pr = path_in_primary(target, eff_cwd, exempts)
            if pr is not None and os.environ.get(REPO_BYPASS_ENV) != "1":
                return Decision("deny",
                    f"Redirection writes {target} inside the PRIMARY clone at "
                    f"{pr.top}. Work in a linked worktree instead.\n"
                    f"Create/use one: {suggest_worktree(pr)}"), warnings

        if not argv:
            continue
        prog = os.path.basename(argv[0])

        if prog == "cd":
            if len(argv) > 1:
                eff_cwd = resolve_path(argv[1], eff_cwd)
            else:
                eff_cwd = Path.home()
            continue

        if prog == "git":
            call = parse_git(argv)
            if call is not None:
                d = check_git_segment(call, eff_cwd, exempts)
                if d is not None:
                    return d, warnings
            continue

        if prog == "gh":
            rest = [a for a in argv[1:] if not a.startswith("-")]
            if len(rest) >= 2 and (rest[0], rest[1]) in GH_GUARDED:
                if os.environ.get(GH_BYPASS_ENV) == "1":
                    log(f"`gh {rest[0]} {rest[1]}` bypass active "
                        f"({GH_BYPASS_ENV}=1). Logged.")
                else:
                    return Decision("ask",
                        f"`gh {rest[0]} {rest[1]}` is a §2.2 destructive "
                        "operation (irreversible). Confirm explicitly."), warnings
            continue

        mode = SHELL_MUTATORS.get(prog)
        if mode is None:
            continue
        pos = [a for a in argv[1:] if not a.startswith("-") and a != ""]
        if mode == "dest":
            pos = pos[-1:]
        elif mode == "skip1":
            pos = pos[1:]
        for raw in pos:
            resolved = resolve_path(raw, eff_cwd)
            if mode == "existing" and not resolved.exists():
                continue
            pr = path_in_primary(raw, eff_cwd, exempts)
            if pr is not None and os.environ.get(REPO_BYPASS_ENV) != "1":
                write_violation("repo-guard",
                                f"`{prog}` targeting {resolved} in primary clone {pr.top}.",
                                "Denied; directed to a linked worktree.")
                return Decision("deny",
                    f"`{prog}` targets {resolved} inside the PRIMARY clone at "
                    f"{pr.top}. The primary clone is read-only; work in a "
                    f"linked worktree.\nCreate/use one: {suggest_worktree(pr)}\n"
                    f"(Session bypass, logged: {REPO_BYPASS_ENV}=1)"), warnings
    return None, warnings


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def run() -> None:
    raw = sys.stdin.read()
    if not raw.strip():
        return
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError:
        return
    if not isinstance(payload, dict):
        return

    hook_event = payload.get("hook_event_name") or payload.get("hookEventName") or ""
    if hook_event and hook_event != "PreToolUse":
        return
    tool = payload.get("tool_name") or payload.get("toolName") or ""
    tool_input = payload.get("tool_input") or payload.get("toolInput") or {}
    if not isinstance(tool_input, dict):
        return
    cwd = Path(payload.get("cwd") or os.getcwd())

    patterns = load_patterns()
    policy = repo_guard_policy()
    exempts = exempt_roots(policy)

    if tool in FILE_TOOLS:
        decision, warnings = check_file_tool(tool_input, cwd, patterns, exempts)
    elif tool in ("Bash", "shell", "execute"):
        command = tool_input.get("command") or ""
        if not isinstance(command, str) or not command:
            return
        decision, warnings = check_bash(command, cwd, patterns, exempts)
    else:
        return

    emit(decision, warnings)


def self_check() -> int:
    try:
        patterns = load_patterns()
        assert patterns, "no patterns loaded"
        branch_guard_policy()
        repo_guard_policy()
        probe(Path.home())
    except Exception as e:  # noqa: BLE001 — self-check reports any failure
        log("self-check FAIL:", e)
        return 1
    log("self-check OK")
    return 0


if __name__ == "__main__":
    if "--self-check" in sys.argv[1:]:
        sys.exit(self_check())
    try:
        run()
    except Exception as e:  # noqa: BLE001 — guard must fail open, never brick
        log("guard-dispatch internal error (failing open):", e)
    sys.exit(0)
gitgovernancebranch-protectionclaude-code
Author convergent-systems-key. Catalog data license CC-BY-4.0.