qbit-scripting/deploy/snapshot_execution.py

# Copyright (C) 2026 Kingsrook, LLC.
# SPDX-License-Identifier: Apache-2.0
"""Copy a closed, hash-verified local candidate context. Never build or download."""
import argparse
import hashlib
import json
import logging
from pathlib import Path
import shutil
from typing import Any
import zipfile

ROOT = Path(__file__).resolve().parent
WASM_SHA = "4b247288d6cdff6783f8087a2a9039935f50af9099451cc841a3639c9625a18c"
EXTISM_SHA = "44eb7ed56c6b5610b97b9aa62a436e6ff7e3175730e487c42bb67105185d7e01"
JNA_SHA = "260c4b1e22b1db9e110ee441c4f13ce115f841fa48c41d78750986214b395557"
JNA_JNI_SHA = "ca07953d595210082339753d9e818a1fdb40509a17a41914d9a2cb0d2df6b6af"
JSON_SHA = "2a3438fbf9293174ea82bdfa35e83d8b965cfe6bddc034a1ac5d60107174c209"
FILES = {
   "supervisor": ("payload/lib/supervisor.jar", None),
   "jna": ("payload/lib/jna-5.18.1.jar", JNA_SHA),
   "json": ("payload/lib/json-20251224.jar", JSON_SHA),
   "extism": ("payload/native/libextism.so", EXTISM_SHA),
   "wasm": ("payload/runtime/qqq_javy_runtime.wasm", WASM_SHA),
   "license": ("metadata/LICENSE", None),
   "notice": ("metadata/NOTICE", None),
   "licenseReview": ("metadata/release-license-review.md", None),
   "cargoInventory": ("metadata/license-inventory.json", None),
}


def digest(path: Path) -> str:
   """Hash local bytes without loading entire native binaries into memory."""
   with path.open("rb") as stream:
      return hashlib.file_digest(stream, "sha256").hexdigest()


