Verify a proof bundle¶

# /// script
# requires-python = ">=3.10"
# dependencies = ["cryptography>=42"]
# ///
"""Verify an ISCC Monitor proof bundle without trusting the monitor.

Checks the hub's Ed25519 signature on the checkpoint, rebuilds the checkpoint's
Merkle root from the record and its RFC 6962 inclusion proof, and confirms that the
record names the bundle's ISCC-ID. When the bundle has an `anchor` member, it checks
that evidence the same way and writes the OpenTimestamps proof for `ots verify`.

    uv run verify_bundle.py MEIGKV5N6NTVF4AB.bundle.json
    uv run verify_bundle.py --resolve MEIGKV5N6NTVF4AB.bundle.json

With --resolve, the script also fetches the hub's did:web document and checks that it
lists the key that signed the checkpoint, so the key does not come from the monitor.
"""

import argparse
import base64
import hashlib
import json
import sys
import urllib.request
from pathlib import Path

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"


def b58decode(text):
    """Decode a base58btc string to bytes."""
    num = 0
    for char in text:
        num = num * 58 + B58.index(char)
    raw = num.to_bytes((num.bit_length() + 7) // 8, "big")
    return b"\x00" * (len(text) - len(text.lstrip("1"))) + raw


def ed25519_key(multibase):
    """Return the 32-byte Ed25519 public key of a did:key style multibase string."""
    raw = b58decode(multibase.removeprefix("z"))
    if not multibase.startswith("z") or raw[:2] != b"\xed\x01" or len(raw) != 34:
        raise ValueError(f"not an Ed25519 multibase key: {multibase}")
    return raw[2:]


def key_hash(name, pubkey):
    """Return the signed-note key hash of an Ed25519 key: 4 bytes of SHA-256."""
    return hashlib.sha256(name.encode() + b"\n\x01" + pubkey).digest()[:4]


def open_checkpoint(note, pubkey):
    """Verify a signed checkpoint note and return its (origin, size, root, key id)."""
    text, sep, sigs = note.partition("\n\n")
    if not sep:
        raise ValueError("checkpoint has no signature block")
    text += "\n"
    for line in sigs.splitlines():
        name, _, sig_b64 = line.removeprefix("— ").rpartition(" ")
        sig = base64.b64decode(sig_b64)
        if sig[:4] != key_hash(name, pubkey):
            continue
        try:
            Ed25519PublicKey.from_public_bytes(pubkey).verify(sig[4:], text.encode())
        except InvalidSignature:
            raise ValueError("checkpoint signature does not verify") from None
        origin, size, root = text.splitlines()[:3]
        return origin, int(size), base64.b64decode(root), sig[:4].hex()
    raise ValueError("no signature by the bundle's key")


def root_from_proof(leaf, index, size, proof):
    """Rebuild a Merkle tree root from a leaf hash and its inclusion proof (RFC 9162, 2.1.3.2)."""
    if index >= size:
        raise ValueError(f"leaf index {index} is outside a tree of size {size}")
    fn, sn, node = index, size - 1, leaf
    for sibling in proof:
        if sn == 0:
            raise ValueError("inclusion proof is too long")
        if fn % 2 == 1 or fn == sn:
            node = hashlib.sha256(b"\x01" + sibling + node).digest()
            while fn % 2 == 0 and fn != 0:
                fn, sn = fn >> 1, sn >> 1
        else:
            node = hashlib.sha256(b"\x01" + node + sibling).digest()
        fn, sn = fn >> 1, sn >> 1
    if sn != 0:
        raise ValueError("inclusion proof is too short")
    return node


def check_evidence(label, checkpoint, inclusion, key, record):
    """Check one checkpoint, its inclusion proof and the record; return the checkpoint root."""
    origin, size, root, key_id = open_checkpoint(checkpoint, ed25519_key(key["multibase"]))
    if inclusion["treeSize"] != size:
        raise ValueError(f"{label}: proof is for size {inclusion['treeSize']}, checkpoint has {size}")
    leaf = hashlib.sha256(b"\x00" + record).digest()
    proof = [base64.b64decode(h) for h in inclusion["inclusionProof"]]
    if root_from_proof(leaf, inclusion["leafIndex"], size, proof) != root:
        raise ValueError(f"{label}: inclusion proof does not rebuild the checkpoint root")
    print(f"ok  {label}: signed by {origin} key {key_id}, record at {inclusion['leafIndex']} of {size}")
    return root


def resolve_keys(domain):
    """Return the Ed25519 multibase keys listed in a hub's did:web document."""
    with urllib.request.urlopen(f"https://{domain}/.well-known/did.json", timeout=10) as resp:  # noqa: S310
        doc = json.load(resp)
    return {vm.get("publicKeyMultibase") for vm in doc.get("verificationMethod", [])}


def main():
    """Verify the bundle named on the command line and report each check."""
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("bundle", type=Path)
    parser.add_argument("--resolve", action="store_true", help="check the key against the hub's did.json")
    args = parser.parse_args()
    bundle = json.loads(args.bundle.read_text(encoding="utf-8"))

    record = base64.b64decode(bundle["record"])
    if json.loads(record)["iscc_id"] != bundle["iscc_id"]:
        raise ValueError("the record does not name the bundle's ISCC-ID")
    print(f"ok  record names {bundle['iscc_id']}")
    check_evidence("checkpoint", bundle["checkpoint"], bundle["inclusion"], bundle["key"], record)

    anchor = bundle.get("anchor")
    if anchor:
        root = check_evidence("anchor", anchor["checkpoint"], anchor["inclusion"], anchor["key"], record)
        ots = args.bundle.with_name(args.bundle.name.split(".")[0] + ".anchor.ots")
        ots.write_bytes(base64.b64decode(anchor["ots"]))
        print(f"    check the Bitcoin anchor: ots verify -d {root.hex()} {ots.name}")

    if args.resolve:
        keys = {bundle["key"]["multibase"]} | ({anchor["key"]["multibase"]} if anchor else set())
        missing = keys - resolve_keys(bundle["hub"]["domain"])
        if missing:
            raise ValueError(f"did:web:{bundle['hub']['domain']} does not list {', '.join(sorted(missing))}")
        print(f"ok  did:web:{bundle['hub']['domain']} lists the signing key")


if __name__ == "__main__":
    try:
        main()
    except (ValueError, KeyError) as exc:
        sys.exit(f"FAIL {exc}")