301 lines
11 KiB
Python
Executable File
301 lines
11 KiB
Python
Executable File
#!/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"
|
|
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}",
|
|
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 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]
|
|
tag_names = {
|
|
str(item.get("name"))
|
|
for item in artifact.get("tags") or []
|
|
if isinstance(item, dict) and item.get("name")
|
|
}
|
|
if expected_tag not in tag_names:
|
|
raise RuntimeError("Harbor artifact does not contain the expected tag")
|
|
|
|
|
|
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())]
|
|
lines[index] = f"{prefix}digest: {digest}{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 _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)
|
|
args = parser.parse_args()
|
|
|
|
validate_destination(args.destination, args.source_revision, args.build_number)
|
|
username, password = _credentials()
|
|
if args.command == "assert-absent":
|
|
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())
|