atlas-iac/services/hermes/scripts/jenkins_image_build_trigger.py

#!/usr/bin/env python3
"""Trigger a bounded reviewed-main Hermes image release job."""

from __future__ import annotations

import argparse
import json
import re
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


JENKINS_ORIGIN = "https://ci.bstein.dev"
JENKINS_BUILD_URL = f"{JENKINS_ORIGIN}/buildByToken/buildWithParameters"
SCM_BROKER_ORIGIN = "http://hermes-scm-broker.hermes-scm.svc.cluster.local:9081"
SCM_METADATA_URL = f"{SCM_BROKER_ORIGIN}/v1/metadata"
SCM_REPOSITORY = "atlas-iac"
MAX_SCM_RESPONSE_BYTES = 512 * 1024
JOBS = {
    "agent": {
        "job": "hermes-agent-image",
        "confirmation": "PUBLISH HERMES AGENT",
    },
    "webui": {
        "job": "hermes-webui-image",
        "confirmation": "PUBLISH HERMES WEBUI",
    },
    "router": {
        "job": "hermes-chat-router-image",
        "confirmation": "PUBLISH HERMES CHAT ROUTER",
    },
    "stt": {
        "job": "hermes-voice-image",
        "confirmation": "PUBLISH HERMES STT",
        "parameters": {"IMAGE_COMPONENT": "stt"},
    },
    "tts": {
        "job": "hermes-voice-image",
        "confirmation": "PUBLISH HERMES TTS",
        "parameters": {"IMAGE_COMPONENT": "tts"},
    },
}
JOB_NAME = JOBS["agent"]["job"]
TOKEN_FILE = Path("/runtime-access/jenkins-image-build-token")
REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
QUEUE_PATH_PATTERN = re.compile(r"^/queue/item/[0-9]+/?$")


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """Keep a queued-build redirect from becoming an unauthorized job read."""

    def redirect_request(self, _request, _file, _code, _message, _headers, _url):
        return None


def _open_without_redirect(request: urllib.request.Request, timeout: int):
    """Return the Build Token Root response, including its expected HTTP 303."""
    opener = urllib.request.build_opener(_NoRedirect())
    try:
        return opener.open(request, timeout=timeout)
    except urllib.error.HTTPError as exc:
        if exc.code == 303:
            return exc
        raise


def _verify_reviewed_revision_exists(revision: str) -> None:
    """Fail before queuing when the exact reviewed commit is absent from SCM."""
    path = f"/api/v1/repos/titan/{SCM_REPOSITORY}/git/commits/{revision}"
    payload = json.dumps({"path": path}, separators=(",", ":")).encode("utf-8")
    request = urllib.request.Request(
        SCM_METADATA_URL,
        data=payload,
        headers={
            "Accept": "application/json",
            "Content-Type": "application/json",
            "User-Agent": "hermes-image-release/1",
        },
        method="POST",
    )
    opener = urllib.request.build_opener(_NoRedirect())
    with opener.open(request, timeout=20) as response:
        if int(response.status) != 200:
            raise RuntimeError("SCM did not confirm the reviewed revision")
        content_type = response.headers.get_content_type()
        if content_type != "application/json":
            raise RuntimeError("SCM returned an unexpected revision response")
        body = response.read(MAX_SCM_RESPONSE_BYTES + 1)
    if len(body) > MAX_SCM_RESPONSE_BYTES:
        raise RuntimeError("SCM revision response exceeded the safe size limit")
    try:
        result = json.loads(body)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise RuntimeError("SCM returned malformed revision evidence") from exc
    if not isinstance(result, dict) or result.get("sha") != revision:
        raise RuntimeError("SCM did not confirm the exact reviewed revision")


def trigger_build(
    revision: str,
    *,
    component: str = "agent",
    token_file: Path = TOKEN_FILE,
    opener=_open_without_redirect,
    revision_verifier=None,
) -> dict[str, str | int]:
    """Post one allow-listed job's fixed parameters using the release token."""
    revision = revision.strip()
    if not REVISION_PATTERN.fullmatch(revision):
        raise ValueError("revision must be a lowercase full 40-character commit")
    job = JOBS.get(component)
    if job is None:
        raise ValueError("component must be agent, router, webui, stt, or tts")
    token = token_file.read_text(encoding="utf-8").strip()
    if not token:
        raise RuntimeError("Jenkins image-build token is empty")
    if revision_verifier is not None:
        revision_verifier(revision)
    fields = {
        "job": job["job"],
        "token": token,
        "PUBLISH_IMAGE": "true",
        "EXPECTED_SOURCE_REVISION": revision,
        "CONFIRM_PUBLISH": job["confirmation"],
    }
    fields.update(job.get("parameters", {}))
    payload = urllib.parse.urlencode(fields).encode("utf-8")
    request = urllib.request.Request(
        JENKINS_BUILD_URL,
        data=payload,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    with opener(request, timeout=20) as response:
        status = int(response.status)
        location = response.headers.get("Location", "")
    if status not in {201, 303}:
        raise RuntimeError(f"Jenkins trigger returned HTTP {status}")
    if not location:
        raise RuntimeError("Jenkins trigger omitted the queue Location")
    queue_url = urllib.parse.urljoin(f"{JENKINS_ORIGIN}/", location)
    parsed_queue = urllib.parse.urlsplit(queue_url)
    expected_origin = urllib.parse.urlsplit(JENKINS_ORIGIN)
    if (
        parsed_queue.scheme != expected_origin.scheme
        or parsed_queue.netloc != expected_origin.netloc
        or parsed_queue.query
        or parsed_queue.fragment
        or not QUEUE_PATH_PATTERN.fullmatch(parsed_queue.path)
    ):
        raise RuntimeError("Jenkins returned an invalid queue Location")
    # Never return the submitted URL: its form body contains the job token.
    queue_path = parsed_queue.path
    return {
        "component": component,
        "follow_command": (
            "/opt/coordinator/hermes_image_release_status.py "
            f"--component {component} --revision {revision} --wait"
        ),
        "job": job["job"],
        "queue_path": queue_path,
        "source_revision": revision,
        "status": status,
    }


def main() -> int:
    """Validate one revision, trigger the bounded job, and print safe metadata."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--component", choices=sorted(JOBS), default="agent", help="image to release"
    )
    parser.add_argument("revision", help="reviewed full commit contained by main")
    args = parser.parse_args()
    try:
        result = trigger_build(
            args.revision,
            component=args.component,
            revision_verifier=_verify_reviewed_revision_exists,
        )
    except (OSError, ValueError, RuntimeError, urllib.error.URLError) as exc:
        print(json.dumps({"error": str(exc)}, sort_keys=True))
        return 1
    print(json.dumps(result, indent=2, sort_keys=True))
    return 0


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