open-standards/wai/signer-conformance/signer_verify.py

#!/usr/bin/env python3
"""Second verifier for signer-conformance/ — key discovery, signing authority and module
authorisation (extensions/jwp-receipts.md §7).

Written from the extension text. It shares no code with wai-rs (`crate::key_log`,
`crate::signer_id`), which generates the corpus. It takes from the claim corpus's own
second verifier (`../claim-conformance/claim_verify.py`) what §7 reuses from §3.1: the
JSON reading of step 1, the canonical form (JCS), BLAKE3 written from its specification,
the strict Ed25519 rule of step 3, the RFC 3339 syntax, and claim verification itself.
Everything §7 adds — the signer_id grammar and base58btc, the key log's rules, the head
walk, anchoring, confirmation, the time comparison and §7.9's module checks — is written
below.

    python3 signer_verify.py [vectors.json]   verify every vector and the canonical fixture
    python3 signer_verify.py fixture          print the fixture's canonical bytes,
                                              signature and revision hash
    python3 signer_verify.py verdicts <file>  print `name<TAB>verdict` for every vector,
                                              ignoring any declared verdict

Exit status 0 only if every vector reaches its declared verdict and the fixture this
script computes equals the literals pinned in wai-rs.
"""
import datetime, hashlib, json, os, re, sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
import claim_verify as CV  # noqa: E402

from cryptography.hazmat.primitives import serialization  # noqa: E402
from cryptography.hazmat.primitives.asymmetric import ed25519  # noqa: E402

D_SET, D_LINK, D_NEXT = b"wai:key-set\x01", b"wai:key-set-link\x01", b"wai:key-set-next\x01"
D_GRANT, D_STATUS = b"wai:module-grant\x01", b"wai:key-status\x01"
GRANT_CEILING, STATUS_CEILING = 86400, 3600
HARDWARE = {"CalibratedInstrument", "OnChipCounter", "HwShunt"}
PINNED_ONLY = {"energy/CalibratedInstrument", "energy/OnChipCounter", "energy/HwShunt",
               "lawful-basis/declaration", "lawful-basis/decision", "aggregate/task"}
REVISION = ["kind", "origin", "seq", "prev", "issued_at", "root", "next_root", "keys", "revoked_modules",
            "grant_max_s", "status_max_age_s", "sig"]
ENTRY = ["id", "key", "may_sign", "status", "until", "compromised_since"]
GRANT = ["kind", "origin", "module", "form", "capabilities", "not_before", "not_after", "signer_pubkey",
         "signer_id", "sig"]
STATUS = ["kind", "origin", "head", "seq", "issued_at", "sig"]
RANK = {"active": 0, "retired": 1, "revoked": 2}


class Stop(Exception):
    """A reason: a log not anchored, a revision not held, an item not confirmed."""

    def __init__(self, reason, level=None, step=None):
        super().__init__(reason)
        self.reason, self.level, self.step = reason, level, step


# ------------------------------------------------------------------ §7.1 signer_id

B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"


def b58encode(b: bytes) -> str:
    n = int.from_bytes(b, "big")
    out = ""
    while n:
        n, r = divmod(n, 58)
        out = B58[r] + out
    return "1" * (len(b) - len(b.lstrip(b"\0"))) + out


