atlas-iac/testing/tests/test_hermes_runtime_access.py

"""Contracts for Hermes' Vault-only runtime access boundary."""

from __future__ import annotations

import importlib.util
import hashlib
import hmac
import json
import sys
import urllib.request
from pathlib import Path

import pytest
import yaml

ROOT = Path(__file__).parents[2]
HERMES = ROOT / "services" / "hermes"
SCRIPTS = HERMES / "scripts"
SCM_SCRIPTS = HERMES / "scm-common" / "scripts"
if str(SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SCRIPTS))
if str(SCM_SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SCM_SCRIPTS))


def _load(name: str):
    root = SCM_SCRIPTS if name == "gitea_api" else SCRIPTS
    spec = importlib.util.spec_from_file_location(name, root / f"{name}.py")
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def test_gitea_api_builds_runtime_authenticated_same_origin_requests():
    gitea_api = _load("gitea_api")

    request = gitea_api.build_request(
        "POST",
        "/api/v1/repos/atlas/cassandra/pulls",
        base_url="https://scm.bstein.dev",
        token="runtime-only-value",
        data={
            "base": "main",
            "body": "Ready for review.",
            "head": "hermes/repair",
            "title": "WIP: Repair semantic review findings",
        },
    )

    assert isinstance(request, urllib.request.Request)
    assert request.full_url == "https://scm.bstein.dev/api/v1/repos/atlas/cassandra/pulls"
    assert request.method == "POST"
    assert request.get_header("Authorization") == "token runtime-only-value"
    assert json.loads(request.data)["title"].startswith("WIP: ")
    assert "draft" not in json.loads(request.data)


@pytest.mark.parametrize(
    "path",
    [
        "https://evil.example/api/v1/repos/atlas/cassandra",
        "/repos/atlas/cassandra",
        "/api/v1/repos/atlas/cassandra#fragment",
    ],
)
def test_gitea_api_rejects_foreign_or_non_api_targets(path: str):
    gitea_api = _load("gitea_api")

    with pytest.raises(ValueError):
        gitea_api.api_url("https://scm.bstein.dev", path)


@pytest.mark.parametrize("method", ["DELETE", "PUT", "OPTIONS"])
def test_gitea_api_rejects_unavailable_methods(method: str):
    gitea_api = _load("gitea_api")

    with pytest.raises(gitea_api.PolicyError):
        gitea_api.authorize_request(
            method, "/api/v1/repos/atlas/cassandra/pulls/1", None
        )


def test_jenkins_evidence_ignores_nested_execution_result():
    evidence_reader = _load("jenkins_build_evidence")
    evidence = evidence_reader.parse_build_xml(
        """
        <flow-build>
          <actions><revision><hash>abc123</hash></revision></actions>
          <timestamp>1000</timestamp><duration>0</duration>
          <execution><result>SUCCESS</result></execution>
        </flow-build>
        """,
        job="demo",
        branch="master",
        number=15,
    )

    assert evidence.revision == "abc123"
    assert evidence.result is None
    assert evidence.building is True


def test_jenkins_evidence_reads_only_top_level_terminal_result():
    evidence_reader = _load("jenkins_build_evidence")
    evidence = evidence_reader.parse_build_xml(
        """
        <flow-build>
          <actions><revision><hash>abc123</hash></revision></actions>
          <timestamp>1000</timestamp><duration>250</duration>
          <result>FAILURE</result>
          <execution><result>SUCCESS</result></execution>
        </flow-build>
        """,
        job="demo",
        branch=None,
        number=4,
    )

    assert evidence.result == "FAILURE"
    assert evidence.building is False
    assert evidence.duration_ms == 250


def test_jenkins_evidence_removes_hidden_links_and_ansi_control_sequences():
    evidence_reader = _load("jenkins_build_evidence")

    value = (
        "before\n\x1b[8mha:////private-jenkins-payload\x1b[0m[Pipeline] sh\n"
        "\x1b[31m5 passed\x1b[0m\n"
    )

    assert evidence_reader.sanitize_log(value) == "before\n[Pipeline] sh\n5 passed\n"


