#!/usr/bin/env python3 """Mirror immutable runtime-cache images and apply their bounded K8s policy.""" from __future__ import annotations import argparse import hashlib import json import re import shlex import subprocess import sys import tarfile import tempfile from dataclasses import dataclass from pathlib import Path from typing import Any, Sequence SCHEMA_VERSION = "awoooi_runtime_image_mirror_policy_v1" RECEIPT_SCHEMA_VERSION = "awoooi_runtime_image_mirror_receipt_v1" STAGING_SCHEMA_VERSION = "awoooi_runtime_image_mirror_staging_policy_v1" STAGING_RECEIPT_SCHEMA_VERSION = "awoooi_runtime_image_mirror_staging_receipt_v1" ALLOWED_KINDS = {"deployment", "daemonset", "statefulset"} ALLOWED_CONTAINER_KINDS = {"container", "init_container"} DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") NAME_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$") TRACE_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{3,160}$") class ControllerError(RuntimeError): """Fail-closed controller error with a public-safe message.""" @dataclass(frozen=True) class Workload: kind: str namespace: str name: str container: str container_kind: str @dataclass(frozen=True) class ImagePolicy: asset_id: str source_index_digest: str platform_digest: str target_registry_digest: str push_ref: str runtime_ref: str workload: Workload @dataclass(frozen=True) class StagingImage: asset_id: str source_index_digest: str platform_digest: str source_config_digest: str push_ref: str runtime_repository: str workload: Workload @dataclass(frozen=True) class Policy: work_item_id: str risk_level: str cache_host: str cache_user: str platform: str images: tuple[ImagePolicy, ...] checksum: str @dataclass(frozen=True) class StagingPolicy: work_item_id: str risk_level: str cache_host: str cache_user: str platform: str runtime_registry_prefix: str images: tuple[StagingImage, ...] checksum: str @dataclass(frozen=True) class RegistryDescriptor: digest: str config_digest: str def _require_text(value: Any, label: str) -> str: if not isinstance(value, str) or not value.strip(): raise ControllerError(f"invalid_{label}") return value.strip() def _require_digest(value: Any, label: str) -> str: digest = _require_text(value, label) if not DIGEST_PATTERN.fullmatch(digest): raise ControllerError(f"invalid_{label}") return digest def _require_name(value: Any, label: str) -> str: name = _require_text(value, label) if not NAME_PATTERN.fullmatch(name): raise ControllerError(f"invalid_{label}") return name def _policy_checksum(payload: dict[str, Any]) -> str: canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() return "sha256:" + hashlib.sha256(canonical).hexdigest() def load_policy(path: Path) -> Policy: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ControllerError("policy_unreadable") from exc if not isinstance(payload, dict) or payload.get("schema_version") != SCHEMA_VERSION: raise ControllerError("policy_schema_invalid") source = payload.get("source_contract") target = payload.get("target_contract") if not isinstance(source, dict) or not isinstance(target, dict): raise ControllerError("policy_contract_missing") if source.get("origin") != "runtime_cache_only": raise ControllerError("runtime_cache_only_required") if source.get("external_pull_allowed") is not False: raise ControllerError("external_pull_must_be_disabled") if source.get("platform") != "linux/amd64": raise ControllerError("unsupported_runtime_platform") if target.get("digest_pin_required") is not True: raise ControllerError("digest_pin_required") push_prefix = _require_text(target.get("push_registry_prefix"), "push_prefix") runtime_prefix = _require_text( target.get("runtime_registry_prefix"), "runtime_prefix" ) if push_prefix != "registry.wooo.work/awoooi/runtime-mirror/": raise ControllerError("push_registry_not_internal") if runtime_prefix != "192.168.0.110:5000/awoooi/runtime-mirror/": raise ControllerError("runtime_registry_not_internal") image_rows = payload.get("images") if not isinstance(image_rows, list) or not image_rows: raise ControllerError("image_policy_empty") images: list[ImagePolicy] = [] asset_ids: set[str] = set() workloads: set[tuple[str, str, str, str, str]] = set() for row in image_rows: if not isinstance(row, dict) or not isinstance(row.get("workload"), dict): raise ControllerError("image_policy_invalid") asset_id = _require_text(row.get("asset_id"), "asset_id") if asset_id in asset_ids: raise ControllerError("duplicate_asset_id") asset_ids.add(asset_id) source_digest = _require_digest( row.get("source_index_digest"), "source_index_digest" ) platform_digest = _require_digest(row.get("platform_digest"), "platform_digest") target_registry_digest = _require_digest( row.get("target_registry_digest"), "target_registry_digest" ) push_ref = _require_text(row.get("push_ref"), "push_ref") runtime_ref = _require_text(row.get("runtime_ref"), "runtime_ref") if not push_ref.startswith(push_prefix) or "@" in push_ref: raise ControllerError("push_ref_invalid") if not runtime_ref.startswith(runtime_prefix): raise ControllerError("runtime_ref_invalid") if not runtime_ref.endswith("@" + target_registry_digest): raise ControllerError("runtime_ref_digest_mismatch") workload_row = row["workload"] kind = _require_text(workload_row.get("kind"), "workload_kind").lower() if kind not in ALLOWED_KINDS: raise ControllerError("workload_kind_not_allowed") workload = Workload( kind=kind, namespace=_require_name(workload_row.get("namespace"), "namespace"), name=_require_name(workload_row.get("name"), "workload_name"), container=_require_name(workload_row.get("container"), "container"), container_kind=_require_text( workload_row.get("container_kind", "container"), "container_kind" ).lower(), ) if workload.container_kind not in ALLOWED_CONTAINER_KINDS: raise ControllerError("container_kind_not_allowed") workload_key = ( workload.kind, workload.namespace, workload.name, workload.container, workload.container_kind, ) if workload_key in workloads: raise ControllerError("duplicate_workload_target") workloads.add(workload_key) images.append( ImagePolicy( asset_id=asset_id, source_index_digest=source_digest, platform_digest=platform_digest, target_registry_digest=target_registry_digest, push_ref=push_ref, runtime_ref=runtime_ref, workload=workload, ) ) return Policy( work_item_id=_require_text(payload.get("work_item_id"), "work_item_id"), risk_level=_require_text(payload.get("risk_level"), "risk_level"), cache_host=_require_text(source.get("cache_host"), "cache_host"), cache_user=_require_name(source.get("cache_user"), "cache_user"), platform="linux/amd64", images=tuple(images), checksum=_policy_checksum(payload), ) def load_staging_policy(path: Path) -> StagingPolicy: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ControllerError("staging_policy_unreadable") from exc if ( not isinstance(payload, dict) or payload.get("schema_version") != STAGING_SCHEMA_VERSION ): raise ControllerError("staging_policy_schema_invalid") source = payload.get("source_contract") target = payload.get("target_contract") if not isinstance(source, dict) or not isinstance(target, dict): raise ControllerError("staging_policy_contract_missing") if source.get("origin") != "runtime_cache_only": raise ControllerError("runtime_cache_only_required") if source.get("external_pull_allowed") is not False: raise ControllerError("external_pull_must_be_disabled") if source.get("platform") != "linux/amd64": raise ControllerError("unsupported_runtime_platform") if target.get("digest_pin_required") is not True: raise ControllerError("digest_pin_required") push_prefix = _require_text(target.get("push_registry_prefix"), "push_prefix") runtime_prefix = _require_text( target.get("runtime_registry_prefix"), "runtime_prefix" ) if push_prefix != "registry.wooo.work/awoooi/runtime-mirror/": raise ControllerError("push_registry_not_internal") if runtime_prefix != "192.168.0.110:5000/awoooi/runtime-mirror/": raise ControllerError("runtime_registry_not_internal") image_rows = payload.get("images") if not isinstance(image_rows, list) or not image_rows: raise ControllerError("staging_image_policy_empty") images: list[StagingImage] = [] asset_ids: set[str] = set() workloads: set[tuple[str, str, str, str, str]] = set() for row in image_rows: if not isinstance(row, dict) or not isinstance(row.get("workload"), dict): raise ControllerError("staging_image_policy_invalid") asset_id = _require_text(row.get("asset_id"), "asset_id") if asset_id in asset_ids: raise ControllerError("duplicate_asset_id") asset_ids.add(asset_id) push_ref = _require_text(row.get("push_ref"), "push_ref") if not push_ref.startswith(push_prefix) or "@" in push_ref: raise ControllerError("push_ref_invalid") push_suffix = push_ref.removeprefix(push_prefix) if ":" not in push_suffix: raise ControllerError("push_ref_tag_required") runtime_repository, push_tag = push_suffix.rsplit(":", 1) if not runtime_repository or any( not NAME_PATTERN.fullmatch(part) for part in runtime_repository.split("/") ): raise ControllerError("runtime_repository_invalid") _require_name(push_tag, "push_tag") workload_row = row["workload"] kind = _require_text(workload_row.get("kind"), "workload_kind").lower() if kind not in ALLOWED_KINDS: raise ControllerError("workload_kind_not_allowed") workload = Workload( kind=kind, namespace=_require_name(workload_row.get("namespace"), "namespace"), name=_require_name(workload_row.get("name"), "workload_name"), container=_require_name(workload_row.get("container"), "container"), container_kind=_require_text( workload_row.get("container_kind", "container"), "container_kind" ).lower(), ) if workload.container_kind not in ALLOWED_CONTAINER_KINDS: raise ControllerError("container_kind_not_allowed") workload_key = ( workload.kind, workload.namespace, workload.name, workload.container, workload.container_kind, ) if workload_key in workloads: raise ControllerError("duplicate_workload_target") workloads.add(workload_key) images.append( StagingImage( asset_id=asset_id, source_index_digest=_require_digest( row.get("source_index_digest"), "source_index_digest" ), platform_digest=_require_digest( row.get("platform_digest"), "platform_digest" ), source_config_digest=_require_digest( row.get("source_config_digest"), "source_config_digest" ), push_ref=push_ref, runtime_repository=runtime_repository, workload=workload, ) ) return StagingPolicy( work_item_id=_require_text(payload.get("work_item_id"), "work_item_id"), risk_level=_require_text(payload.get("risk_level"), "risk_level"), cache_host=_require_text(source.get("cache_host"), "cache_host"), cache_user=_require_name(source.get("cache_user"), "cache_user"), platform="linux/amd64", runtime_registry_prefix=runtime_prefix, images=tuple(images), checksum=_policy_checksum(payload), ) def _run( command: Sequence[str], *, input_text: str | None = None, capture: bool = True, quiet: bool = False, ) -> subprocess.CompletedProcess[str]: stdout: int | None = subprocess.PIPE if capture else None stderr: int | None = subprocess.PIPE if capture else None if quiet: stdout = subprocess.DEVNULL stderr = subprocess.DEVNULL try: completed = subprocess.run( list(command), check=False, text=True, input=input_text, stdout=stdout, stderr=stderr, ) except OSError as exc: raise ControllerError("controlled_command_unavailable") from exc if completed.returncode != 0: raise ControllerError("controlled_command_failed") return completed class RuntimeCache: def __init__( self, policy: Policy | StagingPolicy, ssh_key: Path, known_hosts: Path ) -> None: self.policy = policy self.ssh = [ "ssh", "-i", str(ssh_key), "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=yes", "-o", f"UserKnownHostsFile={known_hosts}", "-o", "ConnectTimeout=10", f"{policy.cache_user}@{policy.cache_host}", ] def _command(self, *args: str) -> list[str]: remote = shlex.join(["sudo", "-n", "k3s", "ctr", *args]) return [*self.ssh, remote] def source_ref(self, expected_digest: str) -> str: output = _run(self._command("images", "list")).stdout or "" matches: list[str] = [] for line in output.splitlines(): parts = line.split() if len(parts) >= 3 and parts[2] == expected_digest: matches.append(parts[0]) if not matches: raise ControllerError("runtime_cache_digest_missing") return sorted(set(matches))[0] def verify_source(self, image: ImagePolicy | StagingImage) -> str: source_ref = self.source_ref(image.source_index_digest) check = _run(self._command("images", "check", f"name=={source_ref}")) if "complete" not in (check.stdout or ""): raise ControllerError("runtime_cache_content_incomplete") content = _run( self._command("content", "get", image.source_index_digest) ).stdout try: manifest = json.loads(content or "{}") except json.JSONDecodeError as exc: raise ControllerError("runtime_cache_index_invalid") from exc selected = [ item.get("digest") for item in manifest.get("manifests", []) if isinstance(item, dict) and item.get("platform", {}).get("os") == "linux" and item.get("platform", {}).get("architecture") == "amd64" ] if selected != [image.platform_digest]: raise ControllerError("runtime_cache_platform_digest_mismatch") return source_ref def source_config_digest(self, platform_digest: str) -> str: content = _run(self._command("content", "get", platform_digest)).stdout try: manifest = json.loads(content or "{}") except json.JSONDecodeError as exc: raise ControllerError("runtime_cache_platform_manifest_invalid") from exc config = manifest.get("config") if not isinstance(config, dict): raise ControllerError("runtime_cache_config_digest_missing") return _require_digest(config.get("digest"), "runtime_cache_config_digest") def export(self, source_ref: str, output: Path) -> None: with output.open("wb") as handle: completed = subprocess.run( self._command( "images", "export", "--platform", self.policy.platform, "-", source_ref, ), check=False, stdout=handle, stderr=subprocess.PIPE, ) if completed.returncode != 0 or output.stat().st_size == 0: raise ControllerError("runtime_cache_export_failed") def _target_repository(push_ref: str) -> str: return push_ref.rsplit(":", 1)[0] def _target_digest_ref(image: ImagePolicy) -> str: return f"{_target_repository(image.push_ref)}@{image.target_registry_digest}" def _registry_digest_present(digest_ref: str) -> bool: try: completed = subprocess.run( [ "docker", "manifest", "inspect", "--insecure", digest_ref, ], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=12, ) except (OSError, subprocess.TimeoutExpired): return False return completed.returncode == 0 def _target_digest_present(image: ImagePolicy) -> bool: return _registry_digest_present(_target_digest_ref(image)) def _registry_descriptor(image_ref: str) -> RegistryDescriptor | None: try: completed = subprocess.run( ["docker", "manifest", "inspect", "--verbose", image_ref], check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=20, ) except (OSError, subprocess.TimeoutExpired) as exc: raise ControllerError("internal_registry_descriptor_unavailable") from exc if completed.returncode != 0: return None try: payload = json.loads(completed.stdout or "{}") except json.JSONDecodeError as exc: raise ControllerError("internal_registry_descriptor_invalid") from exc if not isinstance(payload, dict): raise ControllerError("internal_registry_descriptor_invalid") descriptor = payload.get("Descriptor") manifest = payload.get("SchemaV2Manifest") if not isinstance(descriptor, dict) or not isinstance(manifest, dict): raise ControllerError("internal_registry_descriptor_invalid") config = manifest.get("config") if not isinstance(config, dict): raise ControllerError("internal_registry_descriptor_invalid") return RegistryDescriptor( digest=_require_digest( descriptor.get("digest"), "internal_registry_descriptor_digest" ), config_digest=_require_digest( config.get("digest"), "internal_registry_config_digest" ), ) def _local_config_digest(image_ref: str) -> str: output = _run(["docker", "image", "inspect", "--format={{.Id}}", image_ref]).stdout return _require_digest((output or "").strip(), "local_config_digest") def _local_image_present(image_ref: str) -> bool: try: completed = subprocess.run( ["docker", "image", "inspect", image_ref], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10, ) except (OSError, subprocess.TimeoutExpired): return False return completed.returncode == 0 def _remove_local_image(image_ref: str) -> None: try: subprocess.run( ["docker", "image", "rm", image_ref], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=30, ) except (OSError, subprocess.TimeoutExpired): return def _archive_contains_digest(path: Path, digest: str) -> bool: member = "blobs/sha256/" + digest.removeprefix("sha256:") try: with tarfile.open(path, "r:*") as archive: return member in archive.getnames() except (OSError, tarfile.TarError): return False def mirror_images( policy: Policy, *, trace_id: str, receipt_path: Path, ssh_key: Path, known_hosts: Path, apply: bool, ) -> dict[str, Any]: cache = RuntimeCache(policy, ssh_key, known_hosts) image_receipts: list[dict[str, Any]] = [] missing_count = 0 for image in policy.images: source_ref = cache.verify_source(image) source_fingerprint = "sha256:" + hashlib.sha256(source_ref.encode()).hexdigest() target_present = _target_digest_present(image) execution = "already_present" if not target_present: missing_count += 1 execution = "check_only_missing" if apply: source_preexisting = _local_image_present(source_ref) try: with tempfile.TemporaryDirectory( prefix="awoooi-runtime-image-" ) as temp: archive = Path(temp) / "image.oci.tar" cache.export(source_ref, archive) if not _archive_contains_digest(archive, image.platform_digest): raise ControllerError("runtime_cache_export_digest_missing") _run( ["docker", "load", "--input", str(archive)], quiet=True, ) _run( ["docker", "tag", source_ref, image.push_ref], quiet=True, ) _run(["docker", "push", image.push_ref], quiet=True) finally: _remove_local_image(image.push_ref) if not source_preexisting: _remove_local_image(source_ref) target_present = _target_digest_present(image) if not target_present: raise ControllerError("internal_registry_digest_verifier_failed") execution = "mirrored_from_runtime_cache" image_receipts.append( { "asset_id": image.asset_id, "source_ref_fingerprint": source_fingerprint, "source_index_digest": image.source_index_digest, "platform_digest": image.platform_digest, "target_registry_digest": image.target_registry_digest, "target_digest_ref": _target_digest_ref(image), "source_cache_verified": True, "target_digest_verified": target_present, "execution": execution, } ) verified = all(row["target_digest_verified"] for row in image_receipts) receipt = { "schema_version": RECEIPT_SCHEMA_VERSION, "trace_id": trace_id, "work_item_id": policy.work_item_id, "policy_checksum": policy.checksum, "stage": "runtime_cache_mirror", "mode": "apply" if apply else "check", "external_pull_allowed": False, "source_cache_only": True, "candidate_count": len(image_receipts), "missing_before_count": missing_count, "verified_count": sum( 1 for row in image_receipts if row["target_digest_verified"] ), "state": "verified" if verified else "blocked_with_safe_next_action", "images": image_receipts, } receipt_path.write_text( json.dumps(receipt, ensure_ascii=True, indent=2) + "\n", encoding="utf-8" ) return receipt def stage_images( policy: StagingPolicy, *, trace_id: str, receipt_path: Path, ssh_key: Path, known_hosts: Path, apply: bool, ) -> dict[str, Any]: cache = RuntimeCache(policy, ssh_key, known_hosts) image_receipts: list[dict[str, Any]] = [] missing_count = 0 for image in policy.images: source_ref = cache.verify_source(image) source_fingerprint = "sha256:" + hashlib.sha256(source_ref.encode()).hexdigest() source_config_digest = cache.source_config_digest(image.platform_digest) if source_config_digest != image.source_config_digest: raise ControllerError("runtime_cache_config_digest_mismatch") descriptor = _registry_descriptor(image.push_ref) execution = "already_present" if descriptor is None: missing_count += 1 execution = "check_only_missing" if apply: source_preexisting = _local_image_present(source_ref) try: with tempfile.TemporaryDirectory( prefix="awoooi-runtime-image-stage-" ) as temp: archive = Path(temp) / "image.oci.tar" cache.export(source_ref, archive) if not _archive_contains_digest(archive, image.platform_digest): raise ControllerError("runtime_cache_export_digest_missing") _run( ["docker", "load", "--input", str(archive)], quiet=True, ) if _local_config_digest(source_ref) != source_config_digest: raise ControllerError("local_image_config_digest_mismatch") _run( ["docker", "tag", source_ref, image.push_ref], quiet=True, ) _run(["docker", "push", image.push_ref], quiet=True) finally: _remove_local_image(image.push_ref) if not source_preexisting: _remove_local_image(source_ref) descriptor = _registry_descriptor(image.push_ref) if descriptor is None: raise ControllerError("internal_registry_descriptor_missing") execution = "staged_from_runtime_cache" if descriptor is not None and descriptor.config_digest != source_config_digest: raise ControllerError("internal_registry_provenance_mismatch") target_digest_ref = None runtime_ref = None target_verified = False if descriptor is not None: target_digest_ref = ( f"{_target_repository(image.push_ref)}@{descriptor.digest}" ) runtime_ref = ( f"{policy.runtime_registry_prefix}{image.runtime_repository}" f"@{descriptor.digest}" ) target_verified = _registry_digest_present(target_digest_ref) if apply and not target_verified: raise ControllerError("internal_registry_digest_verifier_failed") image_receipts.append( { "asset_id": image.asset_id, "source_ref_fingerprint": source_fingerprint, "source_index_digest": image.source_index_digest, "platform_digest": image.platform_digest, "source_config_digest": source_config_digest, "target_registry_digest": ( descriptor.digest if descriptor is not None else None ), "target_digest_ref": target_digest_ref, "runtime_ref": runtime_ref, "source_cache_verified": True, "target_digest_verified": target_verified, "provenance_config_digest_match": descriptor is not None, "execution": execution, "workload": { "kind": image.workload.kind, "namespace": image.workload.namespace, "name": image.workload.name, "container": image.workload.container, "container_kind": image.workload.container_kind, }, } ) verified = all(row["target_digest_verified"] for row in image_receipts) receipt = { "schema_version": STAGING_RECEIPT_SCHEMA_VERSION, "trace_id": trace_id, "work_item_id": policy.work_item_id, "policy_checksum": policy.checksum, "stage": "runtime_cache_staging", "mode": "apply" if apply else "check", "risk_level": policy.risk_level, "external_pull_allowed": False, "source_cache_only": True, "candidate_count": len(image_receipts), "missing_before_count": missing_count, "verified_count": sum( 1 for row in image_receipts if row["target_digest_verified"] ), "state": "verified" if verified else "blocked_with_safe_next_action", "images": image_receipts, } receipt_path.write_text( json.dumps(receipt, ensure_ascii=True, indent=2) + "\n", encoding="utf-8" ) return receipt class Kubernetes: def __init__( self, *, use_sudo: bool, kubeconfig: str | None, server: str | None, ) -> None: command = ["kubectl"] if use_sudo: command = ["sudo", "-n", "kubectl"] if kubeconfig: command.extend(["--kubeconfig", kubeconfig]) if server: command.extend(["--server", server]) self.command = command def run( self, args: Sequence[str], *, input_text: str | None = None ) -> subprocess.CompletedProcess[str]: return _run([*self.command, *args], input_text=input_text) def workload(self, item: ImagePolicy) -> dict[str, Any]: output = self.run( [ "get", item.workload.kind, item.workload.name, "-n", item.workload.namespace, "-o", "json", ] ).stdout try: return json.loads(output or "{}") except json.JSONDecodeError as exc: raise ControllerError("workload_readback_invalid") from exc @staticmethod def _spec_container_field(item: ImagePolicy) -> str: if item.workload.container_kind == "init_container": return "initContainers" return "containers" @staticmethod def _status_container_field(item: ImagePolicy) -> str: if item.workload.container_kind == "init_container": return "initContainerStatuses" return "containerStatuses" def current_image(self, item: ImagePolicy) -> str: workload = self.workload(item) container_field = self._spec_container_field(item) containers = ( workload.get("spec", {}) .get("template", {}) .get("spec", {}) .get(container_field, []) ) values = [ row.get("image") for row in containers if isinstance(row, dict) and row.get("name") == item.workload.container ] if len(values) != 1 or not isinstance(values[0], str): raise ControllerError("workload_container_not_found") return values[0] def patch_image(self, item: ImagePolicy, image_ref: str, *, dry_run: bool) -> None: container_field = self._spec_container_field(item) patch = json.dumps( { "spec": { "template": { "spec": { container_field: [ {"name": item.workload.container, "image": image_ref} ] } } } }, separators=(",", ":"), ) args = [ "patch", item.workload.kind, item.workload.name, "-n", item.workload.namespace, "--type=strategic", "--patch", patch, ] if dry_run: args.extend(["--dry-run=server", "-o", "name"]) self.run(args) def rollout(self, item: ImagePolicy) -> None: self.run( [ "rollout", "status", f"{item.workload.kind}/{item.workload.name}", "-n", item.workload.namespace, "--timeout=240s", ] ) def verify_pods(self, item: ImagePolicy) -> int: workload = self.workload(item) labels = workload.get("spec", {}).get("selector", {}).get("matchLabels", {}) if not isinstance(labels, dict) or not labels: raise ControllerError("workload_selector_missing") selector = ",".join(f"{key}={labels[key]}" for key in sorted(labels)) output = self.run( [ "get", "pods", "-n", item.workload.namespace, "-l", selector, "-o", "json", ] ).stdout try: pods = json.loads(output or "{}").get("items", []) except json.JSONDecodeError as exc: raise ControllerError("pod_readback_invalid") from exc active = [ pod for pod in pods if isinstance(pod, dict) and not pod.get("metadata", {}).get("deletionTimestamp") ] if not active: raise ControllerError("workload_pods_missing") verified = 0 for pod in active: status_field = self._status_container_field(item) statuses = pod.get("status", {}).get(status_field, []) matches = [ status for status in statuses if isinstance(status, dict) and status.get("name") == item.workload.container ] if len(matches) != 1: raise ControllerError("pod_container_status_missing") status = matches[0] if item.workload.container_kind == "init_container": terminated = status.get("state", {}).get("terminated", {}) completed = status.get("ready") is True or ( isinstance(terminated, dict) and terminated.get("exitCode") == 0 ) if not completed: raise ControllerError("pod_init_container_not_completed") elif status.get("ready") is not True: raise ControllerError("pod_container_not_ready") image_id = status.get("imageID") if not isinstance(image_id, str) or not image_id.endswith( "@" + item.target_registry_digest ): raise ControllerError("pod_image_digest_mismatch") verified += 1 return verified def write_receipt(self, receipt: dict[str, Any]) -> None: compact = json.dumps(receipt, sort_keys=True, separators=(",", ":")) manifest = self.run( [ "create", "configmap", "awoooi-runtime-image-mirror-receipt", "-n", "awoooi-prod", f"--from-literal=receipt.json={compact}", "--dry-run=client", "-o", "json", ] ).stdout self.run(["apply", "-f", "-"], input_text=manifest) def _read_mirror_receipt(path: Path, policy: Policy, trace_id: str) -> dict[str, Any]: try: receipt = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ControllerError("mirror_receipt_unreadable") from exc if ( not isinstance(receipt, dict) or receipt.get("schema_version") != RECEIPT_SCHEMA_VERSION or receipt.get("trace_id") != trace_id or receipt.get("work_item_id") != policy.work_item_id or receipt.get("policy_checksum") != policy.checksum or receipt.get("state") != "verified" or receipt.get("verified_count") != len(policy.images) ): raise ControllerError("mirror_receipt_not_verified") return receipt def apply_policy( policy: Policy, *, trace_id: str, mirror_receipt_path: Path, apply: bool, use_sudo: bool, kubeconfig: str | None, server: str | None, ) -> dict[str, Any]: mirror_receipt = _read_mirror_receipt(mirror_receipt_path, policy, trace_id) kubernetes = Kubernetes(use_sudo=use_sudo, kubeconfig=kubeconfig, server=server) previous: dict[str, str] = {} changed: list[ImagePolicy] = [] for item in policy.images: previous[item.asset_id] = kubernetes.current_image(item) kubernetes.patch_image(item, item.runtime_ref, dry_run=True) if apply: try: for item in policy.images: if previous[item.asset_id] == item.runtime_ref: continue kubernetes.patch_image(item, item.runtime_ref, dry_run=False) changed.append(item) for item in policy.images: kubernetes.rollout(item) except ControllerError: for item in reversed(changed): try: kubernetes.patch_image(item, previous[item.asset_id], dry_run=False) kubernetes.rollout(item) except ControllerError: pass raise pod_count = 0 compliant_count = 0 for item in policy.images: if kubernetes.current_image(item) == item.runtime_ref: compliant_count += 1 pod_count += kubernetes.verify_pods(item) verified = compliant_count == len(policy.images) receipt = { "schema_version": RECEIPT_SCHEMA_VERSION, "trace_id": trace_id, "work_item_id": policy.work_item_id, "policy_checksum": policy.checksum, "stage": "runtime_policy_apply", "mode": "apply" if apply else "check", "risk_level": policy.risk_level, "sensor_receipt_present": True, "normalized_asset_count": len(policy.images), "source_of_truth_diff_count": sum( 1 for item in policy.images if previous[item.asset_id] != item.runtime_ref ), "ai_candidate_action": "replace_forbidden_registry_with_internal_digest", "policy_decision": "controlled_apply_allowed", "server_dry_run_passed": True, "execution_changed_count": len(changed), "independent_verifier_passed": verified, "verified_workload_count": compliant_count, "verified_pod_count": pod_count, "rollback_performed": False, "external_pull_allowed": False, "mirror_stage_verified": mirror_receipt.get("state") == "verified", "durable_writeback": "kubernetes_configmap", "state": "verified" if verified else "blocked_with_safe_next_action", } if apply and verified: kubernetes.write_receipt(receipt) return receipt def _trace_id(value: str) -> str: if not TRACE_PATTERN.fullmatch(value): raise argparse.ArgumentTypeError("trace_id must be public-safe") return value def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser() parser.add_argument("--policy", type=Path, required=True) parser.add_argument("--trace-id", type=_trace_id, required=True) subparsers = parser.add_subparsers(dest="command", required=True) mirror = subparsers.add_parser("mirror") mirror.add_argument("--apply", action="store_true") mirror.add_argument("--receipt", type=Path, required=True) mirror.add_argument("--ssh-key", type=Path, required=True) mirror.add_argument("--known-hosts", type=Path, required=True) stage = subparsers.add_parser("stage") stage.add_argument("--apply", action="store_true") stage.add_argument("--receipt", type=Path, required=True) stage.add_argument("--ssh-key", type=Path, required=True) stage.add_argument("--known-hosts", type=Path, required=True) apply_parser = subparsers.add_parser("apply") apply_parser.add_argument("--apply", action="store_true") apply_parser.add_argument("--mirror-receipt", type=Path, required=True) apply_parser.add_argument("--kubectl-sudo", action="store_true") apply_parser.add_argument("--kubeconfig") apply_parser.add_argument("--server") return parser def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) try: if args.command == "stage": staging_policy = load_staging_policy(args.policy) receipt = stage_images( staging_policy, trace_id=args.trace_id, receipt_path=args.receipt, ssh_key=args.ssh_key, known_hosts=args.known_hosts, apply=args.apply, ) else: policy = load_policy(args.policy) if args.command == "mirror": receipt = mirror_images( policy, trace_id=args.trace_id, receipt_path=args.receipt, ssh_key=args.ssh_key, known_hosts=args.known_hosts, apply=args.apply, ) elif args.command == "apply": receipt = apply_policy( policy, trace_id=args.trace_id, mirror_receipt_path=args.mirror_receipt, apply=args.apply, use_sudo=args.kubectl_sudo, kubeconfig=args.kubeconfig, server=args.server, ) print(json.dumps(receipt, sort_keys=True, separators=(",", ":"))) return 0 if receipt.get("state") == "verified" else 2 except ControllerError as exc: print( json.dumps( { "schema_version": ( STAGING_RECEIPT_SCHEMA_VERSION if args.command == "stage" else RECEIPT_SCHEMA_VERSION ), "state": "blocked_with_safe_next_action", "error_code": str(exc), "secret_value_exposed": False, "external_pull_attempted": False, }, sort_keys=True, separators=(",", ":"), ), file=sys.stderr, ) return 1 if __name__ == "__main__": raise SystemExit(main())