#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.12,<3.14" # /// """Build the read-only, candidate-bound NutriCoach v1.5 permission package.""" from __future__ import annotations import argparse import shutil import sys from datetime import UTC, datetime from pathlib import Path from typing import cast from uuid import uuid4 sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts.nutricoach_v150_live_upgrade_common import ( JsonValue, LiveTarget, ServiceReader, UpgradeDenied, canonical, load_object, sha256_bytes, sha256_file, string_at, ) from scripts.nutricoach_v150_live_upgrade_state import ( contract_snapshot, profiles_snapshot, service_state, ) from scripts.nutricoach_v150_live_upgrade_boundary import clean_boundary from scripts.nutricoach_v150_prepare_helpers import ( CAPACITY, CANDIDATE, FINAL_QUALIFICATION_SHA, MANIFEST_SHA, OLD_DENIAL_SHA, QUALIFICATION_SHA, changed_paths, cleanup_clone as cleanup_clone, migration, rollback_rehearsal, validate_candidate, write_receipt, ) def prepare_package( target: LiveTarget, manifest_path: Path, output: Path, service_reader: ServiceReader = service_state, now: datetime | None = None, controller_derivation: str = "test-controller-derivation", ) -> dict[str, JsonValue]: _ = validate_candidate(manifest_path) if output.exists() or target.profiles_root.resolve() in output.resolve().parents: raise UpgradeDenied("unsafe_output") output.parent.mkdir(parents=True, exist_ok=True, mode=0o700) before_profiles = profiles_snapshot(target) before_contract = contract_snapshot(target) before_service = service_reader(target.service_name) boundary = clean_boundary(target.profile_root, now or datetime.now(UTC)) registry_path = target.profile_root / "customers/registry.json" registry_before = registry_path.read_bytes() registry_after, migration_receipt = migration(registry_before) rollback = rollback_rehearsal(target, registry_after, output.parent) after_profiles = profiles_snapshot(target) after_contract = contract_snapshot(target) after_service = service_reader(target.service_name) _ = clean_boundary(target.profile_root, now or datetime.now(UTC)) if ( before_profiles["stable_digest"] != after_profiles["stable_digest"] or before_contract["digest"] != after_contract["digest"] or before_service != after_service or registry_path.read_bytes() != registry_before ): raise UpgradeDenied("live_immutability") staging = output.with_name(f".{output.name}.{uuid4().hex}.staging") staging.mkdir(mode=0o700) try: snapshots: dict[str, JsonValue] = { "stable_before_digest": before_profiles["stable_digest"], "stable_after_digest": after_profiles["stable_digest"], "volatile_before_digest": before_profiles["volatile_digest"], "volatile_after_digest": after_profiles["volatile_digest"], "volatile_changed_paths": changed_paths(before_profiles, after_profiles), "contract_before_digest": before_contract["digest"], "contract_after_digest": after_contract["digest"], "service_before": cast(dict[str, JsonValue], before_service), "service_after": cast(dict[str, JsonValue], after_service), } receipts: dict[str, JsonValue] = { "snapshot-before.json": { "profiles": before_profiles, "contract_hashes_only": before_contract, "service": cast(dict[str, JsonValue], before_service), }, "snapshot-after.json": { "profiles": after_profiles, "contract_hashes_only": after_contract, "service": cast(dict[str, JsonValue], after_service), }, "clean-boundary.json": boundary, "migration-dry-run.json": migration_receipt, "rollback-rehearsal.json": rollback, } for name, value in receipts.items(): write_receipt(staging / name, value) evidence: dict[str, JsonValue] = { str(output / name): sha256_file(staging / name) for name in receipts } payload: dict[str, JsonValue] = { "schema": "nutricoach-v150-live-upgrade-package-v1", "status": "AWAITING_AUTHORIZATION", "candidate_digest": CANDIDATE, "candidate_manifest": str(manifest_path.resolve()), "candidate_manifest_sha256": MANIFEST_SHA, "qualification_sha256": QUALIFICATION_SHA, "final_qualification_sha256": FINAL_QUALIFICATION_SHA, "old_manifest_denial_sha256": OLD_DENIAL_SHA, "capacity": CAPACITY, "controller_derivation_sha256": controller_derivation, "read_only_preflight": True, "target": { "profile_root": str(target.profile_root), "profiles_root": str(target.profiles_root), "service_name": target.service_name, "unit_file": str(target.unit_file), "dropin_dir": str(target.dropin_dir), }, "snapshots": snapshots, "evidence": evidence, } package_digest = sha256_bytes(canonical(payload)) phrase = f"AUTHORIZE NUTRICOACH V1.5 LIVE UPGRADE {package_digest}" package: dict[str, JsonValue] = { "payload": payload, "package_digest": package_digest, "approval_phrase": phrase, } write_receipt(staging / "package.json", package) awaiting: dict[str, JsonValue] = { "approval_phrase": phrase, "candidate_digest": CANDIDATE, "package_digest": package_digest, "package_sha256": sha256_file(staging / "package.json"), "status": "AWAITING_AUTHORIZATION", } write_receipt(staging / "awaiting-authorization.json", awaiting) write_receipt( staging / "red-to-green.json", { "red_exit_code": 2, "red_result": "No such file or directory", "green_expected": "AWAITING_AUTHORIZATION", }, ) _ = staging.rename(output) return awaiting except (OSError, UpgradeDenied, ValueError, KeyError): shutil.rmtree(staging, ignore_errors=True) raise def main() -> int: parser = argparse.ArgumentParser(description=__doc__) _ = parser.add_argument("--profile-root", required=True, type=Path) _ = parser.add_argument("--candidate-manifest", required=True, type=Path) _ = parser.add_argument("--output", required=True, type=Path) _ = parser.add_argument("--read-only", required=True, action="store_true") _ = parser.parse_args() profile_root = Path(sys.argv[sys.argv.index("--profile-root") + 1]) candidate_manifest = Path(sys.argv[sys.argv.index("--candidate-manifest") + 1]) output = Path(sys.argv[sys.argv.index("--output") + 1]) expected_profile = Path("/home/cube/.hermes/profiles/dualcoachtest") expected_output = Path( "/home/cube/.hermes/migrations/nutricoach-v1.5.0-combined/" + "preflight-v12-runtime-state" ) derivation_path = Path( "/home/cube/.hermes/migrations/nutricoach-v1.5.0-combined/" + "live-transaction-preseal-v12-runtime-state/" + "controller-derivation.json" ) if profile_root != expected_profile or output != expected_output: raise SystemExit("DENIED:authoritative_live_target") target = LiveTarget( expected_profile, expected_profile.parent, "hermes-gateway-dualcoachtest.service", Path("/home/cube/.config/systemd/user/hermes-gateway-dualcoachtest.service"), Path("/home/cube/.config/systemd/user/hermes-gateway-dualcoachtest.service.d"), ) try: derivation = load_object(derivation_path) controller_derivation = string_at( derivation.get("controller_derivation_sha256"), "controller_derivation", ) receipt = prepare_package( target, candidate_manifest, output, controller_derivation=controller_derivation, ) except (OSError, UpgradeDenied) as exc: raise SystemExit(f"DENIED:{exc}") from exc print(canonical(receipt).decode()) return 0 if __name__ == "__main__": raise SystemExit(main())