def b58decode(s: str):
    if any(c not in B58 for c in s):
        return None
    n = 0
    for c in s:
        n = n * 58 + B58.index(c)
    body = n.to_bytes((n.bit_length() + 7) // 8, "big") if n else b""
    return b"\0" * (len(s) - len(s.lstrip("1"))) + body


LABEL = re.compile(r"[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?\Z")


def host_ok(h):
    labels = h.split(".")
    return (len(h) <= 253 and len(labels) >= 2 and all(LABEL.match(x) for x in labels)
            and not labels[-1].isdigit())


def port_ok(p):
    return re.fullmatch(r"[1-9][0-9]{0,4}", p) is not None and 1 <= int(p) <= 65535 and int(p) != 443


def key_id_ok(k):
    return re.fullmatch(r"[A-Za-z0-9._-]{1,64}", k) is not None


def did_key(pk: bytes) -> str:
    return "did:key:z" + b58encode(b"\xed\x01" + pk)


def classify(sid):
    """('origin', origin, key_id) | ('didkey', key) | ('label',)."""
    if sid.startswith("did:web:"):
        rest = sid[len("did:web:"):]
        if "#" in rest:
            auth, kid = rest.split("#", 1)
            host, port = (auth.split("%3A", 1) + [None])[:2]
            if key_id_ok(kid) and host_ok(host) and (port is None or port_ok(port)):
                return ("origin", "https://" + host + (":" + port if port else ""), kid)
        return ("label",)
    # The base58btc text of 0xed 0x01 and a 32-byte key is 47 characters; a longer one is
    # not its spelling and is not decoded.
    if sid.startswith("did:key:z") and len(sid) - len("did:key:z") <= 47:
        b = b58decode(sid[len("did:key:z"):])
        if b is not None and len(b) == 34 and b[:2] == b"\xed\x01" and did_key(b[2:]) == sid:
            return ("didkey", b[2:])
    return ("label",)


def origin_ok(o):
    m = re.fullmatch(r"https://([^:/]+)(?::([0-9]+))?", o)
    return m is not None and host_ok(m.group(1)) and (m.group(2) is None or port_ok(m.group(2)))


# ------------------------------------------------------------------ times

LOG_TIME = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-5][0-9]Z\Z")


def is_log_time(t):
    return isinstance(t, str) and LOG_TIME.match(t) is not None and CV.is_rfc3339(t)


def secs(t):
    d = datetime.datetime(int(t[0:4]), int(t[5:7]), int(t[8:10]), int(t[11:13]), int(t[14:16]), int(t[17:19]),
                          tzinfo=datetime.timezone.utc)
    return int(d.timestamp())


def instant(t):
    """(minute in UTC, second as written, fraction digits without trailing zeros)."""
    m = re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})[Tt]([0-9]{2}):([0-9]{2}):([0-9]{2})(?:\.([0-9]+))?"
                     r"(?:[Zz]|([+-])([0-9]{2}):([0-9]{2}))", t)
    day = datetime.date(int(m.group(1)), int(m.group(2)), int(m.group(3))).toordinal()
    minute = day * 1440 + int(m.group(4)) * 60 + int(m.group(5))
    if m.group(8):
        off = int(m.group(9)) * 60 + int(m.group(10))
        minute -= off if m.group(8) == "+" else -off
    return minute, int(m.group(6)), (m.group(7) or "").rstrip("0")


def later_than(t, until):
    if not (isinstance(t, str) and CV.is_rfc3339(t)):
        return True
    a, b = instant(t), instant(until)
    if a[:2] != b[:2]:
        return a[:2] > b[:2]
    w = max(len(a[2]), len(b[2]))
    return a[2].ljust(w, "0") > b[2].ljust(w, "0")


# ------------------------------------------------------------------ signatures and hashes

def hexb(v, n):
    if isinstance(v, str) and len(v) == 2 * n and re.fullmatch(r"[0-9a-f]*", v):
        return bytes.fromhex(v)
    return None


def verifies(pk: bytes, sig: bytes, msg: bytes) -> bool:
    try:
        CV.check_signature(pk, sig, msg)
        return True
    except CV.Reject:
        return False


def key_accepted(pk: bytes) -> bool:
    return CV.ED.strict_key(pk)


def commitment(k: bytes) -> bytes:
    return CV.blake3(D_NEXT + k)


def without_sig(d):
    return {k: v for k, v in d.items() if k != "sig"}


def rev_hash(r) -> str:
    return CV.blake3(D_LINK + CV.jcs(r).encode()).hex()


def module_digest(b: bytes) -> str:
    return hashlib.sha256(b).hexdigest()


# ------------------------------------------------------------------ §7.2 the key log

def authority_ok(a):
    return isinstance(a, str) and 1 <= len(a) <= 128 and re.fullmatch(r"[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)*", a)


