"""v2.3 delivery-guarantee tests over the LocalFolderAdapter. Throwaway keys; nothing persists."""
import json, os, shutil, sys
import pytest, pyrage
from cryptography.hazmat.primitives import serialization as S
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, HERE)
from aicep_common import b64, canonical, key_id, Ledger, HOLD_WINDOW, unb64
from aicep_pub import Outbox, publish
from aicep_sub import Receiver, Crash
from aicep_transport import LocalFolderAdapter, TransportAdapter, READY
OBJ = [{"name": "n", "type": "note", "classification": "internal", "body": "hi"}]
class Clock:
def __init__(self, t=1_800_000_000): self.t = t
def __call__(self): return self.t
def raw(k): return k.public_key().public_bytes(S.Encoding.Raw, S.PublicFormat.Raw)
@pytest.fixture
def w(tmp_path):
clk = Clock()
sk, rk, ident = Ed25519PrivateKey.generate(), Ed25519PrivateKey.generate(), pyrage.x25519.Identity.generate()
reg = {key_id(raw(sk)): {"sender_id": "A", "ed25519_pub": b64(raw(sk)), "state": "verified",
"verified_by": "human", "verified_at": clk.t - 10, "expires": clk.t + 86400 * 90, "revoked_at": None,
"folder": "agent-a"}}
bus = LocalFolderAdapter(str(tmp_path / "bus"), now=clk)
ob = Outbox(str(tmp_path / "outbox.json"), sk, "A", str(tmp_path / "s_ledger.jsonl"), now=clk, folder="agent-a")
def recv(**kw):
return Receiver(ident, reg, str(tmp_path / "r"), "c1", rk, kw.pop("bus", bus), now=clk, **kw)
return dict(t=tmp_path, clk=clk, sk=sk, rk=rk, ident=ident, reg=reg, bus=bus, ob=ob, recv=recv,
recip=str(ident.to_public()), work=str(tmp_path / "work"))
def send(w, ttl=3600 * 20, **kw):
return w["ob"].send(OBJ, [w["recip"]], "c1", w["bus"], w["work"], ttl=ttl, **kw)
def test_adapter_is_abstract():
with pytest.raises(TypeError):
TransportAdapter()
assert isinstance(LocalFolderAdapter("/tmp/x"), TransportAdapter)
# 1. commit marker / staging
def test_half_uploaded_batch_is_invisible_then_completes(w):
bad = LocalFolderAdapter(w["bus"].root, now=w["clk"], crash_after_files=2)
with pytest.raises(IOError):
w["ob"].send(OBJ, [w["recip"]], "c1", bad, w["work"], ttl=72000)
assert w["recv"]().run_once() == {} # nothing visible
rows = w["bus"].list_pending("c1")
assert rows and rows[0]["state"] == "_staging" and not rows[0]["ready"]
bid = w["ob"].staged()[0] # still staged in outbox
w["ob"].send(OBJ, [w["recip"]], "c1", w["bus"], w["work"], ttl=72000, batch_id=bid)
assert w["recv"]().run_once()[bid][0] == "committed"
def test_dir_without_ready_marker_skipped(w):
res = send(w); d = res["batch_dir"]
files = {n: open(os.path.join(d, n), "rb").read() for n in os.listdir(d)}
w["bus"].place_raw("c1", "partial-x", files, ready=False, folder="agent-a")
out = w["recv"]().run_once()
assert "partial-x" not in out and out[res["batch_id"]][0] == "committed"
# 2. idempotent redelivery / seq reuse
def test_duplicate_resend_gets_original_receipt(w):
res = send(w); r = w["recv"]()
first = r.run_once()[res["batch_id"]][1]
w["bus"].publish(res["batch_dir"], {"compartment": "c1", "batch_id": res["batch_id"], "folder": "agent-a"}) # re-upload same bytes
shutil.rmtree(os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"]), ignore_errors=True)
w["bus"].place_raw("c1", res["batch_id"], {n: open(os.path.join(res["batch_dir"], n), "rb").read()
for n in os.listdir(res["batch_dir"])}, folder="agent-a")
kind, again = r.run_once()[res["batch_id"]]
assert kind == "duplicate" and again == first
assert any(n["event"] == "receipt_resent" for n in r.notices)
def test_seq_reuse_with_different_manifest_rejected_with_alarm(w):
res = send(w); r = w["recv"](); r.run_once()
st = json.load(open(w["t"] / "outbox.json")); prev = st["last_published"]
evil = publish([{"name": "x", "type": "note", "classification": "internal", "body": "other"}],
[w["recip"]], w["sk"], "A", "c1", str(w["t"] / "evil"), str(w["t"] / "eps.json"),
now=w["clk"].t, seq=1, ttl=3600)
w["bus"].publish(evil["batch_dir"], {"compartment": "c1", "batch_id": evil["batch_id"], "folder": "agent-a"})
out = r.run_once()
assert out[evil["batch_id"]] == "dlq"
assert any(n["event"] == "equivocation" and n["email"] for n in r.notices)
# 3. outbox
def test_outbox_reserves_and_logs_staged_before_publish(w):
bad = LocalFolderAdapter(w["bus"].root, now=w["clk"], crash_after_files=0)
with pytest.raises(IOError):
w["ob"].send(OBJ, [w["recip"]], "c1", bad, w["work"])
st = json.load(open(w["t"] / "outbox.json"))
(bid, e), = st["batches"].items()
assert e["status"] == "staged" and e["seq"] == 1
assert [x["event"] for x in Ledger(str(w["t"] / "s_ledger.jsonl")).entries()] == ["staged"]
res = send(w) # a new batch takes seq 2, not 1
assert res["seq"] == 2
with pytest.raises(ValueError):
send(w, batch_id=res["batch_id"]) # published batch can't be resumed
# 4. dlq
def test_poison_batch_goes_to_dlq_with_signed_reason_and_human_redrive(w):
res = send(w); d = os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"])
f = sorted(x for x in os.listdir(d) if x.endswith(".age") and x != "meta.age")[0]
p = os.path.join(d, f); b = bytearray(open(p, "rb").read()); b[100] ^= 1; open(p, "wb").write(bytes(b))
r = w["recv"]()
assert r.run_once()[res["batch_id"]] == "dlq"
rr = json.load(open(os.path.join(w["bus"].root, "c1", "dlq", res["batch_id"], "reason.json")))
assert rr["record"]["reason"] == "hash_mismatch" and rr["record"]["attempts"] == 1
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
Ed25519PublicKey.from_public_bytes(raw(w["rk"])).verify(unb64(rr["sig"]), canonical(rr["record"]))
assert any(n["event"] == "dlq" and n["email"] for n in r.notices)
with pytest.raises(PermissionError):
w["bus"].redrive("c1", res["batch_id"], operator="")
w["bus"].redrive("c1", res["batch_id"], operator="human-operator")
assert os.path.isdir(os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"]))
def test_transient_errors_retry_three_times_then_dlq(w):
res = send(w); bid = res["batch_id"]
for i in (1, 2):
assert w["recv"](crash_at="transient").run_once()[bid] == f"retry({i})"
w["clk"].t += 60 * i + 1
r = w["recv"](crash_at="transient")
assert r.run_once()[bid] == "dlq"
rr = json.load(open(os.path.join(w["bus"].root, "c1", "dlq", bid, "reason.json")))
assert rr["record"]["reason"] == "ingest_error" and rr["record"]["attempts"] == 3
def test_crypto_failure_dlq_on_first_try(w):
res = send(w); d = os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"])
open(os.path.join(d, "manifest.json.sig"), "w").write(b64(Ed25519PrivateKey.generate().sign(b"x")) + "\n")
assert w["recv"]().run_once()[res["batch_id"]] == "dlq"
rr = json.load(open(os.path.join(w["bus"].root, "c1", "dlq", res["batch_id"], "reason.json")))
assert rr["record"]["reason"] == "sig_fail" and rr["record"]["attempts"] == 1
def test_qop_floor_unencrypted_payload_to_dlq(w):
res = send(w); d = os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"])
open(os.path.join(d, "plain.txt"), "w").write("plaintext")
r = w["recv"](); assert r.run_once()[res["batch_id"]] == "dlq"
rr = json.load(open(os.path.join(w["bus"].root, "c1", "dlq", res["batch_id"], "reason.json")))
assert rr["record"]["reason"] == "qop"
# 5. retry schedule + cumulative receipts
def test_retry_schedule_then_unconfirmed(w):
res = send(w, ttl=3600 * 20); ob, t0 = w["ob"], w["clk"].t
assert ob.due(w["bus"]) == []
for off in (3600, 4 * 3600, 12 * 3600):
w["clk"].t = t0 + off
assert ob.due(w["bus"]) == [("resend", res["batch_id"])]
w["clk"].t = t0 + 13 * 3600
assert ob.due(w["bus"]) == [("unconfirmed", res["batch_id"])]
def test_expiry_marks_unconfirmed(w):
res = send(w, ttl=1800); w["clk"].t += 1800
assert w["ob"].due(w["bus"]) == [("unconfirmed", res["batch_id"])]
def test_cumulative_receipt_acks_all_lower_seqs(w):
a, b = send(w), send(w); r = w["recv"]()
out = r.run_once()
rc = out[b["batch_id"]][1]
assert rc["record"]["acked_through_seq"] == 2 and len(rc["record"]["ledger_head"]) == 64
assert w["ob"].on_receipt(rc["record"], rc["sig"], raw(w["rk"])) == 2
st = json.load(open(w["t"] / "outbox.json"))
assert all(e["status"] == "acked" for e in st["batches"].values())
assert w["ob"].due(w["bus"]) == []
bad = dict(rc["record"], acked_through_seq=99)
with pytest.raises(Exception):
w["ob"].on_receipt(bad, rc["sig"], raw(w["rk"]))
# 6. state machine, crash recovery, lease
@pytest.mark.parametrize("point", ["committing", "after_state"])
def test_crash_injection_recovers_exactly_once(w, point):
res = send(w); bid = res["batch_id"]
with pytest.raises(Crash):
w["recv"](crash_at=point).run_once()
r2 = w["recv"]()
out = r2.run_once()
assert out == {} # recovered in run_once, nothing redone
st = json.load(open(w["t"] / "r" / "receiver.json"))
assert st["phase"][bid]["phase"] == "committed"
assert json.load(open(w["t"] / "r" / "sub_state.json"))["last_seq"][key_id(raw(w["sk"]))] == 1
events = [e["event"] for e in Ledger(str(w["t"] / "r" / "ledger.jsonl")).entries()]
assert events.count("committed") == 1 and ("recovered_commit" in events or "recovered_apply" in events)
def test_crash_before_verify_redelivered_after_lease_expiry(w):
res = send(w); bid = res["batch_id"]
w["bus"].fetch("c1", 10, 300, "r1", folders={"agent-a"}) # r1 leases then dies
assert w["recv"](consumer="r2").run_once() == {} # lease still held: r2 can't take it
w["clk"].t += 301 # lease expires
assert w["recv"](consumer="r2").run_once()[bid][0] == "committed"
def test_lease_blocks_second_consumer(w):
send(w)
a = w["bus"].fetch("c1", 10, 300, "r1", folders={"agent-a"}); b = w["bus"].fetch("c1", 10, 300, "r2", folders={"agent-a"})
assert len(a) == 1 and b == []
# 7. reorder holding, chain, ledgers
def test_out_of_order_held_then_released_in_order(w):
a = send(w); b = send(w)
shutil.move(os.path.join(w["bus"].root, "c1", "pub", "agent-a", a["batch_id"]), str(w["t"] / "delayed"))
r = w["recv"]()
assert r.run_once()[b["batch_id"]] == "held"
shutil.move(str(w["t"] / "delayed"), os.path.join(w["bus"].root, "c1", "pub", "agent-a", a["batch_id"]))
out = r.run_once()
assert out[a["batch_id"]][0] == "committed" and out[b["batch_id"]][0] == "committed"
def test_gap_persisting_past_window_alerts_human(w):
a = send(w); b = send(w)
shutil.rmtree(os.path.join(w["bus"].root, "c1", "pub", "agent-a", a["batch_id"]))
r = w["recv"](); r.run_once()
w["clk"].t += HOLD_WINDOW + 1
assert r.run_once()[b["batch_id"]] == "held_human"
assert any(n["event"] == "gap_timeout" and n["email"] for n in r.notices)
def test_prev_batch_chain_break_rejected(w):
a = send(w); r = w["recv"](); r.run_once()
forged = publish(OBJ, [w["recip"]], w["sk"], "A", "c1", str(w["t"] / "f"), str(w["t"] / "fps.json"),
now=w["clk"].t, seq=2, prev_batch_id="00000000-0000-4000-8000-000000000000", ttl=3600)
w["bus"].publish(forged["batch_dir"], {"compartment": "c1", "batch_id": forged["batch_id"], "folder": "agent-a"})
assert r.run_once()[forged["batch_id"]] == "dlq"
def test_ledgers_hash_chained_and_tamper_evident(w):
send(w); send(w); w["recv"]().run_once()
for p in (w["t"] / "s_ledger.jsonl", w["t"] / "r" / "ledger.jsonl"):
L = Ledger(str(p)); assert L.verify() and len(L.entries()) >= 4
lines = open(p).read().splitlines(); e = json.loads(lines[1]); e["seq"] = 99
lines[1] = json.dumps(e); open(p, "w").write("\n".join(lines) + "\n")
assert not Ledger(str(p)).verify()
def test_no_content_or_keys_in_ledgers_or_notices(w):
send(w); r = w["recv"](); r.run_once()
blob = open(w["t"] / "r" / "ledger.jsonl").read() + json.dumps(r.notices) + open(w["t"] / "s_ledger.jsonl").read()
assert '"hi"' not in blob and "AGE-SECRET-KEY" not in blob and "PRIVATE" not in blob
