Push Guard

Event

PreToolUse

Trigger

tool-name — Bash

Language

python

Side effects

• blocks force-push to protected branches with a deny decision
• emits PUSH-GUARD-VIOLATION sentinel to stderr for unblocked violations

Platforms

linuxmacoswindows

Notes

Requires git. PreToolUse path parses the Bash command from stdin JSON. Wrapper mode reads WRAPPED_ARGV. Protected branches read from ~/.ai/settings.json (key: protectedBranches) with a default of ['main']. Cross-platform via 'ai hooks run'.

Full coverage

Routes every git push through this guard regardless of caller — without the wrapper only AI tool Bash calls are intercepted. ai setup --git-shim

Depends on

hook/lib

Script · push-guard.py

#!/usr/bin/env python3
"""hooks/push-guard.py — block force-pushes to protected branches.

Two enforcement paths:
  PreToolUse  — parses 'git push' from the AI tool Bash JSON payload.
  Wrapper     -- invoked as the git shim (WRAPPED_CMD=git, WRAPPED_ARGV=...)
                 via --mode=wrapper. Intercepts every git push regardless
                 of caller.

Blocks (exit 1 / deny decision) when:
  - --force or -f or --force-with-lease targets a protected branch.
  - The destination branch resolves to a protected branch name.

Protected branches: read from ~/.ai/settings.json key 'protectedBranches'.
Default: ['main'].

Self-check:
  --self-check  exits 0 (no external deps required).
"""
from __future__ import annotations

import json
import os
import re
import shlex
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import _lib  # noqa: E402


# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------

def _protected_branches() -> list[str]:
    settings_path = Path(os.environ.get("AI_ROOT", str(Path.home() / ".ai"))) / "settings.json"
    try:
        data = json.loads(settings_path.read_text(encoding="utf-8"))
        branches = data.get("protectedBranches", [])
        if isinstance(branches, list) and branches:
            return [str(b) for b in branches]
    except (OSError, json.JSONDecodeError, ValueError):
        pass
    return ["main"]


# ---------------------------------------------------------------------------
# Push argument parsing
# ---------------------------------------------------------------------------

_FORCE_FLAGS = {"--force", "-f", "--force-with-lease", "--force-if-includes"}


def _parse_push(tokens: list[str]) -> tuple[bool, str | None]:
    """Return (is_force, destination_branch) from a tokenised git push argv.

    destination_branch is the explicit refspec target or None if not given
    (caller must resolve via current branch name).
    """
    # Strip 'git' and any global git flags before the subcommand.
    argv = list(tokens)
    while argv and argv[0] != "push":
        argv.pop(0)
    if not argv or argv[0] != "push":
        return False, None
    argv.pop(0)  # remove 'push'

    is_force = False
    remote = None
    refspec = None
    i = 0
    while i < len(argv):
        tok = argv[i]
        if tok in _FORCE_FLAGS:
            is_force = True
        elif tok.startswith("--force-with-lease=") or tok.startswith("--force-if-includes="):
            is_force = True
        elif tok.startswith("-") and not tok.startswith("--"):
            # Short flags cluster: -fu means force + set-upstream
            if "f" in tok[1:]:
                is_force = True
        elif not tok.startswith("-"):
            if remote is None:
                remote = tok
            elif refspec is None:
                refspec = tok
        i += 1

    # Refspec format: [+]src:dst or just branch-name
    dest = None
    if refspec:
        # strip leading '+' (force marker)
        ref = refspec.lstrip("+")
        if ":" in ref:
            dest = ref.split(":", 1)[1].removeprefix("refs/heads/")
        else:
            dest = ref
    return is_force, dest


def _current_branch(cwd: str) -> str | None:
    try:
        r = subprocess.run(
            ["git", "branch", "--show-current"],
            capture_output=True, text=True, check=False, cwd=cwd,
        )
        if r.returncode == 0:
            return r.stdout.strip() or None
    except FileNotFoundError:
        pass
    return None


def _is_protected(branch: str | None, protected: list[str]) -> bool:
    return branch in protected if branch else False


# ---------------------------------------------------------------------------
# Command scanning (handles compound shell commands)
# ---------------------------------------------------------------------------

def _git_push_segments(command: str) -> list[list[str]]:
    """Return a list of token lists, one per git push invocation in command."""
    segments: list[list[str]] = []
    for segment in re.split(r"[;&|\n]+", command):
        try:
            tokens = shlex.split(segment.strip(), posix=True)
        except ValueError:
            tokens = segment.strip().split()
        # Strip leading VAR=value assignments
        while tokens and re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", tokens[0]):
            tokens.pop(0)
        # Find git ... push
        for i, tok in enumerate(tokens):
            if tok == "git":
                rest = tokens[i:]
                # Skip global git flags
                j = 1
                while j < len(rest) and rest[j].startswith("-"):
                    j += 1
                if j < len(rest) and rest[j] == "push":
                    segments.append(rest)
    return segments


# ---------------------------------------------------------------------------
# PreToolUse deny helper
# ---------------------------------------------------------------------------

def _deny_pretooluse(reason: str) -> None:
    """Emit a PreToolUse permission-deny decision on stdout and exit 0."""
    print(json.dumps({
        "hookSpecificOutput": {
            "permissionDecision": "deny",
            "permissionDecisionReason": reason,
        }
    }), flush=True)
    sys.exit(0)


# ---------------------------------------------------------------------------
# PreToolUse path
# ---------------------------------------------------------------------------

def _handle_pretooluse(payload: dict) -> None:
    command = (payload.get("tool_input") or {}).get("command", "")
    cwd = payload.get("cwd") or os.getcwd()
    if not command:
        return

    protected = _protected_branches()
    segments = _git_push_segments(command)
    if not segments:
        return

    for tokens in segments:
        is_force, dest = _parse_push(tokens)
        if not is_force:
            continue
        branch = dest or _current_branch(cwd)
        if _is_protected(branch, protected):
            msg = (
                f"push-guard: force-push to protected branch '{branch}' is not allowed. "
                f"Protected branches: {', '.join(protected)}."
            )
            _deny_pretooluse(msg)  # exits


# ---------------------------------------------------------------------------
# Wrapper path
# ---------------------------------------------------------------------------

def _handle_wrapper() -> int:
    argv_json = os.environ.get("WRAPPED_ARGV", "[]")
    cwd = os.getcwd()
    try:
        tokens = ["push"] + json.loads(argv_json)  # WRAPPED_ARGV is args after 'git'
        # Normalise: prepend 'git' so _parse_push can find the push subcommand
        tokens = ["git"] + json.loads(argv_json)
    except (json.JSONDecodeError, ValueError):
        return 0

    protected = _protected_branches()
    is_force, dest = _parse_push(tokens)
    if not is_force:
        return 0

    branch = dest or _current_branch(cwd)
    if _is_protected(branch, protected):
        msg = (
            f"push-guard: force-push to protected branch '{branch}' is not allowed. "
            f"Protected branches: {', '.join(protected)}."
        )
        sys.stderr.write(msg + "\n")
        return 1  # blocks the push
    return 0


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

def main() -> None:
    if "--self-check" in sys.argv:
        _lib.log("self-check OK")
        sys.exit(0)

    if "--mode=wrapper" in sys.argv or os.environ.get("WRAPPED_CMD"):
        sys.exit(_handle_wrapper())

    raw = sys.stdin.read()
    if not raw.strip():
        sys.exit(0)
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError:
        sys.exit(0)

    _handle_pretooluse(payload)


if __name__ == "__main__":
    main()