ai-contract-platform/yuqei-contract-platform/tools/verify_reliable_decision_evidence.py

#!/usr/bin/env python3
"""Offline verifier for reliable-decision-evidence/v2 JSON packages."""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import sys
from pathlib import Path

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec, ed25519, padding, rsa, utils


def canonical_hash(value: dict[str, object]) -> str:
    return hashlib.sha256(
        json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()


def key_fingerprint(public_key_pem: str) -> str:
    key = serialization.load_pem_public_key(public_key_pem.encode("ascii"))
    der = key.public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)
    return hashlib.sha256(der).hexdigest()


def verify_signature(public_key_pem: str, algorithm: str, digest: bytes, signature_b64: str) -> bool:
    try:
        key = serialization.load_pem_public_key(public_key_pem.encode("ascii"))
        signature = base64.b64decode(signature_b64, validate=True)
        if algorithm == "ED25519_SHA256" and isinstance(key, ed25519.Ed25519PublicKey):
            key.verify(signature, digest)
        elif algorithm == "RSA_PSS_SHA_256" and isinstance(key, rsa.RSAPublicKey):
            key.verify(
                signature,
                digest,
                padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=hashes.SHA256().digest_size),
                utils.Prehashed(hashes.SHA256()),
            )
        elif algorithm == "RSA_PKCS1_SHA_256" and isinstance(key, rsa.RSAPublicKey):
            key.verify(signature, digest, padding.PKCS1v15(), utils.Prehashed(hashes.SHA256()))
        elif algorithm in {"ECDSA_SHA_256", "ECC_P256_R1"} and isinstance(key, ec.EllipticCurvePublicKey):
            key.verify(signature, digest, ec.ECDSA(utils.Prehashed(hashes.SHA256())))
        else:
            return False
        return True
    except (ValueError, TypeError, InvalidSignature):
        return False


def verify(package: dict[str, object], trust_bundle: dict[str, object] | None = None) -> dict[str, object]:
    material = dict(package)
    supplied_hash = str(material.pop("package_hash", "") or "")
    material.pop("verification_token", None)
    signature = material.pop("signature", None)
    certificate = material.get("certificate")
    calculated_hash = canonical_hash(material)
    integrity_valid = bool(supplied_hash) and supplied_hash == calculated_hash
    signature_valid = False
    trust_valid = trust_bundle is None
    certificate_status = "snapshot_only" if trust_bundle is None else "untrusted"
    if isinstance(signature, dict) and isinstance(certificate, dict):
        public_key_pem = str(certificate.get("public_key_pem") or "")
        fingerprint = str(certificate.get("public_key_fingerprint") or "")
        algorithm = str(signature.get("algorithm") or "")
        fingerprint_valid = False
        try:
            fingerprint_valid = fingerprint == key_fingerprint(public_key_pem)
        except (ValueError, TypeError):
            pass
        signature_valid = bool(
            integrity_valid
            and fingerprint_valid
            and signature.get("key_id") == certificate.get("key_id")
            and signature.get("key_version") == certificate.get("key_version")
            and algorithm == certificate.get("algorithm")
            and verify_signature(public_key_pem, algorithm, bytes.fromhex(calculated_hash), str(signature.get("value") or ""))
        )
        if trust_bundle is not None:
            keys = trust_bundle.get("keys")
            trusted = next(
                (
                    item for item in keys
                    if isinstance(item, dict) and item.get("key_id") == certificate.get("key_id")
                ),
                None,
            ) if isinstance(keys, list) else None
            if isinstance(trusted, dict):
                certificate_status = str(trusted.get("status") or "unknown")
                trust_valid = bool(
                    certificate_status != "revoked"
                    and trusted.get("public_key_fingerprint") == fingerprint
                    and trusted.get("public_key_pem") == public_key_pem
                )
    valid = integrity_valid and signature_valid and trust_valid
    return {
        "valid": valid,
        "integrity_valid": integrity_valid,
        "signature_valid": signature_valid,
        "trust_valid": trust_valid,
        "certificate_status": certificate_status,
        "package_hash": calculated_hash,
        "format_version": package.get("format_version"),
        "export_id": package.get("export_id"),
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Offline verify a reliable decision evidence package.")
    parser.add_argument("package", type=Path, help="Path to reliable-decision-evidence/v2 JSON")
    parser.add_argument("--trust-bundle", type=Path, help="Optional current trust bundle JSON with revocations")
    args = parser.parse_args()
    try:
        package = json.loads(args.package.read_text(encoding="utf-8"))
        trust_bundle = json.loads(args.trust_bundle.read_text(encoding="utf-8")) if args.trust_bundle else None
        if not isinstance(package, dict) or (trust_bundle is not None and not isinstance(trust_bundle, dict)):
            raise ValueError("JSON root must be an object")
        result = verify(package, trust_bundle)
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        print(json.dumps({"valid": False, "error": str(exc)}, ensure_ascii=False))
        return 2
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if result["valid"] else 1


if __name__ == "__main__":
    sys.exit(main())