tests.py

#!/usr/bin/env python3
"""tests.py - BAHI protocol tests. Each must PASS. Exit 0 = all green.
Pure stdlib. Covers: honest MATCH, attack classes, determinism, quorum,
group binding, corrupt-file handling, and the v1.3 hardening fixes."""
import sys
from chain import BahiChain, receipt_payload, verify_receipt, MIN_WITNESSES, h
from witness import sign_entry

PASS = 0
FAIL = 0
SKIP = 0

def t(name, cond, detail=""):
    global PASS, FAIL
    if cond:
        PASS += 1
        print("  PASS %s" % name)
    else:
        FAIL += 1
        print("  FAIL %s %s" % (name, detail))

def raises(name, exc, fn):
    try:
        fn()
        t(name, False, "no exception raised")
    except exc:
        t(name, True)
    except Exception as ex:
        t(name, False, "wrong exception: %s" % type(ex).__name__)

def make_chain():
    c = BahiChain("G-RAJ-042")
    ts = "2026-08-02T10:00:00"
    c.add_event(1, "contribution", "Sita", 10000, ts)
    c.add_event(2, "contribution", "Geeta", 10000, ts)
    c.add_event(3, "contribution", "Reema", 10000, ts)
    c.add_event(4, "repayment", "Kavita", 10000, ts)
    c.add_event(5, "loan", "Asha", 50000, ts)
    c.add_event(6, "contribution", "Sita", 10000, ts)
    c.add_event(7, "contribution", "Sita", 10000, ts)
    root = c.close_meeting("M07", ts)
    for w in ("Meera", "Laxmi"):
        root["witnesses"].append(sign_entry({"root": root["root_hash"], "meeting": "M07"}, "pass-" + w, w))
    return c, root

def receipt(c, root):
    return receipt_payload("G-RAJ-042", "M07", root, "Sita", chain=c)

t("honest chain -> MATCH", verify_receipt(*[make_chain()[0], receipt(*make_chain())])[0])

# 1. edit past amount
c, root = make_chain()
c.events[6]["amount_paise"] = 1000
ok, det = verify_receipt(c, receipt_payload("G-RAJ-042", "M07", root, "Sita"))
t("edit past amount -> FORK", not ok and "event-hash-mismatch" in det, det)

# 2. delete event
c, root = make_chain()
del c.events[3]
ok, det = verify_receipt(c, receipt_payload("G-RAJ-042", "M07", root, "Sita"))
t("delete event -> FORK", not ok and "FORK" in det, det)

# 3. reorder events
c, root = make_chain()
c.events[4], c.events[5] = c.events[5], c.events[4]
ok, det = verify_receipt(c, receipt_payload("G-RAJ-042", "M07", root, "Sita"))
t("reorder events -> FORK", not ok and "FORK" in det, det)

# 4. tamper meeting root while member already holds original receipt
c, root = make_chain()
r = receipt(c, root)
root["root_hash"] = "0" * 64
ok, det = verify_receipt(c, r)
t("tamper meeting root -> FORK", not ok and "FORK" in det, det)

# 5. witness removed from chain after receipt issued
c, root = make_chain()
r = receipt(c, root)
root["witnesses"] = [root["witnesses"][0]]
ok, det = verify_receipt(c, r)
t("missing witness sig -> quorum-fail", not ok and "quorum-fail" in det, det)

# 6. forged receipt root
c, root = make_chain()
r = receipt(c, root)
r["root"] = "f" * 64
ok, det = verify_receipt(c, r)
t("forged receipt root -> FORK", not ok and "FORK" in det, det)

# 7. ghost meeting id
c, root = make_chain()
ok, det = verify_receipt(c, {"group": "G-RAJ-042", "meeting": "M99", "root": "0"*64, "witnesses": []})
t("ghost meeting -> meeting-root-missing", not ok and "meeting-root-missing" in det, det)

# 8. determinism: two identical chains give identical verdicts
c1, r1 = make_chain()
c2, r2 = make_chain()
v1 = verify_receipt(c1, receipt_payload("G-RAJ-042", "M07", r1, "Sita", chain=c1))
v2 = verify_receipt(c2, receipt_payload("G-RAJ-042", "M07", r2, "Sita", chain=c2))
t("determinism -> identical verdicts", v1 == v2 and v1[0] is True, (v1, v2))

