207 lines
7.7 KiB
Python
207 lines
7.7 KiB
Python
|
"""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")
|