375 lines
14 KiB
Python
Executable File
375 lines
14 KiB
Python
Executable File
#!/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,
|
|
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}")
|
|
|
|
|
|
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 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, 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())
|