scripts/verify-clone-pq

#!/usr/bin/env python3
"""
PAX-Coder Post-Quantum Clone Integrity Verification

Upgrades verify-clone to ML-DSA-44 (CRYSTALS-Dilithium, NIST FIPS 204).

Ed25519 signatures on release manifests are vulnerable to harvest-now-
decrypt-later attacks by quantum adversaries. This script verifies the
release manifest signature using ML-DSA-44 — Shor-resistant, 128-bit
post-quantum security.

Exit codes:
  0 = Integrity verified (ML-DSA-44 signature valid)
  1 = Integrity verification failed
  2 = Script error

Requirements:
  pip install dilithium-py pydantic

Environment:
  PAX_MLDSA_PUBLIC_KEY_HEX  — authority ML-DSA-44 public key (2624 hex chars)
                               OR sovereign/mldsa_authority.pub

Authors: Ahmad Ali Parr, Jessica L. Williams (SNAPKITTYWEST)
"""

import hashlib
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Optional

# ── ML-DSA-44 constants ───────────────────────────────────────────────────────
MLDSA_PK_BYTES  = 1312
MLDSA_SIG_BYTES = 2420
MLDSA_PK_HEX    = MLDSA_PK_BYTES  * 2
MLDSA_SIG_HEX   = MLDSA_SIG_BYTES * 2

EXIT_OK     = 0
EXIT_FAILED = 1
EXIT_ERROR  = 2

# ── ML-DSA-44 verify ──────────────────────────────────────────────────────────

def mldsa_verify(pk: bytes, msg: bytes, sig: bytes) -> bool:
    try:
        from dilithium_py.dilithium import Dilithium2
        return Dilithium2.verify(pk, msg, sig)
    except ImportError:
        pass
    # Fallback: pax_verify_mldsa binary (from worm-engines Rust build)
    binary = os.environ.get("PAX_MLDSA_VERIFY_BIN", "pax_verify_mldsa")
    try:
        import tempfile
        with tempfile.TemporaryDirectory() as d:
            pk_p  = Path(d) / "pk.bin";  pk_p.write_bytes(pk)
            msg_p = Path(d) / "msg.bin"; msg_p.write_bytes(msg)
            sig_p = Path(d) / "sig.bin"; sig_p.write_bytes(sig)
            r = subprocess.run([binary, str(pk_p), str(msg_p), str(sig_p)],
                               capture_output=True, timeout=10)
            return r.returncode == 0
    except Exception:
        return False

# ── Helpers ───────────────────────────────────────────────────────────────────

