76 lines
3.6 KiB
Python
76 lines
3.6 KiB
Python
"""Node credential repair must be targeted, idempotent and silent about secrets."""
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|