def test_jenkins_evidence_falls_back_to_multibranch_suffix(monkeypatch):
    evidence_reader = _load("jenkins_build_evidence")
    calls = []

    def fake_find(job, branch, commit, log_lines):
        calls.append((job, branch, commit, log_lines))
        if job == "demo":
            raise ValueError("not a multibranch job")
        return "evidence"

    monkeypatch.setattr(evidence_reader, "_find_build_for_job", fake_find)

    assert evidence_reader.find_build("demo", "master", "abc", 20) == "evidence"
    assert [call[0] for call in calls] == ["demo", "demo-branches"]


@pytest.mark.parametrize("branch", ["../master", "/master", "feature//unsafe"])
def test_jenkins_evidence_rejects_unsafe_branch_paths(branch: str):
    evidence_reader = _load("jenkins_build_evidence")

    with pytest.raises(ValueError):
        evidence_reader._validate_branch(branch)


def test_agent_runtime_stage_keeps_credentials_in_memory(tmp_path: Path, monkeypatch):
    stage = _load("stage_runtime_access")
    vault = tmp_path / "vault"
    runtime = tmp_path / "runtime"
    home = tmp_path / "home"
    vault.mkdir()
    (home / ".claude").mkdir(parents=True)
    (home / ".claude" / "backups").mkdir()
    (home / ".codex" / "skills").mkdir(parents=True)
    (home / ".codex" / "sessions").mkdir()
    (home / ".claude" / "settings.json").write_text("{}\n", encoding="utf-8")
    (home / ".claude" / "backups" / "credentials.json").write_text(
        "do-not-link\n", encoding="utf-8"
    )
    values = {
        "agent-api-key": "agent-key",
        "execution-pool-key": "e" * 64,
        "chat-relay-key": "relay-key",
        # gitea-token/username are present in Vault but the agent staging must
        # deliberately not copy them (the least-authority broker holds Gitea
        # access); the assertions below prove they never reach the runtime.
        "gitea-token": "gitea-key",
        "gitea-username": "hermes-automation",
        "jenkins-image-build-token": "job-scoped-token",
        "node-ssh-private-key": "private-key",
        "node-ssh-config": "host-config",
        "node-ssh-known-hosts": "known-hosts",
        "claude-credentials": json.dumps(
            {"claudeAiOauth": {"refreshToken": "claude-refresh"}}
        ),
        "codex-auth": json.dumps({"tokens": {"refresh_token": "codex-refresh"}}),
    }
    for name, value in values.items():
        (vault / name).write_text(value + "\n", encoding="utf-8")
    monkeypatch.setattr(stage, "VAULT_ROOT", vault)
    monkeypatch.setattr(stage, "RUNTIME_ROOT", runtime)
    monkeypatch.setattr(stage, "PERSISTENT_HOME", home)
    monkeypatch.setattr(stage.os, "chown", lambda *_args: None)

    stage.stage_agent()

    assert (runtime / "claude/.credentials.json").stat().st_mode & 0o777 == 0o600
    assert (runtime / "codex/auth.json").stat().st_mode & 0o777 == 0o600
    assert (runtime / "jenkins-image-build-token").stat().st_mode & 0o777 == 0o600
    assert (runtime / "claude/settings.json").is_symlink()
    assert (runtime / "codex/skills").is_symlink()
    assert not (runtime / "claude/backups").exists()
    assert not (runtime / "codex/sessions").exists()
    assert not (runtime / "gitea-token").exists()
    assert not (runtime / "gitea-username").exists()
    ssh_config = (runtime / "node-ssh-config").read_text(encoding="utf-8")
    assert ssh_config.startswith("Host titan-*\n  User hermes-agent\n")
    auth = json.loads((runtime / "hermes-auth.json").read_text(encoding="utf-8"))
    assert auth == {"version": 1, "providers": {}, "credential_pool": {}}


