atlas-iac/testing/tests/test_hermes_image_builder.py

"""Safety and artifact contracts for the Hermes agent image release lane."""

from __future__ import annotations

import importlib.util
import io
import json
import re
import urllib.parse
from pathlib import Path

import pytest
import yaml


REPO_ROOT = Path(__file__).resolve().parents[2]
PIPELINE_PATH = REPO_ROOT / "ci/Jenkinsfile.hermes-agent-image"
RELEASE_SCRIPT = REPO_ROOT / "ci/scripts/hermes_image_release.py"
TRIGGER_SCRIPT = REPO_ROOT / "services/hermes/scripts/jenkins_image_build_trigger.py"
HEREDOC_RUNNER = REPO_ROOT / "dockerfiles/hermes-kaniko-heredoc-runner.py"


def _load_release_module():
    spec = importlib.util.spec_from_file_location(
        "hermes_image_release", RELEASE_SCRIPT
    )
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _load_trigger_module():
    spec = importlib.util.spec_from_file_location(
        "jenkins_image_build_trigger", TRIGGER_SCRIPT
    )
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _load_heredoc_runner():
    spec = importlib.util.spec_from_file_location(
        "hermes_kaniko_heredoc_runner", HEREDOC_RUNNER
    )
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _pod_spec() -> dict:
    source = PIPELINE_PATH.read_text(encoding="utf-8")
    pod_yaml = source.split('yaml """', 1)[1].split('"""', 1)[0]
    return yaml.safe_load(pod_yaml)["spec"]


def test_builder_pod_is_daemonless_and_kernel_bounded() -> None:
    """The builder gets only proven build caps, never host, daemon, or K8s access."""
    source = PIPELINE_PATH.read_text(encoding="utf-8")
    spec = _pod_spec()
    assert spec["serviceAccountName"] == "hermes-image-builder"
    assert spec["automountServiceAccountToken"] is False
    assert spec["enableServiceLinks"] is False
    assert "hostPath" not in source
    assert "docker.sock" not in source
    assert "tcp://" not in source
    assert "buildkitd" not in source.lower()
    assert "dind" not in source.lower()
    assert "serviceAccountToken" not in source

    containers = {item["name"]: item for item in spec["containers"]}
    kaniko = containers["kaniko"]
    assert kaniko["image"] == (
        "gcr.io/kaniko-project/executor@sha256:"
        "c3109d5926a997b100c4343944e06c6b30a6804b2f9abe0994d3de6ef92b028e"
    )
    assert kaniko["securityContext"]["capabilities"]["add"] == [
        "CHOWN",
        "FOWNER",
        "DAC_OVERRIDE",
        "SETGID",
        "SETUID",
    ]
    for container in containers.values():
        security = container["securityContext"]
        assert security["allowPrivilegeEscalation"] is False
        assert security["capabilities"]["drop"] == ["ALL"]
        assert security["seccompProfile"]["type"] == "RuntimeDefault"
        assert security.get("privileged", False) is False
    assert "add" not in containers["jnlp"]["securityContext"]["capabilities"]
    assert "add" not in containers["python"]["securityContext"]["capabilities"]


def test_pipeline_requires_reviewed_main_and_runtime_credentials() -> None:
    """Publishing requires explicit confirmation and a reviewed main commit."""
    source = PIPELINE_PATH.read_text(encoding="utf-8")
    assert 'test "${PUBLISH_IMAGE}" = "true"' in source
    assert 'test "${CONFIRM_PUBLISH}" = "PUBLISH HERMES AGENT"' in source
    assert "git rev-parse origin/main" in source
    assert "credentialsId: 'harbor-robot'" in source
    assert "set +x" in source
    assert "umask 077" in source
    assert "unset HARBOR_USER HARBOR_PASSWORD auth" in source
    assert '/busybox/rm -f "${config_path}"' in source
    assert "--digest-file=" in source
    assert "--image-name-tag-with-digest-file=" in source
    assert "--destination=" in source
    assert "assert-absent" in source
    assert "git-${actual_revision}-build-${BUILD_NUMBER}" in source
    assert source.count("--build-arg=HERMES_KANIKO_HEREDOC_COMPAT=1") == 1


