atlas-iac/testing/tests/test_hermes_node_account_hardening.py

"""Core contracts for the dedicated Hermes node SSH identity."""

from __future__ import annotations

import os
from pathlib import Path

import pytest
import yaml

from testing.tests.test_hermes_node_account_support import (
    ROOT,
    SCRIPT,
    _fixture,
    _key,
    _load,
)

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 "audit_membership" in source
    image = spec["containers"][0]["image"]
    assert image == (
        "python@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df"
    )
    assert mounts["host-polkit-share"]["hostPath"]["path"] == "/usr/share/polkit-1"


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()
    host_etc = tmp_path / "etc"
    host_etc.mkdir()
    monkeypatch.setattr(module, "HOST_ETC", host_etc)
    monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid())
    public_key = tmp_path / "public-key"
    public_key.write_text(
        _key(b"synthetic-key") + "\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)