def test_execution_worker_and_mediator_separate_credentials_and_hmac(
    tmp_path: Path, monkeypatch
):
    stage = _load("stage_runtime_access")
    vault = tmp_path / "vault"
    runtime = tmp_path / "runtime"
    worker = tmp_path / "worker"
    provider_access = tmp_path / "provider-access"
    pool_access = tmp_path / "pool-access"
    vault.mkdir()
    (vault / "claude-credentials-1").write_text(
        json.dumps({"claudeAiOauth": {"refreshToken": "claude-refresh"}})
    )
    (vault / "codex-auth-1").write_text(
        json.dumps({"tokens": {"refresh_token": "codex-refresh"}})
    )
    monkeypatch.setattr(stage, "VAULT_ROOT", vault)
    monkeypatch.setattr(stage, "RUNTIME_ROOT", runtime)
    monkeypatch.setattr(stage, "WORKER_ROOT", worker)
    monkeypatch.setattr(stage, "PROVIDER_ACCESS_ROOT", provider_access)
    monkeypatch.setattr(stage, "POOL_ACCESS_ROOT", pool_access)
    monkeypatch.setenv("HERMES_WORKER_ORDINAL", "1")
    monkeypatch.setattr(stage.os, "chown", lambda *_args: None)
    monkeypatch.setattr(stage.os, "fchown", lambda *_args: None)

    stage.stage_execution_worker()

    assert not pool_access.exists()
    assert not (runtime / "execution-pool-key").exists()
    assert not (runtime / "codex/sessions").exists()
    assert not (runtime / "claude/projects").exists()
    assert (provider_access / "codex/auth.json").stat().st_mode & 0o777 == 0o600
    refreshed = {"tokens": {"refresh_token": "provider-rotated"}}
    (provider_access / "codex/auth.json").write_text(json.dumps(refreshed))
    (provider_access / "codex/auth.json").chmod(0o600)
    stage.stage_execution_worker()
    assert json.loads((provider_access / "codex/auth.json").read_text()) == refreshed

    master = "e" * 64
    (vault / "execution-pool-key").write_text(master)
    stage.stage_execution_mediator()
    expected = hmac.new(
        master.encode(), b"hermes-execution-pool-v2:worker:1", hashlib.sha256
    ).hexdigest()
    assert (pool_access / "execution-pool-key").read_text().strip() == expected


def test_execution_worker_fails_closed_without_channel_credential(
    tmp_path: Path, monkeypatch
):
    stage = _load("stage_runtime_access")
    vault = tmp_path / "vault"
    vault.mkdir()
    (vault / "claude-credentials-0").write_text(
        json.dumps({"claudeAiOauth": {"refreshToken": "claude-refresh"}})
    )
    monkeypatch.setattr(stage, "VAULT_ROOT", vault)
    monkeypatch.setattr(stage, "RUNTIME_ROOT", tmp_path / "runtime")
    monkeypatch.setattr(stage, "WORKER_ROOT", tmp_path / "worker")
    monkeypatch.setattr(stage, "PROVIDER_ACCESS_ROOT", tmp_path / "provider-access")
    monkeypatch.setenv("HERMES_WORKER_ORDINAL", "0")
    monkeypatch.setattr(stage.os, "chown", lambda *_args: None)
    monkeypatch.setattr(stage.os, "fchown", lambda *_args: None)

    with pytest.raises(FileNotFoundError):
        stage.stage_execution_worker()


def test_execution_worker_rejects_durable_credential_symlink(
    tmp_path: Path, monkeypatch
):
    stage = _load("stage_runtime_access")
    provider = tmp_path / "provider"
    provider.mkdir()
    (provider / "claude").symlink_to(tmp_path, target_is_directory=True)
    monkeypatch.setattr(stage, "WORKER_ROOT", tmp_path / "worker")
    monkeypatch.setattr(stage, "PROVIDER_ACCESS_ROOT", provider)
    monkeypatch.setenv("HERMES_WORKER_ORDINAL", "0")
    monkeypatch.setattr(stage.os, "chown", lambda *_args: None)
    with pytest.raises(RuntimeError, match="symlink"):
        stage.stage_execution_worker()


def test_invalid_runtime_json_is_removed(tmp_path: Path, monkeypatch):
    stage = _load("stage_runtime_access")
    vault = tmp_path / "vault"
    runtime = tmp_path / "runtime"
    vault.mkdir()
    runtime.mkdir()
    (vault / "credential").write_text("not-json\n", encoding="utf-8")
    monkeypatch.setattr(stage, "VAULT_ROOT", vault)
    monkeypatch.setattr(stage.os, "chown", lambda *_args: None)
    destination = runtime / "credential.json"

    try:
        stage._validated_json("credential", destination, ("token",))
    except RuntimeError:
        pass
    else:
        raise AssertionError("invalid credential JSON should fail staging")

    assert not destination.exists()


