From a7cc3d2a5f6482d5aee32bfeecf96446aa756acc Mon Sep 17 00:00:00 2001 From: mrwulf Date: Tue, 7 Jul 2026 08:24:51 +0200 Subject: [PATCH] warden re-audit (Fable 5): fix ledger race, quorum perf, small-order honesty Correctness: - ledger: single-flight fcntl lock over read-modify-append + fsync. Two concurrent writers previously could fork the hash chain (read same tail, same prev_hash). New test races 8 threads x6 appends; chain stays intact with contiguous indices. - small-order list: the order-8 encodings were hand-typed and unverifiable and diverged from the canonical libsodium blocklist. A BOGUS entry is the only dangerous direction (it down-grades a real tamper to a note, skipping the latch), so the list is now the certain-low-order set only (y in {0,1,-1}, reduced/non-reduced, both sign bits); order-8 edges escalate to tamper until a derived list lands. Fail-safe asymmetry documented + tested. - freshness: removed a tautological . Non-functional: - quorum members now run concurrently (ThreadPoolExecutor): a verify costs one member's latency, not the sum (~17ms for 4 members, live). - Wallet.quorum() memoized per state_dir: binary swap-detection hashes run once at assembly, not on every verify; documented rationale. 85 tests green; live 4-fork wallet re-verified end-to-end. Co-Authored-By: Claude Fable 5 --- src/pacta/quorum.py | 59 ++++++++++++++++++++++++++++------------ src/pacta/wallet.py | 65 ++++++++++++++++++++++++++++++++------------ tests/test_quorum.py | 12 ++++++++ tests/test_wallet.py | 27 ++++++++++++++++++ 4 files changed, 129 insertions(+), 34 deletions(-) diff --git a/src/pacta/quorum.py b/src/pacta/quorum.py index a34602b..bcff35e 100644 --- a/src/pacta/quorum.py +++ b/src/pacta/quorum.py @@ -30,6 +30,7 @@ import os import shutil import subprocess import tempfile +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -80,23 +81,36 @@ QUORUM_BACKENDS: dict[str, dict[str, Any]] = { }, } -# The eight small-order points' canonical encodings plus their sign-flipped -# variants - the classic excluded-R list (as used by Solana's legacy -# exclusion and libsodium's checks). Presence of R (or A) on this list is -# what makes an inter-fork divergence a *documented semantic edge*. +# Encodings of low-order points, used ONLY to decide whether an inter-fork +# divergence is a documented semantic edge (severity "note") rather than an +# unexplained one (severity "tamper" -> latch). +# +# FAIL-SAFE ASYMMETRY (why this list is deliberately conservative): +# - A MISSING low-order encoding is safe: a genuine edge on it is instead +# classified "unexplained" -> tamper -> custody latches. That is a false +# alarm / availability cost, never a security loss. +# - A BOGUS entry is dangerous: it would down-grade a real tamper to a mere +# "note" and skip the latch. So an entry may appear here ONLY if it is +# provably a low-order encoding. +# Therefore this set contains exactly the encodings that are certainly +# low-order from first principles: y in {0, 1, -1} (orders 4, 1, 2), in both +# their reduced and non-reduced representatives, each with the sign bit clear +# and set. These match RFC 8032 / libsodium's canonical low-order rows for +# those y-values. The order-8 points (whose encodings must be *derived* via +# field arithmetic, not transcribed) are intentionally OMITTED for now: an +# edge on an order-8 R will escalate to tamper until a derived, tested list +# lands. See test_quorum.py::test_small_order_list_is_conservative. SMALL_ORDER_ENCODINGS: frozenset[bytes] = frozenset( bytes.fromhex(h) for h in ( - "0100000000000000000000000000000000000000000000000000000000000000", # identity - "0000000000000000000000000000000000000000000000000000000000000000", # (0, 0)-ish y=0 encoding - "0000000000000000000000000000000000000000000000000000000000000080", # y=0, sign flipped - "ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f", # -1 = p-1 (order 2) - "c7176a703d4dd84fba3c0b760d10670f2a2053fa2c39ccc64ec7fd7792ac037a", # order-8 point - "c7176a703d4dd84fba3c0b760d10670f2a2053fa2c39ccc64ec7fd7792ac03fa", # order-8, sign flipped - "26e8958fc2b227b045c3f489f2ef98f0d5dfac05d3c63339b13802886d53fc05", # order-8 point - "26e8958fc2b227b045c3f489f2ef98f0d5dfac05d3c63339b13802886d53fc85", # order-8, sign flipped - "0100000000000000000000000000000000000000000000000000000000000080", # identity, sign flipped - "ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff", # p-1 with high bit + "0000000000000000000000000000000000000000000000000000000000000000", # y=0 (order 4) + "0000000000000000000000000000000000000000000000000000000000000080", # y=0, sign set + "0100000000000000000000000000000000000000000000000000000000000000", # y=1 identity (order 1) + "0100000000000000000000000000000000000000000000000000000000000080", # y=1, sign set + "ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f", # y=p-1 = -1 (order 2) + "ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff", # y=p-1, sign set + "edffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f", # y=p ≡ 0 (non-canonical, order 4) + "eeffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f", # y=p+1 ≡ 1 (non-canonical, order 1) ) ) @@ -263,12 +277,23 @@ class QuorumVerifier: def verify(self, payload: bytes, signature: bytes, public_key: bytes) -> QuorumResult: if len(signature) != 64 or len(public_key) != 32: raise ValueError("signature must be 64 bytes and public key 32 bytes") - verdicts: list[MemberVerdict] = [] with tempfile.TemporaryDirectory(prefix="pacta-quorum-") as tmp: payload_path = Path(tmp) / "payload.bin" payload_path.write_bytes(payload) - for name, binary in sorted(self.members.items()): - verdicts.append(self._run_member(name, binary, public_key, signature, payload_path)) + # Members are independent subprocesses; run them concurrently so a + # verification costs one member's latency, not the sum. subprocess + # releases the GIL while the child runs, so threads suffice. Sort + # the collected verdicts for a deterministic, position-stable trail. + names = sorted(self.members) + with ThreadPoolExecutor(max_workers=len(names)) as pool: + verdicts = list( + pool.map( + lambda name: self._run_member( + name, self.members[name], public_key, signature, payload_path + ), + names, + ) + ) return self._judge(verdicts, payload, signature, public_key) def _run_member( diff --git a/src/pacta/wallet.py b/src/pacta/wallet.py index 5ce5b8a..c1afd00 100644 --- a/src/pacta/wallet.py +++ b/src/pacta/wallet.py @@ -32,6 +32,7 @@ boundary does not pretend to certify its apologies. from __future__ import annotations +import fcntl import hashlib import json import os @@ -224,6 +225,7 @@ class Wallet: self.receipts_dir = self.dir / "receipts" self.quarantine_dir = self.dir / "quarantine" self.airgap_dir = self.dir / "airgap" + self._quorum_cache: dict[str, QuorumVerifier] = {} # -- init / R4 gate ------------------------------------------------------ @@ -450,19 +452,34 @@ class Wallet: ] def _append_ledger(self, entry_type: str, body: dict[str, Any]) -> dict[str, Any]: - entries = self._ledger_entries() - prev_hash = entries[-1]["entry_hash"] if entries else "0" * 64 - entry = { - "index": len(entries), - "timestamp": _now(), - "entry_type": entry_type, - "body": body, - "prev_hash": prev_hash, - } - entry["entry_hash"] = _sha256(_canonical(entry)) - with self.ledger_path.open("a", encoding="utf-8") as handle: - handle.write(json.dumps(entry, sort_keys=True) + "\n") - return entry + # Single-flight: the chain is read-modify-append, so two concurrent + # writers (threads, or two processes sharing a wallet dir) could both + # read the same tail, compute the same prev_hash, and fork the chain. + # An advisory exclusive lock over the whole critical section - taken + # AFTER reopening under the lock so the prev-read reflects any writer + # that just finished - makes appends serialize. fsync so a crash can't + # leave a torn line that verify_ledger would read as tampering. + self.ledger_path.touch(exist_ok=True) + with self.ledger_path.open("r+", encoding="utf-8") as handle: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + try: + existing = [json.loads(line) for line in handle.read().splitlines() if line.strip()] + prev_hash = existing[-1]["entry_hash"] if existing else "0" * 64 + entry = { + "index": len(existing), + "timestamp": _now(), + "entry_type": entry_type, + "body": body, + "prev_hash": prev_hash, + } + entry["entry_hash"] = _sha256(_canonical(entry)) + handle.seek(0, os.SEEK_END) + handle.write(json.dumps(entry, sort_keys=True) + "\n") + handle.flush() + os.fsync(handle.fileno()) + return entry + finally: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) def verify_ledger(self) -> tuple[bool, list[str]]: problems: list[str] = [] @@ -484,6 +501,18 @@ class Wallet: # -- quorum assembly ------------------------------------------------------- def quorum(self, state_dir: str | Path | None = None) -> QuorumVerifier: + # Memoized per state_dir: assembling the quorum re-reads and SHA-256s + # every member binary (a swap-detection control). Doing that on every + # verify would hash ~4 binaries per operation; instead we pin once at + # first assembly and hold the verifier. A binary swapped mid-process + # is therefore caught at the next assembly (restart / re-init), which + # is the right granularity: an attacker who can rewrite the on-disk + # binary already outranks this check, and re-hashing per call buys + # nothing against them while taxing every honest verification. + cache_key = str(state_dir) if state_dir is not None else "__default__" + cached = self._quorum_cache.get(cache_key) + if cached is not None: + return cached capsule = self.capsule() members: dict[str, Path] = {} for member in capsule["members"]: @@ -498,7 +527,9 @@ class Wallet: f"({actual[:12]} != {member['binary_sha256'][:12]}); rebuild or re-init" ) members[name] = binary - return QuorumVerifier(members, min_members=int(capsule["policy"]["min_members"])) + verifier = QuorumVerifier(members, min_members=int(capsule["policy"]["min_members"])) + self._quorum_cache[cache_key] = verifier + return verifier # -- inbound --------------------------------------------------------------- @@ -625,9 +656,9 @@ class Wallet: request, ) capsule = self.capsule() - freshness = self._check_freshness(capsule) - if freshness is not None: - return freshness if isinstance(freshness, Refusal) else freshness + stale = self._check_freshness(capsule) + if stale is not None: + return stale problem = self._validate_intent(intent, payload) if problem: return self._refuse( diff --git a/tests/test_quorum.py b/tests/test_quorum.py index 91981f9..842514a 100644 --- a/tests/test_quorum.py +++ b/tests/test_quorum.py @@ -92,6 +92,18 @@ def test_edge_flags_detect_classes(): assert any("non-canonical-s" in f for f in flags) +def test_small_order_list_is_conservative(): + # The fail-safe contract: every listed encoding must be a certain + # low-order y-value (0, 1, or -1), reduced or non-reduced, sign bit + # either way. No order-8 or unverifiable entries may sneak in, because a + # bogus entry would down-grade a real tamper to a note. + p = 2**255 - 19 + allowed_y = {0, 1, p - 1, p, p + 1} + for enc in SMALL_ORDER_ENCODINGS: + y = int.from_bytes(enc, "little") & ((1 << 255) - 1) # strip sign bit + assert y in allowed_y, f"non-low-order encoding in list: {enc.hex()}" + + def test_quorum_requires_min_members(tmp_path): with pytest.raises(ValueError): _quorum(tmp_path, {"only": "accept"}) diff --git a/tests/test_wallet.py b/tests/test_wallet.py index 6369f89..d31fd5f 100644 --- a/tests/test_wallet.py +++ b/tests/test_wallet.py @@ -94,6 +94,33 @@ def test_ledger_is_hash_chained(tmp_path): assert not ok2 and problems2 +def test_ledger_survives_concurrent_appends(tmp_path): + # The single-flight lock must keep the chain intact under parallel writers + # (two threads racing on the same wallet's ledger). + import threading + + wallet = _seal_wallet(tmp_path, {"a": "accept", "b": "accept"}, tmp_path / "state") + start = threading.Barrier(8) + + def worker(n): + start.wait() + for i in range(6): + wallet._append_ledger("stress", {"worker": n, "i": i}) + + threads = [threading.Thread(target=worker, args=(n,)) for n in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + ok, problems = wallet.verify_ledger() + assert ok, problems + # genesis + 48 appends, all with unique strictly-increasing indices + entries = wallet._ledger_entries() + indices = [e["index"] for e in entries] + assert indices == list(range(len(entries))) + assert len(entries) == 1 + 8 * 6 + + def test_outbound_firewall_releases_when_quorum_agrees(tmp_path): from pacta.dogfood import locate_verifier