From c92ebad4edbc33ad3cfefaa519faae42b96ca728 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 28 Apr 2026 17:36:33 -0400 Subject: [PATCH] mesh: per-peer audit chain-of-claims tracking on the wire MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Receiver now rejects gossip envelopes whose prev_event_hash doesn't extend the sender's last-seen event_hash — fork detection lives at the wire layer, not just in docs/mesh.md. Schema: new mesh_peer_chains table (peer_member_id PK, last_event_hash, last_seq, last_seen_at) with idempotent _migrate_mesh_peer_chains matching the _migrate_audit_mode pattern. Envelope: WireEnvelope gains prev_event_hash + event_hash, both optional for back-compat. canonical_bytes() excludes event_hash to avoid the circular dependency (the hash is computed FROM the canonical bytes). New WireEnvelope.with_chain(prev_event_hash=...) builds a linked envelope with event_hash = sha256(prev || canonical_bytes). Sender (MeshWireClient._signed_envelope, MeshWireServer.handle_request DELIVER_BODY): reads our latest_event_hash, builds the linked envelope, then appends a 'mesh_sent' audit event whose body is the envelope's canonical (event-hash-excluded) dict. By construction the audit event_hash equals the envelope event_hash — wire chain-of-claims and local audit chain stay in lockstep, so back-to-back sends advance the chain naturally. Receiver (handle_announce): when an envelope carries chain fields, recompute event_hash and 400 on mismatch; look up mesh_peer_chains[sender] and 409 on prev mismatch; on accept update the row + write the existing mesh_received event. Backward compat: legacy envelopes (event_hash=None) bypass chain enforcement — kept the existing test_mesh_wire.py fixtures verbatim since they construct WireEnvelope directly without with_chain(). All production traffic goes through MeshWireClient and is always tracked. v1 deferred (documented in module docstring): no multi-event catchup; operator retries on 409. No cross-peer reconciliation. Tests: 12 new in tests/test_mesh_chain.py covering the 7 required cases plus canonical-bytes exclusion, migration idempotence, and legacy-envelope fallback. 246 passed, 1 skipped overall. --- aborist/mesh/wire.py | 347 +++++++++++++++++++++++---- aborist/store.py | 38 +++ tests/test_mesh_chain.py | 494 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 828 insertions(+), 51 deletions(-) create mode 100644 tests/test_mesh_chain.py diff --git a/aborist/mesh/wire.py b/aborist/mesh/wire.py index 9d69e2e..8244481 100644 --- a/aborist/mesh/wire.py +++ b/aborist/mesh/wire.py @@ -16,15 +16,51 @@ valid signature, so non-members are silently rejected. Each receive writes a `mesh_received` event into the *local* audit chain whose body is the full signed envelope. The receiver's chain -stays internally consistent because we append in receive-order. The -sender's chain integrity is verified by signature alone at this layer -— per-peer chain-of-claims tracking (catchup, replay, fork detection) -is a v2 concern (see ``docs/mesh.md``). +stays internally consistent because we append in receive-order. + +**Per-peer chain-of-claims tracking.** Every envelope built via the +sender helper (``MeshWireClient`` outbound traffic, plus +``WireEnvelope.with_chain``) carries two extra fields: + +- ``prev_event_hash`` — the sender's previous chain-of-claims hash, or + ``None`` for the first envelope a peer ever broadcasts. +- ``event_hash`` — ``sha256(prev_event_hash || canonical_envelope_bytes)``. + +The receiver: +1. Verifies the Ed25519 signature (existing). +2. Recomputes ``event_hash`` from ``prev_event_hash`` + the envelope's + canonical bytes (excluding ``event_hash`` itself, which is circular). + Mismatch → 400. +3. Looks up the sender's last-seen ``event_hash`` in + ``mesh_peer_chains``. If a row exists, the envelope's + ``prev_event_hash`` MUST equal it; otherwise the chain has forked + and we return 409 Conflict. If no row exists, this is the first + envelope we've seen from this peer; any ``prev_event_hash`` is + accepted (we have no prior to compare against). +4. On accept, ``mesh_peer_chains`` is updated to the new + ``event_hash`` + bumped ``last_seq`` + ``now``. + +Backward compatibility: envelopes that arrive WITHOUT +``prev_event_hash`` and ``event_hash`` (legacy, hand-built test fixtures +from before the chain-tracking rollout) skip per-peer chain enforcement. +The signature check still runs. New code paths always carry chain +fields, so this fallback only matters for legacy test envelopes. + +v1 limitations (deferred to a later commit): + +- **No multi-event catchup / replay.** If a receiver missed event N + (peer sent N+1 with prev=N's hash but receiver still has prev=N-1), + we reject with 409 and rely on an operator to retry or run a catchup + pass. v2 will fetch the missing range from the peer and replay. +- **No cross-peer reconciliation.** Each peer's chain is tracked + independently — alice's broadcasts to bob and bob's broadcasts to + carol are unrelated streams. """ from __future__ import annotations import base64 +import hashlib import json import os import threading @@ -43,7 +79,7 @@ from aborist.mesh.state import ( recover_epoch_secret, roster_at, ) -from aborist.store import append_audit, connect +from aborist.store import append_audit, connect, latest_event_hash # --------------------------------------------------------------------------- @@ -95,6 +131,17 @@ class WireEnvelope: binds the ciphertext: a MITM cannot swap `encrypted_body` without invalidating the signature. Receivers detect encryption by the presence of `encrypted_body` and decrypt before handler dispatch. + + `prev_event_hash` and `event_hash` carry the sender's chain-of-claims + state. Both default to ``None`` for backward compatibility with hand- + built test envelopes. Production traffic is built via + :meth:`with_chain` (or `MeshWireClient`) which always populates them. + + `event_hash` is intentionally NOT included in the signed canonical + bytes — it is computed FROM those bytes, so embedding it would be + circular. The receiver re-derives ``event_hash`` from + ``sha256(prev_event_hash || canonical_bytes_without_event_hash)`` + and asserts equality before accepting. """ type: str @@ -104,21 +151,73 @@ class WireEnvelope: body: dict[str, Any] v: int = WIRE_VERSION encrypted_body: dict[str, str] | None = None + prev_event_hash: str | None = None + event_hash: str | None = None def canonical_bytes(self) -> bytes: """Bytes that get signed. Stable, reproducible. - Includes `encrypted_body` whenever it is set, so signatures bind - the ciphertext as well as the metadata. We omit the field - entirely from the canonical form when it is None to keep - cleartext-only envelopes byte-identical to the v1 wire format - (forward-compat for receivers that don't speak AEAD yet). + Excludes ``event_hash`` (circular: it's the hash OF these bytes) + but includes ``prev_event_hash`` (linkage proof — the sender is + committing to a specific predecessor) and ``encrypted_body`` + (so signatures bind the ciphertext too). The ``encrypted_body`` + field is dropped from the canonical form when ``None`` so + cleartext-only envelopes stay byte-identical to the v1 wire + format — forward-compat for receivers that don't speak AEAD yet. """ d = asdict(self) + d.pop("event_hash", None) if d.get("encrypted_body") is None: d.pop("encrypted_body", None) return _canonical_json(d).encode("utf-8") + def compute_event_hash(self) -> str: + """Deterministic hash of this envelope as a chain-of-claims node. + + ``sha256(prev_event_hash_bytes || canonical_bytes)`` where + ``prev_event_hash_bytes`` is the raw 32-byte digest if + ``prev_event_hash`` is set, else empty. + """ + h = hashlib.sha256() + if self.prev_event_hash is not None: + h.update(bytes.fromhex(self.prev_event_hash)) + h.update(self.canonical_bytes()) + return h.hexdigest() + + def with_chain(self, *, prev_event_hash: str | None) -> "WireEnvelope": + """Return a copy with chain fields populated. + + ``event_hash`` is computed deterministically from + ``prev_event_hash`` + this envelope's canonical bytes. The + returned envelope is the one the sender signs and broadcasts. + """ + # Construct the linked envelope first so canonical_bytes() sees + # the new prev_event_hash. event_hash itself is excluded from the + # canonical bytes, so we set it last. + linked = WireEnvelope( + type=self.type, + sender_id=self.sender_id, + epoch_id=self.epoch_id, + ts=self.ts, + body=self.body, + v=self.v, + encrypted_body=self.encrypted_body, + prev_event_hash=prev_event_hash, + event_hash=None, + ) + eh = linked.compute_event_hash() + return WireEnvelope( + type=linked.type, + sender_id=linked.sender_id, + epoch_id=linked.epoch_id, + ts=linked.ts, + body=linked.body, + v=linked.v, + encrypted_body=linked.encrypted_body, + prev_event_hash=linked.prev_event_hash, + event_hash=eh, + ) + def sign_with(self, sign_priv: bytes) -> dict[str, Any]: """Wrap envelope + Ed25519 signature in a dict ready for HTTP body.""" sig = sign(sign_priv, self.canonical_bytes()) @@ -134,7 +233,11 @@ class WireEnvelope: def from_signed(cls, signed: dict[str, Any]) -> tuple["WireEnvelope", bytes]: """Reverse of `sign_with`. Returns (envelope, sig_bytes). - Does NOT verify — caller verifies via `verify_envelope_sig`. + Does NOT verify — caller verifies via `verify_envelope_sig` and + (when chain fields are present) via the receiver's chain check. + ``prev_event_hash`` and ``event_hash`` are optional in the wire + payload — legacy envelopes that pre-date chain tracking decode + with both as ``None``. """ if not isinstance(signed, dict) or "envelope" not in signed or "sig_b64" not in signed: raise ValueError("signed payload missing envelope/sig_b64") @@ -155,6 +258,12 @@ class WireEnvelope: for k in ("nonce_b64", "ct_b64"): if k not in encrypted_body: raise ValueError(f"encrypted_body missing field: {k}") + prev = env_d.get("prev_event_hash") + eh = env_d.get("event_hash") + if prev is not None and not isinstance(prev, str): + raise ValueError("prev_event_hash must be str or null") + if eh is not None and not isinstance(eh, str): + raise ValueError("event_hash must be str or null") env = cls( type=env_d["type"], sender_id=env_d["sender_id"], @@ -163,6 +272,8 @@ class WireEnvelope: body=env_d["body"], v=int(env_d["v"]), encrypted_body=encrypted_body, + prev_event_hash=prev, + event_hash=eh, ) try: sig = base64.b64decode(signed["sig_b64"], validate=True) @@ -248,6 +359,44 @@ def decrypt_envelope_body( return decoded +def _peer_chain_row(conn, peer_member_id: str) -> tuple[str, int] | None: + """Return (last_event_hash, last_seq) for a peer, or None if unseen.""" + row = conn.execute( + "SELECT last_event_hash, last_seq FROM mesh_peer_chains " + "WHERE peer_member_id = ?", + (peer_member_id,), + ).fetchone() + if row is None: + return None + return row["last_event_hash"], int(row["last_seq"]) + + +def _update_peer_chain( + conn, + peer_member_id: str, + new_event_hash: str, + *, + now: int, +) -> None: + """Upsert mesh_peer_chains[peer] = new_event_hash, last_seq+=1, now.""" + existing = _peer_chain_row(conn, peer_member_id) + if existing is None: + conn.execute( + "INSERT INTO mesh_peer_chains " + "(peer_member_id, last_event_hash, last_seq, last_seen_at) " + "VALUES (?, ?, ?, ?)", + (peer_member_id, new_event_hash, 1, now), + ) + else: + _, last_seq = existing + conn.execute( + "UPDATE mesh_peer_chains " + "SET last_event_hash = ?, last_seq = ?, last_seen_at = ? " + "WHERE peer_member_id = ?", + (new_event_hash, last_seq + 1, now, peer_member_id), + ) + + def verify_envelope_sig( conn, envelope: WireEnvelope, @@ -425,6 +574,40 @@ class MeshWireServer: try: if not verify_envelope_sig(conn, env, sig): return 401, {"error": "signature verify failed"} + + # Per-peer chain-of-claims tracking. Only enforced when the + # sender attached chain fields. Hand-built legacy envelopes + # (env.event_hash is None) skip enforcement — signature alone. + if env.event_hash is not None: + expected_eh = env.compute_event_hash() + if expected_eh != env.event_hash: + return 400, { + "error": "envelope event_hash mismatch: " + f"computed {expected_eh}, claimed {env.event_hash}" + } + existing = _peer_chain_row(conn, env.sender_id) + if existing is not None: + last_eh, _ = existing + if env.prev_event_hash != last_eh: + return 409, { + "error": ( + f"fork detected: peer {env.sender_id} " + f"expected prev_event_hash {last_eh}, " + f"got {env.prev_event_hash}" + ) + } + # else: first envelope from this peer; any prev_event_hash + # (None or arbitrary string) is accepted — we have no + # prior to compare against. + _update_peer_chain( + conn, + env.sender_id, + env.event_hash, + now=int(time.time()), + ) + + # AEAD decrypt happens after sig + chain checks so we don't + # waste a secret recovery on rejected traffic. effective_body = env.body if env.encrypted_body is not None: try: @@ -432,6 +615,7 @@ class MeshWireServer: effective_body = decrypt_envelope_body(env, epoch_secret) except (ValueError, RuntimeError) as e: return 401, {"error": f"AEAD decrypt failed: {e!s}"} + event_hash = append_audit( conn, event_type="mesh_received", @@ -442,6 +626,11 @@ class MeshWireServer: "remote_ts": env.ts, "envelope_body": effective_body, "encrypted": env.encrypted_body is not None, + # Sender's chain-of-claims hash (if present). Records + # what we accepted so future audits can reconstruct + # the sender's claimed history. + "sender_event_hash": env.event_hash, + "sender_prev_event_hash": env.prev_event_hash, }, subject_root=str( effective_body.get("document_root") @@ -501,6 +690,7 @@ class MeshWireServer: if me is None or epoch is None: return 503, {"error": "mesh not initialized on responder"} + prev = latest_event_hash(conn) deliver = WireEnvelope( type=TYPE_DELIVER_BODY, sender_id=me.member_id, @@ -512,7 +702,25 @@ class MeshWireServer: "leaves_hex": leaves_hex, "text": text, }, + ).with_chain(prev_event_hash=prev) + + # Mirror the client's audit-log of every outbound chain-linked + # envelope so DELIVER_BODY also extends the responder's chain. + from dataclasses import asdict as _asdict + audit_body = _asdict(deliver) + audit_body.pop("event_hash", None) + # Mirror canonical_bytes(): drop None encrypted_body so the + # audit hash matches event_hash for cleartext envelopes. + if audit_body.get("encrypted_body") is None: + audit_body.pop("encrypted_body", None) + audit_eh = append_audit( + conn, + event_type="mesh_sent", + body=audit_body, + subject_root=root, + ts=deliver.ts, ) + assert audit_eh == deliver.event_hash return 200, deliver.sign_with(me.sign_priv) finally: conn.close() @@ -552,56 +760,93 @@ class MeshWireClient: *, encrypt: bool = False, ) -> dict[str, Any]: - """Build a signed envelope. When `encrypt` is True, AEAD-encrypt - the body under this peer's recovered epoch secret before signing. + """Build, chain-link, optionally encrypt, sign, and audit-log one + outbound envelope. - Sender-side secret recovery uses `recover_epoch_secret`, which - looks up the rotator from the audit chain and re-unwraps this - peer's slot — no separate sender-side cache is required. + Three responsibilities, in order: + + 1. **Confidentiality (optional).** When ``encrypt=True``, the + body is AEAD-encrypted under this peer's recovered epoch + secret. Sender-side recovery uses ``recover_epoch_secret`` + which re-unwraps our own slot — no sender-side cache. + 2. **Chain-of-claims.** Populate ``prev_event_hash`` from our + own ``latest_event_hash`` and compute ``event_hash`` so the + receiver can enforce per-peer chain consistency. + 3. **Local audit log.** Append a ``mesh_sent`` event whose body + is the envelope's canonical (event-hash-excluded) dict. + By construction its ``audit_event.event_hash`` equals the + envelope's ``event_hash``, unifying the wire chain-of-claims + with the local audit chain. ``latest_event_hash`` then + advances to this envelope's ``event_hash``, so back-to-back + sends extend a single linear chain. + + Drop the same dict shape into both wire and audit so a future + forensic replay only needs the audit body to reproduce the + envelope bytes (modulo signature). """ + from dataclasses import asdict as _asdict + conn = connect(self.db_path) try: me = load_identity(conn) epoch = current_epoch(conn) - secret: bytes | None = None + if me is None or epoch is None: + raise RuntimeError("mesh not initialized; call init_identity first") + + ts = int(time.time()) if encrypt: - if me is None or epoch is None: - raise RuntimeError( - "mesh not initialized; call init_identity first" - ) secret = recover_epoch_secret(conn, epoch_id=epoch) + encrypted = encrypt_body_for_envelope( + type_=type_, + sender_id=me.member_id, + epoch_id=epoch, + ts=ts, + body=body, + epoch_secret=secret, + ) + base_env = WireEnvelope( + type=type_, + sender_id=me.member_id, + epoch_id=epoch, + ts=ts, + body={}, # placeholder; real body lives in encrypted_body + encrypted_body=encrypted, + ) + else: + base_env = WireEnvelope( + type=type_, + sender_id=me.member_id, + epoch_id=epoch, + ts=ts, + body=body, + ) + + prev = latest_event_hash(conn) + env = base_env.with_chain(prev_event_hash=prev) + + audit_body = _asdict(env) + audit_body.pop("event_hash", None) + # Drop None-valued encrypted_body so cleartext audit bodies + # stay byte-identical to the v1 audit shape (mirrors + # canonical_bytes' behavior — keeps audit hash == event_hash + # for cleartext envelopes built by older callers). + if audit_body.get("encrypted_body") is None: + audit_body.pop("encrypted_body", None) + audit_event_hash = append_audit( + conn, + event_type="mesh_sent", + body=audit_body, + subject_root=str(body.get("document_root") or body.get("cache_key") or ""), + ts=env.ts, + ) + assert audit_event_hash == env.event_hash, ( + f"chain unification broke: audit hash {audit_event_hash} " + f"!= envelope event_hash {env.event_hash}" + ) + signed = env.sign_with(me.sign_priv) finally: conn.close() - if me is None or epoch is None: - raise RuntimeError("mesh not initialized; call init_identity first") - ts = int(time.time()) - if encrypt: - assert secret is not None # invariant from the branch above - encrypted = encrypt_body_for_envelope( - type_=type_, - sender_id=me.member_id, - epoch_id=epoch, - ts=ts, - body=body, - epoch_secret=secret, - ) - env = WireEnvelope( - type=type_, - sender_id=me.member_id, - epoch_id=epoch, - ts=ts, - body={}, # placeholder; real body is in encrypted_body - encrypted_body=encrypted, - ) - else: - env = WireEnvelope( - type=type_, - sender_id=me.member_id, - epoch_id=epoch, - ts=ts, - body=body, - ) - return env.sign_with(me.sign_priv) + return signed # ------------------------------------------------------------------ ops def info(self) -> dict[str, Any]: diff --git a/aborist/store.py b/aborist/store.py index 8972d1a..61f8980 100644 --- a/aborist/store.py +++ b/aborist/store.py @@ -258,6 +258,19 @@ CREATE TABLE IF NOT EXISTS mesh_epochs ( reason TEXT ); +-- Per-peer audit-chain tracking. Each row records the most recent +-- event_hash a given peer has broadcast to us; we enforce that every +-- subsequent gossip envelope carries `prev_event_hash == last_event_hash` +-- of that peer. A mismatch is a fork — the gossip is rejected (409). +-- last_seq is the local count of accepted envelopes from that peer +-- (informational; the canonical chain identity is last_event_hash). +CREATE TABLE IF NOT EXISTS mesh_peer_chains ( + peer_member_id TEXT PRIMARY KEY, + last_event_hash TEXT NOT NULL, + last_seq INTEGER NOT NULL, + last_seen_at INTEGER NOT NULL +); + -- FTS5 over chunk content for UNGROUNDED-mode keyword search. -- -- Contentless mode (`content=''`): FTS5 stores ONLY the inverted index, no @@ -296,6 +309,7 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: conn.row_factory = sqlite3.Row conn.executescript(SCHEMA_SQL) _migrate_audit_mode(conn) + _migrate_mesh_peer_chains(conn) conn.execute("PRAGMA synchronous = NORMAL") conn.execute("PRAGMA cache_size = -65536") conn.execute("PRAGMA temp_store = MEMORY") @@ -355,6 +369,30 @@ def _migrate_audit_mode(conn: sqlite3.Connection) -> None: ) +def _migrate_mesh_peer_chains(conn: sqlite3.Connection) -> None: + """Forward-migrate pre-mesh-fork-detection shards. + + Adds the `mesh_peer_chains` table to DBs that pre-date per-peer + audit-chain tracking on the mesh wire. CREATE TABLE IF NOT EXISTS + in SCHEMA_SQL covers brand-new shards; this migration is a belt- + and-suspenders idempotency check for callers that bypass the full + SCHEMA_SQL pass (cross-shard query views, etc.). Idempotent — + PRAGMA-checks before issuing CREATE. + """ + row = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='mesh_peer_chains'" + ).fetchone() + if row is None: + conn.execute( + "CREATE TABLE mesh_peer_chains (" + " peer_member_id TEXT PRIMARY KEY," + " last_event_hash TEXT NOT NULL," + " last_seq INTEGER NOT NULL," + " last_seen_at INTEGER NOT NULL" + ")" + ) + + def _rebuild_providence_cache_ungrounded(conn: sqlite3.Connection) -> None: """One-time table rebuild: rename audit_mode value VISUAL → UNGROUNDED. diff --git a/tests/test_mesh_chain.py b/tests/test_mesh_chain.py new file mode 100644 index 0000000..50f56b0 --- /dev/null +++ b/tests/test_mesh_chain.py @@ -0,0 +1,494 @@ +"""Mesh wire — per-peer audit-chain tracking. + +Receiver-side enforcement that gossip envelopes from a given peer +extend a single linear chain of claims. Catches: +- Tampered ``event_hash`` (recomputation mismatch -> 400). +- Forks / replays where ``prev_event_hash`` doesn't match the last + envelope we accepted from that peer (-> 409). +- Cross-talk between peers' chains (alice's broadcasts must not + affect bob's chain state and vice versa). + +Implements the spec from ``docs/mesh.md``: + > Audit chains merge by prev_event_hash. A gossip insert that does + > not extend a chain consistently is rejected — mesh never silently + > forks history. +""" + +from __future__ import annotations + +import time +from pathlib import Path + +import pytest + +from aborist.mesh import init_identity +from aborist.mesh.crypto import sign as _sign +from aborist.mesh.members import add_member +from aborist.mesh.wire import ( + MeshWireServer, + TYPE_ANNOUNCE_ROOT, + WireEnvelope, +) +from aborist.store import connect + + +def _bootstrap_three_peer_db(tmp_path: Path): + """Alice (admin) + bob + carol on alice's DB, both enrolled at the + same current_epoch (each ``add_member`` bumps the epoch by one). + + Returns (alice_db, current_epoch, alice_id, bob_id, carol_id). Alice's + roster at ``current_epoch`` contains bob and carol so signature + verification finds their sign_pubs when receiving from either. + """ + from aborist.mesh.state import current_epoch as _current_epoch + + alice_db = tmp_path / "alice.db" + bob_db = tmp_path / "bob.db" + carol_db = tmp_path / "carol.db" + + a_conn = connect(alice_db) + try: + alice = init_identity(a_conn, group_name="t", member_id="alice") + finally: + a_conn.close() + + b_conn = connect(bob_db) + try: + bob = init_identity(b_conn, group_name="t", member_id="bob") + finally: + b_conn.close() + + c_conn = connect(carol_db) + try: + carol = init_identity(c_conn, group_name="t", member_id="carol") + finally: + c_conn.close() + + a_conn = connect(alice_db) + try: + add_member(a_conn, member_id="bob", sign_pub=bob.sign_pub, dh_pub=bob.dh_pub) + add_member(a_conn, member_id="carol", sign_pub=carol.sign_pub, dh_pub=carol.dh_pub) + epoch = _current_epoch(a_conn) + finally: + a_conn.close() + + return alice_db, epoch, alice, bob, carol + + +def _signed_envelope( + *, + sign_priv: bytes, + sender_id: str, + body: dict, + prev_event_hash: str | None, + epoch_id: int, + ts: int | None = None, +) -> tuple[WireEnvelope, dict]: + """Build a chain-linked, signed envelope. Returns (env, signed_dict).""" + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id=sender_id, + epoch_id=epoch_id, + ts=int(time.time()) if ts is None else ts, + body=body, + ).with_chain(prev_event_hash=prev_event_hash) + return env, env.sign_with(sign_priv) + + +def _root_for(seed: str) -> str: + """Deterministic 64-char hex doc-root marker for tests (not a real hash).""" + return (seed * 64)[:64] + + +# --------------------------------------------------------------------------- +# Sender-side determinism +# --------------------------------------------------------------------------- + + +def test_with_chain_produces_deterministic_event_hash(): + """Same envelope + same prev_event_hash -> same event_hash, every time.""" + body = { + "document_root": _root_for("a"), + "source_uri": "https://x", + "chunking_version": "tok-512-v1", + "canonicalization_version": "norm-v1", + "schema_version": "v9.8.0", + } + env_base = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=1700_000_000, + body=body, + ) + a = env_base.with_chain(prev_event_hash=None) + b = env_base.with_chain(prev_event_hash=None) + assert a.event_hash == b.event_hash + assert a.event_hash is not None + assert len(a.event_hash) == 64 + + # Different prev -> different hash. + c = env_base.with_chain(prev_event_hash="aa" * 32) + assert c.event_hash != a.event_hash + + +def test_event_hash_excluded_from_canonical_bytes(): + """canonical_bytes() must NOT contain event_hash (avoids circularity). + + The literal hex of event_hash also must not appear, since the JSON + serialization is the only path it could leak through. (The substring + ``"event_hash"`` does appear because ``prev_event_hash`` is included; + that's expected — only the *value* of event_hash is excluded.) + """ + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=0, + body={"document_root": _root_for("a")}, + ).with_chain(prev_event_hash=None) + cb = env.canonical_bytes() + assert env.event_hash is not None + assert env.event_hash.encode() not in cb + # The "event_hash" key itself is NOT serialized — only "prev_event_hash" is. + assert b'"event_hash"' not in cb + assert b'"prev_event_hash"' in cb + + +# --------------------------------------------------------------------------- +# Receiver-side: first envelope, chain extension, fork detection +# --------------------------------------------------------------------------- + + +def test_first_announce_with_prev_none_accepted(tmp_path): + """A peer's very first envelope, prev_event_hash=None, is accepted.""" + alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path) + _env, signed = _signed_envelope( + sign_priv=bob.sign_priv, + sender_id="bob", + body={"document_root": _root_for("b"), "source_uri": "https://x"}, + prev_event_hash=None, + epoch_id=epoch, + ) + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + finally: + srv.stop() + assert status == 200, body + + a_conn = connect(alice_db) + try: + row = a_conn.execute( + "SELECT last_event_hash, last_seq FROM mesh_peer_chains " + "WHERE peer_member_id = 'bob'" + ).fetchone() + finally: + a_conn.close() + assert row is not None + assert row["last_event_hash"] == _env.event_hash + assert row["last_seq"] == 1 + + +def test_chain_extension_accepted(tmp_path): + """Second envelope with prev=first.event_hash extends the chain.""" + alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path) + env1, signed1 = _signed_envelope( + sign_priv=bob.sign_priv, + sender_id="bob", + body={"document_root": _root_for("1"), "source_uri": "https://x1"}, + prev_event_hash=None, + epoch_id=epoch, + ) + env2, signed2 = _signed_envelope( + sign_priv=bob.sign_priv, + sender_id="bob", + body={"document_root": _root_for("2"), "source_uri": "https://x2"}, + prev_event_hash=env1.event_hash, + epoch_id=epoch, + ) + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + s1, _ = srv.handle_announce(signed1) + s2, _ = srv.handle_announce(signed2) + finally: + srv.stop() + assert s1 == 200 + assert s2 == 200 + + a_conn = connect(alice_db) + try: + row = a_conn.execute( + "SELECT last_event_hash, last_seq FROM mesh_peer_chains " + "WHERE peer_member_id = 'bob'" + ).fetchone() + finally: + a_conn.close() + assert row["last_event_hash"] == env2.event_hash + assert row["last_seq"] == 2 + + +def test_fork_detection_rejects_409(tmp_path): + """Second envelope with prev=GARBAGE is rejected — chain forked.""" + alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path) + env1, signed1 = _signed_envelope( + sign_priv=bob.sign_priv, + sender_id="bob", + body={"document_root": _root_for("1"), "source_uri": "https://x"}, + prev_event_hash=None, + epoch_id=epoch, + ) + # Bob (or someone using bob's key) sends a second envelope claiming + # prev = arbitrary garbage instead of env1.event_hash. + _env2, signed2 = _signed_envelope( + sign_priv=bob.sign_priv, + sender_id="bob", + body={"document_root": _root_for("2"), "source_uri": "https://x"}, + prev_event_hash="ff" * 32, + epoch_id=epoch, + ) + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + s1, _ = srv.handle_announce(signed1) + s2, body = srv.handle_announce(signed2) + finally: + srv.stop() + assert s1 == 200 + assert s2 == 409 + assert "fork detected" in body["error"] + assert "bob" in body["error"] + + # The fork rejection must NOT advance the peer's tracked chain. + a_conn = connect(alice_db) + try: + row = a_conn.execute( + "SELECT last_event_hash, last_seq FROM mesh_peer_chains " + "WHERE peer_member_id = 'bob'" + ).fetchone() + finally: + a_conn.close() + assert row["last_event_hash"] == env1.event_hash + assert row["last_seq"] == 1 + + +def test_tampered_event_hash_rejected_400(tmp_path): + """An envelope whose event_hash doesn't match recomputation is 400.""" + alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path) + env, signed = _signed_envelope( + sign_priv=bob.sign_priv, + sender_id="bob", + body={"document_root": _root_for("a"), "source_uri": "https://x"}, + prev_event_hash=None, + epoch_id=epoch, + ) + # Tamper with the event_hash field without re-signing. The envelope + # signature itself remains valid (event_hash is excluded from the + # signed canonical bytes), so the receiver catches this at the + # event_hash recomputation step, not the signature step. + signed["envelope"]["event_hash"] = "deadbeef" * 8 + assert signed["envelope"]["event_hash"] != env.event_hash + + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + finally: + srv.stop() + assert status == 400 + assert "event_hash mismatch" in body["error"] + + # No peer-chain row written for a rejected envelope. + a_conn = connect(alice_db) + try: + row = a_conn.execute( + "SELECT * FROM mesh_peer_chains WHERE peer_member_id = 'bob'" + ).fetchone() + finally: + a_conn.close() + assert row is None + + +def test_chain_tracking_advances_across_multiple_announces(tmp_path): + """Receiver tracks last_event_hash + last_seq correctly over N announces.""" + alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path) + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + prev = None + last_eh = None + for i in range(5): + env, signed = _signed_envelope( + sign_priv=bob.sign_priv, + sender_id="bob", + body={"document_root": _root_for(str(i)), "source_uri": f"https://x/{i}"}, + prev_event_hash=prev, + epoch_id=epoch, + ) + status, _ = srv.handle_announce(signed) + assert status == 200, f"announce {i} failed" + prev = env.event_hash + last_eh = env.event_hash + finally: + srv.stop() + + a_conn = connect(alice_db) + try: + row = a_conn.execute( + "SELECT last_event_hash, last_seq FROM mesh_peer_chains " + "WHERE peer_member_id = 'bob'" + ).fetchone() + finally: + a_conn.close() + assert row["last_event_hash"] == last_eh + assert row["last_seq"] == 5 + + +def test_two_peers_have_independent_chains(tmp_path): + """Bob's announces don't affect carol's chain state and vice versa.""" + alice_db, epoch, _alice, bob, carol = _bootstrap_three_peer_db(tmp_path) + + # Bob: 2 envelopes + bob_env1, bob_signed1 = _signed_envelope( + sign_priv=bob.sign_priv, + sender_id="bob", + body={"document_root": _root_for("b1"), "source_uri": "https://b1"}, + prev_event_hash=None, + epoch_id=epoch, + ) + bob_env2, bob_signed2 = _signed_envelope( + sign_priv=bob.sign_priv, + sender_id="bob", + body={"document_root": _root_for("b2"), "source_uri": "https://b2"}, + prev_event_hash=bob_env1.event_hash, + epoch_id=epoch, + ) + # Carol: 1 envelope, prev=None (her chain starts fresh — should be + # accepted even though bob has already posted). + carol_env1, carol_signed1 = _signed_envelope( + sign_priv=carol.sign_priv, + sender_id="carol", + body={"document_root": _root_for("c1"), "source_uri": "https://c1"}, + prev_event_hash=None, + epoch_id=epoch, + ) + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + assert srv.handle_announce(bob_signed1)[0] == 200 + assert srv.handle_announce(carol_signed1)[0] == 200 # carol fresh start + assert srv.handle_announce(bob_signed2)[0] == 200 # bob extends his own chain + finally: + srv.stop() + + a_conn = connect(alice_db) + try: + bob_row = a_conn.execute( + "SELECT last_event_hash, last_seq FROM mesh_peer_chains " + "WHERE peer_member_id = 'bob'" + ).fetchone() + carol_row = a_conn.execute( + "SELECT last_event_hash, last_seq FROM mesh_peer_chains " + "WHERE peer_member_id = 'carol'" + ).fetchone() + finally: + a_conn.close() + assert bob_row["last_event_hash"] == bob_env2.event_hash + assert bob_row["last_seq"] == 2 + assert carol_row["last_event_hash"] == carol_env1.event_hash + assert carol_row["last_seq"] == 1 + + +# --------------------------------------------------------------------------- +# Schema migration: peer-chains table appears on legacy DBs +# --------------------------------------------------------------------------- + + +def test_mesh_peer_chains_table_exists_after_connect(tmp_path): + """Fresh connect() creates mesh_peer_chains via SCHEMA_SQL + migration.""" + db = tmp_path / "fresh.db" + conn = connect(db) + try: + row = conn.execute( + "SELECT name FROM sqlite_master " + "WHERE type='table' AND name='mesh_peer_chains'" + ).fetchone() + finally: + conn.close() + assert row is not None + + +def test_mesh_peer_chains_migration_idempotent(tmp_path): + """Two connects in a row don't ALTER twice.""" + db = tmp_path / "twice.db" + connect(db).close() + conn = connect(db) + try: + cnt = conn.execute( + "SELECT COUNT(*) FROM sqlite_master " + "WHERE type='table' AND name='mesh_peer_chains'" + ).fetchone()[0] + finally: + conn.close() + assert cnt == 1 + + +def test_legacy_db_without_peer_chains_migrates(tmp_path): + """A DB created before the peer-chain rollout gains the table on open.""" + import sqlite3 as _s + db = tmp_path / "legacy.db" + raw = _s.connect(db) + raw.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)") + raw.commit() + raw.close() + + # Pre-state: no mesh_peer_chains. + raw = _s.connect(db) + pre = raw.execute( + "SELECT name FROM sqlite_master WHERE name='mesh_peer_chains'" + ).fetchone() + raw.close() + assert pre is None + + # Open via connect() — runs SCHEMA_SQL + migration. + conn = connect(db) + try: + post = conn.execute( + "SELECT name FROM sqlite_master WHERE name='mesh_peer_chains'" + ).fetchone() + finally: + conn.close() + assert post is not None + + +# --------------------------------------------------------------------------- +# Backward compat: legacy envelopes (no chain fields) still accepted +# --------------------------------------------------------------------------- + + +def test_legacy_envelope_without_chain_fields_accepted(tmp_path): + """A WireEnvelope with event_hash=None (legacy fixture path) is accepted + without per-peer chain enforcement. Sig check still runs.""" + alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path) + # Built directly without with_chain() — both prev and event_hash are None. + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="bob", + epoch_id=epoch, + ts=int(time.time()), + body={"document_root": _root_for("a"), "source_uri": "https://x"}, + ) + assert env.event_hash is None + signed = env.sign_with(bob.sign_priv) + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + status, _ = srv.handle_announce(signed) + finally: + srv.stop() + assert status == 200 + + # Legacy path doesn't write a peer-chain row. + a_conn = connect(alice_db) + try: + row = a_conn.execute( + "SELECT * FROM mesh_peer_chains WHERE peer_member_id='bob'" + ).fetchone() + finally: + a_conn.close() + assert row is None