From a1dd519ae7ba38f56065f7cea0a64b81b0583044 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Fri, 5 Jun 2026 12:11:57 -0400 Subject: [PATCH] attest: per-entity FingerprintChain on top of audit_events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit commit_fingerprint() wraps append_audit with event_type='attest_fingerprint' and subject_root=entity_id. Per-entity chain hash is derived at query time by folding sha256(prev_chain || event_hash) over the entity's filtered event sequence, with genesis prev_chain = ZERO_HASH. Two layers of tamper-evidence: mutating any audit row breaks the global event_hash chain (existing invariant); mutating, reordering, or deleting a fingerprint breaks the derived per-entity chain hash. No new SQL table — head-pointer optimization deferred until long chains warrant it. 15 chain tests + 13 fingerprint tests = 28 pass. --- arborist/attest/__init__.py | 17 +++- arborist/attest/chain.py | 167 +++++++++++++++++++++++++++++++++ tests/test_attest_chain.py | 179 ++++++++++++++++++++++++++++++++++++ 3 files changed, 362 insertions(+), 1 deletion(-) create mode 100644 arborist/attest/chain.py create mode 100644 tests/test_attest_chain.py diff --git a/arborist/attest/__init__.py b/arborist/attest/__init__.py index 0adf5da..0d8d82c 100644 --- a/arborist/attest/__init__.py +++ b/arborist/attest/__init__.py @@ -17,6 +17,12 @@ fingerprint distributions are deterministic functions of their sample inputs. No model is in the proof path. """ +from arborist.attest.chain import ( + commit_fingerprint, + entity_chain_hash, + entity_chain_iter, + verify_entity_chain, +) from arborist.attest.fingerprint import ( Fingerprint, canonical_bytes, @@ -24,4 +30,13 @@ from arborist.attest.fingerprint import ( leaf_hash, ) -__all__ = ["Fingerprint", "canonical_bytes", "from_samples", "leaf_hash"] +__all__ = [ + "Fingerprint", + "canonical_bytes", + "commit_fingerprint", + "entity_chain_hash", + "entity_chain_iter", + "from_samples", + "leaf_hash", + "verify_entity_chain", +] diff --git a/arborist/attest/chain.py b/arborist/attest/chain.py new file mode 100644 index 0000000..480b693 --- /dev/null +++ b/arborist/attest/chain.py @@ -0,0 +1,167 @@ +"""Per-entity fingerprint chain on top of `audit_events`. + +A FingerprintChain piggybacks on the existing append-only audit chain +via `arborist.store.append_audit`. Each fingerprint commit becomes +one `audit_events` row with `event_type='attest_fingerprint'` and +`subject_root=fp.entity_id`. + +Per-entity chain hash is derived at query time by folding +``sha256(prev_chain || event_hash)`` over the entity's filtered +event sequence, with the genesis ``prev_chain`` fixed at 32 zero +bytes (``arborist.merkle.ZERO_HASH``). Two layers of +tamper-evidence: + +1. Mutating any audit row breaks the global ``event_hash`` chain + (existing arborist invariant). +2. Mutating, reordering, or deleting a fingerprint breaks the + derived per-entity chain hash. + +No new SQL table at this phase. The per-entity head is computable +from existing columns; adding a head-pointer table is a forward +optimization for entities with very long chains. +""" + +from __future__ import annotations + +import hashlib +import sqlite3 +from dataclasses import asdict +from typing import Iterator + +from arborist.attest.fingerprint import Fingerprint, SCHEMA_VERSION +from arborist.merkle import ZERO_HASH +from arborist.store import append_audit + +EVENT_TYPE = "attest_fingerprint" +GENESIS_PREV_CHAIN = ZERO_HASH.hex() + + +def _fold(prev_chain_hex: str, event_hash_hex: str) -> str: + """One step of the per-entity chain hash.""" + h = hashlib.sha256() + h.update(bytes.fromhex(prev_chain_hex)) + h.update(bytes.fromhex(event_hash_hex)) + return h.hexdigest() + + +def commit_fingerprint( + conn: sqlite3.Connection, + fp: Fingerprint, + *, + ts: int | None = None, +) -> dict: + """Append a Fingerprint to the entity's chain. + + Returns ``{"event_hash", "entity_chain_hash", "seq"}`` where + ``seq`` is the 1-indexed per-entity sequence number after this + insert. + + Raises ``ValueError`` if the fingerprint's ``schema_version`` + does not match this module's pinned ``SCHEMA_VERSION``. This + guards against silently committing fingerprints encoded under + a future schema while the chain math still uses v1 + assumptions. + """ + if fp.schema_version != SCHEMA_VERSION: + raise ValueError( + f"fingerprint schema_version {fp.schema_version!r} != " + f"chain SCHEMA_VERSION {SCHEMA_VERSION!r}" + ) + + body = asdict(fp) + body["error_taxonomy"] = dict(body["error_taxonomy"]) + body["stress_params"] = dict(body["stress_params"]) + + event_hash = append_audit( + conn, + event_type=EVENT_TYPE, + body=body, + subject_root=fp.entity_id, + ts=ts, + ) + + new_chain_hash = entity_chain_hash(conn, fp.entity_id) + assert new_chain_hash is not None, "chain just got an event; must not be None" + seq = _entity_count(conn, fp.entity_id) + return { + "event_hash": event_hash, + "entity_chain_hash": new_chain_hash, + "seq": seq, + } + + +def _entity_count(conn: sqlite3.Connection, entity_id: str) -> int: + row = conn.execute( + "SELECT COUNT(*) FROM audit_events " + "WHERE event_type=? AND subject_root=?", + (EVENT_TYPE, entity_id), + ).fetchone() + return int(row[0]) + + +def _iter_entity_events( + conn: sqlite3.Connection, entity_id: str +) -> Iterator[tuple[int, str, str]]: + """Yield ``(seq, event_hash, body_json)`` for the entity in chain order.""" + cur = conn.execute( + "SELECT seq, event_hash, body FROM audit_events " + "WHERE event_type=? AND subject_root=? ORDER BY seq", + (EVENT_TYPE, entity_id), + ) + for row in cur: + yield int(row[0]), row[1], row[2] + + +def entity_chain_hash(conn: sqlite3.Connection, entity_id: str) -> str | None: + """Per-entity chain hash, derived from the entity's audit-event sequence. + + Returns None if the entity has no fingerprints committed yet. + """ + chain = GENESIS_PREV_CHAIN + any_event = False + for _seq, event_hash, _body in _iter_entity_events(conn, entity_id): + chain = _fold(chain, event_hash) + any_event = True + return chain if any_event else None + + +def entity_chain_iter( + conn: sqlite3.Connection, entity_id: str +) -> Iterator[tuple[int, str, Fingerprint]]: + """Yield ``(entity_seq, event_hash, Fingerprint)`` in chain order. + + ``entity_seq`` is 1-indexed per entity (NOT the global audit + ``seq``). Reconstructs the Fingerprint from canonical body JSON. + """ + import json + + n = 0 + for _global_seq, event_hash, body_json in _iter_entity_events(conn, entity_id): + n += 1 + body = json.loads(body_json) + fp = Fingerprint( + entity_id=body["entity_id"], + session_id=body["session_id"], + timestamp_utc=body["timestamp_utc"], + domain=body["domain"], + host_hash=body["host_hash"], + count=int(body["count"]), + mean_us=int(body["mean_us"]), + std_us=int(body["std_us"]), + p50_us=int(body["p50_us"]), + p90_us=int(body["p90_us"]), + p99_us=int(body["p99_us"]), + skew_q10000=int(body["skew_q10000"]), + error_taxonomy={k: int(v) for k, v in body.get("error_taxonomy", {}).items()}, + stress_params={k: int(v) for k, v in body.get("stress_params", {}).items()}, + schema_version=body["schema_version"], + ) + yield n, event_hash, fp + + +def verify_entity_chain( + conn: sqlite3.Connection, entity_id: str, claimed_chain_hash: str +) -> bool: + """Return True iff the derived chain hash matches the claim.""" + actual = entity_chain_hash(conn, entity_id) + return actual is not None and actual == claimed_chain_hash diff --git a/tests/test_attest_chain.py b/tests/test_attest_chain.py new file mode 100644 index 0000000..b3e7532 --- /dev/null +++ b/tests/test_attest_chain.py @@ -0,0 +1,179 @@ +"""Tests for `arborist.attest.chain`.""" + +from __future__ import annotations + +import hashlib + +import pytest + +from arborist.attest import ( + Fingerprint, + commit_fingerprint, + entity_chain_hash, + entity_chain_iter, + verify_entity_chain, +) +from arborist.attest.chain import ( + EVENT_TYPE, + GENESIS_PREV_CHAIN, + _fold, +) +from arborist.attest.fingerprint import SCHEMA_VERSION +from arborist.store import connect + + +def _fp(entity_id="alice", session_id="s-1", mean_us=1000): + return Fingerprint( + entity_id=entity_id, + session_id=session_id, + timestamp_utc="2026-06-05T15:30:00Z", + domain="python-debug", + host_hash="0" * 64, + count=3, + mean_us=mean_us, + std_us=100, + p50_us=mean_us, + p90_us=mean_us + 100, + p99_us=mean_us + 200, + skew_q10000=0, + error_taxonomy={}, + stress_params={}, + schema_version=SCHEMA_VERSION, + ) + + +def test_commit_returns_event_and_chain_hash(tmp_path): + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp()) + assert set(out) == {"event_hash", "entity_chain_hash", "seq"} + assert len(out["event_hash"]) == 64 + assert len(out["entity_chain_hash"]) == 64 + assert out["seq"] == 1 + + +def test_first_chain_hash_folds_genesis_and_event(tmp_path): + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp()) + expected = _fold(GENESIS_PREV_CHAIN, out["event_hash"]) + assert out["entity_chain_hash"] == expected + + +def test_second_chain_hash_folds_prev_and_event(tmp_path): + conn = connect(tmp_path / "t.db") + out1 = commit_fingerprint(conn, _fp(session_id="s-1")) + out2 = commit_fingerprint(conn, _fp(session_id="s-2", mean_us=1100)) + assert out2["seq"] == 2 + expected = _fold(out1["entity_chain_hash"], out2["event_hash"]) + assert out2["entity_chain_hash"] == expected + + +def test_entity_chain_hash_independent_across_entities(tmp_path): + conn = connect(tmp_path / "t.db") + a1 = commit_fingerprint(conn, _fp(entity_id="alice", session_id="a-1")) + b1 = commit_fingerprint(conn, _fp(entity_id="bob", session_id="b-1")) + assert a1["entity_chain_hash"] != b1["entity_chain_hash"] + assert a1["seq"] == 1 and b1["seq"] == 1 + + +def test_entity_chain_hash_independent_of_global_audit_position(tmp_path): + """Per-entity chain only counts that entity's attest events. + + Even though alice and bob's events are interleaved in the global + audit_events table, alice's chain hash should be a function only + of alice's filtered event sequence. + """ + conn = connect(tmp_path / "t.db") + a1 = commit_fingerprint(conn, _fp(entity_id="alice", session_id="a-1")) + _b1 = commit_fingerprint(conn, _fp(entity_id="bob", session_id="b-1")) + a2 = commit_fingerprint(conn, _fp(entity_id="alice", session_id="a-2", mean_us=1100)) + expected = _fold(a1["entity_chain_hash"], a2["event_hash"]) + assert a2["entity_chain_hash"] == expected + assert a2["seq"] == 2 # second alice event, not third global + + +def test_entity_chain_iter_yields_in_order(tmp_path): + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp(session_id="s-1", mean_us=1000)) + commit_fingerprint(conn, _fp(session_id="s-2", mean_us=1100)) + commit_fingerprint(conn, _fp(session_id="s-3", mean_us=1200)) + seen = list(entity_chain_iter(conn, "alice")) + assert [n for n, _h, _fp in seen] == [1, 2, 3] + assert [fp.session_id for _n, _h, fp in seen] == ["s-1", "s-2", "s-3"] + assert [fp.mean_us for _n, _h, fp in seen] == [1000, 1100, 1200] + + +def test_entity_chain_iter_yields_nothing_for_unknown_entity(tmp_path): + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp(entity_id="alice")) + assert list(entity_chain_iter(conn, "carol")) == [] + + +def test_entity_chain_hash_returns_none_for_unknown_entity(tmp_path): + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp(entity_id="alice")) + assert entity_chain_hash(conn, "carol") is None + + +def test_verify_entity_chain_accepts_correct(tmp_path): + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp()) + assert verify_entity_chain(conn, "alice", out["entity_chain_hash"]) + + +def test_verify_entity_chain_rejects_wrong(tmp_path): + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp()) + assert not verify_entity_chain(conn, "alice", "f" * 64) + + +def test_commit_rejects_unknown_schema_version(tmp_path): + conn = connect(tmp_path / "t.db") + base = _fp() + bad = Fingerprint(**{**base.__dict__, "schema_version": "fingerprint-v999"}) + with pytest.raises(ValueError): + commit_fingerprint(conn, bad) + + +def test_fingerprint_round_trips_through_chain(tmp_path): + conn = connect(tmp_path / "t.db") + original = _fp(session_id="s-1", mean_us=4242) + commit_fingerprint(conn, original) + recovered = next(entity_chain_iter(conn, "alice"))[2] + # Frozen dataclass equality covers all fields including the dict ones. + assert recovered == original + + +def test_event_type_is_namespaced(tmp_path): + """Other event types with same subject_root must not pollute the chain.""" + from arborist.store import append_audit + + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp(entity_id="alice")) + # Inject an unrelated event with the same subject_root. + append_audit(conn, event_type="ingest", body={"x": 1}, subject_root="alice") + commit_fingerprint(conn, _fp(entity_id="alice", session_id="s-2", mean_us=1100)) + # Chain still counts only the two attest_fingerprint events. + seen = list(entity_chain_iter(conn, "alice")) + assert len(seen) == 2 + assert [fp.session_id for _n, _h, fp in seen] == ["s-1", "s-2"] + + +def test_event_hash_matches_global_audit_chain_invariant(tmp_path): + """sha256(prev_global || canonical_body) — the rule in store.py:1665-1675.""" + import json + from dataclasses import asdict + + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp()) + body = asdict(_fp()) + body["error_taxonomy"] = dict(body["error_taxonomy"]) + body["stress_params"] = dict(body["stress_params"]) + body_json = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + expected = hashlib.sha256( + body_json.encode("utf-8", errors="surrogatepass") + ).hexdigest() + assert out["event_hash"] == expected + + +def test_event_type_constant_is_pinned(): + assert EVENT_TYPE == "attest_fingerprint"