atlas-iac/testing/tests/test_hermes_image_builder_vault.py

"""Vault least-privilege and fail-closed seeding tests for image releases."""

from __future__ import annotations

import os
import stat
import subprocess
from pathlib import Path

import pytest
import yaml


REPO_ROOT = Path(__file__).resolve().parents[2]
VAULT_CONFIG = REPO_ROOT / "services/vault/scripts/vault_k8s_auth_configure.sh"
SEEDER = (
    REPO_ROOT
    / "services/vault-hermes-jenkins-token-seed/scripts/vault_hermes_jenkins_build_token_ensure.sh"
)


def test_developer_jenkins_access_is_read_only_and_bound_to_two_consumers() -> None:
    """Only the Jenkins controller and Hermes agent can read the fixed-job token."""
    source = VAULT_CONFIG.read_text(encoding="utf-8")
    assert 'write_policy_and_role "jenkins" "jenkins" "jenkins"' in source
    assert (
        '"jenkins/* shared/harbor-pull quality/sonarqube-oidc '
        'hermes/developer-jenkins" ""'
    ) in source
    assert 'write_policy_and_role "hermes-agent" "hermes" "hermes-agent"' in source
    assert 'hermes/developer-jenkins hermes/developer-ssh" ""' in source
    assert 'write_policy_and_role "hermes-switchyard" "hermes"' in source
    assert '"hermes/chat-telegram" ""' in source
    assert '"hermes/developer-jenkins"' not in source

    jenkins_spc = yaml.safe_load(
        (REPO_ROOT / "services/jenkins/secretproviderclass.yaml").read_text()
    )
    assert jenkins_spc["spec"]["parameters"]["roleName"] == "jenkins-vault-sync"
    switchyard = yaml.safe_load(
        (REPO_ROOT / "services/hermes/switchyard-deployment.yaml").read_text()
    )
    annotations = switchyard["spec"]["template"]["metadata"]["annotations"]
    assert annotations["vault.hashicorp.com/role"] == "hermes-switchyard"


def test_flux_orders_seed_then_jenkins_then_hermes() -> None:
    """The fixed token must be Ready before either consumer is rolled."""
    app_root = REPO_ROOT / "clusters/atlas/flux-system/applications"
    vault = yaml.safe_load((app_root / "vault/kustomization.yaml").read_text())
    seed = yaml.safe_load(
        (app_root / "vault-hermes-jenkins-token-seed/kustomization.yaml").read_text()
    )
    jenkins = yaml.safe_load((app_root / "jenkins/kustomization.yaml").read_text())
    hermes = yaml.safe_load((app_root / "hermes/kustomization.yaml").read_text())
    role_check = {
        "apiVersion": "batch/v1",
        "kind": "Job",
        "name": "vault-k8s-auth-hermes-10",
        "namespace": "vault",
    }
    seed_check = {
        "apiVersion": "batch/v1",
        "kind": "Job",
        "name": "vault-hermes-jenkins-build-token-seed-2",
        "namespace": "vault",
    }
    assert role_check in vault["spec"]["healthChecks"]
    assert {item["name"] for item in seed["spec"]["dependsOn"]} == {"vault"}
    assert seed_check in seed["spec"]["healthChecks"]
    assert jenkins["spec"]["suspend"] is False
    assert {item["name"] for item in jenkins["spec"]["dependsOn"]} >= {
        "helm",
        "harbor",
        "vault-hermes-jenkins-token-seed",
    }
    assert "jenkins" in {item["name"] for item in hermes["spec"]["dependsOn"]}


def test_seed_job_is_revisioned_bounded_and_tracks_fail_closed_script() -> None:
    """The prerequisite is a tracked one-shot on healthy non-reserved capacity."""
    job = yaml.safe_load(
        (
            REPO_ROOT / "services/vault-hermes-jenkins-token-seed/job.yaml"
        ).read_text()
    )
    assert job["metadata"]["name"] == "vault-hermes-jenkins-build-token-seed-2"
    pod = job["spec"]["template"]["spec"]
    assert pod["serviceAccountName"] == "hermes-jenkins-token-seed"
    assert pod["enableServiceLinks"] is False
    assert pod["restartPolicy"] == "Never"
    assert pod["nodeSelector"]["hardware"] == "rpi5"
    expression = pod["affinity"]["nodeAffinity"][
        "requiredDuringSchedulingIgnoredDuringExecution"
    ]["nodeSelectorTerms"][0]["matchExpressions"][0]
    assert set(expression["values"]) >= {
        "titan-04",
        "titan-14",
        "titan-18",
        "titan-19",
        "titan-24",
    }
    container = pod["containers"][0]
    assert "@sha256:" in container["image"]
    security = container["securityContext"]
    assert security["runAsNonRoot"] is True
    assert security["readOnlyRootFilesystem"] is True
    assert security["allowPrivilegeEscalation"] is False
    assert security["capabilities"]["drop"] == ["ALL"]
    source = SEEDER.read_text(encoding="utf-8")
    assert '"options":{"cas":0}' in source
    assert 'vault kv get -field=build_token "${secret_path}"' in source
    assert "kv/data/atlas/hermes/developer-jenkins" in source
    assert '"build_token"[[:space:]]' not in source
    assert "kv patch" not in source
    assert "kv delete" not in source
    assert SEEDER.name not in VAULT_CONFIG.read_text(encoding="utf-8")


