open-standards/wai/http-conformance/http_receipts.py

"""The receipt, key, grant and parameter-set side of http_verify.py (http-delivery §8.1-§8.5,
§10.5).

Written from the extension text. It shares no code with wai-rs. What jwp-receipts defines it
takes from the corpora's own second verifiers, each written from that text and sharing no
code with wai-rs either: group and object receipts, properties and receipt chains from
../jwp-conformance/jwp_verify.py, claims and the strict JSON reading from
../claim-conformance/claim_verify.py, and key logs, anchoring, confirmation and module
grants from ../signer-conformance/signer_verify.py. Those use the `cryptography` package for
Ed25519's equation, so these cases need it; http_verify.py loads this file only for them.
What http-delivery adds is written here: receipt files named by their link hash, the index
read item by item, the parent walk and its bound, what a client holds of the object (the
whole envelope, the cid a head proof established, or nothing), claims by subject and
parent, anchoring only from the origin's own location, logs through caches, grants files as
locators, and parameter sets held only by their pins and jwp-receipts §7.9.
"""
import hashlib, os, sys

HERE = os.path.dirname(os.path.abspath(__file__))
for _d in ("claim-conformance", "jwp-conformance", "signer-conformance", "live-conformance"):
    sys.path.insert(0, os.path.join(HERE, "..", _d))
import claim_verify as CV  # noqa: E402
import jwp_verify as JV  # noqa: E402
import signer_verify as SV  # noqa: E402

MUTATE = None  # set by http_verify.py
_LISTED = []
ORIGIN = "https://origin.example"
KEY_LOG_PATH = "/.well-known/wai-keys"
RCPT = "objects/rcpt"
WALK_BOUND = 64                     # parents fetched for one group, at most (§8.1)
MAX_RECEIPT_FILE_BYTES = 1 << 20    # a receipt file's bytes, at most (§8.1)
MAX_INDEX_BYTES = 4 << 20           # an index's bytes, at most (§8.2)
MAX_INDEX_ITEMS = 256               # delivery items an index lists, at most
MAX_INDEX_CLAIMS = 256              # claims an index lists, at most
FETCH_BUDGET = 1024                 # receipt files fetched for one index, at most
MAX_INDEX_FETCH_BYTES = 64 << 20    # bytes of receipt files read for one index, at most
MAX_KEY_LOG_BYTES = 1 << 20         # a key log's bytes, at most (§8.4)
INDEX_KIND = "wai-receipt-index"
GRANTS_KIND = "wai-module-grants"
ACQUIRED = "2026-10-06T00:00:00Z"
ONE_FILE = {"wai.neural.int_mlicpp"}  # the capabilities the corpus pins: each decodes against one file


class Refused(Exception):
    def __init__(self, code, step=None, walked=None, stop=None):
        super().__init__(code)
        self.code, self.step, self.walked, self.stop = code, step, walked, stop

    def json(self):
        out = {"error": self.code, "step": self.step}
        if self.code == "parent_not_held" and self.walked is not None:
            out["walked"], out["stop"] = self.walked, self.stop
        return out


def pnh(walked, stop):
    return Refused("parent_not_held", None, walked, stop)


class Fault(Exception):
    """A fault of the case itself: no outcome."""


def edit_text(t, edits):
    for e in edits or []:
        if e[0] not in t:
            raise Fault("an edit's text is not in the file")
        t = t.replace(e[0], e[1], 1)
    return t


def read_file(d, path):
    return open(os.path.join(d, path), "rb").read()


def utf8(b):
    try:
        return b.decode("utf-8")
    except UnicodeDecodeError:
        return None


# ------------------------------------------------------------------ receipt files (§8.1)

def read_group_file(name, b):
    """A group receipt's file: JSON as §3.1 step 1 reads it, §3 step 1, then the name.
    Returns the raw receipt and its reading."""
    text = utf8(b)
    if text is None:
        raise Refused("malformed", 1)
    try:
        raw = CV.parse_strict(text)
    except ValueError:
        raise Refused("malformed", 1)
    try:
        g = JV.read_group(raw)
    except JV.Refuse as r:
        raise Refused(r.reason, r.step)
    if not g["labelled"] and MUTATE != "legacy-admitted":
        # The legacy link hash does not cover the parent: the name does not fix the bytes.
        raise Refused("receipt_legacy_group")
    if MUTATE != "rcpt-name-unchecked" and JV.link(g).hex() != name:
        raise Refused("receipt_name_mismatch")
    return raw, g