# 9. quorum: zero-witness close must FAIL
c = BahiChain("G-RAJ-042")
c.add_event(1, "contribution", "Sita", 10000, "t")
root0 = c.close_meeting("M1", "t", witnesses=[])
ok, det = verify_receipt(c, receipt_payload("G-RAJ-042", "M1", root0, "Sita", chain=c))
t("zero-witness close -> quorum-fail", not ok and "quorum-fail" in det, det)

# 10. group binding: receipt for another group must FAIL
c, root = make_chain()
ok, det = verify_receipt(c, receipt_payload("G-OTHER", "M07", root, "Sita"))
t("wrong group receipt -> group-mismatch", not ok and "group-mismatch" in det, det)

# 11. corrupt chain file: load must not raise, verify reports corruption
import json, tempfile, os
c, root = make_chain()
d = c.export()
del d["events"][2]["prev"]
fp = tempfile.mktemp(suffix=".json")
with open(fp, "w") as f:
    json.dump(d, f)
c2 = BahiChain.load(fp)
ok, det = verify_receipt(c2, receipt(c, root))
t("corrupt chain -> structured fail (no crash)", not ok and ("corrupt" in det or "FORK" in det), det)
os.unlink(fp)

# 12. delete the MEETING-CLOSE event but keep roots[] metadata
c, root = make_chain()
r12 = receipt_payload("G-RAJ-042", "M07", root, "Sita")
c.events = [e for e in c.events if e.get("type") != "MEETING-CLOSE"]
ok, det = verify_receipt(c, r12)
t("delete MEETING-CLOSE -> meeting-close-missing", not ok and "meeting-close-missing" in det, det)

# 13. append a fake entry AFTER the close event (hand-edited file: add_event
#     now blocks this at write time, but a tampered file can still smuggle it)
c, root = make_chain()
r13 = receipt_payload("G-RAJ-042", "M07", root, "Sita")
ghost = {"seq": 9, "group": "G-RAJ-042", "type": "contribution", "member": "Ghost",
         "amount_paise": 10000, "ts": "2026-08-02T10:00:00", "prev": c.events[-1]["hash"]}
ghost["hash"] = h(ghost["prev"], "G-RAJ-042", 9, "contribution", "Ghost", 10000, "2026-08-02T10:00:00")
c.events.append(ghost)
ok, det = verify_receipt(c, r13)
t("append after close -> events-after-close", not ok and "events-after-close" in det, det)

# 14. member binding: a CONSISTENT rewrite (attacker re-links all hashes) is
#     caught only by the member receipt binding
c, root = make_chain()
r14 = receipt_payload("G-RAJ-042", "M07", root, "Sita", chain=c)
c.events[0]["amount_paise"] = 5000
prev = h("GENESIS", c.group_id)
for ev in c.events:
    ev["prev"] = prev
    ev["hash"] = h(ev["prev"], c.group_id, ev["seq"], ev["type"], ev["member"], ev["amount_paise"], ev["ts"])
    prev = ev["hash"]
ok, det = verify_receipt(c, r14)
t("consistent rewrite caught (close-hash recompute fires first)",
  not ok and ("member-event-missing-or-tampered" in det or "close hash recompute" in det), det)

# 15. member binding: receipt bound to a member with no events fails
c, root = make_chain()
r15 = receipt_payload("G-RAJ-042", "M07", root, "Nobody", chain=c)
ok, det = verify_receipt(c, r15)
t("member with no events -> member-not-in-chain", not ok and "member-" in det, det)

# ---------------------------------------------------------------- v1.3 fixes
# 16. duplicate seq rejected (A10/A20)
c = BahiChain("G")
c.add_event(1, "contribution", "A", 100, "t")
raises("duplicate seq rejected", ValueError, lambda: c.add_event(1, "contribution", "B", 200, "t"))

# 17. out-of-order seq rejected (A10)
c = BahiChain("G")
c.add_event(1, "contribution", "A", 100, "t")
raises("out-of-order seq rejected", ValueError, lambda: c.add_event(99, "contribution", "C", 300, "t"))

# 18. string / bool seq rejected (A21)
raises("string seq rejected", ValueError, lambda: BahiChain("G").add_event("one", "contribution", "A", 100, "t"))
raises("bool seq rejected", ValueError, lambda: BahiChain("G").add_event(True, "contribution", "A", 100, "t"))

