Bootstrap private node checkouts and operator access from the workstation

This commit is contained in:
deytenit committed 2026-11-08 12:39:49 +03:00
1 parent 702857ec0a
commit ef5be2331d
10 files changed
+616 -137

No files matched your search

-60
View File
@@ -214,63 +214,3 @@ class HostSetup(unittest.TestCase):
self.assertEqual({call.args[0] for call in ownership.call_args_list},
{"HEAD", "node", "other"})
self.assertTrue(all(not call.kwargs["follow_symlinks"] for call in ownership.call_args_list))
def test_operator_provisioning_creates_account_then_preserves_keys_on_rerun(self):
adam = action("config/adam")
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary).resolve()
home = root / "adam"
account = SimpleNamespace(pw_dir=str(home), pw_uid=1001)
key = root / "operator"
subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(key)], check=True)
public = key.with_suffix(".pub")
ctx = SimpleNamespace(sys=Mock(), host=Mock())
ctx.sys.run.side_effect = lambda cmd, **kw: subprocess.run(cmd, input=kw["input"], text=True,
capture_output=True, check=True)
def privileged(command, **kwargs):
if command[0] == "useradd":
home.mkdir()
if command[:6] == ["runuser", "-u", "adam", "--", "sh", "-ec"]:
subprocess.run(command[4:], text=True, input=kwargs["input"], check=True)
ctx.sys.sudo.side_effect = privileged
actual_stat = Path.stat
def metadata(path, *args, **kwargs):
stat = actual_stat(path, *args, **kwargs)
if path == home:
fields = list(stat)
fields[4] = account.pw_uid
return os.stat_result(fields)
return stat
with (patch.object(adam, "HOME", home), patch.object(Path, "stat", metadata),
patch.object(adam.grp, "getgrnam", side_effect=KeyError),
patch.object(adam.pwd, "getpwnam", side_effect=[KeyError, account, account])):
args = SimpleNamespace(public_key_file=str(public))
adam.run(ctx, args)
authorized = home / ".ssh/authorized_keys"
with authorized.open("a") as stream:
stream.write("existing operator entry\n")
adam.run(ctx, args)
self.assertEqual(authorized.read_text().splitlines().count(public.read_text().strip()), 1)
self.assertIn("existing operator entry", authorized.read_text())
commands = [call.args[0] for call in ctx.sys.sudo.call_args_list]
self.assertEqual(sum(cmd[0] == "useradd" for cmd in commands), 1)
self.assertIn(["runuser", "-u", "adam", "--", "sudo", "-n", "true"], commands)
def test_existing_system_account_cannot_be_promoted(self):
adam = action("config/adam")
ctx = SimpleNamespace(sys=Mock())
account = SimpleNamespace(pw_dir="/home/adam", pw_uid=999)
with patch.object(adam.pwd, "getpwnam", return_value=account):
with self.assertRaisesRegex(RuntimeError, "incompatible UID"):
adam.run(ctx, SimpleNamespace(public_key_file=None))
self.assertFalse(ctx.sys.mock_calls)
def test_invalid_operator_key_is_rejected_before_provisioning(self):
adam = action("config/adam")
with tempfile.NamedTemporaryFile(mode="w") as key:
key.write("private key fixture")
key.flush()
ctx = SimpleNamespace(sys=Mock())
with self.assertRaisesRegex(ValueError, "OpenSSH public key"):
adam.run(ctx, SimpleNamespace(public_key_file=key.name))
self.assertFalse(ctx.sys.mock_calls)