"""Contracts for the dedicated, unprivileged Hermes node SSH identity.""" from __future__ import annotations import base64 import importlib.util import os import sys from pathlib import Path import pytest import yaml ROOT = Path(__file__).parents[2] SCRIPT = ROOT / "services/hermes/scripts/node_account_hardening.py" def _load(): spec = importlib.util.spec_from_file_location("node_account_hardening_test", SCRIPT) 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 _fixture(tmp_path: Path, monkeypatch): module = _load() host_etc = tmp_path / "etc" host_home = tmp_path / "home" host_etc.mkdir() host_home.mkdir() originals = { "passwd": ( "root:x:0:0:root:/root:/bin/bash\n" "atlas:x:2000:2000:Atlas:/home/atlas:/bin/bash\n" "oceanus:x:2001:2001:Oceanus:/home/oceanus:/bin/bash\n" ), "group": ( "root:x:0:\n" "atlas:x:2000:\n" "oceanus:x:2001:\n" "disk:x:6:atlas\n" "sudo:x:27:oceanus\n" ), "shadow": ( "root:!:1:0:99999:7:::\n" "atlas:!:1:0:99999:7:::\n" "oceanus:!:1:0:99999:7:::\n" ), "gshadow": ( "root:!::\n" "atlas:!::\n" "oceanus:!::\n" "disk:!::atlas\n" "sudo:!::oceanus\n" ), } for name, value in originals.items(): (host_etc / name).write_text(value, encoding="utf-8") key = "ssh-ed25519 " + base64.b64encode(b"synthetic-hermes-key").decode() other = "ssh-ed25519 " + base64.b64encode(b"human-operator-key").decode() for user in ("atlas", "oceanus"): ssh = host_home / user / ".ssh" ssh.mkdir(parents=True) (ssh / "authorized_keys").write_text( f"{other} {user}\n{key}\n", encoding="utf-8" ) public_key = tmp_path / "public-key" public_key.write_text(key + "\n", encoding="utf-8") monkeypatch.setattr(module, "HOST_ETC", host_etc) monkeypatch.setattr(module, "HOST_HOME", host_home) monkeypatch.setattr(module, "ACCOUNT_UID", os.getuid()) monkeypatch.setattr(module, "ACCOUNT_GID", os.getgid()) # ACL behavior has its own executable test below. These account-database # tests must not depend on the host running pytest with CAP_FOWNER. monkeypatch.setattr(module, "_deny_sensitive_roots", lambda: None) return module, originals, key, other, public_key def test_reconciler_creates_locked_groupless_account_and_moves_only_exact_key( tmp_path: Path, monkeypatch ): module, originals, key, other, public_key = _fixture(tmp_path, monkeypatch) module.reconcile(public_key) first = { name: (module.HOST_ETC / name).read_text(encoding="utf-8") for name in originals } module.reconcile(public_key) second = { name: (module.HOST_ETC / name).read_text(encoding="utf-8") for name in originals } assert first == second expected = module._expected_records() for name, original in originals.items(): assert first[name] == original + ":".join(expected[name]) + "\n" backup = module.HOST_ETC / f"{name}.hermes-boundary-backup" assert backup.read_text(encoding="utf-8") == original assert expected["shadow"][1] == "!" assert expected["group"][-1] == "" assert "disk:x:6:atlas\n" in first["group"] assert "sudo:x:27:oceanus\n" in first["group"] hermes_keys = ( module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys" ).read_text(encoding="utf-8") assert hermes_keys == key + "\n" for user in module.LEGACY_ACCOUNTS: legacy = ( module.HOST_HOME / user / ".ssh/authorized_keys" ).read_text(encoding="utf-8") assert legacy == f"{other} {user}\n" backup = module.HOST_HOME / user / ".ssh/authorized_keys.hermes-boundary-backup" assert backup.read_text(encoding="utf-8") == f"{other} {user}\n{key}\n" def test_identity_conflict_fails_closed_without_editing_account_databases( tmp_path: Path, monkeypatch ): module, originals, _key, _other, public_key = _fixture(tmp_path, monkeypatch) conflict = originals["passwd"] + ( f"unrelated:x:{module.ACCOUNT_UID}:{module.ACCOUNT_GID}:Other:/home/other:/bin/bash\n" ) (module.HOST_ETC / "passwd").write_text(conflict, encoding="utf-8") with pytest.raises(module.HardeningError, match="identity conflicts"): module.reconcile(public_key) assert (module.HOST_ETC / "passwd").read_text(encoding="utf-8") == conflict for name in ("group", "shadow", "gshadow"): assert (module.HOST_ETC / name).read_text(encoding="utf-8") == originals[name] assert not (module.HOST_HOME / module.ACCOUNT).exists() def test_malformed_account_file_fails_closed(tmp_path: Path, monkeypatch): module, originals, _key, _other, public_key = _fixture(tmp_path, monkeypatch) (module.HOST_ETC / "group").write_text("malformed\n", encoding="utf-8") with pytest.raises(module.HardeningError, match="invalid record"): module.reconcile(public_key) assert (module.HOST_ETC / "passwd").read_text(encoding="utf-8") == originals[ "passwd" ] assert (module.HOST_ETC / "group").read_text(encoding="utf-8") == "malformed\n" def test_preexisting_home_fails_before_account_database_or_key_changes( tmp_path: Path, monkeypatch ): module, originals, _key, _other, public_key = _fixture(tmp_path, monkeypatch) home = module.HOST_HOME / module.ACCOUNT home.mkdir() (home / "unrelated").write_text("preserve\n", encoding="utf-8") with pytest.raises(module.HardeningError, match="home already exists"): module.reconcile(public_key) for name, original in originals.items(): assert (module.HOST_ETC / name).read_text(encoding="utf-8") == original assert not (module.HOST_ETC / f"{name}.hermes-boundary-backup").exists() assert (home / "unrelated").read_text(encoding="utf-8") == "preserve\n" def test_flux_daemonset_reconciles_every_node_without_mutating_human_groups(): documents = list( yaml.safe_load_all( (ROOT / "services/hermes/node-ssh-access.yaml").read_text(encoding="utf-8") ) ) daemonset = next(item for item in documents if item["kind"] == "DaemonSet") spec = daemonset["spec"]["template"]["spec"] assert spec["tolerations"] == [{"operator": "Exists"}] command = spec["containers"][0]["args"][0] assert "/opt/node-hardener/node_account_hardening.py" in command assert "sleep 300" in command mounts = {item["name"]: item for item in spec["volumes"]} assert mounts["host-home"]["hostPath"]["path"] == "/home" assert mounts["host-etc"]["hostPath"]["path"] == "/etc" assert mounts["host-k3s"]["hostPath"]["path"] == "/var/lib/rancher/k3s" assert mounts["host-kubelet"]["hostPath"]["path"] == "/var/lib/kubelet" assert mounts["host-run-k3s"]["hostPath"]["path"] == "/run/k3s" assert mounts["host-run-containerd"]["hostPath"]["path"] == "/run/containerd" assert "usermod" not in command assert "groupmod" not in command policies = list( yaml.safe_load_all( (ROOT / "services/hermes/networkpolicy.yaml").read_text(encoding="utf-8") ) ) isolation = next( item for item in policies if item["metadata"]["name"] == "hermes-node-ssh-access-isolation" ) assert isolation["spec"]["ingress"] == [] assert isolation["spec"]["egress"] == [] source = SCRIPT.read_text(encoding="utf-8") assert "ACCOUNT = \"hermes-agent\"" in source assert "ACCOUNT_UID = 1200" in source assert "ACCOUNT_GID = 1200" in source assert 'LEGACY_ACCOUNTS = ("atlas", "oceanus")' in source assert "supplementary" not in source.lower() def test_sensitive_roots_get_explicit_zero_permission_acl_for_hermes( tmp_path: Path, monkeypatch ): module = _load() sensitive = tmp_path / "k3s" sensitive.mkdir(mode=0o755) host_etc = tmp_path / "etc" host_etc.mkdir() monkeypatch.setattr(module, "ACCOUNT_UID", 1200) monkeypatch.setattr(module, "HOST_ETC", host_etc) monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid()) # Production requires uid 0. The test process owns its fixture but runs the # real xattr/ACL implementation using the test owner's uid as that boundary. module._deny_sensitive_root(sensitive) first = module._read_acl(sensitive) module._deny_sensitive_root(sensitive) second = module._read_acl(sensitive) assert first == second entries = module._decode_acl(first, 0o755) assert (module.ACL_USER, 0, 1200) in entries assert (module.ACL_USER_OBJ, 0o7, module.ACL_UNDEFINED_ID) in entries assert (module.ACL_GROUP_OBJ, 0o5, module.ACL_UNDEFINED_ID) in entries assert (module.ACL_OTHER, 0o5, module.ACL_UNDEFINED_ID) in entries backup = host_etc / "hermes-node-boundary/k3s.acl" assert backup.read_bytes() == b"N" assert backup.stat().st_mode & 0o777 == 0o600 def test_sensitive_root_rejects_non_directory_and_non_root_owner( tmp_path: Path, monkeypatch ): module = _load() regular = tmp_path / "not-a-directory" regular.write_text("preserve", encoding="utf-8") with pytest.raises(module.HardeningError, match="unsafe sensitive directory"): module._deny_sensitive_root(regular) directory = tmp_path / "not-root-owned" directory.mkdir() if directory.stat().st_uid == 0: pytest.skip("cannot construct a non-root-owned fixture as root") with pytest.raises(module.HardeningError, match="not root-owned"): module._deny_sensitive_root(directory) def test_acl_parser_rejects_malformed_or_incomplete_values(): module = _load() with pytest.raises(module.HardeningError, match="malformed"): module._decode_acl(b"too short", 0o755) incomplete = module._encode_acl( [(module.ACL_USER_OBJ, 0o7, module.ACL_UNDEFINED_ID)] ) with pytest.raises(module.HardeningError, match="incomplete"): module._decode_acl(incomplete, 0o755) def test_acl_failure_stops_before_installing_the_ssh_key(tmp_path: Path, monkeypatch): module = _load() public_key = tmp_path / "public-key" public_key.write_text( "ssh-ed25519 " + base64.b64encode(b"synthetic-key").decode() + "\n", encoding="utf-8", ) moved = False monkeypatch.setattr(module, "_reconcile_databases", lambda: None) def fail_acl(): raise module.HardeningError("ACL unavailable") def move_key(_path): nonlocal moved moved = True monkeypatch.setattr(module, "_deny_sensitive_roots", fail_acl) monkeypatch.setattr(module, "_move_key", move_key) with pytest.raises(module.HardeningError, match="ACL unavailable"): module.reconcile(public_key) assert moved is False def test_runtime_ssh_config_forces_dedicated_account_for_every_titan(): stage = ( ROOT / "services/hermes/scripts/stage_runtime_access.py" ).read_text(encoding="utf-8") assert '"Host titan-*\\n User hermes-agent\\n"' in stage agent = yaml.safe_load( (ROOT / "services/hermes/agent-deployment.yaml").read_text(encoding="utf-8") ) assert "User atlas" not in yaml.safe_dump(agent) assert "User oceanus" not in yaml.safe_dump(agent)