verify.py

#!/usr/bin/env python3
"""Check a 3D model file against the ModelDirectory provenance registry.

Zero dependencies for the core check: the file is hashed locally and the
Polygon registry contract is read through any public RPC node, so you do
not have to trust modeldirectory.org (or this script's default endpoints)
to confirm a stamp exists and when it was made.

Usage:
  python3 verify.py model.stl
  python3 verify.py model.stl --id 7154ac9f-4092-4d4d-b325-2fbf5d1ff9bf
  python3 verify.py --sha256 1fdb7a90...

With --id the script also fetches the model's signed certificate and its
Bitcoin (OpenTimestamps) status from the ModelDirectory API. Verifying the
certificate signature needs the "cryptography" package (pip install
cryptography); everything else is standard library only.
"""

import argparse
import hashlib
import json
import sys
import urllib.request

# The public registry contract. Also shown on
# https://modeldirectory.org/provenance.html and returned by
# GET https://api.modeldirectory.org/api/blockchain/status
CONTRACT = "0x687F2e6F96288Ac58fe22E2Eb000D1a628aEEB65"

# First 4 bytes of keccak256("getProof(bytes32)"). The contract source is
# published at the address above on PolygonScan if you want to check.
GETPROOF_SELECTOR = "0x1b80bb3a"

# Any Polygon mainnet RPC works. Tried in order; pass --rpc to use your own.
DEFAULT_RPCS = [
    "https://polygon-bor-rpc.publicnode.com",
    "https://polygon.drpc.org",
    "https://1rpc.io/matic",
]

API_BASE = "https://api.modeldirectory.org"

# Field order matters: the server signs json.dumps of exactly these keys in
# exactly this order, no whitespace.
SIGNED_FIELDS = [
    "model_id", "registered_at", "sha256",
    "source_platform", "source_published_at", "source_url",
]


def http_json(url, payload=None, timeout=20):
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request(
        url, data=data,
        headers={"Content-Type": "application/json", "User-Agent": "md-verify/1.0"},
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.load(resp)


def sha256_file(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def check_polygon(sha256_hex, rpcs):
    """Read getProof(bytes32) straight off the chain. Returns (result, rpc_used)."""
    calldata = GETPROOF_SELECTOR + sha256_hex.lower().zfill(64)
    last_err = None
    for rpc in rpcs:
        try:
            out = http_json(rpc, {
                "jsonrpc": "2.0", "id": 1, "method": "eth_call",
                "params": [{"to": CONTRACT, "data": calldata}, "latest"],
            })
            raw = out.get("result", "")
            if not raw or len(raw) < 2 + 192:
                last_err = f"{rpc}: unexpected response {out}"
                continue
            body = raw[2:]
            timestamp = int(body[0:64], 16)
            stamper = "0x" + body[64 + 24:128]
            exists = int(body[128:192], 16) == 1
            return {"exists": exists, "timestamp": timestamp, "stamper": stamper}, rpc
        except Exception as e:
            last_err = f"{rpc}: {e}"
    raise RuntimeError(f"all RPC endpoints failed, last error: {last_err}")


def verify_certificate(cert, expected_sha256):
    """Check the Ed25519 signature on a certificate from the API.

    Returns (ok, detail). ok is None when the cryptography package is not
    installed, True/False otherwise.
    """
    if cert.get("sha256") != expected_sha256:
        return False, "certificate sha256 does not match the file"

    payload = {k: cert.get(k) for k in SIGNED_FIELDS}
    message = json.dumps(payload, separators=(",", ":")).encode()

    try:
        from cryptography.hazmat.primitives.serialization import load_der_public_key
        from cryptography.exceptions import InvalidSignature
    except ImportError:
        return None, "cryptography package not installed, signature not checked"

    import base64
    try:
        key = load_der_public_key(base64.b64decode(cert["server_public_key"]))
        key.verify(base64.b64decode(cert["server_signature"]), message)
        return True, "signature valid"
    except InvalidSignature:
        return False, "SIGNATURE INVALID"
    except Exception as e:
        return False, f"could not verify: {e}"


def main():
    ap = argparse.ArgumentParser(description="Verify a model against the ModelDirectory registry")
    ap.add_argument("file", nargs="?", help="path to the model file (STL, 3MF, ...)")
    ap.add_argument("--sha256", help="check a known hash instead of a file")
    ap.add_argument("--id", help="ModelDirectory model id, enables certificate and Bitcoin checks")
    ap.add_argument("--rpc", help="use this Polygon RPC instead of the built-in list")
    ap.add_argument("--api", default=API_BASE, help="ModelDirectory API base (default: %(default)s)")
    args = ap.parse_args()

    if not args.file and not args.sha256:
        ap.error("give a file or --sha256")

    if args.file:
        digest = sha256_file(args.file)
        print(f"file:      {args.file}")
    else:
        digest = args.sha256.lower().replace("0x", "")
    print(f"sha256:    {digest}")

    rpcs = [args.rpc] if args.rpc else DEFAULT_RPCS
    proof, rpc_used = check_polygon(digest, rpcs)
    print(f"contract:  {CONTRACT} (via {rpc_used})")

    exit_code = 0
    if proof["exists"]:
        from datetime import datetime, timezone
        when = datetime.fromtimestamp(proof["timestamp"], timezone.utc)
        print(f"polygon:   REGISTERED at {when.isoformat()} by {proof['stamper']}")
    else:
        print("polygon:   not found in the registry")
        exit_code = 2

    if args.id:
        try:
            cert = http_json(f"{args.api}/api/models/{args.id}/certificate")
            ok, detail = verify_certificate(cert, digest)
            label = {True: "VALID", False: "FAILED", None: "SKIPPED"}[ok]
            print(f"cert:      {label} ({detail})")
            if ok is False:
                exit_code = 1

            # Cross-check the key inside the certificate against the one the
            # API publishes. A mismatch would mean someone swapped keys.
            pub = http_json(f"{args.api}/api/certificates/public-key")
            if pub.get("public_key") and pub["public_key"] != cert.get("server_public_key"):
                print("cert:      WARNING: certificate key differs from the published server key")
                exit_code = 1
        except Exception as e:
            print(f"cert:      could not fetch ({e})")

        try:
            ots = http_json(f"{args.api}/api/models/{args.id}/verify-ots")
            if ots.get("confirmed"):
                print(f"bitcoin:   confirmed in block {ots['block_height']} ({ots.get('attested_time')})")
                print(f"           raw proof: {ots.get('ots_proof_url')}")
                print("           check it yourself: pip install opentimestamps-client, then 'ots verify'")
            else:
                print(f"bitcoin:   {ots.get('status', 'no timestamp')}")
        except urllib.error.HTTPError as e:
            if e.code == 404:
                print("bitcoin:   no timestamp for this model")
            else:
                print(f"bitcoin:   could not fetch ({e})")
        except Exception as e:
            print(f"bitcoin:   could not fetch ({e})")

    sys.exit(exit_code)


if __name__ == "__main__":
    main()