diff --git a/src/flightdeck/ledger.py b/src/flightdeck/ledger.py index e1d81e1..45ddcfb 100644 --- a/src/flightdeck/ledger.py +++ b/src/flightdeck/ledger.py @@ -14,11 +14,13 @@ import hashlib import json +from collections.abc import Iterator from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path GENESIS = "0" * 64 +_ENTRY_FIELDS = ("seq", "at", "event", "data", "prev", "hash") def _canonical(data: dict) -> str: @@ -30,6 +32,22 @@ def _entry_hash(seq: int, at: str, event: str, data: dict, prev: str) -> str: return hashlib.sha256(payload.encode("utf-8")).hexdigest() +def _entry_shape_error(entry: object) -> str | None: + if not isinstance(entry, dict): + return "invalid entry: expected a JSON object" + missing = [field for field in _ENTRY_FIELDS if field not in entry] + if missing: + return f"invalid entry: missing field(s): {', '.join(missing)}" + if type(entry["seq"]) is not int: + return "invalid entry: field 'seq' must be an integer" + for field in ("at", "event", "prev", "hash"): + if not isinstance(entry[field], str): + return f"invalid entry: field '{field}' must be a string" + if not isinstance(entry["data"], dict): + return "invalid entry: field 'data' must be a JSON object" + return None + + @dataclass class VerifyResult: entries: int @@ -75,28 +93,59 @@ def append(self, event: str, data: dict, at: datetime | None = None) -> dict: return entry def entries(self) -> list[dict]: + return [json.loads(line.decode("utf-8")) for line in self._entry_lines()] + + def _entry_lines(self) -> Iterator[bytes]: if not self.path.exists(): - return [] - out = [] - with self.path.open(encoding="utf-8") as fh: - for line in fh: - line = line.strip() - if line: - out.append(json.loads(line)) - return out + return + with self.path.open("rb") as fh: + for raw_line in fh: + if line := raw_line.strip(): + yield line def verify(self) -> VerifyResult: """Re-walk the chain: sequence must be gapless from 0, each prev must match the previous hash, each hash must recompute. First break wins.""" - entries = self.entries() + entry_count = 0 + failure: tuple[int, str] | None = None prev = GENESIS - for index, entry in enumerate(entries): + for index, line in enumerate(self._entry_lines()): + entry_count += 1 + if failure is not None: + continue + try: + text = line.decode("utf-8") + except UnicodeDecodeError as exc: + failure = (index, f"invalid UTF-8: {exc.reason}") + continue + try: + entry = json.loads(text) + except json.JSONDecodeError as exc: + failure = (index, f"invalid JSON: {exc.msg}") + continue + except (RecursionError, ValueError) as exc: + failure = (index, f"invalid JSON: {exc}") + continue + if shape_error := _entry_shape_error(entry): + failure = (index, shape_error) + continue if entry["seq"] != index: - return VerifyResult(len(entries), False, entry["seq"], "sequence gap or reorder") + failure = (entry["seq"], "sequence gap or reorder") + continue if entry["prev"] != prev: - return VerifyResult(len(entries), False, entry["seq"], "broken chain link") - expected = _entry_hash(entry["seq"], entry["at"], entry["event"], entry["data"], entry["prev"]) + failure = (entry["seq"], "broken chain link") + continue + try: + expected = _entry_hash( + entry["seq"], entry["at"], entry["event"], entry["data"], entry["prev"] + ) + except (RecursionError, UnicodeEncodeError, ValueError) as exc: + failure = (entry["seq"], f"invalid entry: {exc}") + continue if entry["hash"] != expected: - return VerifyResult(len(entries), False, entry["seq"], "entry hash mismatch") + failure = (entry["seq"], "entry hash mismatch") + continue prev = entry["hash"] - return VerifyResult(len(entries), True) + if failure is not None: + return VerifyResult(entry_count, False, *failure) + return VerifyResult(entry_count, True) diff --git a/tests/test_cli_and_html.py b/tests/test_cli_and_html.py index 96b3d92..3de33b6 100644 --- a/tests/test_cli_and_html.py +++ b/tests/test_cli_and_html.py @@ -4,6 +4,7 @@ import json from pathlib import Path +import pytest import yaml from typer.testing import CliRunner @@ -122,6 +123,43 @@ def test_run_feedback_report_loop_offline(tmp_path): assert "chain intact" in result.output +@pytest.mark.parametrize( + ("corruption", "reason"), + [ + ("shape", "invalid entry"), + ("utf16", "invalid UTF-8"), + ("huge_integer", "invalid JSON"), + ("surrogate", "invalid entry"), + ], +) +def test_audit_and_report_survive_an_invalid_ledger_entry(tmp_path, corruption, reason): + root = _init(tmp_path) + org = load_org(root) + if corruption == "utf16": + Ledger(org.ledger_path).append("event", {"n": 0}) + record = org.ledger_path.read_text(encoding="utf-8").strip() + org.ledger_path.write_bytes(record.encode("utf-16")) + elif corruption == "huge_integer": + org.ledger_path.write_text('{"seq":' + "9" * 5000 + "}\n", encoding="utf-8") + elif corruption == "surrogate": + org.ledger_path.write_text( + '{"seq":0,"at":"2026-08-29T20:00:00+00:00","event":"\\ud800",' + '"data":{},"prev":"' + "0" * 64 + '","hash":"x"}\n', + encoding="utf-8", + ) + else: + org.ledger_path.write_text("[]\n", encoding="utf-8") + + audit = invoke("audit", "verify", "--dir", str(root)) + assert audit.exit_code == 1 + assert "INTEGRITY FAILURE" in audit.output + assert reason in audit.output + + report = invoke("report", "--dir", str(root)) + assert report.exit_code == 0, report.output + assert "LEDGER INTEGRITY FAILED" in report.output + + def test_report_json_with_html_keeps_stdout_valid_json(tmp_path): # --json is documented "for pipelines": `report --json --html x > data.json` must # leave stdout a single valid JSON document. The "wrote dashboard/statement" diff --git a/tests/test_ledger.py b/tests/test_ledger.py index f14248a..8d965e2 100644 --- a/tests/test_ledger.py +++ b/tests/test_ledger.py @@ -1,5 +1,7 @@ import json +import pytest + from flightdeck.ledger import Ledger @@ -16,6 +18,108 @@ def test_empty_ledger_verifies(tmp_path): assert result.ok and result.entries == 0 +def test_malformed_json_is_reported_as_integrity_failure(tmp_path): + path = tmp_path / "ledger.jsonl" + Ledger(path).append("event", {"n": 0}) + path.write_text(path.read_text(encoding="utf-8") + '\n{"seq": 1\n', encoding="utf-8") + + result = Ledger(path).verify() + + assert not result.ok + assert result.entries == 2 + assert result.broken_at == 1 + assert result.reason.startswith("invalid JSON:") + + +@pytest.mark.parametrize( + "entry", + [ + None, + [], + {"seq": 0}, + { + "seq": 0, + "at": "2026-08-29T20:00:00+00:00", + "event": "event", + "data": [], + "prev": "0" * 64, + "hash": "0" * 64, + }, + ], +) +def test_invalid_entry_shape_is_reported_as_integrity_failure(tmp_path, entry): + path = tmp_path / "ledger.jsonl" + path.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + result = Ledger(path).verify() + + assert not result.ok + assert result.entries == 1 + assert result.broken_at == 0 + assert result.reason.startswith("invalid entry:") + + +def test_invalid_utf8_is_reported_as_integrity_failure(tmp_path): + path = tmp_path / "ledger.jsonl" + path.write_bytes(b"\xff\n") + + result = Ledger(path).verify() + + assert not result.ok + assert result.entries == 1 + assert result.broken_at == 0 + assert result.reason.startswith("invalid UTF-8:") + + +def test_utf16_record_is_not_accepted_as_utf8(tmp_path): + path = tmp_path / "ledger.jsonl" + Ledger(path).append("event", {"n": 0}) + record = path.read_text(encoding="utf-8").strip() + path.write_bytes(record.encode("utf-16")) + + result = Ledger(path).verify() + + assert not result.ok + assert result.broken_at == 0 + assert result.reason.startswith("invalid UTF-8:") + + +def test_entries_rejects_utf16_records(tmp_path): + path = tmp_path / "ledger.jsonl" + Ledger(path).append("event", {"n": 0}) + record = path.read_text(encoding="utf-8").strip() + path.write_bytes(record.encode("utf-16")) + + with pytest.raises(UnicodeDecodeError): + Ledger(path).entries() + + +def test_oversized_json_integer_is_reported_as_integrity_failure(tmp_path): + path = tmp_path / "ledger.jsonl" + path.write_text('{"seq":' + "9" * 5000 + "}\n", encoding="utf-8") + + result = Ledger(path).verify() + + assert not result.ok + assert result.broken_at == 0 + assert result.reason.startswith("invalid JSON:") + + +def test_invalid_unicode_is_reported_as_integrity_failure(tmp_path): + path = tmp_path / "ledger.jsonl" + path.write_text( + '{"seq":0,"at":"2026-08-29T20:00:00+00:00","event":"\\ud800",' + '"data":{},"prev":"' + "0" * 64 + '","hash":"x"}\n', + encoding="utf-8", + ) + + result = Ledger(path).verify() + + assert not result.ok + assert result.broken_at == 0 + assert result.reason.startswith("invalid entry:") + + def test_tampered_data_breaks_at_that_entry(tmp_path): path = tmp_path / "ledger.jsonl" ledger = Ledger(path)