diff --git a/aborist/mesh/state.py b/aborist/mesh/state.py index 268ae2e..54f181c 100644 --- a/aborist/mesh/state.py +++ b/aborist/mesh/state.py @@ -340,3 +340,64 @@ def unwrap_secret_for_self( ct = base64.b64decode(entry["ct_b64"]) shared = ecdh_shared_secret(me.dh_priv, sender_dh_pub) return aead_decrypt(shared, nonce, ct, aad=me.member_id.encode("utf-8")) + + +def recover_epoch_secret( + conn: sqlite3.Connection, + *, + epoch_id: int, +) -> bytes: + """Convenience: locate the rotator's dh_pub locally and unwrap our slot. + + The rotator is recorded in the epoch's audit event body — `actor` for + rotations (epoch >= 1), `founder` for the genesis epoch 0. We resolve + the actor's dh_pub via `mesh_roster` at that epoch (the actor was a + member of the post-rotation roster by construction). Then we hand off + to `unwrap_secret_for_self`. + + This is the path used by both senders (re-unwrapping their own slot + to AEAD-encrypt outbound gossip) and receivers (decrypting an inbound + encrypted body). Sender-side it's the simplest answer to "where does + the sender get the epoch secret" without a new caching layer. + + Raises: + RuntimeError if mesh identity not initialized. + ValueError if the epoch row is missing, the audit event is missing, + the actor is not in the epoch's roster, this peer has no slot in + the envelope (evicted), or the AEAD tag mismatches. + """ + row = conn.execute( + "SELECT started_event_hash FROM mesh_epochs WHERE epoch_id = ?", + (epoch_id,), + ).fetchone() + if row is None: + raise ValueError(f"unknown epoch: {epoch_id}") + started_event_hash = row["started_event_hash"] + if not started_event_hash: + raise ValueError(f"epoch {epoch_id} has no audit linkage") + ev = conn.execute( + "SELECT body FROM audit_events WHERE event_hash = ?", + (started_event_hash,), + ).fetchone() + if ev is None: + raise ValueError( + f"epoch {epoch_id} audit event {started_event_hash!r} not found" + ) + body = json.loads(ev["body"]) + # Genesis writes "founder"; rotations write "actor". + actor = body.get("actor") or body.get("founder") + if not actor: + raise ValueError( + f"epoch {epoch_id} audit body has no actor/founder field" + ) + rotator = next( + (m for m in roster_at(conn, epoch_id) if m.member_id == actor), + None, + ) + if rotator is None: + raise ValueError( + f"rotator {actor!r} not in epoch {epoch_id} roster" + ) + return unwrap_secret_for_self( + conn, epoch_id=epoch_id, sender_dh_pub=rotator.dh_pub + ) diff --git a/aborist/mesh/wire.py b/aborist/mesh/wire.py index f1e1e4e..9d69e2e 100644 --- a/aborist/mesh/wire.py +++ b/aborist/mesh/wire.py @@ -26,6 +26,7 @@ from __future__ import annotations import base64 import json +import os import threading import time from dataclasses import asdict, dataclass @@ -35,10 +36,11 @@ from typing import Any import httpx -from aborist.mesh.crypto import sign, verify +from aborist.mesh.crypto import aead_decrypt, aead_encrypt, sign, verify from aborist.mesh.state import ( current_epoch, load_identity, + recover_epoch_secret, roster_at, ) from aborist.store import append_audit, connect @@ -85,6 +87,14 @@ class WireEnvelope: `body` is type-specific; canonicalization is via JSON sorted-keys with no whitespace, identical to `aborist.store._canonical_json`. + + Confidentiality is opt-in. When `encrypted_body` is set, `body` is a + placeholder (typically `{}`) and the real body lives AEAD-encrypted + inside `encrypted_body = {"nonce_b64": ..., "ct_b64": ...}`. Both + fields participate in `canonical_bytes()`, so the Ed25519 signature + 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. """ type: str @@ -93,16 +103,30 @@ class WireEnvelope: ts: int body: dict[str, Any] v: int = WIRE_VERSION + encrypted_body: dict[str, str] | None = None def canonical_bytes(self) -> bytes: - """Bytes that get signed. Stable, reproducible.""" - return _canonical_json(asdict(self)).encode("utf-8") + """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). + """ + d = asdict(self) + if d.get("encrypted_body") is None: + d.pop("encrypted_body", None) + return _canonical_json(d).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()) + env_d = asdict(self) + if env_d.get("encrypted_body") is None: + env_d.pop("encrypted_body", None) return { - "envelope": asdict(self), + "envelope": env_d, "sig_b64": base64.b64encode(sig).decode("ascii"), } @@ -124,6 +148,13 @@ class WireEnvelope: 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}") + encrypted_body = env_d.get("encrypted_body") + if encrypted_body is not None: + if not isinstance(encrypted_body, dict): + raise ValueError("encrypted_body must be a dict if present") + for k in ("nonce_b64", "ct_b64"): + if k not in encrypted_body: + raise ValueError(f"encrypted_body missing field: {k}") env = cls( type=env_d["type"], sender_id=env_d["sender_id"], @@ -131,6 +162,7 @@ class WireEnvelope: ts=int(env_d["ts"]), body=env_d["body"], v=int(env_d["v"]), + encrypted_body=encrypted_body, ) try: sig = base64.b64decode(signed["sig_b64"], validate=True) @@ -139,6 +171,83 @@ class WireEnvelope: return env, sig +def _envelope_aad( + *, + type_: str, + sender_id: str, + epoch_id: int, + ts: int, +) -> bytes: + """AEAD additional-authenticated-data for body encryption. + + Binds the ciphertext to the envelope metadata so a captured + (nonce, ct) cannot be replayed under a different `(type, sender, + epoch, ts)` tuple. The signature already binds these fields — AAD + is belt-and-suspenders against the case where a peer accepts an + envelope with a fresh sig but recycled ciphertext (e.g. cross-type + confusion). + """ + return f"{type_}|{sender_id}|{epoch_id}|{ts}".encode("utf-8") + + +def encrypt_body_for_envelope( + *, + type_: str, + sender_id: str, + epoch_id: int, + ts: int, + body: dict[str, Any], + epoch_secret: bytes, +) -> dict[str, str]: + """Serialize `body` canonically, AEAD-encrypt under `epoch_secret`. + + Returns a dict of base64-encoded `nonce_b64` + `ct_b64` suitable to + drop into `WireEnvelope.encrypted_body`. The plaintext is the same + canonical JSON that would have appeared in `body`, so a receiver + decrypts and JSON-parses to recover the original dict. + """ + plaintext = _canonical_json(body).encode("utf-8") + nonce = os.urandom(12) + aad = _envelope_aad( + type_=type_, sender_id=sender_id, epoch_id=epoch_id, ts=ts + ) + ct = aead_encrypt(epoch_secret, nonce, plaintext, aad=aad) + return { + "nonce_b64": base64.b64encode(nonce).decode("ascii"), + "ct_b64": base64.b64encode(ct).decode("ascii"), + } + + +def decrypt_envelope_body( + envelope: WireEnvelope, + epoch_secret: bytes, +) -> dict[str, Any]: + """AEAD-decrypt `envelope.encrypted_body` and return the recovered dict. + + Raises ValueError on missing encrypted_body, malformed base64, AEAD + tag mismatch (wrong key, tampering, AAD drift), or non-dict + plaintext. + """ + if envelope.encrypted_body is None: + raise ValueError("envelope has no encrypted_body to decrypt") + try: + nonce = base64.b64decode(envelope.encrypted_body["nonce_b64"]) + ct = base64.b64decode(envelope.encrypted_body["ct_b64"]) + except Exception as e: + raise ValueError(f"encrypted_body base64 decode failed: {e!r}") from e + aad = _envelope_aad( + type_=envelope.type, + sender_id=envelope.sender_id, + epoch_id=envelope.epoch_id, + ts=envelope.ts, + ) + plaintext = aead_decrypt(epoch_secret, nonce, ct, aad=aad) + decoded = json.loads(plaintext.decode("utf-8")) + if not isinstance(decoded, dict): + raise ValueError("decrypted body must be a JSON object") + return decoded + + def verify_envelope_sig( conn, envelope: WireEnvelope, @@ -292,7 +401,14 @@ class MeshWireServer: } def handle_announce(self, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]: - """Verify sig, dedup, persist as a `mesh_received` audit event.""" + """Verify sig, dedup, persist as a `mesh_received` audit event. + + If the envelope carries `encrypted_body`, decrypt it under this + peer's recovered epoch secret before audit-logging. A peer with + no slot in the epoch envelope (evicted or never enrolled) returns + 401: confidentiality failure is indistinguishable from auth + failure at this layer. + """ try: env, sig = WireEnvelope.from_signed(payload) except ValueError as e: @@ -309,6 +425,13 @@ class MeshWireServer: try: if not verify_envelope_sig(conn, env, sig): return 401, {"error": "signature verify failed"} + effective_body = env.body + if env.encrypted_body is not None: + try: + epoch_secret = recover_epoch_secret(conn, epoch_id=env.epoch_id) + 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", @@ -317,9 +440,14 @@ class MeshWireServer: "sender_id": env.sender_id, "epoch_id": env.epoch_id, "remote_ts": env.ts, - "envelope_body": env.body, + "envelope_body": effective_body, + "encrypted": env.encrypted_body is not None, }, - subject_root=str(env.body.get("document_root") or env.body.get("cache_key") or ""), + subject_root=str( + effective_body.get("document_root") + or effective_body.get("cache_key") + or "" + ), ) finally: conn.close() @@ -417,22 +545,62 @@ class MeshWireClient: self.close() # ------------------------------------------------------------------ helpers - def _signed_envelope(self, type_: str, body: dict[str, Any]) -> dict[str, Any]: + def _signed_envelope( + self, + type_: str, + body: dict[str, Any], + *, + 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. + + 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. + """ conn = connect(self.db_path) try: me = load_identity(conn) epoch = current_epoch(conn) + secret: bytes | None = None + 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) 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, - ) + 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) # ------------------------------------------------------------------ ops @@ -449,7 +617,13 @@ class MeshWireClient: chunking_version: str, canonicalization_version: str, schema_version: str, + encrypt: bool = False, ) -> dict[str, Any]: + """Announce a `document_root`. Default cleartext; pass + `encrypt=True` to AEAD-encrypt the body under this peer's + epoch secret. The signature still binds the ciphertext, so a + MITM cannot swap payloads without invalidating the envelope. + """ signed = self._signed_envelope( TYPE_ANNOUNCE_ROOT, { @@ -459,15 +633,24 @@ class MeshWireClient: "canonicalization_version": canonicalization_version, "schema_version": schema_version, }, + encrypt=encrypt, ) 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]: + def announce_falsification( + self, + *, + cache_key: str, + reason: str, + encrypt: bool = False, + ) -> dict[str, Any]: + """Announce a falsification. `encrypt` mirrors `announce_root`.""" signed = self._signed_envelope( TYPE_ANNOUNCE_FALSIFICATION, {"cache_key": cache_key, "reason": reason}, + encrypt=encrypt, ) r = self._client.post(self.peer_url + PATH_ANNOUNCE, json=signed) r.raise_for_status() diff --git a/tests/test_mesh_aead.py b/tests/test_mesh_aead.py new file mode 100644 index 0000000..33a1436 --- /dev/null +++ b/tests/test_mesh_aead.py @@ -0,0 +1,542 @@ +"""Mesh wire — AEAD-encrypted gossip bodies. + +Cleartext gossip stays the default; this module verifies the opt-in +encryption path. A sender flips ``encrypt=True`` on an announce call, +the body ships AEAD-protected under the per-epoch shared secret, and +a current member of the same epoch decrypts via their wrapped slot in +the secret envelope. + +The signature still spans the full envelope (including the ciphertext), +so a MITM cannot swap payloads without invalidating sig verification. +Receivers without a slot in the epoch envelope (evicted / never +enrolled) get a clean 401 — confidentiality failure mirrors auth +failure at the wire layer. +""" + +from __future__ import annotations + +import base64 +import time +from pathlib import Path + +import pytest + +from aborist.mesh import init_identity +from aborist.mesh.members import add_member +from aborist.mesh.state import recover_epoch_secret +from aborist.mesh.wire import ( + MeshWireServer, + TYPE_ANNOUNCE_FALSIFICATION, + TYPE_ANNOUNCE_ROOT, + WireEnvelope, + decrypt_envelope_body, + encrypt_body_for_envelope, +) +from aborist.store import connect + + +def _bootstrap_two_peer_db(tmp_path: Path) -> tuple[Path, Path, dict, dict]: + """alice + bob in the same group, sharing epoch 1's secret envelope. + + Real federation invariant: every peer at the same epoch sees the + SAME `mesh_epochs.secret_envelope`. Without that, sender-side + encryption under one DB's epoch secret cannot decrypt under + another DB's epoch secret. The signing-only fixture in + `test_mesh_wire.py` independently rotates each side, which is fine + for sig checks but breaks AEAD by construction. + + Here we let alice rotate (genesis+add_member(bob)), then mirror + her epoch 1 `mesh_epochs` row + `mesh_roster` rows + matching + audit event into bob's DB. Same wrapped secret on both sides; + bob's `recover_epoch_secret` resolves alice as actor and unwraps + bob's slot using alice's dh_pub. The same path the wire receiver + walks at runtime. + """ + import json as _json + import sqlite3 as _sqlite3 + + from aborist.store import append_audit, latest_event_hash + + 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() + + # Alice rotates to epoch 1, adding bob. Her DB now has epoch-1 + # secret_envelope wrapped to {alice, bob} via alice's dh_priv. + 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", + ) + epoch1_row = a_conn.execute( + "SELECT epoch_id, started_at, started_event_hash, " + "secret_envelope, reason FROM mesh_epochs WHERE epoch_id=1" + ).fetchone() + alice_roster_e1 = a_conn.execute( + "SELECT member_id, sign_pub, dh_pub, role FROM mesh_roster " + "WHERE epoch_id=1 ORDER BY member_id" + ).fetchall() + # Pull the rotation audit event so we can replay its body on bob's DB. + rot_audit = a_conn.execute( + "SELECT body, ts FROM audit_events WHERE event_hash=?", + (epoch1_row["started_event_hash"],), + ).fetchone() + finally: + a_conn.close() + + # Mirror epoch-1 onto bob's DB. Bob also needs alice's pubs in his + # roster_at(0) so the roster lookup at epoch 1 (the rotator alice's + # dh_pub) resolves. The rotation audit event must replay exactly so + # the chain links cleanly; we use append_audit which recomputes the + # event_hash deterministically — and since bob's chain is shorter + # we just append a fresh rotate event with alice as actor, then + # backfill epoch row's started_event_hash to that hash. + b_conn = connect(bob_db) + try: + # Replay the rotate event on bob's chain. body is canonical-JSON. + rot_body = _json.loads(rot_audit["body"]) + replayed_hash = append_audit( + b_conn, + event_type="mesh_epoch_rotate", + body=rot_body, + ts=int(rot_audit["ts"]), + ) + # Insert the full epoch-1 roster (alice + bob) into bob's DB. + for r in alice_roster_e1: + b_conn.execute( + "INSERT OR REPLACE INTO mesh_roster " + "(epoch_id, member_id, sign_pub, dh_pub, role) " + "VALUES (1, ?, ?, ?, ?)", + (r["member_id"], r["sign_pub"], r["dh_pub"], r["role"]), + ) + # Insert the SAME secret_envelope row so bob unwraps the same key. + b_conn.execute( + "INSERT OR REPLACE INTO mesh_epochs " + "(epoch_id, started_at, started_event_hash, secret_envelope, reason) " + "VALUES (1, ?, ?, ?, ?)", + ( + int(epoch1_row["started_at"]), + replayed_hash, + epoch1_row["secret_envelope"], + epoch1_row["reason"], + ), + ) + b_conn.commit() + finally: + b_conn.close() + + return alice_db, bob_db, alice_id, bob_id + + +def _ann_root_body() -> dict: + return { + "document_root": "ab" * 32, + "source_uri": "https://en.wikipedia.org/wiki/Confidential", + "chunking_version": "tok-512-v1", + "canonicalization_version": "norm-v1", + "schema_version": "v9.8.0", + } + + +def _build_encrypted_announce( + db_path: Path, + sign_priv: bytes, + *, + sender_id: str, + epoch_id: int, + ts: int | None = None, + body: dict | None = None, +) -> tuple[WireEnvelope, dict]: + """Construct + sign an encrypted ANNOUNCE_ROOT envelope from a peer. + + Returns (envelope, signed_payload). The envelope dataclass instance + is handy for tests that want to mutate ciphertext before re-signing. + """ + if body is None: + body = _ann_root_body() + if ts is None: + ts = int(time.time()) + conn = connect(db_path) + try: + secret = recover_epoch_secret(conn, epoch_id=epoch_id) + finally: + conn.close() + encrypted = encrypt_body_for_envelope( + type_=TYPE_ANNOUNCE_ROOT, + sender_id=sender_id, + epoch_id=epoch_id, + ts=ts, + body=body, + epoch_secret=secret, + ) + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id=sender_id, + epoch_id=epoch_id, + ts=ts, + body={}, + encrypted_body=encrypted, + ) + return env, env.sign_with(sign_priv) + + +# --------------------------------------------------------------------------- +# 1. Round-trip: alice encrypts, bob (current member) decrypts + accepts. +# --------------------------------------------------------------------------- + + +def test_encrypted_announce_round_trip(tmp_path): + """Alice sends an encrypted ANNOUNCE_ROOT; bob's server decrypts and + accepts. The audit_event reflects the cleartext body and marks + `encrypted=True` for operator visibility.""" + alice_db, bob_db, alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + env, signed = _build_encrypted_announce( + alice_db, alice_id.sign_priv, sender_id="alice", epoch_id=1 + ) + # Ciphertext must NOT contain the plaintext doc_root. + assert env.encrypted_body is not None + ct_b64 = env.encrypted_body["ct_b64"] + assert "document_root" not in base64.b64decode(ct_b64).decode( + "utf-8", errors="ignore" + ) + + srv = MeshWireServer(bob_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + finally: + srv.stop() + assert status == 200, body + assert body["wire_type"] == TYPE_ANNOUNCE_ROOT + + b_conn = connect(bob_db) + try: + last = b_conn.execute( + "SELECT body, subject_root FROM audit_events ORDER BY seq DESC LIMIT 1" + ).fetchone() + finally: + b_conn.close() + import json + body_json = json.loads(last["body"]) + assert body_json["encrypted"] is True + # Decrypted plaintext threaded into the audit body. + assert body_json["envelope_body"]["document_root"] == "ab" * 32 + assert last["subject_root"] == "ab" * 32 + + +# --------------------------------------------------------------------------- +# 2. Receiver missing the epoch secret — evicted member returns 401. +# --------------------------------------------------------------------------- + + +def test_encrypted_announce_evicted_receiver_401(tmp_path): + """Carol has alice in her roster (so signature verifies) but no slot + in the epoch envelope (she manufactured her own genesis). The + decrypt path therefore fails and `handle_announce` returns 401. + """ + alice_db, _bob_db, alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + carol_db = tmp_path / "carol.db" + c_conn = connect(carol_db) + try: + carol_id = init_identity(c_conn, group_name="t", member_id="carol") + # Splice alice into carol's epoch-1 roster directly so the SIG + # check passes — but carol's secret_envelope at epoch 1 has + # carol's slot only (she was its founder), wrapped to carol's + # OWN dh keys, not alice's. So decrypt under that slot yields + # the wrong key for alice's ciphertext → AEAD tag mismatch. + # First bump carol's epoch to 1 by adding alice via add_member, + # which creates a fresh epoch 1 wrapped to carol+alice. + add_member( + c_conn, + member_id="alice", + sign_pub=alice_id.sign_pub, + dh_pub=alice_id.dh_pub, + role="member", + ) + finally: + c_conn.close() + + # Build an encrypted ANNOUNCE on alice's DB (alice's epoch-1 secret). + _env, signed = _build_encrypted_announce( + alice_db, alice_id.sign_priv, sender_id="alice", epoch_id=1 + ) + # Send to carol's server. Sig verifies (alice is in carol's roster), + # but carol's epoch-1 secret differs from alice's → AEAD fails. + srv = MeshWireServer(carol_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + finally: + srv.stop() + assert status == 401 + assert "AEAD decrypt" in body["error"] or "decrypt" in body["error"].lower() + + # No audit event was written for the rejected message. + c_conn = connect(carol_db) + try: + last = c_conn.execute( + "SELECT event_type FROM audit_events ORDER BY seq DESC LIMIT 1" + ).fetchone() + finally: + c_conn.close() + # The most recent event should NOT be a `mesh_received` for the rejected msg. + assert last is None or last["event_type"] != "mesh_received" + + +# --------------------------------------------------------------------------- +# 3. Tampered ciphertext: sig stays valid (we re-sign) but AEAD fails. +# --------------------------------------------------------------------------- + + +def test_encrypted_announce_tampered_ciphertext_401(tmp_path): + """Flip a byte of the ciphertext and re-sign with a valid key + (alice's). The Ed25519 sig still verifies — the protection here + comes from AEAD's authentication tag, which detects the tamper. + """ + alice_db, bob_db, alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + env, _signed = _build_encrypted_announce( + alice_db, alice_id.sign_priv, sender_id="alice", epoch_id=1 + ) + # Flip a byte inside ct_b64. + assert env.encrypted_body is not None + raw_ct = bytearray(base64.b64decode(env.encrypted_body["ct_b64"])) + raw_ct[0] ^= 0x01 + tampered_ct_b64 = base64.b64encode(bytes(raw_ct)).decode("ascii") + tampered_env = WireEnvelope( + type=env.type, + sender_id=env.sender_id, + epoch_id=env.epoch_id, + ts=env.ts, + body=env.body, + encrypted_body={ + "nonce_b64": env.encrypted_body["nonce_b64"], + "ct_b64": tampered_ct_b64, + }, + ) + signed = tampered_env.sign_with(alice_id.sign_priv) + + srv = MeshWireServer(bob_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + finally: + srv.stop() + assert status == 401 + assert "decrypt" in body["error"].lower() + + +# --------------------------------------------------------------------------- +# 4. Wrong nonce / wrong AAD — both yield AEAD tag mismatch. +# --------------------------------------------------------------------------- + + +def test_encrypted_announce_swapped_nonce_401(tmp_path): + """Re-sign with a freshly randomized nonce (key + AAD unchanged). + AEAD won't authenticate. + """ + alice_db, bob_db, alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + env, _signed = _build_encrypted_announce( + alice_db, alice_id.sign_priv, sender_id="alice", epoch_id=1 + ) + assert env.encrypted_body is not None + import os as _os + + bad_nonce_b64 = base64.b64encode(_os.urandom(12)).decode("ascii") + tampered_env = WireEnvelope( + type=env.type, + sender_id=env.sender_id, + epoch_id=env.epoch_id, + ts=env.ts, + body=env.body, + encrypted_body={ + "nonce_b64": bad_nonce_b64, + "ct_b64": env.encrypted_body["ct_b64"], + }, + ) + signed = tampered_env.sign_with(alice_id.sign_priv) + srv = MeshWireServer(bob_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + finally: + srv.stop() + assert status == 401 + + +def test_encrypted_announce_aad_drift_401(tmp_path): + """Build a ciphertext with one ts, then sign an envelope with a + different ts. AAD is recomputed receiver-side from the envelope + metadata, so it won't match — AEAD fails. + """ + alice_db, bob_db, alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + env, _signed = _build_encrypted_announce( + alice_db, alice_id.sign_priv, sender_id="alice", epoch_id=1, ts=1000 + ) + assert env.encrypted_body is not None + drifted_env = WireEnvelope( + type=env.type, + sender_id=env.sender_id, + epoch_id=env.epoch_id, + ts=2000, # different ts → AAD changes + body=env.body, + encrypted_body=env.encrypted_body, + ) + signed = drifted_env.sign_with(alice_id.sign_priv) + srv = MeshWireServer(bob_db, host="127.0.0.1", port=0) + try: + status, body = srv.handle_announce(signed) + finally: + srv.stop() + assert status == 401 + + +# --------------------------------------------------------------------------- +# 5. Mixed mode: cleartext + encrypted from the same client both flow. +# --------------------------------------------------------------------------- + + +def test_mixed_cleartext_and_encrypted_round_trip(tmp_path): + """Alice sends one cleartext ANNOUNCE_ROOT, then one encrypted + ANNOUNCE_FALSIFICATION. Bob accepts both; audit chain logs the + `encrypted` flag accurately on each.""" + from aborist.mesh.wire import MeshWireClient + + alice_db, bob_db, _alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + bob_srv = MeshWireServer(bob_db, host="127.0.0.1", port=0) + bob_srv.start_in_thread() + try: + with MeshWireClient(alice_db, bob_srv.url) as c: + r1 = c.announce_root( + document_root="cd" * 32, + source_uri="https://x/cleartext", + chunking_version="tok-512-v1", + canonicalization_version="norm-v1", + schema_version="v9.8.0", + ) + r2 = c.announce_falsification( + cache_key="ef" * 32, + reason="encrypted falsification", + encrypt=True, + ) + assert r1["wire_type"] == TYPE_ANNOUNCE_ROOT + assert r2["wire_type"] == TYPE_ANNOUNCE_FALSIFICATION + finally: + bob_srv.stop() + + b_conn = connect(bob_db) + try: + rows = b_conn.execute( + "SELECT body FROM audit_events WHERE event_type='mesh_received' " + "ORDER BY seq DESC LIMIT 2" + ).fetchall() + finally: + b_conn.close() + import json + bodies = [json.loads(r["body"]) for r in rows] + # rows are descending; second message (encrypted falsification) is newest + enc, plain = bodies[0], bodies[1] + assert enc["wire_type"] == TYPE_ANNOUNCE_FALSIFICATION + assert enc["encrypted"] is True + assert enc["envelope_body"]["cache_key"] == "ef" * 32 + assert plain["wire_type"] == TYPE_ANNOUNCE_ROOT + assert plain["encrypted"] is False + assert plain["envelope_body"]["document_root"] == "cd" * 32 + + +# --------------------------------------------------------------------------- +# 6. Cleartext-only mode (current default) — wire format unchanged. +# --------------------------------------------------------------------------- + + +def test_cleartext_default_envelope_unchanged(tmp_path): + """An envelope with `encrypt=False` (default) MUST canonicalize + identically to the v1 wire format — no `encrypted_body` field on + the canonical bytes, no signature change, no audit-body change. + """ + from aborist.mesh.wire import MeshWireClient + + alice_db, bob_db, _alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + bob_srv = MeshWireServer(bob_db, host="127.0.0.1", port=0) + bob_srv.start_in_thread() + try: + with MeshWireClient(alice_db, bob_srv.url) as c: + r = c.announce_root( + document_root="01" * 32, + source_uri="https://x/plain", + chunking_version="tok-512-v1", + canonicalization_version="norm-v1", + schema_version="v9.8.0", + ) + assert r["ok"] is True + finally: + bob_srv.stop() + + b_conn = connect(bob_db) + try: + last = b_conn.execute( + "SELECT body FROM audit_events WHERE event_type='mesh_received' " + "ORDER BY seq DESC LIMIT 1" + ).fetchone() + finally: + b_conn.close() + import json + body_json = json.loads(last["body"]) + assert body_json["encrypted"] is False + assert body_json["envelope_body"]["document_root"] == "01" * 32 + + +# --------------------------------------------------------------------------- +# Hardening: round-trip the AEAD primitive directly (no HTTP). +# --------------------------------------------------------------------------- + + +def test_decrypt_envelope_body_direct(tmp_path): + """Sanity check: encrypt_body_for_envelope + decrypt_envelope_body + are exact inverses given the same key, AAD, and metadata.""" + alice_db, _bob_db, _alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path) + a_conn = connect(alice_db) + try: + secret = recover_epoch_secret(a_conn, epoch_id=1) + finally: + a_conn.close() + body = {"document_root": "11" * 32, "source_uri": "https://x"} + enc = encrypt_body_for_envelope( + type_=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=42, + body=body, + epoch_secret=secret, + ) + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=42, + body={}, + encrypted_body=enc, + ) + decoded = decrypt_envelope_body(env, secret) + assert decoded == body + + +def test_decrypt_envelope_body_rejects_no_encrypted_body(): + env = WireEnvelope( + type=TYPE_ANNOUNCE_ROOT, + sender_id="alice", + epoch_id=1, + ts=0, + body={"document_root": "00" * 32}, + ) + with pytest.raises(ValueError, match="no encrypted_body"): + decrypt_envelope_body(env, b"\x00" * 32)