atlas-iac/testing/tests/test_hermes_node_account_privilege_audit.py

"""Host privilege-policy audits for the Hermes node SSH identity."""

from __future__ import annotations

import shutil
import subprocess
from pathlib import Path

import pytest

from testing.tests.test_hermes_node_account_support import _fixture


@pytest.mark.parametrize(
    "database,record",
    [
        ("group", "disk:x:6:atlas,hermes-agent\n"),
        ("gshadow", "disk:!:hermes-agent:atlas\n"),
        ("gshadow", "disk:!::atlas,hermes-agent\n"),
    ],
)
def test_unexpected_group_or_gshadow_membership_fails_before_writes(
    tmp_path: Path, monkeypatch, database: str, record: str
):
    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
    path = module.HOST_ETC / database
    path.write_text(
        "\n".join(
            line
            for line in originals[database].splitlines()
            if not line.startswith("disk:")
        )
        + "\n"
        + record,
        encoding="utf-8",
    )
    before = {name: (module.HOST_ETC / name).read_bytes() for name in originals}

    with pytest.raises(module.HardeningError, match="group access"):
        module.reconcile(public_key)

    assert {name: (module.HOST_ETC / name).read_bytes() for name in originals} == before


@pytest.mark.parametrize(
    "relative,value",
    [
        ("sudoers.d/hermes", "hermes-agent ALL=(ALL:ALL) NOPASSWD: ALL\n"),
        ("sudoers", "ALL ALL=(ALL:ALL) ALL\n"),
        (
            "polkit-1/rules.d/90-root.rules",
            'if (subject.user == "hermes-agent") return polkit.Result.YES;\n',
        ),
        (
            "polkit-1/rules.d/90-root.rules",
            "unix-user:* org.freedesktop.policykit.exec polkit.Result.YES\n",
        ),
    ],
)
def test_sudo_and_root_equivalent_polkit_authority_fail_closed(
    tmp_path: Path, monkeypatch, relative: str, value: str
):
    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
    policy = module.HOST_ETC / relative
    policy.parent.mkdir(parents=True, exist_ok=True)
    policy.write_text(value, encoding="utf-8")

    with pytest.raises(module.HardeningError, match="sudo|polkit"):
        module.reconcile(public_key)

    assert all(
        (module.HOST_ETC / name).read_text() == original
        for name, original in originals.items()
    )


def test_external_group_authority_source_is_rejected(tmp_path: Path, monkeypatch):
    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
    (module.HOST_ETC / "nsswitch.conf").write_text("passwd: files\ngroup: files ldap\n")
    with pytest.raises(module.HardeningError, match="external group"):
        module.reconcile(public_key)
    assert all(
        (module.HOST_ETC / name).read_text() == original
        for name, original in originals.items()
    )


@pytest.mark.parametrize("database", ["passwd", "group", "initgroups", "shadow"])
def test_external_identity_authority_sources_fail_closed(
    tmp_path: Path, monkeypatch, database: str
):
    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
    (module.HOST_ETC / "nsswitch.conf").write_text(
        f"passwd: files\n{database}: files ldap\n", encoding="utf-8"
    )
    with pytest.raises(module.HardeningError, match="external group/account"):
        module.reconcile(public_key)
    assert all(
        (module.HOST_ETC / name).read_text() == original
        for name, original in originals.items()
    )


@pytest.mark.parametrize(
    "value",
    [
        "#1200 ALL=(ALL:ALL) NOPASSWD: ALL\n",
        "User_Alias HERMES_IDS = #1200\nHERMES_IDS ALL=(ALL:ALL) NOPASSWD: ALL\n",
        "User_Alias HERMES_GROUP = %#1200\nHERMES_GROUP ALL=(ALL:ALL) NOPASSWD: ALL\n",
    ],
)
def test_visudo_valid_numeric_and_alias_grants_fail_closed(
    tmp_path: Path, monkeypatch, value: str
):
    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
    monkeypatch.setattr(module, "ACCOUNT_UID", 1200)
    policy = module.HOST_ETC / "sudoers"
    policy.write_text(value, encoding="utf-8")
    visudo = shutil.which("visudo")
    if visudo is None:
        pytest.skip("visudo is not installed in this test environment")
    validation = subprocess.run(
        [visudo, "-c", "-f", str(policy)],
        check=False,
        capture_output=True,
        text=True,
    )
    assert validation.returncode == 0, validation.stderr

    with pytest.raises(module.HardeningError, match="sudo authority"):
        module.reconcile(public_key)

    assert all(
        (module.HOST_ETC / name).read_text() == original
        for name, original in originals.items()
    )


def test_sudo_include_is_bounded_to_the_audited_standard_directory(
    tmp_path: Path, monkeypatch
):
    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
    sudoers = module.HOST_ETC / "sudoers"
    sudoers.write_text("#includedir /opt/external-sudoers\n", encoding="utf-8")
    with pytest.raises(module.HardeningError, match="unaudited authority source"):
        module.reconcile(public_key)

    sudoers.write_text("#includedir /etc/sudoers.d\n", encoding="utf-8")
    module.reconcile(public_key)
    assert all(
        (module.HOST_ETC / name).read_text().startswith(original)
        for name, original in originals.items()
    )


@pytest.mark.parametrize(
    "relative,value",
    [
        (
            "polkit-1/rules.d/90-hermes.rules",
            "polkit.addRule(function(action, subject) {\n"
            "  if (subject.uid == 1200) return polkit.Result.YES;\n"
            "});\n",
        ),
        (
            "polkit-1/localauthority/50-local.d/hermes.pkla",
            "[Hermes]\nIdentity=unix-user:1200\n"
            "Action=org.freedesktop.policykit.exec\nResultActive=yes\n",
        ),
    ],
)
def test_numeric_polkit_grants_fail_closed(
    tmp_path: Path, monkeypatch, relative: str, value: str
):
    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
    monkeypatch.setattr(module, "ACCOUNT_UID", 1200)
    policy = module.HOST_ETC / relative
    policy.parent.mkdir(parents=True, exist_ok=True)
    policy.write_text(value, encoding="utf-8")

    with pytest.raises(module.HardeningError, match="polkit authority"):
        module.reconcile(public_key)

    assert all(
        (module.HOST_ETC / name).read_text() == original
        for name, original in originals.items()
    )


def test_writable_privilege_policy_is_rejected(tmp_path: Path, monkeypatch):
    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
    sudoers = module.HOST_ETC / "sudoers"
    sudoers.write_text("root ALL=(ALL:ALL) ALL\n")
    sudoers.chmod(0o666)
    with pytest.raises(module.HardeningError, match="unsafe mutation"):
        module.reconcile(public_key)
    assert all(
        (module.HOST_ETC / name).read_text() == original
        for name, original in originals.items()
    )