"""Network, OAuth, Kubernetes, and SSH access contracts for Hermes CLI lanes.""" from __future__ import annotations import yaml from testing.tests.test_hermes_cli_lanes_support import ( FLUX_HERMES, HERMES, KEYCLOAK, ROOT, _agent_deployment, _services, ) def test_broker_services_survive_sibling_container_readiness_loss(): services = _services() for name in ( "hermes-image-broker", "hermes-codex-broker", "hermes-local-image", "hermes-claude-broker", ): assert services[name]["spec"]["publishNotReadyAddresses"] is True def test_agent_dashboard_reconnects_all_transient_websockets(): dockerfile = (HERMES.parents[1] / "dockerfiles/Dockerfile.hermes-agent").read_text( encoding="utf-8" ) assert "eventsRetryAttempt.current" in dockerfile assert "if (!unmounting) setVersion((v) => v + 1);" in dockerfile assert "events feed rejected (${ev.code}) — reload the page" in dockerfile assert 'url = await api.buildWsUrl("/api/pty", params);' in dockerfile assert 'url = await buildWsUrl("/api/events", { channel });' in dockerfile assert dockerfile.count("' await api.getSessions(1, 0,") == 2 assert ".then(() => gw.connect())" in dockerfile assert 'api.getSessions(1, 0, profile ?? "")' in dockerfile assert "dashboard token rotated by a server restart" 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_gate_the_owner_without_node_maintenance(): 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") not 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}, {"protocol": "TCP", "port": 9011}, ], }, ] egress = isolation["spec"]["egress"] assert any( rule.get("ports") == [{"protocol": "TCP", "port": 9081}] for rule in egress ) assert "hermes-scm" in yaml.safe_dump(egress) assert "gitea" in yaml.safe_dump(egress) assert egress != [{}] def test_owner_agent_has_scoped_read_only_kubernetes_context(): config = yaml.safe_load((HERMES / "agent-kubeconfig.yaml").read_text()) assert config["current-context"] == "atlas-observer" assert config["contexts"][0]["context"]["namespace"] == "default" rbac_path = ROOT / "services/hermes-observer-rbac/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": "hermes-agent-cluster-observer-v2", } assert binding["subjects"] == [ {"kind": "ServiceAccount", "name": "hermes-agent", "namespace": "hermes"} ] def test_owner_agent_has_pinned_dedicated_node_ssh_access(): deployment = _agent_deployment() annotations = deployment["spec"]["template"]["metadata"]["annotations"] assert ( annotations["vault.hashicorp.com/agent-inject-secret-node-ssh-private-key"] == "kv/data/atlas/hermes/developer-ssh" ) assert ( annotations["vault.hashicorp.com/agent-inject-secret-node-ssh-config"] == "kv/data/atlas/hermes/developer-ssh" ) assert ( annotations["vault.hashicorp.com/agent-inject-secret-node-ssh-known-hosts"] == "kv/data/atlas/hermes/developer-ssh" ) init = next( item for item in deployment["spec"]["template"]["spec"]["initContainers"] if item["name"] == "init-config" ) command = init["command"][2] assert "ln -s /runtime-access/node-ssh-config /opt/data/home/.ssh/config" in command assert ( "ln -s /runtime-access/node-ssh-known-hosts /opt/data/home/.ssh/known_hosts" in command ) assert "ln -s home/.ssh /opt/data/.ssh" in command assert "chmod 0700 /opt/data/home/.ssh" in command assert ( "ln -s /runtime-access/node-ssh-private-key " "/opt/data/home/.ssh/id_ed25519_atlas_nodes" ) in command config = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())["data"] assert "ssh_config" not in config assert "ssh_known_hosts" not in config assert "ssh-ed25519" not in (HERMES / "agent-configmap.yaml").read_text() resources = yaml.safe_load((HERMES / "kustomization.yaml").read_text())["resources"] assert "node-ssh-access.yaml" in resources access = [ item for item in yaml.safe_load_all((HERMES / "node-ssh-access.yaml").read_text()) if item ] service_account = next(item for item in access if item["kind"] == "ServiceAccount") assert service_account["metadata"]["name"] == "hermes-node-ssh-access" provider = next(item for item in access if item["kind"] == "SecretProviderClass") assert provider["metadata"]["name"] == "hermes-node-ssh-access" assert provider["spec"]["provider"] == "vault" parameters = provider["spec"]["parameters"] assert parameters["roleName"] == "hermes-node-ssh" assert 'secretPath: "kv/data/atlas/hermes/developer-ssh"' in parameters["objects"] assert 'secretKey: "public_key"' in parameters["objects"] daemonset = next(item for item in access if item["kind"] == "DaemonSet") pod = daemonset["spec"]["template"]["spec"] assert pod["serviceAccountName"] == "hermes-node-ssh-access" assert pod["automountServiceAccountToken"] is True host_home = next(item for item in pod["volumes"] if item["name"] == "host-home") assert host_home["hostPath"] == {"path": "/home", "type": "Directory"} vault_secrets = next( item for item in pod["volumes"] if item["name"] == "vault-secrets" ) assert vault_secrets["csi"]["driver"] == "secrets-store.csi.k8s.io" assert vault_secrets["csi"]["volumeAttributes"] == { "secretProviderClass": "hermes-node-ssh-access" } reconciler = pod["containers"][0]["args"][0] assert "/opt/node-hardener/node_account_hardening.py" in reconciler assert "--public-key-file /vault/secrets/node-ssh-public-key" in reconciler assert "sleep 300" in reconciler host_etc = next(item for item in pod["volumes"] if item["name"] == "host-etc") assert host_etc["hostPath"] == {"path": "/etc", "type": "Directory"} hardener = (HERMES / "scripts/node_account_hardening.py").read_text() assert 'ACCOUNT = "hermes-agent"' in hardener assert 'LEGACY_ACCOUNTS = ("atlas", "oceanus")' in hardener assert "ACCOUNT_UID = 1200" in hardener assert "ACCOUNT_GID = 1200" in hardener def test_owner_agent_tracks_no_ssh_identity_or_host_key_material(): """Vault references may be tracked; SSH identities and trust data may not.""" forbidden = ( "BEGIN OPENSSH PRIVATE KEY", "ssh-ed25519 AAAA", "ssh-rsa AAAA", "IdentityFile ", "UserKnownHostsFile ", "StrictHostKeyChecking ", "ssh_config:", "ssh_known_hosts:", ) text_suffixes = { ".conf", ".json", ".md", ".py", ".sh", ".toml", ".yaml", ".yml", } tracked = "\n".join( path.read_text(encoding="utf-8") for path in HERMES.rglob("*") if path.is_file() and path.suffix in text_suffixes ) for marker in forbidden: assert marker not in tracked def test_switchyard_has_a_dedicated_non_owner_identity_and_read_only_catalog(): """Routing must not inherit the owner agent's cluster-admin capability.""" service_accounts = [ item for item in yaml.safe_load_all( (HERMES / "vault-serviceaccount.yaml").read_text() ) if item ] assert any( item["kind"] == "ServiceAccount" and item["metadata"]["name"] == "hermes-switchyard" for item in service_accounts ) switchyard = yaml.safe_load((HERMES / "switchyard-deployment.yaml").read_text()) switchyard_pod = switchyard["spec"]["template"]["spec"] assert switchyard_pod["serviceAccountName"] == "hermes-switchyard" agent_pod = _agent_deployment()["spec"]["template"]["spec"] for pod, container_name in ( (switchyard_pod, "worker-route-broker"), (agent_pod, "claude-broker"), ): container = next( item for item in pod["containers"] if item["name"] == container_name ) catalog = next( item for item in container["volumeMounts"] if item["mountPath"] == "/routing-catalog" ) assert catalog["readOnly"] is True rbac = [ item for item in yaml.safe_load_all( (ROOT / "services/hermes-observer-rbac/rbac.yaml").read_text() ) if item ] binding = next(item for item in rbac if item["kind"] == "ClusterRoleBinding") assert binding["subjects"] == [ {"kind": "ServiceAccount", "name": "hermes-agent", "namespace": "hermes"} ] def test_switchyard_active_state_uses_a_relocatable_rwx_claim(): """A stale node attachment must not strand the routing authority.""" claims = [ item for item in yaml.safe_load_all((HERMES / "switchyard-pvc.yaml").read_text()) if item ] active_claim = next( item for item in claims if item["metadata"]["name"] == "hermes-switchyard-state-rwx" ) assert active_claim["spec"]["accessModes"] == ["ReadWriteMany"] deployment = yaml.safe_load((HERMES / "switchyard-deployment.yaml").read_text()) strategy = deployment["spec"]["strategy"] assert strategy == { "type": "RollingUpdate", "rollingUpdate": {"maxSurge": 1, "maxUnavailable": 0}, } pod = deployment["spec"]["template"]["spec"] state = next(item for item in pod["volumes"] if item["name"] == "state") assert ( state["persistentVolumeClaim"]["claimName"] == active_claim["metadata"]["name"] )