# 19. float amount truncation rejected (A08)
raises("float amount rejected", ValueError, lambda: BahiChain("G").add_event(1, "loan", "A", 10000.9, "t"))

# 20. bool amount rejected (A21)
raises("bool amount rejected", ValueError, lambda: BahiChain("G").add_event(1, "loan", "A", True, "t"))

# 21. MEETING-CLOSE injection rejected (A11)
raises("MEETING-CLOSE injection rejected", ValueError, lambda: BahiChain("G").add_event(1, "MEETING-CLOSE", "__root__", 0, "t"))

# 22. reserved member '__root__' rejected (A18)
raises("__root__ member rejected", ValueError, lambda: BahiChain("G").add_event(1, "contribution", "__root__", 100, "t"))

# 23. re-closing a meeting rejected (A05)
def _reclose():
    c = BahiChain("G"); c.add_event(1, "contribution", "A", 100, "t")
    c.close_meeting("M01", "t"); c.close_meeting("M01", "t")
raises("re-close meeting rejected", ValueError, _reclose)

# 24. group rename detected (N1)
c = BahiChain("G-ORIG")
c.add_event(1, "contribution", "Sita", 10000, "t")
c.close_meeting("M07", "t")
c.group_id = "G-RENAMED"
t("group rename -> verify fails", not c.verify()[0], c.verify()[2])

# 25. empty group rejected at write + verify (N3)
raises("empty group add_event rejected", ValueError, lambda: BahiChain("").add_event(1, "contribution", "A", 100, "t"))
t("empty group verify -> corrupt", BahiChain("").verify()[0] is False and BahiChain("").corrupt)

# 26. over-repayment clamped, excess surfaced (A09)
from loans import balances
c = BahiChain("G"); c.add_event(1, "loan", "Asha", 10000, "t"); c.add_event(2, "repayment", "Asha", 50000, "t")
b = balances(c)["Asha"]
t("over-repayment -> outstanding clamped >= 0", b["outstanding_paise"] == 0 and b["over_repaid_paise"] == 40000, b)

# 27. correction reduces outstanding (N4)
c = BahiChain("G"); c.add_event(1, "loan", "Asha", 10000, "t"); c.add_event(2, "correction", "Asha", 3000, "t")
b = balances(c)["Asha"]
t("correction -> outstanding reduced", b["outstanding_paise"] == 7000, b)

# 28. arbitrary witness strings rejected (A02)
c = BahiChain("G")
for i in range(1, 4): c.add_event(i, "contribution", "Sita", 10000, "t")
root = c.close_meeting("M07", "t")
root["witnesses"] = ["FAKE_SIG_AAAA", "FAKE_SIG_BBBB"]
ok, det = verify_receipt(c, receipt_payload("G", "M07", root, "Sita", chain=c))
t("arbitrary witness strings -> quorum-fail", not ok and "quorum-fail" in det, det)

# 29. witness crypto verification: correct keys -> MATCH (A02)
c, root = make_chain()
r29 = receipt(c, root)
ok, det = verify_receipt(c, r29, witness_keys={"Meera": "pass-Meera", "Laxmi": "pass-Laxmi"})
t("witness crypto verify (correct keys) -> MATCH", ok and det == "MATCH", det)

# 30. witness crypto verification: wrong key -> invalid (A02)
c, root = make_chain()
r30 = receipt(c, root)
ok, det = verify_receipt(c, r30, witness_keys={"Meera": "WRONG", "Laxmi": "pass-Laxmi"})
t("witness crypto verify (wrong key) -> invalid", not ok and "witness-signature-invalid" in det, det)

# 31. receipt is a deep copy (A04): mutating the chain does not rewrite it
c, root = make_chain()
r31 = receipt(c, root)
before = len(r31["witnesses"])
root["witnesses"].append({"witness": "EVIL", "sig": "0" * 64})
t("receipt deep copy (A04)", len(r31["witnesses"]) == before, "aliased: %d -> %d" % (before, len(r31["witnesses"])))