def read_claim_file(name, b):
    """A claim's file: §3.1 steps 1-2, then the name. The verdict of steps 3-4, run later
    in order, is returned beside the claim (None: they pass)."""
    text = utf8(b)
    if text is None:
        raise Refused("malformed", 1)
    later = None
    try:
        CV.verify_claim(text)
    except CV.Reject as r:
        if r.step <= 2:
            raise Refused(r.reason, r.step)
        later = r
    c = CV.parse_strict(text)
    if CV.link_hash(c) != name:
        raise Refused("receipt_name_mismatch")
    return c, later


# ------------------------------------------------------------------ fetching (§8.1, §8.2)

class Fetcher:
    """One index check's receipt files: each name fetched once and what was served kept,
    at most a budget of fetches (never above FETCH_BUDGET) and of bytes read
    (MAX_INDEX_FETCH_BYTES), none over the file cap. Each file is read only as far as
    what remains of the bytes and of its cap plus one byte; the file that reaches the end
    of the bytes is refused, and once they are spent no further file is fetched."""

    PAST = object()

    def __init__(self, get, budget):
        self.get_raw, self.memo, self.fetched, self.bytes = get, {}, 0, 0
        self.budget = min(budget, FETCH_BUDGET)

    def get(self, h):
        if h not in self.memo or MUTATE == "memo-off":
            if self.fetched >= self.budget and MUTATE != "budget-unchecked":
                raise Refused("receipt_fetch_budget_exhausted")
            remaining = MAX_INDEX_FETCH_BYTES - self.bytes
            if remaining <= 0 and MUTATE != "bytes-unchecked":
                raise Refused("receipt_fetch_bytes_exhausted")
            self.fetched += 1
            whole = MUTATE in ("bytes-unchecked", "bytes-read-whole")
            limit = MAX_RECEIPT_FILE_BYTES + 1 if whole else min(remaining, MAX_RECEIPT_FILE_BYTES + 1)
            b = self.get_raw(h)
            if b is not None:
                b = b[:limit]
                self.bytes += len(b)
                past = self.bytes > MAX_INDEX_FETCH_BYTES if MUTATE == "bytes-read-whole" else len(b) == remaining
                if past and MUTATE != "bytes-unchecked":
                    b = Fetcher.PAST
            self.memo[h] = b
        b = self.memo[h]
        if b is None:
            raise Refused("receipt_not_served")
        if b is Fetcher.PAST:
            raise Refused("receipt_fetch_bytes_exhausted")
        if len(b) > MAX_RECEIPT_FILE_BYTES and MUTATE != "caps-unchecked":
            raise Refused("receipt_too_large")
        return b


# ------------------------------------------------------------------ the index (§8.2)

def read_property(p):
    """moq-streaming-format §5's property, as it reads: exactly its five members."""
    keys = {"content_hash", "joules_micro", "acquisition", "acquisition_evidence", "group_receipt"}
    if not (isinstance(p, dict) and set(p) == keys and CV.is_hex(p["content_hash"], 32) and JV.u64(p["joules_micro"])
            and isinstance(p["acquisition"], str) and CV.is_hex(p["acquisition_evidence"]) and CV.is_hex(p["group_receipt"], 32)):
        raise Refused("property_invalid")
    return p


def read_index(b):
    if len(b) > MAX_INDEX_BYTES and MUTATE != "caps-unchecked":
        raise Refused("receipt_index_too_large")
    bad = Refused("receipt_index_malformed")
    text = utf8(b)
    if text is None:
        raise bad
    try:
        v = CV.parse_strict(text)
    except ValueError:
        raise bad
    if not isinstance(v, dict) or set(v) != {"kind", "object", "delivery", "claims"}:
        raise bad
    if v["kind"] != INDEX_KIND or not CV.is_hex(v["object"], 32):
        raise bad
    if not isinstance(v["delivery"], list) or not isinstance(v["claims"], list):
        raise bad
    if (len(v["delivery"]) > MAX_INDEX_ITEMS or len(v["claims"]) > MAX_INDEX_CLAIMS) and MUTATE != "caps-unchecked":
        raise Refused("receipt_index_too_large")
    items = []
    for it in v["delivery"]:
        try:
            if not isinstance(it, dict) or set(it) != {"property", "object_receipt"}:
                raise Refused("receipt_index_malformed")
            p = read_property(it["property"])
            try:
                o = JV.read_object(it["object_receipt"])
            except JV.Refuse as r:
                raise Refused(r.reason, r.step)
            items.append((p, o, it["object_receipt"]))
        except Refused as r:
            items.append(r)
    claims = [c if CV.is_hex(c, 32) else Refused("receipt_index_malformed") for c in v["claims"]]
    return v["object"], items, claims


