"""Owner and Switchyard access-boundary contracts.""" from __future__ import annotations from testing.tests.test_hermes_cli_support import ( HERMES, SCRIPTS, _agent_deployment, yaml, ) 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 "grep -qxF" not 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 = next(item for item in pod["volumes"] if item["name"] == "coordinator") assert hardener["configMap"] == { "name": "hermes-node-account-hardener", "defaultMode": 0o555, } 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((HERMES / "rbac.yaml").read_text()) if item ] bindings = [item for item in rbac if item["kind"] == "ClusterRoleBinding"] assert bindings assert all( subject["name"] not in {"hermes-agent", "hermes-switchyard"} for binding in bindings for subject in binding["subjects"] ) 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" ] def test_worker_route_broker_accepts_pod_network_health_checks(): """Kubelet probes the pod IP, so the broker cannot bind to loopback only.""" script = (SCRIPTS / "worker_route_broker.py").read_text() assert 'ThreadingHTTPServer(("0.0.0.0", PORT), Handler)' in script def test_switchyard_network_boundary_allows_vault_bootstrap(): """The pre-populate init container must reach Vault before routing starts.""" 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-switchyard-isolation" ) assert any( rule.get("to") == [ { "namespaceSelector": { "matchLabels": {"kubernetes.io/metadata.name": "vault"} }, "podSelector": {"matchLabels": {"app": "vault"}}, } ] and rule.get("ports") == [{"protocol": "TCP", "port": 8200}] for rule in isolation["spec"]["egress"] ) def test_switchyard_network_boundary_allows_metrics_scraping(): """VictoriaMetrics may scrape Switchyard without widening its API boundary.""" 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-switchyard-isolation" ) assert any( rule.get("from") == [ { "namespaceSelector": { "matchLabels": { "kubernetes.io/metadata.name": "monitoring" } }, "podSelector": {"matchLabels": {"app": "server"}}, } ] and rule.get("ports") == [{"protocol": "TCP", "port": 9005}] for rule in isolation["spec"]["ingress"] )