atlas-iac/testing/tests/test_hermes_node_audit_coverage.py

"""Behavioral branch coverage for dedicated node-account privilege audits."""

from __future__ import annotations

import os
import struct
import sys
from types import SimpleNamespace

import pytest

from testing.tests.test_hermes_node_account_support import _load


def _audit_module():
    hardening = _load()
    return sys.modules[hardening.audit_membership.__module__]


def test_member_parser_handles_empty_valid_malformed_and_duplicates():
    module = _audit_module()
    assert module._members("", "group") == set()
    assert module._members("one,two", "group") == {"one", "two"}
    for value in ("one,", " one", "one,one"):
        with pytest.raises(module.HardeningError):
            module._members(value, "group")


def test_membership_audit_accepts_unrelated_records_and_rejects_account():
    module = _audit_module()
    module.audit_membership(
        "hermes-agent", [["disk", "x", "6", "atlas"]], [["disk", "!", "atlas", ""]]
    )
    with pytest.raises(module.HardeningError, match="supplementary"):
        module.audit_membership(
            "hermes-agent", [["disk", "x", "6", "hermes-agent"]], []
        )
    with pytest.raises(module.HardeningError, match="gshadow"):
        module.audit_membership("hermes-agent", [], [["disk", "!", "hermes-agent", ""]])


@pytest.mark.parametrize("kind", ["mode", "symlink"])
def test_policy_files_reject_unsafe_directories(tmp_path, kind):
    module = _audit_module()
    host_etc = tmp_path / "etc"
    host_etc.mkdir()
    root = host_etc / "sudoers.d"
    if kind == "mode":
        root.mkdir(mode=0o777)
        root.chmod(0o777)
    else:
        target = tmp_path / "target"
        target.mkdir()
        root.symlink_to(target)
    with pytest.raises(module.HardeningError, match="unsafe sudo/polkit"):
        module._policy_files(host_etc, tmp_path / "share", os.getuid())


def test_policy_file_count_is_bounded(tmp_path):
    module = _audit_module()
    root = tmp_path / "etc/sudoers.d"
    root.mkdir(parents=True)
    for index in range(257):
        (root / str(index)).write_text("# safe\n", encoding="utf-8")
    with pytest.raises(module.HardeningError, match="too many"):
        module._policy_files(tmp_path / "etc", tmp_path / "share", os.getuid())


def _acl(*entries):
    return struct.pack("<I", 2) + b"".join(
        struct.pack("<HHI", *entry) for entry in entries
    )


def test_policy_metadata_accepts_safe_acl_and_rejects_malformed_or_writable():
    module = _audit_module()
    safe = SimpleNamespace(
        uid=0,
        mode=0o600,
        xattrs=(("system.posix_acl_access", _acl((1, 6, 0xFFFFFFFF))),),
    )
    module._audit_policy_metadata(safe, 0, 1200)
    malformed = SimpleNamespace(
        uid=0, mode=0o600, xattrs=(("system.posix_acl_access", b"bad"),)
    )
    with pytest.raises(module.HardeningError, match="ACL is malformed"):
        module._audit_policy_metadata(malformed, 0, 1200)
    writable = SimpleNamespace(
        uid=0, mode=0o600, xattrs=(("system.posix_acl_access", _acl((2, 2, 1200))),)
    )
    with pytest.raises(module.HardeningError, match="grants Hermes write"):
        module._audit_policy_metadata(writable, 0, 1200)


def test_active_sudo_policy_handles_blank_comments_numeric_and_includes():
    module = _audit_module()
    text = (
        "\n# ordinary comment\n#1200 ALL=(ALL) ALL\n"
        "#includedir /etc/sudoers.d\nroot ALL=(ALL) ALL\n"
    )
    active = module._active_sudo_policy(text)
    assert "#1200" in active and "root ALL" in active
    assert "ordinary" not in active and "includedir" not in active
    with pytest.raises(module.HardeningError, match="unaudited"):
        module._active_sudo_policy("@include /external/policy\n")


@pytest.mark.parametrize(
    "directive",
    [
        '#include "/etc/evil path"',
        "#include '/etc/evil path'",
        "#include /etc/evil\\ path",
        "#include\t/etc/evil",
        "#include /etc/evil",
        "#includedir /opt/other",
        '#includedir "/etc/sudoers.d"',
        "@include /external/policy",
        '@include "/etc/evil path"',
        "@includedir /opt/other",
        "#include",
        "@includedir",
    ],
)
def test_every_include_spelling_outside_the_standard_dir_fails_closed(directive: str):
    """No include form may reach the comment branch and stay unaudited."""
    module = _audit_module()
    with pytest.raises(module.HardeningError, match="unaudited"):
        module._active_sudo_policy(directive + "\n")


