diff --git a/src/brigade/guard/git_scan.py b/src/brigade/guard/git_scan.py index 8cbbc2b1..3b88c6ed 100644 --- a/src/brigade/guard/git_scan.py +++ b/src/brigade/guard/git_scan.py @@ -2,12 +2,13 @@ import argparse import json +import re import subprocess import sys from pathlib import Path from .engine import scan_text -from .policy import Policy, load_policy +from .policy import Policy, default_policy, load_policy from .report import to_text @@ -37,12 +38,28 @@ def main(argv: list[str] | None = None) -> int: action="store_true", help="scan content INTRODUCED across commit history (added lines), not just the current tip", ) - parser.add_argument("--range", dest="rev_range", help="revision range for --history, e.g. origin/main..HEAD") - parser.add_argument("--all", action="store_true", help="with --history, scan all reachable commits") + history_source = parser.add_mutually_exclusive_group() + history_source.add_argument( + "--range", dest="rev_range", help="revision range for --history, e.g. origin/main..HEAD" + ) + history_source.add_argument("--all", action="store_true", help="with --history, scan all reachable commits") + history_source.add_argument( + "--revs-stdin", + action="store_true", + help=( + "with --history, read the exact set of commit revisions to scan from " + "stdin (one full 40- or 64-character hex SHA per line) instead of " + "calling git rev-list. Lets a caller batch a precomputed revision " + "set into one scan process and keeps revisions off argv (no ARG_MAX " + "limit, no option injection)." + ), + ) parser.add_argument("--json", action="store_true", help="emit JSON report") args = parser.parse_args(argv) + if args.revs_stdin and not args.history: + parser.error("--revs-stdin requires --history") - policy = load_policy(args.policy) if args.policy else _default_repo_policy() + policy = _load_repo_policy(args.policy) _merge_allow_values_from(policy, args.allow_values_from or []) if args.history: @@ -146,6 +163,20 @@ def _scan_history(policy: Policy, args: argparse.Namespace) -> int: def _history_revs(args: argparse.Namespace) -> list[str]: + if getattr(args, "revs_stdin", False): + revs: list[str] = [] + for line_number, line in enumerate(sys.stdin, start=1): + rev = line.strip() + if not rev: + continue + if re.fullmatch(r"(?:[0-9a-fA-F]{40}|[0-9a-fA-F]{64})", rev) is None: + print( + f"invalid revision on stdin line {line_number}: expected a full 40- or 64-character hex SHA", + file=sys.stderr, + ) + raise SystemExit(2) + revs.append(rev.lower()) + return revs if args.all: cmd = ["git", "rev-list", "--all"] elif args.rev_range: @@ -167,7 +198,9 @@ def _added_lines(rev: str) -> str: check=False, ) if proc.returncode != 0: - return "" + detail = (proc.stderr or "git show failed").strip() + print(f"cannot read commit {rev}: {detail}", file=sys.stderr) + raise SystemExit(2) out = [] for line in proc.stdout.splitlines(): if line.startswith("+") and not line.startswith("+++"): @@ -219,6 +252,16 @@ def _default_repo_policy() -> Policy: ) +def _load_repo_policy(raw: str | None) -> Policy: + if raw is None: + return _default_repo_policy() + path = Path(raw) + if path.is_file() or path.parent != Path("."): + return load_policy(path) + name = path.name if path.suffix == ".json" else f"{path.name}.json" + return load_policy(default_policy(name)) + + def _tracked_paths(*, all_tracked: bool, cwd: Path | None = None) -> list[Path]: if all_tracked: cmd = ["git", "ls-files"] diff --git a/src/brigade/templates/hooks/pre-push b/src/brigade/templates/hooks/pre-push index 272901d0..0cbc4673 100644 --- a/src/brigade/templates/hooks/pre-push +++ b/src/brigade/templates/hooks/pre-push @@ -9,10 +9,37 @@ # # Uses Brigade's embedded content guard. CONTENT_GUARD_DIR is an explicit # compatibility override for an older standalone checkout. +# +# Two scans run per policy: +# 1. tracked tip - working tree files (brigade scrub / content_guard scan) +# 2. push history - content INTRODUCED by the commits being pushed +# (closes the forward-scrub gap: a clean tip can still +# sit on top of commits that leak in their diffs) +# +# For brand-new branches (remote SHA all-zero) the history scan derives the +# exclusion set from `git ls-remote ` rather than local tracking refs, +# which can be stale. Only commits not reachable from any advertised ref are +# scanned, batched into one guard history process per policy via --revs-stdin. set -euo pipefail REPO_ROOT="$(git rev-parse --show-toplevel)" +cd "$REPO_ROOT" + +# Git invokes pre-push as: pre-push . The URL is the +# destination actually being pushed to; it is the only source of truth for +# what the remote currently advertises. Local remote-tracking refs +# (refs/remotes//*) are NOT used to derive the exclusion set, because +# they can be stale: a tracking ref can point at a commit the remote no +# longer holds, or be absent for a commit the remote already holds. Using +# them would over- or under-exclude and either miss a real leak or block on +# already-public content. ${2:-} keeps `set -u` happy when a caller forgets +# the URL (git always passes both). +REMOTE_URL="${2:-}" + rc=0 +PLUMBING_ERR=0 +STDIN_REFS="$(cat)" + if [[ -n "${CONTENT_GUARD_DIR:-}" ]]; then POLICY="${CONTENT_GUARD_POLICY:-$CONTENT_GUARD_DIR/policies/public-repo.json}" if [[ ! -d "$CONTENT_GUARD_DIR" ]]; then @@ -24,7 +51,21 @@ if [[ -n "${CONTENT_GUARD_DIR:-}" ]]; then exit 1 fi echo "pre-push: scanning $REPO_ROOT with explicit content-guard compatibility override" - PYTHONPATH="$CONTENT_GUARD_DIR/src" python3 -m content_guard scan "$REPO_ROOT" --policy "$POLICY" || rc=$? + RUN_TIP() { + local pol="$1" + PYTHONPATH="$CONTENT_GUARD_DIR/src" python3 -m content_guard scan "$REPO_ROOT" --policy "$pol" || rc=$? + } + RUN_HISTORY() { + local pol="$1" + PYTHONPATH="$CONTENT_GUARD_DIR/src" python3 -m content_guard.git_scan \ + --history --revs-stdin --policy "$pol" "${ALLOW_FROM[@]+"${ALLOW_FROM[@]}"}" || rc=$? + } + RUN_HISTORY_RANGE() { + local range="$1" + local pol="$2" + PYTHONPATH="$CONTENT_GUARD_DIR/src" python3 -m content_guard.git_scan \ + --history --range "$range" --policy "$pol" "${ALLOW_FROM[@]+"${ALLOW_FROM[@]}"}" || rc=$? + } else POLICY="${CONTENT_GUARD_POLICY:-public-repo}" if ! command -v brigade >/dev/null 2>&1; then @@ -32,14 +73,142 @@ else exit 1 fi echo "pre-push: scanning $REPO_ROOT with Brigade's embedded guard" - brigade scrub --target "$REPO_ROOT" --policy "$POLICY" --no-receipt || rc=$? + RUN_TIP() { + local pol="$1" + brigade scrub --target "$REPO_ROOT" --policy "$pol" --no-receipt || rc=$? + } + RUN_HISTORY() { + local pol="$1" + brigade guard git --history --revs-stdin --policy "$pol" "${ALLOW_FROM[@]+"${ALLOW_FROM[@]}"}" || rc=$? + } + RUN_HISTORY_RANGE() { + local range="$1" + local pol="$2" + brigade guard git --history --range "$range" --policy "$pol" "${ALLOW_FROM[@]+"${ALLOW_FROM[@]}"}" || rc=$? + } +fi + +# Optional private policy (a private identifier denylist kept out of public +# repos). If present, the hook runs a second scan pass against it. +EXTRA_POLICY="${CONTENT_GUARD_EXTRA_POLICY:-$HOME/.config/content-guard/internal.json}" +ALLOW_FROM=() +[[ -f "$EXTRA_POLICY" ]] && ALLOW_FROM=(--allow-values-from "$EXTRA_POLICY") + +# Track the worst scanner outcome across all scans: BLOCKED (rc=1, a real +# leak verdict) vs SCANNER_ERR (rc>1, the scanner failed to run). Issue #82: +# never mislabel a scanner/plumbing failure as found violations. +BLOCKED=0 +SCANNER_ERR=0 +note_rc() { + if (( rc > 1 )); then SCANNER_ERR=$rc; elif (( rc == 1 )); then BLOCKED=1; fi + rc=0 +} + +ADVERTISED_REFS="" +REMOTE_ENUMERATED=0 + +prepare_new_branch_ranges() { + local needs_remote=0 + local _lref lsha _rref rsha + while read -r _lref lsha _rref rsha; do + [[ -z "${lsha:-}" || "$lsha" =~ ^0+$ ]] && continue + if [[ "$rsha" =~ ^0+$ ]]; then + needs_remote=1 + break + fi + done < <(printf '%s\n' "$STDIN_REFS") + [[ "$needs_remote" -eq 1 ]] || return 0 + + if [[ -z "$REMOTE_URL" ]]; then + echo "pre-push: [history] cannot enumerate remote refs (no destination URL)" >&2 + PLUMBING_ERR=1 + return + fi + + local advertised + # Keep peeled annotated-tag lines (`refs/tags/^{}`), which --refs + # suppresses. The peeled commit can be present locally even when the tag + # object is not, and it still marks that commit as already public. + if ! advertised="$(GIT_TERMINAL_PROMPT=0 git ls-remote "$REMOTE_URL" 2>&1)"; then + echo "pre-push: [history] failed to enumerate advertised refs for $REMOTE_URL" >&2 + while IFS= read -r line; do + [[ -n "$line" ]] && echo "pre-push: $line" >&2 + done <<<"$advertised" + PLUMBING_ERR=1 + return + fi + ADVERTISED_REFS="$advertised" + REMOTE_ENUMERATED=1 +} + +scan_with() { + local pol="$1" + local new_revs new_count range + RUN_TIP "$pol" + note_rc + while read -r _lref lsha _rref rsha; do + [[ -z "${lsha:-}" || "$lsha" =~ ^0+$ ]] && continue # branch deletion + if [[ "$rsha" =~ ^0+$ ]]; then + [[ "$REMOTE_ENUMERATED" -eq 1 ]] || continue + if ! git rev-parse -q --verify "${lsha}^{commit}" >/dev/null; then + echo "pre-push: [history] local commit $lsha does not resolve" >&2 + PLUMBING_ERR=1 + continue + fi + if ! new_revs="$( + printf '%s\n' "$ADVERTISED_REFS" | + awk 'NF { print "^" $1 }' | + git rev-list --ignore-missing "$lsha" --stdin + )"; then + echo "pre-push: [history] failed to resolve local history for $lsha" >&2 + PLUMBING_ERR=1 + continue + fi + if [[ -z "$new_revs" ]]; then + echo "pre-push: [history] no new commits to scan for $lsha" + continue + fi + new_count="$(printf '%s\n' "$new_revs" | grep -c .)" + echo "pre-push: [history] scanning introduced content in $new_count new commit(s) for ${_lref:-$lsha}" + # Batch the exact new-commit set into ONE guard history process via + # --revs-stdin. Revisions are data on stdin, not argv: no shell + # splitting, no option injection, one process per policy regardless + # of how many commits are new. + RUN_HISTORY "$pol" <<<"$new_revs" + note_rc + else + range="$rsha..$lsha" + echo "pre-push: [history] scanning introduced content in $range" + RUN_HISTORY_RANGE "$range" "$pol" + note_rc + fi + done < <(printf '%s\n' "$STDIN_REFS") +} + +prepare_new_branch_ranges +scan_with "$POLICY" +if [[ -n "$EXTRA_POLICY" && -f "$EXTRA_POLICY" ]]; then + scan_with "$EXTRA_POLICY" fi -if [[ "$rc" -eq 0 ]]; then - exit 0 +if (( SCANNER_ERR > 0 )); then + # Any non-1 exit code is the scanner failing to run (missing deps, bad + # policy, crash, or a plumbing failure surfaced as exit 2), not a leak + # verdict. Do not mislabel it as found violations (issue #82). + echo >&2 + echo "pre-push: content-guard failed to run (exit code $SCANNER_ERR); this is a scanner error, not a leak verdict." >&2 + echo "pre-push: re-run it directly to see the error, then push again once the scanner works." >&2 + exit 1 +fi + +if (( PLUMBING_ERR > 0 )); then + echo >&2 + echo "pre-push: remote history could not be determined; this is a git plumbing error, not a leak verdict." >&2 + echo "pre-push: fix the remote or local Git state, then push again." >&2 + exit 1 fi -if [[ "$rc" -eq 1 ]]; then +if [[ "$BLOCKED" -ne 0 ]]; then echo >&2 echo "pre-push: BLOCKED. content-guard found violations." >&2 echo "pre-push: fix the leak, or add an inline allow-tag on the offending line:" >&2 @@ -47,9 +216,4 @@ if [[ "$rc" -eq 1 ]]; then exit 1 fi -# Any other exit code is the scanner failing to run (missing deps, bad policy, -# crash), not a leak verdict. Do not mislabel it as found violations. -echo >&2 -echo "pre-push: content-guard failed to run (exit code $rc); this is a scanner error, not a leak verdict." >&2 -echo "pre-push: re-run it directly to see the error, then push again once the scanner works." >&2 -exit 1 +exit 0 diff --git a/tests/guard/test_cli.py b/tests/guard/test_cli.py index 34210fc6..eda61b15 100644 --- a/tests/guard/test_cli.py +++ b/tests/guard/test_cli.py @@ -291,6 +291,7 @@ def test_git_commit_scan_blocks_approved_agent_coauthor_trailer_by_default(self) "-m", "feat: example", "-m", + # content-guard: allow email "Co-authored-by: Codex ", ], cwd=repo, @@ -655,6 +656,221 @@ def test_n8n_validation_pack_reports_expectation_failure(self) -> None: self.assertFalse(payload["ok"]) self.assertIn("expected blocked", payload["fixtures"][0]["failures"][0]) + def test_git_scan_revs_stdin_scans_exact_set(self) -> None: + # --revs-stdin with --history scans exactly the revisions fed on + # stdin, bypassing `git rev-list`. Lets the pre-push hook batch a + # precomputed new-commit set into one process without putting + # revisions on argv (no ARG_MAX, no option injection). + with TemporaryDirectory() as tmp: + repo = Path(tmp) + self._init_repo(repo) + (repo / "clean.txt").write_text("clean\n") + subprocess.run(["git", "add", "clean.txt"], cwd=repo, check=True) + subprocess.run( + ["git", "commit", "-m", "feat: clean"], + cwd=repo, + check=True, + capture_output=True, + text=True, + ) + # content-guard: allow api-key-assignment + (repo / "leak.txt").write_text('api_key = "AKIAIOSFODNN7EXAMPLE0123456789"\n') + subprocess.run(["git", "add", "leak.txt"], cwd=repo, check=True) + subprocess.run( + ["git", "commit", "-q", "-m", "feat: leak"], + cwd=repo, + check=True, + capture_output=True, + text=True, + ) + leak_sha = subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=repo, capture_output=True, text=True, check=True + ).stdout.strip() + clean_sha = subprocess.run( + ["git", "rev-parse", "HEAD~1"], cwd=repo, capture_output=True, text=True, check=True + ).stdout.strip() + + proc = subprocess.run( + [ + sys.executable, + "-m", + "brigade.guard.git_scan", + "--history", + "--revs-stdin", + "--policy", + str(ROOT / "src" / "brigade" / "guard" / "policies" / "public-repo.json"), + "--json", + ], + cwd=repo, + env={"PYTHONPATH": str(ROOT / "src")}, + input=f"{leak_sha}\n{clean_sha}\n", + capture_output=True, + text=True, + check=False, + ) + + self.assertEqual(proc.returncode, 1, msg=proc.stdout + proc.stderr) + payload = json.loads(proc.stdout) + self.assertTrue(payload["blocked"]) + self.assertEqual(payload["commits_scanned"], 2) + self.assertEqual(payload["commits_with_findings"], 1) + self.assertEqual(payload["commits"][0]["commit"], leak_sha) + + def test_git_scan_revs_stdin_empty_input_scans_nothing(self) -> None: + # Empty stdin -> empty revision set -> clean, zero commits scanned. + with TemporaryDirectory() as tmp: + repo = Path(tmp) + self._init_repo(repo) + proc = subprocess.run( + [ + sys.executable, + "-m", + "brigade.guard.git_scan", + "--history", + "--revs-stdin", + "--policy", + str(ROOT / "src" / "brigade" / "guard" / "policies" / "public-repo.json"), + "--json", + ], + cwd=repo, + env={"PYTHONPATH": str(ROOT / "src")}, + input="", + capture_output=True, + text=True, + check=False, + ) + + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + payload = json.loads(proc.stdout) + self.assertFalse(payload["blocked"]) + self.assertEqual(payload["commits_scanned"], 0) + + def test_git_scan_revs_stdin_rejects_non_sha_input(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + self._init_repo(repo) + proc = subprocess.run( + [ + sys.executable, + "-m", + "brigade.guard.git_scan", + "--history", + "--revs-stdin", + ], + cwd=repo, + env={"PYTHONPATH": str(ROOT / "src")}, + input="--help\n", + capture_output=True, + text=True, + check=False, + ) + + self.assertEqual(proc.returncode, 2, msg=proc.stdout + proc.stderr) + self.assertIn("expected a full 40- or 64-character hex SHA", proc.stderr) + + def test_git_scan_revs_stdin_rejects_unknown_sha(self) -> None: + # A well-formed SHA that does not resolve locally must fail closed + # (exit 2) rather than be silently skipped as clean. + with TemporaryDirectory() as tmp: + repo = Path(tmp) + self._init_repo(repo) + proc = subprocess.run( + [ + sys.executable, + "-m", + "brigade.guard.git_scan", + "--history", + "--revs-stdin", + ], + cwd=repo, + env={"PYTHONPATH": str(ROOT / "src")}, + input=f"{'0' * 40}\n", + capture_output=True, + text=True, + check=False, + ) + + self.assertEqual(proc.returncode, 2, msg=proc.stdout + proc.stderr) + self.assertIn("cannot read commit", proc.stderr) + + def test_git_scan_revs_stdin_requires_history(self) -> None: + proc = subprocess.run( + [sys.executable, "-m", "brigade.guard.git_scan", "--revs-stdin"], + cwd=ROOT, + env={"PYTHONPATH": str(ROOT / "src")}, + input="", + capture_output=True, + text=True, + check=False, + ) + + self.assertEqual(proc.returncode, 2, msg=proc.stdout + proc.stderr) + self.assertIn("--revs-stdin requires --history", proc.stderr) + + def test_git_scan_revs_stdin_accepts_sha256(self) -> None: + # SHA-256 repos emit 64-character SHAs; --revs-stdin must accept them. + with TemporaryDirectory() as tmp: + repo = Path(tmp) + subprocess.run( + ["git", "init", "--object-format=sha256"], cwd=repo, check=True, capture_output=True, text=True + ) + subprocess.run(["git", "config", "user.name", "Example User"], cwd=repo, check=True) + subprocess.run(["git", "config", "user.email", "user@example"], cwd=repo, check=True) + (repo / "README.md").write_text("example\n") + subprocess.run(["git", "add", "README.md"], cwd=repo, check=True) + subprocess.run( + ["git", "commit", "-m", "feat: example"], cwd=repo, check=True, capture_output=True, text=True + ) + sha = subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=repo, capture_output=True, text=True, check=True + ).stdout.strip() + self.assertEqual(len(sha), 64) + + proc = subprocess.run( + [ + sys.executable, + "-m", + "brigade.guard.git_scan", + "--history", + "--revs-stdin", + "--policy", + str(ROOT / "src" / "brigade" / "guard" / "policies" / "public-repo.json"), + "--json", + ], + cwd=repo, + env={"PYTHONPATH": str(ROOT / "src")}, + input=f"{sha}\n", + capture_output=True, + text=True, + check=False, + ) + + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + payload = json.loads(proc.stdout) + self.assertEqual(payload["commits_scanned"], 1) + + def test_git_scan_accepts_packaged_policy_name(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + self._init_repo(repo) + proc = subprocess.run( + [ + sys.executable, + "-m", + "brigade.guard.git_scan", + "--all-tracked", + "--policy", + "public-repo", + ], + cwd=repo, + env={"PYTHONPATH": str(ROOT / "src")}, + capture_output=True, + text=True, + check=False, + ) + + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + def _init_repo(self, repo: Path) -> None: subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True, text=True) subprocess.run(["git", "config", "user.name", "Example User"], cwd=repo, check=True) diff --git a/tests/test_pre_push_hook.py b/tests/test_pre_push_hook.py index d4df1149..758264db 100644 --- a/tests/test_pre_push_hook.py +++ b/tests/test_pre_push_hook.py @@ -2,14 +2,25 @@ from __future__ import annotations +import json +import os +import subprocess +import unittest from pathlib import Path +from tempfile import TemporaryDirectory import brigade +_HOOK = Path(brigade.__file__).resolve().parent / "templates" / "hooks" / "pre-push" +_VENV_BIN = Path(brigade.__file__).resolve().parents[2] / ".venv" / "bin" +_ZERO40 = "0" * 40 +# content-guard: allow api-key-assignment +_LEAK_LINE = 'api_key = "AKIAIOSFODNN7EXAMPLE0123456789"\n' +_CLEAN_LINE = "just a normal change\n" + def _hook_text() -> str: - hook = Path(brigade.__file__).resolve().parent / "templates" / "hooks" / "pre-push" - return hook.read_text() + return _HOOK.read_text() def test_pre_push_hook_captures_exit_code(): @@ -19,9 +30,14 @@ def test_pre_push_hook_captures_exit_code(): def test_pre_push_hook_only_blocks_on_findings_exit_code(): text = _hook_text() - # The "found violations" message is gated on exit code 1 specifically. - assert '"$rc" -eq 1' in text + # Issue #82: the "found violations" (BLOCKED) message is gated on the + # leak verdict (scanner exit 1) specifically, and scanner/plumbing + # failures (exit >1) are reported as a separate "failed to run" message + # rather than mislabeled as leaks. The hook tracks these as distinct + # outcomes (BLOCKED vs SCANNER_ERR) across multiple scans. assert "BLOCKED. content-guard found violations." in text + assert "failed to run" in text + assert "not a leak verdict" in text def test_pre_push_hook_reports_scanner_errors_separately(): @@ -41,3 +57,438 @@ def test_pre_push_hook_keeps_external_checkout_as_explicit_override(): text = _hook_text() assert 'if [[ -n "${CONTENT_GUARD_DIR:-}" ]]' in text assert 'PYTHONPATH="$CONTENT_GUARD_DIR/src"' in text + + +def test_pre_push_hook_history_uses_revs_stdin_for_new_branches(): + text = _hook_text() + assert "--revs-stdin" in text + assert 'git ls-remote "$REMOTE_URL"' in text + assert "git rev-list --ignore-missing" in text + + +def test_pre_push_hook_is_compatible_with_macos_bash_3(): + text = _hook_text() + assert "shopt -s lastpipe" not in text + assert "declare -A" not in text + + +# --------------------------------------------------------------------------- +# Functional tests: actually run the seeded hook against temp repos with a +# bare remote, exercising the new-branch history-scan fix. These verify the +# ls-remote-derived exclusion set, batching, fail-closed behavior, and the +# preserved existing-branch / deletion / anonymous-URL paths. +# --------------------------------------------------------------------------- + + +def _git_env(repo: Path) -> dict: + env = { + **os.environ, + "GIT_CONFIG_GLOBAL": os.devnull, + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_AUTHOR_NAME": "t", + "GIT_AUTHOR_EMAIL": "t@example.com", + "GIT_COMMITTER_NAME": "t", + "GIT_COMMITTER_EMAIL": "t@example.com", + "HOME": str(repo), + "XDG_CONFIG_HOME": str(repo / ".config"), + } + # Put the editable-install brigade on PATH so the hook's `brigade` and + # `brigade guard git` invocations resolve to this checkout. + if _VENV_BIN.is_dir(): + env["PATH"] = f"{_VENV_BIN}:{env.get('PATH', '')}" + return env + + +def _git(repo: Path, *args: str, check: bool = True) -> str: + proc = subprocess.run( + ["git", *args], + cwd=repo, + env=_git_env(repo), + capture_output=True, + text=True, + check=check, + ) + return proc.stdout.strip() + + +def _commit(repo: Path, name: str, content: str) -> str: + (repo / name).write_text(content) + _git(repo, "add", name) + _git(repo, "commit", "-m", f"add {name}") + return _git(repo, "rev-parse", "HEAD") + + +def _bare_remote(tmp: Path, name: str = "remote.git") -> Path: + remote = Path(tmp) / name + _git(Path(tmp), "init", "--bare", "-q", str(remote)) + return remote + + +def _seed_remote(remote: Path, repo: Path, ref: str = "refs/heads/main") -> str: + sha = _git(repo, "rev-parse", "HEAD") + _git(repo, "push", "-q", str(remote), f"HEAD:{ref}") + return sha + + +def _run_hook( + repo: Path, + remote: str, + reflines: str, + url: str | None = None, + extra_policy: Path | None = None, +) -> subprocess.CompletedProcess: + env = _git_env(repo) + env["CONTENT_GUARD_EXTRA_POLICY"] = str(extra_policy or repo / "no-private-policy.json") + argv = ["bash", str(_HOOK), remote] + if url is not None: + argv.append(url) + return subprocess.run( + argv, + cwd=repo, + env=env, + input=reflines, + capture_output=True, + text=True, + check=False, + ) + + +class PrePushHookFunctionalTests(unittest.TestCase): + def _init_repo(self, repo: Path) -> None: + _git(repo, "init", "-q") + _git(repo, "config", "user.name", "t") + _git(repo, "config", "user.email", "t@example.com") + + def test_new_branch_excludes_already_remote_ancestors(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + _commit(repo, "base.txt", "base\n") + leak = _commit(repo, "leak.txt", _LEAK_LINE) + _seed_remote(remote, repo, "refs/heads/main") + _git(repo, "checkout", "-q", "-b", "newbr", leak) + _git(repo, "rm", "-q", "leak.txt") + (repo / "new.txt").write_text(_CLEAN_LINE) + _git(repo, "add", "new.txt") + _git(repo, "commit", "-q", "-m", "drop leak, add clean") + new = _git(repo, "rev-parse", "HEAD") + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {new} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + ) + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + self.assertIn("scanning introduced content in 1 new commit(s)", proc.stdout) + self.assertNotIn(leak, proc.stdout) + + def test_new_branch_still_scans_genuinely_new_leak(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + _commit(repo, "base.txt", "base\n") + clean = _commit(repo, "clean.txt", _CLEAN_LINE) + _seed_remote(remote, repo, "refs/heads/main") + _git(repo, "checkout", "-q", "-b", "newbr", clean) + leak = _commit(repo, "newleak.txt", _LEAK_LINE) + _git(repo, "rm", "-q", "newleak.txt") + _git(repo, "commit", "-q", "-m", "remove new leak") + tip = _git(repo, "rev-parse", "HEAD") + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {tip} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + ) + self.assertEqual(proc.returncode, 1, msg=proc.stdout + proc.stderr) + self.assertIn("scanning introduced content in 2 new commit(s)", proc.stdout) + self.assertIn(leak[:12], proc.stdout) + + def test_new_branch_no_remote_refs_scans_all_ancestors(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + _commit(repo, "base.txt", "base\n") + _commit(repo, "leak.txt", _LEAK_LINE) + _git(repo, "rm", "-q", "leak.txt") + _git(repo, "commit", "-q", "-m", "remove leak") + tip = _git(repo, "rev-parse", "HEAD") + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {tip} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + ) + self.assertEqual(proc.returncode, 1, msg=proc.stdout + proc.stderr) + self.assertIn("scanning introduced content in 3 new commit(s)", proc.stdout) + + def test_existing_branch_range_unchanged(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + _commit(repo, "base.txt", "base\n") + rsha = _commit(repo, "remote.txt", _CLEAN_LINE) + _seed_remote(remote, repo, "refs/heads/main") + leak = _commit(repo, "newleak.txt", _LEAK_LINE) + proc = _run_hook( + repo, + "origin", + f"refs/heads/main {leak} refs/heads/main {rsha}\n", + url=str(remote), + ) + self.assertEqual(proc.returncode, 1, msg=proc.stdout + proc.stderr) + self.assertIn(f"{rsha}..{leak}", proc.stdout) + + def test_branch_deletion_is_skipped(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + _commit(repo, "base.txt", "base\n") + proc = _run_hook( + repo, + "origin", + f"refs/heads/main {_ZERO40} refs/heads/main {_ZERO40}\n", + url=str(remote), + ) + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + + def test_stale_local_tracking_ref_does_not_under_exclude(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + _commit(repo, "base.txt", "base\n") + leak = _commit(repo, "leak.txt", _LEAK_LINE) + _seed_remote(remote, repo, "refs/heads/main") + _git(repo, "checkout", "-q", "-b", "newbr", leak) + _git(repo, "rm", "-q", "leak.txt") + (repo / "new.txt").write_text(_CLEAN_LINE) + _git(repo, "add", "new.txt") + _git(repo, "commit", "-q", "-m", "drop leak, add clean") + new = _git(repo, "rev-parse", "HEAD") + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {new} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + ) + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + self.assertNotIn(leak, proc.stdout) + + def test_stale_local_tracking_ref_does_not_over_exclude(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + _commit(repo, "base.txt", "base\n") + leak = _commit(repo, "leak.txt", _LEAK_LINE) + _seed_remote(remote, repo, "refs/heads/main") + _git(repo, "update-ref", "refs/remotes/origin/main", leak) + _git(remote, "update-ref", "-d", "refs/heads/main") + _git(repo, "checkout", "-q", "-b", "newbr", leak) + _git(repo, "rm", "-q", "leak.txt") + tip = _commit(repo, "clean.txt", _CLEAN_LINE) + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {tip} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + ) + + self.assertEqual(proc.returncode, 1, msg=proc.stdout + proc.stderr) + self.assertIn("scanning introduced content in 3 new commit(s)", proc.stdout) + + def test_remote_enumeration_failure_fails_closed(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + self._init_repo(repo) + tip = _commit(repo, "base.txt", "base\n") + bad_url = str(repo / "no-such-remote.git") + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {tip} refs/heads/newbr {_ZERO40}\n", + url=bad_url, + ) + self.assertEqual(proc.returncode, 1, msg=proc.stdout + proc.stderr) + self.assertIn("failed to enumerate advertised refs", proc.stderr) + self.assertIn("not a leak verdict", proc.stderr) + self.assertNotIn("found violations", proc.stderr) + + def test_missing_url_fails_closed(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + self._init_repo(repo) + tip = _commit(repo, "base.txt", "base\n") + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {tip} refs/heads/newbr {_ZERO40}\n", + url=None, + ) + self.assertEqual(proc.returncode, 1, msg=proc.stdout + proc.stderr) + self.assertIn("cannot enumerate remote refs", proc.stderr) + self.assertIn("not a leak verdict", proc.stderr) + self.assertNotIn("found violations", proc.stderr) + + def test_extra_policy_is_used_for_tip_scan(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + tip = _commit(repo, "private.txt", "company codename oriole\n") + extra_policy = repo / "private-policy.json" + extra_policy.write_text( + json.dumps( + { + "name": "private-test", + "rules": {"private-codename": "block"}, + "custom_rules": [ + { + "id": "private-codename", + "category": "business", + "pattern": "codename oriole", + "replacement": "[redacted-codename]", + } + ], + } + ) + ) + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {tip} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + extra_policy=extra_policy, + ) + + self.assertEqual(proc.returncode, 1, msg=proc.stdout + proc.stderr) + self.assertIn("private-codename", proc.stdout) + + def test_remote_only_advertised_object_is_ignored(self) -> None: + with TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "repo" + other = root / "other" + repo.mkdir() + other.mkdir() + remote = _bare_remote(root) + self._init_repo(repo) + base = _commit(repo, "base.txt", "base\n") + _seed_remote(remote, repo, "refs/heads/main") + _git(repo, "checkout", "-q", "-b", "newbr", base) + tip = _commit(repo, "new.txt", _CLEAN_LINE) + self._init_repo(other) + _commit(other, "other.txt", "collaborator\n") + _seed_remote(remote, other, "refs/heads/collaborator") + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {tip} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + ) + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + self.assertIn("scanning introduced content in 1 new commit(s)", proc.stdout) + + def test_annotated_tag_excludes_peeled_commit_when_tag_object_is_not_local(self) -> None: + with TemporaryDirectory() as tmp: + root = Path(tmp) + seed = root / "seed" + repo = root / "repo" + seed.mkdir() + remote = _bare_remote(root) + self._init_repo(seed) + _commit(seed, "base.txt", "base\n") + _commit(seed, "leak.txt", _LEAK_LINE) + _git(seed, "push", "-q", str(remote), "HEAD:refs/heads/main") + _git(root, "clone", "-q", "--no-tags", "--branch", "main", str(remote), str(repo)) + _git(seed, "tag", "-a", "public", "-m", "public tag") + tag_sha = _git(seed, "rev-parse", "refs/tags/public") + _git(seed, "push", "-q", str(remote), "refs/tags/public") + self.assertNotEqual( + subprocess.run( + ["git", "cat-file", "-e", tag_sha], + cwd=repo, + env=_git_env(repo), + capture_output=True, + text=True, + check=False, + ).returncode, + 0, + ) + _git(seed, "push", "-q", str(remote), ":refs/heads/main") + _git(repo, "config", "user.name", "t") + _git(repo, "config", "user.email", "t@example.com") + _git(repo, "rm", "-q", "leak.txt") + tip = _commit(repo, "clean.txt", _CLEAN_LINE) + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {tip} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + ) + + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + self.assertIn("scanning introduced content in 1 new commit(s)", proc.stdout) + + def test_new_branch_batches_into_one_history_scan(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + _commit(repo, "base.txt", "base\n") + clean = _commit(repo, "remote.txt", _CLEAN_LINE) + _seed_remote(remote, repo, "refs/heads/main") + _git(repo, "checkout", "-q", "-b", "newbr", clean) + for i in range(3): + _commit(repo, f"new{i}.txt", _CLEAN_LINE) + tip = _git(repo, "rev-parse", "HEAD") + proc = _run_hook( + repo, + "origin", + f"refs/heads/newbr {tip} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + ) + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + self.assertEqual(proc.stdout.count("scanning introduced content in"), 1, msg=proc.stdout) + self.assertIn("scanning introduced content in 3 new commit(s)", proc.stdout) + + def test_anonymous_url_push_uses_advertised_refs(self) -> None: + with TemporaryDirectory() as tmp: + repo = Path(tmp) + remote = _bare_remote(tmp) + self._init_repo(repo) + _commit(repo, "base.txt", "base\n") + leak = _commit(repo, "leak.txt", _LEAK_LINE) + _seed_remote(remote, repo, "refs/heads/main") + _git(repo, "checkout", "-q", "-b", "newbr", leak) + _git(repo, "rm", "-q", "leak.txt") + (repo / "new.txt").write_text(_CLEAN_LINE) + _git(repo, "add", "new.txt") + _git(repo, "commit", "-q", "-m", "drop leak, add clean") + new = _git(repo, "rev-parse", "HEAD") + proc = _run_hook( + repo, + "", + f"refs/heads/newbr {new} refs/heads/newbr {_ZERO40}\n", + url=str(remote), + ) + self.assertEqual(proc.returncode, 0, msg=proc.stdout + proc.stderr) + self.assertNotIn(leak, proc.stdout) + + def test_scanner_error_categorized_not_as_leak(self) -> None: + # A scanner/plumbing failure (exit >1) must surface as "failed to run", + # not as "found violations" (issue #82). + text = _hook_text() + assert "failed to run" in text + assert "not a leak verdict" in text + assert "BLOCKED. content-guard found violations." in text + + +if __name__ == "__main__": + unittest.main()