"""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 "git fetch --no-tags origin main" not in source assert 'git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}"' in source assert 'git checkout --detach "${EXPECTED_SOURCE_REVISION}"' in source assert 'test "${actual_revision}" = "${EXPECTED_SOURCE_REVISION}"' 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 "booleanParam('PUBLISH_IMAGE', false" in block assert "stringParam('EXPECTED_SOURCE_REVISION', ''" in block assert "stringParam('CONFIRM_PUBLISH', ''" 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, router, webui, stt, or tts"): module.trigger_build( "b" * 40, component="other", token_file=token_path, opener=opener ) @pytest.mark.parametrize( ("component", "confirmation"), [("stt", "PUBLISH HERMES STT"), ("tts", "PUBLISH HERMES TTS")], ) def test_agent_trigger_binds_private_voice_component( tmp_path: Path, component: str, confirmation: str ) -> None: """Voice releases select one fixed Jenkins job and component parameter.""" 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/85/"} def __enter__(self): return self def __exit__(self, *_args): self.close() def opener(request, timeout): assert timeout == 20 captured["request"] = request return Response(b"") result = module.trigger_build( "c" * 40, component=component, token_file=token_path, opener=opener ) fields = urllib.parse.parse_qs(captured["request"].data.decode("utf-8")) assert fields["job"] == ["hermes-voice-image"] assert fields["IMAGE_COMPONENT"] == [component] assert fields["CONFIRM_PUBLISH"] == [confirmation] assert result["component"] == component 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 )