def exact(d, members):
    return isinstance(d, dict) and set(d) == set(members)


def check_entry(e):
    ok = (exact(e, ENTRY) and isinstance(e["id"], str) and key_id_ok(e["id"]) and hexb(e["key"], 32) is not None
          and key_accepted(bytes.fromhex(e["key"])) and isinstance(e["may_sign"], list) and e["may_sign"]
          and all(authority_ok(a) for a in e["may_sign"]) and len(set(e["may_sign"])) == len(e["may_sign"])
          and e["status"] in RANK and all(e[t] is None or is_log_time(e[t]) for t in ("until", "compromised_since")))
    if not ok:
        return False
    st, u, c = e["status"], e["until"], e["compromised_since"]
    return {"active": u is None and c is None, "retired": u is not None and c is None,
            "revoked": c is not None}[st]


def check_revision(r):
    """Rule 1 on a parsed value."""
    bad = Stop("key_log_malformed")
    if not exact(r, REVISION) or r["kind"] != "wai-key-set" or not isinstance(r["origin"], str):
        raise bad
    if not CV.is_int(r["seq"]) or not (r["prev"] is None or hexb(r["prev"], 32) is not None):
        raise bad
    if (r["seq"] == 0) != (r["prev"] is None) or not is_log_time(r["issued_at"]):
        raise bad
    root, nxt = hexb(r["root"], 32), hexb(r["next_root"], 32)
    if root is None or nxt is None or nxt == commitment(root) or hexb(r["sig"], 64) is None:
        raise bad
    if not (CV.is_int(r["grant_max_s"]) and 1 <= r["grant_max_s"] <= GRANT_CEILING):
        raise bad
    if not (CV.is_int(r["status_max_age_s"]) and 1 <= r["status_max_age_s"] <= STATUS_CEILING):
        raise bad
    if not isinstance(r["keys"], list) or not all(check_entry(e) for e in r["keys"]):
        raise bad
    ids, keys = [e["id"] for e in r["keys"]], [e["key"] for e in r["keys"]]
    if len(set(ids)) != len(ids) or len(set(keys)) != len(keys):
        raise bad
    if any(bytes.fromhex(k) == root or commitment(bytes.fromhex(k)) == nxt for k in keys):
        raise bad
    rm = r["revoked_modules"]
    if not isinstance(rm, list) or not all(hexb(d, 32) is not None for d in rm) or len(set(rm)) != len(rm):
        raise bad
    return r


def parse_revision(text):
    try:
        return check_revision(CV.parse_strict(text))
    except ValueError:
        raise Stop("key_log_malformed")


def check_held(r, origin):
    """Rules 2-3."""
    if r["origin"] != origin:
        raise Stop("origin_mismatch")
    if not verifies(bytes.fromhex(r["root"]), bytes.fromhex(r["sig"]), D_SET + CV.jcs(without_sig(r)).encode()):
        raise Stop("key_log_signature_invalid")


def is_held(r, origin):
    try:
        check_held(r, origin)
        return True
    except Stop:
        return False


def rotation_of(c, p):
    return c["root"] != p["root"] and commitment(bytes.fromhex(c["root"])).hex() == p["next_root"]


def authorised(c, p):
    return c["root"] == p["root"] or rotation_of(c, p)


def check_link(p, c):
    """Rules 4-6."""
    if c["prev"] != rev_hash(p) or c["seq"] != p["seq"] + 1 or c["issued_at"] < p["issued_at"]:
        raise Stop("key_log_chain_break")
    if not ((c["root"] == p["root"] and c["next_root"] == p["next_root"]) or rotation_of(c, p)):
        raise Stop("key_log_root_not_committed")
    pk, ck = p["keys"], c["keys"]
    if len(ck) < len(pk) or c["revoked_modules"][:len(p["revoked_modules"])] != p["revoked_modules"]:
        raise Stop("key_log_rewrites_history")
    for a, b in zip(pk, ck):
        same = a["id"] == b["id"] and a["key"] == b["key"] and a["may_sign"] == b["may_sign"]
        if not same or RANK[b["status"]] < RANK[a["status"]]:
            raise Stop("key_log_rewrites_history")
        for t in ("until", "compromised_since"):
            if a[t] is not None and a[t] != b[t]:
                raise Stop("key_log_rewrites_history")