# ------------------------------------------------------------------ chains and the walk (§8.1)

class Chains:
    """The client's chains (jwp-receipts §3 step 3): every group offered so far, in order,
    run through jwp_verify.chain again for each offer; the last verdict is the offer's."""

    def __init__(self, join):
        self.seq, self.join = [], join

    def admit(self, raw):
        self.seq.append(raw)
        out = JV.chain({"groups": self.seq, "guard": False, "recovery": {"join": True} if self.join else {}})
        last = out[len("chain: "):].split(" / ")[-1]
        if last.startswith("reject"):
            f = dict(x.split("=", 1) for x in last.split()[1:])
            raise Refused(f["reason"], int(f["step"]) if "step" in f else None)
        return last


def admit_walking(ch, raw, g, fetcher, bound):
    try:
        return ch.admit(raw), 0
    except Refused as r:
        if r.code != "parent_not_held":
            raise
        if MUTATE == "walk-skipped":
            return "accept", 0
    if MUTATE != "walk-unclamped":
        bound = min(bound, WALK_BOUND)
    pending, link, fetched = [], g["parent_receipt_hash"], 0
    while True:
        if link is None:
            raise pnh(fetched, "parent_not_held")
        if fetched == bound:
            raise pnh(fetched, "bound")
        try:
            b = fetcher.get(link)
        except Refused as r:
            raise pnh(fetched + 1 if r.code == "receipt_too_large" else fetched, r.code)
        fetched += 1
        try:
            praw, p = read_group_file(link, b)
        except Refused as r:
            raise pnh(fetched, r.code)
        if p["signer_pubkey"] != g["signer_pubkey"] or p["track_alias"] != g["track_alias"]:
            raise pnh(fetched, "another_chain")
        try:
            ch.admit(praw)
            break
        except Refused as r:
            if r.code != "parent_not_held":
                raise pnh(fetched, r.code)
            link = p["parent_receipt_hash"]
            pending.append(praw)
    for praw in reversed(pending):
        try:
            ch.admit(praw)
        except Refused as r:
            raise pnh(fetched, r.code)
    return ch.admit(raw), fetched


# ------------------------------------------------------------------ delivery items (§8.2)

def reading(p, o, g):
    """The property against an object accepted under g (moq-streaming-format §5)."""
    try:
        c = JV.property_class(p)
    except JV.Refuse as r:
        raise Refused(r.reason)
    if bytes.fromhex(p["content_hash"]) != o["content_hash"] or bytes.fromhex(p["group_receipt"]) != JV.link(g):
        raise Refused("property_disagrees")
    if not o["labelled"]:
        return "unmetered" if c is None else "unsigned"
    if o["class"] is None and c is None:
        return "unmetered"
    if o["class"] is not None and c is not None and o["class"] == c and o["joules_micro"] == p["joules_micro"]:
        return f"signed class={c['label']} by={g['signer_pubkey']}"
    raise Refused("property_disagrees")


def check_delivery(cid, item, held, fetcher, ch, bound):
    p, o, raw_o = item
    if p["content_hash"] != cid or o["content_hash"].hex() != cid:
        raise Refused("receipt_not_about_object")
    name = p["group_receipt"]
    b = fetcher.get(name)
    raw, g = read_group_file(name, b)
    admission, walked = admit_walking(ch, raw, g, fetcher, bound)
    if MUTATE == "nothing-as-delivered" and held is None:
        held = (cid, "head-proof")
    if held is not None and held[0] != o["content_hash"].hex():
        raise Refused("content_hash_mismatch", 4)
    try:
        JV.inclusion(o, g)
    except JV.Refuse as r:
        raise Refused(r.reason, r.step)
    if held is not None:
        rd = reading(p, o, g)
    else:
        try:
            JV.property_class(p)
        except JV.Refuse as r:
            raise Refused(r.reason)
        rd = None
    out = {"admission": admission, "walked": walked, "evidence": "delivered-verified" if held is not None else "inclusion-verified",
           "established": held[1] if held is not None else None, "reading": rd}
    return out, {"g": g, "o": o, "raw": raw_o, "held": held}


# ------------------------------------------------------------------ claims (§8.2)

