feat: enhance user account management during VM creation with automatic user creation and metadata preservation
This commit is contained in:
+190
-2
@@ -1,13 +1,15 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import shutil
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from uvm.config import Settings
|
||||
from uvm.errors import UvmError
|
||||
from uvm.errors import UvmError, ValidationError
|
||||
from uvm.images import (
|
||||
_FIRECRACKER_DEMO_PUBLIC_KEY,
|
||||
_GuestFile,
|
||||
@@ -17,7 +19,7 @@ from uvm.images import (
|
||||
ImageStore,
|
||||
)
|
||||
from uvm.integrity import write_manifest
|
||||
from uvm.system import CommandResult
|
||||
from uvm.system import CommandResult, CommandRunner
|
||||
|
||||
|
||||
class PasswordHashRunner:
|
||||
@@ -69,6 +71,16 @@ class ImageStoreTests(unittest.TestCase):
|
||||
self.assertEqual(runner.options["input_text"], "secret-value\n")
|
||||
self.assertTrue(runner.options["sensitive"])
|
||||
|
||||
def test_rejects_invalid_usernames_before_accessing_the_guest_disk(self) -> None:
|
||||
runner = PasswordHashRunner()
|
||||
store = ImageStore(Settings(), runner=runner) # type: ignore[arg-type]
|
||||
|
||||
for username in ("../root", "user\nroot", "user:1000", "user name"):
|
||||
with self.subTest(username=username), self.assertRaises(ValidationError):
|
||||
store.provision_credentials(Path("/tmp/rootfs.ext4"), username, "secret")
|
||||
|
||||
self.assertEqual(runner.calls, [])
|
||||
|
||||
def test_updates_only_the_requested_shadow_entry(self) -> None:
|
||||
original = "root:*:1:0:99999:7:::\nservice:!:1:0:99999:7:::\n"
|
||||
|
||||
@@ -187,3 +199,179 @@ class ImageStoreTests(unittest.TestCase):
|
||||
)
|
||||
|
||||
self.assertEqual(runner.calls, [])
|
||||
|
||||
def test_directory_creation_checks_effects_even_when_debugfs_reports_success(self) -> None:
|
||||
class FailedMkdirRunner:
|
||||
@staticmethod
|
||||
def run(command, **_options) -> CommandResult:
|
||||
return CommandResult(
|
||||
args=tuple(map(str, command)),
|
||||
returncode=0,
|
||||
stderr="File not found by ext2_lookup" if "stat " in str(command[2]) else "",
|
||||
)
|
||||
|
||||
store = ImageStore(Settings(), runner=FailedMkdirRunner()) # type: ignore[arg-type]
|
||||
with self.assertRaisesRegex(UvmError, "could not verify guest rootfs directory"):
|
||||
store._create_guest_directory(Path("/tmp/rootfs.ext4"), "/home/user", 0o700, 1000, 1000)
|
||||
|
||||
|
||||
@unittest.skipUnless(
|
||||
all(shutil.which(tool) for tool in ("mkfs.ext4", "debugfs", "e2fsck", "cp", "openssl", "ssh-keygen")),
|
||||
"e2fsprogs, coreutils, OpenSSL, and OpenSSH are required for offline credential checks",
|
||||
)
|
||||
class OfflineGuestCredentialsTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
temporary = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(temporary.cleanup)
|
||||
self.base = Path(temporary.name)
|
||||
self.template = self.base / "template.ext4"
|
||||
self.disk = self.base / "rootfs.ext4"
|
||||
self.runner = CommandRunner(emit=None)
|
||||
self.store = ImageStore(Settings(base=self.base), self.runner)
|
||||
self.files = {
|
||||
"/etc/passwd": (
|
||||
"root:x:0:0:root:/root:/bin/bash\n"
|
||||
"service:x:999:1002::/nonexistent:/usr/sbin/nologin\n"
|
||||
"ubuntu:x:1000:1000:Ubuntu:/home/ubuntu:/bin/bash\n"
|
||||
"nobody:x:65534:65534:nobody:/nonexistent:/usr/sbin/nologin\n"
|
||||
),
|
||||
"/etc/shadow": (
|
||||
"root:*:1:0:99999:7:::\nservice:!:1:0:99999:7:::\n"
|
||||
"ubuntu:!:2:0:99999:7:::\nnobody:*:1:0:99999:7:::\n"
|
||||
),
|
||||
"/etc/group": "root:x:0:\nshadow:x:42:\nubuntu:x:1000:\nreserved:x:1001:\nnogroup:x:65534:\n",
|
||||
"/etc/gshadow": "root:*::\nshadow:*::\nubuntu:!::\nreserved:!::\nnogroup:*::\n",
|
||||
"/etc/ssh/sshd_config": "PasswordAuthentication no\nPermitRootLogin no\n",
|
||||
"/root/.ssh/authorized_keys": f"{_FIRECRACKER_DEMO_PUBLIC_KEY} demo\nssh-ed25519 AAAA-own-key\n",
|
||||
}
|
||||
|
||||
def make_disk(self) -> None:
|
||||
root = self.base / "guest"
|
||||
for guest_path, contents in self.files.items():
|
||||
path = root / guest_path.lstrip("/")
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(contents, encoding="utf-8")
|
||||
path.chmod(0o640 if guest_path in ("/etc/shadow", "/etc/gshadow") else 0o644)
|
||||
with self.template.open("wb") as image:
|
||||
image.truncate(16 * 1024 * 1024)
|
||||
self.runner.run(("mkfs.ext4", "-q", "-F", "-d", root, self.template), capture=True)
|
||||
for guest_path in self.files:
|
||||
gid = 42 if guest_path in ("/etc/shadow", "/etc/gshadow") else 0
|
||||
for field, value in (("uid", 0), ("gid", gid)):
|
||||
self.runner.run(
|
||||
("debugfs", "-w", "-R", f"set_inode_field {guest_path} {field} {value}", self.template),
|
||||
capture=True,
|
||||
)
|
||||
self.store._copy_sparse(self.template, self.disk)
|
||||
|
||||
def contents(self, guest_path: str) -> str:
|
||||
return self.runner.run(("debugfs", "-R", f"cat {guest_path}", self.disk), capture=True).stdout
|
||||
|
||||
def entry(self, guest_path: str, username: str) -> list[str]:
|
||||
return next(line.split(":") for line in self.contents(guest_path).splitlines()
|
||||
if line.split(":", 1)[0] == username)
|
||||
|
||||
def assert_metadata(self, guest_path: str, mode: int, uid: int, gid: int) -> None:
|
||||
result = self.runner.run(("debugfs", "-R", f"stat {guest_path}", self.disk), capture=True)
|
||||
self.assertRegex(result.stdout, rf"Mode:\s+0*{mode:o}\b")
|
||||
self.assertRegex(result.stdout, rf"User:\s+{uid}\s+Group:\s+{gid}\b")
|
||||
|
||||
def assert_password(self, username: str, password: str) -> None:
|
||||
password_hash = self.entry("/etc/shadow", username)[1]
|
||||
self.assertTrue(password_hash.startswith("$6$"))
|
||||
result = self.runner.run(
|
||||
("openssl", "passwd", "-6", "-salt", password_hash.split("$")[2], "-stdin"),
|
||||
capture=True, input_text=f"{password}\n", sensitive=True,
|
||||
)
|
||||
self.assertEqual(password_hash, result.stdout.strip())
|
||||
|
||||
def test_custom_username_creates_a_login_account_on_the_private_disk(self) -> None:
|
||||
self.make_disk()
|
||||
template_digest = hashlib.sha256(self.template.read_bytes()).hexdigest()
|
||||
day_before = int(time.time() // 86400)
|
||||
|
||||
self.store.provision_credentials(self.disk, "codemate-user", "custom-password")
|
||||
|
||||
# UID 1000 and GIDs 1000/1001 are occupied; GID 1002 is used only in passwd.
|
||||
self.assertEqual(
|
||||
self.contents("/etc/passwd"),
|
||||
self.files["/etc/passwd"] + "codemate-user:x:1001:1003::/home/codemate-user:/bin/sh\n",
|
||||
)
|
||||
self.assertEqual(self.contents("/etc/group"), self.files["/etc/group"] + "codemate-user:x:1003:\n")
|
||||
self.assertEqual(self.contents("/etc/gshadow"), self.files["/etc/gshadow"] + "codemate-user:!::\n")
|
||||
self.assertTrue(self.contents("/etc/shadow").startswith(self.files["/etc/shadow"]))
|
||||
shadow = self.entry("/etc/shadow", "codemate-user")
|
||||
self.assertLessEqual(day_before, int(shadow[2]))
|
||||
self.assertLessEqual(int(shadow[2]), int(time.time() // 86400))
|
||||
self.assertEqual(shadow[3:], ["0", "99999", "7", "", "", ""])
|
||||
self.assert_password("codemate-user", "custom-password")
|
||||
for guest_path in ("/etc/passwd", "/etc/group"):
|
||||
self.assert_metadata(guest_path, 0o644, 0, 0)
|
||||
for guest_path in ("/etc/shadow", "/etc/gshadow"):
|
||||
self.assert_metadata(guest_path, 0o640, 0, 42)
|
||||
self.assert_metadata("/home", 0o755, 0, 0)
|
||||
self.assert_metadata("/home/codemate-user", 0o700, 1001, 1003)
|
||||
sshd_config = self.contents("/etc/ssh/sshd_config")
|
||||
self.assertIn("Match User codemate-user\n PasswordAuthentication yes\n", sshd_config)
|
||||
self.assertIn("PermitRootLogin no", sshd_config)
|
||||
self.assertEqual(self.contents("/root/.ssh/authorized_keys"), "ssh-ed25519 AAAA-own-key\n")
|
||||
self.runner.run(("e2fsck", "-f", "-n", self.disk), capture=True)
|
||||
self.assertEqual(hashlib.sha256(self.template.read_bytes()).hexdigest(), template_digest)
|
||||
|
||||
def test_existing_root_and_non_root_accounts_keep_their_identity_and_home(self) -> None:
|
||||
self.files["/home/ubuntu/data"] = "existing guest data\n"
|
||||
self.make_disk()
|
||||
|
||||
for username in ("root", "ubuntu"):
|
||||
with self.subTest(username=username):
|
||||
self.store._copy_sparse(self.template, self.disk)
|
||||
original_shadow = self.entry("/etc/shadow", username)
|
||||
self.store.provision_credentials(self.disk, username, "changed-password")
|
||||
|
||||
for guest_path in ("/etc/passwd", "/etc/group", "/etc/gshadow", "/home/ubuntu/data"):
|
||||
self.assertEqual(self.contents(guest_path), self.files[guest_path])
|
||||
self.assertEqual(self.entry("/etc/shadow", username)[2:], original_shadow[2:])
|
||||
self.assert_password(username, "changed-password")
|
||||
for line in self.files["/etc/shadow"].splitlines():
|
||||
if line.split(":", 1)[0] != username:
|
||||
self.assertIn(line + "\n", self.contents("/etc/shadow"))
|
||||
|
||||
def test_new_user_reuses_an_existing_named_group(self) -> None:
|
||||
self.files["/etc/group"] += "codemate-user:x:1100:ubuntu\n"
|
||||
self.files["/etc/gshadow"] += "codemate-user:!::ubuntu\n"
|
||||
self.make_disk()
|
||||
|
||||
self.store.provision_credentials(self.disk, "codemate-user", "custom-password")
|
||||
|
||||
self.assertEqual(self.entry("/etc/passwd", "codemate-user")[2:4], ["1001", "1100"])
|
||||
for guest_path in ("/etc/group", "/etc/gshadow"):
|
||||
self.assertEqual(self.contents(guest_path), self.files[guest_path])
|
||||
self.assert_metadata("/home/codemate-user", 0o700, 1001, 1100)
|
||||
|
||||
def test_new_user_supports_images_without_gshadow_and_without_final_newlines(self) -> None:
|
||||
del self.files["/etc/gshadow"]
|
||||
for guest_path in ("/etc/passwd", "/etc/shadow", "/etc/group"):
|
||||
self.files[guest_path] = self.files[guest_path].rstrip("\n")
|
||||
self.files["/home/retained"] = "existing home directory contents\n"
|
||||
self.make_disk()
|
||||
|
||||
self.store.provision_credentials(self.disk, "codemate-user", "custom-password")
|
||||
|
||||
self.assertEqual(self.entry("/etc/passwd", "codemate-user")[2:4], ["1001", "1003"])
|
||||
self.assertEqual(self.entry("/etc/group", "codemate-user"), ["codemate-user", "x", "1003", ""])
|
||||
self.assert_password("codemate-user", "custom-password")
|
||||
self.assertEqual(self.contents("/home/retained"), self.files["/home/retained"])
|
||||
self.assertIsNone(self.store._read_guest_file(
|
||||
self.disk, "/etc/gshadow", self.base / "gshadow", required=False,
|
||||
))
|
||||
|
||||
def test_existing_home_path_is_not_overwritten(self) -> None:
|
||||
self.files["/home/codemate-user"] = "existing guest file\n"
|
||||
self.make_disk()
|
||||
|
||||
with self.assertRaisesRegex(UvmError, "cannot create guest rootfs directory: /home/codemate-user"):
|
||||
self.store.provision_credentials(self.disk, "codemate-user", "custom-password")
|
||||
|
||||
self.assertEqual(self.contents("/home/codemate-user"), self.files["/home/codemate-user"])
|
||||
self.assertEqual(self.contents("/etc/passwd"), self.files["/etc/passwd"])
|
||||
self.assertEqual(self.contents("/etc/shadow"), self.files["/etc/shadow"])
|
||||
|
||||
Reference in New Issue
Block a user