timothyandtimothy
b6b3520bdb
Build ErsatzTV Image / CI toolchain image resolves (push) Successful in 8s
Build ErsatzTV Image / Delimiter ban (release path) (push) Successful in 26s
Build ErsatzTV Image / Build & test (.NET) (push) Successful in 8m32s
Build ErsatzTV Image / EF migration integrity (SQLite + MySql) (push) Successful in 6m12s
Build ErsatzTV Image / Functional E2E (curl + UI contracts) (push) Successful in 6m14s
Build ErsatzTV Image / API docs in sync (OpenAPI + endpoint index) (push) Skipped
Build ErsatzTV Image / Formatting (changed .cs conform to .editorconfig) (push) Skipped
Build ErsatzTV Image / Build & push image (amd64) (push) Successful in 4m32s
Co-authored-by: Timothy <timothy@noreply.gitea.tblindustries.be>
2044 lines
98 KiB
Python
2044 lines
98 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 the
|
|
tracked `.claude/hooks/*.sh`, never listed, per `testing.guard-derives-population-from-source`,
|
|
so a hook staged tomorrow is uninstrumented-and-red rather than silently unobserved. From
|
|
the git index rather than a filesystem walk since ersatztv#806 — an untracked file on one
|
|
machine is not part of the repo, and a guard whose population differs per checkout is one
|
|
nobody trusts.
|
|
|
|
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 pty
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts.tests.hook_fire_isolation import (
|
|
ENV_VAR,
|
|
SHARED_LOG_DIRS,
|
|
isolation_violation,
|
|
resolved_log_dir,
|
|
)
|
|
from scripts.tests.tracked_files import tracked_paths
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
# Directory + patterns resolved against the GIT INDEX, not `Path.glob` (ersatztv#806) — see
|
|
# `scripts/tests/tracked_files.py`. `HOOKS_DIR` survives only for error messages.
|
|
HOOKS = (".claude/hooks", ("*.sh",))
|
|
HUSKY = (".husky", ("*",))
|
|
HOOKS_DIR = REPO_ROOT / ".claude" / "hooks"
|
|
SINK = REPO_ROOT / "scripts" / "hook-fire-log.sh"
|
|
|
|
# CAPTURED AT IMPORT TIME ON PURPOSE — this is a specimen, not an oversight.
|
|
#
|
|
# `{**os.environ}` at module level is the shape #785 shipped: a suite that snapshots the environment
|
|
# once, at collection, and hands that stale mapping to every subprocess it launches. Because the
|
|
# autouse isolation fixture runs at test SETUP, such a snapshot used to predate it and every hook
|
|
# driven from it wrote to `$HOME/.cache/ersatztv/hook-fire/` — 58 records per run, with every
|
|
# assertion green, since the fire-log library is fail-open by design.
|
|
#
|
|
# `test_the_suite_does_not_write_to_the_PRODUCTION_log` asserts this specimen is now SAFE, which is
|
|
# the cross-suite claim that test has always made in its docstring and never checked in its body.
|
|
_IMPORT_TIME_ENV = {**os.environ}
|
|
|
|
# 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 GIT INDEX. Never a list, and never the filesystem (ersatztv#806).
|
|
|
|
The filesystem is not an authoritative source: an untracked `.sh` dropped in `.claude/hooks/` —
|
|
a scratch copy, a half-written hook — used to enter this population and be demanded to carry
|
|
instrumentation, reddening the suite on that checkout while CI, which never sees the file, stayed
|
|
green. That is #778's third shape, and a guard that fails everywhere except where it runs trains
|
|
its readers to ignore it.
|
|
"""
|
|
return tracked_paths(*HOOKS)
|
|
|
|
|
|
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 tracked_paths(*HUSKY):
|
|
if 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 tracked under {HOOKS_DIR} — the derivation 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_removes_EXACTLY_the_preamble_and_nothing_else():
|
|
"""The A/B control must be the hook minus the instrumentation — no more, no less.
|
|
|
|
This compared `strip_instrumentation(hook)` against the hook at the merge base, which asserted
|
|
two unrelated things at once: that the stripper is exact, AND that no hook was edited in this PR
|
|
for any other reason. The second is not a property worth pinning — it went red the moment
|
|
`pretooluse-bom-guard.sh` had a real defect fixed (`xxd`, absent on the CI runner, made it
|
|
fail open) — and coupling them means a legitimate change reads as a broken control.
|
|
|
|
The property that actually protects the differential is structural: every line the stripper
|
|
removes is a preamble line, and it adds nothing. Checked without reference to history, so it
|
|
holds on any branch and on any platform.
|
|
"""
|
|
import difflib
|
|
|
|
allowed = (
|
|
"# ersatztv#776",
|
|
"# git hook:",
|
|
"# Claude hook:",
|
|
"ETV_HOOK_FIRE_LIB=",
|
|
'[ -r "$ETV_HOOK_FIRE_LIB"',
|
|
"type etv_hook_fire_begin",
|
|
"etv_hook_fire_begin ",
|
|
)
|
|
for hook in hook_scripts():
|
|
cur = hook.read_text().splitlines(keepends=True)
|
|
stripped = strip_instrumentation("".join(cur)).splitlines(keepends=True)
|
|
for line in difflib.Differ().compare(cur, stripped):
|
|
if line.startswith("+ "):
|
|
raise AssertionError(f"{hook.name}: the stripper ADDED a line: {line[2:]!r}")
|
|
if line.startswith("- "):
|
|
body = line[2:]
|
|
assert body.strip() == "" or body.startswith(allowed), (
|
|
f"{hook.name}: the stripper removed a NON-preamble line, so the A/B control is "
|
|
f"not this hook: {body!r}"
|
|
)
|
|
|
|
|
|
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' # noqa: E501 - kept on one line so the fixture is greppable against the payload the hook receives
|
|
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"}}', # noqa: E501 - kept on one line so the fixture is greppable against the payload the hook receives
|
|
"bash-allow": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"ls -la"}}', # noqa: E501 - kept on one line so the fixture is greppable against the payload the hook receives
|
|
"git-commit": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Bash","cwd":"%(cwd)s","tool_input":{"command":"git commit -m x"}}', # noqa: E501 - kept on one line so the fixture is greppable against the payload the hook receives
|
|
"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"}}', # noqa: E501 - kept on one line so the fixture is greppable against the payload the hook receives
|
|
"agent-model": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Agent","cwd":"%(cwd)s","tool_input":{"description":"d","prompt":"p","model":"sonnet"}}', # noqa: E501 - kept on one line so the fixture is greppable against the payload the hook receives
|
|
"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}}', # noqa: E501 - kept on one line so the fixture is greppable against the payload the hook receives
|
|
"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"}}', # noqa: E501 - kept on one line so the fixture is greppable against the payload the hook receives
|
|
"ui-edit": '{"session_id":"s","hook_event_name":"PreToolUse","tool_name":"Edit","cwd":"%(cwd)s","tool_input":{"file_path":"web/src/screens/Channels.tsx"}}', # noqa: E501 - kept on one line so the fixture is greppable against the payload the hook receives
|
|
"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):
|
|
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"(?m)^[^\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}"
|
|
)
|
|
|
|
|
|
@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(
|
|
f'#!/usr/bin/env bash\nset -uo pipefail\n. "{SINK}"\netv_hook_fire_begin esc "" capture || true\ninput=$(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` isolates every test, and every MODULE — not just this file, and not just tests.
|
|
|
|
WHAT THIS ASSERTED BEFORE, and why that had to go. It snapshotted the `st_mtime_ns` of every
|
|
`~/.cache/ersatztv/hook-fire/*.jsonl`, drove one hook, and required the two lists equal. That
|
|
directory is the production log EVERY live Claude Code session on the machine writes to, so the
|
|
oracle was global mutable state: an unrelated session firing a hook inside the window failed
|
|
this test with `a test run modified the production hook-fire log` — an accusation about the
|
|
suite when the writer was another process. Observed on three separate branches, green on an
|
|
immediate re-run each time, and corroborated by an independent reviewer who found a file in that
|
|
directory written 13 seconds earlier by someone else (ersatztv#822). The misattribution was the
|
|
expensive part: it points the next reader at the suite, and the honest diagnosis costs a re-run
|
|
plus a directory listing.
|
|
|
|
WHAT IT ASSERTS NOW. The same invariant, structurally, reading nothing under `$HOME`.
|
|
`_IMPORT_TIME_ENV` is the #785 defect's own shape — the environment as a module sees it at
|
|
COLLECTION, before any fixture has run — and a real hook driven with it must land its records in
|
|
the isolated directory that snapshot carries. That is the cross-suite claim: isolation has to be
|
|
in place before a module is IMPORTED, not merely before a test runs, and the same ordering is
|
|
what covers module- and session-scoped fixtures, which no autouse function fixture can reach.
|
|
|
|
It asserts the EFFECT — records in the isolated dir — rather than the shape of the fix. Pinning
|
|
the shape is what `test_worktree_ownership_guard.py` had to correct once already: `_env()` can be
|
|
perfectly correct while a call site passes something else.
|
|
"""
|
|
# Both violations are BOUND before being asserted on, never called inside the `assert`
|
|
# expression: pytest rewrites the expression and prints the repr of every sub-expression, so
|
|
# `assert isolation_violation(os.environ) is None` dumps the whole environment — API keys and
|
|
# tokens included — into the failure output and from there into a CI log. Measured, not feared.
|
|
here = isolation_violation(os.environ)
|
|
assert here is None, here
|
|
|
|
# THE CROSS-SUITE CLAUSE. An autouse fixture alone cannot satisfy this: it runs at test setup,
|
|
# which is after every module in the session has been imported.
|
|
snapshot_violation = isolation_violation(_IMPORT_TIME_ENV)
|
|
assert snapshot_violation is None, (
|
|
"a module-level `{**os.environ}` snapshot resolves to the production hook-fire log. "
|
|
f"conftest.py's `pytest_configure` must set {ENV_VAR} BEFORE collection.\n{snapshot_violation}"
|
|
)
|
|
|
|
snapshot_dir = resolved_log_dir(_IMPORT_TIME_ENV)
|
|
before = {p.name for p in snapshot_dir.glob("*.jsonl")} if snapshot_dir.exists() else set()
|
|
|
|
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":"import-time-snapshot","tool_name":"Bash"}',
|
|
capture_output=True,
|
|
cwd=str(root),
|
|
env=dict(_IMPORT_TIME_ENV),
|
|
timeout=60,
|
|
)
|
|
|
|
# ANTI-VACUITY. Without this the test passes when the hook logged NOTHING AT ALL — which is also
|
|
# what a broken sink looks like, and it would then be asserting the absence of a leak by
|
|
# asserting the absence of any record.
|
|
after = {p.name for p in snapshot_dir.glob("*.jsonl")} if snapshot_dir.exists() else set()
|
|
# The EXACT record, not merely "a new file": the log file is named for the payload's session id,
|
|
# so `after > before` would also be satisfied by an `unknown-session.jsonl` written because
|
|
# payload parsing broke — a hook that logged the wrong thing, passing a test about isolation.
|
|
assert "import-time-snapshot.jsonl" in after - before, (
|
|
f"driving a hook with the import-time snapshot did not add `import-time-snapshot.jsonl` to "
|
|
f"{snapshot_dir} (new files: {sorted(after - before)}). Either the instrumentation stopped "
|
|
"firing — in which case this test proves nothing — or those records went somewhere else, "
|
|
"and the only somewhere else is a shared log this suite must never write."
|
|
)
|
|
|
|
|
|
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. 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"", (
|
|
f"the mutation did not actually disarm the replay, so this proof establishes nothing: stdout={p.stdout!r}"
|
|
)
|
|
|
|
|
|
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}"
|
|
)
|
|
|
|
|
|
def test_NUL_bytes_in_hook_output_survive(sandbox):
|
|
"""A shell variable cannot hold a NUL, so replaying through `$(...)` silently drops them.
|
|
|
|
Pathological for a JSON decision, but the replay must be byte-exact for whatever the hook wrote,
|
|
and "it drops bytes only for content we do not expect" is the reasoning this file exists to
|
|
avoid. The common path streams the file; only the unlinked-file rescue is lossy, and that is
|
|
stated in the code.
|
|
"""
|
|
root, env = sandbox
|
|
hook = root / "nul.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin nul "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
r"printf 'a\000b\n'" + "\n"
|
|
)
|
|
p = subprocess.run(
|
|
["bash", str(hook)], input=b'{"tool_name":"Bash"}', capture_output=True, cwd=str(root), env=env, timeout=60
|
|
)
|
|
assert p.stdout == b"a\x00b\n", f"NUL-containing output was mangled: {p.stdout!r}"
|
|
assert b"null byte" not in p.stderr, f"a warning leaked to the harness: {p.stderr!r}"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"emit,code,expected",
|
|
[
|
|
# Every non-canonical value, not just the ones a restricted character class admits.
|
|
(r"{\"permissionDecision\":\"deny2\"}", 0, "unrecognized"),
|
|
(r"{\"permissionDecision\":\"deny_now\"}", 0, "unrecognized"),
|
|
# Unclassified output plus a FAILING exit is an error, not `output`: the report histograms
|
|
# the decision, so filing it as `output` hid the failure entirely.
|
|
("diagnostic text", 1, "error"),
|
|
],
|
|
)
|
|
def test_odd_values_and_failing_exits_are_not_LAUNDERED(sandbox, emit, code, expected):
|
|
root, env = sandbox
|
|
logdir = root / f"odd-{expected}-{abs(hash((emit, code))) % 9999}"
|
|
hook = root / f"odd-{abs(hash((emit, code))) % 9999}.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin odd "" 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={**env, "ETV_HOOK_FIRE_LOG_DIR": str(logdir)},
|
|
timeout=60,
|
|
)
|
|
rec = [r for r in (logdir / "s1.jsonl").read_text().splitlines() if '"phase":"exit"' in r]
|
|
assert f'"decision":"{expected}"' in rec[0], f"expected {expected}, got {rec[0]}"
|
|
|
|
|
|
def test_an_INHERITED_flushed_flag_does_not_disable_reporting(sandbox):
|
|
"""`ETV_HOOK_FIRE_FLUSHED` was defaulted, never reset.
|
|
|
|
Exported by a parent — or set by an earlier `begin` in the same shell — a stale `1` made the
|
|
first flush return immediately: stdout stayed redirected and no exit record was written. Every
|
|
test assumed a clean environment, which is the assumption a hook running inside someone else's
|
|
process cannot make.
|
|
"""
|
|
root, env = sandbox
|
|
logdir = root / "log-inherited"
|
|
hook = root / "inherited.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin inherited "" 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(root),
|
|
env={**env, "ETV_HOOK_FIRE_LOG_DIR": str(logdir), "ETV_HOOK_FIRE_FLUSHED": "1"},
|
|
timeout=60,
|
|
)
|
|
assert p.stdout == b'{"hookSpecificOutput":{"permissionDecision":"deny"}}', (
|
|
f"an inherited FLUSHED=1 left stdout redirected and swallowed the decision: {p.stdout!r}"
|
|
)
|
|
exits = [r for r in (logdir / "s1.jsonl").read_text().splitlines() if '"phase":"exit"' in r]
|
|
assert exits and '"decision":"deny"' in exits[0], f"no exit record was written: {exits}"
|
|
|
|
|
|
@pytest.mark.parametrize("signame", ["TERM", "INT", "HUP"])
|
|
def test_a_SIGNALLED_hook_behaves_EXACTLY_as_an_uninstrumented_one(sandbox, signame):
|
|
"""The invariant that replaced the signal trap, asserted as an A/B rather than as a rescue.
|
|
|
|
A `trap ... TERM` was added so a killed hook would not lose already-printed output. It caused
|
|
three defects in three rounds, the last of which settles it: bash does not run a trap until the
|
|
current foreground command finishes, so an instrumented hook mid-`curl` took 30s to die where it
|
|
had taken 1s — a TERM-then-KILL supervisor then gets no flush at all AND a 29s stall. The rescue
|
|
was worth less than the transparency it cost.
|
|
|
|
So the contract is the same one the rest of this file asserts: under a signal, instrumented and
|
|
uninstrumented are indistinguishable. Both the STATUS and the TIME TO DIE are compared; the
|
|
latter is the one the trap broke, and no assertion on output could have caught it.
|
|
"""
|
|
import signal
|
|
import time
|
|
|
|
root, env = sandbox
|
|
sig = getattr(signal, f"SIG{signame}")
|
|
body = (
|
|
"input=$(cat)\n"
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"deny\"}}"' + "\n"
|
|
"sleep 25\n"
|
|
)
|
|
instrumented = root / f"sigab-{signame}.sh"
|
|
instrumented.write_text(
|
|
f'#!/usr/bin/env bash\nset -uo pipefail\n. "{SINK}"\netv_hook_fire_begin sigab "" stream || true\n' + body
|
|
)
|
|
control = root / f"sigab-control-{signame}.sh"
|
|
control.write_text("#!/usr/bin/env bash\nset -uo pipefail\n" + body)
|
|
|
|
results = {}
|
|
for tag, script in (("control", control), ("instrumented", instrumented)):
|
|
payload = root / f"sigab-{signame}-{tag}.json"
|
|
payload.write_bytes(b'{"session_id":"s1","tool_name":"Bash"}')
|
|
with open(payload, "rb") as fh:
|
|
# SIGNAL THE PROCESS GROUP, not the pid. A non-interactive bash defers a fatal signal
|
|
# until its foreground command finishes, trapped or not — so signalling the pid alone
|
|
# made BOTH sides take the full sleep and the timing assertion could not distinguish
|
|
# anything. Measured: pid-only gives control 4.0s and trapped 4.1s; group signalling
|
|
# gives control 0.002s and trapped 0.054s. A supervisor kills the group, so this is
|
|
# also the shape that actually occurs.
|
|
p = subprocess.Popen(
|
|
["bash", str(script)],
|
|
stdin=fh,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
cwd=str(root),
|
|
env=env,
|
|
start_new_session=True,
|
|
)
|
|
time.sleep(1.0)
|
|
t0 = time.monotonic()
|
|
os.killpg(os.getpgid(p.pid), sig)
|
|
out, err = p.communicate(timeout=40)
|
|
results[tag] = (p.returncode, time.monotonic() - t0, out, err)
|
|
|
|
(rc_a, dt_a, out_a, err_a), (rc_b, dt_b, out_b, err_b) = (
|
|
results["control"],
|
|
results["instrumented"],
|
|
)
|
|
assert out_a == out_b, f"SIG{signame}: stdout differs under signal: {out_a!r} vs {out_b!r}"
|
|
|
|
# ONE EXEMPTION, measured and narrow. On a group kill with a live foreground child, bash prints
|
|
# its own job-control notice (`Terminated: 15 sleep 5`). The instrumented hook does work in
|
|
# its EXIT handler and therefore lives ~50ms longer than the control, which dies before bash
|
|
# gets to print. Isolated: sourcing the sink alone does not produce it, and a no-op `EXIT` trap
|
|
# does not either — it is doing ANY work at exit that opens the window, which is the sink's
|
|
# whole purpose. Documented as a limit in `testing.hook-reports-its-own-execution` rather than
|
|
# papered over, and scoped to bash's exact notice so any other stderr difference still fails.
|
|
def strip_jobnotice(e: bytes) -> bytes:
|
|
# Two observed forms: bare (`Terminated: 15 sleep 5`) and attributed to whatever line of
|
|
# the sink was executing (`…/hook-fire-log.sh: line 109: 63769 Hangup: 1 sleep 5`). Both
|
|
# require one of bash's four signal words followed by `:` or whitespace, so an ordinary
|
|
# diagnostic still fails the comparison.
|
|
return b"\n".join(
|
|
ln for ln in e.split(b"\n") if not re.search(rb"(^|\s)(Terminated|Hangup|Interrupt|Killed)(:|\s|$)", ln)
|
|
)
|
|
|
|
assert strip_jobnotice(err_a) == strip_jobnotice(err_b), (
|
|
f"SIG{signame}: stderr differs under signal beyond bash's job-control notice: {err_a!r} vs {err_b!r}"
|
|
)
|
|
assert rc_a == rc_b, (
|
|
f"SIG{signame}: exit status differs, control={rc_a} instrumented={rc_b}. git and the "
|
|
"harness both read this status."
|
|
)
|
|
assert dt_b < dt_a + 2.0, (
|
|
f"SIG{signame}: the instrumented hook took {dt_b:.1f}s to die where the uninstrumented one "
|
|
f"took {dt_a:.1f}s. A trap handler does not run until the foreground command finishes, so "
|
|
"trapping the signal makes a killed hook hang for the length of whatever it was doing."
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"locale_env",
|
|
[
|
|
{"LANG": "en_US.UTF-8"},
|
|
{"LC_CTYPE": "en_US.UTF-8"},
|
|
{"LC_CTYPE": "UTF-8"}, # macOS Terminal's default, and the case that broke the fix
|
|
{"LC_ALL": "en_US.UTF-8"},
|
|
],
|
|
)
|
|
def test_an_INVALID_UTF8_byte_in_a_decision_is_still_classified(tmp_path, locale_env):
|
|
"""`local LC_ALL=C` does not export, so the child `sed`/`tr` never saw it — the fix was INERT.
|
|
|
|
One 0xE9 byte in a `permissionDecisionReason` made the classifier print `illegal byte sequence`
|
|
at the harness and file a real `deny` as `output`, and made the payload block file the record
|
|
under `unknown-session` with empty event and tool — manufacturing the exact never-fired vacuity
|
|
#776 exists to abolish.
|
|
|
|
It read as fixed because this author's shell sets `LANG` alone. The locale matrix is
|
|
parametrised because a single ambient locale tests one row of it: with an inherited `LC_CTYPE`,
|
|
`LC_CTYPE` outranks the exported `LANG=C` and the symptom returns in full.
|
|
"""
|
|
name = next(iter(locale_env.values()))
|
|
have = subprocess.run(["locale", "-a"], capture_output=True, text=True).stdout.split()
|
|
if name not in have and name.replace("UTF-8", "utf8") not in have:
|
|
pytest.skip(
|
|
f"{name} is not installed here, so bash itself warns and the assertion would be "
|
|
"measuring the missing locale rather than the export fix. The hazard needs a real "
|
|
"non-C locale to exist."
|
|
)
|
|
|
|
logdir = tmp_path / "loclog"
|
|
hook = tmp_path / "loc.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin loc "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
'printf \'{"hookSpecificOutput":{"permissionDecision":"deny",'
|
|
'"permissionDecisionReason":"caf\\xe9"}}\'\n'
|
|
)
|
|
# HOME copied only if present, like the launch-guard test below: an environment without it is
|
|
# legal (the sink has a `${HOME:-/tmp}` default) and indexing would redden this on a valid
|
|
# checkout for a reason unrelated to what it tests. This is the twin of the site fixed there.
|
|
env = {"PATH": os.environ["PATH"], "ETV_HOOK_FIRE_LOG_DIR": str(logdir), **locale_env}
|
|
if os.environ.get("HOME"):
|
|
env["HOME"] = os.environ["HOME"]
|
|
p = subprocess.run(
|
|
["bash", str(hook)],
|
|
input=b'{"session_id":"s1","tool_name":"Bash"}',
|
|
capture_output=True,
|
|
cwd=str(tmp_path),
|
|
env=env,
|
|
timeout=60,
|
|
)
|
|
|
|
assert p.stderr == b"", f"{locale_env}: a locale diagnostic leaked to the harness: {p.stderr!r}"
|
|
record = (logdir / "s1.jsonl").read_text()
|
|
assert '"decision":"deny"' in record, f"{locale_env}: the decision was misfiled:\n{record}"
|
|
assert '"tool":"Bash"' in record, f"{locale_env}: identity fields were lost:\n{record}"
|
|
assert not (logdir / "unknown-session.jsonl").exists(), (
|
|
f"{locale_env}: the record filed under unknown-session, so this fire reads as never having "
|
|
"happened in the report"
|
|
)
|
|
|
|
|
|
def test_a_NESTED_identity_field_does_not_outrank_the_TOP_LEVEL_one(tmp_path):
|
|
"""A greedy `.*` takes the LAST match, and payloads are one long line.
|
|
|
|
Removing the 64 KB cap is what armed this: a nested `{"session_id":…,"tool_name":…}` inside a
|
|
`tool_response` began outranking the top-level identity, filing the whole invocation under the
|
|
wrong session — so the real session reads as never-fired. The cap had been accidentally
|
|
protecting the right answer, which is the kind of load a bound can be silently carrying.
|
|
"""
|
|
logdir = tmp_path / "nestlog"
|
|
hook = tmp_path / "nest.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin nest "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
)
|
|
import json as _json
|
|
|
|
payload = _json.dumps(
|
|
{
|
|
"session_id": "TOP-SESSION",
|
|
"hook_event_name": "PreToolUse",
|
|
"tool_name": "TopTool",
|
|
"tool_input": {"filler": "x" * 1000},
|
|
"tool_response": {"session_id": "NESTED-SESSION", "tool_name": "NestedTool"},
|
|
}
|
|
)
|
|
subprocess.run(
|
|
["bash", str(hook)],
|
|
input=payload.encode(),
|
|
capture_output=True,
|
|
cwd=str(tmp_path),
|
|
env={**os.environ, "ETV_HOOK_FIRE_LOG_DIR": str(logdir)},
|
|
timeout=60,
|
|
)
|
|
|
|
assert (logdir / "TOP-SESSION.jsonl").exists(), (
|
|
f"records filed under the wrong session: {[f.name for f in logdir.glob('*.jsonl')]}"
|
|
)
|
|
rec = (logdir / "TOP-SESSION.jsonl").read_text()
|
|
assert '"tool":"TopTool"' in rec, f"a nested tool_name outranked the top-level one:\n{rec}"
|
|
|
|
|
|
def test_a_PRESENT_but_empty_decision_is_not_laundered(sandbox):
|
|
"""`{"permissionDecision":""}` extracted as empty, which is indistinguishable from absent, so it
|
|
bypassed `unrecognized` and was filed as `output` — a malformed decision laundered into "the
|
|
hook just printed something"."""
|
|
root, env = sandbox
|
|
logdir = root / "log-empty-decision"
|
|
hook = root / "emptydec.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin emptydec "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
r'printf "{\"hookSpecificOutput\":{\"permissionDecision\":\"\"}}"' + "\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 = [r for r in (logdir / "s1.jsonl").read_text().splitlines() if '"phase":"exit"' in r]
|
|
assert '"decision":"unrecognized"' in rec[0], f"an empty decision was laundered: {rec[0]}"
|
|
|
|
|
|
def test_identity_fields_BEYOND_the_fast_path_cap_are_still_found(tmp_path):
|
|
"""The bound must be an optimisation, never a correctness dependency.
|
|
|
|
A plain 256 KB cap reintroduced the exact defect the unbounded read had just fixed: a payload
|
|
whose filler pushes `session_id` past the cap returns nothing, the record files under
|
|
`unknown-session`, and that fire reads as NEVER HAVING HAPPENED in the report. The nested-field
|
|
test could not see it — its identities sit before only 1,000 bytes of filler.
|
|
|
|
So the cap is a fast path with an unbounded fallback, and this test puts the identity where only
|
|
the fallback can reach it.
|
|
"""
|
|
import json as _json
|
|
|
|
hook = tmp_path / "big.sh"
|
|
hook.write_text(
|
|
f'#!/usr/bin/env bash\nset -uo pipefail\n. "{SINK}"\netv_hook_fire_begin big "" capture || true\ninput=$(cat)\n'
|
|
)
|
|
logdir = tmp_path / "biglog"
|
|
payload = _json.dumps(
|
|
{
|
|
"tool_input": {"filler": "x" * 400_000},
|
|
"session_id": "TOP-SESSION",
|
|
"hook_event_name": "PreToolUse",
|
|
"tool_name": "TopTool",
|
|
}
|
|
)
|
|
subprocess.run(
|
|
["bash", str(hook)],
|
|
input=payload.encode(),
|
|
capture_output=True,
|
|
cwd=str(tmp_path),
|
|
env={**os.environ, "ETV_HOOK_FIRE_LOG_DIR": str(logdir)},
|
|
timeout=90,
|
|
)
|
|
|
|
assert (logdir / "TOP-SESSION.jsonl").exists(), (
|
|
"identity beyond the fast-path cap was not found, so the fire filed under "
|
|
f"{[f.name for f in logdir.glob('*.jsonl')]} and reads as never having happened"
|
|
)
|
|
rec = (logdir / "TOP-SESSION.jsonl").read_text()
|
|
assert '"tool":"TopTool"' in rec and '"event":"PreToolUse"' in rec, rec
|
|
|
|
|
|
def test_the_field_helper_does_not_LEAK_into_the_hooks_namespace(sandbox):
|
|
"""A bash function defined inside another function is still GLOBAL.
|
|
|
|
A short generic name (`_etv_field`) therefore survives `begin` and can collide with, or
|
|
overwrite, a function the hook itself defines — and it would refer to a now-unset variable.
|
|
"""
|
|
root, env = sandbox
|
|
hook = root / "ns.sh"
|
|
hook.write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{SINK}"\n'
|
|
'etv_hook_fire_begin ns "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
"if declare -F _etv_field >/dev/null 2>&1; then echo LEAKED_etv_field; fi\n"
|
|
"if declare -F etv_hook_fire__field >/dev/null 2>&1; then echo LEAKED_namespaced; fi\n"
|
|
"echo CLEAN\n"
|
|
)
|
|
p = subprocess.run(
|
|
["bash", str(hook)],
|
|
input=b'{"session_id":"s1","tool_name":"Bash"}',
|
|
capture_output=True,
|
|
cwd=str(root),
|
|
env=env,
|
|
timeout=60,
|
|
)
|
|
assert b"LEAKED" not in p.stdout, f"a helper leaked into the hook's namespace: {p.stdout!r}"
|
|
assert b"CLEAN" in p.stdout
|
|
|
|
|
|
def test_the_resolver_agrees_with_the_SINKS_OWN_expansion(tmp_path):
|
|
"""`resolved_log_dir` reimplements one shell expansion — so run both and compare, every run.
|
|
|
|
It is the guard's only model of where a child will write, so a disagreement fails OPEN: the
|
|
guard clears a launch, and the sink then points it at the real log. The two `:-` defaults are
|
|
exactly where a Python transliteration goes subtly wrong, because `dict.get(k, default)` returns
|
|
`""` for an empty value and shell's `${k:-default}` does not — which is why the empty cases are
|
|
here rather than only the obvious ones.
|
|
|
|
A differential against the real `scripts/hook-fire-log.sh`, not against a restatement of it: a
|
|
test that asserted the resolver equals a hardcoded path would agree with itself forever after
|
|
the sink moved.
|
|
"""
|
|
cases = [
|
|
{"HOME": str(tmp_path), ENV_VAR: str(tmp_path / "explicit")},
|
|
{"HOME": str(tmp_path)},
|
|
{"HOME": str(tmp_path), ENV_VAR: ""}, # empty is ABSENT to `:-`
|
|
{"HOME": ""}, # ... and so `${HOME:-/tmp}` takes /tmp
|
|
{},
|
|
# Normalisation: shell concatenates strings, `Path` collapses them. These agree on the
|
|
# DIRECTORY and differ in spelling — `/tmp/x/` vs `/tmp/x`, `$HOME//.cache` vs `$HOME/.cache`
|
|
# — which is exactly why the comparison below is the one the guard makes and not `==`.
|
|
{"HOME": str(tmp_path), ENV_VAR: str(tmp_path / "explicit") + "/"},
|
|
{"HOME": str(tmp_path) + "/"},
|
|
{"HOME": str(tmp_path), ENV_VAR: str(tmp_path) + "//double"},
|
|
]
|
|
for case in cases:
|
|
# The case is applied INSIDE the child, not through its launch environment.
|
|
# `etv_hook_fire_log_dir` reads the shell's environment at call time either way, so the
|
|
# differential is unchanged — but the launch itself then carries the isolated log dir like
|
|
# every other launch in the suite, instead of this test needing the guard to make an
|
|
# exception for it. (Several cases deliberately have no `ETV_HOOK_FIRE_LOG_DIR` at all,
|
|
# which is precisely what the guard exists to reject.)
|
|
setup = "\n".join(
|
|
f"export {var}={shlex.quote(case[var])}" if var in case else f"unset {var}" for var in ("HOME", ENV_VAR)
|
|
)
|
|
res = subprocess.run(
|
|
["bash", "-c", f'{setup}\n. "{SINK}"\netv_hook_fire_log_dir'],
|
|
capture_output=True,
|
|
text=True,
|
|
env={"PATH": os.environ["PATH"], ENV_VAR: os.environ[ENV_VAR]},
|
|
timeout=60,
|
|
)
|
|
assert res.returncode == 0, f"could not source the sink for {case}: {res.stderr}"
|
|
# SAME DIRECTORY, not same string. `os.path.realpath` is the comparison `isolation_violation`
|
|
# itself makes, so this asserts the contract the guard depends on; requiring byte equality
|
|
# would redden on a trailing slash that changes nothing about where the records land, and a
|
|
# guard nobody believes is the failure this file exists to avoid. A genuine divergence — a
|
|
# DIFFERENT directory — still fails, because realpath does not collapse those.
|
|
assert os.path.realpath(res.stdout) == os.path.realpath(resolved_log_dir(case)), (
|
|
f"the Python resolver and `etv_hook_fire_log_dir` resolve to DIFFERENT directories for "
|
|
f"{case}: the sink says {res.stdout!r}, resolved_log_dir says "
|
|
f"{str(resolved_log_dir(case))!r}. The guard would clear a launch the sink then points "
|
|
"somewhere else."
|
|
)
|
|
|
|
|
|
def test_a_launch_NOT_CARRYING_the_isolated_log_dir_FAILS_THE_LAUNCH():
|
|
"""The launch guard, driven end to end — the route pre-collection isolation cannot reach.
|
|
|
|
An environment assembled from scratch rather than derived from `os.environ` carries no
|
|
`ETV_HOOK_FIRE_LOG_DIR`, so the sink falls back to its own default. No snapshot is involved, so
|
|
nothing about import ordering helps. This is not hypothetical: three suites build exactly that
|
|
environment for the CI-shaped children they run, and they were found by MEASURING every launch
|
|
the suite makes, not by reading it.
|
|
|
|
BOTH FALLBACK BRANCHES ARE PINNED, because a guard that compared against one resolved production
|
|
path would pass the other. `${HOME:-/tmp}` means a child with no `HOME` lands in
|
|
`/tmp/.cache/ersatztv/hook-fire` — just as shared, just as persistent, and on this suite that was
|
|
the majority case: of the 83 from-scratch launches, 81 also carry no `HOME`.
|
|
|
|
The negative control is the last case — a guard that rejected every launch would satisfy all
|
|
three assertions above it and be useless.
|
|
"""
|
|
guard = getattr(subprocess.Popen, "etv_production_log_guard", None)
|
|
assert guard is not None, (
|
|
"the launch guard is not installed, so this test would pass by never being consulted. "
|
|
"conftest.py's `pytest_configure` should have installed it."
|
|
)
|
|
# INSTALLED IS NOT CONSULTED — but state precisely what this proves, because the obvious
|
|
# reading is wider than the fact. `checked` counts launches the wrapper has already seen, and
|
|
# this module's own `@pytest.mark.parametrize(... hook_scripts())` decorators shell out to
|
|
# `git ls-files` at IMPORT, so it is >= 2 even when this test is selected alone. What that
|
|
# establishes is that installation happened BEFORE this module was collected — i.e. the wrapper
|
|
# is in the path of launches made outside this test. Run alone, those are this module's own
|
|
# collection-time `git ls-files` calls, so it does order installation before collection; in a
|
|
# whole-suite run the counted launches may all belong to other modules, and it establishes only
|
|
# that the wrapper is consulted at all. It does NOT establish whole-suite coverage — nothing
|
|
# here could, since a launch that never reaches `Popen` is invisible by construction.
|
|
assert guard.checked > 0, (
|
|
"the launch guard has been consulted for no launch at all before this test, so it was "
|
|
"installed after this module was collected and cannot have seen collection-time launches"
|
|
)
|
|
|
|
def launch(env, cwd=None):
|
|
return subprocess.run(["bash", "-c", "true"], env=env, cwd=cwd, capture_output=True, timeout=60)
|
|
|
|
# HOME is COPIED IF PRESENT rather than read with `os.environ["HOME"]`: an environment without
|
|
# it is legal — the sink has a `${HOME:-/tmp}` default precisely for that — and indexing would
|
|
# raise KeyError, reddening this test on a valid checkout for a reason unrelated to the guard.
|
|
scratch = {"PATH": os.environ["PATH"]}
|
|
if os.environ.get("HOME"):
|
|
scratch["HOME"] = os.environ["HOME"]
|
|
|
|
# (1) the `$HOME` branch
|
|
with pytest.raises(AssertionError, match="carrying no ETV_HOOK_FIRE_LOG_DIR"):
|
|
launch(scratch)
|
|
|
|
# (2) the `/tmp` branch — no HOME either. This is the one a production-path comparison misses.
|
|
with pytest.raises(AssertionError, match="carrying no ETV_HOOK_FIRE_LOG_DIR"):
|
|
launch({"PATH": os.environ["PATH"]})
|
|
|
|
# (3) the variable set, and set WRONG. Omitting it is not the only way to reach a shared log,
|
|
# and BOTH shared logs are rejected — the likeliest wrong value is the sink's own default
|
|
# transliterated into a `dict.get(k, "<default>")`, which reaches whichever branch that default
|
|
# was copied from. Pinning only the `$HOME` one would leave the `/tmp` one clear.
|
|
for shared in SHARED_LOG_DIRS:
|
|
with pytest.raises(AssertionError, match="inside a shared hook-fire log"):
|
|
launch({**scratch, ENV_VAR: str(shared)})
|
|
|
|
# (4) a SUBDIRECTORY of a shared log. `etv_hook_fire_report` reads its directory with a
|
|
# recursive `find`, so records here are picked up by `hook-fire-log.sh report --dir <shared>`
|
|
# exactly like ones at its root — verified by driving a real hook into a nested directory and
|
|
# seeing it counted. An equality test would call this clean.
|
|
with pytest.raises(AssertionError, match="inside a shared hook-fire log"):
|
|
launch({**scratch, ENV_VAR: str(SHARED_LOG_DIRS[0] / "sub")})
|
|
|
|
# (5) the variable set RELATIVE, with a `cwd` that makes it land on a shared log. Found by
|
|
# cross-family review as a live bypass: a relative value resolves against the CHILD's directory,
|
|
# so a guard anchoring it at the parent's sees an unrelated path under the repo and clears it.
|
|
# The guard raises before `Popen` runs anything, so this launch never enters that directory.
|
|
shared = SHARED_LOG_DIRS[0]
|
|
with pytest.raises(AssertionError, match="inside a shared hook-fire log"):
|
|
launch({**scratch, ENV_VAR: ".cache/ersatztv/hook-fire"}, cwd=shared.parent.parent.parent)
|
|
|
|
# (6) NEGATIVE CONTROL: carrying the isolated dir, the same launch must go through.
|
|
ok = launch({**scratch, ENV_VAR: os.environ[ENV_VAR]})
|
|
assert ok.returncode == 0, "the guard rejected a correctly-isolated launch"
|
|
|
|
|
|
# The specimen the mutation proof below runs in a nested pytest: #785's shape, reduced to the one
|
|
# thing that matters. The hook it drives is written beside the sandbox rather than inline here, so
|
|
# this source carries no nested escaping.
|
|
_SPECIMEN_MODULE = """\
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
_ENV = {**os.environ} # snapshotted at IMPORT, before any fixture has run — this is the defect
|
|
|
|
HOOK = Path(__file__).resolve().parents[2] / "specimen-hook.sh"
|
|
|
|
|
|
def test_specimen_drives_a_hook():
|
|
subprocess.run(
|
|
["bash", str(HOOK)],
|
|
input=b'{"session_id":"specimen"}',
|
|
capture_output=True,
|
|
env=_ENV,
|
|
timeout=60,
|
|
)
|
|
"""
|
|
|
|
|
|
def test_the_guard_JUDGES_every_env_and_cwd_SHAPE_that_Popen_ACCEPTS(tmp_path):
|
|
"""`env` and `cwd` each have several legal shapes; the guard must judge, not crash, on all.
|
|
|
|
`Popen` accepts `cwd` as str, bytes or `PathLike`. The first version of the relative-path
|
|
anchoring called `os.fspath(cwd)` and then `os.path.join`, which raises `TypeError` on mixing
|
|
str with bytes — a guard that raises `TypeError` on a legitimate launch is not a stricter guard,
|
|
it is a broken one, and the failure names the wrong thing entirely.
|
|
|
|
THE PREMISE IS PINNED, not assumed. A file descriptor is NOT a fourth shape: `Popen` calls
|
|
`os.fsencode(cwd)`, which rejects an int. A previous version of the guard added a branch for
|
|
that case; the branch was dead — `os.fspath` already raises on an int — and deleting it left its
|
|
own test green, so it was removed rather than proved. The assertion below is what makes that
|
|
removal safe: if a future CPython starts accepting a descriptor, this goes red and says so.
|
|
"""
|
|
# A RELATIVE value, so the anchoring is actually REACHED. An absolute one short-circuits before
|
|
# `cwd` is consulted at all — an earlier draft of this test used one and passed identically for
|
|
# every shape, including shapes that crashed the guard the moment anything relative arrived.
|
|
relative = {ENV_VAR: "isolated/log"}
|
|
for cwd in (str(tmp_path), tmp_path, os.fsencode(str(tmp_path))):
|
|
assert isolation_violation(relative, cwd) is None, (
|
|
f"a relative log dir under {tmp_path} was rejected for cwd={cwd!r} — it resolves inside "
|
|
"the test's own directory, so this is the guard failing rather than judging"
|
|
)
|
|
|
|
isolated = {ENV_VAR: str(tmp_path / "log")}
|
|
assert isolation_violation(isolated, None) is None, "an absolute isolated dir was rejected"
|
|
|
|
# AND THE ANCHOR MUST BIND TO THE LAUNCH'S cwd, for every shape — not merely fail to crash.
|
|
# The same relative value judged against two different `cwd`s must give two different verdicts;
|
|
# otherwise `cwd` is being ignored, which is the round-3 defect, and the shape loop above would
|
|
# pass while it was.
|
|
shared_parent = SHARED_LOG_DIRS[0].parent.parent.parent
|
|
for shape in (str, Path, os.fsencode):
|
|
onto_shared = shape(str(shared_parent))
|
|
assert isolation_violation({ENV_VAR: ".cache/ersatztv/hook-fire"}, onto_shared) is not None, (
|
|
f"a relative log dir was cleared for cwd={onto_shared!r} even though it resolves onto a "
|
|
"shared log — the anchor is not binding to the launch's cwd for this shape"
|
|
)
|
|
assert isolation_violation({ENV_VAR: ".cache/ersatztv/hook-fire"}, shape(str(tmp_path))) is None, (
|
|
f"the same relative log dir was rejected for cwd={tmp_path} — the anchor is not binding "
|
|
"to the launch's cwd, so both verdicts come from somewhere else"
|
|
)
|
|
|
|
# THE ENV HAS SHAPES TOO. `Popen` accepts a bytes-keyed and/or bytes-valued environment on
|
|
# POSIX (verified: both launch). Unnormalised, a bytes KEY makes the lookup miss and the launch
|
|
# is rejected as carrying no isolated dir; a bytes VALUE makes `Path(...)` raise. Both are reds
|
|
# on a correctly isolated child.
|
|
log = str(tmp_path / "log")
|
|
for env in (
|
|
{os.fsencode(ENV_VAR): os.fsencode(log), b"PATH": b"/usr/bin"},
|
|
{ENV_VAR: os.fsencode(log), "PATH": "/usr/bin"},
|
|
):
|
|
assert isolation_violation(env) is None, (
|
|
f"a correctly isolated child was rejected for a legal env encoding: {env!r}"
|
|
)
|
|
# ...and the decoding must not blunt the guard: the same encoding pointed at a shared log fails.
|
|
assert isolation_violation({os.fsencode(ENV_VAR): os.fsencode(str(SHARED_LOG_DIRS[0]))}) is not None, (
|
|
"normalising the env encoding also stopped the guard seeing a shared log through it"
|
|
)
|
|
|
|
fd = os.open(str(tmp_path), os.O_RDONLY)
|
|
try:
|
|
with pytest.raises(TypeError):
|
|
subprocess.run(
|
|
["bash", "-c", "true"],
|
|
cwd=fd,
|
|
env={"PATH": os.environ["PATH"], **isolated},
|
|
capture_output=True,
|
|
timeout=60,
|
|
)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def test_REMOVING_the_pre_collection_isolation_LETS_an_IMPORT_TIME_snapshot_LEAK(tmp_path):
|
|
"""THE MUTATION PROOF FOR THE CROSS-SUITE CLAUSE — the #785 defect, reproduced and neutralised.
|
|
|
|
Both arms are a real nested pytest over an identical sandbox with a FAKE `HOME`, so nothing here
|
|
touches the machine's own log. The specimen module is the #785 shape verbatim: `{**os.environ}`
|
|
at module level, handed to a hook subprocess.
|
|
|
|
ONE clause differs between the arms — `os.environ[ENV_VAR] = _SESSION_LOG_DIR` in
|
|
`pytest_configure`. Remove it and the specimen's records land in the fake production directory;
|
|
restore it and they do not. That is what makes the pre-collection assignment load-bearing rather
|
|
than decorative, and it is the red #809 asks to witness.
|
|
|
|
WHAT IT NO LONGER TRAVERSES, since the fake directory is handed over explicitly: the sink's
|
|
`${ETV_HOOK_FIRE_LOG_DIR:-${HOME:-/tmp}/…}` fallback. The mutant arm now leaks by INHERITANCE
|
|
rather than by fallback, so deleting that expansion outright leaves both arms unchanged —
|
|
measured. That is covered instead by
|
|
`test_the_resolver_agrees_with_the_SINKS_OWN_expansion`, which reddens on the same deletion.
|
|
The proof remains sensitive to the weakening that matters here: replacing the assignment with
|
|
`os.environ.setdefault(...)` makes the control arm leak.
|
|
|
|
THE MEASUREMENT ARRANGEMENT, stated because it is a deliberate weakening: the launch guard is
|
|
disabled in BOTH arms. Left on, it would reject the specimen's launch in the mutant arm and the
|
|
comparison would be about the guard rather than about the clause under test. Both arms are
|
|
otherwise byte-identical, so the difference measured is the clause's alone.
|
|
"""
|
|
real_conftest = (REPO_ROOT / "scripts" / "tests" / "conftest.py").read_text()
|
|
install_clause = " _GUARD.install()\n"
|
|
isolate_clause = " os.environ[ENV_VAR] = _SESSION_LOG_DIR\n"
|
|
for clause in (install_clause, isolate_clause):
|
|
assert real_conftest.count(clause) == 1, (
|
|
f"the clause {clause!r} has moved or changed in conftest.py, so this proof no longer "
|
|
"targets it. RETARGET it rather than loosening the match — a mutation that silently "
|
|
"stops mutating proves nothing, which is the failure class this file exists to catch."
|
|
)
|
|
|
|
def arm(name: str, conftest_text: str) -> set[str]:
|
|
sb = tmp_path / name
|
|
(sb / "scripts" / "tests").mkdir(parents=True)
|
|
(sb / "scripts" / "__init__.py").write_text("")
|
|
(sb / "scripts" / "tests" / "__init__.py").write_text("")
|
|
shutil.copy(SINK, sb / "scripts" / "hook-fire-log.sh")
|
|
(sb / "specimen-hook.sh").write_text(
|
|
"#!/usr/bin/env bash\nset -uo pipefail\n"
|
|
f'. "{sb}/scripts/hook-fire-log.sh"\n'
|
|
'etv_hook_fire_begin specimen "" capture || true\n'
|
|
"input=$(cat)\n"
|
|
)
|
|
shutil.copy(REPO_ROOT / "scripts" / "tests" / "hook_fire_isolation.py", sb / "scripts" / "tests")
|
|
(sb / "scripts" / "tests" / "conftest.py").write_text(conftest_text)
|
|
(sb / "scripts" / "tests" / "test_specimen.py").write_text(_SPECIMEN_MODULE)
|
|
home = sb / "home"
|
|
home.mkdir()
|
|
leaked = home / ".cache" / "ersatztv" / "hook-fire"
|
|
res = subprocess.run(
|
|
[sys.executable, "-m", "pytest", "-q", "scripts/tests/test_specimen.py"],
|
|
cwd=str(sb),
|
|
capture_output=True,
|
|
text=True,
|
|
# The fake production dir is handed over EXPLICITLY, and `HOME` points at the same tree
|
|
# as a backstop. Passing it explicitly is what lets this launch satisfy the launch guard
|
|
# like every other one, and it sharpens the proof rather than weakening it: the mutant
|
|
# arm inherits this value because its `pytest_configure` no longer overwrites it, the
|
|
# control arm does not because its does. The clause under test IS that overwrite.
|
|
env={"PATH": os.environ["PATH"], "HOME": str(home), ENV_VAR: str(leaked)},
|
|
timeout=300,
|
|
)
|
|
assert res.returncode == 0, f"the {name} arm's specimen run did not pass:\n{res.stdout}\n{res.stderr}"
|
|
return {f.name for f in leaked.glob("*.jsonl")} if leaked.exists() else set()
|
|
|
|
mutant = arm("mutant", real_conftest.replace(install_clause, "").replace(isolate_clause, ""))
|
|
control = arm("control", real_conftest.replace(install_clause, ""))
|
|
|
|
# The mutant arm is also this proof's anti-vacuity check: if the specimen logged nowhere, the
|
|
# control arm's emptiness would be about a dead hook rather than about the isolation.
|
|
assert "specimen.jsonl" in mutant, (
|
|
"removing the pre-collection isolation did NOT make an import-time snapshot leak, so this "
|
|
f"proof establishes nothing about that clause. Fake-home log dir held: {sorted(mutant)}"
|
|
)
|
|
assert control == set(), (
|
|
"an import-time `{**os.environ}` snapshot still reaches the production log with the "
|
|
f"pre-collection isolation in place — it leaked {sorted(control)}. That is #785's defect, "
|
|
"and the cross-suite claim in this file is false."
|
|
)
|