def read_log(origin, text):
    try:
        doc = CV.parse_strict(text)
    except ValueError:
        raise Stop("key_log_malformed")
    if not exact(doc, ["kind", "revisions"]) or doc["kind"] != "wai-key-log":
        raise Stop("key_log_malformed")
    revs = doc["revisions"]
    if not isinstance(revs, list) or not revs:
        raise Stop("key_log_malformed")
    for r in revs:
        check_revision(r)
        check_held(r, origin)
    if revs[0]["seq"] != 0:
        raise Stop("key_log_chain_break")
    for p, c in zip(revs, revs[1:]):
        check_link(p, c)
    return revs


# ------------------------------------------------------------------ §7.4 step 3, the head

def head(held, anchor):
    """Returns (head_hash, head_revision, moved_to or None)."""
    by = {rev_hash(r): r for r in held}
    if anchor not in by:
        raise Stop("origin_not_anchored")

    def kids(h):
        p = by[h]
        return [(k, c) for k, c in sorted(by.items()) if c["prev"] == h and authorised(c, p)]

    line = [anchor]
    while by[line[-1]]["prev"] in by and by[line[-1]]["prev"] not in line:
        line.append(by[line[-1]]["prev"])
    line.reverse()
    start, moved = anchor, None
    for t, n in zip(line, line[1:]):
        if rotation_of(by[n], by[t]):
            continue
        rots = [(k, c) for k, c in kids(t) if k != n and rotation_of(c, by[t])]
        if len(rots) == 1:
            check_link(by[t], rots[0][1])
            start = moved = rots[0][0]
            break
        if len(rots) > 1:
            raise Stop("key_log_fork")
    cur = start
    while True:
        ks = kids(cur)
        if not ks:
            return cur, by[cur], moved
        if len(ks) > 1:
            ks = [kc for kc in ks if rotation_of(kc[1], by[cur])]
            if len(ks) != 1:
                raise Stop("key_log_fork")
        check_link(by[cur], ks[0][1])
        cur = ks[0][0]


# ------------------------------------------------------------------ §7.3 anchoring

_PUBLISHED = None


def is_published(k: bytes) -> bool:
    """Whether k is the public key of a one-byte seed, or the commitment to one."""
    global _PUBLISHED
    if _PUBLISHED is None:
        keys = [ed25519.Ed25519PrivateKey.from_private_bytes(bytes([b]) * 32).public_key().public_bytes(
            serialization.Encoding.Raw, serialization.PublicFormat.Raw) for b in range(256)]
        _PUBLISHED = set(keys) | {commitment(pk) for pk in keys}
    return k in _PUBLISHED


def lists_published(r):
    return any(is_published(bytes.fromhex(x)) for x in [r["root"], r["next_root"]] + [e["key"] for e in r["keys"]])


def anchor_log(origin, text, path, current):
    """('anchored', hash, path) | ('not_anchored', reason) | ('replaced',)."""
    try:
        revs = read_log(origin, text)
    except Stop as s:
        return ("not_anchored", s.reason)
    hs = [rev_hash(r) for r in revs]
    if current is None or path == "pinned":
        return ("anchored", hs[-1], path, hs[-1] if path == "pinned" else None)
    cur, held = current
    if cur["revision"] in hs:
        # Forward: a pinned record stays pinned and keeps the revision that was pinned.
        if cur["path"] == "pinned" and cur.get("pinned_revision") is not None:
            return ("anchored", hs[-1], "pinned", cur["pinned_revision"])
        return ("anchored", hs[-1], path, None)
    hb = {rev_hash(r): r for r in held if is_held(r, origin)}
    trusted = [cur["revision"]]  # newest first
    while trusted[-1] in hb and hb[trusted[-1]]["prev"] in hb and hb[trusted[-1]]["prev"] not in trusted:
        trusted.append(hb[trusted[-1]]["prev"])
    if hs[-1] in trusted:
        return ("not_anchored", "key_log_rolled_back")
    shared = [i for i, x in enumerate(hs) if x in trusted]
    if shared:
        i = shared[-1]
        if i + 1 < len(revs) and rotation_of(revs[i + 1], revs[i]):
            pos = trusted.index(hs[i])
            child = hb.get(trusted[pos - 1]) if pos > 0 else None
            if child is None or not rotation_of(child, revs[i]):
                return ("anchored", hs[-1], path, None)
    return ("replaced",)


