fix(hooks): read the PreToolUse payload, add three forge gates

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 <owner/repo> <pr> --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 @@
 
     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)

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())