Receipts

import base64, hashlib
import requests, rfc8785
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

DOMAIN = "allternit.receipt.v1"

def b64url_decode(s: str) -> bytes:
    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))

def verify_chain(receipts: list[dict], jwks: dict) -> None:
    keys = {
        k["kid"]: Ed25519PublicKey.from_public_bytes(b64url_decode(k["x"]))
        for k in jwks["keys"]
        if k["kty"] == "OKP" and k["crv"] == "Ed25519"
    }
    prev = None
    for i, r in enumerate(receipts):
        body = {k: v for k, v in r.items() if k not in ("content_hash", "signature")}
        digest = "sha256:" + hashlib.sha256(rfc8785.dumps(body)).hexdigest()
        if digest != r["content_hash"]:
            raise ValueError(f"content hash mismatch at index {i}")
        if r["index"] != i or r["prev_hash"] != prev:
            raise ValueError(f"chain link broken at index {i}")

        sig = r.get("signature")
        if sig is None:
            raise ValueError(f"receipt {i} is unsigned")
        if sig["domain"] != DOMAIN:
            raise ValueError(f"unexpected signature domain at index {i}")
        signed = rfc8785.dumps({
            "schema_id": r["schema_id"],
            "schema_version": r["schema_version"],
            "domain": sig["domain"],
            "content_hash": r["content_hash"],
        })
        # raises cryptography.exceptions.InvalidSignature on failure
        keys[sig["key_id"]].verify(b64url_decode(sig["value"]), signed)
        prev = r["content_hash"]

jwks = requests.get("https://api.allternit.com/.well-known/jwks.json").json()
verify_chain(receipts, jwks)  # `receipts` = every page of GET /v1/runs/{run_id}/receipts

import { createHash, createPublicKey, verify } from "node:crypto";
import canonicalize from "canonicalize";

const DOMAIN = "allternit.receipt.v1";

function verifyChain(receipts: any[], jwks: { keys: any[] }): void {
  const keys = new Map(
    jwks.keys
      .filter((k) => k.kty === "OKP" && k.crv === "Ed25519")
      .map((k) => [k.kid, createPublicKey({ key: k, format: "jwk" })]),
  );

  let prev: string | null = null;
  receipts.forEach((r, i) => {
    const { content_hash, signature, ...body } = r;
    const digest = "sha256:" + createHash("sha256").update(canonicalize(body)!).digest("hex");
    if (digest !== content_hash) throw new Error(`content hash mismatch at index ${i}`);
    if (r.index !== i || r.prev_hash !== prev) throw new Error(`chain link broken at index ${i}`);

    if (!signature) throw new Error(`receipt ${i} is unsigned`);
    if (signature.domain !== DOMAIN) throw new Error(`unexpected signature domain at index ${i}`);
    const key = keys.get(signature.key_id);
    if (!key) throw new Error(`unknown key ${signature.key_id}`);
    const signed = canonicalize({
      schema_id: r.schema_id,
      schema_version: r.schema_version,
      domain: signature.domain,
      content_hash,
    })!;
    const ok = verify(null, Buffer.from(signed), key, Buffer.from(signature.value, "base64url"));
    if (!ok) throw new Error(`bad signature at index ${i}`);
    prev = content_hash;
  });
}

const jwks = await (await fetch("https://api.allternit.com/.well-known/jwks.json")).json();
verifyChain(receipts, jwks); // `receipts` = every page of GET /v1/runs/{run_id}/receipts