# ------------------------------------------------------------------ §7.4-§7.5 confirmation

def claim_authorities(c):
    body = c["body"]
    prov = body.get("energy_provenance") if isinstance(body, dict) else None
    if c["class"] == "energy":
        own = "energy/" + prov
    elif c["class"] == "lawful-basis":
        own = "lawful-basis/" + body["record"]
    else:
        own = c["class"]
    out = [own]
    if isinstance(prov, str) and "energy/" + prov not in out:
        out.append("energy/" + prov)
    return out


def figure_authorities(c):
    """§7.5: the energy figure of a `reconstruct-session` or a `transmission` claim is an
    item of its own, needing the claim's class and the energy authority of the class its
    `energy` writes; None for a claim of another class, or one that was not metered."""
    e = c["body"].get("energy") if c["class"] in ("reconstruct-session", "transmission") else None
    return None if e is None else [c["class"], "energy/" + e["energy_provenance"]]


def item_authorities(kind, classes):
    if kind == "group":
        # delivery, and energy/<class> for each class the group's labelled leaves and
        # its labelled total carry (section 7.5).
        out = ["delivery"]
        for c in classes:
            if "energy/" + c not in out:
                out.append("energy/" + c)
        return out
    # receipt, and energy/<class> for each class the receipt's figures carry.
    out = ["receipt"]
    for c in classes:
        if "energy/" + c not in out:
            out.append("energy/" + c)
    return out


def confirm(pubkey, sid, authorities, time, revisions, anchors, expected):
    """`expected` is None or a list of {origin, source}; the origin must be in a list given."""
    form = classify(sid)
    if form[0] == "didkey":
        if form[1] != pubkey:
            return "refused step=1 reason=signer_id_names_another_key"
        return "unconfirmed step=1 reason=signer_id_not_origin_bound"
    if form[0] == "label":
        return "unconfirmed step=1 reason=signer_id_not_origin_bound"
    _, origin, kid = form
    held = [r for r in revisions if is_held(r, origin)]
    anc = next((a for a in anchors if a["origin"] == origin), None)
    if anc is None or anc["revision"] not in [rev_hash(r) for r in held]:
        return "unconfirmed step=2 reason=origin_not_anchored"
    try:
        hh, hr, moved = head(held, anc["revision"])
    except Stop as s:
        return "unconfirmed step=3 reason=" + s.reason
    e = next((x for x in hr["keys"] if x["id"] == kid), None)
    if e is None:
        return "refused step=4 reason=key_not_listed"
    if bytes.fromhex(e["key"]) != pubkey:
        return "refused step=4 reason=key_mismatch"
    if e["status"] == "revoked":
        return "refused step=5 reason=key_revoked"
    if e["status"] == "retired" and time is not None and later_than(time, e["until"]):
        return "refused step=5 reason=key_retired"
    if not all(a in e["may_sign"] for a in authorities):
        return "refused step=6 reason=authority_not_granted"
    pinned = anc["path"] == "pinned" and anc.get("pinned_revision") is not None and moved is None
    if any(a in PINNED_ONLY for a in authorities) and not pinned:
        return "unconfirmed step=6a reason=anchor_not_pinned"
    match = None
    if expected is not None:
        match = next((x for x in expected if x["origin"] == origin), None)
        if match is None:
            return "refused step=7 reason=origin_not_expected"
    v = f"confirmed origin={origin} key_id={kid} head={hh} anchor={anc['path']}"
    if moved:
        v += " moved_to=" + moved
    if e["status"] == "retired":
        v += " retired_until=" + e["until"]
        if time is None:
            v += " untimed=true"
    if match is not None:
        v += " expected=" + match["source"]
    if any(a.startswith("energy/") and a[len("energy/"):] in HARDWARE for a in authorities):
        v += " also_delivers=" + ("true" if "delivery" in e["may_sign"] else "false")
    return v