# 16. THE PR9 CRITICAL: close-content swap with VALID seqs. The attacker
#     changes the close event's own content (re-hashes it) but leaves
#     root_meta pointing at the ORIGINAL root. verify() stays self-consistent,
#     Sita's member events are untouched, yet the recomputed close hash no
#     longer equals the receipt root -> caught by the close-hash recompute tie.
c, root = make_chain()
r16 = receipt_payload("G-RAJ-042", "M07", root, "Sita", chain=c)
close_idx = len(c.events) - 1
tamper_ts = "2026-01-01T00:00:00"
c.events[close_idx]["ts"] = tamper_ts
c.events[close_idx]["hash"] = h(c.events[close_idx]["prev"], c.group_id,
                                c.events[close_idx]["seq"], "MEETING-CLOSE", "__root__", 0, tamper_ts)
okc, _, _ = c.verify()   # self-consistent (seqs valid, hash re-linked)
ok, det = verify_receipt(c, r16)
t("close-content swap -> FORK via close-hash recompute",
  okc and not ok and "close hash recompute" in det, det)

# 17. C1 quorum gaming: one witness signing twice must NOT satisfy quorum
from witness import sign as _sign
c = BahiChain("G")
for i in range(1, 4): c.add_event(i, "contribution", "Sita", 10000, "t")
root = c.close_meeting("M07", "t")
root["witnesses"] = [{"witness": "Meera", "sig": _sign({"root": root["root_hash"], "meeting": "M07"}, "pass-Meera", "Meera")},
                     {"witness": "Meera", "sig": _sign({"root": root["root_hash"], "meeting": "M07"}, "pass-Meera-2", "Meera")}]
ok, det = verify_receipt(c, receipt_payload("G", "M07", root, "Sita", chain=c))
t("quorum gaming (same witness twice) -> quorum-fail", not ok and "quorum-fail" in det, det)

# 18. C1b: duplicate witness names rejected at close_meeting
def _dup_close():
    cc = BahiChain("G"); cc.add_event(1, "contribution", "Sita", 10000, "t")
    cc.close_meeting("M07", "t", witnesses=[
        {"witness": "Meera", "sig": "0"*64}, {"witness": "Meera", "sig": "1"*64}])
raises("duplicate witness at close rejected", ValueError, _dup_close)

# 19. R1: legacy receipt (no member_events) must be rejected
c, root = make_chain()
legacy = {"group": "G-RAJ-042", "meeting": "M07", "root": root["root_hash"],
          "root_seq": root["root_seq"], "member": "Sita", "root_ts": "t",
          "witnesses": [dict(w) for w in root["witnesses"]], "member_events": None}
ok, det = verify_receipt(c, legacy)
t("legacy receipt (no member_events) -> rejected", not ok and "member_events-missing" in det, det)

# 20. B2 corpus insolvency + B1 repeat borrower + B3 orphan correction hints
from exporter import hint_flags as _hf
c = BahiChain("G")
c.add_event(1, "contribution", "Sita", 10000, "t")
c.add_event(2, "loan", "Asha", 100000, "t")   # insolvent (loan > corpus)
c.add_event(3, "loan", "Asha", 50000, "t")    # repeat borrower, no repayment
c.add_event(4, "correction", "Asha", 12345, "t")  # orphan (no matching prior amount)
c.close_meeting("M01", "t")
hints = {f["hint"] for f in _hf(c)}
t("B2 corpus insolvency hint", "corpus_insolvency" in hints, hints)
t("B1 repeat borrower hint", "repeat_borrower" in hints, hints)
t("B3 orphan correction hint", "orphan_correction" in hints, hints)

# ------------------------------------------------- Ed25519 witness signatures
# The HMAC-symmetric caveat: offline members can verify an Ed25519 witness
# signature with ONLY the public key embedded in the receipt (no shared secret).
from witness import (generate_keypair as _gen, sign_entry_ed25519 as _sed,
                     verify_sig_ed25519 as _ved, ed25519_available as _ea,
                     is_valid_verify_key as _ivk, is_valid_ed_sig as _ies)
from chain import BahiChain as _BC

def _ed_chain():
    c = _BC("G-ED")
    for i in range(1, 4):
        c.add_event(i, "contribution", "Sita", 10000, "t")
    root = c.close_meeting("M1", "t")
    kps = {"Meera": _gen(), "Laxmi": _gen()}
    for name, kp in kps.items():
        root["witnesses"].append(_sed({"root": root["root_hash"], "meeting": "M1"},
                                      kp["signing_key"], name))
    return c, root, kps