def test_kaniko_replays_only_the_exact_reviewed_heredoc_contract() -> None:
    """Pinned Kaniko's ignored inline files are replayed in exact source order."""
    module = _load_heredoc_runner()
    dockerfile = (REPO_ROOT / "dockerfiles/Dockerfile.hermes-agent").read_text()
    blocks = module.extract_blocks(dockerfile)
    assert tuple(command[0] for command, _body in blocks) == (
        "node",
        "python",
        "python",
        "python",
        "python",
        "python",
        "python",
        "python",
        "node",
    )
    assert "session message total" in blocks[-1][1]
    compat = dockerfile.split("ARG HERMES_KANIKO_HEREDOC_COMPAT=0", 1)[1]
    assert compat.count("python /tmp/hermes-kaniko-heredoc-runner.py") == 9
    assert [int(value) for value in re.findall(r"--block-index (\d+)", compat)] == list(
        range(1, 10)
    )
    ignored = (
        REPO_ROOT / "dockerfiles/Dockerfile.hermes-agent.dockerignore"
    ).read_text()
    assert "!dockerfiles/Dockerfile.hermes-agent" in ignored
    assert "!dockerfiles/hermes-kaniko-heredoc-runner.py" in ignored


def test_kaniko_heredoc_runner_rejects_drift_and_executes_separately(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Missing blocks fail closed and each reviewed body gets a fresh process."""
    module = _load_heredoc_runner()
    source = (REPO_ROOT / "dockerfiles/Dockerfile.hermes-agent").read_text()
    with pytest.raises(ValueError, match="contract changed"):
        module.extract_blocks(source.replace("RUN node <<'NODE'", "RUN node", 1))

    calls = []

    def run(command, **kwargs):
        calls.append((command, kwargs))

    monkeypatch.setattr(module.subprocess, "run", run)
    dockerfile = tmp_path / "Dockerfile"
    dockerfile.write_text(source, encoding="utf-8")
    for block_index in range(1, 10):
        module.replay(dockerfile, block_index)
    assert len(calls) == 9
    assert all(call[1]["check"] is True for call in calls)
    assert all(call[1]["text"] is True for call in calls)
    assert all(call[1]["input"].endswith("\n") for call in calls)


def test_jenkins_job_is_manual_and_reads_pipeline_from_main() -> None:
    """JCasC must not publish unreviewed branch contents or poll automatically."""
    config = yaml.safe_load(
        (REPO_ROOT / "services/jenkins/configmap-jcasc.yaml").read_text(
            encoding="utf-8"
        )
    )
    jobs = config["data"]["jobs.yaml"]
    block = jobs.split("pipelineJob('hermes-agent-image')", 1)[1].split(
        "pipelineJob(", 1
    )[0]
    assert "branches('*/main')" in block
    assert "scriptPath('ci/Jenkinsfile.hermes-agent-image')" in block
    assert (
        "authenticationToken(System.getenv('HERMES_AGENT_IMAGE_BUILD_TOKEN'))" in block
    )
    assert "pipelineTriggers" not in block
    assert "scmTrigger" not in block


def test_agent_trigger_is_limited_to_the_image_job(tmp_path: Path) -> None:
    """Agent Hermes gets one job token, fixed parameters, and no Jenkins admin API."""
    module = _load_trigger_module()
    revision = "a" * 40
    token_path = tmp_path / "token"
    token_path.write_text("private-job-token\n", encoding="utf-8")
    captured = {}

    class Response(io.BytesIO):
        status = 201
        headers = {"Location": "https://ci.bstein.dev/queue/item/42/"}

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            self.close()

    def opener(request, timeout):
        captured["request"] = request
        captured["timeout"] = timeout
        return Response(b"")

    result = module.trigger_build(revision, token_file=token_path, opener=opener)
    request = captured["request"]
    fields = urllib.parse.parse_qs(request.data.decode("utf-8"))
    assert request.full_url == module.JENKINS_BUILD_URL
    assert fields == {
        "CONFIRM_PUBLISH": ["PUBLISH HERMES AGENT"],
        "EXPECTED_SOURCE_REVISION": [revision],
        "PUBLISH_IMAGE": ["true"],
        "job": ["hermes-agent-image"],
        "token": ["private-job-token"],
    }
    assert captured["timeout"] == 20
    assert "private-job-token" not in json.dumps(result)
    assert result["source_revision"] == revision


def test_agent_trigger_rejects_unsafe_revision_and_empty_token(tmp_path: Path) -> None:
    """No user-controlled job, URL, or abbreviated revision reaches Jenkins."""
    module = _load_trigger_module()
    token_path = tmp_path / "token"
    token_path.write_text("token\n", encoding="utf-8")
    with pytest.raises(ValueError):
        module.trigger_build("main", token_file=token_path)
    token_path.write_text("\n", encoding="utf-8")
    with pytest.raises(RuntimeError, match="empty"):
        module.trigger_build("a" * 40, token_file=token_path)


def test_agent_trigger_can_select_only_the_bounded_webui_job(tmp_path: Path) -> None:
    """Hermes can release WebUI without gaining a caller-controlled Jenkins job."""
    module = _load_trigger_module()
    token_path = tmp_path / "token"
    token_path.write_text("private-job-token\n", encoding="utf-8")
    captured = {}

    class Response(io.BytesIO):
        status = 201
        headers = {"Location": "https://ci.bstein.dev/queue/item/84/"}

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            self.close()

    def opener(request, timeout):
        captured["request"] = request
        captured["timeout"] = timeout
        return Response(b"")

    result = module.trigger_build(
        "b" * 40, component="webui", token_file=token_path, opener=opener
    )
    fields = urllib.parse.parse_qs(captured["request"].data.decode("utf-8"))
    assert fields["job"] == ["hermes-webui-image"]
    assert fields["CONFIRM_PUBLISH"] == ["PUBLISH HERMES WEBUI"]
    assert result["component"] == "webui"
    with pytest.raises(ValueError, match="agent or webui"):
        module.trigger_build(
            "b" * 40, component="other", token_file=token_path, opener=opener
        )


def test_agent_trigger_accepts_existing_queue_redirect(tmp_path: Path) -> None:
    """HTTP 303 means the exact release is already queued, not a trigger failure."""
    module = _load_trigger_module()
    token_path = tmp_path / "token"
    token_path.write_text("token\n", encoding="utf-8")

    class Response(io.BytesIO):
        status = 303
        headers = {"Location": "https://ci.bstein.dev/queue/item/7/"}

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            self.close()

    result = module.trigger_build(
        "e" * 40, token_file=token_path, opener=lambda *_args, **_kwargs: Response()
    )
    assert result["status"] == 303
    assert result["queue_path"] == "/queue/item/7/"


def test_job_token_is_generated_and_injected_only_at_runtime() -> None:
    """The fixed-job credential stays in Vault and pod-lifetime memory."""
    plugins = (REPO_ROOT / "services/jenkins/configmap-plugins.yaml").read_text(
        encoding="utf-8"
    )
    vault = (
        REPO_ROOT / "services/vault/scripts/vault_k8s_auth_configure.sh"
    ).read_text(encoding="utf-8")
    seeder = (
        REPO_ROOT
        / "services/vault-hermes-jenkins-token-seed/scripts/vault_hermes_jenkins_build_token_ensure.sh"
    ).read_text(encoding="utf-8")
    jenkins = (REPO_ROOT / "services/jenkins/deployment.yaml").read_text(
        encoding="utf-8"
    )
    agent = (REPO_ROOT / "services/hermes/agent-deployment.yaml").read_text(
        encoding="utf-8"
    )
    stage = (REPO_ROOT / "services/hermes/scripts/stage_runtime_access.py").read_text(
        encoding="utf-8"
    )
    assert "build-token-root:365.v717f8685a_09e" in plugins
    assert 'write_raw_policy "hermes-jenkins-token-seed"' in vault
    assert "sys/tools/random/32 format=hex" in seeder
    assert '"options":{"cas":0}' in seeder
    assert "kv/data/atlas/hermes/developer-jenkins" in seeder
    assert "HERMES_AGENT_IMAGE_BUILD_TOKEN={{ .Data.data.build_token }}" in jenkins
    assert "agent-inject-secret-jenkins-image-build-token" in agent
    assert '"jenkins-image-build-token"' in stage


def test_release_renderer_preserves_manifest_and_emits_safe_artifacts(
    tmp_path: Path,
) -> None:
    """The release artifact is exact, reviewable, and contains no credentials."""
    module = _load_release_module()
    old_digest = "sha256:" + "1" * 64
    new_digest = "sha256:" + "2" * 64
    revision = "a" * 40
    build_number = "17"
    manifest = tmp_path / "kustomization.yaml"
    manifest.write_text(
        "apiVersion: kustomize.config.k8s.io/v1beta1\n"
        "kind: Kustomization\n"
        "images:\n"
        f"  - name: {module.DEFAULT_IMAGE}\n"
        f"    digest: {old_digest}\n",
        encoding="utf-8",
    )
    output = tmp_path / "out"
    metadata = module.write_release_artifacts(
        digest=f"{new_digest}\n",
        source_revision=revision,
        build_number=build_number,
        destination=f"{module.DEFAULT_IMAGE}:git-{revision}-build-{build_number}",
        kustomization=manifest,
        output_dir=output,
    )

    assert manifest.read_text(encoding="utf-8").endswith(f"{old_digest}\n")
    assert (
        (output / "hermes-kustomization.yaml")
        .read_text(encoding="utf-8")
        .endswith(f"{new_digest}\n")
    )
    patch = (output / "hermes-image-update.patch").read_text(encoding="utf-8")
    assert f"-    digest: {old_digest}" in patch
    assert f"+    digest: {new_digest}" in patch
    assert (
        json.loads((output / "hermes-agent-image.json").read_text(encoding="utf-8"))
        == metadata
    )
    assert set(metadata) == {
        "digest",
        "build_number",
        "flux_image",
        "image",
        "published_tag",
        "source_revision",
    }


@pytest.mark.parametrize(
    ("digest", "revision", "destination"),
    [
        (
            "latest",
            "a" * 40,
            "registry.bstein.dev/bstein/hermes-agent:git-" + "a" * 40 + "-build-1",
        ),
        (
            "sha256:" + "1" * 64,
            "short",
            "registry.bstein.dev/bstein/hermes-agent:git-short-build-1",
        ),
        (
            "sha256:" + "1" * 64,
            "a" * 40,
            "registry.bstein.dev/bstein/hermes-agent:latest",
        ),
    ],
)
def test_release_renderer_rejects_unpinned_inputs(
    tmp_path: Path, digest: str, revision: str, destination: str
) -> None:
    """Only exact digests and immutable source-derived tags are accepted."""
    module = _load_release_module()
    manifest = tmp_path / "kustomization.yaml"
    manifest.write_text(
        "images:\n"
        f"  - name: {module.DEFAULT_IMAGE}\n"
        "    digest: sha256:" + "0" * 64 + "\n",
        encoding="utf-8",
    )
    with pytest.raises(ValueError):
        module.write_release_artifacts(
            digest=digest,
            source_revision=revision,
            build_number="1",
            destination=destination,
            kustomization=manifest,
            output_dir=tmp_path / "out",
        )


def test_release_renderer_fails_closed_on_manifest_drift(tmp_path: Path) -> None:
    """Missing, duplicate, or already-current image entries require human review."""
    module = _load_release_module()
    digest = "sha256:" + "f" * 64
    missing = "images:\n  - name: example.invalid/other\n    digest: " + digest + "\n"
    with pytest.raises(ValueError, match="found 0"):
        module.render_kustomization(missing, digest)

    duplicate = (
        "images:\n"
        f"  - name: {module.DEFAULT_IMAGE}\n    digest: {digest}\n"
        f"  - name: {module.DEFAULT_IMAGE}\n    digest: {digest}\n"
    )
    with pytest.raises(ValueError, match="found 2"):
        module.render_kustomization(duplicate, digest)

    manifest = tmp_path / "kustomization.yaml"
    manifest.write_text(
        f"images:\n  - name: {module.DEFAULT_IMAGE}\n    digest: {digest}\n",
        encoding="utf-8",
    )
    revision = "b" * 40
    with pytest.raises(ValueError, match="already matches"):
        module.write_release_artifacts(
            digest=digest,
            source_revision=revision,
            build_number="1",
            destination=f"{module.DEFAULT_IMAGE}:git-{revision}-build-1",
            kustomization=manifest,
            output_dir=tmp_path / "out",
        )


def test_release_cli_reads_digest_file(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The pipeline CLI uses the Kaniko digest file as its sole digest input."""
    module = _load_release_module()
    digest = "sha256:" + "c" * 64
    revision = "d" * 40
    manifest = tmp_path / "kustomization.yaml"
    manifest.write_text(
        f"images:\n  - name: {module.DEFAULT_IMAGE}\n    digest: sha256:"
        + "0" * 64
        + "\n",
        encoding="utf-8",
    )
    digest_file = tmp_path / "digest"
    digest_file.write_text(digest + "\n", encoding="utf-8")
    destination = f"{module.DEFAULT_IMAGE}:git-{revision}-build-9"
    image_file = tmp_path / "image"
    image_file.write_text(f"{destination}@{digest}\n", encoding="utf-8")
    output = tmp_path / "out"
    monkeypatch.setenv("HARBOR_USER", "robot")
    monkeypatch.setenv("HARBOR_PASSWORD", "secret")
    monkeypatch.setattr(
        module, "verify_registry_digest", lambda *_args, **_kwargs: None
    )
    monkeypatch.setattr(
        "sys.argv",
        [
            "hermes_image_release.py",
            "render",
            "--digest-file",
            str(digest_file),
            "--image-file",
            str(image_file),
            "--source-revision",
            revision,
            "--build-number",
            "9",
            "--destination",
            destination,
            "--kustomization",
            str(manifest),
            "--output-dir",
            str(output),
        ],
    )
    assert module.main() == 0
    assert (
        json.loads((output / "hermes-agent-image.json").read_text(encoding="utf-8"))[
            "digest"
        ]
        == digest
    )