diff --git a/arborist/attest/__init__.py b/arborist/attest/__init__.py index d2d4d0a..e7da74f 100644 --- a/arborist/attest/__init__.py +++ b/arborist/attest/__init__.py @@ -18,6 +18,7 @@ sample inputs. No model is in the proof path. """ from arborist.attest.chain import ( + blast_radius, commit_fingerprint, entity_chain_hash, entity_chain_iter, @@ -41,6 +42,7 @@ from arborist.attest.fingerprint import ( __all__ = [ "Fingerprint", "KSResult", + "blast_radius", "canonical_bytes", "commit_fingerprint", "entity_chain_hash", diff --git a/arborist/attest/chain.py b/arborist/attest/chain.py index 480b693..e653a43 100644 --- a/arborist/attest/chain.py +++ b/arborist/attest/chain.py @@ -154,6 +154,7 @@ def entity_chain_iter( 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()}, + cites=tuple(body.get("cites", ())), schema_version=body["schema_version"], ) yield n, event_hash, fp @@ -165,3 +166,36 @@ def verify_entity_chain( """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 + + +def blast_radius(conn: sqlite3.Connection, falsified_chain_hash: str) -> list[str]: + """Distinct entity_ids whose fingerprints cite the falsified chain hash. + + When an entity's chain hash is later falsified (KS / Mahalanobis + drift, manual quarantine, etc.), this returns the set of entities + who had declared a citation on that exact chain state via + ``Fingerprint.cites``. Their trust posture is now contingent on + the falsification and should be re-examined. + + Pure read; does NOT mutate state. Routing decisions (quarantine, + retest, downgrade) belong to a caller higher up the stack — the + blast graph is the input, not the action. + + Linear scan over ``audit_events`` of type ``attest_fingerprint``; + O(N) per call. Acceptable while N is small; if it ever isn't, a + derived ``fingerprint_cites(falsified_chain, citing_entity_id)`` + index is the next step. Don't build that until the pressure is + real. + """ + import json + + affected: set[str] = set() + cur = conn.execute( + "SELECT body FROM audit_events WHERE event_type=?", + (EVENT_TYPE,), + ) + for (body_json,) in cur: + body = json.loads(body_json) + if falsified_chain_hash in body.get("cites", ()): + affected.add(body["entity_id"]) + return sorted(affected) diff --git a/arborist/attest/fingerprint.py b/arborist/attest/fingerprint.py index 44f2eb9..65f0e87 100644 --- a/arborist/attest/fingerprint.py +++ b/arborist/attest/fingerprint.py @@ -1,6 +1,6 @@ """Fingerprint schema + canonical encoder + Merkle leaf hash. -Schema version: ``fingerprint-v1``. +Schema version: ``fingerprint-v2``. All numeric fields are integers. Latency lives in microseconds. Skew is quantized to a 4-decimal-place integer (``skew_q10000``). @@ -13,6 +13,16 @@ the quantization boundary at unlucky moments, two hosts could differ by ±1 in the last digit of ``skew_q10000`` or one of the std/mean fields. Acceptable for v1; a future revision can compute moments in integer arithmetic if required. + +**v2 delta: optional ``cites`` field.** A fingerprint may declare a +``cites`` tuple of other entities' chain hashes that it vouches for +or depends on. The citation is a *claim of dependency*, not a +verification — at commit time arborist does not re-verify the cited +chain. The point is that if a cited chain hash is later falsified +(KS / Mahalanobis drift, manual quarantine, etc.), the audit graph +yields a deterministic blast radius via ``chain.blast_radius()``. +Default empty tuple → equivalent to v1 semantics. Legacy v1 bodies +in storage (no ``cites`` key) are reconstructed with ``cites=()``. """ from __future__ import annotations @@ -25,7 +35,7 @@ from typing import Mapping, Sequence from arborist.merkle import hash_leaf -SCHEMA_VERSION = "fingerprint-v1" +SCHEMA_VERSION = "fingerprint-v2" @dataclass(frozen=True) @@ -58,6 +68,12 @@ class Fingerprint: - ``stress_params``: stress parameters in effect during the session (``concurrent_tasks``, ``time_limit_s``, ``interrupts``, etc.); caller-defined keys, integer values only. + - ``cites``: optional tuple of other entities' chain hashes this + fingerprint declares a dependency on. Order is significant + (caller decides) — citations are NOT sorted at canonicalization, + because the order itself carries meaning ("I observed A then B"). + Default empty tuple. See module docstring for blast-radius + semantics. - ``schema_version``: pinned to ``SCHEMA_VERSION``. """ @@ -75,6 +91,7 @@ class Fingerprint: skew_q10000: int error_taxonomy: Mapping[str, int] = field(default_factory=dict) stress_params: Mapping[str, int] = field(default_factory=dict) + cites: tuple[str, ...] = () schema_version: str = SCHEMA_VERSION @@ -135,6 +152,7 @@ def from_samples( latency_samples_us: Sequence[int], error_taxonomy: Mapping[str, int] | None = None, stress_params: Mapping[str, int] | None = None, + cites: Sequence[str] = (), ) -> Fingerprint: """Compute a Fingerprint from a list of integer microsecond latencies. @@ -165,6 +183,7 @@ def from_samples( skew_q10000=int(round(skew_f * 10000)), error_taxonomy=dict(error_taxonomy or {}), stress_params=dict(stress_params or {}), + cites=tuple(cites), ) diff --git a/tests/test_attest_chain.py b/tests/test_attest_chain.py index b3e7532..a226aac 100644 --- a/tests/test_attest_chain.py +++ b/tests/test_attest_chain.py @@ -8,6 +8,7 @@ import pytest from arborist.attest import ( Fingerprint, + blast_radius, commit_fingerprint, entity_chain_hash, entity_chain_iter, @@ -22,7 +23,7 @@ from arborist.attest.fingerprint import SCHEMA_VERSION from arborist.store import connect -def _fp(entity_id="alice", session_id="s-1", mean_us=1000): +def _fp(entity_id="alice", session_id="s-1", mean_us=1000, cites=()): return Fingerprint( entity_id=entity_id, session_id=session_id, @@ -38,6 +39,7 @@ def _fp(entity_id="alice", session_id="s-1", mean_us=1000): skew_q10000=0, error_taxonomy={}, stress_params={}, + cites=tuple(cites), schema_version=SCHEMA_VERSION, ) @@ -177,3 +179,93 @@ def test_event_hash_matches_global_audit_chain_invariant(tmp_path): def test_event_type_constant_is_pinned(): assert EVENT_TYPE == "attest_fingerprint" + + +def test_blast_radius_empty_when_no_events(tmp_path): + conn = connect(tmp_path / "t.db") + assert blast_radius(conn, "a" * 64) == [] + + +def test_blast_radius_empty_when_no_citations(tmp_path): + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp(entity_id="alice")) + assert blast_radius(conn, out["entity_chain_hash"]) == [] + + +def test_blast_radius_finds_single_citer(tmp_path): + conn = connect(tmp_path / "t.db") + a = commit_fingerprint(conn, _fp(entity_id="alice")) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)), + ) + assert blast_radius(conn, a["entity_chain_hash"]) == ["bob"] + + +def test_blast_radius_finds_multiple_citers_distinct_and_sorted(tmp_path): + conn = connect(tmp_path / "t.db") + a = commit_fingerprint(conn, _fp(entity_id="alice")) + commit_fingerprint( + conn, + _fp(entity_id="carol", session_id="c-1", cites=(a["entity_chain_hash"],)), + ) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)), + ) + assert blast_radius(conn, a["entity_chain_hash"]) == ["bob", "carol"] + + +def test_blast_radius_deduplicates_repeated_citers(tmp_path): + conn = connect(tmp_path / "t.db") + a = commit_fingerprint(conn, _fp(entity_id="alice")) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)), + ) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-2", mean_us=1100, cites=(a["entity_chain_hash"],)), + ) + # Same entity citing the same chain hash from two fingerprints → one row. + assert blast_radius(conn, a["entity_chain_hash"]) == ["bob"] + + +def test_blast_radius_misses_unrelated_chain_hash(tmp_path): + conn = connect(tmp_path / "t.db") + a = commit_fingerprint(conn, _fp(entity_id="alice")) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)), + ) + assert blast_radius(conn, "f" * 64) == [] + + +def test_legacy_v1_body_without_cites_reconstructs_with_empty_tuple(tmp_path): + """v1 events in storage (no `cites` key) must round-trip through entity_chain_iter.""" + from arborist.store import append_audit + + conn = connect(tmp_path / "t.db") + legacy_body = { + "entity_id": "alice", + "session_id": "s-legacy", + "timestamp_utc": "2026-06-05T15:30:00Z", + "domain": "python-debug", + "host_hash": "0" * 64, + "count": 3, + "mean_us": 1000, + "std_us": 100, + "p50_us": 1000, + "p90_us": 1100, + "p99_us": 1200, + "skew_q10000": 0, + "error_taxonomy": {}, + "stress_params": {}, + "schema_version": "fingerprint-v1", + } + append_audit(conn, event_type=EVENT_TYPE, body=legacy_body, subject_root="alice") + seen = list(entity_chain_iter(conn, "alice")) + assert len(seen) == 1 + fp = seen[0][2] + assert fp.cites == () + assert fp.schema_version == "fingerprint-v1"