open-standards/wai/live-conformance/holdings_verify.py

#!/usr/bin/env python3
"""Independent verifier of the holdings corpus (extensions/holdings.md).

Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and
never runs it. It takes from the second implementations of the texts this extension builds on
what they define: group receipts and their link hashes from the JWP corpus's
(`../jwp-conformance/jwp_verify.py`, jwp-receipts §3), and Ed25519 verification as SPEC §7
*Signatures* states it from the routine every Python verifier of WAI's corpora shares
(`../claim-conformance/_ed25519.py`). Everything this extension adds is written here: the
record's layout and rules (§4), its verdict at an instant against group receipts (§5.1 to
§5.3), queries and an asker's plan (§5.4).

For every case it requires the outcome the corpus states, field for field; that every reason of
§7's table appears in the corpus; then it mutates copies of the corpus, one defect each, and
requires every copy to fail; and it drops each of its own checks in turn and requires the corpus
to catch the weakened verifier.

    python3 holdings_verify.py <live-conformance/holdings>

Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier is
caught. Python 3.8 or later, the standard library and the JWP verifier's own requirements.
"""

import copy
import hashlib
import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
for d in ("claim-conformance", "jwp-conformance"):
    sys.path.insert(0, os.path.join(HERE, "..", d))
import _ed25519 as ED  # noqa: E402
import jwp_verify as JV  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "holdings.md")
MAGIC = b"WHD1"
DOMAIN = b"wai:holdings\x01"
MAX_LEN = 1 << 20
MAX_INT = (1 << 53) - 1

# A weakened run drops one of these checks (see `WEAKENINGS`); None checks everything.
DROPPED = None


def drops(name):
    return DROPPED == name


def reasons_table():
    """§7's codes, in order."""
    text = open(EXTENSION, encoding="utf-8").read()
    at = text.index("\n## 7. Reasons")
    end = text.find("\n## ", at + 1)
    rows = text[at:end if end >= 0 else len(text)].splitlines()
    return [r.split("`")[1] for r in rows if r.startswith("| `")]


class Refused(Exception):
    def __init__(self, reason):
        super().__init__(reason)
        self.reason = reason


def bit(bits, i):
    return i // 8 < len(bits) and bits[i // 8] & (0x80 >> (i % 8)) != 0


# ------------------------------------------------------------------ §4: the record