if _ea():
    t("ed25519 backend present (cryptography)", True)
else:
    SKIP += 1
    print("  SKIP ed25519 backend present (cryptography; zero-dep mode)")

if _ea():
    # E1: offline verify MATCH - a member holding ONLY the receipt (no keys) can
    #     cryptographically verify every witness signature via the embedded public key.
    c, root, kps = _ed_chain()
    ok, det = verify_receipt(c, receipt_payload("G-ED", "M1", root, "Sita", chain=c))
    t("E1 ed25519 offline verify -> MATCH (no witness_keys)", ok and det == "MATCH", det)

    # E2: well-formed keypair + signature encoding
    kp = _gen()
    t("E2 keypair verify_key is 64-hex", _ivk(kp["verify_key"]))
    sig = _sed({"root": "x", "meeting": "y"}, kp["signing_key"], "Meera")
    t("E2 ed25519 sig is valid base64", _ies(sig["sig"]))
    t("E2 ed25519 round-trip verify", _ved({"root": "x", "meeting": "y"}, sig["sig"], kp["verify_key"]))

    # E3: witness signed the WRONG payload (root R2 vs receipt R1). Signature is
    #     well-formed and subset-matches, but crypto verification fails offline.
    c, root, kps = _ed_chain()
    r = receipt_payload("G-ED", "M1", root, "Sita", chain=c)
    forged = _sed({"root": "0" * 64, "meeting": "M1"}, kps["Meera"]["signing_key"], "Meera")
    root["witnesses"][0] = forged                      # swap into chain too
    r["witnesses"][0] = dict(forged)                    # and receipt (subset passes)
    ok, det = verify_receipt(c, r)
    t("E3 wrong-payload ed25519 sig -> witness-signature-invalid", not ok and "witness-signature-invalid" in det, det)

    # E4: an attacker replaces Meera's public key with her OWN and signs with her own
    #     key. Crypto cannot detect this (her signature IS valid under her key) - this
    #     is the identity-binding gap that requires an out-of-band trust anchor (KYC),
    #     documented in docs/PRODUCT-DECISIONS.md. We assert the honest current limit.
    c, root, kps = _ed_chain()
    r = receipt_payload("G-ED", "M1", root, "Sita", chain=c)
    mallory = _gen()
    evil = _sed({"root": root["root_hash"], "meeting": "M1"}, mallory["signing_key"], "Meera")
    root["witnesses"][0] = dict(evil)
    r["witnesses"][0] = dict(evil)
    ok, det = verify_receipt(c, r)
    t("E4 key-substitution passes crypto (documented identity gap)", ok and det == "MATCH", det)

    # E5: mixed quorum - one Ed25519, one legacy HMAC. Without witness_keys the
    #     Ed25519 sig is still fully verified, the HMAC sig structural-only.
    c = _BC("G-MIX")
    for i in range(1, 4):
        c.add_event(i, "contribution", "Sita", 10000, "t")
    root = c.close_meeting("M1", "t")
    root["witnesses"].append(_sed({"root": root["root_hash"], "meeting": "M1"},
                                  _gen()["signing_key"], "Meera"))
    from witness import sign as _sig
    root["witnesses"].append({"witness": "Laxmi", "sig": _sig({"root": root["root_hash"], "meeting": "M1"}, "pass-Laxmi", "Laxmi")})
    ok, det = verify_receipt(c, receipt_payload("G-MIX", "M1", root, "Sita", chain=c),
                             witness_keys={"Laxmi": "pass-Laxmi"})
    t("E5 mixed ed25519+hmac quorum -> MATCH", ok and det == "MATCH", det)
    # ...and with the WRONG hmac key the hmac half fails
    ok, det = verify_receipt(c, receipt_payload("G-MIX", "M1", root, "Sita", chain=c),
                             witness_keys={"Laxmi": "WRONG"})
    t("E5 mixed quorum wrong hmac key -> invalid", not ok and "witness-signature-invalid" in det, det)

else:
    SKIP += 8
    print("  SKIP E1-E5 ed25519 signature suite (cryptography backend absent, zero-dep mode; HMAC witness path covered above)")

print("\n%d/%d PASSED (%d failed, %d skipped)" % (PASS, PASS + FAIL, FAIL, SKIP))
sys.exit(0 if FAIL == 0 else 1)