timothyandClaude Opus 5
b9f031f52a
Round 3, from two independent re-reviews of the round-2 fix commit. The fix for the lost-output-on-SIGTERM defect introduced a worse one than the defect. The signal handler ended in `exit "$?"`, and `$?` at signal time is the status of whatever command was running — normally 0 for a guard still gathering evidence. Measured on the real guard, `prepush-donewhen.sh` SIGTERMed mid-curl with an unticked Done-when box: uninstrumented rc=-15 git ABORTS the push instrumented rc=0 git PUSHES to main That converts a killed guard into consent, in the permissive direction, in the hook that protects `main`. The handler now flushes, restores the default disposition and re-raises the signal, so the process dies the way it would have died. test_output_survives_a_SIGTERM could not see it: it asserts stdout content, never reads the return code, and drives a capture-mode hook so no exit-code guard is ever signalled. Also from this round: - The flush ran TWICE on the signal path (signal handler, then EXIT trap), writing two contradictory exit records for one invocation — a real decision then a phantom no-op — so the report mis-histogrammed the very case the trap was added for. It is idempotent now. - `exec 4<&- 2>/dev/null`: the round-1 blocker pattern again, eight lines below the comment forbidding it. Rather than fix the third instance and wait for a fourth, test_no_exec_in_the_sink_carries_a_STDERR_REDIRECT now enforces it. - The log write leaked `Operation not permitted` to stderr on an unwritable log, because `>> "$file" 2>/dev/null` opens the file BEFORE redirecting stderr. The fail-open test used a path that dies at mkdir and never reached this. - Classification: an unrecognised value is no longer lowercased into a valid decision, exit 2 now dominates a printed `allow` (it is the harness's block channel, and reporting a permit for a refused call is the one direction this log must never be wrong in), and a non-zero exit no longer annotates a real decision with an invented `+error`. - Classification and replay read the same bytes; they disagreed when a hook used fd 4, so the harness saw `deny` while the log recorded `no-op`. - The stderr normaliser rewrote `line N:` globally, collapsing genuinely different diagnostics; it is anchored to the script-path prefix now. - docs 5.4 claimed the two zero rows were "trigger did not fire", justified by unit tests that invoke the scripts directly and therefore prove nothing about harness dispatch. It now separates what is established from what is not, and stops contradicting its own decision record. Refs: #776 Decisions-Edit: yes Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1284 lines
61 KiB
Python
1284 lines
61 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.
|
|
"""
|
|
# The two fields are normalised TOGETHER, as one anchored prefix, not independently. A global
|
|
# `line \d+:` substitution also rewrites application text that happens to contain that phrase,
|
|
# and cold review demonstrated it collapsing two genuinely different diagnostics ("highest
|
|
# private fd line 0" vs "line 4") into one — hiding exactly the kind of fd-state difference this
|
|
# comparison exists to catch.
|
|
return re.sub(rb"[^\s:]*/[^\s:]*\.sh: line \d+:", b"<script>: line <n>:", 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",
|
|
[
|
|
# A NON-CANONICAL value is recorded as non-canonical, not laundered into a valid decision.
|
|
# This case went both ways before settling: first filed as unclassified prose, then
|
|
# lowercased into a clean `deny` (manufacturing a decision the harness may never honour).
|
|
(r'{\"permissionDecision\":\"Deny\"}', 0, "unrecognized"),
|
|
# A non-zero status neither ERASES a printed decision nor annotates it. The status has its
|
|
# own field; the decision field states the decision and nothing else.
|
|
(r'{\"permissionDecision\":\"deny\"}', 1, "deny"),
|
|
# A failure with nothing classifiable IS an error.
|
|
("", 1, "error"),
|
|
# Exit 2 is the harness's block channel and DOMINATES the printed JSON. Recording the
|
|
# printed `allow` would report a permit for a call that was actually refused — the one
|
|
# direction a log of security decisions must never be wrong in.
|
|
(r'{\"permissionDecision\":\"allow\"}', 2, "deny-exit2"),
|
|
],
|
|
)
|
|
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 = ''' [ -n "$out" ] && printf '%s' "$out"'''
|
|
assert marker in text, (
|
|
"the replay line has moved; this mutation no longer targets it. Retarget it rather than "
|
|
"loosening the match — a mutation that silently stops mutating proves nothing, which is "
|
|
"the whole failure class this file exists to catch."
|
|
)
|
|
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}"
|
|
)
|
|
|
|
|
|
def test_a_SIGNALLED_stream_hook_still_FAILS(sandbox):
|
|
"""A signal must not become consent. THE regression from the round-2 signal trap.
|
|
|
|
Measured on the real guard: `prepush-donewhen.sh` SIGTERMed mid-`curl` exits -15 without
|
|
instrumentation, so git aborts the push; with an `exit "$?"`-based signal handler it returned 0
|
|
and git PUSHED TO MAIN, because `$?` at signal time was the status of whatever command was
|
|
running — 0 for a guard still gathering evidence.
|
|
|
|
`test_output_survives_a_SIGTERM` could not see it: that test asserts stdout CONTENT and never
|
|
reads the return code, and it drives a capture-mode hook, so no exit-code guard is ever
|
|
signalled. This one asserts the only thing git looks at.
|
|
"""
|
|
import signal
|
|
import time
|
|
|
|
root, env = sandbox
|
|
for sig, name in ((signal.SIGTERM, "TERM"), (signal.SIGINT, "INT"), (signal.SIGHUP, "HUP")):
|
|
hook = root / f"sig-{name}.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
f'etv_hook_fire_begin sig{name} "" stream || true\n'
|
|
"input=$(cat)\n"
|
|
"sleep 30\n"
|
|
"exit 1\n"
|
|
)
|
|
payload = root / f"sig-{name}.json"
|
|
payload.write_bytes(b'{"tool_name":"Bash"}')
|
|
with open(payload, "rb") as fh:
|
|
p = subprocess.Popen(["bash", str(hook)], stdin=fh, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, cwd=str(root), env=env)
|
|
time.sleep(1.5)
|
|
p.send_signal(sig)
|
|
p.communicate(timeout=30)
|
|
|
|
assert p.returncode != 0, (
|
|
f"a hook killed by SIG{name} reported SUCCESS ({p.returncode}). git reads this status: "
|
|
"for a pre-push guard that is the difference between aborting the push and letting it "
|
|
"through, so the instrumentation would have turned a killed guard into consent."
|
|
)
|
|
assert p.returncode == -sig, (
|
|
f"SIG{name}: expected death by signal ({-sig}), got {p.returncode}. The handler is "
|
|
"inventing an exit status instead of re-raising."
|
|
)
|
|
|
|
|
|
def test_the_exit_record_is_written_ONCE_on_the_signal_path(sandbox):
|
|
"""Bash runs the signal handler AND then the EXIT trap, so the flush must be idempotent.
|
|
|
|
Without the guard this wrote two contradictory `exit` records for one invocation — a real
|
|
decision followed by a phantom `no-op` — and the report histograms exit records, so the product
|
|
mis-reported precisely the case the signal trap was added to capture.
|
|
"""
|
|
import signal
|
|
import time
|
|
|
|
root, env = sandbox
|
|
logdir = root / "log-signal-once"
|
|
hook = root / "sigonce.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin sigonce "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}"' + "\n"
|
|
"sleep 30\n"
|
|
)
|
|
payload = root / "sigonce.json"
|
|
payload.write_bytes(b'{"session_id":"s1","tool_name":"Bash"}')
|
|
with open(payload, "rb") as fh:
|
|
p = subprocess.Popen(["bash", str(hook)], stdin=fh, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, cwd=str(root),
|
|
env={**env, "ETV_HOOK_FIRE_LOG_DIR": str(logdir)})
|
|
time.sleep(1.5)
|
|
p.send_signal(signal.SIGTERM)
|
|
out, _ = p.communicate(timeout=30)
|
|
|
|
records = (logdir / "s1.jsonl").read_text().splitlines()
|
|
exits = [r for r in records if '"phase":"exit"' in r]
|
|
assert len(exits) == 1, f"expected exactly one exit record, got {len(exits)}:\n" + "\n".join(records)
|
|
assert '"decision":"deny"' in exits[0], f"the surviving record lost the decision: {exits[0]}"
|
|
assert b'"permissionDecision":"deny"' in out, "the decision was not replayed to the harness"
|
|
|
|
|
|
def test_no_exec_in_the_sink_carries_a_STDERR_REDIRECT():
|
|
"""The rule that has now been broken twice, enforced instead of restated.
|
|
|
|
`exec` with redirections and no command applies them to the SHELL, permanently. `exec ... 2>/dev/null`
|
|
therefore silences the hook's whole stderr rather than suppressing an error from that one
|
|
redirection. It shipped once in the stdin replay (silencing every husky guard's user-facing
|
|
output), was fixed, and was then reintroduced eight lines below the comment forbidding it.
|
|
A grep is a weak detector in general; here the pattern is exact and the cost of missing it is a
|
|
guard that blocks while explaining nothing.
|
|
"""
|
|
offenders = [
|
|
f"{SINK.name}:{i}: {ln.strip()}"
|
|
for i, ln in enumerate(SINK.read_text().splitlines(), 1)
|
|
if re.match(r"\s*exec\s[^|&;#]*2>", ln) and not ln.lstrip().startswith("#")
|
|
]
|
|
assert not offenders, (
|
|
"these `exec` lines carry a stderr redirection, which applies to the whole shell and "
|
|
"silences the hook:\n " + "\n ".join(offenders)
|
|
)
|
|
|
|
|
|
def test_the_log_and_the_REPLAY_agree_when_the_hook_uses_fd_4(sandbox):
|
|
"""Classification and replay must read the same bytes.
|
|
|
|
They did not: the classifier read fd 4 while the replay preferred the file, so a hook that used
|
|
fd 4 for its own purposes replayed its `deny` correctly to the harness and filed it in the log
|
|
as `no-op` — the record contradicting the decision it exists to record.
|
|
"""
|
|
root, env = sandbox
|
|
logdir = root / "log-fd4"
|
|
hook = root / "fd4.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin fd4 "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
'exec 4</dev/null\n' # the hook takes fd 4 for itself
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}"' + "\n"
|
|
)
|
|
p = 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)
|
|
assert b'"permissionDecision":"deny"' in p.stdout, f"replay lost the decision: {p.stdout!r}"
|
|
exits = [r for r in (logdir / "s1.jsonl").read_text().splitlines() if '"phase":"exit"' in r]
|
|
assert '"decision":"deny"' in exits[0], (
|
|
f"the harness saw a deny but the log recorded something else: {exits[0]}"
|
|
)
|
|
|
|
|
|
def test_an_UNWRITABLE_log_file_is_still_stderr_transparent(sandbox, tmp_path):
|
|
"""Fail-open must be SILENT, not merely non-fatal.
|
|
|
|
`printf ... >> "$file" 2>/dev/null` looks like it suppresses a failure to open the log. It does
|
|
not: redirections apply left to right, so bash opens the file first and reports the failure on
|
|
the stderr still in force — the hook prints `Operation not permitted` straight at the harness.
|
|
`test_logging_failure_does_not_break_the_hook` could not see it because its path dies at `mkdir`
|
|
and never reaches an existing-but-unwritable destination.
|
|
"""
|
|
_root, env = sandbox
|
|
logdir = tmp_path / "ro-log"
|
|
logdir.mkdir()
|
|
victim = logdir / "s1.jsonl"
|
|
victim.write_text("")
|
|
victim.chmod(0o400)
|
|
|
|
hook = tmp_path / "ro.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin ro "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}"' + "\n"
|
|
)
|
|
p = subprocess.run(["bash", str(hook)], input=b'{"session_id":"s1","tool_name":"Bash"}',
|
|
capture_output=True, cwd=str(tmp_path),
|
|
env={**env, "ETV_HOOK_FIRE_LOG_DIR": str(logdir)}, timeout=60)
|
|
assert p.returncode == 0
|
|
assert p.stdout == b'{"hookSpecificOutput":{"permissionDecision":"deny"}}'
|
|
assert p.stderr == b"", (
|
|
"an unwritable log leaked a diagnostic to the harness. Observability that cannot write must "
|
|
f"be silent, not chatty: {p.stderr!r}"
|
|
)
|