"""design review v2.3 review: each test pins a real bug. All six fixed in v2.4 (xfail markers removed). Originally xfail(strict=True) so the suite stayed green
until the source is fixed; when a fix lands the test XPASSes -> fails -> remove the marker."""
import json, os, shutil
import sys
import pytest, pyrage
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import aicep_sub
from aicep_common import Ledger, Reject, b64, canonical
from aicep_pub import publish
from test_delivery import w, send, OBJ # noqa: F401 (fixture)
from test_tamper import env, pub, sub, resign # noqa: F401 (fixture)
BUG = pytest.mark.xfail(strict=True, reason="design review v2.3 finding (open)") # unused since v2.4
# fixed v2.4: V1 must: reorder-hold trusts an UNVERIFIED manifest seq
def test_unsigned_forged_high_seq_is_not_held(w):
res = send(w); d = res["batch_dir"]
files = {n: open(os.path.join(d, n), "rb").read() for n in os.listdir(d)}
m = json.loads(files["manifest.json"]); m["seq"] = 999
files["manifest.json"] = canonical(m) # signature now invalid
w["bus"].place_raw("c1", "forged-x", files, folder="agent-a")
shutil.rmtree(os.path.join(w["bus"].root, "c1", "pub", "agent-a", res["batch_id"]))
assert w["recv"]().run_once()["forged-x"] == "dlq" # actual: "held" (then gap_timeout email)
# fixed v2.4: V2 must: lease is not re-checked before commit
def test_expired_lease_holder_cannot_commit(w):
send(w); r = w["recv"](consumer="r1")
(d,) = w["bus"].fetch("c1", 10, 300, "r1", folders={"agent-a"})
w["clk"].t += 301 # lease gone; r2 may now take it
assert w["bus"].fetch("c1", 10, 300, "r2", folders={"agent-a"}) # r2 holds it too
assert r.handle(d)[0] != "committed"
# fixed v2.4: V3 should: v2 sender can drop prev_batch_id and skip the chain check (downgrade)
def test_missing_prev_batch_id_after_chain_started_rejected(w):
send(w); r = w["recv"](); r.run_once()
b = publish(OBJ, [w["recip"]], w["sk"], "A", "c1", str(w["t"] / "f"), str(w["t"] / "fps.json"),
now=w["clk"].t, seq=2, ttl=3600)
w["bus"].publish(b["batch_dir"], {"compartment": "c1", "batch_id": b["batch_id"], "folder": "agent-a"})
assert r.run_once()[b["batch_id"]] == "dlq"
# fixed v2.4: V4 should: size cap trusts declared size_padded, decrypts before checking real size
def test_size_cap_checks_bytes_on_disk_before_decrypt(env, monkeypatch):
big = [{"name": "b", "type": "note", "classification": "internal", "body": "x" * 200_000}]
d = publish(big, [env["recip"]], env["sk"], "swarmA", "c1", str(env["pub"]), env["pstate"], now=1_800_000_000, enforce_limits=False)["batch_dir"]
resign(env, d, lambda m: [f.__setitem__("size_padded", 4096) for f in m["files"]])
called = []
real = pyrage.decrypt
monkeypatch.setattr(aicep_sub.pyrage, "decrypt", lambda *a: called.append(1) or real(*a))
with pytest.raises(Reject):
sub(env, d)
assert not called # actual: decrypts 256 KiB first
# fixed v2.4: V5 should: non-int signed fields raise ValueError (not Reject) -> CLI traceback / receiver poison loop
def test_non_int_created_is_reject_not_crash(env):
d = pub(env); resign(env, d, lambda m: m.__setitem__("created", "soon"))
reg = json.loads(json.dumps(env["registry"]))
for v in reg.values(): v["revoked_at"] = 1_800_000_000 + 1000
with pytest.raises(Reject):
sub(env, d, registry=reg)
# fixed v2.4: V6 should: ledger verify() can't detect tail truncation (no external anchor)
def test_ledger_tail_truncation_detected(w):
send(w); send(w); w["recv"]().run_once()
p = w["t"] / "r" / "ledger.jsonl"; head = Ledger(str(p)).head()
lines = open(p).read().splitlines(); open(p, "w").write("\n".join(lines[:-2]) + "\n")
L = Ledger(str(p))
assert not L.verify() or L.head() == head # actual: verify() True, history silently shorter