def declaration_text(link, fetcher):
    """A declaration a claim names, fetched by its link and read as §8.1 reads every file
    fetched by name (steps 1-2, then the name): not served is context_not_held; its steps
    3-4 are step 6's, which reports a declaration that does not verify (context_invalid)."""
    try:
        b = fetcher.get(link)
    except Refused as r:
        if r.code == "receipt_not_served":
            raise Refused("context_not_held", 6)
        raise
    if MUTATE != "declaration-unnamed":
        read_claim_file(link, b)
    t = utf8(b)
    return t if t is not None else "\u0000"  # not UTF-8: no claim reads from it (context_invalid)


def step6(c, cid, passed, fetcher, exact_algs):
    """§3.1 step 6 over HTTP (§8.2): against the delivery items that passed and the
    declarations the claim names."""
    if MUTATE == "step6-skipped":
        return
    cls, b = c["class"], c["body"]
    delivered = [v for v in passed if v["held"] is not None]
    try:
        if cls == "energy":
            for v in delivered:
                CV.check_context(c, {"object": v["raw"]})
            if not delivered or b["oer_receipt"] is not None:
                raise Refused("context_not_held", 6)
        elif cls in ("measurement", "operating-point") and not (cls == "operating-point" and MUTATE == "operating-point-unchecked"):
            v = next((v for v in delivered if JV.link(v["g"]).hex() == b["group"]), None)
            if v is None:
                raise Refused("figure_not_held" if cls == "measurement" else "context_not_held", 6)
            fig = {k: v["raw"][k] for k in JV.CLASS_KEYS if k in v["raw"]}
            fig.update(index=v["o"]["index"], joules_micro=v["o"]["joules_micro"])
            view = {"profile": "moq-jwp:object", "identity": cid, "group": JV.link(v["g"]).hex(), "figures": [fig],
                    "signer": v["g"]["signer_pubkey"]}
            CV.check_figure(c, CV.figure_view({"view": view}))
        elif cls == "delivery-gap":
            at = (b["track_alias"], b["group_id"], b["object_id"])
            v = next((v for v in passed if (v["o"]["track_alias"], v["o"]["group_id"], v["o"]["object_id"]) == at), None)
            if v is None:
                raise Refused("context_not_held", 6)
            CV.check_context(c, {"gap_object": {"track_alias": v["o"]["track_alias"], "group_id": v["o"]["group_id"],
                                                "object_id": v["o"]["object_id"], "content_hash": v["o"]["content_hash"].hex()}})
        elif cls == "lawful-basis" and b["record"] == "decision" and MUTATE != "decision-unchecked":
            CV.check_context(c, {"declaration": declaration_text(bytes.fromhex(b["declaration"]).hex(), fetcher)})
        elif cls == "binding-recovery" and MUTATE != "recovery-record-unchecked":
            if not passed:
                raise Refused("context_not_held", 6)
            t = declaration_text(b["declaration"], fetcher)
            signers = [v["g"]["signer_pubkey"] for v in passed]
            try:
                d = CV.parse_strict(t)
                ds = d.get("signer_pubkey") if isinstance(d, dict) else None
            except ValueError:
                ds = None
            pub = ds if ds in signers else signers[0]
            CV.check_context(c, {"declaration": t, "publisher": pub, "exact_match_algs": exact_algs})
        elif cls == "reconstruct-session" and MUTATE != "session-accepted":
            raise Refused("context_not_held", 6)
        elif cls in ("transmission", "reconciliation"):
            # bearer-receipts §4.3, §5.3: against the group of an item its subject names, and a
            # reconciliation also against the object receipts of the items listed under that
            # group here; one naming an object this index does not list is context_not_held.
            import bearer_verify as BV
            v = next((v for v in passed if JV.link(v["g"]).hex() == c["subject"]), None)
            if v is None:
                raise Refused("context_not_held", 6)
            if cls == "transmission":
                BV.check_transmission(c, v["g"])
            else:
                objs = [w["o"] for w in passed if JV.link(w["g"]).hex() == c["subject"]]
                try:
                    BV.check_reconciliation(c, v["g"], objs)
                except CV.Reject as r:
                    if r.reason == "reconciliation_objects_not_held":
                        raise Refused("context_not_held", 6)
                    raise
        elif cls == "schedule":
            # carousel §4.3: against the group of an item its subject names and the object
            # receipts of the items listed under it here, with no envelope bytes.
            import carousel_verify as KV
            v = next((v for v in passed if JV.link(v["g"]).hex() == c["subject"]), None)
            if v is None:
                raise Refused("context_not_held", 6)
            KV.check_schedule(c, v["g"], [(w["o"], None) for w in passed if JV.link(w["g"]).hex() == c["subject"]])
    except CV.Reject as r:
        raise Refused(r.reason, r.step)


