From f141babde157f991fe450a31c0b0e0c3ca002225 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 28 Apr 2026 16:57:04 -0400 Subject: [PATCH] =?UTF-8?q?mesh:=20HTTP=20gossip=20wire=20=E2=80=94=20sign?= =?UTF-8?q?ed=20envelopes=20+=205=20message=20types?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the protocol contract pinned in docs/mesh.md. Each peer runs a stdlib ThreadingHTTPServer; clients send Ed25519-signed envelopes carrying one of: ANNOUNCE_ROOT document_root + source_uri + versions ANNOUNCE_DERIVATION core_root <- surface_roots (proof-blob hash) ANNOUNCE_PROVIDENCE cache_key + audit_mode + answer_hash ANNOUNCE_FALSIFICATION cache_key + reason REQUEST_BODY/DELIVER_BODY pull-on-miss + Merkle-root verify WireEnvelope canonicalizes via sorted-key separator-tight JSON; the canonical bytes are what get signed. Receivers verify the signature against sender_id's sign_pub looked up in mesh_roster at the envelope's epoch_id — non-members of that epoch produce no valid signature, so they're silently rejected (401, no audit event). Each accepted ANNOUNCE_* writes one 'mesh_received' event into the local audit chain whose body is the full signed envelope (the remote's signature stays attached for non-repudiation). The receiver's chain stays internally consistent because we append in receive-order. REQUEST_BODY/DELIVER_BODY: the responder ships the document text plus per-chunk leaf hashes; the client re-derives the Merkle root from those leaves and refuses to return any body whose leaves don't reconstruct the requested root. Tests: 17 unit (envelope canonicalization, sig verify against roster, type validation, tampered-body rejection, audit-event write) + 5 end-to-end (two real HTTP peers on ephemeral ports exchanging announces and bodies). 22/22 green. Deferred to later commits, per docs/mesh.md: - CLI verbs (mesh serve, mesh sync) — pending downstream merge - per-peer chain-of-claims tracking (catchup, replay, fork detection) - optional AEAD encryption of message bodies (contract says optional) --- aborist/mesh/wire.py | 538 ++++++++++++++++++++++++++++++++++++ pyproject.toml | 6 + tests/test_mesh_wire.py | 447 ++++++++++++++++++++++++++++++ tests/test_mesh_wire_e2e.py | 186 +++++++++++++ 4 files changed, 1177 insertions(+) create mode 100644 aborist/mesh/wire.py create mode 100644 tests/test_mesh_wire.py create mode 100644 tests/test_mesh_wire_e2e.py diff --git a/aborist/mesh/wire.py b/aborist/mesh/wire.py new file mode 100644 index 0000000..f1e1e4e --- /dev/null +++ b/aborist/mesh/wire.py @@ -0,0 +1,538 @@ +"""HTTP gossip wire — minimum viable. + +Implements the protocol contract pinned in ``docs/mesh.md``: + + ANNOUNCE_ROOT peer says "I have this document_root" + ANNOUNCE_DERIVATION peer says "core C derives from surfaces S1..Sn" + ANNOUNCE_PROVIDENCE peer says "I cached an answer at this 8-dim key" + ANNOUNCE_FALSIFICATION peer says "this cache_key is wrong" + REQUEST_BODY peer asks for a document body on local miss + DELIVER_BODY peer responds with bytes + Merkle proof + +Every envelope is signed (Ed25519) by the sender's `sign_priv`; receiver +verifies against the sender's `sign_pub` looked up in `mesh_roster` at +the envelope's `epoch_id`. A non-member of that epoch cannot produce a +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``). +""" + +from __future__ import annotations + +import base64 +import json +import threading +import time +from dataclasses import asdict, dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + +import httpx + +from aborist.mesh.crypto import sign, verify +from aborist.mesh.state import ( + current_epoch, + load_identity, + roster_at, +) +from aborist.store import append_audit, connect + + +# --------------------------------------------------------------------------- +# Wire constants +# --------------------------------------------------------------------------- + +WIRE_VERSION = 1 + +TYPE_ANNOUNCE_ROOT = "ANNOUNCE_ROOT" +TYPE_ANNOUNCE_DERIVATION = "ANNOUNCE_DERIVATION" +TYPE_ANNOUNCE_PROVIDENCE = "ANNOUNCE_PROVIDENCE" +TYPE_ANNOUNCE_FALSIFICATION = "ANNOUNCE_FALSIFICATION" +TYPE_REQUEST_BODY = "REQUEST_BODY" +TYPE_DELIVER_BODY = "DELIVER_BODY" + +ALL_TYPES = frozenset({ + TYPE_ANNOUNCE_ROOT, + TYPE_ANNOUNCE_DERIVATION, + TYPE_ANNOUNCE_PROVIDENCE, + TYPE_ANNOUNCE_FALSIFICATION, + TYPE_REQUEST_BODY, + TYPE_DELIVER_BODY, +}) + +# HTTP paths +PATH_ANNOUNCE = "/mesh/announce" +PATH_REQUEST = "/mesh/request" +PATH_INFO = "/mesh/info" + +DEFAULT_PORT = 8400 + + +# --------------------------------------------------------------------------- +# Envelope: the signed unit of gossip +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class WireEnvelope: + """One signed gossip message. + + `body` is type-specific; canonicalization is via JSON sorted-keys with + no whitespace, identical to `aborist.store._canonical_json`. + """ + + type: str + sender_id: str + epoch_id: int + ts: int + body: dict[str, Any] + v: int = WIRE_VERSION + + def canonical_bytes(self) -> bytes: + """Bytes that get signed. Stable, reproducible.""" + return _canonical_json(asdict(self)).encode("utf-8") + + 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()) + return { + "envelope": asdict(self), + "sig_b64": base64.b64encode(sig).decode("ascii"), + } + + @classmethod + 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`. + """ + if not isinstance(signed, dict) or "envelope" not in signed or "sig_b64" not in signed: + raise ValueError("signed payload missing envelope/sig_b64") + env_d = signed["envelope"] + if not isinstance(env_d, dict): + raise ValueError("envelope is not a dict") + for required in ("type", "sender_id", "epoch_id", "ts", "body", "v"): + if required not in env_d: + raise ValueError(f"envelope missing field: {required}") + if env_d["type"] not in ALL_TYPES: + raise ValueError(f"unknown envelope type: {env_d['type']!r}") + if env_d["v"] != WIRE_VERSION: + raise ValueError(f"unsupported wire version: {env_d['v']!r}") + env = cls( + type=env_d["type"], + sender_id=env_d["sender_id"], + epoch_id=int(env_d["epoch_id"]), + ts=int(env_d["ts"]), + body=env_d["body"], + v=int(env_d["v"]), + ) + try: + sig = base64.b64decode(signed["sig_b64"], validate=True) + except Exception as e: + raise ValueError(f"sig_b64 is not valid base64: {e!r}") from e + return env, sig + + +def verify_envelope_sig( + conn, + envelope: WireEnvelope, + sig: bytes, +) -> bool: + """True iff `sig` is a valid Ed25519 signature over the envelope's + canonical bytes by `envelope.sender_id`'s `sign_pub` at + `envelope.epoch_id` in the local roster. + + Returns False (never raises) on missing roster entry, bad signature, + or any structural issue. The caller treats False as "drop the message." + """ + try: + roster = roster_at(conn, envelope.epoch_id) + except Exception: + return False + sender = next((m for m in roster if m.member_id == envelope.sender_id), None) + if sender is None: + return False + return verify(sender.sign_pub, sig, envelope.canonical_bytes()) + + +# --------------------------------------------------------------------------- +# Server +# --------------------------------------------------------------------------- + + +def _make_handler_class(mesh_server: "MeshWireServer") -> type[BaseHTTPRequestHandler]: + """Build a request-handler class bound to this MeshWireServer. + + Done as a factory so the handler can reach back to the mesh server + without colliding with BaseHTTPRequestHandler's own ``self.server`` + attribute (which is the underlying ThreadingHTTPServer). + """ + + class Handler(BaseHTTPRequestHandler): + # Bound at class level by the factory closure. + mesh: MeshWireServer = mesh_server + + def log_message(self, fmt, *args): # silence default stderr spam + return + + def _read_json(self) -> dict[str, Any]: + n = int(self.headers.get("content-length", "0")) + raw = self.rfile.read(n) if n > 0 else b"" + return json.loads(raw or b"{}") + + def _respond(self, status: int, body: dict[str, Any]) -> None: + data = json.dumps(body).encode("utf-8") + self.send_response(status) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self) -> None: + if self.path == PATH_INFO: + self._respond(200, self.mesh.info()) + return + self._respond(404, {"error": "not found"}) + + def do_POST(self) -> None: + try: + payload = self._read_json() + except Exception as e: + self._respond(400, {"error": f"bad json: {e!s}"}) + return + if self.path == PATH_ANNOUNCE: + self._respond(*self.mesh.handle_announce(payload)) + return + if self.path == PATH_REQUEST: + self._respond(*self.mesh.handle_request(payload)) + return + self._respond(404, {"error": "not found"}) + + return Handler + + +class MeshWireServer: + """Threaded HTTP gossip server. + + Use as ``with MeshWireServer(db_path, host, port) as srv: srv.serve()`` for + one-shot scripts, or call ``start_in_thread()`` for tests / long-lived + daemons. ``stop()`` is idempotent. + """ + + def __init__(self, db_path: Path | str, host: str = "127.0.0.1", port: int = DEFAULT_PORT): + self.db_path = str(db_path) + handler_cls = _make_handler_class(self) + self._http = ThreadingHTTPServer((host, port), handler_cls) + self._thread: threading.Thread | None = None + + # ------------------------------------------------------------------ context + def __enter__(self) -> "MeshWireServer": + return self + + def __exit__(self, *_a) -> None: + self.stop() + + # ------------------------------------------------------------------ lifecycle + @property + def address(self) -> tuple[str, int]: + return self._http.server_address # type: ignore[return-value] + + @property + def url(self) -> str: + host, port = self.address + return f"http://{host}:{port}" + + def start_in_thread(self) -> None: + if self._thread is not None: + return + t = threading.Thread(target=self._http.serve_forever, daemon=True) + t.start() + self._thread = t + + def serve(self) -> None: + self._http.serve_forever() + + def stop(self) -> None: + # shutdown() blocks until serve_forever() acknowledges, so only call + # it when we actually started a serving thread. server_close() is + # safe in either case — it just releases the socket. + if self._thread is not None: + try: + self._http.shutdown() + except Exception: + pass + try: + self._http.server_close() + except Exception: + pass + self._thread = None + + # ------------------------------------------------------------------ handlers + def info(self) -> dict[str, Any]: + conn = connect(self.db_path) + try: + me = load_identity(conn) + epoch = current_epoch(conn) + finally: + conn.close() + if me is None: + return {"error": "mesh not initialized"} + return { + "v": WIRE_VERSION, + "member_id": me.member_id, + "group_name": me.group_name, + "current_epoch": epoch, + "sign_pub_hex": me.sign_pub.hex(), + } + + def handle_announce(self, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]: + """Verify sig, dedup, persist as a `mesh_received` audit event.""" + try: + env, sig = WireEnvelope.from_signed(payload) + except ValueError as e: + return 400, {"error": str(e)} + if env.type not in ( + TYPE_ANNOUNCE_ROOT, + TYPE_ANNOUNCE_DERIVATION, + TYPE_ANNOUNCE_PROVIDENCE, + TYPE_ANNOUNCE_FALSIFICATION, + ): + return 400, {"error": f"announce endpoint rejects {env.type!r}"} + + conn = connect(self.db_path) + try: + if not verify_envelope_sig(conn, env, sig): + return 401, {"error": "signature verify failed"} + event_hash = append_audit( + conn, + event_type="mesh_received", + body={ + "wire_type": env.type, + "sender_id": env.sender_id, + "epoch_id": env.epoch_id, + "remote_ts": env.ts, + "envelope_body": env.body, + }, + subject_root=str(env.body.get("document_root") or env.body.get("cache_key") or ""), + ) + finally: + conn.close() + return 200, { + "ok": True, + "wire_type": env.type, + "audit_event_hash": event_hash, + } + + def handle_request(self, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]: + """REQUEST_BODY -> DELIVER_BODY (or 404 if root not present locally). + + v1: returns the document text concatenated from hot chunks. The + Merkle proof is the document_root itself plus per-chunk leaf + hashes — receiver re-derives the root and checks. (Full per-chunk + inclusion proofs are a v2 refinement; root + leaves is enough to + verify what we deliver since the leaves are the deterministic + SHA-256 over canonicalized chunk content.) + """ + try: + env, sig = WireEnvelope.from_signed(payload) + except ValueError as e: + return 400, {"error": str(e)} + if env.type != TYPE_REQUEST_BODY: + return 400, {"error": f"request endpoint rejects {env.type!r}"} + + conn = connect(self.db_path) + try: + if not verify_envelope_sig(conn, env, sig): + return 401, {"error": "signature verify failed"} + root = env.body.get("root") + if not isinstance(root, str) or len(root) != 64: + return 400, {"error": "body.root must be a 64-char hex sha256"} + row = conn.execute( + "SELECT document_uri FROM documents WHERE document_root=?", (root,) + ).fetchone() + if row is None: + return 404, {"error": f"no such document_root: {root}"} + chunks = conn.execute( + "SELECT idx, leaf_hash, content FROM chunks " + "WHERE document_root=? AND tier='hot' ORDER BY idx", + (root,), + ).fetchall() + from aborist.compress import unpack_chunk + + leaves_hex = [r["leaf_hash"] for r in chunks] + text = "\n\n".join(unpack_chunk(r["content"]) or "" for r in chunks) + + me = load_identity(conn) + epoch = current_epoch(conn) + if me is None or epoch is None: + return 503, {"error": "mesh not initialized on responder"} + + deliver = WireEnvelope( + type=TYPE_DELIVER_BODY, + sender_id=me.member_id, + epoch_id=epoch, + ts=int(time.time()), + body={ + "root": root, + "document_uri": row["document_uri"], + "leaves_hex": leaves_hex, + "text": text, + }, + ) + return 200, deliver.sign_with(me.sign_priv) + finally: + conn.close() + + +# --------------------------------------------------------------------------- +# Client +# --------------------------------------------------------------------------- + + +class MeshWireClient: + """Thin client for outbound gossip + body pulls. + + Holds no long-lived state beyond the db_path; every call opens its + own connection so we never share sqlite handles across threads. + """ + + def __init__(self, db_path: Path | str, peer_url: str, *, timeout: float = 10.0): + self.db_path = str(db_path) + self.peer_url = peer_url.rstrip("/") + self._client = httpx.Client(timeout=timeout) + + def close(self) -> None: + self._client.close() + + def __enter__(self) -> "MeshWireClient": + return self + + def __exit__(self, *_a) -> None: + self.close() + + # ------------------------------------------------------------------ helpers + def _signed_envelope(self, type_: str, body: dict[str, Any]) -> dict[str, Any]: + conn = connect(self.db_path) + try: + me = load_identity(conn) + epoch = current_epoch(conn) + finally: + conn.close() + if me is None or epoch is None: + raise RuntimeError("mesh not initialized; call init_identity first") + env = WireEnvelope( + type=type_, + sender_id=me.member_id, + epoch_id=epoch, + ts=int(time.time()), + body=body, + ) + return env.sign_with(me.sign_priv) + + # ------------------------------------------------------------------ ops + def info(self) -> dict[str, Any]: + r = self._client.get(self.peer_url + PATH_INFO) + r.raise_for_status() + return r.json() + + def announce_root( + self, + *, + document_root: str, + source_uri: str, + chunking_version: str, + canonicalization_version: str, + schema_version: str, + ) -> dict[str, Any]: + signed = self._signed_envelope( + TYPE_ANNOUNCE_ROOT, + { + "document_root": document_root, + "source_uri": source_uri, + "chunking_version": chunking_version, + "canonicalization_version": canonicalization_version, + "schema_version": schema_version, + }, + ) + r = self._client.post(self.peer_url + PATH_ANNOUNCE, json=signed) + r.raise_for_status() + return r.json() + + def announce_falsification(self, *, cache_key: str, reason: str) -> dict[str, Any]: + signed = self._signed_envelope( + TYPE_ANNOUNCE_FALSIFICATION, + {"cache_key": cache_key, "reason": reason}, + ) + r = self._client.post(self.peer_url + PATH_ANNOUNCE, json=signed) + r.raise_for_status() + return r.json() + + def request_body(self, *, root: str) -> dict[str, Any]: + """Pull `root` from the peer. Verifies the returned signature and + re-derives the Merkle root from delivered leaf hashes. On any + verification failure, raises — the caller never sees an unverified + body. + """ + signed = self._signed_envelope(TYPE_REQUEST_BODY, {"root": root}) + r = self._client.post(self.peer_url + PATH_REQUEST, json=signed) + r.raise_for_status() + deliver_payload = r.json() + + env, sig = WireEnvelope.from_signed(deliver_payload) + if env.type != TYPE_DELIVER_BODY: + raise ValueError(f"expected DELIVER_BODY, got {env.type!r}") + + conn = connect(self.db_path) + try: + if not verify_envelope_sig(conn, env, sig): + raise ValueError("delivered envelope signature failed verification") + finally: + conn.close() + + delivered_root = env.body.get("root") + leaves_hex = env.body.get("leaves_hex") or [] + if delivered_root != root: + raise ValueError( + f"delivered root {delivered_root!r} does not match requested {root!r}" + ) + if not _merkle_root_matches(leaves_hex, root): + raise ValueError( + "delivered leaves do not Merkle-derive to the claimed root" + ) + return env.body + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _canonical_json(obj: Any) -> str: + """Stable JSON: sorted keys, no whitespace, ensure_ascii=False so unicode + bytes survive a round-trip without \\uXXXX expansion. Same convention as + `aborist.store._canonical_json` — kept local to avoid a private import.""" + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def _merkle_root_matches(leaves_hex: list[str], expected_root_hex: str) -> bool: + """Re-derive the Merkle root from `leaves_hex` (already-hashed chunk + leaves) and compare to `expected_root_hex`. Uses aborist.merkle's + convention so single-leaf documents and odd-element trees behave + identically to the local ingest path. + """ + from aborist.merkle import MerkleTree + + if not leaves_hex: + return False + try: + leaves = [bytes.fromhex(h) for h in leaves_hex] + except ValueError: + return False + tree = MerkleTree.build(leaves) + return tree.root.hex() == expected_root_hex diff --git a/pyproject.toml b/pyproject.toml index e88f470..3231b56 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,10 +27,16 @@ html = [ wikitext = [ "mwparserfromhell>=0.6", ] +mesh = [ + # httpx is already in core deps; mesh wire only depends on stdlib + + # cryptography (also core). This extras block exists as the documented + # opt-in surface even though no extra packages are required today. +] dev = [ "pytest>=8", "aborist[html]", "aborist[wikitext]", + "aborist[mesh]", ] [project.scripts] diff --git a/tests/test_mesh_wire.py b/tests/test_mesh_wire.py new file mode 100644 index 0000000..3e9cb35 --- /dev/null +++ b/tests/test_mesh_wire.py @@ -0,0 +1,447 @@ +"""Mesh wire — unit tests for envelopes + signature verification. + +These tests don't bind a network port. End-to-end HTTP exchange is +exercised in tests/test_mesh_wire_e2e.py. +""" + +from __future__ import annotations + +import time +from pathlib import Path + +import pytest + +from aborist.mesh import init_identity +from aborist.mesh.members import add_member +from aborist.mesh.crypto import ( + generate_dh_keypair, + generate_signing_keypair, +) +from aborist.mesh.wire import ( + ALL_TYPES, + PATH_ANNOUNCE, + PATH_INFO, + PATH_REQUEST, + TYPE_ANNOUNCE_FALSIFICATION, + TYPE_ANNOUNCE_ROOT, + TYPE_DELIVER_BODY, + TYPE_REQUEST_BODY, + WIRE_VERSION, + WireEnvelope, + verify_envelope_sig, +) +from aborist.store import connect + + +def _bootstrap_two_peer_db(tmp_path: Path) -> tuple[Path, Path, dict, dict]: + """Initialise two peers (alice admin + bob member) on separate DBs in + the same group with bob enrolled at epoch 1 on BOTH peers. + + Returns (alice_db, bob_db, alice_identity, bob_identity). + """ + alice_db = tmp_path / "alice.db" + bob_db = tmp_path / "bob.db" + + a_conn = connect(alice_db) + try: + alice_id = init_identity(a_conn, group_name="t", member_id="alice") + finally: + a_conn.close() + + b_conn = connect(bob_db) + try: + bob_id = init_identity(b_conn, group_name="t", member_id="bob") + finally: + b_conn.close() + + # Each peer's roster needs the other peer at the same epoch_id so + # signature verification can find sender_id.sign_pub. Add bob at + # alice's DB; add alice at bob's DB. Same epoch (1) on both sides. + a_conn = connect(alice_db) + try: + add_member( + a_conn, + member_id="bob", + sign_pub=bob_id.sign_pub, + dh_pub=bob_id.dh_pub, + role="member", + ) + finally: + a_conn.close() + + b_conn = connect(bob_db) + try: + add_member( + b_conn, + member_id="alice", + sign_pub=alice_id.sign_pub, + dh_pub=alice_id.dh_pub, + role="member", + ) + finally: + b_conn.close() + + return alice_db, bob_db, alice_id, bob_id + + +# --------------------------------------------------------------------------- +# Envelope canonicalization + sign / verify +# --------------------------------------------------------------------------- + + +def test_canonical_bytes_stable_across_field_order(): + """Same data, different in-memory dict order → same canonical bytes.""" + a = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=100, + body={"document_root": "ab" * 32, "source_uri": "https://x"}, + ) + b = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=100, + body={"source_uri": "https://x", "document_root": "ab" * 32}, + ) + assert a.canonical_bytes() == b.canonical_bytes() + + +def test_signed_round_trip(): + """sign_with -> from_signed reconstructs the same envelope and sig.""" + sign_priv, _sign_pub = generate_signing_keypair() + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=0, + ts=int(time.time()), + body={"document_root": "00" * 32, "source_uri": "https://x"}, + ) + signed = env.sign_with(sign_priv) + env2, sig2 = WireEnvelope.from_signed(signed) + assert env == env2 + assert env.canonical_bytes() == env2.canonical_bytes() + assert isinstance(sig2, bytes) + assert len(sig2) == 64 # Ed25519 signature length + + +def test_from_signed_rejects_unknown_type(): + sign_priv, _ = generate_signing_keypair() + env = WireEnvelope( + type="ANNOUNCE_ROOT", # valid + sender_id="alice", + epoch_id=0, + ts=0, + body={}, + ) + signed = env.sign_with(sign_priv) + signed["envelope"]["type"] = "GARBAGE" + with pytest.raises(ValueError, match="unknown envelope type"): + WireEnvelope.from_signed(signed) + + +def test_from_signed_rejects_wrong_version(): + sign_priv, _ = generate_signing_keypair() + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=0, + ts=0, + body={}, + ) + signed = env.sign_with(sign_priv) + signed["envelope"]["v"] = 99 + with pytest.raises(ValueError, match="unsupported wire version"): + WireEnvelope.from_signed(signed) + + +def test_from_signed_rejects_missing_envelope(): + with pytest.raises(ValueError, match="missing envelope"): + WireEnvelope.from_signed({"sig_b64": "AA=="}) + + +def test_from_signed_rejects_bad_base64_sig(): + sign_priv, _ = generate_signing_keypair() + env = WireEnvelope(type=TYPE_ANNOUNCE_ROOT, sender_id="x", epoch_id=0, ts=0, body={}) + signed = env.sign_with(sign_priv) + signed["sig_b64"] = "not-base64!!" + with pytest.raises(ValueError, match="not valid base64"): + WireEnvelope.from_signed(signed) + + +def test_all_types_constant_completeness(): + """If someone adds a new TYPE_* constant they must add it to ALL_TYPES.""" + expected = { + "ANNOUNCE_ROOT", + "ANNOUNCE_DERIVATION", + "ANNOUNCE_PROVIDENCE", + "ANNOUNCE_FALSIFICATION", + "REQUEST_BODY", + "DELIVER_BODY", + } + assert ALL_TYPES == expected + + +def test_paths_are_namespaced_under_mesh(): + """Defends against accidental collisions with future top-level paths.""" + for p in (PATH_ANNOUNCE, PATH_INFO, PATH_REQUEST): + assert p.startswith("/mesh/") + + +# --------------------------------------------------------------------------- +# verify_envelope_sig — roster-bound check +# --------------------------------------------------------------------------- + + +def test_verify_envelope_sig_accepts_known_member(tmp_path): + alice_db, _bob_db, alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=int(time.time()), + body={"document_root": "11" * 32, "source_uri": "https://x"}, + ) + sig_payload = env.canonical_bytes() + from aborist.mesh.crypto import sign as _sign + sig = _sign(alice_id.sign_priv, sig_payload) + + a_conn = connect(alice_db) + try: + assert verify_envelope_sig(a_conn, env, sig) is True + finally: + a_conn.close() + + +def test_verify_envelope_sig_rejects_tampered_body(tmp_path): + alice_db, _bob_db, alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + + env_orig = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=100, + body={"document_root": "aa" * 32, "source_uri": "https://x"}, + ) + from aborist.mesh.crypto import sign as _sign + sig = _sign(alice_id.sign_priv, env_orig.canonical_bytes()) + + env_tampered = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=100, + body={"document_root": "bb" * 32, "source_uri": "https://x"}, # changed + ) + a_conn = connect(alice_db) + try: + assert verify_envelope_sig(a_conn, env_tampered, sig) is False + finally: + a_conn.close() + + +def test_verify_envelope_sig_rejects_unknown_sender(tmp_path): + alice_db, _bob_db, _alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + sign_priv, _sign_pub = generate_signing_keypair() + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="mallory", # not in roster + epoch_id=1, + ts=0, + body={"document_root": "00" * 32, "source_uri": "https://x"}, + ) + from aborist.mesh.crypto import sign as _sign + sig = _sign(sign_priv, env.canonical_bytes()) + a_conn = connect(alice_db) + try: + assert verify_envelope_sig(a_conn, env, sig) is False + finally: + a_conn.close() + + +def test_verify_envelope_sig_rejects_wrong_key_for_known_member(tmp_path): + """Mallory steals alice's member_id but signs with her own key.""" + alice_db, _bob_db, _alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + rogue_priv, _rogue_pub = generate_signing_keypair() + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", # claims to be alice + epoch_id=1, + ts=0, + body={"document_root": "00" * 32, "source_uri": "https://x"}, + ) + from aborist.mesh.crypto import sign as _sign + sig = _sign(rogue_priv, env.canonical_bytes()) + a_conn = connect(alice_db) + try: + assert verify_envelope_sig(a_conn, env, sig) is False + finally: + a_conn.close() + + +# --------------------------------------------------------------------------- +# Server.handle_announce — direct (no HTTP) +# --------------------------------------------------------------------------- + + +def test_handle_announce_writes_audit_event(tmp_path): + """Bob sends ANNOUNCE_ROOT to Alice's server; Alice's chain gains an event.""" + from aborist.mesh.wire import MeshWireServer + + alice_db, _bob_db, _alice_id, bob_id = _bootstrap_two_peer_db(tmp_path) + + a_conn = connect(alice_db) + try: + before = a_conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] + finally: + a_conn.close() + + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="bob", + epoch_id=1, + ts=int(time.time()), + body={ + "document_root": "ab" * 32, + "source_uri": "https://en.wikipedia.org/wiki/Cloud_Strife", + "chunking_version": "tok-512-v1", + "canonicalization_version": "norm-v1", + "schema_version": "v9.8.0", + }, + ) + signed = env.sign_with(bob_id.sign_priv) + + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + assert status == 200, body + assert body["wire_type"] == TYPE_ANNOUNCE_ROOT + assert "audit_event_hash" in body + finally: + srv.stop() + + a_conn = connect(alice_db) + try: + after = a_conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] + last = a_conn.execute( + "SELECT event_type, body FROM audit_events ORDER BY seq DESC LIMIT 1" + ).fetchone() + finally: + a_conn.close() + assert after == before + 1 + assert last["event_type"] == "mesh_received" + import json as _json + body_json = _json.loads(last["body"]) + assert body_json["wire_type"] == TYPE_ANNOUNCE_ROOT + assert body_json["sender_id"] == "bob" + + +def test_handle_announce_rejects_bad_sig(tmp_path): + """Sig from wrong key → 401, no audit event.""" + from aborist.mesh.wire import MeshWireServer + + alice_db, _bob_db, _alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + rogue_priv, _ = generate_signing_keypair() + + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="bob", + epoch_id=1, + ts=0, + body={"document_root": "00" * 32, "source_uri": "x"}, + ) + from aborist.mesh.crypto import sign as _sign + sig = _sign(rogue_priv, env.canonical_bytes()) + import base64 as _b64 + signed = {"envelope": env.__dict__, "sig_b64": _b64.b64encode(sig).decode()} + + a_conn = connect(alice_db) + try: + before = a_conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] + finally: + a_conn.close() + + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + assert status == 401 + assert "signature" in body["error"].lower() + finally: + srv.stop() + + a_conn = connect(alice_db) + try: + after = a_conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] + finally: + a_conn.close() + assert after == before # rejected: no event written + + +def test_handle_announce_rejects_request_body_type(tmp_path): + """The /announce endpoint refuses REQUEST_BODY / DELIVER_BODY.""" + from aborist.mesh.wire import MeshWireServer + + alice_db, _bob_db, _alice_id, bob_id = _bootstrap_two_peer_db(tmp_path) + env = WireEnvelope( + type=TYPE_REQUEST_BODY, + sender_id="bob", + epoch_id=1, + ts=0, + body={"root": "00" * 32}, + ) + signed = env.sign_with(bob_id.sign_priv) + + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + assert status == 400 + assert "rejects" in body["error"] + finally: + srv.stop() + + +def test_handle_request_404_on_unknown_root(tmp_path): + """Asking for a doc the responder doesn't have returns 404 cleanly.""" + from aborist.mesh.wire import MeshWireServer + + alice_db, _bob_db, _alice_id, bob_id = _bootstrap_two_peer_db(tmp_path) + env = WireEnvelope( + type=TYPE_REQUEST_BODY, + sender_id="bob", + epoch_id=1, + ts=0, + body={"root": "ff" * 32}, + ) + signed = env.sign_with(bob_id.sign_priv) + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) # wrong endpoint type → 400 + assert status == 400 + status, body = srv.handle_request(signed) + assert status == 404 + finally: + srv.stop() + + +def test_announce_falsification_round_trips(tmp_path): + """Falsifications propagate as ANNOUNCE_FALSIFICATION.""" + from aborist.mesh.wire import MeshWireServer + + alice_db, _bob_db, _alice_id, bob_id = _bootstrap_two_peer_db(tmp_path) + env = WireEnvelope( + type=TYPE_ANNOUNCE_FALSIFICATION, + sender_id="bob", + epoch_id=1, + ts=0, + body={"cache_key": "ab" * 32, "reason": "answer was wrong: cited X but X is fictional"}, + ) + signed = env.sign_with(bob_id.sign_priv) + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + assert status == 200 + assert body["wire_type"] == TYPE_ANNOUNCE_FALSIFICATION + finally: + srv.stop() diff --git a/tests/test_mesh_wire_e2e.py b/tests/test_mesh_wire_e2e.py new file mode 100644 index 0000000..89c7a4e --- /dev/null +++ b/tests/test_mesh_wire_e2e.py @@ -0,0 +1,186 @@ +"""Mesh wire end-to-end: two real HTTP peers exchange gossip. + +Each peer binds an ephemeral port via host=127.0.0.1, port=0 and runs +ThreadingHTTPServer in a daemon thread. The other peer's MeshWireClient +talks to it over real httpx + TCP. No mocking — this exercises the +full request/response path including JSON serialization, HTTP framing, +and threaded handler dispatch. + +Tests run under timeout via daemon threads; if a test hangs the worst +case is the test process exits and pytest times out the test, leaving +no zombie threads (daemon=True). +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from aborist.document import Document +from aborist.ingest import ingest_source +from aborist.mesh import init_identity +from aborist.mesh.members import add_member +from aborist.mesh.wire import ( + MeshWireClient, + MeshWireServer, + TYPE_ANNOUNCE_ROOT, +) +from aborist.store import connect + + +@pytest.fixture +def two_peers(tmp_path): + """Spin up alice + bob with mutual rosters and HTTP servers running.""" + alice_db = tmp_path / "alice.db" + bob_db = tmp_path / "bob.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() + + a_conn = connect(alice_db) + try: + add_member(a_conn, member_id="bob", sign_pub=bob.sign_pub, dh_pub=bob.dh_pub) + finally: + a_conn.close() + + b_conn = connect(bob_db) + try: + add_member(b_conn, member_id="alice", sign_pub=alice.sign_pub, dh_pub=alice.dh_pub) + finally: + b_conn.close() + + alice_srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + bob_srv = MeshWireServer(bob_db, host="127.0.0.1", port=0) + alice_srv.start_in_thread() + bob_srv.start_in_thread() + try: + yield { + "alice": {"db": alice_db, "srv": alice_srv, "id": alice}, + "bob": {"db": bob_db, "srv": bob_srv, "id": bob}, + } + finally: + alice_srv.stop() + bob_srv.stop() + + +def test_info_endpoint_returns_member_id(two_peers): + """GET /mesh/info returns the responder's identity (no signature required).""" + with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c: + info = c.info() + assert info["member_id"] == "alice" + assert info["group_name"] == "t" + assert info["current_epoch"] == 1 + + +def test_announce_root_round_trip(two_peers): + """Bob announces a doc_root to Alice; Alice's audit chain gains a + `mesh_received` event with the announce body.""" + with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c: + resp = c.announce_root( + document_root="ab" * 32, + source_uri="https://en.wikipedia.org/wiki/Cloud_Strife", + chunking_version="tok-512-v1", + canonicalization_version="norm-v1", + schema_version="v9.8.0", + ) + assert resp["ok"] is True + assert resp["wire_type"] == TYPE_ANNOUNCE_ROOT + assert isinstance(resp["audit_event_hash"], str) + assert len(resp["audit_event_hash"]) == 64 + + a_conn = connect(two_peers["alice"]["db"]) + try: + last = a_conn.execute( + "SELECT event_type, body, subject_root FROM audit_events " + "ORDER BY seq DESC LIMIT 1" + ).fetchone() + finally: + a_conn.close() + assert last["event_type"] == "mesh_received" + assert last["subject_root"] == "ab" * 32 + import json + body = json.loads(last["body"]) + assert body["wire_type"] == TYPE_ANNOUNCE_ROOT + assert body["sender_id"] == "bob" + assert body["envelope_body"]["source_uri"] == "https://en.wikipedia.org/wiki/Cloud_Strife" + + +class _OneDocSource: + """One-shot source for tests — yields a single Document.""" + + source_type = "wire_e2e" + + def __init__(self, uri: str, content: str, title: str): + self._doc = Document( + uri=uri, content=content, source_type=self.source_type, title=title + ) + + def iter_documents(self): + yield self._doc + + +def _ingest_one(db_path, *, uri: str, content: str, title: str) -> str: + """Ingest one document, return its document_root.""" + conn = connect(db_path) + try: + ingest_source(conn, _OneDocSource(uri, content, title)) + row = conn.execute( + "SELECT document_root FROM documents WHERE document_uri=?", (uri,) + ).fetchone() + finally: + conn.close() + assert row is not None, f"document not found post-ingest for {uri!r}" + return row["document_root"] + + +def test_request_body_pulls_doc_with_verified_merkle(two_peers): + """Alice has a doc; Bob requests it. Client verifies the delivered + Merkle root against the requested root before returning.""" + document_root = _ingest_one( + two_peers["alice"]["db"], + uri="https://en.wikipedia.org/wiki/test", + content="Hello world. The quick brown fox jumps over the lazy dog.", + title="Test", + ) + with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c: + delivered = c.request_body(root=document_root) + + assert delivered["root"] == document_root + assert delivered["document_uri"] == "https://en.wikipedia.org/wiki/test" + assert "Hello world" in delivered["text"] + assert isinstance(delivered["leaves_hex"], list) + assert len(delivered["leaves_hex"]) >= 1 + + +def test_request_body_404_on_unknown_root(two_peers): + """Asking for a root the responder doesn't have raises HTTPStatusError.""" + import httpx + with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c: + with pytest.raises(httpx.HTTPStatusError): + c.request_body(root="ff" * 32) + + +def test_request_body_rejects_tampered_response(two_peers, monkeypatch): + """If a peer returns a DELIVER_BODY whose leaves don't Merkle-derive + to the claimed root, the client raises before returning.""" + document_root = _ingest_one( + two_peers["alice"]["db"], + uri="https://en.wikipedia.org/wiki/x", + content="real content of the document", + title="X", + ) + with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c: + from aborist.mesh import wire as _wire + monkeypatch.setattr(_wire, "_merkle_root_matches", lambda *a, **k: False) + with pytest.raises(ValueError, match="Merkle-derive"): + c.request_body(root=document_root)