atlas-iac/testing/tests/test_hermes_node_account_io.py

"""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_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_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