def check_claim(cid, passed, held, name, fetcher, recovery):
    b = fetcher.get(name)
    c, later = read_claim_file(name, b)
    groups = [JV.link(v["g"]).hex() for v in passed]
    if MUTATE == "groups-from-listed":
        groups += _LISTED
    # §3.2's P: a recovery through a soft binding needs the object in hand; of the keys
    # held for the subject, the declaration's own signer's where held, else the first.
    pubs = (recovery or {}).get("publishers") or [] if held is not None else []
    keys = [k for h, k in pubs if h == c["subject"]]
    publisher = None
    if keys:
        publisher = keys[0] if MUTATE == "first-publisher" else next((k for k in keys if k == c["signer_pubkey"]), keys[0])
    if c["subject_type"] == "object" and c["subject"] == cid:
        attests = "object"
    elif c["subject_type"] == "receipt" and c["subject"] in groups:
        attests = "receipt"
    elif c["subject_type"] == "object" and c["class"] == "soft-binding" and (publisher is not None or MUTATE == "ancestor-any"):
        attests = "ancestor"
    else:
        raise Refused("receipt_not_about_object")
    if later is not None:
        raise Refused(later.reason, later.step)
    parent = c["parent_receipt_hash"]
    if parent is not None:
        try:
            pb = fetcher.get(parent)
        except Refused as r:
            raise pnh(1 if r.code == "receipt_too_large" else 0, r.code)
        try:
            prev, plater = read_claim_file(parent, pb)
        except Refused as r:
            raise pnh(1, r.code)
        if plater is not None:
            raise pnh(1, plater.reason)
        try:
            CV.check_chain(c, prev)
        except CV.Reject as r:
            raise Refused(r.reason, r.step)
    algs = (recovery or {}).get("exact_match_algs") or []
    if attests == "ancestor":
        if MUTATE != "ancestor-any":
            try:
                CV.recover({"declaration": b.decode("utf-8"), "publisher": publisher, "observed_hash": cid,
                            "observed_value": recovery["observed_value"], "exact_match_algs": algs})
            except CV.Reject as r:
                raise Refused(r.reason, r.step)
    else:
        step6(c, cid, passed, fetcher, algs)
    return {"class": c["class"], "attests": attests, "subject": c["subject"], "parent": parent}, c


# ------------------------------------------------------------------ keys (§8.4)

class KeyState:
    def __init__(self):
        self.anchor, self.held = None, []

    def add(self, revs):
        have = {SV.rev_hash(r) for r in self.held}
        for r in revs:
            if SV.rev_hash(r) not in have:
                have.add(SV.rev_hash(r))
                self.held.append(r)

    def step(self, d, s):
        """What the step did, or its refusal ({"error": …}); a step refused changes nothing."""
        origin = s.get("origin", ORIGIN)
        f = s.get("anchor") or s.get("cached") or s.get("pin")
        body = edit_text(read_file(d, f).decode("utf-8"), s.get("edit"))
        if s.get("pad") is not None:
            body = body + " " * max(0, s["pad"] - len(body.encode("utf-8")))
        if s.get("pin") is not None:
            out = SV.anchor_log(origin, body, "pinned", None)
            if out[0] != "anchored":
                raise Fault("a pinned log that does not anchor")
            self.held = SV.read_log(origin, body)
            self.anchor = {"origin": origin, "revision": out[1], "path": "pinned", "pinned_revision": out[3]}
            return {"anchored": out[1], "path": "pinned"}
        if s.get("cached") is not None:
            if self.anchor is None:
                raise Fault("a cached log before any anchor")
            return self.cached(origin, body)
        loc = origin + KEY_LOG_PATH
        r = s.get("retrieval") or {}
        requested = r.get("requested", loc)
        answered = r.get("answered", requested)
        status = r.get("status", 200)
        if not origin.startswith("https://") or requested != loc or (answered != loc and MUTATE != "anchor-follows-redirect") or status != 200:
            return {"error": "key_log_not_from_origin"}
        if len(body.encode("utf-8")) > MAX_KEY_LOG_BYTES and MUTATE != "key-log-uncapped":
            return {"error": "key_log_too_large"}
        current = (self.anchor, self.held) if self.anchor is not None else None
        out = SV.anchor_log(origin, body, "fetched", current)
        if out[0] == "not_anchored":
            return {"error": out[1]}
        if out[0] == "replaced":
            return {"error": "key_log_replaced"}
        try:
            self.add(SV.read_log(origin, body))
        except SV.Stop:
            pass
        self.anchor = {"origin": origin, "revision": out[1], "path": out[2], "pinned_revision": out[3]}
        return {"anchored": out[1], "path": out[2]}

    def cached(self, origin, body):
        if self.anchor["origin"] != origin:
            return {"error": "origin_not_anchored"}
        if len(body.encode("utf-8")) > MAX_KEY_LOG_BYTES and MUTATE != "key-log-uncapped":
            return {"error": "key_log_too_large"}
        try:
            revs = SV.read_log(origin, body)
        except SV.Stop as e:
            return {"error": e.reason}
        by = {SV.rev_hash(r): r for r in self.held if SV.is_held(r, origin)}
        trusted = [self.anchor["revision"]]
        while trusted[-1] in by and by[trusted[-1]]["prev"] in by and by[trusted[-1]]["prev"] not in trusted:
            trusted.append(by[trusted[-1]]["prev"])
        latest = SV.rev_hash(revs[-1])
        if latest != self.anchor["revision"] and latest in trusted and MUTATE != "cached-rollback-unchecked":
            return {"error": "key_log_rolled_back"}
        before = len(self.held)
        self.add(revs)
        return {"added": len(self.held) - before}


