|
| 1 | +"""Control-plane BootReleaseSet planner. |
| 2 | +
|
| 3 | +This module consumes the canonical SourceOS control-plane BootReleaseSet shape |
| 4 | +from `SourceOS-Linux/sourceos-spec` and turns it into a safe, non-mutating boot |
| 5 | +plan for sourceos-boot. |
| 6 | +
|
| 7 | +It deliberately performs no network, disk, kexec, install, rollback, or key |
| 8 | +rotation side effects. It validates the boot intent and emits the operations a |
| 9 | +future executor would need to perform, together with the proof reports required |
| 10 | +by policy. |
| 11 | +""" |
| 12 | + |
| 13 | +from __future__ import annotations |
| 14 | + |
| 15 | +from dataclasses import dataclass |
| 16 | +from typing import Any |
| 17 | + |
| 18 | + |
| 19 | +SAFE_DIGEST_PREFIXES = ("sha256:", "sha384:", "sha512:") |
| 20 | +BOOT_ACTION_BY_CHANNEL = { |
| 21 | + "live": "boot-live-environment", |
| 22 | + "installer": "plan-install", |
| 23 | + "rescue": "plan-rescue", |
| 24 | + "rollback": "plan-rollback", |
| 25 | + "bootstrap": "plan-bootstrap", |
| 26 | +} |
| 27 | + |
| 28 | + |
| 29 | +@dataclass(frozen=True) |
| 30 | +class ControlPlaneBootPlan: |
| 31 | + """Side-effect-free plan derived from a control-plane BootReleaseSet.""" |
| 32 | + |
| 33 | + boot_release_set_id: str |
| 34 | + base_release_set_ref: str |
| 35 | + boot_mode: str |
| 36 | + boot_channel: str |
| 37 | + action: str |
| 38 | + status: str |
| 39 | + policy_ref: str |
| 40 | + platform_entrypoints: list[dict[str, Any]] |
| 41 | + artifact_refs: dict[str, str | None] |
| 42 | + signing: dict[str, Any] |
| 43 | + boot_capabilities: dict[str, Any] |
| 44 | + proof_reports: list[str] |
| 45 | + offline_fallback: dict[str, Any] |
| 46 | + verification_gates: list[str] |
| 47 | + execute: bool = False |
| 48 | + |
| 49 | + def to_dict(self) -> dict[str, Any]: |
| 50 | + return { |
| 51 | + "boot_release_set_id": self.boot_release_set_id, |
| 52 | + "base_release_set_ref": self.base_release_set_ref, |
| 53 | + "boot_mode": self.boot_mode, |
| 54 | + "boot_channel": self.boot_channel, |
| 55 | + "action": self.action, |
| 56 | + "status": self.status, |
| 57 | + "policy_ref": self.policy_ref, |
| 58 | + "platform_entrypoints": self.platform_entrypoints, |
| 59 | + "artifact_refs": self.artifact_refs, |
| 60 | + "signing": self.signing, |
| 61 | + "boot_capabilities": self.boot_capabilities, |
| 62 | + "proof_reports": self.proof_reports, |
| 63 | + "offline_fallback": self.offline_fallback, |
| 64 | + "verification_gates": self.verification_gates, |
| 65 | + "execute": self.execute, |
| 66 | + } |
| 67 | + |
| 68 | + |
| 69 | +class ControlPlaneBootReleaseSetError(ValueError): |
| 70 | + """Raised when a control-plane BootReleaseSet is unsafe or malformed.""" |
| 71 | + |
| 72 | + |
| 73 | +def _require_str(data: dict[str, Any], key: str) -> str: |
| 74 | + value = data.get(key) |
| 75 | + if not isinstance(value, str) or not value.strip(): |
| 76 | + raise ControlPlaneBootReleaseSetError(f"{key} must be a non-empty string") |
| 77 | + return value |
| 78 | + |
| 79 | + |
| 80 | +def _require_dict(data: dict[str, Any], key: str) -> dict[str, Any]: |
| 81 | + value = data.get(key) |
| 82 | + if not isinstance(value, dict): |
| 83 | + raise ControlPlaneBootReleaseSetError(f"{key} must be an object") |
| 84 | + return value |
| 85 | + |
| 86 | + |
| 87 | +def _require_list(data: dict[str, Any], key: str) -> list[Any]: |
| 88 | + value = data.get(key) |
| 89 | + if not isinstance(value, list): |
| 90 | + raise ControlPlaneBootReleaseSetError(f"{key} must be a list") |
| 91 | + return value |
| 92 | + |
| 93 | + |
| 94 | +def _artifact_refs(artifacts: dict[str, Any]) -> dict[str, str | None]: |
| 95 | + required = ["manifest_ref", "kernel_ref", "initrd_ref", "rootfs_ref"] |
| 96 | + refs: dict[str, str | None] = {} |
| 97 | + for key in required: |
| 98 | + refs[key] = _require_str(artifacts, key) |
| 99 | + for key in ["bootloader_ref", "installer_metadata_ref"]: |
| 100 | + value = artifacts.get(key) |
| 101 | + if value is not None and not isinstance(value, str): |
| 102 | + raise ControlPlaneBootReleaseSetError(f"artifacts.{key} must be a string or null") |
| 103 | + refs[key] = value |
| 104 | + return refs |
| 105 | + |
| 106 | + |
| 107 | +def _validate_signing(signing: dict[str, Any]) -> None: |
| 108 | + signature_ref = _require_str(signing, "signature_ref") |
| 109 | + signer_ref = _require_str(signing, "signer_ref") |
| 110 | + signature_algorithm = _require_str(signing, "signature_algorithm") |
| 111 | + manifest_digest = _require_dict(signing, "manifest_digest") |
| 112 | + digest_value = _require_str(manifest_digest, "value") |
| 113 | + if not signature_ref.startswith("urn:srcos:signature:"): |
| 114 | + raise ControlPlaneBootReleaseSetError("signing.signature_ref must be a SourceOS signature URN") |
| 115 | + if not signer_ref.startswith("urn:srcos:key:"): |
| 116 | + raise ControlPlaneBootReleaseSetError("signing.signer_ref must be a SourceOS key URN") |
| 117 | + if signature_algorithm not in {"rsa-pss-sha256", "ed25519", "ecdsa-p256-sha256"}: |
| 118 | + raise ControlPlaneBootReleaseSetError("signing.signature_algorithm is unsupported") |
| 119 | + if not digest_value.startswith(SAFE_DIGEST_PREFIXES): |
| 120 | + raise ControlPlaneBootReleaseSetError("signing.manifest_digest.value must be sha256/sha384/sha512 prefixed") |
| 121 | + |
| 122 | + |
| 123 | +def _verification_gates(plan: ControlPlaneBootPlan) -> list[str]: |
| 124 | + gates = [ |
| 125 | + "verify-boot-release-set-status-ready", |
| 126 | + "verify-manifest-signature", |
| 127 | + "verify-artifact-refs-present", |
| 128 | + "verify-policy-ref-present", |
| 129 | + "verify-proof-reporting-required", |
| 130 | + ] |
| 131 | + if plan.offline_fallback.get("enabled"): |
| 132 | + gates.append("verify-offline-fallback-signature-required") |
| 133 | + if plan.boot_capabilities.get("kexec_allowed") is False: |
| 134 | + gates.append("verify-kexec-denied") |
| 135 | + if plan.boot_capabilities.get("disk_write") in {"installer-scoped", "recovery-scoped"}: |
| 136 | + gates.append("verify-disk-write-scope") |
| 137 | + return gates |
| 138 | + |
| 139 | + |
| 140 | +def build_control_plane_boot_plan(doc: dict[str, Any]) -> ControlPlaneBootPlan: |
| 141 | + """Build a side-effect-free plan from canonical control-plane BootReleaseSet JSON.""" |
| 142 | + |
| 143 | + boot_release_set_id = _require_str(doc, "boot_release_set_id") |
| 144 | + base_release_set_ref = _require_str(doc, "base_release_set_ref") |
| 145 | + boot_mode = _require_str(doc, "boot_mode") |
| 146 | + boot_channel = _require_str(doc, "boot_channel") |
| 147 | + status = _require_str(doc, "status") |
| 148 | + policy_ref = _require_str(doc, "policy_ref") |
| 149 | + |
| 150 | + if status != "ready": |
| 151 | + raise ControlPlaneBootReleaseSetError("BootReleaseSet status must be ready before planning") |
| 152 | + if boot_channel not in BOOT_ACTION_BY_CHANNEL: |
| 153 | + raise ControlPlaneBootReleaseSetError(f"unsupported boot_channel={boot_channel!r}") |
| 154 | + if boot_mode not in {"installer", "recovery", "ephemeral", "bootstrap"}: |
| 155 | + raise ControlPlaneBootReleaseSetError(f"unsupported boot_mode={boot_mode!r}") |
| 156 | + if not boot_release_set_id.startswith("urn:srcos:boot-release-set:"): |
| 157 | + raise ControlPlaneBootReleaseSetError("boot_release_set_id must be a SourceOS BootReleaseSet URN") |
| 158 | + if not base_release_set_ref.startswith("urn:srcos:release-set:"): |
| 159 | + raise ControlPlaneBootReleaseSetError("base_release_set_ref must be a SourceOS ReleaseSet URN") |
| 160 | + if not policy_ref.startswith("urn:srcos:policy:"): |
| 161 | + raise ControlPlaneBootReleaseSetError("policy_ref must be a SourceOS policy URN") |
| 162 | + |
| 163 | + platform_entrypoints = _require_list(doc, "platform_entrypoints") |
| 164 | + if not platform_entrypoints: |
| 165 | + raise ControlPlaneBootReleaseSetError("platform_entrypoints must not be empty") |
| 166 | + for index, entrypoint in enumerate(platform_entrypoints): |
| 167 | + if not isinstance(entrypoint, dict): |
| 168 | + raise ControlPlaneBootReleaseSetError(f"platform_entrypoints[{index}] must be an object") |
| 169 | + _require_str(entrypoint, "platform") |
| 170 | + _require_str(entrypoint, "entrypoint_kind") |
| 171 | + _require_str(entrypoint, "entrypoint_ref") |
| 172 | + |
| 173 | + artifacts = _artifact_refs(_require_dict(doc, "artifacts")) |
| 174 | + signing = _require_dict(doc, "signing") |
| 175 | + _validate_signing(signing) |
| 176 | + |
| 177 | + boot_capabilities = _require_dict(doc, "boot_capabilities") |
| 178 | + offline_fallback = _require_dict(doc, "offline_fallback") |
| 179 | + proof_reporting = _require_dict(doc, "proof_reporting") |
| 180 | + proof_reports = proof_reporting.get("reports") |
| 181 | + if proof_reporting.get("required") is not True: |
| 182 | + raise ControlPlaneBootReleaseSetError("proof_reporting.required must be true") |
| 183 | + if not isinstance(proof_reports, list) or not proof_reports: |
| 184 | + raise ControlPlaneBootReleaseSetError("proof_reporting.reports must be a non-empty list") |
| 185 | + if offline_fallback.get("enabled") and offline_fallback.get("requires_signature_verification") is not True: |
| 186 | + raise ControlPlaneBootReleaseSetError("enabled offline fallback must require signature verification") |
| 187 | + if offline_fallback.get("allows_unsigned_artifacts") is not False: |
| 188 | + raise ControlPlaneBootReleaseSetError("offline fallback must not allow unsigned artifacts") |
| 189 | + |
| 190 | + plan = ControlPlaneBootPlan( |
| 191 | + boot_release_set_id=boot_release_set_id, |
| 192 | + base_release_set_ref=base_release_set_ref, |
| 193 | + boot_mode=boot_mode, |
| 194 | + boot_channel=boot_channel, |
| 195 | + action=BOOT_ACTION_BY_CHANNEL[boot_channel], |
| 196 | + status=status, |
| 197 | + policy_ref=policy_ref, |
| 198 | + platform_entrypoints=platform_entrypoints, |
| 199 | + artifact_refs=artifacts, |
| 200 | + signing=signing, |
| 201 | + boot_capabilities=boot_capabilities, |
| 202 | + proof_reports=[str(item) for item in proof_reports], |
| 203 | + offline_fallback=offline_fallback, |
| 204 | + verification_gates=[], |
| 205 | + execute=False, |
| 206 | + ) |
| 207 | + return ControlPlaneBootPlan( |
| 208 | + boot_release_set_id=plan.boot_release_set_id, |
| 209 | + base_release_set_ref=plan.base_release_set_ref, |
| 210 | + boot_mode=plan.boot_mode, |
| 211 | + boot_channel=plan.boot_channel, |
| 212 | + action=plan.action, |
| 213 | + status=plan.status, |
| 214 | + policy_ref=plan.policy_ref, |
| 215 | + platform_entrypoints=plan.platform_entrypoints, |
| 216 | + artifact_refs=plan.artifact_refs, |
| 217 | + signing=plan.signing, |
| 218 | + boot_capabilities=plan.boot_capabilities, |
| 219 | + proof_reports=plan.proof_reports, |
| 220 | + offline_fallback=plan.offline_fallback, |
| 221 | + verification_gates=_verification_gates(plan), |
| 222 | + execute=False, |
| 223 | + ) |
0 commit comments