atlas-iac/ci/scripts/hermes_image_release.py

#!/usr/bin/env python3
"""Verify and render a reviewable Hermes agent image release."""

from __future__ import annotations

import argparse
import base64
import difflib
import json
import os
import re
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any, Callable


DEFAULT_IMAGE = "registry.bstein.dev/bstein/hermes-agent"
HARBOR_API_ORIGIN = "https://registry.bstein.dev/api/v2.0"
HARBOR_PROJECT = "bstein"
HARBOR_REPOSITORY = "hermes-agent"
IMMUTABLE_REPOSITORY_PATTERN = "hermes-agent"
IMMUTABLE_TAG_PATTERN = "git-*-build-*"
DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
BUILD_PATTERN = re.compile(r"^[1-9][0-9]*$")
DESTINATION_PATTERN = re.compile(
    r"^registry\.bstein\.dev/bstein/hermes-agent:"
    r"git-([0-9a-f]{40})-build-([1-9][0-9]*)$"
)


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 _validated(value: str, pattern: re.Pattern[str], label: str) -> str:
    """Return a normalized value when it matches the release contract."""
    normalized = value.strip()
    if not pattern.fullmatch(normalized):
        raise ValueError(f"invalid {label}: expected {pattern.pattern}")
    return normalized


def validate_destination(
    destination: str, source_revision: str, build_number: str
) -> tuple[str, str]:
    """Bind one unique build tag to the reviewed revision and Jenkins build."""
    revision = _validated(source_revision, REVISION_PATTERN, "source revision")
    build = _validated(build_number, BUILD_PATTERN, "build number")
    match = DESTINATION_PATTERN.fullmatch(destination.strip())
    if not match or match.groups() != (revision, build):
        raise ValueError(
            "destination must bind the reviewed revision and unique Jenkins build"
        )
    return revision, build


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}")


def verify_registry_digest(
    destination: str,
    digest: str,
    *,
    username: str,
    password: str,
    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
) -> None:
    """Verify Harbor independently resolves the pushed tag to Kaniko's digest."""
    digest = _validated(digest, DIGEST_PATTERN, "image digest")
    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")


def render_kustomization(source: str, digest: str, image: str = DEFAULT_IMAGE) -> str:
    """Replace exactly one matching Kustomize image digest without reformatting."""
    digest = _validated(digest, DIGEST_PATTERN, "image digest")
    lines = source.splitlines(keepends=True)
    matches: list[int] = []

    for index, line in enumerate(lines):
        if line.strip() != f"- name: {image}":
            continue
        name_indent = len(line) - len(line.lstrip())
        for candidate_index in range(index + 1, len(lines)):
            candidate = lines[candidate_index]
            stripped = candidate.strip()
            candidate_indent = len(candidate) - len(candidate.lstrip())
            if stripped.startswith("- name:") and candidate_indent == name_indent:
                break
            if stripped.startswith("digest:") and candidate_indent > name_indent:
                matches.append(candidate_index)
                break

    if len(matches) != 1:
        raise ValueError(
            f"expected exactly one digest for image {image!r}; found {len(matches)}"
        )

    index = matches[0]
    newline = "\n" if lines[index].endswith("\n") else ""
    prefix = lines[index][: len(lines[index]) - len(lines[index].lstrip())]
    value = lines[index].strip().removeprefix("digest:").strip()
    _current_digest, separator, comment = value.partition(" #")
    suffix = f" #{comment}" if separator else ""
    lines[index] = f"{prefix}digest: {digest}{suffix}{newline}"
    return "".join(lines)


def write_release_artifacts(
    *,
    digest: str,
    source_revision: str,
    build_number: str,
    destination: str,
    kustomization: Path,
    output_dir: Path,
) -> dict[str, str]:
    """Write a rendered manifest, patch, and credential-free release metadata."""
    digest = _validated(digest, DIGEST_PATTERN, "image digest")
    source_revision, build_number = validate_destination(
        destination, source_revision, build_number
    )
    source = kustomization.read_text(encoding="utf-8")
    rendered = render_kustomization(source, digest)
    relative_name = kustomization.name
    patch = "".join(
        difflib.unified_diff(
            source.splitlines(keepends=True),
            rendered.splitlines(keepends=True),
            fromfile=f"a/services/hermes/{relative_name}",
            tofile=f"b/services/hermes/{relative_name}",
        )
    )
    if not patch:
        raise ValueError("published digest already matches the Flux manifest")
    output_dir.mkdir(parents=True, exist_ok=True)
    (output_dir / "hermes-kustomization.yaml").write_text(rendered, encoding="utf-8")
    (output_dir / "hermes-image-update.patch").write_text(patch, encoding="utf-8")
    metadata = {
        "build_number": build_number,
        "digest": digest,
        "flux_image": f"{DEFAULT_IMAGE}@{digest}",
        "image": DEFAULT_IMAGE,
        "published_tag": destination,
        "source_revision": source_revision,
    }
    (output_dir / "hermes-agent-image.json").write_text(
        json.dumps(metadata, indent=2, sort_keys=True) + "\n", encoding="utf-8"
    )
    return metadata


def validate_release_artifacts(
    *,
    digest_file: Path,
    image_file: Path,
    source_revision: str,
    build_number: str,
    destination: str,
    kustomization: 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,
    )
    source_revision, build_number = validate_destination(
        destination, source_revision, build_number
    )
    expected_names = {
        "hermes-agent-image.json",
        "hermes-image-update.patch",
        "hermes-kustomization.yaml",
    }
    entries = list(output_dir.iterdir())
    if {entry.name for entry in entries} != expected_names or not all(
        entry.is_file() and not entry.is_symlink() for entry in entries
    ):
        raise ValueError("release output must contain exactly three evidence files")
    source = kustomization.read_text(encoding="utf-8")
    rendered = render_kustomization(source, digest)
    relative_name = kustomization.name
    patch = "".join(
        difflib.unified_diff(
            source.splitlines(keepends=True),
            rendered.splitlines(keepends=True),
            fromfile=f"a/services/hermes/{relative_name}",
            tofile=f"b/services/hermes/{relative_name}",
        )
    )
    metadata = {
        "build_number": build_number,
        "digest": digest,
        "flux_image": f"{DEFAULT_IMAGE}@{digest}",
        "image": DEFAULT_IMAGE,
        "published_tag": destination,
        "source_revision": source_revision,
    }
    expected = {
        "hermes-agent-image.json": json.dumps(metadata, indent=2, sort_keys=True)
        + "\n",
        "hermes-image-update.patch": patch,
        "hermes-kustomization.yaml": rendered,
    }
    for name, expected_text in expected.items():
        if (output_dir / name).read_text(encoding="utf-8") != expected_text:
            raise ValueError(f"release evidence is incomplete or mismatched: {name}")


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("--kustomization", 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("--kustomization", 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,
            kustomization=args.kustomization,
            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, username=username, password=password
    )
    write_release_artifacts(
        digest=digest,
        source_revision=args.source_revision,
        build_number=args.build_number,
        destination=args.destination,
        kustomization=args.kustomization,
        output_dir=args.output_dir,
    )
    return 0


if __name__ == "__main__":  # pragma: no cover - exercised through main()
    raise SystemExit(main())