# ------------------------------------------------------------------ §7.9 modules

def parse_grant(text):
    try:
        g = CV.parse_strict(text)
    except ValueError:
        return None
    if not exact(g, GRANT) or g["kind"] != "wai-module-grant":
        return None
    caps = g["capabilities"]
    ok = (isinstance(g["origin"], str) and origin_ok(g["origin"]) and hexb(g["module"], 32) is not None
          and g["form"] in ("decoder", "reconstruct") and isinstance(caps, list) and caps
          and all(isinstance(c, str) and CV.is_token(c) for c in caps) and len(set(caps)) == len(caps)
          and is_log_time(g["not_before"]) and is_log_time(g["not_after"]) and g["not_before"] <= g["not_after"]
          and hexb(g["signer_pubkey"], 32) is not None and isinstance(g["signer_id"], str)
          and classify(g["signer_id"])[:2] == ("origin", g["origin"])
          and hexb(g["sig"], 64) is not None)
    if not ok or not verifies(bytes.fromhex(g["signer_pubkey"]), bytes.fromhex(g["sig"]),
                              D_GRANT + CV.jcs(without_sig(g)).encode()):
        return None
    return g


def parse_status(text):
    try:
        s = CV.parse_strict(text)
    except ValueError:
        return None
    if (not exact(s, STATUS) or s["kind"] != "wai-key-status" or not isinstance(s["origin"], str)
            or hexb(s["head"], 32) is None or not CV.is_int(s["seq"]) or not is_log_time(s["issued_at"])
            or hexb(s["sig"], 64) is None):
        return None
    return s


def revoked_by(digest, revisions, anchors):
    """The origin of the first anchor whose head (or, where none can be reached, whose
    anchor revision) lists digest in revoked_modules, or None (7.9 check 5)."""
    for a in anchors:
        ah = [r for r in revisions if is_held(r, a["origin"])]
        try:
            lst = head(ah, a["revision"])[1]["revoked_modules"]
        except Stop:
            ar = next((r for r in ah if rev_hash(r) == a["revision"]), None)
            lst = ar["revoked_modules"] if ar is not None else []
        if digest in lst:
            return a["origin"]
    return None


def module(c):
    v = module_checks(c)
    if c["object"] != "parameter-set" or v.startswith("run") or "reason=module_revoked" in v:
        return v
    # SPEC 3.1 check 5: a pinned set whose digest a held anchor's head revokes is refused
    # as revoked, whichever check a grant failed first, and with no grant at all.
    o = revoked_by(module_digest(bytes.fromhex(c["module"])), held_revisions(c["revisions"]), c["anchors"])
    if o is not None:
        fb = "fallback" if c["fallback"] else "inert"
        return f"{fb} step=5 reason=module_revoked revoked_by={o} refusal=prior_revoked"
    return v


