timothyandClaude Opus 5
183df2bb3d
Round 2, from two independent cold reviews. The headline defect was found by
the test the reviews prompted rather than by the reviews themselves:
exec 0<"$sin" 2>/dev/null
reads as "suppress errors from this redirection". It is not. `exec` with
redirections and no command applies them to the shell PERMANENTLY, so this sent
each hook's ENTIRE stderr to /dev/null. Every husky guard's user-facing output
is stderr — the H6 "push to main BLOCKED" message, the BOM guard's remediation
text, the missing-trailer message — so the guards went on blocking correctly
while telling the human nothing about why. Invisible to the original A/B, which
compared only stdout and exit status.
The finding that exposed it was ranked Low ("the A/B ignores stderr").
Also fixed, each with an executed proof:
- The fd restore sat inside the same conditional as the replay, and the replay
read the PATH, so an unlinked-but-open scratch file meant a swallowed `deny`
at exit 0. Restore is now unconditional and the replay reads through a held
fd, which survives unlink.
- A SIGTERMed hook lost output it had already printed (52 bytes -> 0). Bash runs
no EXIT trap for an untrapped fatal signal; INT/TERM/HUP are trapped now.
- The classifier misreported an uppercase decision as prose, read the bare word
additionalContext in free text as a decision, and let a non-zero exit ERASE a
decision the hook really printed.
- The session id reached the log FILENAME unscrubbed. The first fix reused the
record scrubber, which deliberately keeps `/` and `.`, so `../../escaped` went
through untouched: passing a value through a scrub is not the same as passing
it through the right scrub.
- Preamble line 1 could abort a `set -e` hook, turning a fail-open guard
fail-closed; and a missing sink printed `command not found` to stderr.
The big one is test quality. `test_instrumentation_changes_NOTHING_the_harness_
can_see` was 160/165 empty-vs-empty and 165/165 exit-0-vs-exit-0, so deleting
the whole replay left it green for four hooks, two of which issue `deny`. Three
places claimed the matrix reached every deciding branch, including the decision
record. It now constructs the missing cases — a foreign worktree marker, a BOM
`.cs` in an *ersatztv*-named repo, stubbed memory_pressure, a real bare remote
for the three pre-push guards, a stub Gitea for prepush-donewhen — and
test_the_AB_is_not_VACUOUS_for_any_hook fails unless each hook has a case where
the control side actually says something.
Running pytest also wrote 115 synthetic fires into the production log, so the
report described the test suite while looking like a record of real sessions.
scripts/tests/conftest.py isolates every test; verified live, a full suite run
now leaves the real log untouched.
Refs: #776
Decisions-Edit: yes
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1103 lines
52 KiB
Python
1103 lines
52 KiB
Python
"""Every hook reports that it fired, and the reporting changes nothing else (ersatztv#776).
|
|
|
|
Two separable claims, and conflating them is how instrumentation ships as a regression:
|
|
|
|
1. COVERAGE — every hook script records its own execution. The population is DERIVED from
|
|
`.claude/hooks/*.sh`, never listed, per `testing.guard-derives-population-from-source`, so a
|
|
hook added tomorrow is uninstrumented-and-red rather than silently unobserved.
|
|
|
|
2. TRANSPARENCY — the wrapper is invisible to the harness. It slurps stdin and replays it, and it
|
|
diverts stdout and replays it, which puts it directly in the path of the most load-bearing
|
|
guards in this repo. If it drops a byte of a `deny` payload or swallows an exit code, it
|
|
disables a guard while reporting that the guard fired. That is strictly worse than the
|
|
blindness #776 set out to fix, so the differential test below drives EVERY hook, with and
|
|
without the instrumentation, over a payload matrix and demands byte-equal stdout and equal
|
|
exit status.
|
|
|
|
The mutation proof for claim 1 is the contrapositive form allowed by
|
|
`testing.guard-ships-with-mutation-proof` for a checker-guard: the defect the guard exists to catch
|
|
(an uninstrumented hook) is introduced into an isolated copy of the guarded artifact, and the named
|
|
check must go red. See `test_a_hook_that_LOSES_its_instrumentation_is_DETECTED`.
|
|
|
|
Claim 2's proof is not a mutation of an assertion but a real A/B over the real scripts, which is the
|
|
only thing that can establish it — a unit test of the wrapper in isolation would pass on a wrapper
|
|
that is fine alone and lethal in context.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import pty
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
HOOKS_DIR = REPO_ROOT / ".claude" / "hooks"
|
|
SINK = REPO_ROOT / "scripts" / "hook-fire-log.sh"
|
|
|
|
# The three lines a hook must carry. Matched structurally rather than as one frozen blob so that a
|
|
# comment reflow does not redden the suite, but the two things that matter — sourcing the single
|
|
# shared sink, and calling begin with this hook's own name — are both pinned.
|
|
_SOURCES_SINK = re.compile(r'^\[ -r "\$ETV_HOOK_FIRE_LIB" \] && \. "\$ETV_HOOK_FIRE_LIB" \|\| true$', re.M)
|
|
_BEGINS = re.compile(r"^etv_hook_fire_begin (\S+) .*\|\| true$", re.M)
|
|
|
|
|
|
def hook_scripts() -> list[Path]:
|
|
"""THE POPULATION, from the filesystem. Never a list."""
|
|
return sorted(HOOKS_DIR.glob("*.sh"))
|
|
|
|
|
|
def expected_mode(name: str) -> str:
|
|
"""DERIVED from the wiring, not listed.
|
|
|
|
A hook registered in `.claude/settings.json` is a Claude hook: it always exits 0 and decides by
|
|
PRINTING JSON, so its stdout must be captured or the decision is unobservable. A hook registered
|
|
in `.husky/*` is a git hook: it decides by exit status and its stdout is progress text a human
|
|
watches live, so capturing it would hold the output back until the end.
|
|
|
|
Reading this from the settings files rather than a hardcoded set means a hook that changes
|
|
wiring gets the right expectation automatically — and, unlike a list, it cannot be quietly
|
|
edited to match a mistake.
|
|
"""
|
|
if name in (REPO_ROOT / ".claude" / "settings.json").read_text():
|
|
return "capture"
|
|
for husky in (REPO_ROOT / ".husky").iterdir():
|
|
if husky.is_file() and name in husky.read_text():
|
|
return "stream"
|
|
return "capture"
|
|
|
|
|
|
def instrumentation_faults(text: str, name: str) -> list[str]:
|
|
"""The check itself, factored out so a mutation can be fed to the same code the suite runs."""
|
|
faults = []
|
|
if not _SOURCES_SINK.search(text):
|
|
faults.append(f"{name}: does not source scripts/hook-fire-log.sh")
|
|
|
|
# The sink PATH, not just the sourcing line. Repointing `ETV_HOOK_FIRE_LIB` at /dev/null leaves
|
|
# the `[ -r ] && .` line untouched and disables all reporting silently — a mutation the first
|
|
# version of this checker passed clean.
|
|
lib = re.search(r"^ETV_HOOK_FIRE_LIB=(.*)$", text, re.M)
|
|
if not lib:
|
|
faults.append(f"{name}: no ETV_HOOK_FIRE_LIB assignment")
|
|
elif "/scripts/hook-fire-log.sh" not in lib.group(1):
|
|
faults.append(f"{name}: ETV_HOOK_FIRE_LIB does not point at the shared sink: {lib.group(1)}")
|
|
|
|
# A hook that disables its own reporting reads as instrumented and reports nothing.
|
|
if re.search(r"^\s*(export\s+)?ETV_HOOK_FIRE_DISABLE=1", text, re.M):
|
|
faults.append(f"{name}: sets ETV_HOOK_FIRE_DISABLE=1, so it never reports")
|
|
|
|
m = _BEGINS.search(text)
|
|
if not m:
|
|
faults.append(f"{name}: never calls etv_hook_fire_begin")
|
|
elif m.group(1) != name:
|
|
faults.append(f"{name}: calls etv_hook_fire_begin for {m.group(1)!r}, not for itself")
|
|
else:
|
|
# MODE. Flipping a Claude hook from `capture` to `stream` stops its stdout being read, so
|
|
# every decision it makes is recorded as `pass`/`blocked` from the exit status it never
|
|
# uses — the log stays full and becomes wrong. Nothing else in the suite notices.
|
|
mode = re.search(rf"^etv_hook_fire_begin {re.escape(name)} \S+ (\S+)", text, re.M)
|
|
want = expected_mode(name)
|
|
if not mode:
|
|
faults.append(f"{name}: etv_hook_fire_begin names no stdout mode")
|
|
elif mode.group(1) != want:
|
|
faults.append(
|
|
f"{name}: begins in {mode.group(1)!r} mode but its wiring implies {want!r}"
|
|
)
|
|
|
|
# Order is the whole point: 8 of the hooks slurp stdin with `input=$(cat)`, and a begin
|
|
# placed after that read would find the pipe already drained — recording a fire with no
|
|
# tool name, and replaying nothing into a hook that has already consumed its payload.
|
|
stdin_read = text.find("$(cat)")
|
|
if 0 <= stdin_read < m.start():
|
|
faults.append(f"{name}: reads stdin before etv_hook_fire_begin")
|
|
return faults
|
|
|
|
|
|
def strip_instrumentation(text: str) -> str:
|
|
"""Reconstruct the pre-#776 script. Used BOTH as the mutation and as the A/B control.
|
|
|
|
The blank line that separates the preamble from the code above it is dropped too. Leaving it
|
|
made the reconstruction differ from the original by one blank line in all 13 hooks — harmless
|
|
behaviourally, but the A/B control is only trustworthy insofar as it IS the original, and
|
|
"differs only in ways I judged harmless" is a claim, not a property.
|
|
`test_the_stripper_reproduces_the_ORIGINAL_byte_for_byte` turns it into a property.
|
|
"""
|
|
out, skipping = [], False
|
|
for line in text.splitlines(keepends=True):
|
|
if line.startswith("# ersatztv#776 —"):
|
|
skipping = True
|
|
if out and out[-1].strip() == "":
|
|
out.pop()
|
|
continue
|
|
if skipping:
|
|
if line.startswith(("#", "ETV_HOOK_FIRE_LIB=", "[ -r ", "type etv_hook_fire_begin",
|
|
"etv_hook_fire_begin ")):
|
|
continue
|
|
skipping = False
|
|
out.append(line)
|
|
return "".join(out)
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------
|
|
# ANTI-VACUITY FIRST — a glob that stopped matching would make every assertion below iterate over
|
|
# nothing and report full coverage. That is the characteristic failure of a completeness check, and
|
|
# this repo has shipped it (#631, #751), so it is asserted before anything depends on it.
|
|
# ------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def test_the_population_is_not_empty():
|
|
hooks = hook_scripts()
|
|
assert len(hooks) >= 10, (
|
|
f"only found {len(hooks)} hook scripts under {HOOKS_DIR} — the glob has stopped matching, "
|
|
"so every coverage assertion in this file is vacuous."
|
|
)
|
|
assert SINK.exists(), "the shared sink is missing; the instrumentation cannot work"
|
|
assert os.access(SINK, os.X_OK), "the sink is not executable, so its report side cannot be run"
|
|
|
|
|
|
def test_the_stripper_reproduces_the_ORIGINAL_byte_for_byte():
|
|
"""The A/B control must BE the pre-#776 script, not merely resemble it.
|
|
|
|
`test_the_stripper_actually_strips` only asserts the preamble is gone. That leaves the property
|
|
the differential test actually depends on — that the control side is the real original —
|
|
unpinned, so a future edit to the preamble format could make the control silently diverge in a
|
|
way that matters while every assertion stayed green. Compared against git rather than against a
|
|
second copy in this file, because a second copy is the thing that drifts.
|
|
"""
|
|
base = subprocess.run(
|
|
["git", "merge-base", "HEAD", "origin/main"],
|
|
capture_output=True, cwd=str(REPO_ROOT), text=True,
|
|
).stdout.strip()
|
|
if not base:
|
|
pytest.skip("no origin/main to compare against")
|
|
|
|
for hook in hook_scripts():
|
|
rel = hook.relative_to(REPO_ROOT)
|
|
orig = subprocess.run(
|
|
["git", "show", f"{base}:{rel}"],
|
|
capture_output=True, cwd=str(REPO_ROOT),
|
|
)
|
|
if orig.returncode != 0:
|
|
continue # a hook added after the merge base has no pre-#776 form
|
|
assert strip_instrumentation(hook.read_text()) == orig.stdout.decode(), (
|
|
f"{hook.name}: strip_instrumentation() does not reproduce the original. The A/B "
|
|
"control is therefore not the pre-#776 script, and the differential test is comparing "
|
|
"against something nobody wrote."
|
|
)
|
|
|
|
|
|
def test_the_stripper_actually_strips():
|
|
"""The A/B control and the mutation are the same function. If it were a no-op, the differential
|
|
test would compare each hook against itself and pass on a wrapper that breaks everything."""
|
|
for hook in hook_scripts():
|
|
text = hook.read_text()
|
|
stripped = strip_instrumentation(text)
|
|
assert stripped != text, f"{hook.name}: strip_instrumentation() removed nothing"
|
|
assert "etv_hook_fire_begin" not in stripped, f"{hook.name}: begin survived the strip"
|
|
assert not instrumentation_faults(text, hook.stem), f"{hook.name} is not instrumented"
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------
|
|
# CLAIM 1 — COVERAGE
|
|
# ------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def test_every_hook_reports_that_it_fired():
|
|
faults = []
|
|
for hook in hook_scripts():
|
|
faults += instrumentation_faults(hook.read_text(), hook.stem)
|
|
assert not faults, (
|
|
"these hooks do not report their own execution:\n " + "\n ".join(faults) +
|
|
"\n\nAdd the three-line preamble after the `set -` line and BEFORE any stdin read. A hook "
|
|
"that does not report is one whose firing we can only infer, which is ersatztv#776."
|
|
)
|
|
|
|
|
|
def test_a_hook_that_LOSES_its_instrumentation_is_DETECTED():
|
|
"""THE MUTATION PROOF (contrapositive form, for a checker-guard).
|
|
|
|
Introduce the defect this guard exists to catch — an uninstrumented hook — into an isolated copy
|
|
of each hook in turn, and the check must report it. Every hook, not a sample: the interesting
|
|
loss is whichever file someone actually edits.
|
|
"""
|
|
for hook in hook_scripts():
|
|
mutated = strip_instrumentation(hook.read_text())
|
|
faults = instrumentation_faults(mutated, hook.stem)
|
|
assert faults, (
|
|
f"stripping the instrumentation from {hook.name} left the check GREEN. The check is "
|
|
"not load-bearing — it would not notice a hook silently losing its reporting."
|
|
)
|
|
|
|
|
|
def test_begin_placed_AFTER_the_stdin_read_is_DETECTED():
|
|
"""The subtler mutation: present but too late. Ordering is the property that makes it work."""
|
|
late = 'set -euo pipefail\ninput=$(cat)\nETV_HOOK_FIRE_LIB="x"\n[ -r "$ETV_HOOK_FIRE_LIB" ] && . "$ETV_HOOK_FIRE_LIB" || true\netv_hook_fire_begin demo "" capture || true\n'
|
|
faults = instrumentation_faults(late, "demo")
|
|
assert any("reads stdin before" in f for f in faults), (
|
|
"a begin call placed after `input=$(cat)` was accepted. It would record a fire with no "
|
|
f"payload and replay nothing. faults={faults}"
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------
|
|
# CLAIM 2 — TRANSPARENCY. The wrapper must be invisible to the harness.
|
|
# ------------------------------------------------------------------------------------------------
|
|
|
|
# A BROAD, CHEAP matrix. It reaches the deciding branch of only a few hooks; the ones it cannot
|
|
# reach with a bare payload get a constructed positive case in `positive_cases()` below, and
|
|
# `test_the_AB_is_not_VACUOUS_for_any_hook` refuses to let any hook rely on this matrix alone.
|
|
#
|
|
# Stated because the first version of this file claimed the opposite in three places: this matrix
|
|
# alone produced empty-vs-empty comparisons for 8 of 13 hooks and a zero exit status for all 13, so
|
|
# deleting the entire stdout replay left the differential test green for four hooks — two of which
|
|
# issue `deny`. An A/B over silent allows proves transparency on the one path where there is nothing
|
|
# to be transparent about.
|
|
PAYLOADS = {
|
|
"bash-deny": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"ETV_UPDATE_GOLDENS=1 dotnet test"}}',
|
|
"bash-allow": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"ls -la"}}',
|
|
"git-commit": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"git commit -m x"}}',
|
|
"nav-deny": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"mcp__x__navigate","cwd":"%(cwd)s","tool_input":{"url":"http://h/iptv/channels.m3u"}}',
|
|
"agent-no-model": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Agent","cwd":"%(cwd)s","tool_input":{"description":"d","prompt":"p"}}',
|
|
"agent-model": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Agent","cwd":"%(cwd)s","tool_input":{"description":"d","prompt":"p","model":"sonnet"}}',
|
|
"merge": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"mcp__gitea__pull_request_write","cwd":"%(cwd)s","tool_input":{"method":"merge","owner":"timothy","repo":"ersatztv","pull_number":1}}',
|
|
"worktree-add": '{"session_id":"s","hook_event_name":"PostToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"git worktree add /tmp/nope-%(nonce)s HEAD"}}',
|
|
"ui-edit": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Edit","cwd":"%(cwd)s","tool_input":{"file_path":"web/src/screens/Channels.tsx"}}',
|
|
"empty": "",
|
|
"garbage": "not json at all",
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def sandbox(tmp_path_factory):
|
|
"""A controlled environment so the A/B pair cannot differ for reasons unrelated to the wrapper.
|
|
|
|
`memory_pressure` is stubbed because `pretooluse-agent-ram.sh` reads live free RAM: a real
|
|
reading that crossed the 10% or 20% threshold between the two runs would make the A and B halves
|
|
disagree for a reason that has nothing to do with instrumentation, and a differential test that
|
|
flakes is one that gets rerun until green.
|
|
"""
|
|
root = tmp_path_factory.mktemp("hookfire")
|
|
bindir = root / "bin"
|
|
bindir.mkdir()
|
|
stub = bindir / "memory_pressure"
|
|
stub.write_text("#!/bin/sh\necho 'System-wide memory free percentage: 42%'\n")
|
|
stub.chmod(0o755)
|
|
|
|
env = dict(os.environ)
|
|
env["PATH"] = f"{bindir}:{env['PATH']}"
|
|
env["ETV_HOOK_FIRE_LOG_DIR"] = str(root / "log")
|
|
env["CLAUDE_PROJECT_DIR"] = str(REPO_ROOT)
|
|
# No creds: `pretooluse-merge-consent.sh` must take its documented "cannot derive state" path
|
|
# rather than making live Gitea calls from a unit test.
|
|
for k in ("ETV_GITEA_TOKEN", "ETV_GITEA_BASICAUTH"):
|
|
env.pop(k, None)
|
|
return root, env
|
|
|
|
|
|
def _run(script: Path, payload: str, env: dict, cwd: Path, arg: str | None = None):
|
|
cmd = ["bash", str(script)] + ([arg] if arg else [])
|
|
p = subprocess.run(
|
|
cmd, input=payload.encode(), capture_output=True, cwd=str(cwd), env=env, timeout=60
|
|
)
|
|
return p.returncode, p.stdout
|
|
|
|
|
|
def _git(cwd: Path, *args: str) -> None:
|
|
subprocess.run(
|
|
["git", *args], cwd=str(cwd), check=True, capture_output=True,
|
|
env={**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@e",
|
|
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@e"},
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def positives(sandbox, tmp_path_factory):
|
|
"""CONSTRUCTED cases that drive hooks the bare payload matrix cannot reach to a DECIDING branch.
|
|
|
|
Each entry is (case-name, payload, arg, env, cwd). Building real git repositories is the point:
|
|
the hooks these cover — the BOM guard, the worktree-ownership guard, the three pre-push guards —
|
|
all decide by inspecting a repository, so a payload alone can only ever reach their early exit.
|
|
"""
|
|
_root, base_env = sandbox
|
|
w = tmp_path_factory.mktemp("positives")
|
|
cases: dict[str, list[tuple]] = {}
|
|
|
|
def pay(**kw) -> str:
|
|
import json as _json
|
|
return _json.dumps({"session_id": "me", "hook_event_name": "PreToolUse",
|
|
"tool_name": "Bash", **kw})
|
|
|
|
# --- worktree-guard: a marker naming ANOTHER session, on a `git commit` -------------------
|
|
wt = w / "sibling"
|
|
wt.mkdir()
|
|
_git(wt, "init", "-q", ".")
|
|
(wt / ".claude-worktree-owner").write_text("SOME-OTHER-SESSION\n")
|
|
cases["pretooluse-worktree-guard"] = [(
|
|
"foreign-marker-deny",
|
|
pay(cwd=str(wt), tool_input={"command": "git commit -m x"}), None, base_env, wt,
|
|
)]
|
|
|
|
# --- bom-guard: repo path must match `*ersatztv*`, with a staged BOM-carrying .cs ----------
|
|
br = w / "ersatztv-scratch"
|
|
br.mkdir()
|
|
_git(br, "init", "-q", ".")
|
|
(br / "Bad.cs").write_bytes(b"\xef\xbb\xbfclass A {}\n")
|
|
_git(br, "add", "Bad.cs")
|
|
cases["pretooluse-bom-guard"] = [(
|
|
"staged-bom-deny",
|
|
pay(cwd=str(br), tool_input={"command": "git commit -m x"}), None, base_env, br,
|
|
)]
|
|
|
|
# --- agent-ram: both thresholds, via a stubbed `memory_pressure` ---------------------------
|
|
ram_cases = []
|
|
for pct, label in ((5, "deny"), (15, "ask")):
|
|
bindir = w / f"bin{pct}"
|
|
bindir.mkdir()
|
|
stub = bindir / "memory_pressure"
|
|
stub.write_text(f"#!/bin/sh\necho 'System-wide memory free percentage: {pct}%'\n")
|
|
stub.chmod(0o755)
|
|
env = dict(base_env)
|
|
env["PATH"] = f"{bindir}:{os.environ['PATH']}"
|
|
ram_cases.append((
|
|
f"free-{pct}pct-{label}",
|
|
pay(tool_name="Agent", cwd=str(w), tool_input={"prompt": "p", "model": "sonnet"}),
|
|
None, env, w,
|
|
))
|
|
cases["pretooluse-agent-ram"] = ram_cases
|
|
|
|
# --- decisions-guard: a validator that prints and blocks, and one that prints and passes ---
|
|
dg = w / "dg"
|
|
(dg / "scripts").mkdir(parents=True)
|
|
_git(dg, "init", "-q", ".")
|
|
dg_cases = []
|
|
for rc, label in ((1, "block"), (0, "pass")):
|
|
d = w / f"dg{rc}"
|
|
(d / "scripts").mkdir(parents=True)
|
|
_git(d, "init", "-q", ".")
|
|
(d / "scripts" / "decisions_validate.py").write_text(
|
|
f"import sys\nprint('VALIDATOR SAID SOMETHING')\nsys.exit({rc})\n"
|
|
)
|
|
dg_cases.append((f"validator-{label}", "", None, base_env, d))
|
|
cases["decisions-guard"] = dg_cases
|
|
|
|
# --- pre-push guards: a real remote, so `git fetch origin main` resolves -------------------
|
|
bare = w / "remote.git"
|
|
_git(w, "init", "-q", "--bare", "--initial-branch=main", str(bare))
|
|
|
|
def clone(name: str) -> Path:
|
|
p = w / name
|
|
subprocess.run(["git", "clone", "-q", str(bare), str(p)], check=True, capture_output=True)
|
|
_git(p, "config", "user.email", "t@e")
|
|
_git(p, "config", "user.name", "t")
|
|
return p
|
|
|
|
seed = clone("seed")
|
|
(seed / "file.txt").write_text("one\n")
|
|
_git(seed, "add", "-A")
|
|
_git(seed, "commit", "-qm", "seed")
|
|
_git(seed, "push", "-q", "origin", "main")
|
|
|
|
# rebase-check: local HEAD is an ANCESTOR of a since-advanced origin/main -> blocks.
|
|
behind = clone("behind")
|
|
_git(behind, "checkout", "-qb", "feat")
|
|
ahead = clone("ahead")
|
|
(ahead / "file.txt").write_text("two\n")
|
|
_git(ahead, "add", "-A")
|
|
_git(ahead, "commit", "-qm", "advance main")
|
|
_git(ahead, "push", "-q", "origin", "main")
|
|
head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(behind),
|
|
capture_output=True, text=True).stdout.strip()
|
|
cases["prepush-rebase-check"] = [(
|
|
"behind-origin-main-blocks",
|
|
f"refs/heads/feat {head} refs/heads/feat {'0' * 40}\n", None, base_env, behind,
|
|
)]
|
|
|
|
# clean-worktree-check: a file both MODIFIED in the tree and present in the pushed set.
|
|
dirty = clone("dirty")
|
|
(dirty / "file.txt").write_text("committed change\n")
|
|
_git(dirty, "add", "-A")
|
|
_git(dirty, "commit", "-qm", "change file")
|
|
(dirty / "file.txt").write_text("uncommitted change\n")
|
|
dhead = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(dirty),
|
|
capture_output=True, text=True).stdout.strip()
|
|
cases["prepush-clean-worktree-check"] = [(
|
|
"dirty-file-in-pushed-set-blocks",
|
|
f"refs/heads/main {dhead} refs/heads/main {'0' * 40}\n", None, base_env, dirty,
|
|
)]
|
|
|
|
|
|
# --- prepush-donewhen: a push to main closing an issue with an unticked Done-when box -------
|
|
#
|
|
# This one needs a Gitea, so it gets a stub HTTP server rather than an exemption. Without it the
|
|
# hook's only reachable path in a test is its no-creds fail-open, and the guard that blocks
|
|
# pushes to `main` would be the one hook whose transparency nothing checks — precisely the
|
|
# "covered everywhere except where it matters" shape this repo keeps paying for.
|
|
import http.server
|
|
import json as _json
|
|
import threading
|
|
|
|
body = _json.dumps({"body": "## Done-when\n\n- [ ] not finished\n- [x] finished\n"}).encode()
|
|
|
|
class _Stub(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self): # noqa: N802
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *a): # keep pytest output clean
|
|
pass
|
|
|
|
srv = http.server.HTTPServer(("127.0.0.1", 0), _Stub)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
|
|
dw = clone("donewhen")
|
|
(dw / "src.cs").write_text("class A {}\n") # NOT docs-only, so the exemption does not apply
|
|
_git(dw, "add", "-A")
|
|
_git(dw, "commit", "-qm", "feat: thing\n\nfixes #999")
|
|
dw_head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=str(dw),
|
|
capture_output=True, text=True).stdout.strip()
|
|
dw_prev = subprocess.run(["git", "rev-parse", "HEAD~1"], cwd=str(dw),
|
|
capture_output=True, text=True).stdout.strip()
|
|
dw_env = dict(base_env)
|
|
dw_env["ETV_GITEA_URL"] = f"http://127.0.0.1:{srv.server_address[1]}"
|
|
dw_env["ETV_GITEA_BASICAUTH"] = "stub:stub"
|
|
cases["prepush-donewhen"] = [(
|
|
"unticked-donewhen-blocks-push-to-main",
|
|
f"refs/heads/main {dw_head} refs/heads/main {dw_prev}\n", None, dw_env, dw,
|
|
)]
|
|
|
|
return cases
|
|
|
|
|
|
def _ab_cases(hook: Path, sandbox, positives) -> list[tuple]:
|
|
"""(label, payload, arg, env, cwd) for the generic matrix PLUS this hook's constructed cases."""
|
|
root, env = sandbox
|
|
cases: list[tuple] = []
|
|
args = [None, "start", "finish"] if hook.stem == "design-sync-reminder" else [None]
|
|
for name, template in PAYLOADS.items():
|
|
for arg in args:
|
|
cases.append((
|
|
f"matrix:{name}:{arg}", template % {"cwd": str(root), "nonce": name},
|
|
arg, env, root,
|
|
))
|
|
for label, payload, arg, penv, cwd in positives.get(hook.stem, []):
|
|
cases.append((f"positive:{label}", payload, arg, penv, cwd))
|
|
return cases
|
|
|
|
|
|
def _ab_run(hook: Path, sandbox, case, tag_root: str):
|
|
"""Run the control and the instrumented hook for one case. Returns both (rc, stdout, stderr)."""
|
|
root = sandbox[0]
|
|
control = root / f"control-{hook.name}"
|
|
if not control.exists():
|
|
control.write_text(strip_instrumentation(hook.read_text()))
|
|
control.chmod(0o755)
|
|
_label, payload, arg, env, cwd = case
|
|
|
|
# A SEPARATE TMPDIR PER SIDE, not per pair. `design-sync-reminder.sh` throttles itself with a
|
|
# marker file under $TMPDIR, so a shared one let the control fire and then silently suppress the
|
|
# instrumented run — the A/B reported a swallowed decision that was really the hook's own
|
|
# one-shot behaviour working correctly. Sharing state between the two halves of a differential
|
|
# test makes it measure the state, not the difference.
|
|
def side(t: str) -> dict:
|
|
e = dict(env)
|
|
e["TMPDIR"] = str(root / f"tmp-{tag_root}-{t}")
|
|
Path(e["TMPDIR"]).mkdir(parents=True, exist_ok=True)
|
|
return e
|
|
|
|
def one(script: Path, t: str):
|
|
cmd = ["bash", str(script)] + ([arg] if arg else [])
|
|
p = subprocess.run(cmd, input=payload.encode(), capture_output=True,
|
|
cwd=str(cwd), env=side(t), timeout=90)
|
|
return p.returncode, p.stdout, p.stderr
|
|
|
|
return one(control, "control"), one(hook, "instrumented")
|
|
|
|
|
|
def _normalise_stderr(err: bytes) -> bytes:
|
|
"""Erase the two things that MUST differ between the control and the instrumented hook.
|
|
|
|
bash prefixes its own diagnostics with `$0` and a line number. The control is a copy at a
|
|
different path, and the instrumentation adds four lines to the top, so *every* bash-emitted
|
|
message differs in exactly those two fields for reasons that have nothing to do with
|
|
transparency. Only the script path and `line N:` are normalised — narrowly, so that a genuine
|
|
difference in the message body still fails.
|
|
"""
|
|
err = re.sub(rb"[^\s:]*/[^\s:]*\.sh:", b"<script>:", err)
|
|
err = re.sub(rb"line \d+:", b"line <n>:", err)
|
|
return err
|
|
|
|
|
|
@pytest.mark.parametrize("hook", hook_scripts(), ids=lambda h: h.stem)
|
|
def test_instrumentation_changes_NOTHING_the_harness_can_see(hook, sandbox, positives):
|
|
"""THE REGRESSION NET. Every hook, every case: byte-equal stdout, byte-equal stderr, equal status.
|
|
|
|
This is what stands between #776 and having silently disarmed the guard set while adding a log
|
|
that cheerfully reports every hook firing.
|
|
|
|
STDERR is compared too. It was not, and that hid a real change: when the sink is unreadable the
|
|
preamble's `etv_hook_fire_begin` was an undefined command, so the hook printed `command not
|
|
found` to stderr and the A/B could not see it. Anything the harness or a human can observe
|
|
belongs in the comparison, not just the channel the decision travels on.
|
|
"""
|
|
for i, case in enumerate(_ab_cases(hook, sandbox, positives)):
|
|
(rc_a, out_a, err_a), (rc_b, out_b, err_b) = _ab_run(hook, sandbox, case, f"{hook.stem}-{i}")
|
|
label = f"{hook.name} {case[0]}"
|
|
assert rc_a == rc_b, f"{label}: exit status changed {rc_a} -> {rc_b}"
|
|
assert out_a == out_b, f"{label}: stdout changed.\n without: {out_a!r}\n with: {out_b!r}"
|
|
na, nb = _normalise_stderr(err_a), _normalise_stderr(err_b)
|
|
assert na == nb, f"{label}: stderr changed.\n without: {na!r}\n with: {nb!r}"
|
|
|
|
|
|
@pytest.mark.parametrize("hook", hook_scripts(), ids=lambda h: h.stem)
|
|
def test_the_AB_is_not_VACUOUS_for_any_hook(hook, sandbox, positives):
|
|
"""ANTI-VACUITY FOR CLAIM 2 — the assertion the first version of this file was missing.
|
|
|
|
`test_instrumentation_changes_NOTHING_the_harness_can_see` compares the two halves against each
|
|
other. If a hook emits nothing and exits 0 on every case, that comparison is `("", 0) == ("", 0)`
|
|
and passes on a wrapper that discards all output. Measured on the original matrix: 160 of 165
|
|
comparisons were empty-vs-empty and 165 of 165 were `0 == 0`, so deleting the entire replay block
|
|
left the differential green for four hooks, two of which issue `deny`.
|
|
|
|
So each hook must have at least one case where the CONTROL side actually says something —
|
|
non-empty stdout, or a non-zero exit. The population is the hook glob, so a new hook arrives
|
|
here uncovered-and-red rather than quietly riding on a matrix that never reaches it.
|
|
|
|
ONE HOOK IS EXEMPT, and the exemption is narrow, derived and stated rather than assumed:
|
|
`posttooluse-worktree-marker.sh` emits nothing and exits 0 on every input BY DESIGN — its whole
|
|
output is a marker file. Testing it for non-empty stdout would be testing for a bug. Its
|
|
transparency is asserted instead through its side effect, in
|
|
`test_the_worktree_markers_SIDE_EFFECT_is_unchanged`.
|
|
"""
|
|
if hook.stem == "posttooluse-worktree-marker":
|
|
pytest.skip("emits nothing by design; covered by its side-effect test instead")
|
|
|
|
productive = []
|
|
for i, case in enumerate(_ab_cases(hook, sandbox, positives)):
|
|
(rc_a, out_a, _), _ = _ab_run(hook, sandbox, case, f"vac-{hook.stem}-{i}")
|
|
if out_a or rc_a != 0:
|
|
productive.append((case[0], len(out_a), rc_a))
|
|
|
|
assert productive, (
|
|
f"{hook.name}: NOT ONE case makes this hook emit output or exit non-zero, so every "
|
|
"comparison in the differential test is empty-vs-empty and it would stay green with the "
|
|
"instrumentation's entire replay path deleted. Add a constructed case to `positives()` that "
|
|
"drives this hook to its deciding branch."
|
|
)
|
|
|
|
|
|
def test_the_worktree_markers_SIDE_EFFECT_is_unchanged(sandbox, tmp_path):
|
|
"""The one hook whose whole output is a file, compared on that file rather than on stdout."""
|
|
env = sandbox[1]
|
|
hook = HOOKS_DIR / "posttooluse-worktree-marker.sh"
|
|
results = []
|
|
for tag in ("control", "instrumented"):
|
|
repo = tmp_path / tag
|
|
repo.mkdir()
|
|
_git(repo, "init", "-q", ".")
|
|
(repo / "f").write_text("x")
|
|
_git(repo, "add", "-A")
|
|
_git(repo, "commit", "-qm", "c")
|
|
target = tmp_path / f"wt-{tag}"
|
|
script = hook
|
|
if tag == "control":
|
|
script = tmp_path / "control-marker.sh"
|
|
script.write_text(strip_instrumentation(hook.read_text()))
|
|
import json as _json
|
|
payload = _json.dumps({
|
|
"session_id": "SESSION-XYZ", "hook_event_name": "PostToolUse", "tool_name": "Bash",
|
|
"cwd": str(repo),
|
|
"tool_input": {"command": f"git worktree add {target} HEAD"},
|
|
})
|
|
_git(repo, "worktree", "add", "-q", str(target), "HEAD")
|
|
_run(script, payload, env, repo)
|
|
marker = target / ".claude-worktree-owner"
|
|
results.append(marker.read_text() if marker.exists() else None)
|
|
|
|
assert results[0] == results[1] == "SESSION-XYZ\n", (
|
|
f"the marker the hook writes differs with and without instrumentation: {results}"
|
|
)
|
|
|
|
|
|
def test_stdout_is_replayed_BYTE_EXACT(sandbox):
|
|
"""Pinned separately from the A/B, because the A/B cannot see this class of loss.
|
|
|
|
`out=$(cat file)` strips trailing newlines, so replaying stdout through a variable delivers the
|
|
hook's JSON a newline short. Every JSON parser accepts the short form, and the A/B compares the
|
|
two halves against each other rather than against what the hook wrote — so a wrapper that
|
|
truncated BOTH would pass it. Only comparing against the literal expected bytes catches this.
|
|
"""
|
|
root, env = sandbox
|
|
hook = root / "trailing.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -euo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
"etv_hook_fire_begin trailing \"\" capture || true\n"
|
|
"input=$(cat)\n"
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}\n\n"' + "\n"
|
|
)
|
|
rc, out = _run(hook, '{"tool_name":"Bash"}', env, root)
|
|
assert rc == 0
|
|
assert out == b'{"hookSpecificOutput":{"permissionDecision":"deny"}}\n\n', (
|
|
f"trailing bytes were altered: {out!r}"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("code", [0, 1, 2, 3])
|
|
def test_the_exit_code_is_preserved(sandbox, code):
|
|
"""Exit 2 is the harness's block channel and non-zero is git's. A wrapper that normalised the
|
|
status to 0 would turn every git hook into a rubber stamp while logging that it ran."""
|
|
root, env = sandbox
|
|
hook = root / f"exit{code}.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
f'etv_hook_fire_begin exit{code} "" capture || true\n'
|
|
f"exit {code}\n"
|
|
)
|
|
rc, _ = _run(hook, "{}", env, root)
|
|
assert rc == code, f"exit {code} came back as {rc}"
|
|
|
|
|
|
def test_stdin_reaches_the_hook_intact(sandbox):
|
|
"""The wrapper reads stdin before the hook does. If it did not replay it faithfully, every
|
|
guard that parses `.tool_input` would see an empty payload and fail open, silently."""
|
|
root, env = sandbox
|
|
payload = '{"tool_name":"Bash","tool_input":{"command":"echo \\"quoted\\" && ls"},"session_id":"s"}'
|
|
hook = root / "echo-stdin.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -euo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin echo-stdin "" capture || true\n'
|
|
"input=$(cat)\nprintf '%s' \"$input\"\n"
|
|
)
|
|
rc, out = _run(hook, payload, env, root)
|
|
assert rc == 0
|
|
assert out.decode() == payload, f"stdin was altered:\n sent: {payload}\n saw: {out.decode()}"
|
|
|
|
|
|
def test_a_TTY_stdin_is_not_slurped(sandbox):
|
|
"""The hang this nearly shipped with.
|
|
|
|
An interactive `git commit` hands its hooks a terminal on fd 0. `cat` on a tty blocks until the
|
|
user types EOF, so an unconditional slurp would freeze every commit — instrumentation hanging
|
|
the very thing it was added to observe. Driven through a real pty, because the `[ ! -t 0 ]`
|
|
guard is precisely what a pipe-based test cannot exercise.
|
|
"""
|
|
root, env = sandbox
|
|
hook = root / "tty.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin ttyhook "" stream || true\n'
|
|
"echo ALIVE\n"
|
|
)
|
|
master, slave = pty.openpty()
|
|
try:
|
|
p = subprocess.Popen(
|
|
["bash", str(hook)], stdin=slave, stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL, cwd=str(root), env=env,
|
|
)
|
|
os.close(slave)
|
|
try:
|
|
out, _ = p.communicate(timeout=20)
|
|
except subprocess.TimeoutExpired:
|
|
p.kill()
|
|
pytest.fail(
|
|
"the hook hung with a tty on stdin — the instrumentation slurped a terminal, "
|
|
"which would freeze every interactive `git commit`."
|
|
)
|
|
finally:
|
|
os.close(master)
|
|
assert b"ALIVE" in out
|
|
|
|
|
|
def test_logging_failure_does_not_break_the_hook(sandbox):
|
|
"""Fail-open, in the one direction that is correct here. This file is observability; if it
|
|
cannot write, the guard must still guard. The inverse — a logging bug that swallows a `deny` —
|
|
is the failure mode the A/B test above exists to prevent."""
|
|
root, env = sandbox
|
|
bad = dict(env)
|
|
bad["ETV_HOOK_FIRE_LOG_DIR"] = "/dev/null/cannot-exist"
|
|
hook = root / "failopen.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -euo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin failopen "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}"' + "\n"
|
|
)
|
|
rc, out = _run(hook, '{"tool_name":"Bash"}', bad, root)
|
|
assert rc == 0
|
|
assert out == b'{"hookSpecificOutput":{"permissionDecision":"deny"}}', (
|
|
f"an unwritable log dir changed the hook's decision output: {out!r}"
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------
|
|
# THE RECORD ITSELF, and the report over it
|
|
# ------------------------------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"emit,mode,expected",
|
|
[
|
|
(r'{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}', "capture", "deny"),
|
|
(r'{\"hookSpecificOutput\":{\"permissionDecision\":\"ask\"}}', "capture", "ask"),
|
|
(r'{\"hookSpecificOutput\":{\"permissionDecision\":\"allow\"}}', "capture", "allow"),
|
|
(r'{\"hookSpecificOutput\":{\"additionalContext\":\"hi\"}}', "capture", "context"),
|
|
(r'{\"decision\":\"block\",\"reason\":\"r\"}', "capture", "block"),
|
|
("", "capture", "no-op"),
|
|
("", "stream", "pass"),
|
|
],
|
|
)
|
|
def test_the_recorded_decision_is_the_one_the_hook_emitted(sandbox, emit, mode, expected):
|
|
"""The decision in the log is derived from the bytes the harness receives, never from something
|
|
the hook author remembered to declare. That is the whole difference between this and inference.
|
|
"""
|
|
root, env = sandbox
|
|
logdir = root / f"log-{expected}-{mode}"
|
|
runenv = dict(env)
|
|
runenv["ETV_HOOK_FIRE_LOG_DIR"] = str(logdir)
|
|
|
|
hook = root / f"decide-{expected}-{mode}.sh"
|
|
body = f'printf "{emit}"\n' if emit else ""
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
f'etv_hook_fire_begin decider "" {mode} || true\n'
|
|
"input=$(cat)\n" + body
|
|
)
|
|
rc, _ = _run(hook, '{"session_id":"sess1","tool_name":"Bash"}', runenv, root)
|
|
assert rc == 0
|
|
|
|
records = (logdir / "sess1.jsonl").read_text().splitlines()
|
|
exits = [r for r in records if '"phase":"exit"' in r]
|
|
assert len(exits) == 1, f"expected one exit record, got {records}"
|
|
assert f'"decision":"{expected}"' in exits[0], f"expected {expected} in {exits[0]}"
|
|
assert any('"phase":"fire"' in r for r in records), "no fire record was written"
|
|
|
|
|
|
def test_a_stream_mode_hook_that_BLOCKS_is_recorded_as_blocked(sandbox):
|
|
"""A git hook decides by exit code; recording a non-zero exit as `pass` would put the report
|
|
back into the business of inferring, with the answer inverted on the only path that matters."""
|
|
root, env = sandbox
|
|
logdir = root / "log-blocked"
|
|
runenv = dict(env)
|
|
runenv["ETV_HOOK_FIRE_LOG_DIR"] = str(logdir)
|
|
hook = root / "blocker.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin blocker "" stream || true\n'
|
|
"input=$(cat)\nexit 1\n"
|
|
)
|
|
rc, _ = _run(hook, '{"session_id":"sess1"}', runenv, root)
|
|
assert rc == 1
|
|
assert '"decision":"blocked"' in (logdir / "sess1.jsonl").read_text()
|
|
|
|
|
|
def test_the_report_names_hooks_that_NEVER_fired(sandbox):
|
|
"""The finding #776 asks for is the ZERO row, so the report must derive its rows from the hooks
|
|
that exist rather than from the hooks that happen to appear in the log. A report built from the
|
|
log alone can only ever show hooks that fired, and would print a clean table while the broken
|
|
hook is the one missing from it."""
|
|
root, env = sandbox
|
|
logdir = root / "log-report"
|
|
logdir.mkdir()
|
|
(logdir / "sessX.jsonl").write_text(
|
|
'{"ts":"t","session":"sessX","pid":"1","hook":"pretooluse-bash-guard","label":"",'
|
|
'"event":"PreToolUse","tool":"Bash","phase":"fire","code":"","decision":""}\n'
|
|
'{"ts":"t","session":"sessX","pid":"1","hook":"pretooluse-bash-guard","label":"",'
|
|
'"event":"PreToolUse","tool":"Bash","phase":"exit","code":"0","decision":"deny"}\n'
|
|
)
|
|
p = subprocess.run(
|
|
["bash", str(SINK), "report", "--all", "--dir", str(logdir)],
|
|
capture_output=True, cwd=str(REPO_ROOT), env=env, timeout=60,
|
|
)
|
|
out = p.stdout.decode()
|
|
assert p.returncode == 0, p.stderr.decode()
|
|
assert "pretooluse-bash-guard" in out and "deny=1" in out
|
|
|
|
for hook in hook_scripts():
|
|
assert hook.stem in out, (
|
|
f"{hook.stem} is absent from the report. Only hooks present in the log were listed, so "
|
|
"a hook that never fires — the finding this report exists to surface — is invisible."
|
|
)
|
|
zero_rows = [ln for ln in out.splitlines() if re.search(r"\s0\s", ln)]
|
|
assert zero_rows, "no never-fired rows shown despite only one hook appearing in the log"
|
|
|
|
|
|
def test_the_report_REFUSES_an_empty_population(tmp_path, sandbox):
|
|
"""Anti-vacuity on the read side. A report over zero hooks would print a tidy empty table and
|
|
read as 'all hooks accounted for' — the shape of #751's 6-second green.
|
|
|
|
The sink must be COPIED into a hook-less tree, not merely pointed at one via
|
|
`CLAUDE_PROJECT_DIR`. The script deliberately falls back to its own location when that variable
|
|
names a tree with no hooks — that fallback is correct, it is what makes `hook-fire-log.sh
|
|
report` work from any directory — but it means setting the variable alone lands the report back
|
|
on the real repo's 13 live hooks, and the test would then assert against a full population while
|
|
claiming to exercise the empty one.
|
|
"""
|
|
_, env = sandbox
|
|
fake_scripts = tmp_path / "scripts"
|
|
fake_scripts.mkdir()
|
|
copied = fake_scripts / "hook-fire-log.sh"
|
|
copied.write_text(SINK.read_text())
|
|
copied.chmod(0o755)
|
|
|
|
runenv = dict(env)
|
|
runenv.pop("CLAUDE_PROJECT_DIR", None)
|
|
p = subprocess.run(
|
|
["bash", str(copied), "report", "--all", "--dir", str(tmp_path / "log")],
|
|
capture_output=True, cwd=str(tmp_path), env=runenv, timeout=60,
|
|
)
|
|
assert p.returncode == 2, (
|
|
"the report exited 0 over an empty hook population. It must refuse rather than print a "
|
|
f"clean table of nothing. stdout={p.stdout.decode()!r}"
|
|
)
|
|
assert b"refusing to report" in p.stderr
|
|
|
|
|
|
# ------------------------------------------------------------------------------------------------
|
|
# PROOFS FOR THE ROUND-2 FIXES. Each of these went red before its fix and green after; each names
|
|
# the failure it pins, because "this test exists for a reason" is not a reason.
|
|
# ------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def test_stderr_is_NOT_silenced(sandbox):
|
|
"""The worst defect this change shipped, and the cheapest to have missed.
|
|
|
|
`exec 0<"$sin" 2>/dev/null` reads as "suppress errors from this redirection". It is not: `exec`
|
|
with redirections and no command applies them to the shell PERMANENTLY, so it sent the HOOK'S
|
|
ENTIRE STDERR to /dev/null for the rest of its life. Every husky guard's user-facing output is
|
|
stderr — the H6 "push to main BLOCKED" message, the BOM guard's remediation text — so the guards
|
|
went on blocking while telling the human nothing about why.
|
|
|
|
Invisible to the original A/B, which compared only stdout and exit status.
|
|
"""
|
|
root, env = sandbox
|
|
hook = root / "noisy.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin noisy "" stream || true\n'
|
|
"input=$(cat)\n"
|
|
'printf "IMPORTANT DIAGNOSTIC\\n" >&2\n'
|
|
"exit 1\n"
|
|
)
|
|
p = subprocess.run(["bash", str(hook)], input=b'{"tool_name":"Bash"}',
|
|
capture_output=True, cwd=str(root), env=env, timeout=60)
|
|
assert p.returncode == 1
|
|
assert b"IMPORTANT DIAGNOSTIC" in p.stderr, (
|
|
f"the hook's stderr was swallowed by the instrumentation: {p.stderr!r}"
|
|
)
|
|
|
|
|
|
def test_output_SURVIVES_a_vanished_stdout_tempfile(sandbox):
|
|
"""The swallowed-`deny` path, found by cold review by stubbing `mktemp`.
|
|
|
|
Two coupled mistakes: the fd restore lived inside the same conditional as the replay, so a
|
|
missing file left stdout still pointing at the temp target; and the replay read the PATH, so an
|
|
unlinked-but-open file was unrecoverable even though its bytes existed. The hook here deletes
|
|
its own scratch after printing, which is what a $TMPDIR reaper does to it from outside.
|
|
"""
|
|
root, env = sandbox
|
|
hook = root / "vanish.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin vanish "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}"' + "\n"
|
|
'rm -f "${TMPDIR:-/tmp}"/etv-hook-stdout.* 2>/dev/null || true\n'
|
|
)
|
|
tmp = root / "vanish-tmp"
|
|
tmp.mkdir(exist_ok=True)
|
|
p = subprocess.run(["bash", str(hook)], input=b'{"tool_name":"Bash"}', capture_output=True,
|
|
cwd=str(root), env={**env, "TMPDIR": str(tmp)}, timeout=60)
|
|
assert p.returncode == 0
|
|
assert b'"permissionDecision":"deny"' in p.stdout, (
|
|
"the guard's deny was DISCARDED when its stdout scratch file vanished mid-run. The hook "
|
|
f"exited 0 with no output, which the harness reads as allow. stdout={p.stdout!r}"
|
|
)
|
|
|
|
|
|
def test_output_survives_a_SIGTERM(sandbox):
|
|
"""A hook killed by the harness timeout must still deliver what it already printed.
|
|
|
|
Measured before the fix: 52 bytes delivered without instrumentation, 0 with it. Bash runs no
|
|
EXIT trap for an UNTRAPPED fatal signal, so the decision sat in a temp file nobody read. Every
|
|
capture-mode hook today prints and exits immediately, so the window is sub-millisecond — but
|
|
that is a property of today's hooks, not of the mechanism, and nothing else would tell a future
|
|
editor they had just made it reachable.
|
|
"""
|
|
import signal
|
|
import time
|
|
|
|
root, env = sandbox
|
|
hook = root / "slow.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin slow "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}"' + "\n"
|
|
"sleep 30\n"
|
|
)
|
|
payload_file = root / "sigterm-stdin.json"
|
|
payload_file.write_bytes(b'{"tool_name":"Bash"}')
|
|
with open(payload_file, "rb") as fh:
|
|
p = subprocess.Popen(["bash", str(hook)], stdin=fh, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, cwd=str(root), env=env)
|
|
time.sleep(2.0)
|
|
p.send_signal(signal.SIGTERM)
|
|
out, _ = p.communicate(timeout=30)
|
|
assert b'"permissionDecision":"deny"' in out, (
|
|
f"a SIGTERMed hook lost the decision it had already printed: {out!r}"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"emit,code,expected",
|
|
[
|
|
# Uppercase: a real decision was filed as unclassified prose because the class was [a-z] only.
|
|
(r'{\"permissionDecision\":\"Deny\"}', 0, "deny"),
|
|
# A non-zero status must not ERASE a decision the hook actually printed; the status has its
|
|
# own field in the record.
|
|
(r'{\"permissionDecision\":\"deny\"}', 1, "deny+error"),
|
|
# ...but a failure with nothing classifiable is still an error.
|
|
("", 1, "error"),
|
|
],
|
|
)
|
|
def test_the_classifier_does_not_MISREPORT(sandbox, emit, code, expected):
|
|
root, env = sandbox
|
|
logdir = root / f"cls-{expected}-{code}"
|
|
runenv = {**env, "ETV_HOOK_FIRE_LOG_DIR": str(logdir)}
|
|
hook = root / f"cls-{expected}-{code}.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin cls "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
+ (f'printf "{emit}"\n' if emit else "")
|
|
+ f"exit {code}\n"
|
|
)
|
|
subprocess.run(["bash", str(hook)], input=b'{"session_id":"s1","tool_name":"Bash"}',
|
|
capture_output=True, cwd=str(root), env=runenv, timeout=60)
|
|
rec = [ln for ln in (logdir / "s1.jsonl").read_text().splitlines() if '"phase":"exit"' in ln]
|
|
assert f'"decision":"{expected}"' in rec[0], f"expected {expected}, got {rec[0]}"
|
|
|
|
|
|
def test_the_word_additionalContext_in_PROSE_is_not_a_decision(sandbox):
|
|
"""`grep -q additionalContext` matched the bare word anywhere, including human text."""
|
|
root, env = sandbox
|
|
logdir = root / "cls-prose"
|
|
hook = root / "prose.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin prose "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
'printf "note: this hook does not use additionalContext at all\\n"\n'
|
|
)
|
|
subprocess.run(["bash", str(hook)], input=b'{"session_id":"s1","tool_name":"Bash"}',
|
|
capture_output=True, cwd=str(root),
|
|
env={**env, "ETV_HOOK_FIRE_LOG_DIR": str(logdir)}, timeout=60)
|
|
rec = [ln for ln in (logdir / "s1.jsonl").read_text().splitlines() if '"phase":"exit"' in ln]
|
|
assert '"decision":"output"' in rec[0], f"prose was classified as a decision: {rec[0]}"
|
|
|
|
|
|
def test_the_session_id_cannot_ESCAPE_the_log_directory(sandbox, tmp_path):
|
|
"""The log FILENAME was built from an unscrubbed payload field while the log FIELD was scrubbed.
|
|
|
|
Unreachable with harness-generated UUIDs, fixed anyway: a file whose stated thesis is "restrict
|
|
the value space so there is no escaping bug to have" should not exempt the one use that becomes
|
|
a path.
|
|
"""
|
|
_root, env = sandbox
|
|
logdir = tmp_path / "logs"
|
|
hook = tmp_path / "esc.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin esc "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
)
|
|
evil = '{"session_id":"../../escaped","tool_name":"Bash"}'
|
|
subprocess.run(["bash", str(hook)], input=evil.encode(), capture_output=True,
|
|
cwd=str(tmp_path), env={**env, "ETV_HOOK_FIRE_LOG_DIR": str(logdir)}, timeout=60)
|
|
written = list(logdir.glob("*.jsonl"))
|
|
assert written, "nothing was logged at all"
|
|
for f in written:
|
|
assert f.parent == logdir, f"a record escaped the log directory: {f}"
|
|
assert not (tmp_path / "escaped.jsonl").exists()
|
|
|
|
|
|
def test_the_suite_does_not_write_to_the_PRODUCTION_log(sandbox):
|
|
"""`conftest.py` must isolate every test, not just this file's.
|
|
|
|
Before it, running `pytest scripts/tests` put 115 synthetic fires into
|
|
`$HOME/.cache/ersatztv/hook-fire/` — 96 merge-consent including two `deny`s — so
|
|
`hook-fire-log.sh report` described the test suite while looking exactly like a record of real
|
|
sessions. That is the inference problem this change exists to abolish, one layer up.
|
|
"""
|
|
assert os.environ.get("ETV_HOOK_FIRE_LOG_DIR"), (
|
|
"ETV_HOOK_FIRE_LOG_DIR is not set for this test, so any hook it drives writes to the real "
|
|
"log. scripts/tests/conftest.py should be setting it for every test."
|
|
)
|
|
default = Path(os.path.expanduser("~/.cache/ersatztv/hook-fire"))
|
|
assert Path(os.environ["ETV_HOOK_FIRE_LOG_DIR"]).resolve() != default.resolve()
|
|
|
|
before = sorted(p.stat().st_mtime_ns for p in default.glob("*.jsonl")) if default.exists() else []
|
|
root, env = sandbox
|
|
hook = root / "prod-check.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin prodcheck "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
)
|
|
subprocess.run(["bash", str(hook)], input=b'{"session_id":"s1","tool_name":"Bash"}',
|
|
capture_output=True, cwd=str(root), env=env, timeout=60)
|
|
after = sorted(p.stat().st_mtime_ns for p in default.glob("*.jsonl")) if default.exists() else []
|
|
assert before == after, "a test run modified the production hook-fire log"
|
|
|
|
|
|
def test_DELETING_the_replay_makes_the_differential_go_RED(sandbox, positives, tmp_path):
|
|
"""THE MUTATION PROOF FOR THE TRANSPARENCY CLAUSE.
|
|
|
|
The coverage clause has its own proof (`test_a_hook_that_LOSES_its_instrumentation_is_DETECTED`).
|
|
That one says nothing about whether the differential test is load-bearing, and the inventory row
|
|
claims BOTH. So: disarm the replay in an isolated copy of the sink, point a real hook at it, and
|
|
the same comparison the differential performs must fail. Without this, "the A/B would catch a
|
|
broken wrapper" is an argument, and this repo's record is that arguments of that shape have been
|
|
wrong repeatedly — including for this very file, whose A/B was 97% empty-vs-empty on arrival.
|
|
"""
|
|
root, env = sandbox
|
|
mutated = tmp_path / "hook-fire-log.sh"
|
|
text = SINK.read_text()
|
|
marker = ' cat "$ETV_HOOK_FIRE_STDOUT_TMP" 2>/dev/null || true'
|
|
assert marker in text, "the replay line has moved; this mutation no longer targets it"
|
|
mutated.write_text(text.replace(marker, " : # MUTATED: replay removed", 1))
|
|
|
|
hook = tmp_path / "victim.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{mutated}"\n'
|
|
'etv_hook_fire_begin victim "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}"' + "\n"
|
|
)
|
|
p = subprocess.run(["bash", str(hook)], input=b'{"tool_name":"Bash"}', capture_output=True,
|
|
cwd=str(root), env={**env, "TMPDIR": str(tmp_path)}, timeout=60)
|
|
assert p.stdout == b"", (
|
|
"the mutation did not actually disarm the replay, so this proof establishes nothing: "
|
|
f"stdout={p.stdout!r}"
|
|
)
|