atlas-iac/testing/tests/test_node_admin_access.py

"""Node credential repair must be targeted, idempotent and silent about secrets."""
import importlib.util
from pathlib import Path
import unittest
import tempfile
from unittest.mock import patch, Mock

spec = importlib.util.spec_from_file_location(
    "node_admin_access", Path(__file__).resolve().parents[2] / "scripts/node_admin_access.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)


class NodeAccessTests(unittest.TestCase):
    def setUp(self):
        self.payload = {"hostname": "titan-test", "passwords": {"atlas": "synthetic-only"}}
        self.root = patch.object(module.os, "geteuid", return_value=0)
        self.host = patch.object(module.socket, "gethostname", return_value="titan-test")
        self.account = patch.object(module.pwd, "getpwnam")
        for item in [self.root, self.host, self.account]:
            item.start()
            self.addCleanup(item.stop)

    def test_wrong_host_never_changes_password(self):
        with patch.object(module.subprocess, "run") as run:
            self.payload["hostname"] = "wrong-node"
            with self.assertRaisesRegex(ValueError, "hostname_mismatch"):
                module.run(self.payload, True)
            run.assert_not_called()

    def test_validate_all_accounts_before_mutation(self):
        with patch.object(module.subprocess, "run") as run:
            self.payload["passwords"]["other"] = "synthetic"
            with self.assertRaisesRegex(ValueError, "account_not_allowed"):
                module.run(self.payload, True)
            run.assert_not_called()

    def test_password_record_injection_rejected(self):
        self.payload["passwords"]["atlas"] = "bad\nroot:injected"
        with self.assertRaisesRegex(ValueError, "invalid_password"):
            module.run(self.payload, True)

    def test_audit_is_read_only(self):
        with patch.object(module, "password_status", return_value={"vault_password_matches": False}), \
             patch.object(module.subprocess, "run") as run:
            result = module.run(self.payload)
            run.assert_not_called()
            self.assertEqual(result["changed_accounts"], [])

    def test_matching_password_is_unchanged(self):
        with patch.object(module, "password_status", return_value={"vault_password_matches": True}), \
             patch.object(module.subprocess, "run") as run:
            module.run(self.payload, True)
            run.assert_not_called()

    def test_restore_uses_stdin_and_returns_no_secret(self):
        with patch.object(module, "password_status", side_effect=[
            {"vault_password_matches": False, "locked": True},
            {"vault_password_matches": True, "locked": False}]), \
             patch.object(module.subprocess, "run", return_value=Mock(returncode=0)) as run:
            result = module.run(self.payload, True)
            self.assertEqual(run.call_args.args[0], ["/usr/sbin/chpasswd"])
            self.assertEqual(run.call_args.kwargs["input"], "atlas:synthetic-only\n")
            self.assertNotIn("synthetic-only", str(result))
            self.assertEqual(result["changed_accounts"], ["atlas"])
            self.assertTrue(result["after"]["atlas"]["vault_password_matches"])

    def test_failed_update_does_not_echo_subprocess(self):
        with patch.object(module, "password_status", return_value={"vault_password_matches": False}), \
             patch.object(module.subprocess, "run", return_value=Mock(returncode=1, stderr="SECRET")):
            with self.assertRaisesRegex(ValueError, "^password_update_failed$"):
                module.run(self.payload, True)

    def test_legacy_grant_retirement_keeps_a_rollback_copy(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            grant = root / "etc/sudoers.d/90-hecate-atlas"
            grant.parent.mkdir(parents=True)
            grant.write_text("atlas ALL=(ALL) NOPASSWD: /usr/bin/systemctl, /usr/sbin/poweroff, "
                             "/sbin/poweroff, /usr/local/bin/hecate\n")
            result = {"after": {"atlas": {"vault_password_matches": True}}}
            with patch.object(module, "Path", side_effect=lambda p: root / p.lstrip("/")), \
                 patch.object(module.subprocess, "run", return_value=Mock(returncode=0)):
                module.retire_legacy_sudo(result)
                module.retire_legacy_sudo(result)
            self.assertFalse(grant.exists())
            self.assertTrue((root / "var/lib/atlas-maintenance/legacy-sudo-20261004/90-hecate-atlas").exists())

    def test_custom_sudo_rule_is_never_removed(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            grant = root / "etc/sudoers.d/90-hecate-atlas"
            grant.parent.mkdir(parents=True)
            grant.write_text("reviewed custom rule")
            result = {"after": {"atlas": {"vault_password_matches": True}}}
            with patch.object(module, "Path", side_effect=lambda p: root / p.lstrip("/")):
                with self.assertRaisesRegex(ValueError, "legacy_grant_modified_requires_review"):
                    module.retire_legacy_sudo(result)
            self.assertEqual(grant.read_text(), "reviewed custom rule")

    def test_legacy_grant_requires_working_password(self):
        with self.assertRaisesRegex(ValueError, "atlas_password_not_verified"):
            module.retire_legacy_sudo({"after": {"atlas": {"vault_password_matches": False}}})


if __name__ == "__main__":
    unittest.main()