diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index ef0f9a6ed..5009bbb42 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -1647,6 +1647,20 @@ jobs: echo "BLOCKER runtime_image_cache_host_key_unavailable" exit 1 } + STAGING_TRACE_ID="aia-p0-006-02b-${GITHUB_RUN_ID:-${GITEA_RUN_NUMBER:-unknown}}" + STAGING_RECEIPT_PATH="/tmp/awoooi-runtime-image-staging-receipt.json" + rm -f "${STAGING_RECEIPT_PATH}" + python3 scripts/security/runtime_image_mirror_controller.py \ + --policy config/security/runtime-image-mirror-staging-policy.json \ + --trace-id "${STAGING_TRACE_ID}" \ + stage \ + --apply \ + --receipt "${STAGING_RECEIPT_PATH}" \ + --ssh-key "${HOME}/.ssh/deploy_key" \ + --known-hosts "${CACHE_KNOWN_HOSTS}" + test -s "${STAGING_RECEIPT_PATH}" + echo "runtime_image_staging_stage=verified" + TRACE_ID="aia-p0-006-02a-${GITHUB_RUN_ID:-${GITEA_RUN_NUMBER:-unknown}}" RECEIPT_PATH="/tmp/awoooi-runtime-image-mirror-receipt.json" rm -f "${RECEIPT_PATH}" diff --git a/config/security/runtime-image-mirror-staging-policy.json b/config/security/runtime-image-mirror-staging-policy.json new file mode 100644 index 000000000..34b373a88 --- /dev/null +++ b/config/security/runtime-image-mirror-staging-policy.json @@ -0,0 +1,33 @@ +{ + "schema_version": "awoooi_runtime_image_mirror_staging_policy_v1", + "work_item_id": "AIA-P0-006-02B", + "risk_level": "high", + "source_contract": { + "origin": "runtime_cache_only", + "external_pull_allowed": false, + "cache_host": "192.168.0.120", + "cache_user": "wooo", + "platform": "linux/amd64" + }, + "target_contract": { + "push_registry_prefix": "registry.wooo.work/awoooi/runtime-mirror/", + "runtime_registry_prefix": "192.168.0.110:5000/awoooi/runtime-mirror/", + "digest_pin_required": true + }, + "images": [ + { + "asset_id": "runtime-image:argocd-v3.3.6-canary", + "source_index_digest": "sha256:16b92ba472fbb9287459cc52e0ecff07288dff461209955098edb56ce866fe49", + "platform_digest": "sha256:2f25a42949ea69c0dd33f4ce1918c6a01039d6c14a7ecc1d19088504a9d3e94f", + "source_config_digest": "sha256:86219ff1fefea5c8e09a4791d1093f06fca7ac8b1b2d1c56f94d3cb560a77603", + "push_ref": "registry.wooo.work/awoooi/runtime-mirror/argocd:v3.3.6-linux-amd64", + "workload": { + "kind": "deployment", + "namespace": "argocd", + "name": "argocd-dex-server", + "container": "copyutil", + "container_kind": "init_container" + } + } + ] +} diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index 24bac02c4..3b44746f9 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -160,12 +160,12 @@ spec: - name: AWOOOI_BUILD_COMMIT_SHA # 2026-06-29 Codex: CD rewrites this to the deployed image tag so # production deploy readback does not rely on a stale static snapshot. - value: "3de0ba821e1414842fdfd3a3813c3591efc9684a" + value: "af03e863cb0737737542bf90b941eceed84d703d" - name: AWOOOI_DESIRED_API_IMAGE_TAG # 2026-06-30 Codex: CD rewrites this alongside AWOOOI_BUILD_COMMIT_SHA. # Production readback compares runtime image truth against this # GitOps desired tag instead of doing a slow Gitea raw fetch. - value: "3de0ba821e1414842fdfd3a3813c3591efc9684a" + value: "af03e863cb0737737542bf90b941eceed84d703d" - name: DATABASE_POOL_SIZE # 2026-07-01 Codex: production role `awoooi` currently has a low # connection limit. Keep API pool conservative until DB role diff --git a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml index 5f87e04a4..b020aa2fb 100644 --- a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml +++ b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml @@ -66,7 +66,7 @@ spec: - name: DATABASE_NULL_POOL value: "true" - name: AWOOOI_BUILD_COMMIT_SHA - value: "3de0ba821e1414842fdfd3a3813c3591efc9684a" + value: "af03e863cb0737737542bf90b941eceed84d703d" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "false" - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index 5a256b340..d3daba937 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -173,7 +173,7 @@ spec: - name: DATABASE_BOOTSTRAP_DDL_ENABLED value: "false" - name: AWOOOI_BUILD_COMMIT_SHA - value: "3de0ba821e1414842fdfd3a3813c3591efc9684a" + value: "af03e863cb0737737542bf90b941eceed84d703d" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "true" - name: ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER diff --git a/k8s/awoooi-prod/kustomization.yaml b/k8s/awoooi-prod/kustomization.yaml index 26d44ee0c..fae68ff4f 100644 --- a/k8s/awoooi-prod/kustomization.yaml +++ b/k8s/awoooi-prod/kustomization.yaml @@ -40,9 +40,9 @@ resources: # ⚠️ 重要: name 必須與 deployment YAML 中的 image 完全匹配 (含 tag) # newName + newTag 由 CI 透過 kustomize edit set image 注入 images: -- digest: sha256:8c45722e1162540ca88d705e25fe9da706c3feb6f89994f548cca47c04e460d1 +- digest: sha256:5295557b830877fcf8e93506bc7790679b9b8ef347b3144fa5a3557e5f26d7f3 name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api -- digest: sha256:04a87a039743115f2db48951a78431bc446fcff1b7259789808cd8674af10c4e +- digest: sha256:d367bfc1c7ab31fecf6fb147b0da572627f74d88267c206c0e9c99a519b956c9 name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web diff --git a/scripts/reboot-recovery/deploy-agent99-via-windows99-ssh.sh b/scripts/reboot-recovery/deploy-agent99-via-windows99-ssh.sh index 5f4b18d5d..76b773ff0 100755 --- a/scripts/reboot-recovery/deploy-agent99-via-windows99-ssh.sh +++ b/scripts/reboot-recovery/deploy-agent99-via-windows99-ssh.sh @@ -417,116 +417,18 @@ from pathlib import Path envelope = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) source_revision = envelope["sourceRevision"] manifest_sha256 = envelope["manifestSha256"] -fixed_names = [row["name"] for row in envelope["files"]] -fixed_rows = ",\n ".join("'" + name.replace("'", "''") + "'" for name in fixed_names) stage_identity = f"check||||{source_revision}" stage_token = hashlib.sha256(stage_identity.encode("utf-8")).hexdigest()[:20] -script = r'''$ErrorActionPreference = "Stop" -$agentRoot = "C:\Wooo\Agent99" -$binDir = Join-Path $agentRoot "bin" -$manifestPath = Join-Path $agentRoot "state\runtime-manifest.json" -$fixedFiles = @( - __FIXED_ROWS__ -) -$current = [pscustomobject]@{ - exists = $false - sourceRevision = "" - runtimeMatched = $false - recordedRuntimeMatched = $false - fileCount = 0 - mismatchCount = -1 - recordedMismatchCount = -1 - parseError = "" -} -if (Test-Path -LiteralPath $manifestPath -PathType Leaf) { - try { - $manifest = Get-Content -LiteralPath $manifestPath -Raw | ConvertFrom-Json - $rows = @($manifest.files) - $byName = @{} - $shapeValid = [bool]($rows.Count -eq $fixedFiles.Count) - foreach ($row in $rows) { - $name = [string]$row.name - if (-not $name -or $byName.ContainsKey($name)) { - $shapeValid = $false - continue - } - $byName[$name] = $row - } - $actualMismatchCount = 0 - foreach ($name in $fixedFiles) { - if (-not $byName.ContainsKey($name)) { - $shapeValid = $false - $actualMismatchCount += 1 - continue - } - $recordedSha = ([string]$byName[$name].runtimeSha256).ToLowerInvariant() - $runtimePath = Join-Path $binDir $name - $actualSha = if (Test-Path -LiteralPath $runtimePath -PathType Leaf) { - (Get-FileHash -LiteralPath $runtimePath -Algorithm SHA256).Hash.ToLowerInvariant() - } else { - "missing" - } - if (-not $recordedSha -or $recordedSha -ne $actualSha) { - $actualMismatchCount += 1 - } - } - $runtimeMatched = [bool]( - $shapeValid -and - [bool]$manifest.runtimeMatched -and - [int]$manifest.fileCount -eq $fixedFiles.Count -and - [int]$manifest.mismatchCount -eq 0 -and - $actualMismatchCount -eq 0 - ) - $current = [pscustomobject]@{ - exists = $true - sourceRevision = [string]$manifest.sourceRevision - runtimeMatched = $runtimeMatched - recordedRuntimeMatched = [bool]$manifest.runtimeMatched - fileCount = [int]$manifest.fileCount - mismatchCount = $actualMismatchCount - recordedMismatchCount = [int]$manifest.mismatchCount - parseError = "" - } - } catch { - $current.exists = $true - $current.parseError = $_.Exception.GetType().Name - } -} -[pscustomobject]@{ - schemaVersion = "agent99_remote_atomic_deploy_receipt_v1" - status = "check_ready" - mode = "check" - sourceRevision = "__SOURCE_REVISION__" - expectedRuntimeFileCount = 14 - manifestSha256 = "__MANIFEST_SHA256__" - plannedStagingPath = "C:\Wooo\Agent99\deploy\remote-source-__STAGE_TOKEN__" - runtimeManifest = $current - livePreflight = [pscustomobject]@{ executed = $false; reason = "apply_only_after_validate_only" } - transportPayloadMode = "control_command_no_stdin" - transientTransportPayloadStored = $false - transientTransportPayloadDeletedBeforeReceiver = $false - transportPayloadContainsSecrets = $false - operationBoundaries = [pscustomobject]@{ - remoteWritePerformed = $false - liveRuntimeWritePerformed = $false - livePromotionAttempted = $false - livePromotionPerformed = $false - secretValueRead = $false - privateKeyValueRead = $false - tokenValueRead = $false - environmentSecretRead = $false - uiInteraction = $false - vmPowerChange = $false - hostReboot = $false - serviceRestart = $false - scheduledTaskRestart = $false - scheduledTaskModification = $false - } - nextSafeAction = "rerun_with_apply_and_trace_run_work_item_after_source_commit_and_transport_check" -} | ConvertTo-Json -Compress -Depth 8''' +script = r'''$ErrorActionPreference='Stop';$ProgressPreference='SilentlyContinue' +$a='C:\Wooo\Agent99';$b=Join-Path $a 'bin';$p=Join-Path $a 'state\runtime-manifest.json' +$r=[ordered]@{exists=$false;sourceRevision='';runtimeMatched=$false;recordedRuntimeMatched=$false;fileCount=0;mismatchCount=-1;recordedMismatchCount=-1;parseError=''} +if(Test-Path -LiteralPath $p -PathType Leaf){try{$m=Get-Content -LiteralPath $p -Raw|ConvertFrom-Json;$rows=@($m.files);$seen=@{};$bad=0;foreach($x in $rows){$n=[string]$x.name;if(!$n-or[IO.Path]::GetFileName($n)-ne$n-or$seen.ContainsKey($n)){$bad++;continue};$seen[$n]=1;$q=Join-Path $b $n;$h=if(Test-Path -LiteralPath $q -PathType Leaf){(Get-FileHash -LiteralPath $q -Algorithm SHA256).Hash.ToLowerInvariant()}else{'missing'};if(([string]$x.runtimeSha256).ToLowerInvariant()-ne$h){$bad++}};$r.exists=$true;$r.sourceRevision=[string]$m.sourceRevision;$r.recordedRuntimeMatched=[bool]$m.runtimeMatched;$r.fileCount=[int]$m.fileCount;$r.mismatchCount=$bad;$r.recordedMismatchCount=[int]$m.mismatchCount;$r.runtimeMatched=[bool]($m.runtimeMatched-and$m.fileCount-eq 14-and$m.mismatchCount-eq 0-and$rows.Count-eq 14-and$seen.Count-eq 14-and$bad-eq 0)}catch{$r.exists=$true;$r.parseError=$_.Exception.GetType().Name}} +$c=[ordered]@{executed=$false;ok=$false;exitCode=-1;errorType=''} +try{& powershell.exe -NoProfile -NonInteractive -ExecutionPolicy Bypass -File (Join-Path $b 'agent99-contract-check.ps1') -SourceRoot $b 2>$null|Out-Null;$c.executed=$true;$c.exitCode=$LASTEXITCODE;$c.ok=[bool]($c.exitCode-eq 0)}catch{$c.executed=$true;$c.errorType=$_.Exception.GetType().Name} +$z=[ordered]@{};foreach($n in 'remoteWritePerformed,liveRuntimeWritePerformed,livePromotionAttempted,livePromotionPerformed,secretValueRead,privateKeyValueRead,tokenValueRead,environmentSecretRead,uiInteraction,vmPowerChange,hostReboot,serviceRestart,scheduledTaskRestart,scheduledTaskModification'.Split(',')){$z[$n]=$false} +[ordered]@{schemaVersion='agent99_remote_atomic_deploy_receipt_v1';status='check_ready';mode='check';sourceRevision='__SOURCE_REVISION__';expectedRuntimeFileCount=14;manifestSha256='__MANIFEST_SHA256__';plannedStagingPath='C:\Wooo\Agent99\deploy\remote-source-__STAGE_TOKEN__';runtimeManifest=[pscustomobject]$r;runtimeContract=[pscustomobject]$c;livePreflight=[ordered]@{executed=$false;reason='apply_only_after_validate_only'};transportPayloadMode='control_command_no_stdin';transientTransportPayloadStored=$false;transientTransportPayloadDeletedBeforeReceiver=$false;transportPayloadContainsSecrets=$false;operationBoundaries=$z;nextSafeAction='rerun_with_apply_and_trace_run_work_item_after_source_commit_and_transport_check'}|ConvertTo-Json -Compress -Depth 8''' script = ( - script.replace("__FIXED_ROWS__", fixed_rows) - .replace("__SOURCE_REVISION__", source_revision) + script.replace("__SOURCE_REVISION__", source_revision) .replace("__MANIFEST_SHA256__", manifest_sha256) .replace("__STAGE_TOKEN__", stage_token) ) diff --git a/scripts/reboot-recovery/tests/test_agent99_remote_atomic_deploy_transport_contract.py b/scripts/reboot-recovery/tests/test_agent99_remote_atomic_deploy_transport_contract.py index cca4b9e73..522ec2a0b 100644 --- a/scripts/reboot-recovery/tests/test_agent99_remote_atomic_deploy_transport_contract.py +++ b/scripts/reboot-recovery/tests/test_agent99_remote_atomic_deploy_transport_contract.py @@ -1,5 +1,6 @@ from __future__ import annotations +import base64 import os import re import shlex @@ -658,11 +659,31 @@ def test_check_mode_uses_bounded_control_command_without_ssh_stdin( assert result.returncode == 0, result.stderr assert '"status":"check_ready"' in result.stdout args = args_path.read_text(encoding="utf-8") + remote_command = args.splitlines()[-1] assert "BatchMode=yes" in args assert "StrictHostKeyChecking=yes" in args assert "PasswordAuthentication=no" in args assert "Administrator@192.168.0.99" in args assert str(identity) in args + assert len(remote_command) < 8191 + encoded_command = remote_command.split("-EncodedCommand ", 1)[1] + check_script = base64.b64decode(encoded_command).decode("utf-16le") + assert "agent99-contract-check.ps1" in check_script + assert "Get-FileHash" in check_script + assert "runtimeContract" in check_script + assert check_script.count("ProgressPreference") == 1 + assert "remoteWritePerformed" in check_script + assert "$z[$n]=$false" in check_script + for write_command in ( + "New-Item", + "Set-Content", + "Add-Content", + "Out-File", + "Remove-Item", + "Move-Item", + "Copy-Item", + ): + assert write_command not in check_script stdin_bytes = int(bytes_path.read_text(encoding="utf-8")) assert stdin_bytes == 0 diff --git a/scripts/security/runtime_image_mirror_controller.py b/scripts/security/runtime_image_mirror_controller.py index 7565356cd..6438bf040 100644 --- a/scripts/security/runtime_image_mirror_controller.py +++ b/scripts/security/runtime_image_mirror_controller.py @@ -19,7 +19,10 @@ 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}$") @@ -35,6 +38,7 @@ class Workload: namespace: str name: str container: str + container_kind: str @dataclass(frozen=True) @@ -48,6 +52,17 @@ class ImagePolicy: 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 @@ -59,6 +74,24 @@ class Policy: 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}") @@ -119,7 +152,7 @@ def load_policy(path: Path) -> Policy: raise ControllerError("image_policy_empty") images: list[ImagePolicy] = [] asset_ids: set[str] = set() - workloads: set[tuple[str, str, str, 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") @@ -152,12 +185,18 @@ def load_policy(path: Path) -> Policy: 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") @@ -185,6 +224,121 @@ def load_policy(path: Path) -> Policy: ) +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], *, @@ -214,7 +368,9 @@ def _run( class RuntimeCache: - def __init__(self, policy: Policy, ssh_key: Path, known_hosts: Path) -> None: + def __init__( + self, policy: Policy | StagingPolicy, ssh_key: Path, known_hosts: Path + ) -> None: self.policy = policy self.ssh = [ "ssh", @@ -246,7 +402,7 @@ class RuntimeCache: raise ControllerError("runtime_cache_digest_missing") return sorted(set(matches))[0] - def verify_source(self, image: ImagePolicy) -> str: + 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 ""): @@ -269,6 +425,17 @@ class RuntimeCache: 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( @@ -296,7 +463,7 @@ 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: +def _registry_digest_present(digest_ref: str) -> bool: try: completed = subprocess.run( [ @@ -304,7 +471,7 @@ def _target_digest_present(image: ImagePolicy) -> bool: "manifest", "inspect", "--insecure", - _target_digest_ref(image), + digest_ref, ], check=False, stdout=subprocess.DEVNULL, @@ -316,6 +483,52 @@ def _target_digest_present(image: ImagePolicy) -> bool: 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( @@ -437,6 +650,127 @@ def mirror_images( 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, @@ -476,13 +810,26 @@ class Kubernetes: 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("containers", []) + .get(container_field, []) ) values = [ row.get("image") @@ -494,12 +841,13 @@ class Kubernetes: 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": { - "containers": [ + container_field: [ {"name": item.workload.container, "image": image_ref} ] } @@ -566,7 +914,8 @@ class Kubernetes: raise ControllerError("workload_pods_missing") verified = 0 for pod in active: - statuses = pod.get("status", {}).get("containerStatuses", []) + status_field = self._status_container_field(item) + statuses = pod.get("status", {}).get(status_field, []) matches = [ status for status in statuses @@ -576,7 +925,14 @@ class Kubernetes: if len(matches) != 1: raise ControllerError("pod_container_status_missing") status = matches[0] - if status.get("ready") is not True: + 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( @@ -715,6 +1071,12 @@ def build_parser() -> argparse.ArgumentParser: 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) @@ -727,7 +1089,18 @@ def build_parser() -> argparse.ArgumentParser: def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) try: - policy = load_policy(args.policy) + 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, @@ -737,7 +1110,7 @@ def main(argv: Sequence[str] | None = None) -> int: known_hosts=args.known_hosts, apply=args.apply, ) - else: + elif args.command == "apply": receipt = apply_policy( policy, trace_id=args.trace_id, @@ -753,7 +1126,11 @@ def main(argv: Sequence[str] | None = None) -> int: print( json.dumps( { - "schema_version": RECEIPT_SCHEMA_VERSION, + "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, diff --git a/scripts/security/tests/test_runtime_image_mirror_controller.py b/scripts/security/tests/test_runtime_image_mirror_controller.py index e3a715c14..0a5e9b375 100644 --- a/scripts/security/tests/test_runtime_image_mirror_controller.py +++ b/scripts/security/tests/test_runtime_image_mirror_controller.py @@ -14,6 +14,7 @@ from unittest.mock import patch ROOT = Path(__file__).resolve().parents[3] CONTROLLER_PATH = ROOT / "scripts/security/runtime_image_mirror_controller.py" POLICY_PATH = ROOT / "config/security/runtime-image-mirror-policy.json" +STAGING_POLICY_PATH = ROOT / "config/security/runtime-image-mirror-staging-policy.json" def _load_controller(): @@ -31,6 +32,15 @@ controller = _load_controller() class RuntimeImageMirrorControllerTest(unittest.TestCase): + def _image_with_container_kind(self, container_kind: str): + payload = json.loads(POLICY_PATH.read_text(encoding="utf-8")) + payload["images"] = [payload["images"][0]] + payload["images"][0]["workload"]["container_kind"] = container_kind + with tempfile.TemporaryDirectory() as temp: + path = Path(temp) / "policy.json" + path.write_text(json.dumps(payload), encoding="utf-8") + return controller.load_policy(path).images[0] + def test_policy_is_internal_digest_pinned_and_runtime_cache_only(self) -> None: policy = controller.load_policy(POLICY_PATH) raw = POLICY_PATH.read_text(encoding="utf-8") @@ -61,6 +71,37 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): for image in policy.images ) ) + self.assertTrue( + all(image.workload.container_kind == "container" for image in policy.images) + ) + + def test_policy_loads_and_validates_init_container_kind(self) -> None: + image = self._image_with_container_kind("init_container") + + self.assertEqual(image.workload.container_kind, "init_container") + + with self.assertRaisesRegex( + controller.ControllerError, "container_kind_not_allowed" + ): + self._image_with_container_kind("ephemeral_container") + + def test_staging_policy_is_runtime_cache_only_and_init_canary_scoped( + self, + ) -> None: + policy = controller.load_staging_policy(STAGING_POLICY_PATH) + raw = STAGING_POLICY_PATH.read_text(encoding="utf-8") + + self.assertEqual(policy.work_item_id, "AIA-P0-006-02B") + self.assertEqual(policy.risk_level, "high") + self.assertEqual(len(policy.images), 1) + image = policy.images[0] + self.assertEqual(image.runtime_repository, "argocd") + self.assertEqual(image.workload.name, "argocd-dex-server") + self.assertEqual(image.workload.container, "copyutil") + self.assertEqual(image.workload.container_kind, "init_container") + self.assertNotIn("github.com", raw) + self.assertNotIn("ghcr.io", raw) + self.assertIn('"external_pull_allowed": false', raw) def test_policy_rejects_external_pull_and_mutable_runtime_ref(self) -> None: payload = json.loads(POLICY_PATH.read_text(encoding="utf-8")) @@ -152,6 +193,8 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): self.assertNotIn('"images", "pull"', controller_source) self.assertIn("Mirror Runtime Cache Images to Harbor", workflow) self.assertIn("Apply Runtime Image Mirror Policy", workflow) + self.assertIn("runtime-image-mirror-staging-policy.json", workflow) + self.assertIn("runtime_image_staging_stage=verified", workflow) self.assertIn("runtime_image_mirror_controller.py", workflow) mirror_block = workflow[ workflow.index( @@ -160,6 +203,145 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): ] self.assertNotIn("HARBOR_PASSWORD", mirror_block) + def test_staging_reuses_verified_internal_artifact_without_export(self) -> None: + policy = controller.load_staging_policy(STAGING_POLICY_PATH) + image = policy.images[0] + descriptor = controller.RegistryDescriptor( + digest="sha256:" + "1" * 64, + config_digest=image.source_config_digest, + ) + with tempfile.TemporaryDirectory() as temp: + receipt_path = Path(temp) / "receipt.json" + with ( + patch.object( + controller.RuntimeCache, + "verify_source", + return_value="quay.io/argoproj/argocd:v3.3.6", + ), + patch.object( + controller.RuntimeCache, + "source_config_digest", + return_value=image.source_config_digest, + ), + patch.object(controller.RuntimeCache, "export") as export, + patch.object( + controller, "_registry_descriptor", return_value=descriptor + ), + patch.object(controller, "_registry_digest_present", return_value=True), + patch.object(controller, "_run") as run, + ): + receipt = controller.stage_images( + policy, + trace_id="aia-p0-006-02b-existing", + receipt_path=receipt_path, + ssh_key=Path("/unused/key"), + known_hosts=Path("/unused/known-hosts"), + apply=True, + ) + + self.assertEqual(receipt["state"], "verified") + self.assertEqual(receipt["missing_before_count"], 0) + self.assertEqual(receipt["images"][0]["execution"], "already_present") + self.assertTrue(receipt["images"][0]["provenance_config_digest_match"]) + export.assert_not_called() + run.assert_not_called() + + def test_staging_exports_cached_source_and_verifies_remote_provenance( + self, + ) -> None: + policy = controller.load_staging_policy(STAGING_POLICY_PATH) + image = policy.images[0] + descriptor = controller.RegistryDescriptor( + digest="sha256:" + "2" * 64, + config_digest=image.source_config_digest, + ) + with tempfile.TemporaryDirectory() as temp: + receipt_path = Path(temp) / "receipt.json" + with ( + patch.object( + controller.RuntimeCache, + "verify_source", + return_value="quay.io/argoproj/argocd:v3.3.6", + ), + patch.object( + controller.RuntimeCache, + "source_config_digest", + return_value=image.source_config_digest, + ), + patch.object(controller.RuntimeCache, "export") as export, + patch.object( + controller, + "_registry_descriptor", + side_effect=[None, descriptor], + ), + patch.object(controller, "_local_image_present", return_value=False), + patch.object(controller, "_archive_contains_digest", return_value=True), + patch.object( + controller, + "_local_config_digest", + return_value=image.source_config_digest, + ), + patch.object(controller, "_registry_digest_present", return_value=True), + patch.object(controller, "_remove_local_image") as remove, + patch.object(controller, "_run") as run, + ): + receipt = controller.stage_images( + policy, + trace_id="aia-p0-006-02b-stage", + receipt_path=receipt_path, + ssh_key=Path("/unused/key"), + known_hosts=Path("/unused/known-hosts"), + apply=True, + ) + + self.assertEqual(receipt["state"], "verified") + self.assertEqual(receipt["missing_before_count"], 1) + self.assertEqual(receipt["images"][0]["execution"], "staged_from_runtime_cache") + self.assertEqual( + receipt["images"][0]["target_registry_digest"], descriptor.digest + ) + export.assert_called_once() + commands = [call.args[0] for call in run.call_args_list] + self.assertEqual([command[1] for command in commands], ["load", "tag", "push"]) + self.assertTrue(all("pull" not in command for command in commands)) + self.assertEqual(remove.call_count, 2) + + def test_staging_rejects_internal_artifact_with_wrong_provenance(self) -> None: + policy = controller.load_staging_policy(STAGING_POLICY_PATH) + image = policy.images[0] + descriptor = controller.RegistryDescriptor( + digest="sha256:" + "3" * 64, + config_digest="sha256:" + "4" * 64, + ) + with tempfile.TemporaryDirectory() as temp: + with ( + patch.object( + controller.RuntimeCache, + "verify_source", + return_value="quay.io/argoproj/argocd:v3.3.6", + ), + patch.object( + controller.RuntimeCache, + "source_config_digest", + return_value=image.source_config_digest, + ), + patch.object( + controller, "_registry_descriptor", return_value=descriptor + ), + self.assertRaisesRegex( + controller.ControllerError, + "internal_registry_provenance_mismatch", + ), + ): + controller.stage_images( + policy, + trace_id="aia-p0-006-02b-wrong-provenance", + receipt_path=Path(temp) / "receipt.json", + ssh_key=Path("/unused/key"), + known_hosts=Path("/unused/known-hosts"), + apply=True, + ) + def test_verified_mirror_receipt_is_bound_to_trace_and_policy(self) -> None: policy = controller.load_policy(POLICY_PATH) receipt = { @@ -182,6 +364,123 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): ): controller._read_mirror_receipt(path, policy, "different-trace") + def test_init_container_read_and_patch_use_init_container_spec(self) -> None: + image = self._image_with_container_kind("init_container") + kubernetes = controller.Kubernetes( + use_sudo=False, + kubeconfig=None, + server=None, + ) + workload = { + "spec": { + "template": { + "spec": { + "containers": [{"name": "other", "image": "other:image"}], + "initContainers": [ + { + "name": image.workload.container, + "image": "source:init", + } + ], + } + } + } + } + with patch.object(kubernetes, "workload", return_value=workload): + self.assertEqual(kubernetes.current_image(image), "source:init") + + with patch.object(kubernetes, "run") as run: + kubernetes.patch_image(image, image.runtime_ref, dry_run=True) + + args = run.call_args.args[0] + patch_payload = json.loads(args[args.index("--patch") + 1]) + pod_spec = patch_payload["spec"]["template"]["spec"] + self.assertNotIn("containers", pod_spec) + self.assertEqual( + pod_spec["initContainers"], + [{"name": image.workload.container, "image": image.runtime_ref}], + ) + self.assertIn("--dry-run=server", args) + + def test_init_container_verifier_accepts_completed_digest_pinned_status( + self, + ) -> None: + image = self._image_with_container_kind("init_container") + kubernetes = controller.Kubernetes( + use_sudo=False, + kubeconfig=None, + server=None, + ) + workload = { + "spec": {"selector": {"matchLabels": {"app": "runtime-mirror-test"}}} + } + pods = { + "items": [ + { + "metadata": {"name": "runtime-mirror-test-1"}, + "status": { + "initContainerStatuses": [ + { + "name": image.workload.container, + "ready": False, + "state": {"terminated": {"exitCode": 0}}, + "imageID": "registry.local/repo@" + + image.target_registry_digest, + } + ] + }, + } + ] + } + with ( + patch.object(kubernetes, "workload", return_value=workload), + patch.object(kubernetes, "run") as run, + ): + run.return_value.stdout = json.dumps(pods) + self.assertEqual(kubernetes.verify_pods(image), 1) + + def test_init_container_verifier_rejects_incomplete_or_wrong_digest(self) -> None: + image = self._image_with_container_kind("init_container") + workload = { + "spec": {"selector": {"matchLabels": {"app": "runtime-mirror-test"}}} + } + cases = ( + ( + { + "state": {"running": {}}, + "imageID": "repo@" + image.target_registry_digest, + }, + "pod_init_container_not_completed", + ), + ( + { + "state": {"terminated": {"exitCode": 0}}, + "imageID": "repo@sha256:" + "0" * 64, + }, + "pod_image_digest_mismatch", + ), + ) + for status, error_code in cases: + with self.subTest(error_code=error_code): + status["name"] = image.workload.container + pods = { + "items": [ + {"metadata": {}, "status": {"initContainerStatuses": [status]}} + ] + } + kubernetes = controller.Kubernetes( + use_sudo=False, + kubeconfig=None, + server=None, + ) + with ( + patch.object(kubernetes, "workload", return_value=workload), + patch.object(kubernetes, "run") as run, + self.assertRaisesRegex(controller.ControllerError, error_code), + ): + run.return_value.stdout = json.dumps(pods) + kubernetes.verify_pods(image) + if __name__ == "__main__": unittest.main()