#!/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" ALLOWED_KINDS = {"deployment", "daemonset", "statefulset"} 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 @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 Policy: work_item_id: str risk_level: str cache_host: str cache_user: str platform: str images: tuple[ImagePolicy, ...] checksum: 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]] = 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"), ) workload_key = ( workload.kind, workload.namespace, workload.name, workload.container, ) 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 _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, 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) -> 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 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 _target_digest_present(image: ImagePolicy) -> bool: try: completed = subprocess.run( [ "docker", "manifest", "inspect", "--insecure", _target_digest_ref(image), ], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=12, ) except (OSError, subprocess.TimeoutExpired): return False return completed.returncode == 0 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 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 def current_image(self, item: ImagePolicy) -> str: workload = self.workload(item) containers = ( workload.get("spec", {}) .get("template", {}) .get("spec", {}) .get("containers", []) ) 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: patch = json.dumps( { "spec": { "template": { "spec": { "containers": [ {"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: statuses = pod.get("status", {}).get("containerStatuses", []) 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 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) 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: 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, ) else: 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": 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())