russell@unturf.com
01340013dd
Per fox 2026-04-29: the wire protocol has had ANNOUNCE_FALSIFICATION
since the foundation commit (and MeshWireClient.announce_falsification
since AEAD landed), but the user-facing `mesh sync` only ever fired
ANNOUNCE_ROOT. Falsifications local to one peer never reached others
unless an operator hand-rolled a Python script.
Now `mesh sync` enumerates BOTH categories:
- ANNOUNCE_ROOT most-recent --limit documents (existing path)
- ANNOUNCE_FALSIFICATION most-recent --limit falsifications (new)
Receivers verify Ed25519 sig + per-peer chain-of-claims as before,
write one mesh_received audit event per accepted envelope. Result
JSON now reports both counts:
announced_roots: N (was: announced)
announced_falsifications: N
sent_roots: [...] (was: sent)
sent_falsifications: [...]
Two opt-out flags so operators can scope the broadcast:
--no-roots only push falsifications
--no-falsifications only push roots
Burns are deliberately NOT propagated. Burn semantics are local
kindergarten cleanup ("delete a leaf I shouldn't have written") —
other peers may have legitimately ingested the doc independently.
Falsify is the audit-preserving alternative whose broadcast IS the
right cross-peer signal for "this answer is wrong."
Caveat (deferred): no per-peer dedup state yet. Re-running sync
re-broadcasts the same most-recent N falsifications; receivers get
duplicate mesh_received audit-log entries (no state corruption,
just log noise). A `mesh_sync_state` table tracking
last_falsify_announced_ts per peer URL is the natural follow-up
when the falsification volume grows.
Tests:
- existing test_sync_announces_local_roots updated for new field
names (announced_roots, announced_falsifications)
- new test_sync_announces_falsifications: bob falsifies, syncs,
alice's chain has the ANNOUNCE_FALSIFICATION envelope
- new test_sync_no_falsifications_flag_skips_them: --no-falsifications
skips the broadcast cleanly
339 passed, 1 skipped.
313 lines
10 KiB
Python
313 lines
10 KiB
Python
"""CLI integration: `aborist mesh serve` and `aborist mesh sync`.
|
|
|
|
Wire-layer protocol behavior is covered in test_mesh_wire.py and
|
|
test_mesh_wire_e2e.py — these tests just exercise the argparse plumbing,
|
|
the gating checks (mesh initialized + enabled), and the round-trip
|
|
through MeshWireClient that the sync verb performs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from aborist.cli import build_parser
|
|
from aborist.document import Document
|
|
from aborist.ingest import ingest_source
|
|
from aborist.mesh import init_identity, set_enabled
|
|
from aborist.mesh.members import add_member
|
|
from aborist.mesh.wire import MeshWireServer
|
|
from aborist.store import connect
|
|
|
|
|
|
class _OneDocSource:
|
|
source_type = "mesh_cli_test"
|
|
|
|
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(db_path, *, uri: str, content: str, title: str) -> str:
|
|
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()
|
|
return row["document_root"]
|
|
|
|
|
|
def _bootstrap_two_peers(tmp_path: Path):
|
|
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")
|
|
set_enabled(a_conn, True)
|
|
finally:
|
|
a_conn.close()
|
|
b_conn = connect(bob_db)
|
|
try:
|
|
bob = init_identity(b_conn, group_name="t", member_id="bob")
|
|
set_enabled(b_conn, True)
|
|
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()
|
|
return alice_db, bob_db, alice, bob
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gating: mesh must be initialized AND enabled
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_serve_refuses_when_mesh_not_initialized(tmp_path, capsys):
|
|
db = tmp_path / "x.db"
|
|
parser = build_parser()
|
|
args = parser.parse_args(["--db", str(db), "mesh", "serve", "--port", "0"])
|
|
rc = args.func(args)
|
|
assert rc == 2
|
|
assert "not initialized" in capsys.readouterr().err
|
|
|
|
|
|
def test_serve_refuses_when_mesh_disabled(tmp_path, capsys):
|
|
db = tmp_path / "x.db"
|
|
conn = connect(db)
|
|
try:
|
|
init_identity(conn, group_name="t", member_id="alice")
|
|
# Don't call set_enabled — flag stays off by default.
|
|
finally:
|
|
conn.close()
|
|
parser = build_parser()
|
|
args = parser.parse_args(["--db", str(db), "mesh", "serve", "--port", "0"])
|
|
rc = args.func(args)
|
|
assert rc == 2
|
|
assert "enabled" in capsys.readouterr().err
|
|
|
|
|
|
def test_sync_refuses_when_mesh_disabled(tmp_path, capsys):
|
|
db = tmp_path / "x.db"
|
|
conn = connect(db)
|
|
try:
|
|
init_identity(conn, group_name="t", member_id="alice")
|
|
finally:
|
|
conn.close()
|
|
parser = build_parser()
|
|
args = parser.parse_args(["--db", str(db), "mesh", "sync", "--peer", "http://0:0"])
|
|
rc = args.func(args)
|
|
assert rc == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# sync: announces local document_roots to a peer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_sync_announces_local_roots(tmp_path, capsys):
|
|
"""Bob's CLI sync sends ANNOUNCE_ROOT for each local doc to Alice's
|
|
server; Alice's audit chain records each receive."""
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
|
|
# Bob ingests two documents locally — these are what he'll announce.
|
|
_ingest(bob_db, uri="https://example.com/a", content="Doc A content here.", title="A")
|
|
_ingest(bob_db, uri="https://example.com/b", content="Doc B content here.", title="B")
|
|
|
|
srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
parser = build_parser()
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", srv.url,
|
|
"--limit", "100",
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 0
|
|
finally:
|
|
srv.stop()
|
|
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["status"] == "synced"
|
|
assert payload["peer_member_id"] == "alice"
|
|
assert payload["announced_roots"] == 2
|
|
assert payload["announced_falsifications"] == 0 # bob has none
|
|
assert payload["errors"] == 0
|
|
|
|
# Alice's chain gained one mesh_received event per announce.
|
|
a_conn = connect(alice_db)
|
|
try:
|
|
rcvd = a_conn.execute(
|
|
"SELECT COUNT(*) FROM audit_events WHERE event_type='mesh_received'"
|
|
).fetchone()[0]
|
|
finally:
|
|
a_conn.close()
|
|
assert rcvd == 2
|
|
|
|
|
|
def test_sync_announces_falsifications(tmp_path, capsys):
|
|
"""Bob falsifies a local cache record; mesh sync to alice broadcasts
|
|
the falsification. Alice's audit chain gains one mesh_received event
|
|
whose body carries an ANNOUNCE_FALSIFICATION envelope."""
|
|
import time as _time
|
|
from aborist.store import append_audit, transaction
|
|
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
|
|
# Seed a falsification on bob's side.
|
|
b_conn = connect(bob_db)
|
|
try:
|
|
with transaction(b_conn):
|
|
event_hash = append_audit(
|
|
b_conn,
|
|
event_type="falsify",
|
|
subject_root="ab" * 32,
|
|
body={"cache_key": "ab" * 32, "to_state": "failed", "reason": "wrong"},
|
|
)
|
|
b_conn.execute(
|
|
"INSERT INTO falsifications "
|
|
"(cache_key, state, reason, by_actor, at, audit_event_hash) "
|
|
"VALUES (?, 'failed', 'wrong', 'bob', ?, ?)",
|
|
("ab" * 32, int(_time.time()), event_hash),
|
|
)
|
|
finally:
|
|
b_conn.close()
|
|
|
|
srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
parser = build_parser()
|
|
# --no-roots so we isolate the falsification path.
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", srv.url,
|
|
"--limit", "10",
|
|
"--no-roots",
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 0
|
|
finally:
|
|
srv.stop()
|
|
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["announced_roots"] == 0
|
|
assert payload["announced_falsifications"] == 1
|
|
assert payload["errors"] == 0
|
|
|
|
# Alice's chain gained one mesh_received event with the falsification body.
|
|
import json as _json
|
|
a_conn = connect(alice_db)
|
|
try:
|
|
last = a_conn.execute(
|
|
"SELECT event_type, body FROM audit_events "
|
|
"WHERE event_type = 'mesh_received' "
|
|
"ORDER BY seq DESC LIMIT 1"
|
|
).fetchone()
|
|
finally:
|
|
a_conn.close()
|
|
body = _json.loads(last["body"])
|
|
assert body["wire_type"] == "ANNOUNCE_FALSIFICATION"
|
|
assert body["envelope_body"]["cache_key"] == "ab" * 32
|
|
assert body["envelope_body"]["reason"] == "wrong"
|
|
|
|
|
|
def test_sync_no_falsifications_flag_skips_them(tmp_path, capsys):
|
|
"""--no-falsifications skips the falsification broadcast. Bob's
|
|
falsification is NOT announced. Alice gets only the root broadcasts."""
|
|
import time as _time
|
|
from aborist.store import append_audit, transaction
|
|
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
b_conn = connect(bob_db)
|
|
try:
|
|
with transaction(b_conn):
|
|
event_hash = append_audit(
|
|
b_conn, event_type="falsify",
|
|
subject_root="cd" * 32,
|
|
body={"cache_key": "cd" * 32, "to_state": "failed", "reason": "x"},
|
|
)
|
|
b_conn.execute(
|
|
"INSERT INTO falsifications "
|
|
"(cache_key, state, reason, by_actor, at, audit_event_hash) "
|
|
"VALUES (?, 'failed', 'x', 'bob', ?, ?)",
|
|
("cd" * 32, int(_time.time()), event_hash),
|
|
)
|
|
finally:
|
|
b_conn.close()
|
|
|
|
srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
parser = build_parser()
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", srv.url,
|
|
"--limit", "10",
|
|
"--no-falsifications",
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 0
|
|
finally:
|
|
srv.stop()
|
|
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["announced_falsifications"] == 0
|
|
|
|
|
|
def test_sync_reports_unreachable_peer(tmp_path, capsys):
|
|
alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path)
|
|
parser = build_parser()
|
|
# Bind 127.0.0.1:1 — almost certainly nothing listening here.
|
|
args = parser.parse_args([
|
|
"--db", str(bob_db), "mesh", "sync",
|
|
"--peer", "http://127.0.0.1:1",
|
|
])
|
|
rc = args.func(args)
|
|
assert rc == 2
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["status"] == "peer_unreachable"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# serve: smoke-test the server starts and responds via /mesh/info
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_serve_responds_on_info_when_started_in_thread(tmp_path):
|
|
"""We don't actually invoke `mesh serve` (which blocks); we exercise
|
|
MeshWireServer's start_in_thread path the verb relies on."""
|
|
import httpx
|
|
|
|
db = tmp_path / "x.db"
|
|
conn = connect(db)
|
|
try:
|
|
init_identity(conn, group_name="t", member_id="alice")
|
|
set_enabled(conn, True)
|
|
finally:
|
|
conn.close()
|
|
|
|
srv = MeshWireServer(db, host="127.0.0.1", port=0)
|
|
srv.start_in_thread()
|
|
try:
|
|
r = httpx.get(srv.url + "/mesh/info", timeout=5.0)
|
|
r.raise_for_status()
|
|
info = r.json()
|
|
finally:
|
|
srv.stop()
|
|
assert info["member_id"] == "alice"
|
|
assert info["group_name"] == "t"
|