"""Freeze the fresh V15 runtime-authority deployment package.""" from __future__ import annotations import hashlib import json import os import shutil import stat import subprocess import sys import zipfile from pathlib import Path from pydantic import JsonValue, TypeAdapter sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from gateway.platforms.task26_runtime_authority import ( build_runtime_authority_pin, load_task26_production_authority, ) from scripts.nutricoach_v150_live_upgrade_common import canonical from scripts.nutricoach_v150_runtime_ops import dependency_snapshot_digest _OBJECT = TypeAdapter(dict[str, JsonValue]) SOURCE = Path("/home/cube/projects/richard/.worktrees/nutricoach-v150-combined") BASE = Path("/home/cube/.hermes/migrations/nutricoach-v1.5.0-combined") PROFILE = Path("/home/cube/.hermes/profiles/dualcoachtest") UNIT = Path.home() / ".config/systemd/user/hermes-gateway-dualcoachtest.service" DROPIN = UNIT.parent / f"{UNIT.name}.d/task26-authority.conf" PRESEAL = BASE / "live-transaction-preseal-v15-runtime-authority-r21" PREFLIGHT = BASE / "preflight-v15-runtime-authority-r21" AUTHORITY_ID = "nutricoach-v150-v15-runtime-authority-r21" PRIOR_PRESEAL = BASE / "live-transaction-preseal-v14-live-representative-r20" PRIOR_PREFLIGHT = BASE / "preflight-v14-live-representative-r20" def _load(path: Path) -> dict[str, JsonValue]: return _OBJECT.validate_json(path.read_bytes()) def _write(path: Path, value: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) document = _OBJECT.validate_python(value) _ = path.write_bytes(canonical(document) + b"\n") def _sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _record_sha(path: Path) -> str: with zipfile.ZipFile(path) as wheel: record = next(name for name in wheel.namelist() if name.endswith("/RECORD")) return hashlib.sha256(wheel.read(record)).hexdigest() def _service_state() -> dict[str, JsonValue]: fields = ("ActiveState", "SubState", "MainPID", "ExecStart", "NRestarts") result = subprocess.run( ( "/usr/bin/systemctl", "--user", "show", "hermes-gateway-dualcoachtest.service", *(f"--property={field}" for field in fields), ), check=True, capture_output=True, text=True, ) rows = dict( line.split("=", 1) for line in result.stdout.splitlines() if "=" in line ) return {field: rows.get(field, "") for field in fields} def _credential_paths() -> tuple[Path, Path]: values: dict[str, Path] = {} for line in DROPIN.read_text(encoding="utf-8").splitlines(): if not line.startswith("LoadCredential=") or ":" not in line: continue name, raw_path = line.split("=", 1)[1].split(":", 1) values[name] = Path(raw_path) return ( values["task26-authority-pin.json"], values["task26-candidate-digest"], ) def _authority_baseline() -> dict[str, JsonValue]: pin_path, candidate_path = _credential_paths() old_pin = os.environ.get("TASK26_AUTHORITY_PIN") old_candidate = os.environ.get("TASK26_CANDIDATE_DIGEST_FILE") os.environ["TASK26_AUTHORITY_PIN"] = str(pin_path) os.environ["TASK26_CANDIDATE_DIGEST_FILE"] = str(candidate_path) try: source, candidate = load_task26_production_authority( profile_root=PROFILE, package_root=BASE, ) finally: if old_pin is None: _ = os.environ.pop("TASK26_AUTHORITY_PIN", None) else: os.environ["TASK26_AUTHORITY_PIN"] = old_pin if old_candidate is None: _ = os.environ.pop("TASK26_CANDIDATE_DIGEST_FILE", None) else: os.environ["TASK26_CANDIDATE_DIGEST_FILE"] = old_candidate registry = source.root / "candidate-authority/registry.json" ledger = source.root / "candidate-authority/qualification-ledger.json" pin = build_runtime_authority_pin(source.root) registry_document = _load(registry) events = registry_document.get("events") if not isinstance(events, list) or not events: raise RuntimeError("authority_events") latest = events[-1] if not isinstance(latest, dict): raise RuntimeError("authority_event") historical_pass = latest.get("historical_pass_digest") if not isinstance(historical_pass, str): raise RuntimeError("authority_historical_pass") return _OBJECT.validate_python({ "authority_root": str(source.root), "candidate_digest": candidate, "candidate_path": str(candidate_path), "candidate_file_sha256": _sha(candidate_path), "event_count": pin["event_count"], "genesis_sha256": pin["genesis_sha256"], "historical_pass_digest": historical_pass, "ledger_file_sha256": _sha(ledger), "ledger_head_sha256": pin["ledger_head_sha256"], "pin_path": str(pin_path), "pin_file_sha256": _sha(pin_path), "registry_file_sha256": _sha(registry), "registry_head_sha256": pin["registry_head_sha256"], "source_id": pin["source_id"], }) def _controller_files(candidate: dict[str, JsonValue]) -> dict[str, JsonValue]: rows = candidate.get("source_inventory") if not isinstance(rows, list): raise RuntimeError("candidate_source_inventory") files: dict[str, JsonValue] = {} for row in rows: if not isinstance(row, dict) or not isinstance(row.get("path"), str): raise RuntimeError("candidate_source_entry") relative = Path(str(row["path"])) product = Path(*relative.parts[2:]) if (SOURCE / product).is_file(): files[product.as_posix()] = _sha(SOURCE / product) for relative in ( "scripts/execute_nutricoach_v150_sealed_live.py", "scripts/nutricoach_v150_controller_worker.py", "scripts/nutricoach_v150_detached_bootstrap.py", "scripts/prepare_nutricoach_v150_v15_preseal.py", "scripts/verify_nutricoach_v150_preseal_v15.py", ): files[relative] = _sha(SOURCE / relative) return dict(sorted(files.items())) def main() -> int: if len(sys.argv) != 4: raise RuntimeError("usage: preparer CANDIDATE_ROOT HERMES_WHEEL PROFILE_WHEEL") candidate_root, hermes_wheel, profile_wheel = map(Path, sys.argv[1:]) if PRESEAL.exists() or PREFLIGHT.exists(): raise RuntimeError("v15_paths_exist") candidate = _load(candidate_root / "manifest.json") digest = candidate.get("candidate_digest") if not isinstance(digest, str): raise RuntimeError("candidate_digest") for wheel in (hermes_wheel, profile_wheel): if not wheel.is_file(): raise RuntimeError("wheel") PREFLIGHT.mkdir(parents=True) PRESEAL.mkdir(parents=True) protected = PREFLIGHT / "snapshot-before.json" _ = shutil.copy2(PRIOR_PREFLIGHT / "snapshot-before.json", protected) dependency_root = PRESEAL / "dependencies/site-packages" predecessor_site = next( (PROFILE / ".strict-runtime/6c9c4394-v132/venv/lib").glob( "python*/site-packages" ) ) for name in ("croniter", "croniter-6.0.0.dist-info"): _ = shutil.copytree( predecessor_site / name, dependency_root / name, ignore=shutil.ignore_patterns("__pycache__", "*.pyc", "*.pyo"), ) controller_files = _controller_files(candidate) controller_root = PRESEAL / "controller-source" for relative in controller_files: destination = controller_root / relative destination.parent.mkdir(parents=True, exist_ok=True) _ = shutil.copy2(SOURCE / relative, destination) _write( PRESEAL / "controller-source-manifest.json", { "schema": "nutricoach-v150-controller-source-manifest-v15", "files": controller_files, }, ) baseline = _authority_baseline() target_root = PROFILE / f".strict-runtime/{digest[:8]}-v150" wheels = [ {"path": str(path), "sha256": _sha(path), "record_sha256": _record_sha(path)} for path in (hermes_wheel, profile_wheel) ] target_binding = { "authority_id": AUTHORITY_ID, "profile_root": str(PROFILE), "current_runtime": str(PROFILE / ".strict-runtime/6c9c4394-v132/venv"), "successor_runtime": str(target_root / "venv"), "service_name": "hermes-gateway-dualcoachtest.service", "unit": str(UNIT), "dropin": str(DROPIN), "execution_root": str(BASE / "live-executions-v15" / AUTHORITY_ID), "global_approval_ledger": str(BASE / "live-authorization-v15" / AUTHORITY_ID), "protected_inventory": str(protected), "protected_inventory_sha256": _sha(protected), "dependency_snapshot": str(dependency_root), "dependency_snapshot_sha256": dependency_snapshot_digest(dependency_root), "registry_sha256": _sha(PROFILE / "customers/registry.json"), } payload = _OBJECT.validate_python({ "schema": "nutricoach-v150-live-upgrade-package-v15", "candidate_digest": digest, "candidate_manifest": str(candidate_root / "manifest.json"), "candidate_manifest_sha256": _sha(candidate_root / "manifest.json"), "controller_derivation_sha256": hashlib.sha256( canonical(controller_files) ).hexdigest(), "authority_baseline": baseline, "target_binding": target_binding, "service": _service_state(), "wheels": wheels, }) package_digest = hashlib.sha256(canonical(payload)).hexdigest() phrase = f"AUTHORIZE NUTRICOACH V1.5 LIVE UPGRADE {package_digest}" package = { "package_digest": package_digest, "approval_phrase": phrase, "payload": payload, } _write(PREFLIGHT / "package.json", package) target = { **target_binding, "schema": "nutricoach-v150-sealed-live-target-v15", "approval_phrase": phrase, "package_digest": package_digest, "candidate_digest": digest, "candidate_manifest": str(candidate_root / "manifest.json"), "candidate_manifest_sha256": _sha(candidate_root / "manifest.json"), "controller_derivation_sha256": payload["controller_derivation_sha256"], "permission_package": str(PREFLIGHT / "package.json"), "permission_package_sha256": _sha(PREFLIGHT / "package.json"), "authority_baseline": baseline, "wheels": wheels, } _write(PRESEAL / "sealed-target.json", target) prior = _load(PRIOR_PRESEAL / "package-supersession.json") rows = prior.get("superseded") if not isinstance(rows, list): raise RuntimeError("supersession") rows.append({ "package_digest": prior["active_package_digest"], "approval_phrase_reusable": False, "reason": "r20 consumed and rolled back after external authority rejection", }) _write( PRESEAL / "package-supersession.json", { "schema": "nutricoach-v150-package-supersession-v15", "active_package_digest": package_digest, "active_package_sha256": _sha(PREFLIGHT / "package.json"), "active_approval_phrase": phrase, "superseded": rows, }, ) entries = { path.relative_to(PRESEAL).as_posix(): _sha(path) for path in PRESEAL.rglob("*") if path.is_file() and path.name != "package-manifest.json" } _write( PRESEAL / "package-manifest.json", { "schema": "nutricoach-v150-detached-preseal-package-v15", "status": "FROZEN_AWAITING_AUDIT", "self_exclusion": "package-manifest.json only", "new_package_digest": package_digest, "new_approval_phrase": phrase, "approval_consumed": False, "entries": dict(sorted(entries.items())), }, ) for path in PRESEAL.rglob("*"): if path.is_file() and not path.is_symlink(): path.chmod(0o444) for path in sorted( (item for item in PRESEAL.rglob("*") if item.is_dir()), reverse=True ): path.chmod(0o555) PRESEAL.chmod(0o555) if any( stat.S_IMODE(path.stat(follow_symlinks=False).st_mode) & 0o222 for path in (PRESEAL, *PRESEAL.rglob("*")) ): raise RuntimeError("immutable_mode") print(json.dumps({"candidate_digest": digest, "package_digest": package_digest})) return 0 if __name__ == "__main__": raise SystemExit(main())