def test_runtime_refresh_sync_uses_cas_and_preserves_other_fields(
    tmp_path: Path, monkeypatch
):
    sync = _load("sync_runtime_credentials")
    claude = tmp_path / "claude.json"
    codex = tmp_path / "codex.json"
    claude.write_text(
        json.dumps({"claudeAiOauth": {"refreshToken": "new-claude"}}),
        encoding="utf-8",
    )
    codex.write_text(
        json.dumps({"tokens": {"refresh_token": "new-codex"}}),
        encoding="utf-8",
    )
    monkeypatch.setattr(
        sync,
        "CREDENTIALS",
        {
            "claude_credentials_json": (
                claude,
                ("claudeAiOauth", "refreshToken"),
            ),
            "codex_auth_json": (codex, ("tokens", "refresh_token")),
        },
    )
    writes = []

    def request(method, path, payload=None, *, token=""):
        assert token == "vault-token"
        if method == "GET":
            return {
                "data": {
                    "data": {
                        "claude_credentials_json": "old",
                        "codex_auth_json": "old",
                        "agent_api_key": "preserve-me",
                    },
                    "metadata": {"version": 7},
                }
            }
        writes.append((path, payload))
        return {}

    monkeypatch.setattr(sync, "_request", request)

    assert sync.sync_once("vault-token") == [
        "claude_credentials_json",
        "codex_auth_json",
    ]
    assert writes[0][1]["options"] == {"cas": 7}
    assert writes[0][1]["data"]["agent_api_key"] == "preserve-me"


def test_subprocess_patches_strip_and_redact_runtime_credentials(tmp_path: Path):
    boundary = _load("patch_subprocess_secret_boundary")
    process = _load("patch_process_output_redaction")
    local_source = tmp_path / "local.py"
    local_output = tmp_path / "patched-local.py"
    local_source.write_text("prefix\n" + boundary.BEFORE + "suffix\n", encoding="utf-8")
    boundary.patch(local_source, local_output)
    patched_local = local_output.read_text(encoding="utf-8")
    for name in (
        "API_SERVER_KEY",
        "CLAUDE_CODE_OAUTH_TOKEN",
        "GITEA_TOKEN",
        "HERMES_IMAGE_BROKER_KEY",
    ):
        assert f'"{name}"' in patched_local

    process_source = tmp_path / "process.py"
    process_output = tmp_path / "patched-process.py"
    process_source.write_text(
        "prefix\n" + process.BEFORE + "suffix\n", encoding="utf-8"
    )
    process.patch(process_source, process_output)
    assert "code_file=False" in process_output.read_text(encoding="utf-8")