def module_checks(c):
    fb = "fallback" if c["fallback"] else "inert"
    if c["grant"] is None and c["object"] == "parameter-set":
        return f"{fb} no_grant refusal=prior_not_authorised"

    def no(step, reason, revoked_by=None):
        v = f"{fb} step={step} reason={reason}"
        if revoked_by is not None:
            v += " revoked_by=" + revoked_by
        if c["object"] == "parameter-set":
            # SPEC 3.1 check 5: a revoked digest, or any other failure of 7.9.
            v += " refusal=" + ("prior_revoked" if reason == "module_revoked" else "prior_not_authorised")
        return v

    g = parse_grant(c["grant"])
    if g is None:
        return no(1, "module_grant_invalid")
    if module_digest(bytes.fromhex(c["module"])) != g["module"]:
        return no(2, "module_digest_mismatch")
    revisions = held_revisions(c["revisions"])
    v = confirm(bytes.fromhex(g["signer_pubkey"]), g["signer_id"], ["module/" + g["form"]], g["not_before"],
                revisions, c["anchors"], c["module_origins"])
    if not v.startswith("confirmed"):
        return no(3, v.split("reason=")[1])
    fields = dict(x.split("=", 1) for x in v.split()[1:])
    origin, hh = fields["origin"], fields["head"]
    held = [r for r in revisions if is_held(r, origin)]
    hr = next(r for r in held if rev_hash(r) == hh)
    now = c["now"]
    if not is_log_time(now):
        return no(4, "module_status_stale")
    # The head and its held ancestors: a status naming anything else is not held.
    by = {rev_hash(r): r for r in held}
    line = [hh]
    while by[line[-1]]["prev"] in by and by[line[-1]]["prev"] not in line:
        line.append(by[line[-1]]["prev"])
    valid = []
    for t in c["statuses"]:
        s = parse_status(t)
        if s is None or s["origin"] != origin or s["head"] not in line:
            continue
        for r in held:
            if (rev_hash(r) == s["head"] and r["seq"] == s["seq"]
                    and verifies(bytes.fromhex(r["root"]), bytes.fromhex(s["sig"]),
                                 D_STATUS + CV.jcs(without_sig(s)).encode())):
                valid.append(s)
                break
    current = [s["issued_at"] for s in valid if s["head"] == hh]
    if not current:
        return no(4, "module_status_not_descended" if valid else "module_status_stale")
    age = secs(now) - secs(max(current))
    if age < 0 or age > hr["status_max_age_s"]:
        return no(4, "module_status_stale")
    if g["module"] in hr["revoked_modules"]:
        return no(5, "module_revoked")
    o = revoked_by(g["module"], revisions, c["anchors"])
    if o is not None:
        return no(5, "module_revoked", o)
    if now < g["not_before"]:
        return no(6, "module_grant_not_yet_valid")
    if now > g["not_after"]:
        return no(6, "module_grant_expired")
    if secs(g["not_after"]) - secs(g["not_before"]) > hr["grant_max_s"]:
        return no(7, "module_grant_too_long")
    if c["capability"] not in g["capabilities"]:
        return no(8, "module_capability_not_granted")
    return (f"run origin={origin} key_id={fields['key_id']} head={hh} anchor={fields['anchor']} "
            f"expected={fields['expected']} kill_latency_s={hr['grant_max_s'] + hr['status_max_age_s']}")


# ------------------------------------------------------------------ vectors

def held_revisions(texts):
    out = []
    for t in texts:
        try:
            out.append(parse_revision(t))
        except Stop:
            pass  # a revision that fails rule 1 is not held
    return out


def verdict(v):
    c = v["context"]
    if v["procedure"] == "confirm":
        if "receipt" in v:
            try:
                cl = CV.verify_claim(v["receipt"])
            except CV.Reject as r:
                return f"reject step={r.step} reason={r.reason}"
            pub, sid, auth, time = bytes.fromhex(cl["signer_pubkey"]), cl["signer_id"], claim_authorities(cl), cl["issued_at"]
            figure = figure_authorities(cl)
        else:
            it = c["item"]
            pub, sid, time = bytes.fromhex(it["signer_pubkey"]), it["signer_id"], it["time"]
            auth, figure = item_authorities(it["kind"], it["classes"]), None
        exp = c.get("expected_origin")
        held, expected = held_revisions(c["revisions"]), None if exp is None else [exp]
        out = confirm(pub, sid, auth, time, held, c["anchors"], expected)
        if figure is not None:
            out += "; figure " + confirm(pub, sid, figure, time, held, c["anchors"], expected)
        return out
    if v["procedure"] == "anchor":
        cur = c.get("current")
        current = (cur["anchor"], held_revisions(cur["held"])) if cur else None
        out = anchor_log(c["origin"], c["log"], c["path"], current)
        if out[0] == "anchored":
            latest = read_log(c["origin"], c["log"])[-1]
            note = " note=published_test_key" if lists_published(latest) else ""
            pinned = f" pinned_revision={out[3]}" if out[3] is not None and out[3] != out[1] else ""
            return f"anchored revision={out[1]} anchor={out[2]}{pinned}{note}"
        if out[0] == "not_anchored":
            return "not_anchored reason=" + out[1]
        return "replaced reason=key_log_replaced"
    if v["procedure"] == "module":
        return module(c)
    raise ValueError("unknown procedure")


