scripts/verify_release.py

#!/usr/bin/env python3
"""Offline integrity and privacy-boundary verifier for the QC67 Cosmos kit."""
from __future__ import annotations

import argparse
import hashlib
import json
import sys
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parent
MANIFEST = ROOT / "RELEASE_MANIFEST.json"


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def check_manifest(strict: bool) -> list[str]:
    errors: list[str] = []
    release = json.loads(MANIFEST.read_text(encoding="utf-8"))
    expected = set()
    for entry in release.get("files", []):
        rel = entry["path"]
        expected.add(rel)
        path = ROOT / rel
        if not path.is_file():
            errors.append(f"missing: {rel}")
            continue
        size = path.stat().st_size
        if size != int(entry["bytes"]):
            errors.append(f"size mismatch: {rel} ({size} != {entry['bytes']})")
            continue
        actual = sha256(path)
        if actual != entry["sha256"]:
            errors.append(f"hash mismatch: {rel}")

    if strict:
        ignored = {"RELEASE_MANIFEST.json"}
        actual = {
            path.relative_to(ROOT).as_posix()
            for path in ROOT.rglob("*")
            if path.is_file()
            and path.relative_to(ROOT).as_posix() not in ignored
            and not path.relative_to(ROOT).as_posix().startswith("downloads/")
        }
        for rel in sorted(actual - expected):
            errors.append(f"unmanifested file: {rel}")
    return errors


def check_blank_credentials() -> list[str]:
    errors: list[str] = []
    config = json.loads(
        (ROOT / "genesis_engine" / "config.json").read_text(encoding="utf-8")
    )
    for key in ("ibm_token", "azure_connection_string"):
        if str(config.get(key) or "").strip():
            errors.append(f"credential field is not blank: genesis_engine/config.json:{key}")
    forbidden_names = ("oauth2_tokens.json", ".env", "credentials.json")
    for path in ROOT.rglob("*"):
        if path.is_file() and path.name.casefold() in forbidden_names:
            errors.append(f"forbidden credential file present: {path.relative_to(ROOT)}")
    return errors


def check_public_archive() -> tuple[list[str], dict]:
    errors: list[str] = []
    archive = ROOT / "data" / "quantum_measurements_public.jsonl"
    data_manifest = json.loads(
        (ROOT / "data" / "quantum_measurements_manifest.json").read_text(
            encoding="utf-8"
        )
    )
    records = Counter()
    samples = Counter()
    total = 0
    for line_number, line in enumerate(
        archive.open(encoding="utf-8", errors="strict"), 1
    ):
        try:
            row = json.loads(line)
        except Exception as exc:
            errors.append(f"archive line {line_number}: invalid JSON ({exc})")
            continue
        counts = row.get("counts")
        if not isinstance(counts, dict) or not counts:
            errors.append(f"archive line {line_number}: missing counts")
            continue
        observed = sum(int(value) for value in counts.values())
        declared = int(row.get("total_shots", -1))
        if observed != declared:
            errors.append(
                f"archive line {line_number}: shot mismatch {observed} != {declared}"
            )
        category = str(row.get("provider_class") or "missing")
        records[category] += 1
        samples[category] += observed
        total += observed

    expected = data_manifest["summary"]
    if total != int(expected["total_samples"]):
        errors.append(
            f"archive total mismatch: {total} != {expected['total_samples']}"
        )
    for category, expected_count in expected["records_by_provider_class"].items():
        if records[category] != int(expected_count):
            errors.append(
                f"archive record count mismatch for {category}: "
                f"{records[category]} != {expected_count}"
            )
    return errors, {
        "records": sum(records.values()),
        "samples": total,
        "records_by_class": dict(records),
        "samples_by_class": dict(samples),
    }


def check_model_metadata() -> list[str]:
    errors: list[str] = []
    metadata = json.loads(
        (ROOT / "weights" / "cosmos_born.meta.json").read_text(encoding="utf-8")
    )
    if int(metadata.get("params", 0)) != 1_842_432:
        errors.append("unexpected cosmos_born parameter count")
    if str(metadata.get("base_model") or "").upper().split()[0] != "NONE":
        errors.append("cosmos_born metadata no longer reports a from-scratch base")
    return errors


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--no-strict",
        action="store_true",
        help="allow extra files not listed in the release manifest",
    )
    args = parser.parse_args()
    if not MANIFEST.is_file():
        print("[FAIL] RELEASE_MANIFEST.json is missing")
        return 1

    errors = []
    errors.extend(check_manifest(strict=not args.no_strict))
    errors.extend(check_blank_credentials())
    archive_errors, archive_stats = check_public_archive()
    errors.extend(archive_errors)
    errors.extend(check_model_metadata())

    if errors:
        print(f"[FAIL] {len(errors)} release check(s) failed")
        for error in errors:
            print("  -", error)
        return 1

    print("[OK] release manifest hashes verified")
    print("[OK] shipped cloud credential fields are blank")
    print("[OK] cosmos_born metadata is internally consistent")
    print(
        "[OK] public archive:",
        f"{archive_stats['records']:,} records,",
        f"{archive_stats['samples']:,} samples",
    )
    for category in sorted(archive_stats["records_by_class"]):
        print(
            "    ",
            category,
            f"{archive_stats['records_by_class'][category]:,} records /",
            f"{archive_stats['samples_by_class'][category]:,} samples",
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())