Nick
5458585e84
scripts/capture_failure.py: one command turns a raw failure payload into a sanitised fixture under tests/fixtures/production/<class>/<id>/ plus a manifest.json entry carrying the payload's sha256 (tamper-evidence for the generated regression tests). Redaction reuses #122's filer patterns (query creds, DSN userinfo, bearer, assignments) — copied with attribution because the filer module ships on the failure-auto-filing branch; parity is pinned on the filer's own test literals. After redaction an independent residual scan aborts the capture loudly (exit 3, nothing persisted) when a credential-shaped span survives — over-redaction is a shape witness, a leaked fixture is an incident.
262 lines
11 KiB
Python
262 lines
11 KiB
Python
"""Tests for scripts/capture_failure.py — production failure → sanitised fixture (#157).
|
|
|
|
The capture pipeline (plan 2026-09-07-157): a payload file + a failure class +
|
|
a source become a redacted fixture under tests/fixtures/production/<class>/<id>/
|
|
plus a manifest entry carrying the payload's sha256 (tamper-evidence — the
|
|
generated regression tests assert it). Redaction reuses #122's filer patterns;
|
|
after redaction an INDEPENDENT residual scan must abort the capture loudly
|
|
rather than persist a leaked fixture (plan design decision 1: when in doubt,
|
|
redact — a leaked fixture is an incident).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _load_module():
|
|
spec = importlib.util.spec_from_file_location("capture_failure", ROOT / "scripts" / "capture_failure.py")
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
cf = _load_module()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Redaction parity with #122's filer (shared cases)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# These literals are the shared redaction contract with
|
|
# src/rune/failure_auto_filer.py::redact (commit 9105ca8, #122). They are
|
|
# copied verbatim from that module's tests so drift on either side is caught:
|
|
# capture_failure.redact must behave identically on every case the filer
|
|
# defines. (The filer module itself is not importable from this branch — it
|
|
# ships via the failure-auto-filing branch — so parity is pinned by literals.
|
|
# When that branch merges, switch to importing it directly and asserting
|
|
# equality on these cases.)
|
|
PARITY_CASES = [
|
|
# query-string credential params
|
|
(
|
|
"GET https://forge.example.com/api?token=supersecret&page=2 failed",
|
|
"GET https://forge.example.com/api?token=<redacted>&page=2 failed",
|
|
),
|
|
# DSN userinfo (postgres/redis/mysql, with and without username)
|
|
(
|
|
"connect failed for postgres://alice:hunter2@db.example.com:5432/rune",
|
|
"connect failed for postgres://<redacted>:<redacted>@db.example.com:5432/rune",
|
|
),
|
|
("redis://:s3cr3t@cache.example.com:6379/0", "redis://<redacted>:<redacted>@cache.example.com:6379/0"),
|
|
# Authorization / bearer headers
|
|
("Authorization: Bearer eyJhbGciOi.eyJzdWIi.SflKxwRJSMeK", "Authorization: Bearer <redacted>"),
|
|
# assignment forms (note: the trailing quote of a quoted value survives —
|
|
# the span replaces only the credential value, exactly as the filer does)
|
|
(
|
|
"login failed with password='hunter2' and secret=abcdef1234",
|
|
"login failed with password=<redacted>' and secret=<redacted>",
|
|
),
|
|
# clean text is a no-op
|
|
("plain error, nothing sensitive", "plain error, nothing sensitive"),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(("raw", "expected"), PARITY_CASES, ids=lambda v: v[:40])
|
|
def test_redact_matches_filer_on_shared_cases(raw: str, expected: str) -> None:
|
|
assert cf.redact(raw) == expected
|
|
|
|
|
|
def test_redact_keeps_error_shape_visible() -> None:
|
|
out = cf.redact("RuntimeError: connection to https://user:pass@forge.example.com refused (port 443)")
|
|
assert out.startswith("RuntimeError: ")
|
|
assert "refused (port 443)" in out # non-credential detail survives
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Capture: payload → redacted fixture + manifest entry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _capture(tmp_path: Path, payload: str, failure_class: str = "ci-crash", source: str = "manual", **kwargs):
|
|
payload_file = tmp_path / "payload.txt"
|
|
payload_file.write_text(payload, encoding="utf-8")
|
|
fixtures_root = tmp_path / "fixtures"
|
|
return cf.capture(
|
|
payload_file,
|
|
failure_class=failure_class,
|
|
source=source,
|
|
fixtures_root=fixtures_root,
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
def test_capture_writes_redacted_fixture_and_manifest_entry(tmp_path: Path) -> None:
|
|
raw = "broker boom: GET https://forge.example.com/api?token=supersecret&page=2 failed"
|
|
entry = _capture(tmp_path, raw, failure_class="ci-crash", source="ci")
|
|
|
|
# derived slug: class + first words of the (redacted) first line
|
|
assert entry["id"].startswith("ci-crash-broker-boom")
|
|
fixture_file = tmp_path / "fixtures" / "ci-crash" / entry["id"] / "payload.json"
|
|
assert fixture_file.exists(), "fixture payload.json must be written under <class>/<id>/"
|
|
|
|
stored = json.loads(fixture_file.read_text(encoding="utf-8"))
|
|
assert "supersecret" not in stored["payload"], "credential must be redacted in the stored payload"
|
|
assert "token=<redacted>" in stored["payload"]
|
|
assert stored["failure_class"] == "ci-crash"
|
|
assert stored["source"] == "ci"
|
|
assert stored["date"] # capture date recorded
|
|
assert stored["original_shape"] == "text"
|
|
assert stored["redactions_applied"] >= 1
|
|
assert "query-creds" in stored["redaction_kinds"]
|
|
|
|
# manifest entry carries the sha256 of the stored payload text
|
|
manifest = json.loads((tmp_path / "fixtures" / "manifest.json").read_text(encoding="utf-8"))
|
|
ids = [e["id"] for e in manifest["fixtures"]]
|
|
assert entry["id"] in ids
|
|
recorded = next(e for e in manifest["fixtures"] if e["id"] == entry["id"])
|
|
assert recorded["sha256"] == hashlib.sha256(stored["payload"].encode("utf-8")).hexdigest()
|
|
assert recorded["class"] == "ci-crash"
|
|
assert recorded["source"] == "ci"
|
|
|
|
|
|
def test_capture_records_json_original_shape(tmp_path: Path) -> None:
|
|
raw = '{"error": "KeyError: False", "step": "convergence"}'
|
|
_capture(tmp_path, raw, fixture_id="json-shape-case")
|
|
fixture_file = tmp_path / "fixtures" / "ci-crash" / "json-shape-case" / "payload.json"
|
|
stored = json.loads(fixture_file.read_text(encoding="utf-8"))
|
|
assert stored["original_shape"] == "json"
|
|
assert stored["payload"] == raw # JSON payloads are stored verbatim (redaction still applied)
|
|
|
|
|
|
def test_capture_is_idempotent_on_recapture(tmp_path: Path) -> None:
|
|
raw = "boom with password=hunter2 in the log"
|
|
_capture(tmp_path, raw, fixture_id="same-id")
|
|
_capture(tmp_path, raw, fixture_id="same-id")
|
|
manifest = json.loads((tmp_path / "fixtures" / "manifest.json").read_text(encoding="utf-8"))
|
|
assert [e["id"] for e in manifest["fixtures"]].count("same-id") == 1, "re-capture updates, never duplicates"
|
|
|
|
|
|
def test_capture_clean_payload_records_zero_redactions(tmp_path: Path) -> None:
|
|
entry = _capture(tmp_path, "harmless failure text", fixture_id="clean-case")
|
|
assert entry["redactions_applied"] == 0
|
|
assert entry["redaction_kinds"] == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Residual-secret abort (the loud backstop)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_residual_scan_flags_surviving_credentials() -> None:
|
|
# A credential shape the redaction patterns cannot clean (embedded quote
|
|
# defeats the assignment span) must be FLAGGED by the independent scan.
|
|
assert cf.residual_secrets("config had secret=ab'cd embedded") != []
|
|
assert cf.residual_secrets("password=abcd") != []
|
|
assert cf.residual_secrets("https://x.example/?key=zzz") != []
|
|
assert cf.residual_secrets("Authorization: Bearer eyJ.abc") != []
|
|
assert cf.residual_secrets("postgres://alice:hunter2@db/x") != []
|
|
|
|
|
|
def test_residual_scan_is_quiet_on_redacted_text() -> None:
|
|
redacted = cf.redact(
|
|
"token=supersecret&page=2 postgres://alice:hunter2@db/x Authorization: Bearer eyJ.abc password=hunter2"
|
|
)
|
|
assert cf.residual_secrets(redacted) == [], "redaction output must pass the residual scan"
|
|
|
|
|
|
def test_capture_aborts_loudly_on_residual_secret(tmp_path: Path) -> None:
|
|
raw = "config had secret=ab'cd embedded" # survives redaction
|
|
with pytest.raises(cf.ResidualSecretError) as excinfo:
|
|
_capture(tmp_path, raw, fixture_id="leaky-case")
|
|
assert "secret" in str(excinfo.value).lower(), "abort message must name the residual secret"
|
|
|
|
# nothing persisted: no fixture directory, no manifest entry
|
|
assert not (tmp_path / "fixtures" / "ci-crash" / "leaky-case").exists()
|
|
manifest_file = tmp_path / "fixtures" / "manifest.json"
|
|
if manifest_file.exists():
|
|
manifest = json.loads(manifest_file.read_text(encoding="utf-8"))
|
|
assert "leaky-case" not in [e["id"] for e in manifest["fixtures"]]
|
|
|
|
|
|
def test_capture_rejects_unknown_source(tmp_path: Path) -> None:
|
|
with pytest.raises(ValueError):
|
|
_capture(tmp_path, "boom", source="twitter")
|
|
|
|
|
|
def test_capture_rejects_unsafe_class_or_id(tmp_path: Path) -> None:
|
|
with pytest.raises(ValueError):
|
|
_capture(tmp_path, "boom", failure_class="../escape")
|
|
with pytest.raises(ValueError):
|
|
_capture(tmp_path, "boom", fixture_id="../escape")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI surface
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cli_captures_and_exits_zero(tmp_path: Path, capsys) -> None:
|
|
payload_file = tmp_path / "raw.txt"
|
|
payload_file.write_text("boom: token=supersecret leaked", encoding="utf-8")
|
|
fixtures_root = tmp_path / "fixtures"
|
|
code = cf.main(
|
|
[
|
|
str(payload_file),
|
|
"--class",
|
|
"ci-crash",
|
|
"--source",
|
|
"ci",
|
|
"--id",
|
|
"cli-case",
|
|
"--fixtures-root",
|
|
str(fixtures_root),
|
|
]
|
|
)
|
|
assert code == 0
|
|
assert (fixtures_root / "ci-crash" / "cli-case" / "payload.json").exists()
|
|
out = capsys.readouterr().out
|
|
assert "cli-case" in out
|
|
|
|
|
|
def test_cli_aborts_nonzero_on_residual_secret(tmp_path: Path, capsys) -> None:
|
|
payload_file = tmp_path / "raw.txt"
|
|
payload_file.write_text("config had secret=ab'cd embedded", encoding="utf-8")
|
|
code = cf.main(
|
|
[
|
|
str(payload_file),
|
|
"--class",
|
|
"ci-crash",
|
|
"--source",
|
|
"manual",
|
|
"--id",
|
|
"leaky-cli",
|
|
"--fixtures-root",
|
|
str(tmp_path / "fixtures"),
|
|
]
|
|
)
|
|
assert code == cf.EXIT_RESIDUAL_SECRET, "residual secret must abort with a distinct exit code"
|
|
assert "secret" in capsys.readouterr().err.lower()
|
|
|
|
|
|
def test_cli_missing_payload_file_exits_nonzero(tmp_path: Path) -> None:
|
|
code = cf.main(
|
|
[
|
|
str(tmp_path / "does-not-exist.txt"),
|
|
"--class",
|
|
"ci-crash",
|
|
"--source",
|
|
"manual",
|
|
"--fixtures-root",
|
|
str(tmp_path / "fixtures"),
|
|
]
|
|
)
|
|
assert code == cf.EXIT_USAGE
|