weichen
b84667a697
Why: independent review found shared submodule metadata risk, quoted header leakage, stale JSON receipt blindness and missing abandon coverage. Evidence: SELF 92 governance regressions pass; the original review receipt and replay output are recorded in the task card. Impact: preserves external component worktrees, handles both evidence formats and adds targeted failure-path coverage; no product behavior changes. Refs: .task/feature-governance-reliability.md; .task/reviews/governance-reliability.json
358 lines
17 KiB
Python
358 lines
17 KiB
Python
"""Exercise the evidence recorder against disposable real Git repositories."""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
|
|
|
|
|
|
# Literal delimiters exercise header, JSON, repr and assignment boundaries.
|
|
AUTHORIZATION_FORMS = (
|
|
'Authorization: {scheme} {secret}',
|
|
'{{"Authorization":"{scheme} {secret}"}}',
|
|
"{{'Authorization': '{scheme} {secret}'}}",
|
|
'Authorization: "{scheme} {secret}"',
|
|
"Authorization: '{scheme} {secret}'",
|
|
'AUTHORIZATION = {scheme} {secret}',
|
|
'"authorization" = "{scheme} {secret}"',
|
|
"'aUtHoRiZaTiOn' = '{scheme} {secret}'",
|
|
'authorization={scheme} {secret}',
|
|
'{{ "Authorization"\t:\t"{scheme}\t{secret}", "ok": true }}',
|
|
)
|
|
|
|
|
|
def authorization_cases():
|
|
for scheme in ("Bearer", "tOkEn", "bAsIc"):
|
|
for index, form in enumerate(AUTHORIZATION_FORMS):
|
|
secret = f"syntheticAuthValue{scheme}{index}"
|
|
yield (form.format(scheme=scheme, secret=secret), secret,
|
|
form.format(scheme=scheme, secret="<redacted>"))
|
|
|
|
|
|
def authorization_command(headers):
|
|
# Use a shell quote that leaves each credential spelling intact in CMD.
|
|
args = [('"' + header + '"') if "'" in header else ("'" + header + "'")
|
|
for header in headers]
|
|
return "printf '%s\\n' " + " ".join(args)
|
|
|
|
|
|
class EvidenceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.base = Path(self.temp.name)
|
|
self.repo = self.base / "repo"
|
|
self.repo.mkdir()
|
|
self.git("init", "-b", "main")
|
|
self.git("config", "user.name", "Evidence Test")
|
|
self.git("config", "user.email", "evidence@example.invalid")
|
|
(self.repo / "tracked").write_text("initial\n")
|
|
self.git("add", "tracked")
|
|
self.git("commit", "-qm", "Initial")
|
|
self.head = self.git("rev-parse", "HEAD")
|
|
self.git("update-ref", "refs/remotes/origin/main", self.head)
|
|
self.receipt = self.base / "receipt.json"
|
|
self.card = self.repo / ".task/main.md"
|
|
self.card.parent.mkdir()
|
|
self.card.write_text("# Test task\n")
|
|
|
|
def git(self, *args, cwd=None):
|
|
return subprocess.check_output(
|
|
["git", *args], cwd=cwd or self.repo, text=True,
|
|
stderr=subprocess.DEVNULL).strip()
|
|
|
|
def run_evidence(self, command=None, cwd=None, **values):
|
|
env = os.environ.copy()
|
|
for key in ("CMD", "SRC", "RECEIPT", "IMPORT_RECEIPT", "VALIDATE_RECEIPT"):
|
|
env.pop(key, None)
|
|
if command is not None:
|
|
env["CMD"] = command
|
|
env.update({key: str(value) for key, value in values.items()})
|
|
return subprocess.run(
|
|
["bash", str(SCRIPTS / "evidence.sh")], cwd=cwd or self.repo,
|
|
env=env, capture_output=True, text=True)
|
|
|
|
def capture(self, command="printf 'hello\\n'", **kwargs):
|
|
result = self.run_evidence(command, RECEIPT=self.receipt, **kwargs)
|
|
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
|
|
self.assertTrue(self.receipt.exists(), "capture must write a receipt")
|
|
return json.loads(self.receipt.read_text())
|
|
|
|
def import_receipt(self, **kwargs):
|
|
return self.run_evidence(IMPORT_RECEIPT=self.receipt, **kwargs)
|
|
|
|
def resign(self, data):
|
|
payload = {key: value for key, value in data.items() if key != "receipt_sha256"}
|
|
data["receipt_sha256"] = hashlib.sha256(json.dumps(
|
|
payload, sort_keys=True, ensure_ascii=False, separators=(",", ":")
|
|
).encode()).hexdigest()
|
|
self.receipt.write_text(json.dumps(data))
|
|
|
|
def test_pipeline_failure_is_recorded_and_returned(self):
|
|
result = self.run_evidence("false | tail -n 1")
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertIn('"exit_code": 1', self.card.read_text())
|
|
|
|
def test_child_failure_code_is_preserved(self):
|
|
result = self.run_evidence("exit 7")
|
|
self.assertEqual(result.returncode, 7)
|
|
self.assertIn('"exit_code": 7', self.card.read_text())
|
|
|
|
def test_capture_without_card_and_import_without_rerun(self):
|
|
self.card.unlink()
|
|
data = self.capture("printf 'once\\n' >> invocation; cat invocation")
|
|
self.assertEqual(data["before"]["head"], self.head)
|
|
self.assertEqual(data["output"], "once\n")
|
|
self.assertEqual(data["source"], "SELF")
|
|
self.assertFalse(self.card.exists())
|
|
self.card.write_text("# Test task\n")
|
|
result = self.import_receipt()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual((self.repo / "invocation").read_text(), "once\n")
|
|
self.assertIn(data["receipt_sha256"], self.card.read_text())
|
|
|
|
def test_recorder_environment_does_not_leak_into_nested_commands(self):
|
|
data = self.capture("printf '%s,%s,%s' \"${CMD-unset}\" \"${RECEIPT-unset}\" \"${SRC-unset}\"", SRC="SELF")
|
|
self.assertEqual(data["output"], "unset,unset,unset")
|
|
|
|
def test_missing_card_does_not_execute(self):
|
|
self.card.unlink()
|
|
result = self.run_evidence("touch should-not-exist")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse((self.repo / "should-not-exist").exists())
|
|
|
|
def test_import_missing_card_fails_without_losing_receipt(self):
|
|
self.capture()
|
|
self.card.unlink()
|
|
result = self.import_receipt()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertTrue(self.receipt.exists())
|
|
|
|
def test_linked_worktree_import_preserves_source_identity(self):
|
|
data = self.capture()
|
|
other = self.base / "linked"
|
|
self.git("worktree", "add", "-qb", "feature/linked", str(other))
|
|
card = other / ".task/feature-linked.md"
|
|
card.parent.mkdir()
|
|
card.write_text("# Linked task\n")
|
|
result = self.import_receipt(cwd=other)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn(str(self.repo), card.read_text())
|
|
self.assertIn(data["before"]["head"], card.read_text())
|
|
|
|
def test_unrelated_repo_cannot_import(self):
|
|
self.capture()
|
|
other = self.base / "unrelated"
|
|
subprocess.check_call(["git", "clone", "-q", str(self.repo), str(other)])
|
|
card = other / ".task/main.md"
|
|
card.parent.mkdir()
|
|
card.write_text("# Other task\n")
|
|
result = self.import_receipt(cwd=other)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(card.read_text(), "# Other task\n")
|
|
|
|
def test_all_persisted_and_displayed_text_is_sanitized(self):
|
|
secrets = ["syntheticBearerValue", "syntheticEnvValue", "syntheticUrlPassword",
|
|
"syntheticQueryValue", "syntheticHeaderValue", "syntheticPlainValue"]
|
|
command = "printf '%s\\n' " + " ".join([
|
|
"'Authorization: Bearer syntheticBearerValue'",
|
|
"'GITEA_TOKEN=syntheticEnvValue'",
|
|
"'https://user:syntheticUrlPassword@example.invalid'",
|
|
"'https://example.invalid?access_token=syntheticQueryValue&ok=1'",
|
|
"'Authorization: token syntheticHeaderValue'",
|
|
"'password: syntheticPlainValue'",
|
|
])
|
|
data = self.capture(command)
|
|
imported = self.import_receipt()
|
|
all_text = self.receipt.read_text() + self.card.read_text() + imported.stdout + imported.stderr
|
|
for secret in secrets:
|
|
self.assertNotIn(secret, all_text)
|
|
self.assertIn("<redacted>", data["command"])
|
|
self.assertEqual(data["output_sha256"], hashlib.sha256(data["output"].encode()).hexdigest())
|
|
self.assertEqual(data["command_sha256"], hashlib.sha256(command.encode()).hexdigest())
|
|
|
|
def test_authorization_forms_are_redacted_in_capture_console_and_import(self):
|
|
cases = list(authorization_cases())
|
|
command = authorization_command([header for header, _, _ in cases])
|
|
result = self.run_evidence(command, RECEIPT=self.receipt)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
data = json.loads(self.receipt.read_text())
|
|
imported = self.import_receipt()
|
|
self.assertEqual(imported.returncode, 0, imported.stderr)
|
|
persisted = self.receipt.read_text() + self.card.read_text()
|
|
displayed = result.stdout + result.stderr + imported.stdout + imported.stderr
|
|
for header, secret, expected in cases:
|
|
with self.subTest(header=header):
|
|
self.assertNotIn(secret, persisted + displayed)
|
|
self.assertIn(expected, data["output"].splitlines())
|
|
self.assertEqual(data["command_sha256"], hashlib.sha256(command.encode()).hexdigest())
|
|
self.assertEqual(data["output_sha256"], hashlib.sha256(data["output"].encode()).hexdigest())
|
|
self.assertEqual(data["source"], "SELF")
|
|
self.assertEqual(data["schema"], 1)
|
|
|
|
def test_authorization_forms_are_redacted_on_direct_card_capture(self):
|
|
cases = list(authorization_cases())
|
|
command = authorization_command([header for header, _, _ in cases])
|
|
result = self.run_evidence(command)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
visible = self.card.read_text() + result.stdout + result.stderr
|
|
for header, secret, _ in cases:
|
|
with self.subTest(header=header):
|
|
self.assertNotIn(secret, visible)
|
|
self.assertIn("<redacted>", visible)
|
|
|
|
def test_import_rejects_unredacted_authorization_with_valid_checksums(self):
|
|
original = self.capture()
|
|
for header, _, _ in authorization_cases():
|
|
with self.subTest(header=header):
|
|
self.card.write_text("# Test task\n")
|
|
data = dict(original, output=header)
|
|
data["output_sha256"] = hashlib.sha256(header.encode()).hexdigest()
|
|
self.resign(data)
|
|
result = self.import_receipt()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("unredacted credential syntax", result.stderr)
|
|
self.assertEqual(self.card.read_text(), "# Test task\n")
|
|
|
|
def test_known_secret_environment_value_is_redacted_without_label(self):
|
|
data = self.capture("printf %s \"$GITEA_TOKEN\"", GITEA_TOKEN="unlabelledSyntheticCredential")
|
|
self.assertNotIn("unlabelledSyntheticCredential", self.receipt.read_text())
|
|
self.assertIn("<redacted>", data["output"])
|
|
|
|
def test_quoted_credentials_with_spaces_are_fully_redacted(self):
|
|
secret = "synthetic phrase with spaces"
|
|
data = self.capture('printf \'password="synthetic phrase with spaces"\\n\'')
|
|
self.assertNotIn(secret, self.receipt.read_text())
|
|
self.assertNotIn("phrase with spaces", data["command"] + data["output"])
|
|
|
|
def test_command_line_token_arguments_are_redacted(self):
|
|
data = self.capture("printf '%s\\n' 'tool --token syntheticArgumentValue'")
|
|
self.assertNotIn("syntheticArgumentValue", self.receipt.read_text())
|
|
self.assertIn("<redacted>", data["command"])
|
|
|
|
def test_source_cannot_be_promoted_by_environment(self):
|
|
for source in ("GATE", "REPLAY"):
|
|
with self.subTest(source=source):
|
|
result = self.run_evidence("touch should-not-exist", SRC=source)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse((self.repo / "should-not-exist").exists())
|
|
|
|
def test_tamper_is_rejected_before_card_append(self):
|
|
data = self.capture()
|
|
data["output"] = "changed output"
|
|
self.receipt.write_text(json.dumps(data))
|
|
result = self.import_receipt()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(self.card.read_text(), "# Test task\n")
|
|
|
|
def test_invalid_field_types_and_forged_source_are_rejected(self):
|
|
original = self.capture()
|
|
for field, value in (("exit_code", True), ("source", "GATE"), ("before", [])):
|
|
with self.subTest(field=field):
|
|
changed = dict(original, **{field: value})
|
|
self.resign(changed)
|
|
self.assertNotEqual(self.import_receipt().returncode, 0)
|
|
self.assertEqual(self.card.read_text(), "# Test task\n")
|
|
|
|
def test_output_digest_checked_even_if_receipt_digest_recomputed(self):
|
|
data = self.capture()
|
|
data["output_sha256"] = "0" * 64
|
|
self.resign(data)
|
|
self.assertNotEqual(self.import_receipt().returncode, 0)
|
|
|
|
def test_missing_required_snapshot_fields_are_rejected(self):
|
|
original = self.capture()
|
|
for field in ("target_head", "status"):
|
|
with self.subTest(field=field):
|
|
self.card.write_text("# Test task\n")
|
|
data = json.loads(json.dumps(original))
|
|
del data["before"][field]
|
|
self.resign(data)
|
|
result = self.import_receipt()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertNotIn("Traceback", result.stderr)
|
|
self.assertEqual(self.card.read_text(), "# Test task\n")
|
|
|
|
def test_before_after_heads_do_not_relabel_changed_command(self):
|
|
data = self.capture("printf changed > tracked; git add tracked; git commit -qm Changed")
|
|
self.assertEqual(data["before"]["head"], self.head)
|
|
self.assertEqual(data["after"]["head"], self.git("rev-parse", "HEAD"))
|
|
self.assertNotEqual(data["before"]["head"], data["after"]["head"])
|
|
|
|
def test_dirty_and_untracked_contents_have_object_digests(self):
|
|
(self.repo / "tracked").write_text("modified\n")
|
|
(self.repo / "extra").write_text("untracked\n")
|
|
data = self.capture()
|
|
self.assertIn("tracked", data["before"]["status"])
|
|
self.assertEqual(data["before"]["untracked"]["extra"], hashlib.sha256(b"untracked\n").hexdigest())
|
|
self.assertNotEqual(data["before"]["tracked_diff_sha256"], hashlib.sha256(b"").hexdigest())
|
|
|
|
def test_nested_cwd_records_untracked_files_from_repository_root(self):
|
|
nested = self.repo / "nested"
|
|
nested.mkdir()
|
|
(nested / "extra").write_text("nested source\n")
|
|
data = self.capture(cwd=nested)
|
|
self.assertEqual(data["before"]["untracked"]["nested/extra"],
|
|
hashlib.sha256(b"nested source\n").hexdigest())
|
|
|
|
def test_after_snapshot_failure_keeps_command_exit_and_output(self):
|
|
result = self.run_evidence("mv .git unavailable-git; printf captured; exit 7", RECEIPT=self.receipt)
|
|
self.assertEqual(result.returncode, 7)
|
|
self.assertTrue(self.receipt.exists())
|
|
data = json.loads(self.receipt.read_text())
|
|
self.assertIsNone(data["after"])
|
|
self.assertTrue(data["after_error"])
|
|
self.assertEqual(data["output"], "captured")
|
|
|
|
def test_import_retry_is_idempotent(self):
|
|
self.capture()
|
|
self.assertEqual(self.import_receipt().returncode, 0)
|
|
recorded = self.card.read_text()
|
|
self.assertEqual(self.import_receipt().returncode, 0)
|
|
self.assertEqual(self.card.read_text(), recorded)
|
|
|
|
def test_full_output_is_preserved_not_only_tail(self):
|
|
data = self.capture("seq 1 30", TAIL=2)
|
|
self.assertTrue(data["output"].startswith("1\n2\n"))
|
|
self.assertTrue(data["output"].endswith("29\n30\n"))
|
|
|
|
def test_receipt_write_failure_cannot_claim_success(self):
|
|
bad = self.base / "missing" / "receipt.json"
|
|
result = self.run_evidence("printf hello", RECEIPT=bad)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
|
|
def test_failed_child_exit_survives_receipt_write_failure(self):
|
|
bad = self.base / "missing" / "receipt.json"
|
|
result = self.run_evidence("exit 7", RECEIPT=bad)
|
|
self.assertEqual(result.returncode, 7)
|
|
|
|
def test_capture_does_not_overwrite_an_existing_receipt(self):
|
|
self.receipt.write_text("preserve me")
|
|
result = self.run_evidence("touch should-not-exist", RECEIPT=self.receipt)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(self.receipt.read_text(), "preserve me")
|
|
self.assertFalse((self.repo / "should-not-exist").exists())
|
|
|
|
def test_validate_precheck_checks_result_and_current_base(self):
|
|
(self.repo / "Makefile").write_text("precheck:\n\t@echo '[verdict] READY — fixture'\n")
|
|
self.capture("make precheck")
|
|
valid = self.run_evidence(VALIDATE_RECEIPT=self.receipt)
|
|
self.assertEqual(valid.returncode, 0, valid.stderr)
|
|
self.git("commit", "--allow-empty", "-qm", "Advance base")
|
|
stale = self.run_evidence(VALIDATE_RECEIPT=self.receipt)
|
|
self.assertNotEqual(stale.returncode, 0)
|
|
|
|
def test_validate_precheck_rejects_successful_arbitrary_command(self):
|
|
self.capture("printf '[verdict] READY\\n'")
|
|
result = self.run_evidence(VALIDATE_RECEIPT=self.receipt)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|