# ------------------------------------------------------------------ canonical fixture

def seed_key(b):
    return ed25519.Ed25519PrivateKey.from_private_bytes(bytes([b]) * 32)


def pub(sk):
    return sk.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)


def fixture():
    """The revision wai-rs pins: computed here from its members and the seeds 0x61 (root),
    0x62 (committed) and 0x63 (its one key)."""
    sk = seed_key(0x61)
    r = {
        "status_max_age_s": 600, "keys": [{"until": None, "status": "active", "key": pub(seed_key(0x63)).hex(),
                                           "may_sign": ["delivery", "energy/OnChipCounter"], "id": "d1",
                                           "compromised_since": None}],
        "root": pub(sk).hex(), "seq": 0, "prev": None, "origin": "https://publisher.example",
        "next_root": commitment(pub(seed_key(0x62))).hex(), "kind": "wai-key-set",
        "issued_at": "2026-09-01T00:00:00Z", "revoked_modules": [], "grant_max_s": 3600,
    }
    r["sig"] = sk.sign(D_SET + CV.jcs(r).encode()).hex()
    return r, CV.jcs(r), rev_hash(r)


def fixture_matches_wai_rs():
    src_path = os.path.join(HERE, "..", "wai-rs", "src", "key_log", "corpus.rs")
    try:
        src = open(src_path, encoding="utf-8").read()
    except OSError as e:
        return [f"cannot read {src_path}: {e}"]
    r, canon, h = fixture()
    fails = []
    m = re.search(r'const CANONICAL: &str = "((?:[^"\\]|\\.)*)";', src)
    pinned = re.sub(r'\\(["\\])', r"\1", m.group(1)) if m else None
    if pinned != canon:
        fails.append("canonical bytes differ from wai-rs")
    m = re.search(r'const HASH: &str = "([0-9a-f]{64})";', src)
    if not m or m.group(1) != h:
        fails.append("revision hash differs from wai-rs")
    if r["sig"] not in src:
        fails.append("signature differs from wai-rs")
    return fails


def main(argv):
    if argv[1:2] == ["verdicts"] and len(argv) == 3:
        doc = json.load(open(argv[2], encoding="utf-8"))
        for v in doc["vectors"]:
            try:
                got = verdict(v)
            except (ValueError, KeyError, TypeError):
                got = "error"
            print(f"{v['name']}\t{got}")
        return 0
    if argv[1:2] == ["fixture"]:
        r, canon, h = fixture()
        print("sig          ", r["sig"])
        print("canonical    ", canon)
        print("revision_hash", h)
        return 0
    fail = 0
    pk11 = pub(seed_key(0x11))
    sid = did_key(pk11) == "did:key:z6MktULudTtAsAhRegYPiZ6631RV3viv12qd4GQF8z1xB22S"
    print(("PASS" if sid else "FAIL") + " did:key of seed 0x11 reproduces its literal")
    fail += not sid
    ff = fixture_matches_wai_rs()
    print(("PASS" if not ff else "FAIL") + " canonical fixture equals the literals pinned in wai-rs " + ",".join(ff))
    fail += bool(ff)
    path = argv[1] if len(argv) > 1 else os.path.join(HERE, "vectors.json")
    doc = json.load(open(path, encoding="utf-8"))
    for v in doc["vectors"]:
        got = verdict(v)
        ok = got == v["expected"]
        fail += not ok
        print(f"{'PASS' if ok else 'FAIL'} {v['name']}  expected={v['expected']}  got={got}")
    print(f"{len(doc['vectors'])} vectors, {fail} failure(s)")
    return 1 if fail else 0


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