108 lines
5.4 KiB
Python
108 lines
5.4 KiB
Python
"""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()
|