e54e0f3a27
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled
1149 lines
42 KiB
Python
1149 lines
42 KiB
Python
#!/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())
|