def snapshot(inputs: dict[str, Any], destination: Path) -> dict[str, Any]:
   """Verify the closed input set, then copy into a new cache directory."""
   if set(inputs) != set(FILES):
      raise ValueError("Exact named local inputs required")
   # Validate everything before creating the previously nonexistent context.
   verified = {}
   for name, (relative, pinned) in FILES.items():
      item = inputs[name]
      if not isinstance(item, dict) or set(item) != {"path", "sha256"}:
         raise ValueError("Each input needs only path and SHA256")
      path = Path(item["path"])
      expected = item["sha256"]
      if (not path.is_absolute() or not path.is_file() or path.is_symlink()
            or path.stat().st_size > 256 * 1024 * 1024
            or not isinstance(expected, str) or len(expected) != 64
            or any(c not in "0123456789abcdef" for c in expected)
            or (pinned is not None and expected != pinned)
            or digest(path) != expected):
         raise ValueError("Local input verification failed: " + name)
      verified[name] = (path, relative, expected)
   destination = Path(destination)
   cache = (Path.home() / ".cache").resolve()
   if (not destination.is_absolute() or destination.exists()
         or not destination.resolve().is_relative_to(cache)):
      raise ValueError("A new absolute output directory inside the current user's cache is required")
   with zipfile.ZipFile(verified["supervisor"][0]) as jar:
      names = jar.namelist()
      if ("com/kingsrook/scripting/supervisor/SupervisorMain.class" not in names
            or any(n.endswith(".class") and not n.startswith("com/kingsrook/scripting/supervisor/") for n in names)):
         raise ValueError("Standalone supervisor JAR required; QQQ/QBit classes refused")
   with zipfile.ZipFile(verified["jna"][0]) as jar:
      jni = jar.read("com/sun/jna/linux-x86-64/libjnidispatch.so")
   if (jni[:6] != b"\x7fELF\x02\x01" or jni[18:20] != b"\x3e\x00"
         or len(jni) > 4 * 1024 * 1024 or hashlib.sha256(jni).hexdigest() != JNA_JNI_SHA):
      raise ValueError("Pinned Linux JNI payload invalid")
   with verified["extism"][0].open("rb") as stream:
      header = stream.read(20)
   if header[:6] != b"\x7fELF\x02\x01" or header[18:20] != b"\x3e\x00":
      raise ValueError("Linux x86-64 Extism required")
   with verified["wasm"][0].open("rb") as stream:
      if stream.read(8) != b"\0asm\x01\0\0\0":
         raise ValueError("Fixed Wasm header invalid")
   destination.mkdir(parents=True)
   records = []
   for name, (source, relative, expected) in verified.items():
      target = destination / relative
      target.parent.mkdir(parents=True, exist_ok=True)
      shutil.copyfile(source, target)
      if digest(target) != expected:
         raise ValueError("Snapshot changed during copy: " + name)
      records.append({"path": relative, "sha256": expected, "bytes": target.stat().st_size})
   jni_path = destination / "payload/native/libjnidispatch.so"
   jni_path.write_bytes(jni)
   records.append({"path": jni_path.relative_to(destination).as_posix(),
               "sha256": digest(jni_path), "bytes": len(jni)})
   for filename in ("execution.Containerfile", "NOTICE-GAPS.md"):
      relative = "Containerfile" if filename.endswith("Containerfile") else "metadata/NOTICE-GAPS.md"
      shutil.copyfile(ROOT / filename, destination / relative)
      records.append({"path": relative, "sha256": digest(destination / relative)})
   (destination / ".dockerignore").write_text("*\n!Containerfile\n!payload/\n!payload/**\n!metadata/\n!metadata/**\n", encoding="utf-8")
   records.append({"path": ".dockerignore", "sha256": digest(destination / ".dockerignore")})
   manifest = {"productionQualified": False, "redistributionCleared": False,
            "architecture": "linux/amd64", "files": records,
            "supervisorClasses": sorted(n for n in names if n.endswith(".class")),
            "jnaNativeEntry": "com/sun/jna/linux-x86-64/libjnidispatch.so",
            "nativeSha256": EXTISM_SHA, "runtimeSha256": WASM_SHA,
            "supervisorSha256": verified["supervisor"][2], "jnaJniSha256": digest(jni_path)}
   (destination / "metadata/candidate.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
   verify_snapshot(destination)
   return manifest


def verify_snapshot(directory: Path) -> None:
   """Refuse unexpected context files or changed bytes immediately before a build."""
   directory = directory.resolve(strict=True)
   manifest_path = directory / "metadata/candidate.json"
   if manifest_path.stat().st_size > 128 * 1024:
      raise ValueError("Candidate manifest too large")
   manifest = json.loads(manifest_path.read_text(encoding="utf-8"), object_pairs_hook=no_duplicates)
   expected = {relative for relative, _ in FILES.values()} | {
      "payload/native/libjnidispatch.so", "Containerfile", "metadata/NOTICE-GAPS.md", ".dockerignore",
   }
   rows = manifest["files"]
   if (manifest["productionQualified"] is not False or manifest["redistributionCleared"] is not False
         or manifest["architecture"] != "linux/amd64" or manifest["runtimeSha256"] != WASM_SHA
         or manifest["nativeSha256"] != EXTISM_SHA or not isinstance(rows, list)
         or len(rows) != len(expected) or {row["path"] for row in rows} != expected):
      raise ValueError("Invalid candidate inventory")
   actual = set()
   for path in directory.rglob("*"):
      if path.is_symlink() or not path.resolve().is_relative_to(directory):
         raise ValueError("Links are not permitted in the build context")
      if path.is_file():
         actual.add(path.relative_to(directory).as_posix())
   if actual != expected | {"metadata/candidate.json"}:
      raise ValueError("Unexpected or missing build-context file")
   hashes = {row["path"]: row["sha256"] for row in rows}
   for relative in expected:
      if digest(directory / relative) != hashes[relative]:
         raise ValueError("Changed build-context bytes")
   for relative, pinned in FILES.values():
      if pinned is not None and hashes[relative] != pinned:
         raise ValueError("Pinned input identity changed")
   if (hashes["payload/lib/supervisor.jar"] != manifest["supervisorSha256"]
         or hashes["payload/native/libjnidispatch.so"] != manifest["jnaJniSha256"]
         or manifest["jnaJniSha256"] != JNA_JNI_SHA
         or hashes["Containerfile"] != digest(ROOT / "execution.Containerfile")
         or hashes["metadata/NOTICE-GAPS.md"] != digest(ROOT / "NOTICE-GAPS.md")):
      raise ValueError("Candidate linkage mismatch")


def no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
   """Reject ambiguous input descriptors rather than last-key-wins parsing."""
   result = {}
   for key, value in pairs:
      if key in result:
         raise ValueError("Duplicate input key")
      result[key] = value
   return result


def main() -> None:
   """Create local packaging evidence without a build or network operation."""
   parser = argparse.ArgumentParser(description=__doc__)
   parser.add_argument("inputs", type=Path, nargs="?")
   parser.add_argument("output", type=Path, nargs="?")
   parser.add_argument("--verify", type=Path)
   args = parser.parse_args()
   if args.verify is not None:
      if args.inputs is not None or args.output is not None:
         raise ValueError("Verification accepts only the existing context directory")
      verify_snapshot(args.verify)
      logging.warning("Closed candidate context verified; no build performed")
      return
   if args.inputs is None or args.output is None:
      raise ValueError("Local input descriptor and new cache output are required")
   if args.inputs.stat().st_size > 32 * 1024:
      raise ValueError("Input descriptor too large")
   snapshot(json.loads(args.inputs.read_text(encoding="utf-8"), object_pairs_hook=no_duplicates), args.output)
   logging.warning("Local candidate snapshot verified; unqualified; no build performed")


if __name__ == "__main__":
   main()