def test_seed_vault_policy_is_exact_create_read_and_rng_only() -> None:
    """The seed identity cannot rotate, delete, list, or touch unrelated KV."""
    source = VAULT_CONFIG.read_text(encoding="utf-8")
    policy = source.split("hermes_jenkins_token_seed_policy='", 1)[1].split("'", 1)[0]
    assert 'path "kv/data/atlas/hermes/developer-jenkins"' in policy
    assert 'capabilities = ["create", "read"]' in policy
    assert 'path "sys/tools/random/32"' in policy
    assert 'capabilities = ["update"]' in policy
    for forbidden in ("delete", "patch", "list", 'kv/data/atlas/hermes/*'):
        assert forbidden not in policy
    assert 'bound_service_account_names="hermes-jenkins-token-seed"' in source
    assert 'bound_service_account_namespaces="vault"' in source


def _fake_vault_tools(tmp_path: Path) -> tuple[Path, Path, Path]:
    """Create deterministic Vault/sleep fakes and return their evidence paths."""
    binary_dir = tmp_path / "bin"
    binary_dir.mkdir(parents=True)
    calls = tmp_path / "calls"
    stdin_capture = tmp_path / "stdin"
    vault = binary_dir / "vault"
    vault.write_text(
        """#!/bin/sh
set -eu
printf '%s\n' "$*" >> "$FAKE_CALL_LOG"
scenario="$FAKE_SCENARIO"
if [ "$1" = "write" ] && [ "$2" = "-field=random_bytes" ] && [ "$3" = "sys/tools/random/32" ]; then
  printf '%064d\n' 0
  exit 0
fi
if [ "$1" = "kv" ] && [ "$2" = "get" ] && [ "$3" = "-field=build_token" ]; then
  case "$scenario" in
    existing)
      printf '%064d\n' 1
      exit 0
      ;;
    cas-race)
      printf '%064d\n' 2
      exit 0
      ;;
    field-prefix)
      printf 'prefix%064d\n' 3
      exit 0
      ;;
    field-suffix)
      printf '%064dsuffix\n' 4
      exit 0
      ;;
    field-multiline)
      printf '%064d\nextra\n' 5
      exit 0
      ;;
    missing-field|metadata-only|custom-metadata)
      echo 'No value found at build_token' >&2
      exit 2
      ;;
  esac
fi
if [ "$1" = "write" ] && [ "$2" = "kv/data/atlas/hermes/developer-jenkins" ]; then
  payload="${3#@}"
  cp "$payload" "$FAKE_STDIN_CAPTURE"
  if [ "$scenario" = "cas-race" ]; then
    : > "$FAKE_WINNER"
    echo 'check-and-set parameter did not match' >&2
    exit 2
  fi
  exit 0
fi
if [ "$1" = "read" ] && [ "$2" = "-format=json" ]; then
  case "$scenario" in
    existing)
      printf '{"data":{"data":{"build_token":"%064d"},"metadata":{"build_token":"decoy"}}}\n' 1
      exit 0
      ;;
    missing-field)
      printf '{"data":{"data":{"other":"preserve-me"}}}\n'
      exit 0
      ;;
    metadata-only)
      printf '{"data":{"data":{"other":"preserve-me"},"metadata":{"build_token":"%064d"}}}\n' 6
      exit 0
      ;;
    custom-metadata)
      printf '{"data":{"data":{"other":"preserve-me"},"metadata":{"custom_metadata":{"build_token":"%064d"}}}}\n' 7
      exit 0
      ;;
    field-prefix|field-suffix|field-multiline)
      printf '{"data":{"data":{"build_token":"present"}}}\n'
      exit 0
      ;;
    absent)
      echo 'Code: 404' >&2
      exit 2
      ;;
    cas-race)
      if [ -f "$FAKE_WINNER" ]; then
        printf '{"data":{"data":{"build_token":"%064d"}}}\n' 2
        exit 0
      fi
      echo 'Code: 404' >&2
      exit 2
      ;;
    transient)
      count=0
      if [ -f "$FAKE_COUNT" ]; then count="$(cat "$FAKE_COUNT")"; fi
      count=$((count + 1))
      printf '%s' "$count" > "$FAKE_COUNT"
      if [ "$count" -lt 3 ]; then echo 'temporary upstream error' >&2; exit 2; fi
      echo 'Code: 404' >&2
      exit 2
      ;;
    read-error)
      echo 'permission denied' >&2
      exit 2
      ;;
  esac
fi
echo "unexpected fake Vault call: $*" >&2
exit 99
""",
        encoding="utf-8",
    )
    sleep = binary_dir / "sleep"
    sleep.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
    for executable in (vault, sleep):
        executable.chmod(executable.stat().st_mode | stat.S_IXUSR)
    return binary_dir, calls, stdin_capture


