From f4416b48ea065dab418eeb6dea54406429087474 Mon Sep 17 00:00:00 2001 From: stevenmih Date: Tue, 8 Sep 2026 21:01:29 -0700 Subject: [PATCH] test(gates): end-to-end tests proving check_append_only.main() rejects broken appends Adds TestCheckAppendOnlyEndToEnd to test_registry_gates.py. The three cases call check_append_only.main() against a real temporary git repository (the same execution path CI uses), not just the extracted comparison logic: - test_tampered_entry_is_rejected: commits a registry entry, then replaces its merkle_root with a different hash (broken hash-chain link / wrong prev pointer) and asserts main() exits 1. - test_deleted_entry_is_rejected: commits two entries, then removes the first (omission attack) and asserts main() exits 1. - test_pure_append_passes: appends a new entry to an existing file and asserts main() exits 0, confirming the check is not over-broad. All three pass locally. The import of check_append_only at the top of the module is also new; the two existing test classes are unchanged. --- tests/test_registry_gates.py | 163 +++++++++++++++++++++++++++++++++++ 1 file changed, 163 insertions(+) diff --git a/tests/test_registry_gates.py b/tests/test_registry_gates.py index 1faaca4..32b06bb 100644 --- a/tests/test_registry_gates.py +++ b/tests/test_registry_gates.py @@ -9,6 +9,7 @@ from __future__ import annotations import json +import subprocess import sys import tempfile import unittest @@ -17,6 +18,7 @@ REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "tools")) +import check_append_only # noqa: E402 import validate_registry # noqa: E402 @@ -189,5 +191,166 @@ def test_new_file_no_old_lines(self): self.assertFalse(violation) +class TestCheckAppendOnlyEndToEnd(unittest.TestCase): + """End-to-end tests for check_append_only.main() against a real git repository. + + The simulation in TestCheckAppendOnly exercises the comparison logic in + isolation. These tests call main() the same way CI does — with an actual + git repository and a real BASE_SHA — so they prove the tool itself catches + each violation, not just that the logic is sound in the abstract. + + Three cases mirror the three things CI guards against: + (a) a committed entry whose content was changed after the fact + (tampered hash-chain link / wrong content), + (b) a line removed from the registry file (omission attack), and + (c) a valid pure-append, which must pass so the check is not + over-broad. + """ + + # --------------------------------------------------------------------------- + # Helpers + # --------------------------------------------------------------------------- + + @staticmethod + def _git(repo: Path, *args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["git", *args], + cwd=repo, + capture_output=True, + text=True, + ) + + def _make_repo(self, tmp: Path) -> tuple[Path, str]: + """Initialise a bare git repo with one committed registry entry. + + Returns the repo root and the SHA of the initial commit (BASE_SHA for + the append-only check). + """ + repo = tmp / "repo" + repo.mkdir() + + self._git(repo, "init") + self._git(repo, "config", "user.email", "ci@example.com") + self._git(repo, "config", "user.name", "CI Test") + self._git(repo, "commit", "--allow-empty", "-m", "root") + + ndjson = repo / "registry" / "2026" / "06" / "12.ndjson" + ndjson.parent.mkdir(parents=True) + entry = json.dumps(VALID_ENTRY) + ndjson.write_text(entry + "\n", encoding="utf-8") + + self._git(repo, "add", "registry/") + self._git(repo, "commit", "-m", "add registry entry") + + base_result = self._git(repo, "rev-parse", "HEAD") + base_sha = base_result.stdout.strip() + return repo, base_sha + + def _run_check(self, repo: Path, base_sha: str) -> int: + """Call check_append_only.main() with the given repo as cwd. + + check_append_only resolves REPO_ROOT from __file__ (the tools/ + directory), so we monkey-patch it to point at the temp repo instead. + """ + original_root = check_append_only.REPO_ROOT + try: + check_append_only.REPO_ROOT = repo + return check_append_only.main([base_sha]) + finally: + check_append_only.REPO_ROOT = original_root + + # --------------------------------------------------------------------------- + # (c) Pure append: a new line added — must PASS + # --------------------------------------------------------------------------- + + def test_pure_append_passes(self): + """Appending a new line to an existing registry file is allowed.""" + with tempfile.TemporaryDirectory() as tmp_s: + tmp = Path(tmp_s) + repo, base_sha = self._make_repo(tmp) + + ndjson = repo / "registry" / "2026" / "06" / "12.ndjson" + second = {**VALID_ENTRY, "ts": "2026-06-12T20:00:00Z", "batch_id": "2026-06-12-002"} + with ndjson.open("a", encoding="utf-8") as f: + f.write(json.dumps(second) + "\n") + + self._git(repo, "add", "registry/") + self._git(repo, "commit", "-m", "append second entry") + + new_sha = self._git(repo, "rev-parse", "HEAD").stdout.strip() + exit_code = self._run_check(repo, base_sha) + self.assertEqual(exit_code, 0, "a pure append must be accepted") + + # --------------------------------------------------------------------------- + # (a) Tampered entry: existing line content changed — must FAIL + # --------------------------------------------------------------------------- + + def test_tampered_entry_is_rejected(self): + """Modifying a committed registry entry (e.g. swapping the merkle_root) + is a broken-append-only violation and must be caught. + + This is the canonical broken-hash-chain case: the previously committed + merkle_root value — which anchors the Merkle leaf content — has been + replaced with a different hash. The append-only check must reject it + and exit non-zero. + """ + with tempfile.TemporaryDirectory() as tmp_s: + tmp = Path(tmp_s) + repo, base_sha = self._make_repo(tmp) + + ndjson = repo / "registry" / "2026" / "06" / "12.ndjson" + # Replace the committed entry with one whose merkle_root differs — + # simulating a post-hoc substitution of a hash-chain link. + tampered = {**VALID_ENTRY, "merkle_root": "sha256:" + "ff" * 32} + ndjson.write_text(json.dumps(tampered) + "\n", encoding="utf-8") + + self._git(repo, "add", "registry/") + self._git(repo, "commit", "-m", "tamper: swap merkle_root in existing entry") + + exit_code = self._run_check(repo, base_sha) + self.assertEqual( + exit_code, 1, + "a tampered entry (wrong merkle_root) must be caught as an " + "append-only violation", + ) + + # --------------------------------------------------------------------------- + # (b) Omission: a line deleted from the registry — must FAIL + # --------------------------------------------------------------------------- + + def test_deleted_entry_is_rejected(self): + """Removing a previously committed registry line is an append-only + violation and must be caught. + + An attacker with write access to the NDJSON file could drop an entry to + hide it from the record. The check must reject this and exit non-zero. + """ + with tempfile.TemporaryDirectory() as tmp_s: + tmp = Path(tmp_s) + repo, base_sha = self._make_repo(tmp) + + # Commit a second entry so the file has two lines. + ndjson = repo / "registry" / "2026" / "06" / "12.ndjson" + second = {**VALID_ENTRY, "ts": "2026-06-12T20:00:00Z", "batch_id": "2026-06-12-002"} + with ndjson.open("a", encoding="utf-8") as f: + f.write(json.dumps(second) + "\n") + self._git(repo, "add", "registry/") + self._git(repo, "commit", "-m", "append second entry") + + # Now record this two-line state as the new base. + two_entry_sha = self._git(repo, "rev-parse", "HEAD").stdout.strip() + + # Delete the first entry, leaving only the second. + ndjson.write_text(json.dumps(second) + "\n", encoding="utf-8") + self._git(repo, "add", "registry/") + self._git(repo, "commit", "-m", "omission: delete first entry") + + exit_code = self._run_check(repo, two_entry_sha) + self.assertEqual( + exit_code, 1, + "deleting a committed entry must be caught as an append-only violation", + ) + + if __name__ == "__main__": unittest.main()