"""Offline rehydration and sealed expected-state checks for Task26 bundles.""" from __future__ import annotations import argparse import hashlib import json import os import shutil import stat import subprocess import sys import tempfile from pathlib import Path from typing import Any, Mapping, cast def canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _private(path: Path, *, directory: bool = False) -> None: info = path.lstat() expected = stat.S_ISDIR if directory else stat.S_ISREG modes = {0o700, 0o500} if directory else {0o600, 0o400} if ( stat.S_ISLNK(info.st_mode) or not expected(info.st_mode) or info.st_uid != os.geteuid() or (not directory and info.st_nlink != 1) or stat.S_IMODE(info.st_mode) not in modes ): raise ValueError(f"sealed private path is invalid: {path}") def _object(value: object, label: str) -> dict[str, Any]: if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): raise ValueError(f"{label} is invalid") return cast(dict[str, Any], value) def verify_expected_state( bundle: Path, expected_path: Path, *, hermes_wheel: Path, profile_wheel: Path, ) -> dict[str, Any]: """Validate the hash-bound portable expectation before creating a runtime.""" _private(bundle, directory=True) _private(expected_path) _private(hermes_wheel) _private(profile_wheel) expected = _object(json.loads(expected_path.read_text(encoding="utf-8")), "sealed expected state") if set(expected) != { "schema", "candidate_digest", "runtime_portable", "hermes_wheel", "profile_wheel", "verifier", "provenance_helper", "expected_status", "expected_result_sha256", "authority", "document_sha256", } or expected.get("schema") != "task26-sealed-final-state-v1": raise ValueError("sealed expected-state schema is invalid") claimed = expected.get("document_sha256") unsigned = {key: value for key, value in expected.items() if key != "document_sha256"} if claimed != hashlib.sha256(canonical(unsigned)).hexdigest(): raise ValueError("sealed expected-state digest is invalid") for label, path in (("hermes_wheel", hermes_wheel), ("profile_wheel", profile_wheel)): binding = _object(expected.get(label), label) if binding != {"filename": path.name, "sha256": sha256_file(path)}: raise ValueError(f"sealed {label} digest differs") for label in ("verifier", "provenance_helper"): binding = _object(expected.get(label), label) relative = binding.get("relative_path") if not isinstance(relative, str) or Path(relative).is_absolute() or ".." in Path(relative).parts: raise ValueError(f"sealed {label} path is invalid") target = bundle / relative _private(target) if binding.get("sha256") != sha256_file(target): raise ValueError(f"sealed {label} digest differs") authority = _object(expected.get("authority"), "sealed authority") for name in ("registry.json", "qualification-ledger.json"): target = bundle / "candidate-authority" / name _private(target) if authority.get(name.removesuffix(".json") + "_sha256") != sha256_file(target): raise ValueError("sealed authority digest differs") return expected def _portable_verifier_result(result: Mapping[str, object]) -> dict[str, object]: return { key: value for key, value in result.items() if key not in {"package_authority"} } def build_expected_state( bundle: Path, *, candidate_digest: str, runtime_portable: Mapping[str, object], hermes_wheel: Path, profile_wheel: Path, verifier_relative_path: str, provenance_relative_path: str, verifier_result: Mapping[str, object], ) -> dict[str, object]: """Build, but do not write or freeze, one successor expected-state document.""" authority_result = _object(verifier_result.get("candidate_authority"), "candidate authority result") document: dict[str, object] = { "schema": "task26-sealed-final-state-v1", "candidate_digest": candidate_digest, "runtime_portable": dict(runtime_portable), "hermes_wheel": {"filename": hermes_wheel.name, "sha256": sha256_file(hermes_wheel)}, "profile_wheel": {"filename": profile_wheel.name, "sha256": sha256_file(profile_wheel)}, "verifier": {"relative_path": verifier_relative_path, "sha256": sha256_file(bundle / verifier_relative_path)}, "provenance_helper": {"relative_path": provenance_relative_path, "sha256": sha256_file(bundle / provenance_relative_path)}, "expected_status": verifier_result.get("status"), "expected_result_sha256": hashlib.sha256( canonical(_portable_verifier_result(verifier_result)) ).hexdigest(), "authority": { "registry_sha256": sha256_file(bundle / "candidate-authority/registry.json"), "qualification-ledger_sha256": sha256_file(bundle / "candidate-authority/qualification-ledger.json"), "registry_head_sha256": authority_result.get("registry_head_sha256"), "ledger_head_sha256": authority_result.get("ledger_head_sha256"), }, } document["document_sha256"] = hashlib.sha256(canonical(document)).hexdigest() return document def _portable_runtime(runtime: Mapping[str, object]) -> dict[str, object]: distributions = _object(runtime.get("distributions"), "runtime distributions") portable_distributions: dict[str, object] = {} for role in ("profile", "hermes"): distribution = _object(distributions.get(role), f"{role} distribution") raw_inventory = distribution.get("installed_inventory") if not isinstance(raw_inventory, list): raise ValueError(f"{role} installed inventory is invalid") if any(not isinstance(row, dict) for row in raw_inventory): raise ValueError(f"{role} installed inventory row is invalid") inventory = [ row for row in raw_inventory if ".." not in Path(str(row.get("path", ""))).parts ] portable_distribution: dict[str, object] = { key: distribution.get(key) for key in ( "distribution_name", "distribution_version", "metadata_sha256", "wheel_filename", "wheel_sha256", "wheel_record_sha256", ) } portable_distribution["installed_inventory"] = inventory portable_distribution["installed_inventory_sha256"] = hashlib.sha256( canonical(inventory) ).hexdigest() portable_distributions[role] = portable_distribution interpreter = _object(runtime.get("interpreter"), "runtime interpreter") return { "schema": "installed-golden-runtime-portable-v1", "interpreter": { "sha256": interpreter.get("sha256"), "version": interpreter.get("version"), }, "distributions": portable_distributions, } def _clean_environment() -> dict[str, str]: environment = dict(os.environ) environment.pop("PYTHONPATH", None) environment.pop("PYTHONHOME", None) environment["PYTHONNOUSERSITE"] = "1" environment["PIP_NO_INDEX"] = "1" return environment def rehydrate_and_verify( bundle: Path, expected_path: Path, *, hermes_wheel: Path, profile_wheel: Path, wheelhouse: Path | None = None, temp_parent: Path | None = None, ) -> dict[str, object]: """Create an offline temp runtime, rerun the verifier, and always scrub it.""" expected = verify_expected_state( bundle, expected_path, hermes_wheel=hermes_wheel, profile_wheel=profile_wheel ) temporary = Path(tempfile.mkdtemp(prefix="task26-rehydrate-", dir=temp_parent)) try: venv = temporary / "venv" environment = _clean_environment() subprocess.run( [sys.executable, "-m", "venv", str(venv)], check=True, env=environment, capture_output=True, text=True, ) python = venv / ("Scripts/python.exe" if os.name == "nt" else "bin/python") install = [str(python), "-m", "pip", "install", "--no-index"] if wheelhouse is None: install.append("--no-deps") else: _private(wheelhouse, directory=True) install.extend(("--find-links", str(wheelhouse))) install.extend((str(profile_wheel), str(hermes_wheel))) subprocess.run(install, check=True, env=environment, capture_output=True, text=True) site_packages_text = subprocess.run( [str(python), "-c", "import sysconfig; print(sysconfig.get_paths()['purelib'])"], check=True, env=environment, capture_output=True, text=True, ).stdout.strip() site_packages = Path(site_packages_text) runtime_path = temporary / "runtime.json" helper = bundle / str(_object(expected["provenance_helper"], "provenance helper")["relative_path"]) probe = ( "import importlib.util,json,pathlib,sys;" "p=pathlib.Path(sys.argv[1]);s=importlib.util.spec_from_file_location('sealed_provenance',p);" "m=importlib.util.module_from_spec(s);s.loader.exec_module(m);" "r=m.collect_installed_runtime(venv=pathlib.Path(sys.argv[2]),site_packages=pathlib.Path(sys.argv[3])," "profile_wheel=pathlib.Path(sys.argv[4]),hermes_wheel=pathlib.Path(sys.argv[5]));" "import checkin_cli,gateway;" "print(json.dumps(r,sort_keys=True,separators=(',',':')))" ) runtime = subprocess.run( [str(python), "-c", probe, str(helper), str(venv), str(site_packages), str(profile_wheel), str(hermes_wheel)], check=True, env=environment, capture_output=True, text=True, ).stdout runtime_document = _object(json.loads(runtime), "rehydrated runtime") if _portable_runtime(runtime_document) != expected.get("runtime_portable"): raise ValueError("rehydrated runtime differs from sealed portable bytes") runtime_path.write_text(runtime, encoding="utf-8") runtime_path.chmod(0o600) verifier = bundle / str(_object(expected["verifier"], "verifier")["relative_path"]) completed = subprocess.run( [str(python), str(verifier), str(bundle), "--rehydrated-runtime", str(runtime_path)], check=True, env=environment, capture_output=True, text=True, ) result = _object(json.loads(completed.stdout), "rehydrated verifier result") authority = _object(result.get("candidate_authority"), "rehydrated authority") expected_authority = _object(expected.get("authority"), "expected authority") if ( result.get("status") != expected.get("expected_status") or result.get("candidate_digest") != expected.get("candidate_digest") or authority.get("registry_head_sha256") != expected_authority.get("registry_head_sha256") or authority.get("ledger_head_sha256") != expected_authority.get("ledger_head_sha256") or hashlib.sha256(canonical(_portable_verifier_result(result))).hexdigest() != expected.get("expected_result_sha256") ): raise ValueError("rehydrated verifier result differs from sealed expected state") return result finally: if temporary.exists(): for path in temporary.rglob("*"): try: if path.is_symlink(): continue path.chmod(0o700 if path.is_dir() else 0o600) except OSError: pass shutil.rmtree(temporary, ignore_errors=False) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("bundle", type=Path) parser.add_argument("expected_state", type=Path) parser.add_argument("--hermes-wheel", type=Path, required=True) parser.add_argument("--profile-wheel", type=Path, required=True) parser.add_argument("--wheelhouse", type=Path) parser.add_argument("--temp-parent", type=Path) args = parser.parse_args() try: result = rehydrate_and_verify( args.bundle, args.expected_state, hermes_wheel=args.hermes_wheel, profile_wheel=args.profile_wheel, wheelhouse=args.wheelhouse, temp_parent=args.temp_parent, ) except (OSError, ValueError, subprocess.SubprocessError, json.JSONDecodeError) as exc: print(json.dumps({"status": "TASK26_REHYDRATION_FAIL", "reason": str(exc)}, sort_keys=True), file=sys.stderr) return 1 print(json.dumps(result, sort_keys=True, separators=(",", ":"))) return 0 if __name__ == "__main__": raise SystemExit(main())