429 lines
21 KiB
Python
429 lines
21 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",
|
||
"VALIDATE_RECEIPT_PURPOSE", "TASK_WORK_MODE"):
|
||
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_validation_purpose_is_not_injected_into_measured_command(self):
|
||
data = self.capture("printf '%s' \"${VALIDATE_RECEIPT_PURPOSE-unset}\"",
|
||
VALIDATE_RECEIPT_PURPOSE="resume-local")
|
||
self.assertEqual(data["output"], "unset")
|
||
|
||
def test_fixed_candidate_capture_preserves_checkout(self):
|
||
self.git("add", ".task/main.md")
|
||
self.git("commit", "-qm", "Save fixture card before validation")
|
||
head = self.git("rev-parse", "HEAD")
|
||
card = self.card.read_bytes()
|
||
index = self.git("ls-files", "--stage")
|
||
data = self.capture("printf 'fixed candidate check\\n'")
|
||
self.assertEqual(data["before"]["head"], head)
|
||
self.assertEqual(data["after"]["head"], head)
|
||
self.assertEqual(data["output"], "fixed candidate check\n")
|
||
self.assertEqual(self.git("rev-parse", "HEAD"), head)
|
||
self.assertEqual(self.git("ls-files", "--stage"), index)
|
||
self.assertEqual(self.card.read_bytes(), card)
|
||
self.assertEqual(self.git("status", "--porcelain"), "")
|
||
self.assertEqual(self.receipt.stat().st_mode & 0o777, 0o600)
|
||
|
||
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)
|
||
|
||
def test_child_receipts_record_root_and_immediate_targets(self):
|
||
self.card.write_text("# Test task\n- 分支类型：child；initiative_id：feature-alpha；root_target：main；"
|
||
"immediate_target：feature/alpha；accepting_requirements：false\n")
|
||
data = self.capture()
|
||
self.assertEqual((data["root_target"], data["immediate_target"]), ("main", "feature/alpha"))
|
||
self.assertEqual(self.import_receipt().returncode, 0)
|
||
self.assertIn(data["receipt_sha256"], self.card.read_text())
|
||
checked = subprocess.run(["python3", str(SCRIPTS / "check-evidence.py"), str(self.card)],
|
||
cwd=self.repo, capture_output=True, text=True)
|
||
self.assertNotIn("WARN", checked.stdout + checked.stderr)
|
||
|
||
def archive_card(self, status):
|
||
self.card.unlink()
|
||
archived = self.repo / ".task/archive/feature-done.md"
|
||
archived.parent.mkdir()
|
||
archived.write_text(f"# Done task\n任务主记录：PR #3 fixture；状态：{status}；owner：test\n"
|
||
"- 工作区：fixture（worktree）；分支：main\n"
|
||
"- 分支类型：child；initiative_id：feature-alpha；root_target：main；"
|
||
"immediate_target：feature/alpha；accepting_requirements：false\n")
|
||
return archived
|
||
|
||
def test_evidence_appends_to_pre_merge_closed_archived_card(self):
|
||
archived = self.archive_card("已完成（合入前闭合）")
|
||
result = self.run_evidence("printf 'archived\\n'")
|
||
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
|
||
text = archived.read_text()
|
||
self.assertIn("## 证据（Evidence）", text)
|
||
self.assertIn('"immediate_target": "feature/alpha"', text)
|
||
|
||
def test_evidence_ignores_archived_card_with_other_status(self):
|
||
archived = self.archive_card("已完成（回填）")
|
||
original = archived.read_text()
|
||
result = self.run_evidence("touch should-not-exist")
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertFalse((self.repo / "should-not-exist").exists())
|
||
self.assertEqual(archived.read_text(), original)
|
||
|
||
def test_receipts_without_target_metadata_default_to_main(self):
|
||
data = self.capture()
|
||
self.assertEqual((data["root_target"], data["immediate_target"]), ("main", "main"))
|
||
|
||
def test_receipt_target_fields_must_come_in_pairs(self):
|
||
original = self.capture()
|
||
partial = dict(original)
|
||
del partial["immediate_target"]
|
||
self.resign(partial)
|
||
self.assertNotEqual(self.import_receipt().returncode, 0)
|
||
self.assertEqual(self.card.read_text(), "# Test task\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|