def sha256_file(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()

def sha256_dir(repo_root: Path) -> str:
    """Deterministic SHA-256 of all tracked files (git ls-files order)."""
    try:
        r = subprocess.run(["git", "ls-files", "-z"],
                           capture_output=True, cwd=repo_root, timeout=30)
        files = [f for f in r.stdout.split(b"\x00") if f]
    except Exception:
        return ""
    h = hashlib.sha256()
    for f in sorted(files):
        path = repo_root / f.decode()
        if path.is_file():
            h.update(f + b"\x00")
            h.update(path.read_bytes())
    return h.hexdigest()

def git_head(repo_root: Path) -> Optional[str]:
    try:
        r = subprocess.run(["git", "rev-parse", "HEAD"],
                           capture_output=True, text=True, cwd=repo_root, timeout=10)
        return r.stdout.strip() if r.returncode == 0 else None
    except Exception:
        return None

# ── Main verification ─────────────────────────────────────────────────────────

def verify(repo_root: Path) -> int:
    sovereign = repo_root / "sovereign"

    print("========================================")
    print("PAX-CODER PQ CLONE VERIFICATION")
    print("ML-DSA-44 / NIST FIPS 204")
    print("========================================")
    print()

    # [1] Release metadata
    print("[1] Reading release metadata...")
    release_file = sovereign / "release.json"
    if not release_file.exists():
        print("  ERROR: sovereign/release.json not found")
        return EXIT_FAILED
    try:
        release = json.loads(release_file.read_text())
    except json.JSONDecodeError as e:
        print(f"  ERROR: Cannot parse release.json: {e}")
        return EXIT_ERROR
    print(f"  Repository: {release.get('repository','')}")
    print(f"  Version:    {release.get('release_version','')}")
    print(f"  Timestamp:  {release.get('release_timestamp_utc','')}")
    print("  OK")
    print()

    # [2] Git commit
    print("[2] Verifying git commit...")
    current = git_head(repo_root)
    if current is None:
        print("  ERROR: Not a git repository")
        return EXIT_ERROR
    expected = release.get("git_commit", "")
    if current != expected:
        print(f"  FAILED: Commit mismatch")
        print(f"    Expected: {expected}")
        print(f"    Current:  {current}")
        return EXIT_FAILED
    print(f"  OK: {current[:12]}")
    print()

    # [3] Manifest hash
    print("[3] Verifying manifest hash...")
    manifest_file = sovereign / "manifest.json"
    expected_hash = release.get("manifest_sha256", "")
    if manifest_file.exists() and expected_hash:
        computed = sha256_file(manifest_file)
        if computed != expected_hash:
            print(f"  FAILED: Manifest hash mismatch")
            print(f"    Expected: {expected_hash}")
            print(f"    Computed: {computed}")
            return EXIT_FAILED
        print(f"  OK: {computed[:16]}...")
    else:
        print("  SKIP: No manifest.json or no expected hash")
    print()

    # [4] File tree hash
    print("[4] Verifying tracked file tree...")
    expected_tree = release.get("tree_sha256", "")
    if expected_tree:
        computed_tree = sha256_dir(repo_root)
        if computed_tree != expected_tree:
            print("  FAILED: File tree hash mismatch — files may have been tampered")
            return EXIT_FAILED
        print(f"  OK: {computed_tree[:16]}...")
    else:
        print("  SKIP: No tree_sha256 in release.json")
    print()

    # [5] ML-DSA-44 release signature
    print("[5] Verifying ML-DSA-44 release signature...")
    sig_file = sovereign / "release.sig.hex"
    sig_hex  = os.environ.get("PAX_RELEASE_SIG_HEX", "")
    if not sig_hex and sig_file.exists():
        sig_hex = sig_file.read_text().strip()
    if not sig_hex:
        print("  SKIP: No ML-DSA-44 release signature found")
        print("        Set PAX_RELEASE_SIG_HEX or create sovereign/release.sig.hex")
        print("  (Clone integrity cannot be fully verified without signature)")
        print()
        print("========================================")
        print("STATUS: PARTIAL — commit + hash verified")
        print("        ML-DSA-44 signature not present")
        print("========================================")
        return EXIT_OK

    pk_hex = os.environ.get("PAX_MLDSA_PUBLIC_KEY_HEX", "")
    if not pk_hex:
        pk_file = sovereign / "mldsa_authority.pub"
        if pk_file.exists():
            pk_hex = pk_file.read_text().strip()
    if not pk_hex:
        print("  ERROR: No ML-DSA-44 authority public key")
        print("         Set PAX_MLDSA_PUBLIC_KEY_HEX or create sovereign/mldsa_authority.pub")
        return EXIT_ERROR

    if len(sig_hex) != MLDSA_SIG_HEX:
        print(f"  ERROR: Signature must be {MLDSA_SIG_HEX} hex chars (ML-DSA-44)")
        return EXIT_FAILED

    # Message signed: canonical JSON of release metadata
    message = json.dumps(release, separators=(",",":"), sort_keys=True).encode()

    try:
        pk  = bytes.fromhex(pk_hex)
        sig = bytes.fromhex(sig_hex)
    except ValueError:
        print("  ERROR: Non-hex characters in key or signature")
        return EXIT_ERROR

    if mldsa_verify(pk, message, sig):
        print("  OK: ML-DSA-44 signature VALID")
    else:
        print("  FAILED: ML-DSA-44 signature INVALID")
        print("          This clone may have been tampered with or is unsigned")
        return EXIT_FAILED
    print()

    print("========================================")
    print("STATUS: AUTHENTIC PAX-CODER RELEASE")
    print("        ML-DSA-44 verified (FIPS 204)")
    print("========================================")
    return EXIT_OK


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--repo-root", type=Path,
                        default=Path(__file__).resolve().parent)
    args = parser.parse_args()
    sys.exit(verify(args.repo_root))