def test_manifests_never_seed_access_material_into_persistent_env():
    agent = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
    chat = yaml.safe_load((HERMES / "chat-statefulset.yaml").read_text())
    triage = yaml.safe_load((HERMES / "deployment.yaml").read_text())
    for workload in (agent, chat, triage):
        pod = workload["spec"]["template"]["spec"]
        runtime = next(item for item in pod["volumes"] if item["name"] == "runtime-access")
        assert runtime["emptyDir"]["medium"] == "Memory"
        assert not any(item["name"] == "provider-auth" for item in pod["volumes"])
        init = next(item for item in pod["initContainers"] if item["name"] == "init-config")
        command = init["command"][2]
        for key in (
            "ANTHROPIC_API_KEY",
            "API_SERVER_KEY",
            "CLAUDE_API_KEY",
            "CLAUDE_CODE_OAUTH_TOKEN",
            "GITEA_TOKEN",
            "HERMES_IMAGE_BROKER_KEY",
            "OPENAI_API_KEY",
        ):
            assert f"printf '{key}=" not in command
            assert f"upsert_env {key}" not in command

    annotations = agent["spec"]["template"]["metadata"]["annotations"]
    agent_runtime = next(
        item
        for item in agent["spec"]["template"]["spec"]["volumes"]
        if item["name"] == "runtime-access"
    )
    assert agent_runtime["emptyDir"] == {
        "medium": "Memory",
        "sizeLimit": "128Mi",
    }
    assert "vault.hashicorp.com/agent-inject-secret-anthropic-token" not in annotations
    assert annotations["vault.hashicorp.com/agent-inject-secret-claude-credentials"] == (
        "kv/data/atlas/hermes/agent-tokens"
    )
    assert annotations["vault.hashicorp.com/agent-inject-secret-codex-auth"] == (
        "kv/data/atlas/hermes/agent-tokens"
    )
    credential_sync = next(
        item
        for item in agent["spec"]["template"]["spec"]["containers"]
        if item["name"] == "credential-sync"
    )
    assert {
        (item["mountPath"], item.get("subPath"))
        for item in credential_sync["volumeMounts"]
        if item["name"] == "runtime-access"
    } == {
        ("/runtime-access/claude", "claude"),
        ("/runtime-access/codex", "codex"),
    }
    triage_annotations = triage["spec"]["template"]["metadata"]["annotations"]
    assert "vault.hashicorp.com/agent-inject-secret-anthropic-token" not in (
        triage_annotations
    )
    assert triage_annotations[
        "vault.hashicorp.com/agent-inject-secret-triage-api-key"
    ] == "kv/data/atlas/hermes/triage-api"
    gitea_api = (
        ROOT / "services/hermes/scm-common/scripts/gitea_api.py"
    ).read_text(encoding="utf-8")
    assert "scm_broker_client" in gitea_api
    assert "GITEA_TOKEN" not in gitea_api
    assert "gitea-token" not in annotations


def test_chat_media_reads_one_raw_runtime_secret(tmp_path: Path):
    module = _load("telegram_media_server")
    secret = tmp_path / "relay"
    secret.write_text("relay-value\n", encoding="utf-8")
    assert module.read_relay_key(secret) == "relay-value"


def test_ariadne_receives_the_triage_key_directly_from_vault():
    maintenance = ROOT / "services" / "maintenance" / "apps" / "ariadne-deployment.yaml"
    deployment = yaml.safe_load(maintenance.read_text(encoding="utf-8"))
    annotations = deployment["spec"]["template"]["metadata"]["annotations"]
    template = annotations["vault.hashicorp.com/agent-inject-template-ariadne-env.sh"]
    assert 'secret "kv/data/atlas/hermes/triage-api"' in template
    assert 'export ARIADNE_HERMES_API_KEY="{{ .Data.data.api_key }}"' in template
    assert 'secret "kv/data/atlas/hermes/developer-gitea"' in template
    assert 'export ARIADNE_HERMES_GITEA_TOKEN="{{ .Data.data.token }}"' in template
    container = deployment["spec"]["template"]["spec"]["containers"][0]
    assert not any(item["name"] == "ARIADNE_HERMES_API_KEY" for item in container["env"])
    assert not any(
        item["name"] == "ARIADNE_HERMES_GITEA_TOKEN" for item in container["env"]
    )

    vault_policy = (
        ROOT / "services" / "vault" / "scripts" / "vault_k8s_auth_configure.sh"
    ).read_text(encoding="utf-8")
    assert '"hermes/triage-oidc hermes/agent-tokens hermes/triage-api"' in vault_policy
    assert 'hermes-credential-sync' in vault_policy
    assert 'hermes/triage-api hermes/developer-gitea' in vault_policy


def test_gitea_bootstrap_reuses_valid_vault_tokens():
    bootstrap = (
        ROOT / "services" / "gitea" / "scripts" / "gitea_atlas_identity_ensure.sh"
    ).read_text(encoding="utf-8")

    assert "vault_login\n\nreconciler_token=$(vault_read_field" in bootstrap
    assert 'if ! token_is_valid "${reconciler_user}"' in bootstrap
    assert 'if ! token_is_valid "${hermes_user}"' in bootstrap
    assert bootstrap.count('generate_token "${reconciler_user}"') == 1
    assert bootstrap.count('generate_token "${hermes_user}"') == 1