Files
nest/tests/test_images.py
T

449 lines
22 KiB
Python

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, ValidationError
from uvm.images import (
_FIRECRACKER_DEMO_PUBLIC_KEY,
_GuestFile,
_enable_ssh_password_authentication,
_set_shadow_password,
_without_firecracker_demo_key,
ImageStore,
)
from uvm.integrity import write_manifest
from uvm.system import CommandResult, CommandRunner
class PasswordHashRunner:
def __init__(self) -> None:
self.command: tuple[str, ...] | None = None
self.calls: list[tuple[str, ...]] = []
self.options: dict[str, object] = {}
def run(self, command, **options) -> CommandResult:
self.command = tuple(str(part) for part in command)
self.calls.append(self.command)
self.options = options
stdout = "$6$salt$password-hash\n" if self.command[0] == "openssl" else ""
return CommandResult(self.command, 0, stdout=stdout)
class ImageStoreTests(unittest.TestCase):
def test_uses_the_persisted_install_manifest_when_strict_mode_is_enabled(self) -> None:
with tempfile.TemporaryDirectory() as temporary_directory:
settings = Settings(base=Path(temporary_directory), allow_unverified_downloads=False)
settings.images_dir.mkdir(parents=True)
settings.kernel_image.write_bytes(b"kernel")
settings.rootfs_image.write_bytes(b"rootfs")
write_manifest(
settings.integrity_manifest_path,
{
"kernel": hashlib.sha256(b"kernel").hexdigest(),
"rootfs": hashlib.sha256(b"rootfs").hexdigest(),
"firecracker": "0" * 64,
"jailer": "1" * 64,
},
verified=True,
)
assets = ImageStore(settings).installed_assets()
self.assertEqual(assets.kernel.name, "vmlinux")
self.assertEqual(assets.rootfs.name, "ubuntu.ext4")
def test_hashes_password_through_stdin_without_putting_it_in_argv(self) -> None:
runner = PasswordHashRunner()
store = ImageStore(Settings(), runner=runner) # type: ignore[arg-type]
password_hash = store._password_hash("secret-value")
self.assertEqual(password_hash, "$6$salt$password-hash")
self.assertEqual(runner.command, ("openssl", "passwd", "-6", "-stdin"))
self.assertNotIn("secret-value", runner.command)
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"
updated = _set_shadow_password(original, "root", "$6$salt$hash")
self.assertEqual(
updated,
"root:$6$salt$hash:1:0:99999:7:::\nservice:!:1:0:99999:7:::\n",
)
with self.assertRaises(UvmError):
_set_shadow_password(original, "missing", "$6$salt$hash")
def test_enables_root_password_login_before_existing_sshd_settings(self) -> None:
updated = _enable_ssh_password_authentication(
"HostKey /insecure/shared-key\n"
"PasswordAuthentication no\n"
"PermitRootLogin prohibit-password\n",
"root",
)
self.assertTrue(
updated.startswith(
"# Managed by uvm\n"
"HostKey /etc/ssh/ssh_host_rsa_key\n"
"HostKey /etc/ssh/ssh_host_ecdsa_key\n"
"HostKey /etc/ssh/ssh_host_ed25519_key\n"
"PasswordAuthentication no\n"
"KbdInteractiveAuthentication no\n"
"ChallengeResponseAuthentication no\n"
"PermitRootLogin yes\n"
"Match User root\n"
" PasswordAuthentication yes\n"
"Match all\n"
)
)
self.assertNotIn("/insecure/shared-key", updated)
self.assertNotIn("PermitRootLogin prohibit-password", updated)
def test_non_root_credentials_preserve_the_existing_root_login_policy(self) -> None:
updated = _enable_ssh_password_authentication(
"PermitRootLogin no\nChallengeResponseAuthentication yes\n",
"service",
)
self.assertIn("PermitRootLogin no", updated)
self.assertIn("Match User service\n PasswordAuthentication yes", updated)
self.assertNotIn("ChallengeResponseAuthentication yes", updated)
def test_full_ubuntu_sshd_global_directives_stay_outside_match_blocks(self) -> None:
updated = _enable_ssh_password_authentication(
"Include /etc/ssh/sshd_config.d/*.conf\n"
"UsePAM yes\nSubsystem sftp /usr/lib/openssh/sftp-server\n"
"Match Group restricted\n X11Forwarding no\n PasswordAuthentication no\n",
"codemate-user",
)
self.assertLess(updated.index("UsePAM yes"), updated.index("Match User codemate-user"))
self.assertLess(updated.index("Subsystem sftp"), updated.index("Match User codemate-user"))
self.assertLess(updated.index("Match User codemate-user"), updated.index("Match Group restricted"))
self.assertIn("Match Group restricted\n X11Forwarding no", updated)
@unittest.skipUnless(
shutil.which("sshd") and shutil.which("ssh-keygen") and Path("/run/sshd").is_dir(),
"OpenSSH server tools and the privilege-separation directory are required",
)
def test_openssh_accepts_the_full_config_and_limits_password_auth_to_the_guest_user(self) -> None:
runner = CommandRunner(emit=None)
with tempfile.TemporaryDirectory() as temporary:
key = Path(temporary) / "host_key"
runner.run(("ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", key), capture=True)
contents = _enable_ssh_password_authentication(
"UsePAM yes\nSubsystem sftp internal-sftp\nPermitRootLogin no\n",
"codemate-user",
)
for key_type in ("rsa", "ecdsa", "ed25519"):
contents = contents.replace(f"/etc/ssh/ssh_host_{key_type}_key", str(key))
config = Path(temporary) / "sshd_config"
config.write_text(contents, encoding="utf-8")
for username, expected in (("codemate-user", "yes"), ("root", "no")):
with self.subTest(username=username):
result = runner.run((
shutil.which("sshd"), "-T", "-f", config,
"-C", f"user={username},host=localhost,addr=127.0.0.1",
), capture=True)
self.assertIn(f"passwordauthentication {expected}\n", result.stdout)
def test_removes_only_the_public_firecracker_demo_key(self) -> None:
own_key = "ssh-ed25519 AAAA-own-key developer@example"
updated = _without_firecracker_demo_key(
f"{_FIRECRACKER_DEMO_PUBLIC_KEY} demo\n{own_key}\n"
)
self.assertEqual(updated, f"{own_key}\n")
def test_generates_three_unique_guest_host_key_types(self) -> None:
runner = PasswordHashRunner()
store = ImageStore(Settings(), runner=runner) # type: ignore[arg-type]
with (
tempfile.TemporaryDirectory() as temporary_directory,
patch.object(store, "_read_guest_file", return_value=None),
patch.object(store, "_write_guest_file") as write_guest_file,
):
store._replace_guest_host_keys(
Path("/tmp/rootfs.ext4"),
Path(temporary_directory),
)
keygen_calls = [call for call in runner.calls if call[0] == "ssh-keygen"]
self.assertEqual([call[3] for call in keygen_calls], ["rsa", "ecdsa", "ed25519"])
self.assertEqual(write_guest_file.call_count, 6)
def test_optional_guest_file_inspection_fails_closed_on_debugfs_errors(self) -> None:
class FailedRunner:
@staticmethod
def run(_command, **_options) -> CommandResult:
return CommandResult(
args=("debugfs",),
returncode=0,
stderr="filesystem checksum failure",
)
store = ImageStore(Settings(), runner=FailedRunner()) # type: ignore[arg-type]
with tempfile.TemporaryDirectory() as temporary_directory:
with self.assertRaises(UvmError):
store._read_guest_file(
Path("/tmp/rootfs.ext4"),
"/root/.ssh/authorized_keys",
Path(temporary_directory) / "authorized_keys",
required=False,
)
def test_refuses_to_replace_files_with_unpreserved_security_metadata(self) -> None:
runner = PasswordHashRunner()
store = ImageStore(Settings(), runner=runner) # type: ignore[arg-type]
for guest_file in (
_GuestFile(Path("/tmp/shadow"), 0o640, 0, 42, links=2),
_GuestFile(
Path("/tmp/shadow"),
0o640,
0,
42,
has_extended_attributes=True,
),
):
with self.subTest(guest_file=guest_file):
with self.assertRaises(UvmError):
store._write_guest_file(
Path("/tmp/rootfs.ext4"),
"/etc/shadow",
guest_file,
)
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/sudoers": "Defaults env_reset\nroot ALL=(ALL:ALL) ALL\n%sudo ALL=(ALL:ALL) ALL\n",
"/usr/bin/sudo": "sudo binary fixture\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")
mode = {"/etc/shadow": 0o640, "/etc/gshadow": 0o640, "/etc/sudoers": 0o440, "/usr/bin/sudo": 0o4755}
path.chmod(mode.get(guest_path, 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")
self.assertIn("codemate-user ALL=(ALL:ALL) PASSWD: ALL\n", self.contents("/etc/sudoers"))
self.assertIn("Defaults:codemate-user !rootpw, !runaspw, !targetpw\n", self.contents("/etc/sudoers"))
self.assert_metadata("/etc/sudoers", 0o440, 0, 0)
self.assert_metadata("/usr/bin/sudo", 0o4755, 0, 0)
if shutil.which("visudo"):
sudoers = self.base / "sudoers"
sudoers.write_text(self.contents("/etc/sudoers"), encoding="utf-8")
sudoers.chmod(0o440)
self.runner.run(("visudo", "-c", "-f", sudoers), capture=True)
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")
if username != "root":
self.assertIn(f"{username} ALL=(ALL:ALL) PASSWD: ALL\n", self.contents("/etc/sudoers"))
else:
self.assertEqual(self.contents("/etc/sudoers"), self.files["/etc/sudoers"])
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"])
def test_custom_users_fail_with_an_actionable_error_on_stripped_images(self) -> None:
del self.files["/usr/bin/sudo"]
del self.files["/etc/sudoers"]
self.make_disk()
with self.assertRaisesRegex(UvmError, "guest image needs setuid-root sudo.*uvm install --force"):
self.store.provision_credentials(self.disk, "codemate-user", "custom-password")
self.assertEqual(self.contents("/etc/shadow"), self.files["/etc/shadow"])
def test_sudo_without_setuid_permission_does_not_produce_a_broken_administrator(self) -> None:
self.make_disk()
self.runner.run(("debugfs", "-w", "-R", "set_inode_field /usr/bin/sudo mode 0100755", self.disk), capture=True)
with self.assertRaisesRegex(UvmError, "guest image needs setuid-root sudo"):
self.store.provision_credentials(self.disk, "codemate-user", "custom-password")