diff --git a/README.md b/README.md index 1fc2626..7fa2c81 100644 --- a/README.md +++ b/README.md @@ -15,13 +15,29 @@ notices until later. ## Behaviour -Blocks a push when the resolved target is a protected branch. The target is read from: - -- an explicit refspec — `git push origin main`, `git push origin feature:main` -- the upstream of the current branch, when no refspec is given - -It also blocks `git push --delete origin `, on the grounds that deleting an -integration branch is worse than pushing to one. +Blocks a push when **any** destination it resolves is a protected branch. The command +line is parsed as argv (quotes honoured), and every refspec of every `git push` in it is +checked: + +- explicit refspecs — `git push origin main`, `git push origin feature:main`, + `git push origin feat/a main` (the second refspec is caught), `+HEAD:main`, + `refs/heads/x:refs/heads/main` +- a bare `HEAD` refspec, which means the current branch +- the current branch, when no refspec is given (`git push`, `git push origin`). The + current branch name stands in for the upstream; `push.default` is not consulted +- chained commands — `git push origin feat && git push origin main`, `a; b` +- `git -C push ...` and other git global options before `push` +- option values are skipped, not mistaken for refspecs (`-o `, + `--receive-pack `, `--push-option=`) + +It also blocks deleting a protected branch — `git push --delete origin ` +(the flag applies to every refspec) and the empty-source refspec form — on the grounds +that deleting an integration branch is worse than pushing to one. + +**Not interpreted** (these are allowed): `--all` and `--mirror`, wildcard refspecs +(`refs/heads/*:refs/heads/*`), tag refspecs, backtick substitution, shell aliases and +`git` wrappers, and pushes made without going through the Bash tool. Unbalanced quotes +fall back to a plain whitespace split. It allows everything else: feature branches, ad-hoc branches, and any command it cannot confidently parse. **The hook fails open by design** — a parser bug should never stand @@ -70,6 +86,32 @@ Then register it as a `PreToolUse` hook matching the `Bash` tool. The payload it Exit codes: `0` allow, `2` block (the message is written to stderr). +## Self-check + +`--self-check` prints a JSON receipt and exits, without reading stdin, running `git`, or +contacting any remote: + +```bash +python3 block-direct-integration-push.py --self-check +``` + +The receipt carries the script path and SHA-256, the effective protected branches and +where they came from (`default` or `env:GIT_PROTECTED_BRANCHES`), whether the bypass +variable is set, the advisory/fail-open contract, and a `registration` block. The hook +is looked for in Claude Code `settings.json` / `settings.local.json` files in the current +directory, its parents, and `~/.claude`. `registration.status` is one of: + +| status | meaning | exit | +|---|---|---| +| `active` | registered as a `PreToolUse` hook matching `Bash`, and the script exists | 0 | +| `registered-script-missing` | registered for `Bash` but the script path does not exist | 1 | +| `registered-not-for-bash` | registered, but its matcher does not cover `Bash` | 1 | +| `not-registered` | settings files found, none registers this script | 1 | +| `no-settings` | no readable settings file (other wrappers are not inspected) | 1 | + +Registration is detected by the script's file name, so a renamed copy reads as +`not-registered`. `remote_mutation` is always `false`. + ## Tests ```bash diff --git a/block-direct-integration-push.py b/block-direct-integration-push.py index a61adc2..62d0de2 100644 --- a/block-direct-integration-push.py +++ b/block-direct-integration-push.py @@ -12,16 +12,24 @@ a follow-up commit lands directly on it. Nothing is destroyed; review is simply skipped, silently. -Configure the branch list with PROTECTED_BRANCHES below, or the +Configure the branch list with DEFAULT_PROTECTED_BRANCHES below, or the GIT_PROTECTED_BRANCHES environment variable (comma-separated). + +Run with `--self-check` (no stdin, no git, no network, no push) to print a +JSON receipt of the effective protected branches and whether this script is +registered as a PreToolUse hook for the Bash tool. Exit 0 = active, +1 = not active. Hook mode itself is advisory and fails open. """ from __future__ import annotations +import hashlib import json import os import re +import shlex import subprocess import sys +from pathlib import Path DEFAULT_PROTECTED_BRANCHES = {"main", "master", "develop", "trunk"} BYPASS_ENV = "ALLOW_INTEGRATION_PUSH" @@ -49,34 +57,291 @@ def get_current_branch(cwd: str) -> str | None: return None +# `git push` options that consume the NEXT argv entry as their value. +# (Values attached with `=` are part of the option token and need no entry.) +_PUSH_VALUE_OPTS = {"--repo", "--receive-pack", "--exec", "--push-option"} +# `git` global options that consume the next argv entry (before the subcommand). +_GIT_VALUE_OPTS = {"-C", "-c", "--git-dir", "--work-tree", "--namespace"} +_SHELL_PUNCT = ";&|()" +_REDIRECT = re.compile(r"^\d*(>>?|<)&?$") + + +def _tokenize(cmd: str) -> list[str]: + """Split a shell command line into argv-ish tokens. + + Quotes are honoured; `;`, `&&`, `||`, `|`, `&` and parentheses become + their own tokens. On an unbalanced quote, fall back to a whitespace split + so that a malformed line is still checked rather than waved through. + """ + try: + lex = shlex.shlex(cmd, posix=True, punctuation_chars=_SHELL_PUNCT) + lex.whitespace_split = True + lex.commenters = "" + return list(lex) + except ValueError: + return cmd.split() + + +def _push_segments(tokens: list[str]) -> list[list[str]]: + """Return the argument list after `push` for every `git push` in tokens.""" + segments: list[list[str]] = [[]] + for tok in tokens: + if tok and all(c in _SHELL_PUNCT for c in tok): + segments.append([]) + else: + segments[-1].append(tok) + out: list[list[str]] = [] + for seg in segments: + git_idx = next( + (i for i, t in enumerate(seg) if t == "git" or t.endswith("/git")), None + ) + if git_idx is None: + continue + i = git_idx + 1 + while i < len(seg): + tok = seg[i] + if tok == "push": + out.append(seg[i + 1:]) + break + if not tok.startswith("-"): + break # some other git subcommand + i += 2 if tok in _GIT_VALUE_OPTS else 1 + return out + + +def _parse_push_args(args: list[str]) -> tuple[list[str], bool]: + """Split push args into (positionals, delete_flag).""" + positional: list[str] = [] + is_delete = False + i = 0 + while i < len(args): + tok = args[i] + i += 1 + if tok == "--": + positional.extend(args[i:]) + break + if _REDIRECT.match(tok): + i += 1 # bare redirection operator: its target is not a refspec + continue + if re.match(r"^\d*[<>]", tok): + continue # redirection with an attached target, e.g. 2>/dev/null + if tok.startswith("--"): + if tok == "--delete": + is_delete = True + elif tok in _PUSH_VALUE_OPTS: + i += 1 + elif tok.startswith("-") and len(tok) > 1: + for pos, ch in enumerate(tok[1:], start=1): + if ch == "d": + is_delete = True + elif ch == "o": # -o , or -o attached + if pos == len(tok) - 1: + i += 1 + break + else: + positional.append(tok) + return positional, is_delete + + +def _refspec_destination( + spec: str, current: str | None +) -> tuple[str | None, bool]: + """Return (destination branch or None, is_deletion_refspec).""" + spec = spec.lstrip("+") + if ":" in spec: + src, dst = spec.split(":", 1) + deletion = src == "" + else: + dst, deletion = spec, False + if dst == "HEAD": + dst = current or "" + if dst.startswith("refs/heads/"): + dst = dst[len("refs/heads/"):] + return (dst or None), deletion + + +def parse_push_targets(cmd: str, current: str | None) -> list[tuple[str, bool]]: + """Return [(destination_branch, is_delete), ...] for every `git push`. + + Every refspec of every `git push` on the command line is reported. + Handles: + git push / git push origin -> current branch + git push origin a b:c -> a, c + git push origin HEAD -> current branch + git push origin :b -> b (deletion refspec) + git push --delete origin a b -> a, b (delete) + option values (-o x, --repo x), quoting, `&&` / `;` chains, + `git -C dir push ...` + + Not interpreted (treated as allowed): --all / --mirror, wildcard + refspecs, `push.default` / upstream name mapping (the current branch name + stands in for the upstream), backtick substitution. + An empty list means "nothing to check"; callers allow it. + """ + targets: list[tuple[str, bool]] = [] + for args in _push_segments(_tokenize(cmd)): + positional, delete_flag = _parse_push_args(args) + refspecs = positional[1:] # positional[0] is the remote + if not refspecs: + if current: + targets.append((current, delete_flag)) + continue + skip_next = False + for spec in refspecs: + if skip_next: + skip_next = False + continue + if spec == "tag": # `git push origin tag ` + skip_next = True + continue + dst, deletion = _refspec_destination(spec, current) + if dst: + targets.append((dst, delete_flag or deletion)) + return targets + + def parse_push_target(cmd: str, current: str | None) -> tuple[str | None, bool]: - """Return (target_branch, is_delete). + """Return (target_branch, is_delete) for the most relevant destination. - Handles the common shapes: - git push -> upstream of current branch - git push origin -> upstream of current branch - git push origin -> - git push origin : -> - git push --delete origin -> , is_delete=True + Kept for callers of the original single-target API: returns the first + protected destination if there is one, else the first destination. Use + parse_push_targets to see every destination. - Returns target_branch=None when the parse is ambiguous; the caller + Returns target_branch=None when nothing could be parsed; the caller treats None as "allow" so we never block on a command we can't read. """ - tokens = re.split(r"\s+", cmd.strip()) - is_delete = "--delete" in tokens or "-d" in tokens + targets = parse_push_targets(cmd, current) + if not targets: + return None, False + protected = protected_branches() + for dst, is_delete in targets: + if dst in protected: + return dst, is_delete + return targets[0] + + +HOOK_NAME = "block-direct-integration-push" +CONTRACT = ( + "advisory client-side guard; fails open on anything it cannot parse; " + "not a security control and not a substitute for server-side protection" +) + + +def _settings_files(cwd: Path) -> list[Path]: + """Claude Code settings files that could register the hook, deduplicated.""" + files = [d / ".claude" / n for d in [cwd, *cwd.parents] + for n in ("settings.json", "settings.local.json")] + home = Path(os.path.expanduser("~")) + files += [home / ".claude" / n for n in ("settings.json", "settings.local.json")] + seen: set[Path] = set() + out = [] + for f in files: + if f not in seen: + seen.add(f) + out.append(f) + return out + + +def _covers_bash(matcher: object) -> bool: + if not matcher or matcher == "*": + return True try: - push_idx = tokens.index("push") + return re.fullmatch(str(matcher), "Bash") is not None + except re.error: + return False + + +def _script_exists(command: str, project_dir: Path) -> bool: + try: + words = shlex.split(command) except ValueError: - return None, is_delete - args = [t for t in tokens[push_idx + 1:] if not t.startswith("-")] - if len(args) >= 2: - spec = args[1] - dst = spec.split(":", 1)[1] if ":" in spec else spec - return dst.replace("refs/heads/", "") or None, is_delete - return current, is_delete + words = command.split() + for w in words: + if HOOK_NAME in w: + for var in ("CLAUDE_PROJECT_DIR",): + w = w.replace("${%s}" % var, str(project_dir)) + w = w.replace("$" + var, str(project_dir)) + return Path(os.path.expandvars(os.path.expanduser(w))).exists() + return False + + +def self_check() -> int: + """Print a JSON receipt of the effective configuration. Read-only. + + Reads this script and Claude Code settings files only. It runs no git + command, reads no stdin and touches no remote. Exit 0 when the hook is + registered for the Bash tool and its script exists, 1 otherwise. + """ + cwd = Path.cwd() + checked: list[dict] = [] + entries: list[dict] = [] + readable = 0 + for f in _settings_files(cwd): + if not f.is_file(): + continue + rec: dict = {"path": str(f), "exists": True} + checked.append(rec) + try: + data = json.loads(f.read_text()) + groups = data.get("hooks", {}).get("PreToolUse", []) + readable += 1 + except Exception as exc: # unreadable or wrong shape: report, go on + rec["error"] = f"{type(exc).__name__}: {exc}" + continue + project_dir = f.parent.parent if f.parent.name == ".claude" else f.parent + for group in groups if isinstance(groups, list) else []: + if not isinstance(group, dict): + continue + for h in group.get("hooks") or []: + cmd = h.get("command", "") if isinstance(h, dict) else "" + if HOOK_NAME in cmd: + entries.append({ + "settings": str(f), + "matcher": group.get("matcher", ""), + "command": cmd, + "covers_bash": _covers_bash(group.get("matcher")), + "script_exists": _script_exists(cmd, project_dir), + }) + if readable == 0: + status = "no-settings" + elif any(e["covers_bash"] and e["script_exists"] for e in entries): + status = "active" + elif any(e["covers_bash"] for e in entries): + status = "registered-script-missing" + elif entries: + status = "registered-not-for-bash" + else: + status = "not-registered" + + configured = os.environ.get(BRANCHES_ENV, "").strip() + script = Path(__file__).resolve() + receipt = { + "receipt": "push-guard-self-check/1", + "script": str(script), + "script_sha256": hashlib.sha256(script.read_bytes()).hexdigest(), + "python": sys.version.split()[0], + "protected_branches": sorted(protected_branches()), + "protected_branches_source": ( + f"env:{BRANCHES_ENV}" if configured else "default" + ), + "bypass_env": {"name": BYPASS_ENV, + "set_in_environment": os.environ.get(BYPASS_ENV) == "1"}, + "contract": CONTRACT, + "registration": { + "status": status, + "entries": entries, + "settings_checked": checked, + }, + "remote_mutation": False, + } + json.dump(receipt, sys.stdout, indent=2) + sys.stdout.write("\n") + return 0 if status == "active" else 1 def main() -> int: + if "--self-check" in sys.argv[1:]: + return self_check() try: payload = json.load(sys.stdin) except Exception: @@ -94,13 +359,15 @@ def main() -> int: if os.environ.get(BYPASS_ENV) == "1": return 0 - # Match the literal `git push` token sequence (not `gh push`, - # not arbitrary substrings). - if not re.search(r"\bgit\s+push\b", cmd): + # Cheap prefilter (not `gh push`, not arbitrary substrings); the real + # work is argv-aware parsing, which also handles `git -C dir push`. + if not re.search(r"\bgit\b.*\bpush\b", cmd, re.DOTALL): return 0 cwd = payload.get("cwd") or os.getcwd() current = get_current_branch(cwd) + # Prefers a protected destination, so a protected ref anywhere in a + # multi-refspec or chained command is found. target, is_delete = parse_push_target(cmd, current) if target and target in protected_branches(): diff --git a/test_block_direct_integration_push.py b/test_block_direct_integration_push.py index c30c360..9aed6cb 100644 --- a/test_block_direct_integration_push.py +++ b/test_block_direct_integration_push.py @@ -79,3 +79,238 @@ def test_malformed_payload_fails_open() -> None: [sys.executable, str(HOOK)], input="not json", text=True, capture_output=True ) assert proc.returncode == 0 + + +# --- PG-1: argv-aware parsing, every refspec checked ----------------------- + + +def test_second_refspec_to_protected_branch_is_blocked() -> None: + assert run_hook("git push origin feat/x main") == 2 + + +def test_later_refspec_colon_form_is_blocked() -> None: + assert run_hook("git push origin feat/x feat/y:master") == 2 + + +def test_multiple_feature_refspecs_are_allowed() -> None: + assert run_hook("git push origin feat/a feat/b:feat/c") == 0 + + +def test_plus_prefixed_refspec_is_read() -> None: + assert run_hook("git push origin +HEAD:main") == 2 + + +def test_delete_colon_form_is_blocked_as_delete() -> None: + assert push_guard.parse_push_targets("git push origin :main", "feat") == [ + ("main", True) + ] + + +def test_delete_flag_applies_to_every_refspec() -> None: + assert push_guard.parse_push_targets( + "git push --delete origin old main", "feat" + ) == [("old", True), ("main", True)] + + +def test_option_values_are_not_mistaken_for_remote_or_refspec() -> None: + cmd = "git push -o ci.skip --receive-pack x origin feat/x main" + assert push_guard.parse_push_targets(cmd, "feat") == [ + ("feat/x", False), + ("main", False), + ] + + +def test_attached_option_values_are_skipped() -> None: + cmd = "git push --push-option=main:abc123 origin feat/x" + assert push_guard.parse_push_targets(cmd, "feat") == [("feat/x", False)] + + +def test_quoted_refspec_is_unquoted() -> None: + assert run_hook("git push origin 'HEAD:main'") == 2 + assert run_hook('git push origin "main"') == 2 + + +def test_quoted_operator_inside_argument_is_not_a_second_command() -> None: + assert run_hook("git push origin 'feat/a && main'") == 0 + + +def test_second_command_on_the_line_is_checked() -> None: + assert run_hook("git push origin feat/x && git push origin main") == 2 + assert run_hook("git push origin feat/x; git push origin master") == 2 + + +def test_git_global_options_before_push_are_skipped() -> None: + assert run_hook("git -C /tmp/repo push origin main") == 2 + assert run_hook("git -c user.name=x push origin feat/x") == 0 + + +def test_env_prefix_before_git_is_handled() -> None: + assert run_hook("GIT_TRACE=1 git push origin main") == 2 + + +def test_bare_head_refspec_means_current_branch() -> None: + assert push_guard.parse_push_targets("git push origin HEAD", "main") == [ + ("main", False) + ] + assert push_guard.parse_push_targets("git push origin HEAD", "feat") == [ + ("feat", False) + ] + + +def test_implicit_destination_uses_current_branch() -> None: + assert push_guard.parse_push_targets("git push", "main") == [("main", False)] + assert push_guard.parse_push_targets("git push origin", "main") == [ + ("main", False) + ] + assert push_guard.parse_push_targets("git push -u origin", "feat") == [ + ("feat", False) + ] + + +def test_refs_heads_prefix_is_stripped_on_every_refspec() -> None: + assert push_guard.parse_push_targets( + "git push origin a refs/heads/b:refs/heads/main", "feat" + ) == [("a", False), ("main", False)] + + +def test_tag_refspec_named_like_a_branch_is_not_a_branch_push() -> None: + assert run_hook("git push origin refs/tags/main") == 0 + + +def test_unbalanced_quote_still_checks_with_whitespace_split() -> None: + assert run_hook("git push origin main 'oops") == 2 + + +def test_parse_push_target_returns_protected_target_when_any() -> None: + target, is_delete = push_guard.parse_push_target( + "git push origin feat main", "feat" + ) + assert target == "main" + assert is_delete is False + + +def test_non_push_git_command_yields_no_targets() -> None: + assert push_guard.parse_push_targets("git status main", "main") == [] + assert push_guard.parse_push_targets("gh pr create --base main", "main") == [] + + +# --- PG-2: --self-check, read-only registration receipt -------------------- + + +def run_self_check( + cwd: Path, home: Path, env: dict[str, str] | None = None +) -> tuple[int, dict]: + proc = subprocess.run( + [sys.executable, str(HOOK), "--self-check"], + text=True, + capture_output=True, + cwd=str(cwd), + env={"PATH": "/usr/bin:/bin", "HOME": str(home), **(env or {})}, + stdin=subprocess.DEVNULL, + ) + return proc.returncode, json.loads(proc.stdout) + + +def write_settings(path: Path, hooks: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"hooks": hooks})) + + +def make_dirs(tmp_path: Path) -> tuple[Path, Path]: + proj, home = tmp_path / "proj", tmp_path / "home" + proj.mkdir() + home.mkdir() + return proj, home + + +def bash_entry(command: str, matcher: str = "Bash") -> dict: + return { + "PreToolUse": [ + {"matcher": matcher, "hooks": [{"type": "command", "command": command}]} + ] + } + + +def test_self_check_without_any_settings_reports_no_settings(tmp_path: Path) -> None: + proj, home = make_dirs(tmp_path) + rc, receipt = run_self_check(proj, home) + assert rc == 1 + assert receipt["registration"]["status"] == "no-settings" + assert receipt["remote_mutation"] is False + + +def test_self_check_settings_without_the_hook_is_not_registered( + tmp_path: Path, +) -> None: + proj, home = make_dirs(tmp_path) + write_settings(proj / ".claude" / "settings.json", bash_entry("/x/other-hook.py")) + rc, receipt = run_self_check(proj, home) + assert rc == 1 + assert receipt["registration"]["status"] == "not-registered" + assert receipt["registration"]["entries"] == [] + + +def test_self_check_registered_for_bash_is_active(tmp_path: Path) -> None: + proj, home = make_dirs(tmp_path) + write_settings(home / ".claude" / "settings.json", bash_entry(f"python3 {HOOK}")) + rc, receipt = run_self_check(proj, home) + assert rc == 0 + reg = receipt["registration"] + assert reg["status"] == "active" + assert reg["entries"][0]["matcher"] == "Bash" + assert reg["entries"][0]["script_exists"] is True + + +def test_self_check_registered_for_another_tool_is_not_active( + tmp_path: Path, +) -> None: + proj, home = make_dirs(tmp_path) + write_settings( + proj / ".claude" / "settings.local.json", bash_entry(str(HOOK), matcher="Edit") + ) + rc, receipt = run_self_check(proj, home) + assert rc == 1 + assert receipt["registration"]["status"] == "registered-not-for-bash" + + +def test_self_check_registered_with_missing_script_is_not_active( + tmp_path: Path, +) -> None: + proj, home = make_dirs(tmp_path) + write_settings( + proj / ".claude" / "settings.json", + bash_entry("/nonexistent/block-direct-integration-push.py"), + ) + rc, receipt = run_self_check(proj, home) + assert rc == 1 + assert receipt["registration"]["status"] == "registered-script-missing" + + +def test_self_check_reports_protected_branches_and_contract(tmp_path: Path) -> None: + proj, home = make_dirs(tmp_path) + _, receipt = run_self_check(proj, home) + assert receipt["protected_branches"] == ["develop", "main", "master", "trunk"] + assert receipt["protected_branches_source"] == "default" + assert "fails open" in receipt["contract"] + assert len(receipt["script_sha256"]) == 64 + _, receipt = run_self_check( + proj, home, env={"GIT_PROTECTED_BRANCHES": "release, prod"} + ) + assert receipt["protected_branches"] == ["prod", "release"] + assert receipt["protected_branches_source"] == "env:GIT_PROTECTED_BRANCHES" + + +def test_self_check_unreadable_settings_does_not_crash(tmp_path: Path) -> None: + proj, home = make_dirs(tmp_path) + (proj / ".claude").mkdir() + (proj / ".claude" / "settings.json").write_text("{not json") + rc, receipt = run_self_check(proj, home) + assert rc == 1 + assert receipt["registration"]["status"] == "no-settings" + assert receipt["registration"]["settings_checked"][0]["error"] + + +def test_self_check_needs_no_stdin_and_no_git(tmp_path: Path) -> None: + proj, home = make_dirs(tmp_path) + _, receipt = run_self_check(proj, home, env={"PATH": "/nonexistent"}) + assert receipt["receipt"] == "push-guard-self-check/1"