"""Behavioral branch coverage for node-storage ACL hardening.""" from __future__ import annotations import errno import os import pytest from testing.tests.test_hermes_node_account_support import _fixture, _load def test_acl_decoder_rejects_version_permissions_and_adds_missing_mask(): module = _load() valid = [ (module.ACL_USER_OBJ, 7, module.ACL_UNDEFINED_ID), (module.ACL_GROUP_OBJ, 5, module.ACL_UNDEFINED_ID), (module.ACL_OTHER, 5, module.ACL_UNDEFINED_ID), ] wrong_version = module.ACL_HEADER.pack(99) + b"".join( module.ACL_ENTRY.pack(*entry) for entry in valid ) with pytest.raises(module.HardeningError, match="version"): module._decode_acl(wrong_version, 0o755) bad_permission = module._encode_acl([*valid, (module.ACL_USER, 8, 1200)]) with pytest.raises(module.HardeningError, match="permission"): module._decode_acl(bad_permission, 0o755) updated = module._decode_acl( module._acl_with_deny(module._encode_acl(valid), 0o755), 0o755 ) assert (module.ACL_MASK, 5, module.ACL_UNDEFINED_ID) in updated assert (module.ACL_USER, 0, module.ACCOUNT_UID) in updated def test_read_acl_handles_absence_and_propagates_other_errors(monkeypatch, tmp_path): module = _load() path = tmp_path / "root" path.mkdir() def missing(*_args, **_kwargs): raise OSError(errno.ENODATA, "missing") monkeypatch.setattr(module.os, "getxattr", missing) assert module._read_acl(path) == b"" def denied(*_args, **_kwargs): raise OSError(errno.EPERM, "denied") monkeypatch.setattr(module.os, "getxattr", denied) with pytest.raises(OSError): module._read_acl(path) @pytest.mark.parametrize("value", [b"", b"X", b"A", b"Nextra"]) def test_acl_backup_validator_rejects_malformed_encodings(value): module = _load() with pytest.raises(module.HardeningError, match="backup is malformed"): module._validate_acl_backup(value) module._validate_acl_backup(b"N") valid = module._encode_acl( [ (module.ACL_USER_OBJ, 7, module.ACL_UNDEFINED_ID), (module.ACL_GROUP_OBJ, 5, module.ACL_UNDEFINED_ID), (module.ACL_OTHER, 5, module.ACL_UNDEFINED_ID), ] ) module._validate_acl_backup(b"A" + valid) def test_acl_backup_reuses_valid_existing_copy(tmp_path, monkeypatch): module = _load() host_etc = tmp_path / "etc" root = host_etc / "hermes-node-boundary" root.mkdir(parents=True) backup = root / "k3s.acl" backup.write_bytes(b"N") monkeypatch.setattr(module, "HOST_ETC", host_etc) monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid()) module._acl_backup_once(tmp_path / "k3s", b"") assert backup.read_bytes() == b"N" def test_acl_backup_short_write_cleans_temporary(tmp_path, monkeypatch): module = _load() host_etc = tmp_path / "etc" host_etc.mkdir() monkeypatch.setattr(module, "HOST_ETC", host_etc) monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid()) monkeypatch.setattr(module.os, "write", lambda _fd, _value: 0) with pytest.raises(module.HardeningError, match="short sensitive"): module._acl_backup_once(tmp_path / "k3s", b"") assert not list((host_etc / "hermes-node-boundary").glob(".*.hermes-*")) def test_acl_backup_tolerates_concurrent_first_writer(tmp_path, monkeypatch): module = _load() host_etc = tmp_path / "etc" host_etc.mkdir() monkeypatch.setattr(module, "HOST_ETC", host_etc) monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid()) real_link = os.link def race(source, target, **kwargs): real_link(source, target, **kwargs) raise FileExistsError() monkeypatch.setattr(module.os, "link", race) module._acl_backup_once(tmp_path / "k3s", b"") assert (host_etc / "hermes-node-boundary/k3s.acl").read_bytes() == b"N" def test_sensitive_root_failure_restores_existing_acl(tmp_path, monkeypatch): module = _load() root = tmp_path / "k3s" root.mkdir() monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) current = module._encode_acl( [ (module.ACL_USER_OBJ, 7, module.ACL_UNDEFINED_ID), (module.ACL_GROUP_OBJ, 5, module.ACL_UNDEFINED_ID), (module.ACL_MASK, 5, module.ACL_UNDEFINED_ID), (module.ACL_OTHER, 5, module.ACL_UNDEFINED_ID), ] ) reads = iter((current, current)) writes = [] monkeypatch.setattr(module, "_read_acl", lambda _path: next(reads)) monkeypatch.setattr(module, "_acl_backup_once", lambda *_a, **_k: None) monkeypatch.setattr( module.os, "setxattr", lambda *args, **kwargs: writes.append(args[2]) ) with pytest.raises(module.HardeningError, match="validation failed"): module._deny_sensitive_root(root) assert writes[-1] == current @pytest.mark.parametrize("remove_errno", [errno.ENODATA, errno.EPERM]) def test_sensitive_root_failure_removes_new_acl_or_propagates_cleanup_error( tmp_path, monkeypatch, remove_errno ): module = _load() root = tmp_path / "k3s" root.mkdir() monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) monkeypatch.setattr(module, "_read_acl", lambda _path: b"") monkeypatch.setattr(module, "_acl_backup_once", lambda *_a, **_k: None) monkeypatch.setattr( module.os, "setxattr", lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("set failed")), ) monkeypatch.setattr( module.os, "removexattr", lambda *_a, **_k: (_ for _ in ()).throw(OSError(remove_errno, "remove")), ) expected = OSError if remove_errno == errno.EPERM else RuntimeError with pytest.raises(expected): module._deny_sensitive_root(root) def test_database_validation_failure_rolls_back_all_writes(tmp_path, monkeypatch): module, originals, _key_value, _other, _public = _fixture(tmp_path, monkeypatch) real_read = module._read_regular calls = 0 def missing_expected(path, maximum=module.MAX_ACCOUNT_FILE): nonlocal calls calls += 1 if calls == 1: return originals["passwd"].encode(), path.stat(follow_symlinks=False) return real_read(path, maximum) monkeypatch.setattr(module, "_read_regular", missing_expected) with pytest.raises(module.HardeningError, match="validation failed"): module._reconcile_databases() for name, value in originals.items(): assert (module.HOST_ETC / name).read_text() == value def test_preexisting_target_key_removal_is_verified(tmp_path, monkeypatch): module, _originals, key, _other, public = _fixture(tmp_path, monkeypatch) target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys" target.parent.mkdir(parents=True) target.write_text(key + "\n", encoding="utf-8") real_write = module.atomic_write def ignore_preinstall_target(path, value, metadata): if path == target and value == b"": return real_write(path, value, metadata) monkeypatch.setattr(module, "atomic_write", ignore_preinstall_target) with pytest.raises(module.HardeningError, match="preexisting Hermes authorization"): module._move_key(public) def test_existing_target_without_key_does_not_need_precleanup(tmp_path, monkeypatch): module, _originals, _key_value, other, public = _fixture(tmp_path, monkeypatch) target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys" target.parent.mkdir(parents=True) target.write_text(other + "\n", encoding="utf-8") module._move_key(public) assert target.read_text().endswith("\n")