378 lines
17 KiB
Python
378 lines
17 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_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/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"])
|