254 lines
8.8 KiB
Python
254 lines
8.8 KiB
Python
"""Crash-safety and metadata contracts for Hermes node-account writes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from testing.tests.test_hermes_node_account_support import (
|
|
ROOT,
|
|
_fixture,
|
|
_load,
|
|
_typed_key,
|
|
)
|
|
|
|
def test_key_identity_ignores_options_and_comments_and_removes_legacy_first(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
module, _originals, key, other, public_key = _fixture(tmp_path, monkeypatch)
|
|
key_fields = key.split()
|
|
decorated = f'restrict,command="echo denied" {key_fields[0]} {key_fields[1]} old-comment'
|
|
for user in module.LEGACY_ACCOUNTS:
|
|
path = module.HOST_HOME / user / ".ssh/authorized_keys"
|
|
path.write_text(f"{other} {user}\n{decorated}\n", encoding="utf-8")
|
|
writes: list[Path] = []
|
|
real_write = module.atomic_write
|
|
|
|
def recording_write(path, value, metadata):
|
|
writes.append(path)
|
|
real_write(path, value, metadata)
|
|
|
|
monkeypatch.setattr(module, "atomic_write", recording_write)
|
|
module.reconcile(public_key)
|
|
|
|
key_writes = [path for path in writes if path.name == "authorized_keys"]
|
|
assert key_writes[-1].parts[-3:] == (module.ACCOUNT, ".ssh", "authorized_keys")
|
|
assert all(module.ACCOUNT not in path.parts for path in key_writes[:-1])
|
|
|
|
|
|
def test_unrelated_modern_human_key_type_is_preserved(tmp_path: Path, monkeypatch):
|
|
module, _originals, key, _other, public_key = _fixture(tmp_path, monkeypatch)
|
|
modern = _typed_key("sk-ssh-ed25519@openssh.com", b"human-security-key", "human")
|
|
for user in module.LEGACY_ACCOUNTS:
|
|
path = module.HOST_HOME / user / ".ssh/authorized_keys"
|
|
path.write_text(f"{modern}\n{key}\n", encoding="utf-8")
|
|
|
|
module.reconcile(public_key)
|
|
|
|
for user in module.LEGACY_ACCOUNTS:
|
|
path = module.HOST_HOME / user / ".ssh/authorized_keys"
|
|
assert path.read_text() == modern + "\n"
|
|
|
|
|
|
def test_malformed_unrelated_authorized_key_lines_are_preserved(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
module, _originals, key, _other, public_key = _fixture(tmp_path, monkeypatch)
|
|
malformed = b"ssh-ed25519 YWJj human-key-that-must-survive\noperator-note-\xff\n"
|
|
for user in module.LEGACY_ACCOUNTS:
|
|
path = module.HOST_HOME / user / ".ssh/authorized_keys"
|
|
path.write_bytes(malformed + key.encode("ascii") + b"\n")
|
|
|
|
module.reconcile(public_key)
|
|
|
|
for user in module.LEGACY_ACCOUNTS:
|
|
path = module.HOST_HOME / user / ".ssh/authorized_keys"
|
|
assert path.read_bytes() == malformed
|
|
|
|
|
|
def test_crash_during_legacy_key_removal_never_installs_duplicate_key(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
module, _originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
|
|
real_write = module.atomic_write
|
|
key_writes = 0
|
|
|
|
def crash_on_second_key(path, value, metadata):
|
|
nonlocal key_writes
|
|
if path.name == "authorized_keys":
|
|
key_writes += 1
|
|
if key_writes == 2:
|
|
raise module.HardeningError("synthetic crash")
|
|
real_write(path, value, metadata)
|
|
|
|
monkeypatch.setattr(module, "atomic_write", crash_on_second_key)
|
|
with pytest.raises(module.HardeningError, match="synthetic crash"):
|
|
module.reconcile(public_key)
|
|
assert not (module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys").exists()
|
|
|
|
|
|
def test_preexisting_target_key_is_removed_before_legacy_cleanup_and_reinstalled(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
module, _originals, key, _other, public_key = _fixture(tmp_path, monkeypatch)
|
|
target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text(key + " existing\n")
|
|
writes: list[Path] = []
|
|
real_write = module.atomic_write
|
|
|
|
def recording_write(path, value, metadata):
|
|
if path.name == "authorized_keys":
|
|
writes.append(path)
|
|
real_write(path, value, metadata)
|
|
|
|
monkeypatch.setattr(module, "atomic_write", recording_write)
|
|
module._move_key(public_key)
|
|
|
|
assert writes[0] == target
|
|
assert writes[-1] == target
|
|
assert all(path != target for path in writes[1:-1])
|
|
assert target.read_text() == key + "\n"
|
|
|
|
|
|
def test_reconcile_reads_csi_projected_public_key(tmp_path: Path, monkeypatch):
|
|
module, _originals, key, _other, public_key = _fixture(tmp_path, monkeypatch)
|
|
data = public_key.parent / "..2026_08_18.1"
|
|
data.mkdir()
|
|
(data / public_key.name).write_text(key + "\n", encoding="utf-8")
|
|
(public_key.parent / "..data").symlink_to(data.name)
|
|
public_key.unlink()
|
|
public_key.symlink_to(Path("..data") / public_key.name)
|
|
|
|
module.reconcile(public_key)
|
|
|
|
target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys"
|
|
assert target.read_text() == key + "\n"
|
|
|
|
|
|
def test_account_database_write_preserves_mode_owner_and_xattrs(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
|
|
passwd = module.HOST_ETC / "passwd"
|
|
passwd.chmod(0o640)
|
|
os.setxattr(passwd, "user.hermes-test", b"preserve")
|
|
before = passwd.stat()
|
|
|
|
module.reconcile(public_key)
|
|
|
|
after = passwd.stat()
|
|
assert after.st_mode & 0o777 == 0o640
|
|
assert (after.st_uid, after.st_gid) == (before.st_uid, before.st_gid)
|
|
assert os.getxattr(passwd, "user.hermes-test") == b"preserve"
|
|
backup = module.HOST_ETC / "passwd.hermes-boundary-backup"
|
|
assert os.getxattr(backup, "user.hermes-test") == b"preserve"
|
|
assert backup.read_text() == originals["passwd"]
|
|
|
|
|
|
def test_atomic_write_restores_acl_and_security_label_xattrs(tmp_path: Path, monkeypatch):
|
|
module = _load()
|
|
io_module = sys.modules[module.atomic_write.__module__]
|
|
target = tmp_path / "passwd"
|
|
target.write_bytes(b"old\n")
|
|
metadata = io_module.FileSnapshot(
|
|
value=b"old\n",
|
|
device=target.stat().st_dev,
|
|
inode=target.stat().st_ino,
|
|
mode=0o600,
|
|
uid=os.getuid(),
|
|
gid=os.getgid(),
|
|
size=4,
|
|
mtime_ns=target.stat().st_mtime_ns,
|
|
ctime_ns=target.stat().st_ctime_ns,
|
|
xattrs=(
|
|
("system.posix_acl_access", b"synthetic-acl"),
|
|
("security.selinux", b"synthetic-label"),
|
|
("user.audit", b"synthetic-xattr"),
|
|
),
|
|
)
|
|
restored: list[tuple[str, bytes]] = []
|
|
monkeypatch.setattr(
|
|
io_module.os,
|
|
"setxattr",
|
|
lambda _fd, name, value: restored.append((name, value)),
|
|
)
|
|
|
|
io_module.atomic_write(target, b"new\n", metadata)
|
|
|
|
assert target.read_bytes() == b"new\n"
|
|
assert restored == list(metadata.xattrs)
|
|
|
|
|
|
def test_concurrent_database_change_is_detected_before_any_account_write(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
|
|
real_backup = module.backup_once
|
|
backup_count = 0
|
|
writes: list[Path] = []
|
|
|
|
def racing_backup(path, snapshot, maximum):
|
|
nonlocal backup_count
|
|
result = real_backup(path, snapshot, maximum)
|
|
backup_count += 1
|
|
if backup_count == 4:
|
|
group = module.HOST_ETC / "group"
|
|
group.write_text(originals["group"] + "race:x:4000:\n")
|
|
return result
|
|
|
|
monkeypatch.setattr(module, "backup_once", racing_backup)
|
|
monkeypatch.setattr(module, "atomic_write", lambda path, *_args: writes.append(path))
|
|
with pytest.raises(module.HardeningError, match="concurrent host account change"):
|
|
module.reconcile(public_key)
|
|
assert writes == []
|
|
|
|
|
|
def test_flux_orders_observer_rbac_before_hermes_prunes_old_authority():
|
|
observer = yaml.safe_load(
|
|
(ROOT / "clusters/atlas/flux-system/applications/hermes-observer-rbac/kustomization.yaml").read_text()
|
|
)
|
|
hermes = yaml.safe_load(
|
|
(ROOT / "clusters/atlas/flux-system/applications/hermes/kustomization.yaml").read_text()
|
|
)
|
|
bindings = yaml.safe_load(
|
|
(
|
|
ROOT
|
|
/ "clusters/atlas/flux-system/applications/hermes-observer-bindings/kustomization.yaml"
|
|
).read_text()
|
|
)
|
|
assert "dependsOn" not in observer["spec"]
|
|
assert {item["name"] for item in hermes["spec"]["dependsOn"]} >= {
|
|
"hermes-observer-rbac",
|
|
"hermes-scm-broker",
|
|
}
|
|
assert {item["name"] for item in bindings["spec"]["dependsOn"]} == {
|
|
"hermes-observer-rbac",
|
|
"hermes",
|
|
}
|
|
assert "hermes-observer-bindings" not in {
|
|
item["name"] for item in hermes["spec"]["dependsOn"]
|
|
}
|
|
|
|
|
|
def test_sensitive_root_acl_backups_have_unique_host_path_names(monkeypatch):
|
|
module = _load()
|
|
seen = []
|
|
monkeypatch.setattr(
|
|
module,
|
|
"_deny_sensitive_root",
|
|
lambda path, backup_name=None: seen.append((path, backup_name)),
|
|
)
|
|
module._deny_sensitive_roots()
|
|
assert [name for _path, name in seen] == [
|
|
"var-lib-rancher-k3s",
|
|
"var-lib-kubelet",
|
|
"run-k3s",
|
|
"run-containerd",
|
|
]
|
|
assert len({name for _path, name in seen}) == 4
|