open-standards/wai/examples/verify-receipts/python/verify_receipts.py

#!/usr/bin/env python3
"""Verify WAI delivery receipts in a directory, with nothing but Python's standard library.

    python3 verify_receipts.py <dir>

For each receipt index `<cid>.wai.receipts` in <dir> (laid out as http-delivery §8 lays them
out beside an object, as `wai hello` writes them), each delivery item is checked:

1. the group receipt is read from `rcpt/<link hash>.json`, and its link hash, recomputed, is
   its name;
2. its Ed25519 signature verifies over its signed payload (jwp-receipts §2.3, the labelled
   form), under the strict rule every WAI verifier holds;
3. the object `<cid>.wai` is there and hashes to the receipt's content hash, which is its
   name, and the object's leaf reaches the group's Merkle root by its proof (§3 steps 4, 5).

This example reads groups whose objects are explicitly unmetered (energy not measured), as
`wai hello` writes; a receipt with a measured energy class is reported, not read. It is a
second implementation: it shares no code with the Rust one in ../rust/. Exit 0 when every
item of every index verifies.
"""
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _blake3 import blake3  # noqa: E402
from _ed25519 import verify as ed25519_verify  # noqa: E402

LEAF = b"moq-jwp:merkle-leaf\x02"
NODE = b"moq-jwp:merkle-node\x01"
SIGN = b"moq-jwp:group-sign\x02"
LINK = b"moq-jwp:receipt-link\x01"
OBJECT = b"moq-jwp:object\x01"
UNMETERED = b"\x00"  # the signed encoding of "not measured" (energy-measurement §2.7)


def u32(n):
    return n.to_bytes(4, "big")


def u64(n):
    return n.to_bytes(8, "big")


def text(s):
    b = s.encode()
    return u32(len(b)) + b


def signed_payload(g):
    """The labelled group payload (jwp-receipts §2.3)."""
    parent = g["parent_receipt_hash"]
    return (SIGN + u64(g["track_alias"]) + u64(g["group_id"]) + bytes.fromhex(g["root_hash"])
            + u64(g["leaf_count"]) + u64(g["joules_total"]) + text(g["group_start"]) + text(g["group_end"])
            + bytes.fromhex(g["signer_pubkey"]) + text(g["signer_id"])
            + UNMETERED + b"\x00"  # the class, and its coverage: whole
            + ((b"\x01" + bytes.fromhex(parent)) if parent else b"\x00" + bytes(32)))


def check_item(d, cid, item):
    """One delivery item; raises ValueError with the reason when it does not verify."""
    o = item["object_receipt"]
    gname = item["property"]["group_receipt"]
    with open(os.path.join(d, "rcpt", gname + ".json"), "rb") as f:
        g = json.load(f)
    if g.get("kind") != "group" or g.get("sign_version") != 2:
        raise ValueError("not a group receipt in the labelled form")
    if any(k.startswith("energy_") and k != "energy_coverage" for k in g) or o.get("leaf_version") != 2 or any(k.startswith("energy_") for k in o):
        raise ValueError("a measured energy class: this example reads unmetered receipts alone")
    payload = signed_payload(g)
    if blake3(LINK + payload + bytes.fromhex(g["sig"])).hex() != gname:
        raise ValueError("receipt_name_mismatch: the group's link hash is not its file's name")
    why = ed25519_verify(bytes.fromhex(g["signer_pubkey"]), payload, bytes.fromhex(g["sig"]))
    if why is not None:
        raise ValueError(f"{why}: the group's signature")
    with open(os.path.join(d, cid + ".wai"), "rb") as f:
        envelope = f.read()
    content = blake3(OBJECT + envelope)
    if content.hex() != o["content_hash"] or content.hex() != cid:
        raise ValueError("content_hash_mismatch: the object is not the one the receipt names")
    # The object's leaf, up its proof to the root.
    p = o["merkle_proof"]
    if o["object_id"] != p["leaf_index"] or p["direction_bits"] != p["leaf_index"] or o["group_id"] != g["group_id"]:
        raise ValueError("proof_position_mismatch")
    if o["joules_micro"] != 0:
        raise ValueError("object_label_invalid: an unmetered object states no energy")
    cur = blake3(LEAF + content + u64(0) + UNMETERED + text(o["origin"]))
    for i, s in enumerate(p["siblings"]):
        s = bytes.fromhex(s)
        cur = blake3(NODE + s + cur) if (p["direction_bits"] >> i) & 1 else blake3(NODE + cur + s)
    if cur.hex() != g["root_hash"]:
        raise ValueError("inclusion_proof_failed")
    return g


def main(d):
    names = sorted(n for n in os.listdir(d) if n.endswith(".wai.receipts"))
    if not names:
        print(f"no receipt index (*.wai.receipts) in {d}")
        return 2
    ok = True
    for n in names:
        cid = n[: -len(".wai.receipts")]
        with open(os.path.join(d, n), "rb") as f:
            index = json.load(f)
        if index.get("kind") != "wai-receipt-index" or index.get("object") != cid:
            print(f"FAIL {n}: not this object's receipt index")
            ok = False
            continue
        for item in index["delivery"]:
            try:
                g = check_item(d, cid, item)
                print(f"ok   {cid[:16]}: delivered and verified; group {item['property']['group_receipt'][:16]} signed by {g['signer_pubkey'][:16]}… ({g['signer_id'][:24]}…)")
            except (ValueError, OSError, KeyError) as e:
                print(f"FAIL {cid[:16]}: {e}")
                ok = False
    return 0 if ok else 1


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(__doc__.split("\n\n")[1], file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv[1]))