def _run_seeder(
    tmp_path: Path, scenario: str
) -> tuple[subprocess.CompletedProcess, str]:
    binary_dir, calls, stdin_capture = _fake_vault_tools(tmp_path)
    env = {
        **os.environ,
        "PATH": f"{binary_dir}:{os.environ['PATH']}",
        "VAULT_TOKEN": "test-token",
        "FAKE_SCENARIO": scenario,
        "FAKE_CALL_LOG": str(calls),
        "FAKE_STDIN_CAPTURE": str(stdin_capture),
        "FAKE_WINNER": str(tmp_path / "winner"),
        "FAKE_COUNT": str(tmp_path / "count"),
        "TMPDIR": str(tmp_path),
    }
    result = subprocess.run(
        ["sh", str(SEEDER)],
        env=env,
        text=True,
        capture_output=True,
        check=False,
    )
    call_text = calls.read_text(encoding="utf-8") if calls.exists() else ""
    return result, call_text


@pytest.mark.parametrize(
    "scenario",
    ["existing", "missing-field", "metadata-only", "custom-metadata", "read-error"],
)
def test_seeder_never_overwrites_existing_or_ambiguous_state(
    tmp_path: Path, scenario: str
) -> None:
    """Existing fields and persistent read failures can never become a put."""
    result, calls = _run_seeder(tmp_path, scenario)
    assert (result.returncode == 0) is (scenario == "existing")
    assert "write kv/data/atlas/hermes/developer-jenkins" not in calls
    assert "sys/tools/random" not in calls


@pytest.mark.parametrize("scenario", ["field-prefix", "field-suffix", "field-multiline"])
def test_seeder_requires_the_entire_exact_data_field(
    tmp_path: Path, scenario: str
) -> None:
    """Prefix, suffix, or multiline values cannot satisfy readiness."""
    result, calls = _run_seeder(tmp_path, scenario)
    assert result.returncode != 0
    assert "kv get -field=build_token kv/atlas/hermes/developer-jenkins" in calls
    assert "write kv/data/atlas/hermes/developer-jenkins" not in calls
    assert "sys/tools/random" not in calls


def test_seeder_uses_single_cas_create_and_stdin_for_absent_secret(
    tmp_path: Path,
) -> None:
    """A confirmed 404 produces one create-only write without a token argument."""
    result, calls = _run_seeder(tmp_path, "absent")
    assert result.returncode == 0, result.stderr
    assert calls.count("write kv/data/atlas/hermes/developer-jenkins") == 1
    assert "@" in calls
    assert "0000000000000000000000000000000000000000000000000000000000000000" not in calls
    payload = (tmp_path / "stdin").read_text(encoding="utf-8")
    assert '"options":{"cas":0}' in payload
    assert '"build_token":"' in payload
    assert not (tmp_path / "hermes-jenkins-token.json").exists()


def test_seeder_retries_reads_and_accepts_only_a_verified_cas_winner(
    tmp_path: Path,
) -> None:
    """Transient reads retry; a competing create is accepted only after reread."""
    transient, transient_calls = _run_seeder(tmp_path / "transient", "transient")
    assert transient.returncode == 0, transient.stderr
    assert transient_calls.count("read -format=json kv/data/atlas/hermes/developer-jenkins") == 3
    assert transient_calls.count("write kv/data/atlas/hermes/developer-jenkins") == 1

    race, race_calls = _run_seeder(tmp_path / "race", "cas-race")
    assert race.returncode == 0, race.stderr
    assert race_calls.count("write kv/data/atlas/hermes/developer-jenkins") == 1
    assert race_calls.count("read -format=json kv/data/atlas/hermes/developer-jenkins") == 2
    assert "another seeder won CAS" in race.stderr