atlas-iac/testing/tests/test_hermes_agent_security.py

"""Agent authentication, networking, and Flux health contracts."""

from __future__ import annotations

from testing.tests.test_hermes_cli_support import (
    FLUX_HERMES,
    HERMES,
    KEYCLOAK,
    _agent_deployment,
    yaml,
)


def test_agent_image_runs_execution_safety_patch_and_regressions():
    dockerfiles = HERMES.parents[1] / "dockerfiles"
    dockerfile = (dockerfiles / "Dockerfile.hermes-agent").read_text(encoding="utf-8")
    dockerignore = (dockerfiles / "Dockerfile.hermes-agent.dockerignore").read_text(
        encoding="utf-8"
    )

    for name in (
        "patch-hermes-execution-safety.py",
        "hermes-execution-safety-regression.py",
        "hermes_execution_patch_support.py",
        "patch_hermes_run_safety.py",
        "patch_hermes_decomposition_safety.py",
        "hermes_execution_regression_support.py",
        "hermes_run_safety_regression.py",
        "hermes_decomposition_safety_regression.py",
    ):
        assert f"COPY dockerfiles/{name}" in dockerfile
        assert f"!dockerfiles/{name}" in dockerignore
    assert "/opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py" in (
        dockerfile
    )
    assert "/opt/hermes/.venv/bin/python /tmp/hermes-execution-safety-regression.py" in (
        dockerfile
    )
    assert "COPY services/hermes/scripts/cli_lane_*.py" in dockerfile
    assert "!services/hermes/scripts/cli_lane_*.py" in dockerignore
    assert "HERMES_CLI_LANE_SOURCE=/tmp/hermes-lane-regression" in dockerfile


def test_agent_refreshes_routes_after_restoring_cli_logins():
    deployment = _agent_deployment()
    init_containers = {
        item["name"]: item
        for item in deployment["spec"]["template"]["spec"]["initContainers"]
    }
    configure = init_containers["configure-agent-clients"]
    command = configure["command"][-1]
    assert "configure_agent_clients.py" in command
    assert command.index("configure_agent_clients.py") < command.index(
        "hermes_coordinator.py --once"
    )
    env = {item["name"]: item["value"] for item in configure["env"]}
    assert env["HERMES_AUTH_FILE"] == "/runtime-access/hermes-auth.json"
    assert env["PYTHONPATH"] == "/opt/hermes"
    for name in ("bootstrap-coordinator", "configure-agent-clients"):
        route_env = {
            item["name"]: item["value"]
            for item in init_containers[name]["env"]
        }
        assert route_env["CODEX_HOME"] == "/runtime-access/codex"
        assert route_env["CLAUDE_CONFIG_DIR"] == "/runtime-access/claude"
        assert "/opt/data/tools/bin" in route_env["PATH"]

    assert "patch-web-session-activity" in init_containers
    web_patch = init_containers["patch-web-session-activity"]
    assert "/opt/coordinator/patch_web_session_activity.py" in web_patch["command"]

    containers = {
        item["name"]: item
        for item in deployment["spec"]["template"]["spec"]["containers"]
    }
    steward_env = {
        item["name"]: item["value"]
        for item in containers["model-steward"]["env"]
    }
    assert steward_env["CODEX_HOME"] == "/runtime-access/codex"
    assert steward_env["CLAUDE_CONFIG_DIR"] == "/runtime-access/claude"
    assert "/opt/data/tools/bin" in steward_env["PATH"]
    hermes_mounts = {
        (item["name"], item["mountPath"], item.get("subPath"))
        for item in containers["hermes"]["volumeMounts"]
    }
    assert (
        "web-server-patch",
        "/opt/hermes/hermes_cli/web_server.py",
        "web_server.py",
    ) in hermes_mounts


def test_flux_health_checks_follow_the_owner_oauth_sidecar():
    flux = yaml.safe_load(FLUX_HERMES.read_text())
    checks = {
        (item["kind"], item["name"])
        for item in flux["spec"]["healthChecks"]
    }
    assert ("Deployment", "hermes-agent") in checks
    assert ("DaemonSet", "hermes-node-ssh-access") in checks
    assert ("Deployment", "oauth2-proxy-hermes-agent") not in checks


def test_agent_auth_is_bstein_group_and_email_bounded():
    deployment = _agent_deployment()
    oauth = next(
        item for item in deployment["spec"]["template"]["spec"]["containers"]
        if item["name"] == "oauth2-proxy"
    )
    args = oauth["args"]
    assert "--user-id-claim=sub" in args
    assert "--oidc-groups-claim=groups" in args
    assert "--allowed-group=/hermes-owner" in args
    assert "--authenticated-emails-file=/etc/oauth2-proxy/allowed-emails" in args
    script = (KEYCLOAK / "scripts/hermes_access_oidc_ensure.sh").read_text()
    assert 'group_name="hermes-owner"' in script
    assert "username=bstein&exact=true" in script
    assert '"full.path":"true"' in script


def test_agent_network_boundary_allows_only_authenticated_and_metrics_surfaces():
    documents = [
        item
        for item in yaml.safe_load_all((HERMES / "networkpolicy.yaml").read_text())
        if item
    ]
    isolation = next(item for item in documents if item.get("metadata", {}).get("name") == "hermes-agent-isolation")
    assert isolation["spec"]["ingress"] == [
        {
            "from": [
                {
                    "namespaceSelector": {
                        "matchLabels": {
                            "kubernetes.io/metadata.name": "traefik"
                        }
                    },
                    "podSelector": {
                        "matchLabels": {"app.kubernetes.io/name": "traefik"}
                    },
                }
            ],
            "ports": [{"protocol": "TCP", "port": 4180}],
        },
        {
            "from": [
                {
                    "podSelector": {
                        "matchLabels": {"app": "hermes-chat-tenant"}
                    }
                }
            ],
            "ports": [
                {"protocol": "TCP", "port": 9002},
                {"protocol": "TCP", "port": 9003},
            ],
        },
        {
            "from": [
                {
                    "podSelector": {
                        "matchLabels": {"app": "hermes-switchyard"}
                    }
                }
            ],
            "ports": [
                {"protocol": "TCP", "port": 9003},
                {"protocol": "TCP", "port": 9006},
            ],
        },
        {
            "from": [
                {
                    "namespaceSelector": {
                        "matchLabels": {
                            "kubernetes.io/metadata.name": "monitoring"
                        }
                    },
                    "podSelector": {"matchLabels": {"app": "server"}},
                }
            ],
            "ports": [{"protocol": "TCP", "port": 9010}],
        },
    ]
    assert isolation["spec"]["egress"] == [{}]


def test_owner_agent_has_cluster_admin_kubernetes_context():
    config = yaml.safe_load((HERMES / "agent-kubeconfig.yaml").read_text())
    assert config["current-context"] == "atlas-owner"
    assert config["contexts"][0]["context"]["namespace"] == "default"
    rbac_path = HERMES / "agent-rbac.yaml"
    documents = [item for item in yaml.safe_load_all(rbac_path.read_text()) if item]
    binding = next(item for item in documents if item["kind"] == "ClusterRoleBinding")
    assert binding["roleRef"] == {
        "apiGroup": "rbac.authorization.k8s.io",
        "kind": "ClusterRole",
        "name": "cluster-admin",
    }
    assert binding["subjects"] == [
        {"kind": "ServiceAccount", "name": "hermes-agent", "namespace": "hermes"}
    ]