@pytest.mark.parametrize(
    "directive",
    ["#includedir /etc/sudoers.d", "@includedir /etc/sudoers.d"],
)
def test_standard_includedir_is_the_only_allowed_include(directive: str):
    module = _audit_module()
    active = module._active_sudo_policy(directive + "\nroot ALL=(ALL) ALL\n")
    assert active == "root ALL=(ALL) ALL"


@pytest.mark.parametrize(
    "line",
    ["#included by an operator note", "#include-guard notes", "#includes history"],
)
def test_comment_lines_that_only_resemble_includes_stay_comments(line: str):
    module = _audit_module()
    active = module._active_sudo_policy(line + "\nroot ALL=(ALL) ALL\n")
    assert active == "root ALL=(ALL) ALL"


def test_identity_matcher_covers_account_numeric_and_unrelated_values():
    module = _audit_module()
    assert module._mentions_dedicated_identity("hermes-agent ALL", "hermes-agent", 1200)
    assert module._mentions_dedicated_identity("#1200 ALL", "hermes-agent", 1200)
    assert not module._mentions_dedicated_identity(
        "hermes-agent-extra 12001", "hermes-agent", 1200
    )


def _snapshot(value: bytes, *, mode=0o600, uid=0, xattrs=()):
    return SimpleNamespace(value=value, mode=mode, uid=uid, xattrs=tuple(xattrs))


def test_privilege_audit_accepts_files_systemd_nss_and_comments(tmp_path, monkeypatch):
    module = _audit_module()
    host_etc = tmp_path / "etc"
    host_etc.mkdir()
    nss = host_etc / "nsswitch.conf"
    nss.write_text(
        "passwd: files systemd\nhosts: files dns\n# comment\n", encoding="utf-8"
    )
    sudoers = host_etc / "sudoers"
    sudoers.write_text("# safe comment\nroot ALL=(ALL) ALL\n", encoding="utf-8")
    monkeypatch.setattr(
        module,
        "read_regular",
        lambda path, maximum: _snapshot(path.read_bytes(), uid=os.getuid()),
    )
    module.audit_privilege_policies(
        "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
    )


@pytest.mark.parametrize(
    ("value", "match"),
    [
        (b"passwd: \n", "external group/account"),
        (b"passwd: files\xff\n", "not UTF-8"),
    ],
)
def test_nsswitch_audit_rejects_empty_or_non_utf8_sources(
    tmp_path, monkeypatch, value, match
):
    module = _audit_module()
    host_etc = tmp_path / "etc"
    host_etc.mkdir()
    path = host_etc / "nsswitch.conf"
    path.write_bytes(value)
    monkeypatch.setattr(
        module,
        "read_regular",
        lambda _path, _maximum: _snapshot(value, uid=os.getuid()),
    )
    with pytest.raises(module.HardeningError, match=match):
        module.audit_privilege_policies(
            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
        )


def test_policy_input_rejects_non_utf8_and_total_size(tmp_path, monkeypatch):
    module = _audit_module()
    host_etc = tmp_path / "etc"
    host_etc.mkdir()
    sudoers = host_etc / "sudoers"
    sudoers.write_bytes(b"\xff")
    monkeypatch.setattr(
        module,
        "read_regular",
        lambda _path, _maximum: _snapshot(b"\xff", uid=os.getuid()),
    )
    with pytest.raises(module.HardeningError, match="not UTF-8"):
        module.audit_privilege_policies(
            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
        )

    monkeypatch.setattr(module, "_policy_files", lambda *_a: [sudoers] * 9)
    monkeypatch.setattr(
        module,
        "read_regular",
        lambda _path, _maximum: _snapshot(b"#" * (256 * 1024), uid=os.getuid()),
    )
    with pytest.raises(module.HardeningError, match="exceeds safe limit"):
        module.audit_privilege_policies(
            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
        )


def test_broad_sudo_and_pkla_paths_are_rejected(tmp_path, monkeypatch):
    module = _audit_module()
    host_etc = tmp_path / "etc"
    host_etc.mkdir()
    sudoers = host_etc / "sudoers"
    sudoers.write_text("%ALL ALL=(ALL) ALL\n", encoding="utf-8")
    monkeypatch.setattr(
        module,
        "read_regular",
        lambda path, maximum: _snapshot(path.read_bytes(), uid=os.getuid()),
    )
    with pytest.raises(module.HardeningError, match="broad sudo"):
        module.audit_privilege_policies(
            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
        )

    sudoers.unlink()
    policy = host_etc / "polkit-1/localauthority/policy.pkla"
    policy.parent.mkdir(parents=True)
    policy.write_text(
        "Identity=unix-user:*\nAction=org.freedesktop.policykit.exec\nResultAny=yes\n",
        encoding="utf-8",
    )
    with pytest.raises(module.HardeningError, match="root-equivalent broad polkit"):
        module.audit_privilege_policies(
            "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "share"
        )