def read(b):
    """§4.4: the layout and rules in the order the bytes are read, then the signature. Returns
    (holder, as_of, expires, entries, signed bytes); each entry (group, leaf_count, bits or None
    for whole)."""
    bad = Refused("holdings_malformed")
    if len(b) > MAX_LEN:
        raise bad
    at = 0

    def take(n):
        nonlocal at
        if at + n > len(b):
            raise bad
        s = b[at:at + n]
        at += n
        return s

    if take(4) != MAGIC or take(1) != b"\x01":
        raise bad
    holder = take(32)
    as_of = int.from_bytes(take(8), "big")
    expires = int.from_bytes(take(8), "big")
    if (as_of >= expires and not drops("window")) or expires > MAX_INT:
        raise bad
    n = int.from_bytes(take(4), "big")
    if n == 0:
        raise bad
    entries = []
    for _ in range(n):
        group = take(32)
        if entries and entries[-1][0] >= group and not drops("order"):
            raise bad
        leaf_count = int.from_bytes(take(8), "big")
        if leaf_count == 0 or leaf_count > MAX_INT:
            raise bad
        form = take(1)[0]
        if form == 0:
            bits = take((leaf_count + 7) // 8)
            past = any(bit(bits, i) for i in range(leaf_count, 8 * len(bits)))
            held = sum(1 for i in range(leaf_count) if bit(bits, i))
            if (past and not drops("past")) or held == 0 or (held == leaf_count and not drops("canonical")):
                raise bad
            entries.append((group, leaf_count, bits))
        elif form == 1:
            entries.append((group, leaf_count, None))
        else:
            raise bad
    signed = b[:at]
    sig = take(64)
    if at != len(b):
        raise bad
    if not drops("signature") and ED.verify(holder, DOMAIN + signed, sig) is not None:
        raise Refused("holdings_signature_invalid")
    return holder, as_of, expires, entries, signed


def digest(rec):
    return hashlib.sha256(rec[4]).digest()


# ------------------------------------------------------------------ §5

def holds(rec, group, i):
    """§5.4: None when held, else the reason."""
    for g, n, bits in rec[3]:
        if g == group:
            if i >= n:
                return "object_out_of_range"
            if bits is None or bit(bits, i):
                return None
            return "object_not_held"
    return "group_not_advertised"


def verdict(records, receipts, t):
    """§5.1 to §5.3: each record's label, and the records that stand by position."""
    held = {}
    for r in receipts:
        try:
            g = JV.read_group(r)
            JV.verify_group(g)
        except JV.Refuse:
            continue
        held[JV.link(g)] = g["leaf_count"]
    labels, read_ = [], {}
    for k, b in enumerate(records):
        try:
            rec = read(b)
            if t < rec[1] and not drops("not_yet"):
                raise Refused("holdings_not_yet")
            if t >= rec[2] and not drops("expired"):
                raise Refused("holdings_expired")
            if not drops("leaf_count") and any(g in held and held[g] != n for g, n, _ in rec[3]):
                raise Refused("holdings_leaf_count_mismatch")
            read_[k] = rec
            labels.append("stands")
        except Refused as r:
            labels.append("reject reason=" + r.reason)
    by_holder = {}
    for k, rec in read_.items():
        by_holder.setdefault(rec[0], []).append(k)
    for ks in by_holder.values():
        latest = max(read_[k][1] for k in ks)
        at_latest = [k for k in ks if read_[k][1] == latest]
        contradicted = len({digest(read_[k]) for k in at_latest}) > 1 and not drops("contradiction")
        for k in ks:
            if read_[k][1] < latest and not drops("supersede"):
                labels[k] = "superseded"
            elif read_[k][1] == latest and contradicted:
                labels[k] = "set-aside reason=holdings_contradicted"
    standing = [read_[k] for k in sorted(read_) if labels[k] == "stands"]
    return labels, standing


def plan(standing, wanted):
    """§5.4: each wanted object's holders, in the asker's order, and those no holder holds."""
    holders, origin = [], []
    for w, (group, i) in enumerate(wanted):
        who = []
        for rec in standing:
            if holds(rec, group, i) is None and rec[0].hex() not in who:
                who.append(rec[0].hex())
        if not who:
            origin.append(w)
        holders.append(who)
    return holders, origin


# ------------------------------------------------------------------ the corpus

def record_outcome(b):
    try:
        rec = read(b)
    except Refused as r:
        return {"outcome": "refuse", "reason": r.reason}
    return {"outcome": "read", "digest": digest(rec).hex(), "holder": rec[0].hex(), "entries": len(rec[3])}


def run(kind, c):
    if kind == "records":
        return record_outcome(bytes.fromhex(c["bytes"]))
    if kind == "verdicts":
        labels, _ = verdict([bytes.fromhex(x) for x in c["records"]], c["receipts"], c["at_ms"])
        return {"records": labels}
    if kind == "queries":
        why = holds(read(bytes.fromhex(c["record"])), bytes.fromhex(c["group"]), c["index"])
        return "held" if why is None else why
    labels, standing = verdict([bytes.fromhex(x) for x in c["records"]], c["receipts"], c["at_ms"])
    holders, origin = plan(standing, [(bytes.fromhex(w["group"]), w["index"]) for w in c["wanted"]])
    return {"records": labels, "holders": holders, "from_origin": origin}


KINDS = ("records", "verdicts", "queries", "plans")


def strings(v, out):
    if isinstance(v, str):
        out.append(v)
    elif isinstance(v, list):
        for x in v:
            strings(x, out)
    elif isinstance(v, dict):
        for x in v.values():
            strings(x, out)


def check(doc, quiet=False):
    """The failures of a corpus document: each case's outcome, and every reason of §7."""
    fails = []
    for kind in KINDS:
        for c in doc[kind]:
            try:
                got = run(kind, c)
            except (Refused, KeyError, TypeError, ValueError, IndexError) as e:
                got = {"error": repr(e)}
            if got != c["outcome"]:
                fails.append(c["name"])
                if not quiet:
                    print(f"FAIL {c['name']}\n  stated {json.dumps(c['outcome'])}\n  got    {json.dumps(got)}")
    stated = []
    for kind in KINDS:
        for c in doc[kind]:
            strings(c["outcome"], stated)
    for code in reasons_table():
        if not any(s == code or s.endswith("reason=" + code) for s in stated):
            fails.append("no case gives " + code)
    return fails


def mutations(doc):
    """Copies of the corpus, one defect each, every one of which must fail."""
    def case(d, name):
        return next(c for k in KINDS for c in d[k] if c["name"] == name)

    def m_bit(d):
        c = case(d, "whole_and_partial")
        b = bytearray.fromhex(c["bytes"])
        b[-70] ^= 0x01
        c["bytes"] = b.hex()

    def m_digest(d):
        case(d, "several_groups")["outcome"]["digest"] = "00" * 32

    def m_instant(d):
        case(d, "last_instant")["at_ms"] += 1

    def m_superseded(d):
        case(d, "later_record_stands")["outcome"]["records"][0] = "stands"

    def m_query(d):
        case(d, "not_held")["index"] = 8

    def m_plan(d):
        case(d, "three_neighbours")["outcome"]["from_origin"] = []

    def m_order(d):
        c = case(d, "three_neighbours")
        c["records"] = list(reversed(c["records"]))

    def m_receipt(d):
        case(d, "leaf_count_mismatch")["receipts"] = []

    def m_reason_gone(d):
        d["verdicts"].remove(case(d, "contradicted"))

    return [m_bit, m_digest, m_instant, m_superseded, m_query, m_plan, m_order, m_receipt, m_reason_gone]


WEAKENINGS = ["window", "order", "past", "canonical", "signature", "not_yet", "expired", "leaf_count", "contradiction",
              "supersede"]


def main(argv):
    global DROPPED
    if len(argv) != 2:
        print(__doc__)
        return 2
    doc = json.load(open(os.path.join(argv[1], "vectors.json"), encoding="utf-8"))
    extra = sorted(set(os.listdir(argv[1])) - {"vectors.json", "README.md"})
    fails = check(doc)
    if extra:
        fails.append(f"files the corpus does not hold: {extra}")
    # §4.4's bound, which no vector carries (a megabyte of hex): checked here.
    try:
        read(b"\x00" * (MAX_LEN + 1))
        fails.append("a record past 1,048,576 bytes was read")
    except Refused as r:
        if r.reason != "holdings_malformed":
            fails.append("a record past the bound was refused " + r.reason)
    print(f"{sum(len(doc[k]) for k in KINDS)} cases ({', '.join(f'{len(doc[k])} {k}' for k in KINDS)}): {len(fails)} failure(s)")
    caught = 0
    for m in mutations(doc):
        d = copy.deepcopy(doc)
        m(d)
        if check(d, quiet=True):
            caught += 1
        else:
            fails.append(f"mutation {m.__name__} was not caught")
    print(f"{caught} of {len(mutations(doc))} mutated copies fail")
    weak = 0
    for w in WEAKENINGS:
        DROPPED = w
        if check(doc, quiet=True):
            weak += 1
        else:
            fails.append(f"a verifier without its {w} check passes the corpus")
    DROPPED = None
    print(f"{weak} of {len(WEAKENINGS)} weakened verifiers are caught")
    for f in fails:
        print("FAIL", f)
    return 1 if fails else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))