def key_material(d, steps):
    st = KeyState()
    for s in steps or []:
        r = st.step(d, s)
        if "error" in r:
            raise Fault("the case's key material does not anchor")
    return st.held, ([st.anchor] if st.anchor is not None else [])


def read_grants_file(module, b):
    bad = Refused("grants_file_malformed")
    text = utf8(b)
    if text is None:
        raise bad
    try:
        v = CV.parse_strict(text)
    except ValueError:
        raise bad
    if not isinstance(v, dict) or set(v) != {"kind", "module", "grants"} or v["kind"] != GRANTS_KIND:
        raise bad
    if not (CV.is_hex(v["module"], 32) and (v["module"] == module or MUTATE == "grants-module-unchecked")):
        raise bad
    if not isinstance(v["grants"], list):
        raise bad
    out = []
    for g in v["grants"]:
        try:
            out.append(CV.jcs(g))
        except ValueError:
            out.append("null")  # outside the canonical subset: no grant reads from it
    return out


# ------------------------------------------------------------------ parameter sets (§8.5)

def module_check_digest(digest, grant_text, cap, origins, revisions, anchors, statuses, now):
    """jwp-receipts §7.9's checks with the pin's digest standing for the bytes' (check 2):
    ("run",) or ("not_held", step, reason, revoked_by)."""
    def no(step, reason, by=None):
        return ("not_held", step, reason, by)
    g = SV.parse_grant(grant_text)
    if g is None:
        return no("1", "module_grant_invalid")
    if digest != g["module"]:
        return no("2", "module_digest_mismatch")
    v = SV.confirm(bytes.fromhex(g["signer_pubkey"]), g["signer_id"], ["module/" + g["form"]], g["not_before"], revisions, anchors, 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 SV.is_held(r, origin)]
    hr = next(r for r in held if SV.rev_hash(r) == hh)
    if now is None or not SV.is_log_time(now):
        return no("4", "module_status_stale")
    by = {SV.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 statuses:
        s = SV.parse_status(t)
        if s is None or s["origin"] != origin or s["head"] not in line:
            continue
        for r in held:
            if (SV.rev_hash(r) == s["head"] and r["seq"] == s["seq"]
                    and SV.verifies(bytes.fromhex(r["root"]), bytes.fromhex(s["sig"]), SV.D_STATUS + CV.jcs(SV.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 = SV.secs(now) - SV.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 = SV.revoked_by(g["module"], revisions, 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 SV.secs(g["not_after"]) - SV.secs(g["not_before"]) > hr["grant_max_s"]:
        return no("7", "module_grant_too_long")
    if cap not in g["capabilities"]:
        return no("8", "module_capability_not_granted")
    return ("run",)


def check_set(cap, digest, revisions, anchors, statuses, grants, hold_to, now):
    """SPEC §3.1 check 5 at a sink that holds a key log: None where the set is held or
    the check does not apply, else the refusal."""
    revoked = SV.revoked_by(digest, revisions, anchors)
    if hold_to is None:
        return None if revoked is None else {"error": "prior_revoked", "revoked_by": revoked}
    if MUTATE == "set-grants-ignored":
        return None
    origins = [{"origin": o, "source": "relying-party"} for o in hold_to]

    def rank(o):
        return 255 if o[0] == "run" or o[2] == "module_revoked" else int(o[1])
    reported = None
    for g in grants:
        pg = SV.parse_grant(g)
        if pg is not None and pg["form"] != "decoder":
            continue
        out = module_check_digest(digest, g, cap, origins, revisions, anchors, statuses, now)
        if out[0] == "run":
            return None
        if reported is None or rank(out) > rank(reported):
            reported = out
    if reported is not None and reported[2] == "module_revoked":
        return {"error": "prior_revoked", "revoked_by": reported[3]}
    if revoked is not None:
        return {"error": "prior_revoked", "revoked_by": revoked}
    if reported is None:
        return {"error": "prior_not_authorised", "check": None, "reason": None}
    return {"error": "prior_not_authorised", "check": reported[1], "reason": reported[2]}


def param_set(d, inp):
    cap, sha = inp["capability"], inp["sha256"]
    if not CV.is_hex(sha, 32):
        return {"error": "pin_malformed"}
    if cap not in ONE_FILE:
        raise Fault("a capability whose set shape this verifier does not read")
    served = read_file(d, inp["served"])
    revisions, anchors = key_material(d, inp.get("keys"))
    statuses = [read_file(d, p).decode("utf-8") for p in inp.get("statuses") or []]
    grants = []
    if inp.get("grants") is not None:
        try:
            grants = read_grants_file(sha, read_file(d, inp["grants"]))
        except Refused:
            raise Fault("the case's grants file does not read")
    r = check_set(cap, sha, revisions, anchors, statuses, grants, inp.get("hold_to"), inp.get("now"))
    if r is not None:
        return r
    if hashlib.sha256(served).hexdigest() != sha and MUTATE != "set-by-name":
        return {"error": "prior_not_held"}
    return {"held": True}


# ------------------------------------------------------------------ confirmation (§7.4)

def confirm_group(g, o, revisions, anchors):
    classes = []
    if g["labelled"] and g["cls"] is not None:
        classes.append(g["cls"]["label"])
    if o["labelled"] and o["class"] is not None:
        classes.append(o["class"]["label"])
    return SV.confirm(bytes.fromhex(g["signer_pubkey"]), g["signer_id"], SV.item_authorities("group", classes), g["group_end"],
                      revisions, anchors, None)


def confirm_claim(c, revisions, anchors):
    return SV.confirm(bytes.fromhex(c["signer_pubkey"]), c["signer_id"], SV.claim_authorities(c), c["issued_at"], revisions, anchors, None)


# ------------------------------------------------------------------ the ops

class Served:
    """A case's receipt files: the `rcpt` directories searched for <hash>.json, hashes
    `omit`ted, hashes served `as` another file, and files `pad`ded with spaces to a
    length, or every file to `pad_all`."""

    def __init__(self, d, v):
        self.d = d
        self.dirs = v.get("rcpt") or [RCPT]
        self.omit = set(v.get("omit") or [])
        self.swap = v.get("as") or {}
        self.pad = v.get("pad") or {}
        self.pad_all = v.get("pad_all")

    def get(self, h):
        if h in self.omit:
            return None
        b = None
        if h in self.swap:
            p = os.path.join(self.d, self.swap[h])
            b = open(p, "rb").read() if os.path.exists(p) else None
        else:
            for dd in self.dirs:
                p = os.path.join(self.d, dd, h + ".json")
                if os.path.exists(p):
                    b = open(p, "rb").read()
                    break
        if b is not None and h in self.pad:
            b = b + b" " * max(0, self.pad[h] - len(b))
        elif b is not None and self.pad_all is not None:
            b = b + b" " * max(0, self.pad_all - len(b))
        return b


def receipt_file(d, inp):
    path = inp["file"]
    t = edit_text(read_file(d, path).decode("utf-8"), inp.get("edit"))
    name = inp.get("name") or path.rsplit("/", 1)[-1][:-len(".json")]
    try:
        if inp["kind"] == "group":
            _, g = read_group_file(name, t.encode("utf-8"))
            return {"link": JV.link(g).hex()}
        c, _ = read_claim_file(name, t.encode("utf-8"))
        return {"link": CV.link_hash(c)}
    except Refused as r:
        return r.json()


def receipts(d, objs, inp, fetch_object, content_hash):
    served = Served(d, inp.get("serve") or {})
    ch = Chains(inp.get("join") is True)
    bound = WALK_BOUND if inp.get("max_walk") is None else inp["max_walk"]
    budget = FETCH_BUDGET if inp.get("fetch_budget") is None else inp["fetch_budget"]
    keys = key_material(d, inp["keys"]) if inp.get("keys") is not None else None
    rec_in = inp.get("recovery")
    publishers = []
    out = []
    for chk in inp["checks"]:
        if "object_cid" in chk:
            b, cid = None, chk["object_cid"]
        else:
            b = read_file(d, chk["object_file"]) if "object_file" in chk else objs[chk["object"]]
            cid = content_hash(b).hex()
        ixb = edit_text(read_file(d, chk.get("index") or f"objects/{cid}.wai.receipts").decode("utf-8"), chk.get("edit_index")).encode("utf-8")
        if "pad_index" in chk:
            ixb = ixb + b" " * max(0, chk["pad_index"] - len(ixb))
        fetched = None
        h = chk["held"]
        if h == "none":
            held = None
        elif h == "whole":
            hb = objs[chk["held_object"]] if "held_object" in chk else b
            held = (content_hash(hb).hex(), "envelope")
        else:
            held, fetched = fetch_object(b, h["fetch"])
        try:
            obj, items, claims = read_index(ixb)
            if obj != cid:
                raise Refused("receipt_not_about_object")
        except Refused as r:
            entry = {"index": r.code}
            if fetched is not None:
                entry["fetched"] = fetched
            out.append(entry)
            continue
        fetcher = Fetcher(served.get, budget)
        dl, passed = [], []
        global _LISTED  # what the mutation groups-from-listed attaches claims through
        _LISTED = [it[0]["group_receipt"] for it in items if not isinstance(it, Refused)]
        for it in items:
            if isinstance(it, Refused):
                dl.append(it.json())
                continue
            try:
                v, ctx = check_delivery(cid, it, held, fetcher, ch, bound)
                if keys is not None:
                    v["confirm"] = confirm_group(ctx["g"], it[1], *keys)
                dl.append(v)
                passed.append(ctx)
            except Refused as r:
                dl.append(r.json())
        # A recovery through a soft binding runs only for an object the client holds
        # (check_claim gates it); the matchers serve a recovery record's step 6 either way.
        recovery = None
        if rec_in is not None and (held is not None or MUTATE != "matchers-need-object"):
            recovery = {"publishers": publishers, "observed_value": rec_in["observed_value"],
                        "exact_match_algs": rec_in.get("exact_match_algs") or []}
        cl = []
        for c in claims:
            if isinstance(c, Refused):
                cl.append(c.json())
                continue
            try:
                v, claim = check_claim(cid, passed, held, c, fetcher, recovery)
                if keys is not None:
                    v["confirm"] = confirm_claim(claim, *keys)
                cl.append(v)
            except Refused as r:
                cl.append(r.json())
        entry = {"index": "ok", "delivery": dl, "claims": cl, "files_fetched": fetcher.fetched}
        if fetched is not None:
            entry["fetched"] = fetched
        out.append(entry)
        # The delivery receipts this check verified over the object's bytes: publisher keys
        # a later check may recover through.
        publishers += [(cid, v["g"]["signer_pubkey"]) for v in passed if v["held"] is not None]
    return {"checks": out}


def key_log(d, inp):
    st = KeyState()
    return {"steps": [st.step(d, s) for s in inp["steps"]]}


def grants_file(d, inp):
    path = inp["file"]
    t = edit_text(read_file(d, path).decode("utf-8"), inp.get("edit"))
    module = inp.get("module") or path.rsplit("/", 1)[-1][:-len(".json")]
    try:
        gs = read_grants_file(module, t.encode("utf-8"))
    except Refused as r:
        return {"error": r.code}
    return {"grants": ["read" if SV.parse_grant(g) is not None else "module_grant_invalid" for g in gs]}


def evaluate(d, objs, op, inp, fetch_object, content_hash):
    if op == "receipt_file":
        return receipt_file(d, inp)
    if op == "receipts":
        return receipts(d, objs, inp, fetch_object, content_hash)
    if op == "key_log":
        return key_log(d, inp)
    if op == "grants_file":
        return grants_file(d, inp)
    if op == "param_set":
        return param_set(d, inp)
    raise Fault("unknown op " + op)