"""Capacity migration, systemd postimages, and post-fence operations.""" from __future__ import annotations from pathlib import Path from pydantic import JsonValue, TypeAdapter from scripts.nutricoach_v150_live_models import CANDIDATE_DIGEST from scripts.nutricoach_v150_sealed_authority import atomic_write, digest from scripts.nutricoach_v150_sealed_target import ( DisposableService, HostError, HostPaths, Service, ) from scripts.nutricoach_v150_weekly_authority import ( build_weekly_authority_postimage, initialize_canonical_weekly_files, ) _OBJECT = TypeAdapter(dict[str, JsonValue]) def capacity_after( registry: Path, candidate_digest: str = CANDIDATE_DIGEST, ) -> bytes: """Return exact capacity-five registry postimage.""" from checkin_cli.multi_customer_admission_migration import ( propose_admission_migration, ) after, _ = propose_admission_migration( registry.read_bytes(), candidate_digest=candidate_digest, max_enabled_customers=5, ) return after def protected_paths(paths: HostPaths) -> tuple[Path, ...]: """Return stable protected files for read-only drift verification.""" inventory = paths.protected_inventory if inventory is None: return paths.mutable document = _OBJECT.validate_json(inventory.read_bytes()) profiles = document.get("profiles") if not isinstance(profiles, dict): raise HostError("protected_inventory") rows = profiles.get("stable") if not isinstance(rows, list): raise HostError("protected_inventory") values: list[Path] = list(paths.mutable) for raw in rows: relative = raw.get("path") if isinstance(raw, dict) else None if ( isinstance(raw, dict) and raw.get("profile") == paths.profile.name and isinstance(relative, str) ): path = paths.profile / relative if path.is_file() and not path.is_symlink() and path not in values: values.append(path) return tuple(values) def snapshot_paths(paths: HostPaths) -> tuple[Path, ...]: """Return only paths the transaction can mutate and must restore.""" return paths.mutable def apply_migrations( paths: HostPaths, after: bytes, candidate_digest: str = CANDIDATE_DIGEST, created_files: tuple[Path, ...] = (), weekly_authority: Path | None = None, *, receipt_preserving: bool = False, ) -> None: """Apply capacity and canonical weekly reminder authority postimages.""" from checkin_cli.customer_admin import ( commit_multi_customer_admission_migration, ) from checkin_cli.multi_customer_admission_migration import ( propose_admission_migration, ) expected_after, proposal = propose_admission_migration( paths.registry.read_bytes(), candidate_digest=candidate_digest, max_enabled_customers=5, ) if expected_after != after: raise HostError("capacity_postimage") if receipt_preserving: committed = commit_multi_customer_admission_migration( paths.profile, proposal, proposal.approval_phrase, ) if committed != proposal.after_sha256: raise HostError("capacity_commit") else: atomic_write(paths.registry, after) initialize_canonical_weekly_files(created_files) atomic_write( paths.config, build_weekly_authority_postimage( paths, candidate_digest=candidate_digest, authority_path=weekly_authority, ), ) def switch_postimages( paths: HostPaths, candidate_digest: str = CANDIDATE_DIGEST, authority_pin: bytes | None = None, ) -> dict[Path, str]: """Publish exact unit, drop-in, and credential postimages.""" credentials = publish_runtime_authority_credentials( paths, candidate_digest, authority_pin, ) current = str(paths.current_runtime).encode() successor = str(paths.successor_runtime).encode() unit_before = paths.unit.read_bytes() if current not in unit_before: raise HostError("runtime_switch_source") unit_payload = unit_before.replace(current, successor) if current in unit_payload: raise HostError("runtime_switch") atomic_write(paths.unit, unit_payload) lines: list[bytes] = [] for line in paths.dropin.read_bytes().splitlines(): target = line.replace(current, successor) if target.startswith(b"LoadCredential=") and b":" in target: name = target.split(b"=", 1)[1].split(b":", 1)[0] destination = credentials / name.decode() target = b"LoadCredential=" + name + b":" + str(destination).encode() lines.append(target) atomic_write(paths.dropin, b"\n".join(lines) + b"\n") values = (paths.unit, paths.dropin, *sorted(credentials.iterdir())) return {path: digest(path.read_bytes()) for path in values} def publish_runtime_authority_credentials( paths: HostPaths, candidate_digest: str, authority_pin: bytes | None, ) -> Path: """Create canonical successor credential files before service switching.""" credential_root = paths.successor_runtime.parent / "runtime-authority" if credential_root.exists(): if not credential_root.is_dir() or credential_root.is_symlink(): raise HostError("runtime_authority_credentials") else: credential_root.mkdir(mode=0o700) for line in paths.dropin.read_bytes().splitlines(): if not line.startswith(b"LoadCredential=") or b":" not in line: continue name, source_text = line.split(b"=", 1)[1].split(b":", 1) source = Path(source_text.decode()) if name == b"task26-authority-pin.json" and authority_pin is not None: payload = authority_pin elif name in {b"candidate-digest", b"task26-candidate-digest"}: payload = candidate_digest.encode() + b"\n" else: payload = source.read_bytes() atomic_write(credential_root / name.decode(), payload, 0o400) return credential_root def service_state(service: Service, paths: HostPaths) -> dict[str, str]: """Return real/faithful service fields with exact unit ExecStart.""" state = service.observe() if isinstance(service, DisposableService): state["ExecStart"] = paths.unit.read_text(encoding="utf-8") return state def verify_postimages(postimages: dict[Path, str]) -> bool: """Verify every precomputed systemd and credential postimage.""" return bool(postimages) and all( digest(path.read_bytes()) == expected for path, expected in postimages.items() ) def post_fence( paths: HostPaths, state: dict[str, str], postimages: dict[Path, str], candidate_digest: str = CANDIDATE_DIGEST, ) -> None: """Fence capacity, runtime, postimages, and actual service identity.""" document = _OBJECT.validate_json(paths.registry.read_bytes()) policy = document.get("admission_policy") if ( not isinstance(policy, dict) or policy.get("max_enabled_customers") != 5 or policy.get("candidate_digest") != candidate_digest ): raise HostError("capacity_post_fence") if str(paths.successor_runtime).encode() not in paths.unit.read_bytes(): raise HostError("runtime_post_fence") if ( state["ActiveState"] != "active" or state["SubState"] != "running" or int(state["MainPID"]) <= 0 or str(paths.successor_runtime) not in state["ExecStart"] or not verify_postimages(postimages) ): raise HostError("systemd_post_fence")