From e6c42773b0d42bc3bcb9b849a9a59f63ff3d7ac2 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Fri, 5 Jun 2026 11:58:30 -0400 Subject: [PATCH] attest: Fingerprint schema + canonical encoder + Merkle leaf hash First piece of digital-identity-proof-of-overlearning. arborist/attest/ ships an integer-only Fingerprint (latency in microseconds, skew quantized to 4dp int) so canonical bytes stay bit-stable across hosts. Commitment-not-occurrence bound is a hard invariant in arborist/attest/__init__.py: a fingerprint commits the encoded metrics under this schema; it does NOT assert occurrence or body-binding. Sibling repos close those gaps. 13 tests pass. --- arborist/attest/__init__.py | 27 +++++ arborist/attest/fingerprint.py | 178 +++++++++++++++++++++++++++++++ tests/test_attest_fingerprint.py | 159 +++++++++++++++++++++++++++ 3 files changed, 364 insertions(+) create mode 100644 arborist/attest/__init__.py create mode 100644 arborist/attest/fingerprint.py create mode 100644 tests/test_attest_fingerprint.py diff --git a/arborist/attest/__init__.py b/arborist/attest/__init__.py new file mode 100644 index 0000000..0adf5da --- /dev/null +++ b/arborist/attest/__init__.py @@ -0,0 +1,27 @@ +"""Consistency attestation primitives. + +A `Fingerprint` is a canonical, integer-only summary of a session's +observable behavior (latency moments, error taxonomy, stress params, +host hash). Fingerprints hash via the project Merkle conventions +(`arborist.merkle.hash_leaf`) and chain into the existing audit +substrate. + +**Commitment-not-occurrence bound.** A fingerprint commitment asserts +that the committed metrics were canonically encoded under this +schema. It does NOT assert that the underlying behavior occurred, +that the metrics describe a specific physical body, or that the +session was unobserved. Body-binding is out of scope. + +**Pure-stats verifier.** Consistency tests (K-S / Mahalanobis) over +fingerprint distributions are deterministic functions of their +sample inputs. No model is in the proof path. +""" + +from arborist.attest.fingerprint import ( + Fingerprint, + canonical_bytes, + from_samples, + leaf_hash, +) + +__all__ = ["Fingerprint", "canonical_bytes", "from_samples", "leaf_hash"] diff --git a/arborist/attest/fingerprint.py b/arborist/attest/fingerprint.py new file mode 100644 index 0000000..44f2eb9 --- /dev/null +++ b/arborist/attest/fingerprint.py @@ -0,0 +1,178 @@ +"""Fingerprint schema + canonical encoder + Merkle leaf hash. + +Schema version: ``fingerprint-v1``. + +All numeric fields are integers. Latency lives in microseconds. +Skew is quantized to a 4-decimal-place integer (``skew_q10000``). +This keeps canonical bytes bit-stable across hosts with different +libm float repr — float math runs once at fingerprint construction, +quantizes to int, never re-enters the canonical form. + +Cross-host stability has a known limit: if libm float drift crosses +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. +""" + +from __future__ import annotations + +import hashlib +import json +import statistics +from dataclasses import asdict, dataclass, field +from typing import Mapping, Sequence + +from arborist.merkle import hash_leaf + +SCHEMA_VERSION = "fingerprint-v1" + + +@dataclass(frozen=True) +class Fingerprint: + """Canonical, integer-only summary of one observed session. + + Field-by-field: + + - ``entity_id``: opaque caller-supplied identifier. Treated as a + bytestring; this module does not interpret it. + - ``session_id``: opaque per-session identifier. + - ``timestamp_utc``: ISO-8601 UTC string (no timezone offset + shortcuts, no microseconds — second precision; canonical form + is enforced by the caller). + - ``domain``: scope tag (e.g. ``"python-debug"``); fingerprints + from different domains are not statistically comparable. + - ``host_hash``: hex-encoded SHA-256 of the canonical host + identity bytes (caller decides what those are: hostname + + machine-id + user, opus-mt-style hash-pinning, etc.). + Defends against multi-host stitching attacks where an + attacker would otherwise replay sessions captured under + different hardware as one continuous profile. + - ``count``: number of samples folded into the moments. + - ``mean_us`` / ``std_us``: latency mean and population std, + microseconds, integer. + - ``p50_us`` / ``p90_us`` / ``p99_us``: latency percentiles, + microseconds, integer. + - ``skew_q10000``: ``round(sample_skew * 10000)`` as int. + - ``error_taxonomy``: error-kind → count, integer counts only. + - ``stress_params``: stress parameters in effect during the + session (``concurrent_tasks``, ``time_limit_s``, ``interrupts``, + etc.); caller-defined keys, integer values only. + - ``schema_version``: pinned to ``SCHEMA_VERSION``. + """ + + entity_id: str + session_id: str + timestamp_utc: str + domain: str + host_hash: str + count: int + mean_us: int + std_us: int + p50_us: int + p90_us: int + p99_us: int + skew_q10000: int + error_taxonomy: Mapping[str, int] = field(default_factory=dict) + stress_params: Mapping[str, int] = field(default_factory=dict) + schema_version: str = SCHEMA_VERSION + + +def _canonical_json(obj) -> str: + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def canonical_bytes(fp: Fingerprint) -> bytes: + """Canonical UTF-8 bytes for a fingerprint. + + Deterministic across runs and hosts when every payload field is + an integer or a fixed string. + """ + payload = asdict(fp) + payload["error_taxonomy"] = dict(payload["error_taxonomy"]) + payload["stress_params"] = dict(payload["stress_params"]) + return _canonical_json(payload).encode("utf-8", errors="surrogatepass") + + +def leaf_hash(fp: Fingerprint) -> bytes: + """SHA-256 leaf hash with project Merkle leaf prefix.""" + return hash_leaf(canonical_bytes(fp)) + + +def _sample_skew(samples: Sequence[int], mean: float, std: float) -> float: + """Population skewness; returns 0.0 when std is 0 or n<2.""" + n = len(samples) + if n < 2 or std == 0.0: + return 0.0 + s = 0.0 + for x in samples: + z = (x - mean) / std + s += z * z * z + return s / n + + +def _percentile(sorted_samples: Sequence[int], q: float) -> int: + """Nearest-rank percentile on a pre-sorted integer sequence.""" + n = len(sorted_samples) + if n == 0: + raise ValueError("no samples") + if n == 1: + return sorted_samples[0] + # Nearest-rank: ceil(q * n), 1-indexed. + import math + + rank = max(1, min(n, math.ceil(q * n))) + return sorted_samples[rank - 1] + + +def from_samples( + *, + entity_id: str, + session_id: str, + timestamp_utc: str, + domain: str, + host_hash: str, + latency_samples_us: Sequence[int], + error_taxonomy: Mapping[str, int] | None = None, + stress_params: Mapping[str, int] | None = None, +) -> Fingerprint: + """Compute a Fingerprint from a list of integer microsecond latencies. + + Raises ValueError on empty samples. + """ + n = len(latency_samples_us) + if n == 0: + raise ValueError("latency_samples_us is empty") + for v in latency_samples_us: + if not isinstance(v, int): + raise TypeError("latency samples must be int (microseconds)") + sorted_samples = sorted(latency_samples_us) + mean_f = statistics.fmean(latency_samples_us) + std_f = statistics.pstdev(latency_samples_us) if n > 1 else 0.0 + skew_f = _sample_skew(latency_samples_us, mean_f, std_f) + return Fingerprint( + entity_id=entity_id, + session_id=session_id, + timestamp_utc=timestamp_utc, + domain=domain, + host_hash=host_hash, + count=n, + mean_us=int(round(mean_f)), + std_us=int(round(std_f)), + p50_us=_percentile(sorted_samples, 0.50), + p90_us=_percentile(sorted_samples, 0.90), + p99_us=_percentile(sorted_samples, 0.99), + skew_q10000=int(round(skew_f * 10000)), + error_taxonomy=dict(error_taxonomy or {}), + stress_params=dict(stress_params or {}), + ) + + +def host_hash_from_parts(*parts: str) -> str: + """Helper: hex SHA-256 over canonical bytes of a list of strings. + + Caller decides what parts make up host identity. Example: + ``host_hash_from_parts(hostname, machine_id, user)``. + """ + payload = _canonical_json(list(parts)).encode("utf-8", errors="surrogatepass") + return hashlib.sha256(payload).hexdigest() diff --git a/tests/test_attest_fingerprint.py b/tests/test_attest_fingerprint.py new file mode 100644 index 0000000..6058035 --- /dev/null +++ b/tests/test_attest_fingerprint.py @@ -0,0 +1,159 @@ +"""Tests for `arborist.attest.fingerprint`.""" + +from __future__ import annotations + +import hashlib + +import pytest + +from arborist.attest import ( + Fingerprint, + canonical_bytes, + from_samples, + leaf_hash, +) +from arborist.attest.fingerprint import ( + SCHEMA_VERSION, + _percentile, + _sample_skew, + host_hash_from_parts, +) +from arborist.merkle import LEAF_PREFIX + + +def _fp(**overrides): + base = dict( + entity_id="alice", + session_id="s-1", + 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=SCHEMA_VERSION, + ) + base.update(overrides) + return Fingerprint(**base) + + +def test_canonical_bytes_is_deterministic(): + fp = _fp() + assert canonical_bytes(fp) == canonical_bytes(fp) + + +def test_canonical_bytes_is_key_sorted(): + """Constructor key order must not affect canonical bytes.""" + fp_a = _fp(error_taxonomy={"a": 1, "b": 2}, stress_params={"x": 1, "y": 2}) + fp_b = _fp(error_taxonomy={"b": 2, "a": 1}, stress_params={"y": 2, "x": 1}) + assert canonical_bytes(fp_a) == canonical_bytes(fp_b) + + +def test_canonical_bytes_distinguishes_payloads(): + assert canonical_bytes(_fp()) != canonical_bytes(_fp(mean_us=1001)) + assert canonical_bytes(_fp()) != canonical_bytes(_fp(entity_id="bob")) + assert canonical_bytes(_fp()) != canonical_bytes(_fp(domain="js-debug")) + assert canonical_bytes(_fp()) != canonical_bytes(_fp(host_hash="f" * 64)) + + +def test_leaf_hash_uses_project_merkle_prefix(): + fp = _fp() + expected = hashlib.sha256(LEAF_PREFIX + canonical_bytes(fp)).digest() + assert leaf_hash(fp) == expected + assert len(leaf_hash(fp)) == 32 + + +def test_leaf_hash_distinguishes_distinct_fingerprints(): + assert leaf_hash(_fp()) != leaf_hash(_fp(session_id="s-2")) + + +def test_from_samples_computes_moments(): + samples = [1000, 1100, 1200, 1300, 1400] + fp = from_samples( + entity_id="alice", + session_id="s-1", + timestamp_utc="2026-06-05T15:30:00Z", + domain="python-debug", + host_hash="0" * 64, + latency_samples_us=samples, + ) + assert fp.count == 5 + assert fp.mean_us == 1200 + # Nearest-rank percentile at q=0.50 on n=5 = rank ceil(2.5)=3 → 1200. + assert fp.p50_us == 1200 + # q=0.90 → rank ceil(4.5)=5 → 1400. + assert fp.p90_us == 1400 + # q=0.99 → rank ceil(4.95)=5 → 1400. + assert fp.p99_us == 1400 + # Symmetric arithmetic progression → skew = 0 → q10000 = 0. + assert fp.skew_q10000 == 0 + + +def test_from_samples_rejects_empty(): + with pytest.raises(ValueError): + from_samples( + entity_id="alice", + session_id="s-1", + timestamp_utc="2026-06-05T15:30:00Z", + domain="python-debug", + host_hash="0" * 64, + latency_samples_us=[], + ) + + +def test_from_samples_rejects_floats(): + with pytest.raises(TypeError): + from_samples( + entity_id="alice", + session_id="s-1", + timestamp_utc="2026-06-05T15:30:00Z", + domain="python-debug", + host_hash="0" * 64, + latency_samples_us=[1.0, 2.0, 3.0], # type: ignore[list-item] + ) + + +def test_sample_skew_zero_on_constant(): + assert _sample_skew([5, 5, 5], 5.0, 0.0) == 0.0 + + +def test_sample_skew_positive_for_right_tail(): + # Long right tail → positive skew. + samples = [1, 1, 1, 1, 1, 100] + import statistics + + mean = statistics.fmean(samples) + std = statistics.pstdev(samples) + assert _sample_skew(samples, mean, std) > 0 + + +def test_percentile_nearest_rank(): + s = [10, 20, 30, 40, 50] + assert _percentile(s, 0.0) == 10 + assert _percentile(s, 0.5) == 30 + assert _percentile(s, 1.0) == 50 + + +def test_host_hash_from_parts_is_deterministic_and_order_sensitive(): + h_ab = host_hash_from_parts("host", "machine-id") + h_ba = host_hash_from_parts("machine-id", "host") + assert h_ab == host_hash_from_parts("host", "machine-id") + assert h_ab != h_ba # order matters; caller picks the canonical order + + +def test_schema_version_is_pinned_in_default_construction(): + fp = from_samples( + entity_id="alice", + session_id="s-1", + timestamp_utc="2026-06-05T15:30:00Z", + domain="python-debug", + host_hash="0" * 64, + latency_samples_us=[1000], + ) + assert fp.schema_version == SCHEMA_VERSION