hooks: PreToolUse Bash guard flags masked exit status (U-H3)
PreToolUse hook for the Bash tool. When the user writes a pipeline whose
left side changes state (git push, kubectl apply, curl -X PUT, ...) and
whose exit status is then consumed by && / || / an if-condition, the
hook exits 2 and surfaces an advisory naming the masking command and
suggesting ${PIPESTATUS[0]} or dropping the pipe.
Always exits 0 on parse failure, missing python3, empty input, malformed
JSON, or anything it does not recognise — a hook that breaks Bash is
worse than the bug it prevents.
Real-process tests in hooks/tests/ feed JSON on stdin as a subprocess and
include mutation cases (six targeted matcher breakages) to prove each
assertion can actually fail.
This commit is contained in:
140
hooks/bash-pipeline-guard.sh
Executable file
140
hooks/bash-pipeline-guard.sh
Executable file
@@ -0,0 +1,140 @@
|
||||
#!/usr/bin/env bash
|
||||
# PreToolUse hook for Bash — flags pipelines whose left side changes state but
|
||||
# whose exit status is then consumed by `&&`, `||`, or an `if`-condition.
|
||||
#
|
||||
# bug or subtlety: `git push -q origin main 2>&1 | tail -1 && echo "pushed"`
|
||||
# reads tail's exit status, not git's. A push that was rejected
|
||||
# as non-fast-forward still reports "pushed".
|
||||
#
|
||||
# Contract:
|
||||
# - exit 0 for anything it does not understand, and for any parse failure.
|
||||
# A hook that breaks Bash is far worse than the bug it prevents.
|
||||
# - exit 2 with a short stderr explanation when it sees the pattern.
|
||||
# Advisory only — never block.
|
||||
#
|
||||
# Reads tool_input.command from stdin JSON (same shape post-edit-lint.sh parses).
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
INPUT="$(cat 2>/dev/null || true)"
|
||||
[ -n "$INPUT" ] || exit 0
|
||||
command -v python3 >/dev/null 2>&1 || exit 0
|
||||
|
||||
# Inline the matcher as a python script in a temp file. Single-quoted heredoc
|
||||
# prevents shell expansion of $-characters in the python source.
|
||||
TMPPY="$(mktemp -t bpg.XXXXXX 2>/dev/null || mktemp -t bpg.XXXXXX)" || exit 0
|
||||
[ -n "$TMPPY" ] || exit 0
|
||||
trap 'rm -f "$TMPPY"' EXIT INT TERM HUP
|
||||
|
||||
cat >"$TMPPY" <<'PYTHON_EOF'
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
|
||||
raw = sys.stdin.read()
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
except Exception:
|
||||
# Parse failure → silent pass
|
||||
sys.exit(0)
|
||||
|
||||
if not isinstance(data, dict):
|
||||
sys.exit(0)
|
||||
|
||||
tool_input = data.get("tool_input")
|
||||
if not isinstance(tool_input, dict):
|
||||
sys.exit(0)
|
||||
|
||||
cmd = tool_input.get("command", "")
|
||||
if not isinstance(cmd, str) or not cmd:
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
def strip_quotes(s):
|
||||
# Strip double-quoted and single-quoted strings so we don't match a verb
|
||||
# that lives inside a quoted argument or echo payload.
|
||||
s = re.sub(r'"[^"]*"', "", s)
|
||||
s = re.sub(r"'[^']*'", "", s)
|
||||
return s
|
||||
|
||||
|
||||
# Author has handled the masked-status problem → no need to flag.
|
||||
if re.search(r"set\s+-o\s+pipefail", cmd):
|
||||
sys.exit(0)
|
||||
if re.search(r"\$\{?PIPESTATUS", cmd):
|
||||
sys.exit(0)
|
||||
|
||||
# Walk the command once and record single-pipe positions, skipping `||`.
|
||||
pipes = []
|
||||
i = 0
|
||||
while i < len(cmd):
|
||||
c = cmd[i]
|
||||
if c == "|" and i + 1 < len(cmd) and cmd[i + 1] == "|":
|
||||
i += 2
|
||||
continue
|
||||
if c == "|":
|
||||
pipes.append(i)
|
||||
i += 1
|
||||
|
||||
if not pipes:
|
||||
sys.exit(0)
|
||||
|
||||
# Left side = everything before the first pipe (the segment whose status we
|
||||
# care about).
|
||||
left = strip_quotes(cmd[: pipes[0]])
|
||||
|
||||
# State-changing verbs. Matched conservatively — these are the verbs the brief
|
||||
# enumerates, matched as `verb` (not as a prefix).
|
||||
state_verbs = [
|
||||
(r"\bgit\s+push\b", "git push"),
|
||||
(r"\bgit\s+commit\b", "git commit"),
|
||||
(r"\bgit\s+merge\b", "git merge"),
|
||||
(r"\bgit\s+rebase\b", "git rebase"),
|
||||
(r"\bkubectl\s+apply\b", "kubectl apply"),
|
||||
(r"\bkubectl\s+delete\b", "kubectl delete"),
|
||||
(r"\bkubectl\s+patch\b", "kubectl patch"),
|
||||
(r"\bdocker\s+push\b", "docker push"),
|
||||
(r"\bhelm\s+upgrade\b", "helm upgrade"),
|
||||
(r"\bhelm\s+install\b", "helm install"),
|
||||
]
|
||||
verb = None
|
||||
for pat, name in state_verbs:
|
||||
if re.search(pat, left):
|
||||
verb = name
|
||||
break
|
||||
|
||||
# curl is special: only state-changing when paired with a write method or a
|
||||
# data payload. Read-only `curl ... | jq` must not be flagged.
|
||||
if not verb and re.search(r"\bcurl\b", left):
|
||||
if re.search(r"-X\s*(POST|PUT|DELETE)", left, re.IGNORECASE) or re.search(
|
||||
r"--data", left
|
||||
):
|
||||
verb = "curl"
|
||||
|
||||
if not verb:
|
||||
sys.exit(0)
|
||||
|
||||
# Is the pipeline's status consumed by `&&`, `||`, or an `if`-test whose
|
||||
# `then` clause follows the pipeline?
|
||||
rest = cmd[pipes[-1] + 1 :]
|
||||
if re.search(r"&&|\|\|", rest):
|
||||
print(verb)
|
||||
sys.exit(0)
|
||||
if re.search(r"\bthen\b", rest):
|
||||
print(verb)
|
||||
sys.exit(0)
|
||||
|
||||
sys.exit(0)
|
||||
PYTHON_EOF
|
||||
|
||||
# Pipe the raw hook input through the python matcher; capture stdout (the verb
|
||||
# name to surface in the advisory). Any non-zero exit from python (e.g. a real
|
||||
# crash) is treated as a silent pass — same policy as a parse failure.
|
||||
RESULT="$(printf '%s' "$INPUT" | python3 "$TMPPY" 2>/dev/null)" || exit 0
|
||||
[ -n "$RESULT" ] || exit 0
|
||||
|
||||
{
|
||||
printf 'Pipeline masks %s exit status: &&/|| reads the LAST command in the pipe, not the first. Use ${PIPESTATUS[0]} or drop the pipe.\n' \
|
||||
"$RESULT"
|
||||
} >&2
|
||||
exit 2
|
||||
Reference in New Issue
Block a user