diff --git a/scripts/validate_git_command.py b/scripts/validate_git_command.py index 0e8dde9..b68bdf0 100755 --- a/scripts/validate_git_command.py +++ b/scripts/validate_git_command.py @@ -5,9 +5,14 @@ """ import json +import os import re import sys +# Enough of a body to count wrapped lines in; a cap so an accidentally huge +# file cannot stall the hook. +BODY_READ_LIMIT = 256 * 1024 + # Conventional commit pattern CONVENTIONAL_COMMIT_PATTERN = ( r"^(feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert)(\(.+\))?!?:\s.+" @@ -44,6 +49,171 @@ ] +# --------------------------------------------------------------------------- +# Gates. Unlike the advisory checks above these refuse the call, because each +# one describes an action that silently does the wrong thing rather than one +# that merely reads badly. +# --------------------------------------------------------------------------- + +# Bodies posted to a forge: gh pr/issue/release create|edit|comment. +FORGE_BODY = re.compile( + r"\bgh\s+(pr|issue|release)\s+(create|edit|comment)\b", re.IGNORECASE +) +BODY_FILE = re.compile(r"--(?:body|notes)-file[= ]+(\S+)") +BODY_INLINE = re.compile(r"--(?:body|notes)[= ]+(['\"])(.*?)\1", re.DOTALL) + +# Replying to a review comment needs the PR number in the path: +# repos/O/R/pulls/{pr}/comments/{id}/replies. Without it GitHub answers 404 and +# the reply is silently not posted. Two deliberate limits: only the /replies +# subresource is checked (`pulls/comments/{id}` is a legitimate read endpoint), +# and only a segment that actually invokes gh/curl counts — matching the path +# anywhere would block writing about it in an echo or a commit message. +REPLY_WITHOUT_PR = re.compile(r"/pulls/comments/[^/\s'\"]+/replies\b") +# The anchor matters: matching the path anywhere would block writing ABOUT it +# in an echo or a commit message. But anchoring on gh/curl alone let any +# prefix through -- `env FOO=1 gh api …` and `sudo gh api …` both slipped the +# gate -- so leading assignments and the usual wrapper words are skipped over +# first. Still anchored, so quoted prose stays unaffected. +INVOKES_FORGE_API = re.compile( + r"^\s*(?:(?:[A-Za-z_][A-Za-z0-9_]*=\S*|sudo|env|time|command|nohup|xargs)\s+)*" + r"(?:gh\s+api|curl)\b" +) + +POLL_LOOP = re.compile(r"\b(?:until|while)\b.*?\bsleep\b", re.DOTALL) +FOR_LOOP_POLL = re.compile(r"\bfor\b[^\n]*\bin\b[^\n]*\bseq\b.*?\bsleep\b", re.DOTALL) +POLLS_PR = re.compile( + r"\bgh\s+pr\s+(?:view|checks|status)\b" + r"|\bgh\s+api\b[^\n]*?/pulls/" + r"|\bpr-status\.sh\b" +) + + +def read_command(data) -> str: + """Pull the command out of a PreToolUse payload. + + Claude Code sends {"tool_name": ..., "tool_input": {"command": ...}}. An + earlier version read a top-level "command" key, which that payload does not + have, so the hook returned silently on every invocation and none of the + checks below ever ran. + """ + if not isinstance(data, dict): + return "" + tool_input = data.get("tool_input") + if isinstance(tool_input, dict) and tool_input.get("command"): + return tool_input["command"] + return data.get("command", "") or "" + + +def deny(reason: str) -> None: + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": reason, + } + } + ) + ) + + +def hard_wrapped(text: str) -> int: + """Count prose lines that look hard-wrapped at a fixed column. + + Only consecutive prose counts: a short line followed by more prose is the + signature of a fixed-width wrap. Tables, lists, quotes, headings, link + references and fenced code keep their own line structure and are skipped, + as is a lone short line (a real one-line paragraph). + """ + lines = text.split("\n") + fenced = False + hits = 0 + for i, ln in enumerate(lines): + s = ln.strip() + if s.startswith(("```", "~~~")): + fenced = not fenced + continue + if fenced or not s: + continue + if re.match(r"^([-*+>#|]|\d+[.)]|\[)", s) or "|" in s: + continue + nxt = lines[i + 1].strip() if i + 1 < len(lines) else "" + if not nxt or re.match(r"^([-*+>#|`]|\d+[.)]|\[)", nxt): + continue + # A prose line that stops in the 55-85 column band while the paragraph + # continues on the next line was wrapped by hand, not by the renderer. + if 55 <= len(ln.rstrip()) <= 85: + hits += 1 + return hits + + +def forge_body_hard_wrapped(cmd: str) -> str | None: + if not FORGE_BODY.search(cmd): + return None + bodies = [] + for m in BODY_FILE.finditer(cmd): + p = m.group(1).strip("'\"") + try: + # Regular files only, and only the first chunk. `--body-file` can + # name a pipe -- process substitution (`--body-file <(...)`) hands + # over /dev/fd/N -- and reading one here blocks until a writer this + # process cannot see appears. A hook that hangs is worse than one + # that misses a finding, so a non-regular path is skipped. + if not os.path.isfile(p): + continue + with open(p, encoding="utf-8") as fh: + bodies.append((p, fh.read(BODY_READ_LIMIT))) + except OSError: + pass + for m in BODY_INLINE.finditer(cmd): + bodies.append(("--body", m.group(2))) + for name, text in bodies: + n = hard_wrapped(text) + if n >= 3: + return ( + f"{name} carries {n} hard-wrapped prose lines. Bodies posted to " + "GitHub/GitLab/Jira must NOT be wrapped at a fixed column: write " + "each paragraph as ONE long line and let the renderer reflow it. " + "Hard breaks read ragged in the web UI, break on mobile, and " + "corrupt every later quote or diff — and in release notes they " + "survive verbatim, unlike a CHANGELOG where markdown reflows. " + "Tables, lists and fenced code keep their own line structure. " + "(Commit messages are the exception and stay wrapped at ~72.)" + ) + return None + + +def reply_path_without_pr(cmd: str) -> str | None: + for segment in re.split(r"(?:\|\||&&|[;|&\n])", cmd): + if INVOKES_FORGE_API.match(segment) and REPLY_WITHOUT_PR.search(segment): + return ( + "A review-comment reply needs the PR number in the path — this " + "one would 404 and post nothing:\n\n" + " repos/{owner}/{repo}/pulls/{pr}/comments/{comment_id}/replies\n\n" + "`repos/{owner}/{repo}/pulls/comments/{comment_id}` (without " + "/replies) is the valid form for READING one comment, which is " + "where the shorter path comes from." + ) + return None + + +def handrolled_pr_poll(cmd: str) -> str | None: + if "--watch" in cmd or not POLLS_PR.search(cmd): + return None + if not (POLL_LOOP.search(cmd) or FOR_LOOP_POLL.search(cmd)): + return None + return ( + "Hand-rolled poll over pull-request state. Use " + "`pr-status.sh -R --watch` instead: it returns at the " + "FIRST actionable event — a check that failed, a review that arrived, a " + "thread that needs an answer — where a loop written here waits for the " + "one outcome it was told about and sleeps through the rest. A loop that " + "exited only on `merge` slept through the review it was waiting for, and " + "the operator had to ask what was happening." + ) + + def check_conventional_commit(message: str) -> str | None: """Validate commit message follows conventional commits.""" if not re.match(CONVENTIONAL_COMMIT_PATTERN, message): @@ -147,11 +317,23 @@ def main(): try: data = json.loads(input_data) - command = data.get("command", "") + command = read_command(data) except (json.JSONDecodeError, TypeError): command = input_data - if not command or "git" not in command.lower(): + if not command: + return + + # Gates that refuse the call outright. Checked before the advisory + # warnings because a denied command never runs, so warning about its + # style would be noise. + for gate in (forge_body_hard_wrapped, reply_path_without_pr, handrolled_pr_poll): + reason = gate(command) + if reason: + deny(reason) + return + + if "git" not in command.lower(): return warnings = check_command(command) diff --git a/tests/test_validate_git_command.py b/tests/test_validate_git_command.py new file mode 100755 index 0000000..7ce1878 --- /dev/null +++ b/tests/test_validate_git_command.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""Cases for scripts/validate_git_command.py. + +Run: python3 tests/test_validate_git_command.py + +The gates deny commands, so a regression here silently either blocks +legitimate work or stops catching the thing it was written for. Each case +names the failure it stands for. +""" + +import json +import os +import subprocess +import sys +import tempfile + +HOOK = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "scripts", + "validate_git_command.py", +) + +WRAPPED = ( + "A paragraph wrapped by hand at roughly the seventy-two column mark,\n" + "which is the shape this gate exists to catch before it is posted and\n" + "the breaks survive verbatim in the rendered release notes forever.\n" + "A fourth line so the run is unambiguously a wrapped paragraph." +) + +CASES = [ + # (name, expected, command) + # The bug that made every check below unreachable: the payload is nested. + ("nested payload reaches the checks", "REMINDER", 'git commit -m "stuff"'), + ("conventional message stays quiet", "PASS", 'git commit -m "fix: handle null"'), + ( + "reply path without the PR number", + "DENY", + "gh api repos/o/r/pulls/comments/123/replies -f body=x", + ), + # Legitimate read endpoint - same prefix, no /replies. + ( + "reading one comment is allowed", + "PASS", + "gh api repos/o/r/pulls/comments/123", + ), + # Writing ABOUT the path must not be blocked, only invoking it. + ( + "the path inside an echo is not a call", + "PASS", + "echo 'use repos/o/r/pulls/comments/1/replies'", + ), + # A prefix must not become a bypass: both of these slipped the anchor. + ( + "env assignment before the call", + "DENY", + "env FOO=1 gh api repos/o/r/pulls/comments/1/replies -f body=x", + ), + ( + "sudo before the call", + "DENY", + "sudo gh api repos/o/r/pulls/comments/1/replies -f body=x", + ), + ( + "bare assignment before the call", + "DENY", + "GH_TOKEN=x gh api repos/o/r/pulls/comments/1/replies -f body=x", + ), + # Second segment of a chain still counts. + ( + "after && still counts", + "DENY", + "git status && gh api repos/o/r/pulls/comments/1/replies -f body=x", + ), + ( + "sleep-loop over PR state", + "DENY", + "until [ x = y ]; do gh pr view 1 --json state; sleep 30; done", + ), + ( + "pr-status.sh --watch is the fix, not the fault", + "PASS", + "pr-status.sh -R o/r 1 --watch", + ), + ( + "a single gh pr view is not a poll", + "PASS", + "gh pr view 1 --repo o/r --json state", + ), + ("hard-wrapped inline body", "DENY", f'gh pr create --title x --body "{WRAPPED}"'), + ( + "one long line is what we want", + "PASS", + 'gh pr create --title x --body "One long line that the renderer reflows by itself."', + ), + ("unrelated command", "PASS", "ls -la"), +] + + +def run(command: str) -> str: + payload = json.dumps({"tool_name": "Bash", "tool_input": {"command": command}}) + # check=False: a non-zero exit is itself something the cases assert on, + # not a reason to abort the run. + proc = subprocess.run( + [sys.executable, HOOK], + input=payload, + capture_output=True, + text=True, + timeout=10, + check=False, + ) + out = proc.stdout.strip() + if not out: + return "PASS" + if out.startswith("{"): + decision = json.loads(out)["hookSpecificOutput"]["permissionDecision"] + return "DENY" if decision == "deny" else "PASS" + return "REMINDER" + + +def fifo_case() -> bool: + """A --body-file naming a pipe must not block the hook. + + `gh pr create --body-file <(generate)` hands over /dev/fd/N. Opening it + here waits for a writer this process cannot see, and the hook hangs. + """ + tmp = tempfile.mkdtemp() + path = os.path.join(tmp, "fifo") + os.mkfifo(path) + try: + run(f"gh pr create --title x --body-file {path}") + return True + except subprocess.TimeoutExpired: + return False + finally: + os.unlink(path) + os.rmdir(tmp) + + +def main() -> int: + fails = 0 + for name, expected, command in CASES: + got = run(command) + ok = got == expected + fails += 0 if ok else 1 + print(f" {'OK ' if ok else 'FAIL'} {name:<44} want={expected:<9} got={got}") + + ok = fifo_case() + fails += 0 if ok else 1 + print( + f" {'OK ' if ok else 'FAIL'} {'--body-file on a pipe returns':<44} want=no-hang " + f"got={'no-hang' if ok else 'HUNG'}" + ) + + print(f" ---- failures: {fails}") + return 1 if fails else 0 + + +if __name__ == "__main__": + raise SystemExit(main())