#!/usr/bin/env python3 """Verify and render a reviewable Hermes WebUI image release.""" from __future__ import annotations import argparse import base64 import json import os import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any, Callable from hermes_webui_flux_release import ( DEFAULT_IMAGE as DEFAULT_IMAGE, DESTINATION_PATTERN, DIGEST_PATTERN, REVISION_PATTERN, render_hux_build_metadata as render_hux_build_metadata, render_workload as render_workload, validate_destination, validate_release_artifacts as validate_flux_release_artifacts, validated as _validated, write_release_artifacts, ) HARBOR_API_ORIGIN = "https://registry.bstein.dev/api/v2.0" HARBOR_PROJECT = "bstein" HARBOR_REPOSITORY = "hermes-webui" IMMUTABLE_REPOSITORY_PATTERN = "hermes-webui" IMMUTABLE_TAG_PATTERN = "git-*-build-*" class _NoRedirect(urllib.request.HTTPRedirectHandler): """Never send registry credentials to a redirect target.""" def redirect_request(self, _request, _file, _code, _message, _headers, _url): return None def validate_kaniko_evidence( *, digest_text: str, image_text: str, destination: str ) -> str: """Cross-check both independent Kaniko output files against the destination.""" digest_lines = digest_text.splitlines() image_lines = image_text.splitlines() if len(digest_lines) != 1: raise ValueError("Kaniko digest evidence must contain exactly one line") if len(image_lines) != 1: raise ValueError("Kaniko image evidence must contain exactly one line") digest = _validated(digest_lines[0], DIGEST_PATTERN, "image digest") if image_lines[0].strip() != f"{destination}@{digest}": raise ValueError("Kaniko image evidence does not match destination and digest") return digest def _registry_request(request: urllib.request.Request, timeout: int) -> Any: """Make a registry request without following redirects.""" opener = urllib.request.build_opener(_NoRedirect()) try: return opener.open(request, timeout=timeout) except urllib.error.HTTPError as exc: return exc def _artifact_response( destination: str, *, username: str, password: str, opener: Callable[[urllib.request.Request, int], Any] = _registry_request, ) -> tuple[int, bytes]: """Read one exact Harbor artifact by tag with bounded response size.""" match = DESTINATION_PATTERN.fullmatch(destination) if not match: raise ValueError("invalid destination") if not username or not password: raise RuntimeError("Harbor credentials are empty") tag = destination.rsplit(":", 1)[1] encoded_tag = urllib.parse.quote(tag, safe="") auth = base64.b64encode(f"{username}:{password}".encode()).decode("ascii") request = urllib.request.Request( f"{HARBOR_API_ORIGIN}/projects/{HARBOR_PROJECT}/repositories/" f"{HARBOR_REPOSITORY}/artifacts/{encoded_tag}" "?with_immutable_status=true", headers={"Accept": "application/json", "Authorization": f"Basic {auth}"}, method="GET", ) with opener(request, 20) as response: body = response.read(1_048_577) if len(body) > 1_048_576: raise RuntimeError("Harbor artifact response exceeded the size limit") return int(response.status), body def _immutable_rules_response( *, username: str, password: str, opener: Callable[[urllib.request.Request, int], Any] = _registry_request, ) -> tuple[int, bytes, dict[str, str]]: """Read the project policy with the same least-privilege publish identity.""" if not username or not password: raise RuntimeError("Harbor credentials are empty") auth = base64.b64encode(f"{username}:{password}".encode()).decode("ascii") request = urllib.request.Request( f"{HARBOR_API_ORIGIN}/projects/{HARBOR_PROJECT}/immutabletagrules" "?page=1&page_size=100", headers={"Accept": "application/json", "Authorization": f"Basic {auth}"}, method="GET", ) with opener(request, 20) as response: body = response.read(1_048_577) if len(body) > 1_048_576: raise RuntimeError("Harbor immutable rule response exceeded the size limit") return int(response.status), body, dict(response.headers) def _require_complete_rule_page( rules: list[dict[str, Any]], headers: dict[str, str] ) -> None: """Require proof that the bounded first page contains every rule.""" raw_total = next( (value for key, value in headers.items() if key.lower() == "x-total-count"), None, ) if raw_total is None or not str(raw_total).isdecimal(): raise RuntimeError("Harbor immutable rule list omitted a valid total count") if int(raw_total) != len(rules): raise RuntimeError("Harbor immutable rule list was truncated") def _normalized_immutable_rule(rule: dict[str, Any]) -> dict[str, Any]: """Select only fields that bind the server-side build-tag policy.""" return { "disabled": bool(rule.get("disabled", False)), "action": rule.get("action"), "template": rule.get("template"), "tag_selectors": [ { "kind": item.get("kind"), "decoration": item.get("decoration"), "pattern": item.get("pattern"), } for item in rule.get("tag_selectors") or [] if isinstance(item, dict) ], "scope_selectors": { "repository": [ { "kind": item.get("kind"), "decoration": item.get("decoration"), "pattern": item.get("pattern"), } for item in (rule.get("scope_selectors") or {}).get("repository", []) if isinstance(item, dict) ] }, } def verify_immutable_policy( *, username: str, password: str, opener: Callable[[urllib.request.Request, int], Any] = _registry_request, ) -> None: """Fail closed before build unless the exact Harbor rule is active.""" status, body, headers = _immutable_rules_response( username=username, password=password, opener=opener ) if status != 200: raise RuntimeError(f"Harbor immutable policy preflight returned HTTP {status}") try: rules = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise RuntimeError("Harbor returned invalid immutable rule JSON") from exc if not isinstance(rules, list) or not all(isinstance(item, dict) for item in rules): raise RuntimeError("Harbor immutable rule list has an invalid shape") _require_complete_rule_page(rules, headers) expected = { "disabled": False, "action": "immutable", "template": "immutable_template", "tag_selectors": [ { "kind": "doublestar", "decoration": "matches", "pattern": IMMUTABLE_TAG_PATTERN, } ], "scope_selectors": { "repository": [ { "kind": "doublestar", "decoration": "repoMatches", "pattern": IMMUTABLE_REPOSITORY_PATTERN, } ] }, } matches = [ _normalized_immutable_rule(item) for item in rules if _normalized_immutable_rule(item)["tag_selectors"] == expected["tag_selectors"] and _normalized_immutable_rule(item)["scope_selectors"] == expected["scope_selectors"] ] if matches != [expected]: raise RuntimeError("Harbor immutable build-tag policy is absent or not exact") def assert_tag_absent( destination: str, *, username: str, password: str, opener: Callable[[urllib.request.Request, int], Any] = _registry_request, ) -> None: """Reject replay before Kaniko can push an already-used immutable identity.""" status, _body = _artifact_response( destination, username=username, password=password, opener=opener ) if status == 404: return if status == 200: raise RuntimeError("Harbor destination tag already exists; refusing overwrite") raise RuntimeError(f"Harbor destination preflight returned HTTP {status}") OCI_REVISION_LABEL = "org.opencontainers.image.revision" def _child_artifact_response( child_digest: str, *, username: str, password: str, opener: Callable[[urllib.request.Request, int], Any] = _registry_request, ) -> tuple[int, bytes]: """Read one per-arch child artifact of a multi-arch index by its digest.""" if not username or not password: raise RuntimeError("Harbor credentials are empty") encoded = urllib.parse.quote(child_digest, safe="") auth = base64.b64encode(f"{username}:{password}".encode()).decode("ascii") request = urllib.request.Request( f"{HARBOR_API_ORIGIN}/projects/{HARBOR_PROJECT}/repositories/" f"{HARBOR_REPOSITORY}/artifacts/{encoded}" "?with_immutable_status=true", headers={"Accept": "application/json", "Authorization": f"Basic {auth}"}, method="GET", ) with opener(request, 20) as response: body = response.read(1_048_577) if len(body) > 1_048_576: raise RuntimeError("Harbor child artifact response exceeded the size limit") return int(response.status), body def _config_labels(artifact: dict[str, Any]) -> Any: """Extract the OCI config labels Harbor reports for one artifact, if any.""" return ((artifact.get("extra_attrs") or {}).get("config") or {}).get("Labels") def _verify_source_revision_label( artifact: dict[str, Any], source_revision: str, *, username: str, password: str, opener: Callable[[urllib.request.Request, int], Any], ) -> None: """Assert the published image carries the reviewed source revision label. A single-arch image exposes ``org.opencontainers.image.revision`` on its own config. A multi-arch manifest list has no top-level config, so Harbor reports the label on each per-arch child instead; verify every child in that case. """ labels = _config_labels(artifact) if isinstance(labels, dict): if labels.get(OCI_REVISION_LABEL) != source_revision: raise RuntimeError("Harbor OCI source-revision label does not match") return references = artifact.get("references") if not isinstance(references, list) or not references: raise RuntimeError("Harbor artifact omitted OCI image labels") for reference in references: child_digest = str((reference or {}).get("child_digest") or "").strip() if not DIGEST_PATTERN.fullmatch(child_digest): raise RuntimeError("Harbor index reference omitted a valid child digest") status, body = _child_artifact_response( child_digest, username=username, password=password, opener=opener ) if status != 200: raise RuntimeError( f"Harbor child artifact verification returned HTTP {status}" ) try: child = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise RuntimeError("Harbor returned invalid child artifact JSON") from exc child_labels = _config_labels(child) if not isinstance(child_labels, dict): raise RuntimeError("Harbor artifact omitted OCI image labels") if child_labels.get(OCI_REVISION_LABEL) != source_revision: raise RuntimeError("Harbor OCI source-revision label does not match") def verify_registry_digest( destination: str, digest: str, source_revision: str, *, username: str, password: str, opener: Callable[[urllib.request.Request, int], Any] = _registry_request, ) -> None: """Verify Harbor resolves the tag, digest, and persisted source revision.""" digest = _validated(digest, DIGEST_PATTERN, "image digest") source_revision = _validated( source_revision, REVISION_PATTERN, "source revision" ) status, body = _artifact_response( destination, username=username, password=password, opener=opener ) if status != 200: raise RuntimeError(f"Harbor manifest verification returned HTTP {status}") try: artifact = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise RuntimeError("Harbor returned invalid artifact JSON") from exc harbor_digest = str(artifact.get("digest") or "").strip() if not DIGEST_PATTERN.fullmatch(harbor_digest): raise RuntimeError("Harbor response omitted a valid artifact digest") if harbor_digest != digest: raise RuntimeError("Harbor digest does not match Kaniko evidence") expected_tag = destination.rsplit(":", 1)[1] matching_tags = [ item for item in artifact.get("tags") or [] if isinstance(item, dict) and item.get("name") == expected_tag ] if len(matching_tags) != 1: raise RuntimeError("Harbor artifact does not contain the expected tag") if matching_tags[0].get("immutable") is not True: raise RuntimeError("Harbor did not enforce the expected tag as immutable") _verify_source_revision_label( artifact, source_revision, username=username, password=password, opener=opener, ) def validate_release_artifacts( *, digest_file: Path, image_file: Path, source_revision: str, build_number: str, destination: str, chat_manifest: Path, dashboard_manifest: Path, output_dir: Path, ) -> None: """Revalidate the exact successful-build evidence without rewriting it.""" digest = validate_kaniko_evidence( digest_text=digest_file.read_text(encoding="utf-8"), image_text=image_file.read_text(encoding="utf-8"), destination=destination, ) validate_flux_release_artifacts( digest=digest, source_revision=source_revision, build_number=build_number, destination=destination, chat_manifest=chat_manifest, dashboard_manifest=dashboard_manifest, output_dir=output_dir, ) def _credentials() -> tuple[str, str]: """Read the masked, runtime-only Jenkins credential environment.""" username = os.environ.get("HARBOR_USER", "") password = os.environ.get("HARBOR_PASSWORD", "") if not username or not password: raise RuntimeError("Harbor credentials are unavailable") return username, password def _common_arguments(parser: argparse.ArgumentParser) -> None: parser.add_argument("--source-revision", required=True) parser.add_argument("--build-number", required=True) parser.add_argument("--destination", required=True) def main() -> int: """Fail closed around the unique tag, then verify and render after push.""" parser = argparse.ArgumentParser() commands = parser.add_subparsers(dest="command", required=True) absent = commands.add_parser("assert-absent") _common_arguments(absent) render = commands.add_parser("render") _common_arguments(render) render.add_argument("--digest-file", required=True, type=Path) render.add_argument("--image-file", required=True, type=Path) render.add_argument("--chat-manifest", required=True, type=Path) render.add_argument("--dashboard-manifest", required=True, type=Path) render.add_argument("--output-dir", required=True, type=Path) verify = commands.add_parser("verify-evidence") _common_arguments(verify) verify.add_argument("--digest-file", required=True, type=Path) verify.add_argument("--image-file", required=True, type=Path) verify.add_argument("--chat-manifest", required=True, type=Path) verify.add_argument("--dashboard-manifest", required=True, type=Path) verify.add_argument("--output-dir", required=True, type=Path) args = parser.parse_args() validate_destination(args.destination, args.source_revision, args.build_number) if args.command == "verify-evidence": validate_release_artifacts( digest_file=args.digest_file, image_file=args.image_file, source_revision=args.source_revision, build_number=args.build_number, destination=args.destination, chat_manifest=args.chat_manifest, dashboard_manifest=args.dashboard_manifest, output_dir=args.output_dir, ) return 0 username, password = _credentials() if args.command == "assert-absent": verify_immutable_policy(username=username, password=password) assert_tag_absent(args.destination, username=username, password=password) return 0 digest = validate_kaniko_evidence( digest_text=args.digest_file.read_text(encoding="utf-8"), image_text=args.image_file.read_text(encoding="utf-8"), destination=args.destination, ) verify_registry_digest( args.destination, digest, args.source_revision, username=username, password=password, ) write_release_artifacts( digest=digest, source_revision=args.source_revision, build_number=args.build_number, destination=args.destination, chat_manifest=args.chat_manifest, dashboard_manifest=args.dashboard_manifest, output_dir=args.output_dir, ) return 0 if __name__ == "__main__": # pragma: no cover - exercised through main() raise SystemExit(main())