943 lines
42 KiB
Python
943 lines
42 KiB
Python
"""Guest asset lookup and per-VM writable root filesystem handling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import tempfile
|
|
import time
|
|
from dataclasses import dataclass
|
|
from ipaddress import IPv4Address
|
|
from pathlib import Path
|
|
|
|
from . import guest_agent
|
|
from .config import MAX_STORAGE_MIB, MIN_STORAGE_MIB, Settings
|
|
from .errors import UvmError, ValidationError
|
|
from .integrity import load_manifest, verify_file
|
|
from .system import CommandRunner
|
|
from .validation import parse_username
|
|
|
|
|
|
_FIRECRACKER_DEMO_PUBLIC_KEY = (
|
|
"ssh-rsa "
|
|
"AAAAB3NzaC1yc2EAAAADAQABAAABAQCirWKrc1zDyvZufHGinIRNoeIot+C3idANxtqZyDHL9mYm"
|
|
"NeQzx9CjbjMgSDJ3xhIPP9mu3MP4Py/u3X5Wey98zN3EPboKdGRf6T2fFviK4i0q85LueDtsK0"
|
|
"IoWR459w87tC9NMwPb27C8jPqFod6nWfccdhEdM+veKkFh4Dk5TrPfYHDayXDPFEdz7jl0GedEH"
|
|
"fP9w11LPfa66D7731CdD3tMHAWLYxYmeXo58RXUaP6AgUK8uF/hL+E21q+wgTNPOQuRn2ekOjdu"
|
|
"J34oHkJ2i48tLqKdGKU6RwfFc3rW3TkeYTUqi3UY9EnwNWFJicez+nZ5bhr5KvRsfSZ2QvBj"
|
|
)
|
|
_MODE_PATTERN = re.compile(r"Mode:\s+0*([0-7]+)")
|
|
_OWNER_PATTERN = re.compile(r"User:\s+(\d+)\s+Group:\s+(\d+)")
|
|
_FLAGS_PATTERN = re.compile(r"Flags:\s+0x([0-9a-fA-F]+)")
|
|
_LINKS_PATTERN = re.compile(r"Links:\s+(\d+)")
|
|
_HOST_KEY_TYPES = (
|
|
("rsa", "3072"),
|
|
("ecdsa", "256"),
|
|
("ed25519", None),
|
|
)
|
|
|
|
# Keep the kernel's proto-boot default route until networkd takes over. On
|
|
# Ubuntu 22.04, KeepConfiguration=static preserves addresses but drops that route.
|
|
GUEST_NETWORK_TEMPLATE = (
|
|
"[Match]\nName=eth0\n\n[Network]\n"
|
|
"KeepConfiguration=yes\nDHCP=no\nLinkLocalAddressing=no\nIPv6AcceptRA=no\n"
|
|
)
|
|
_LEGACY_GUEST_NETWORK_TEMPLATE = GUEST_NETWORK_TEMPLATE.replace(
|
|
"KeepConfiguration=yes", "KeepConfiguration=static",
|
|
)
|
|
|
|
GUEST_AGENT_SERVICE = """\
|
|
[Unit]
|
|
Description=UVM guest command execution agent
|
|
After=local-fs.target network.target
|
|
ConditionPathExists=/etc/uvm-agent.json
|
|
StartLimitIntervalSec=0
|
|
|
|
[Service]
|
|
Type=simple
|
|
ExecStartPre=/bin/sh /usr/lib/uvm/prepare-agent.sh
|
|
ExecStart=/usr/bin/python3 -I /usr/lib/uvm/agent.py
|
|
Restart=on-failure
|
|
RestartSec=5
|
|
TimeoutStartSec=600
|
|
TimeoutStopSec=5
|
|
KillMode=control-group
|
|
Delegate=yes
|
|
UMask=0077
|
|
|
|
[Install]
|
|
WantedBy=multi-user.target
|
|
"""
|
|
|
|
# Runs only INSIDE the VM. Old Jammy guests can acquire the runtime without SSH
|
|
# or executing a retained guest's binaries in a privileged host chroot.
|
|
GUEST_AGENT_PREPARE = """\
|
|
set -eu
|
|
check_python() {
|
|
/usr/bin/python3 -I -c 'import sys, json, math, pwd, selectors, socket, struct, subprocess, threading, uuid; assert sys.version_info >= (3, 10)' 2>/dev/null
|
|
}
|
|
if check_python; then
|
|
exit 0
|
|
fi
|
|
export DEBIAN_FRONTEND=noninteractive LC_ALL=C
|
|
/usr/bin/apt-get -o DPkg::Lock::Timeout=300 -o APT::Update::Error-Mode=any update
|
|
/usr/bin/apt-get -o DPkg::Lock::Timeout=300 -y --no-install-recommends install python3
|
|
check_python
|
|
"""
|
|
|
|
|
|
# The original v4 template shipped this exact agent without an ownership
|
|
# manifest. Recognize that payload when creating from a cached v4 template.
|
|
_V4_TEMPLATE_AGENT_SHA256 = "dfdd4bd6ee3dc06fbdbda6402cb2c54921cff22190d8da889209559800f8b948"
|
|
|
|
|
|
def guest_agent_files() -> dict[str, bytes]:
|
|
"""The same standalone payload is used by the builder and offline upgrades."""
|
|
return {
|
|
"/usr/lib/uvm/agent.py": Path(guest_agent.__file__).read_bytes(),
|
|
"/usr/lib/uvm/prepare-agent.sh": GUEST_AGENT_PREPARE.encode("utf-8"),
|
|
"/etc/systemd/system/uvm-agent.service": GUEST_AGENT_SERVICE.encode("utf-8"),
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class GuestAssets:
|
|
kernel: Path
|
|
rootfs: Path
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _GuestFile:
|
|
path: Path
|
|
mode: int
|
|
uid: int
|
|
gid: int
|
|
flags: int | None = None
|
|
links: int = 1
|
|
has_extended_attributes: bool = False
|
|
|
|
|
|
class ImageStore:
|
|
"""Treat downloaded guest assets as templates, never as writable VM disks."""
|
|
|
|
def __init__(self, settings: Settings, runner: CommandRunner | None = None) -> None:
|
|
self._settings = settings
|
|
self._runner = runner or CommandRunner()
|
|
|
|
def installed_assets(self) -> GuestAssets:
|
|
kernel = self.installed_kernel()
|
|
rootfs = self._settings.rootfs_image
|
|
if not kernel.exists() or not rootfs.exists():
|
|
raise UvmError("guest assets missing. Run: sudo uvm install")
|
|
manifest = load_manifest(self._settings.integrity_manifest_path)
|
|
rootfs_checksum = self._settings.rootfs_sha256
|
|
if manifest.verified or self._settings.allow_unverified_downloads:
|
|
rootfs_checksum = rootfs_checksum or manifest.checksums.get("rootfs")
|
|
verify_file(
|
|
rootfs,
|
|
rootfs_checksum,
|
|
"guest root filesystem",
|
|
"UVM_ROOTFS_SHA256",
|
|
allow_unverified=self._settings.allow_unverified_downloads,
|
|
)
|
|
return GuestAssets(kernel=kernel, rootfs=rootfs)
|
|
|
|
def installed_kernel(self) -> Path:
|
|
"""A retained private disk does not depend on the original rootfs template."""
|
|
kernel = self._settings.kernel_image
|
|
if not kernel.is_file():
|
|
raise UvmError("guest kernel missing. Run: sudo uvm install")
|
|
manifest = load_manifest(self._settings.integrity_manifest_path)
|
|
checksum = self._settings.kernel_sha256
|
|
if manifest.verified or self._settings.allow_unverified_downloads:
|
|
checksum = checksum or manifest.checksums.get("kernel")
|
|
verify_file(kernel, checksum, "guest kernel", "UVM_KERNEL_SHA256",
|
|
allow_unverified=self._settings.allow_unverified_downloads)
|
|
return kernel
|
|
|
|
def create_vm_disk(
|
|
self, source: Path, destination: Path, *, storage_mib: int = MIN_STORAGE_MIB,
|
|
) -> Path:
|
|
"""Copy the template before Firecracker opens it read-write for a VM."""
|
|
|
|
self._validate_storage(storage_mib)
|
|
if destination.exists():
|
|
raise UvmError(f"VM disk already exists: {destination}")
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
self._copy_sparse(source, destination)
|
|
destination.chmod(0o600)
|
|
self._resize_filesystem(destination, storage_mib)
|
|
except BaseException as error:
|
|
try:
|
|
destination.unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
if isinstance(error, OSError):
|
|
raise UvmError(f"could not create VM disk {destination}: {error}") from error
|
|
raise
|
|
return destination
|
|
|
|
def disk_size_bytes(self, disk: Path) -> int:
|
|
"""Read logical disk capacity, including a resize published before state was saved."""
|
|
try:
|
|
return disk.stat().st_size
|
|
except OSError as error:
|
|
raise UvmError(f"could not inspect private disk capacity: {disk}: {error}") from error
|
|
|
|
def resize_vm_disk(self, disk: Path, storage_mib: int) -> None:
|
|
"""Resize a proven-offline private disk on a copy, then publish atomically."""
|
|
self._validate_storage(storage_mib)
|
|
metadata = disk.lstat()
|
|
if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1
|
|
or disk.resolve() == self._settings.rootfs_image.resolve()):
|
|
raise UvmError(f"private VM disk is unsafe to resize: {disk}")
|
|
target_bytes = storage_mib * 1024 * 1024
|
|
if metadata.st_size == target_bytes and self._filesystem_size(disk) == target_bytes:
|
|
return # A previous resize may have published its disk before saving metadata.
|
|
|
|
descriptor, name = tempfile.mkstemp(prefix=".rootfs-resize-", suffix=".ext4", dir=disk.parent)
|
|
os.close(descriptor)
|
|
temporary = Path(name)
|
|
try:
|
|
self._copy_sparse(disk, temporary)
|
|
temporary.chmod(0o600)
|
|
self._resize_filesystem(temporary, storage_mib)
|
|
os.replace(temporary, disk)
|
|
directory = os.open(disk.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
except OSError as error:
|
|
raise UvmError(f"could not resize VM disk {disk}: {error}") from error
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
def configure_guest_network(self, disk: Path, guest_ip: str) -> bool:
|
|
"""Specialize an unedited UVM profile on a proven-offline private disk.
|
|
|
|
Lifecycle owns the operation lock and confirms process exit. Stage and
|
|
check a sparse copy so failed migrations retain the original disk. A
|
|
completed migration, custom image, or guest-edited profile is a no-op.
|
|
"""
|
|
try:
|
|
address = IPv4Address(guest_ip)
|
|
except ValueError as error:
|
|
raise ValidationError("guest network configuration requires a valid IPv4 address") from error
|
|
if address not in self._settings.network or address in {
|
|
self._settings.network.network_address,
|
|
self._settings.network.broadcast_address,
|
|
self._settings.gateway,
|
|
}:
|
|
raise ValidationError("guest network configuration requires a usable IP inside the UVM subnet")
|
|
|
|
try:
|
|
metadata = disk.lstat()
|
|
if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1
|
|
or disk.resolve() == self._settings.rootfs_image.resolve()):
|
|
raise UvmError(f"private VM disk is unsafe to configure: {disk}")
|
|
with tempfile.TemporaryDirectory(prefix="uvm-guest-network-") as name:
|
|
files = Path(name)
|
|
if self._unconfigured_guest_network(disk, files) is None:
|
|
return False
|
|
|
|
descriptor, candidate = tempfile.mkstemp(prefix=".rootfs-network-", suffix=".ext4", dir=disk.parent)
|
|
os.close(descriptor)
|
|
temporary = Path(candidate)
|
|
try:
|
|
self._copy_sparse(disk, temporary)
|
|
temporary.chmod(0o600)
|
|
checked = self._runner.run(("e2fsck", "-f", "-p", temporary), check=False, capture=True)
|
|
if checked.returncode not in (0, 1):
|
|
detail = checked.stderr.strip() or checked.stdout.strip()
|
|
raise UvmError(f"filesystem check failed before guest network configuration: {detail}")
|
|
# Journal replay may reveal guest edits not visible to the
|
|
# initial debugfs read. Recheck ownership on the recovered copy.
|
|
profile = self._unconfigured_guest_network(temporary, files)
|
|
if profile is None:
|
|
return False
|
|
profile.path.write_text(
|
|
GUEST_NETWORK_TEMPLATE
|
|
+ f"Address={address}/{self._settings.network.prefixlen}\n"
|
|
+ f"Gateway={self._settings.gateway}\n",
|
|
encoding="utf-8",
|
|
)
|
|
self._write_guest_file(temporary, "/etc/systemd/network/10-uvm.network", profile)
|
|
self._runner.run(("e2fsck", "-f", "-n", temporary), capture=True)
|
|
with temporary.open("rb") as image:
|
|
os.fsync(image.fileno())
|
|
os.replace(temporary, disk)
|
|
directory = os.open(disk.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
return True
|
|
except (OSError, UnicodeError) as error:
|
|
raise UvmError(f"could not configure guest networking: {error}") from error
|
|
|
|
def _unconfigured_guest_network(self, disk: Path, temporary: Path) -> _GuestFile | None:
|
|
marker = self._read_guest_file(
|
|
disk, "/etc/uvm-image", temporary / "marker", required=False, regular_only=True,
|
|
)
|
|
if marker is None or marker.path.read_text(encoding="utf-8") not in {
|
|
"uvm-ubuntu-jammy-v1\n", "uvm-ubuntu-jammy-v2\n", "uvm-ubuntu-jammy-v3\n",
|
|
"uvm-ubuntu-jammy-v4\n", "uvm-ubuntu-jammy-v5\n",
|
|
}:
|
|
return None
|
|
profile = self._read_guest_file(
|
|
disk, "/etc/systemd/network/10-uvm.network", temporary / "network", required=False, regular_only=True,
|
|
)
|
|
if (profile is None or (profile.uid, profile.gid, profile.mode, profile.links) != (0, 0, 0o644, 1)
|
|
or profile.has_extended_attributes):
|
|
return None
|
|
if profile.path.read_text(encoding="utf-8") not in {
|
|
_LEGACY_GUEST_NETWORK_TEMPLATE, GUEST_NETWORK_TEMPLATE,
|
|
}:
|
|
return None
|
|
return profile
|
|
|
|
def provision_guest_agent(self, disk: Path, vm_id: str, username: str, *, install: bool = False) -> bool:
|
|
"""Ensure every guest has its vsock agent before booting its private disk.
|
|
|
|
The lifecycle lock and confirmed VMM exit are required. A checked sparse
|
|
copy preserves the original disk on failure and makes retries atomic.
|
|
Return False only when the managed files and service are already current.
|
|
An explicit install request always stages and rechecks the installation.
|
|
"""
|
|
parse_username(username)
|
|
if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,80}", vm_id) is None:
|
|
raise ValidationError("invalid VM ID for guest agent configuration")
|
|
try:
|
|
metadata = disk.lstat()
|
|
if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1
|
|
or disk.resolve() == self._settings.rootfs_image.resolve()):
|
|
raise UvmError("private VM disk is unsafe for guest agent installation")
|
|
with tempfile.TemporaryDirectory(prefix="uvm-guest-agent-") as name:
|
|
files = Path(name)
|
|
if not install and self._guest_agent_current(disk, files, vm_id, username):
|
|
return False
|
|
descriptor, candidate = tempfile.mkstemp(prefix=".rootfs-agent-", suffix=".ext4", dir=disk.parent)
|
|
os.close(descriptor)
|
|
temporary = Path(candidate)
|
|
try:
|
|
self._copy_sparse(disk, temporary)
|
|
temporary.chmod(0o600)
|
|
checked = self._runner.run(("e2fsck", "-f", "-p", temporary), check=False, capture=True)
|
|
if checked.returncode not in (0, 1):
|
|
raise UvmError("filesystem check failed before guest agent installation")
|
|
# All ownership/content checks happen AFTER journal replay.
|
|
self._install_agent_files(temporary, files, vm_id, username)
|
|
self._runner.run(("e2fsck", "-f", "-n", temporary), capture=True)
|
|
with temporary.open("rb") as image:
|
|
os.fsync(image.fileno())
|
|
os.replace(temporary, disk)
|
|
directory = os.open(disk.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
return True
|
|
except (OSError, UnicodeError) as error:
|
|
raise UvmError(f"could not provision guest agent: {error}") from error
|
|
|
|
@staticmethod
|
|
def _agent_payloads(vm_id: str, username: str) -> dict[str, bytes]:
|
|
payloads = guest_agent_files()
|
|
payloads["/etc/uvm-agent.json"] = (json.dumps({"vm_id": vm_id, "username": username}) + "\n").encode("utf-8")
|
|
return payloads
|
|
|
|
def _guest_agent_current(self, disk: Path, files: Path, vm_id: str, username: str) -> bool:
|
|
"""Read-only fast path; changed installations are checked again after journal replay."""
|
|
payloads = self._agent_payloads(vm_id, username)
|
|
for index, (path, content) in enumerate(payloads.items()):
|
|
existing = self._agent_file(disk, path, files / f"current-{index}")
|
|
mode = 0o600 if path.endswith(".json") else 0o644
|
|
if (existing is None or (existing.mode, existing.gid) != (mode, 0)
|
|
or existing.path.read_bytes() != content):
|
|
return False
|
|
manifest = self._agent_file(disk, "/etc/uvm-agent-files.json", files / "current-manifest")
|
|
if manifest is None or (manifest.mode, manifest.gid) != (0o600, 0):
|
|
return False
|
|
try:
|
|
hashes = json.loads(manifest.path.read_bytes())
|
|
except (ValueError, UnicodeError):
|
|
return False
|
|
if hashes != {path: hashlib.sha256(content).hexdigest() for path, content in payloads.items()}:
|
|
return False
|
|
service = self._runner.run((
|
|
"debugfs", "-R", "stat /etc/systemd/system/multi-user.target.wants/uvm-agent.service", disk,
|
|
), capture=True)
|
|
return (re.search(r"Type:\s+symlink\b", service.stdout) is not None
|
|
and 'Fast link dest: "/etc/systemd/system/uvm-agent.service"' in service.stdout)
|
|
|
|
def _install_agent_files(self, disk: Path, files: Path, vm_id: str, username: str) -> None:
|
|
for binary in ("/usr/lib/systemd/systemd", "/usr/bin/apt-get"):
|
|
result = self._runner.run(("debugfs", "-R", f"stat {binary}", disk), capture=True)
|
|
if re.search(r"Type:\s+regular\b", result.stdout) is None:
|
|
raise UvmError(
|
|
"automatic vsock guest-agent setup requires a systemd/apt image (Ubuntu 22.04 or newer); "
|
|
"use the default Ubuntu rootfs or a compatible custom image"
|
|
)
|
|
for directory in ("/etc", "/usr", "/usr/lib", "/usr/lib/uvm", "/etc/systemd",
|
|
"/etc/systemd/system", "/etc/systemd/system/multi-user.target.wants"):
|
|
self._agent_directory(disk, directory)
|
|
payloads = self._agent_payloads(vm_id, username)
|
|
manifest_path = "/etc/uvm-agent-files.json"
|
|
previous = self._agent_file(disk, manifest_path, files / "manifest")
|
|
old_hashes = {}
|
|
if previous is not None:
|
|
try:
|
|
old_hashes = json.loads(previous.path.read_bytes())
|
|
except (ValueError, UnicodeError) as error:
|
|
raise UvmError("invalid guest agent ownership manifest") from error
|
|
if (not isinstance(old_hashes, dict) or set(old_hashes) - set(payloads)
|
|
or any(not isinstance(value, str) or re.fullmatch(r"[0-9a-f]{64}", value) is None
|
|
for value in old_hashes.values())):
|
|
raise UvmError("invalid guest agent ownership manifest")
|
|
hashes = {path: hashlib.sha256(content).hexdigest() for path, content in payloads.items()}
|
|
for index, (path, content) in enumerate(payloads.items()):
|
|
existing = self._agent_file(disk, path, files / f"existing-{index}")
|
|
if existing is not None:
|
|
digest = hashlib.sha256(existing.path.read_bytes()).hexdigest()
|
|
allowed = {hashes[path], old_hashes.get(path)}
|
|
if not old_hashes and path == "/usr/lib/uvm/agent.py":
|
|
allowed.add(_V4_TEMPLATE_AGENT_SHA256)
|
|
if digest not in allowed:
|
|
raise UvmError(f"refusing to overwrite a customized guest agent file: {path}")
|
|
source = files / f"payload-{index}"
|
|
source.write_bytes(content)
|
|
self._write_guest_file(disk, path, _GuestFile(source, 0o600 if path.endswith(".json") else 0o644, 0, 0))
|
|
source = files / "new-manifest"
|
|
source.write_text(json.dumps(hashes, sort_keys=True) + "\n", encoding="utf-8")
|
|
self._write_guest_file(disk, manifest_path, _GuestFile(source, 0o600, 0, 0))
|
|
link = "/etc/systemd/system/multi-user.target.wants/uvm-agent.service"
|
|
target = "/etc/systemd/system/uvm-agent.service"
|
|
result = self._runner.run(("debugfs", "-R", f"stat {link}", disk), capture=True)
|
|
if "file not found" in (result.stdout + result.stderr).lower():
|
|
self._runner.run(("debugfs", "-w", "-R", f"symlink {link} {target}", disk), capture=True)
|
|
result = self._runner.run(("debugfs", "-R", f"stat {link}", disk), capture=True)
|
|
if (re.search(r"Type:\s+symlink\b", result.stdout) is None
|
|
or f'Fast link dest: "{target}"' not in result.stdout):
|
|
raise UvmError("could not enable guest agent without replacing a customized service link")
|
|
|
|
def _agent_directory(self, disk: Path, path: str) -> None:
|
|
result = self._runner.run(("debugfs", "-R", f"stat {path}", disk), capture=True)
|
|
if "file not found" in (result.stdout + result.stderr).lower():
|
|
self._create_guest_directory(disk, path, 0o755, 0, 0)
|
|
return
|
|
mode, owner = _MODE_PATTERN.search(result.stdout), _OWNER_PATTERN.search(result.stdout)
|
|
if (re.search(r"Type:\s+directory\b", result.stdout) is None or mode is None or owner is None
|
|
or int(owner.group(1)) != 0 or int(mode.group(1), 8) & 0o022):
|
|
raise UvmError(f"unsafe guest agent directory: {path}")
|
|
|
|
def _agent_file(self, disk: Path, path: str, destination: Path) -> _GuestFile | None:
|
|
result = self._runner.run(("debugfs", "-R", f"stat {path}", disk), capture=True)
|
|
if "file not found" in (result.stdout + result.stderr).lower():
|
|
return None
|
|
if re.search(r"Type:\s+regular\b", result.stdout) is None:
|
|
raise UvmError(f"unsafe guest agent file: {path}")
|
|
source = self._read_guest_file(disk, path, destination)
|
|
if (source.uid != 0 or source.mode & 0o022 or source.links != 1 or source.has_extended_attributes):
|
|
raise UvmError(f"unsafe guest agent file: {path}")
|
|
return source
|
|
|
|
@staticmethod
|
|
def _validate_storage(storage_mib: int) -> None:
|
|
if type(storage_mib) is not int or not MIN_STORAGE_MIB <= storage_mib <= MAX_STORAGE_MIB:
|
|
raise ValidationError("storage must be a supported whole-MiB capacity of at least 30 GiB")
|
|
|
|
def _copy_sparse(self, source: Path, destination: Path) -> None:
|
|
self._runner.run(("cp", "--reflink=auto", "--sparse=always", "--", source, destination), capture=True)
|
|
|
|
def _filesystem_size(self, disk: Path) -> int:
|
|
result = self._runner.run(("env", "LC_ALL=C", "dumpe2fs", "-h", disk), capture=True)
|
|
blocks = re.search(r"^Block count:\s+(\d+)\s*$", result.stdout, re.MULTILINE)
|
|
block_size = re.search(r"^Block size:\s+(\d+)\s*$", result.stdout, re.MULTILINE)
|
|
if blocks is None or block_size is None:
|
|
raise UvmError(f"could not determine ext4 filesystem capacity: {disk}")
|
|
return int(blocks.group(1)) * int(block_size.group(1))
|
|
|
|
def _resize_filesystem(self, disk: Path, storage_mib: int) -> None:
|
|
"""Grow the file before ext4; shrink ext4 before truncating the file."""
|
|
checked = self._runner.run(("e2fsck", "-f", "-p", disk), check=False, capture=True)
|
|
if checked.returncode not in (0, 1):
|
|
detail = checked.stderr.strip() or checked.stdout.strip()
|
|
raise UvmError(f"filesystem check failed before resize ({checked.returncode}): {detail}")
|
|
target_bytes = storage_mib * 1024 * 1024
|
|
if disk.stat().st_size < target_bytes:
|
|
with disk.open("r+b") as image:
|
|
image.truncate(target_bytes)
|
|
self._runner.run(("resize2fs", disk, f"{storage_mib}M"), capture=True)
|
|
if self._filesystem_size(disk) != target_bytes:
|
|
raise UvmError("resized ext4 filesystem does not match the requested capacity")
|
|
with disk.open("r+b") as image:
|
|
image.truncate(target_bytes)
|
|
image.flush()
|
|
os.fsync(image.fileno())
|
|
|
|
def provision_credentials(self, disk: Path, username: str, password: str) -> None:
|
|
"""Create a missing guest account and set its password in an offline ext4 disk."""
|
|
|
|
parse_username(username)
|
|
password_hash = self._password_hash(password)
|
|
try:
|
|
with tempfile.TemporaryDirectory(prefix="uvm-credentials-") as temporary_name:
|
|
temporary = Path(temporary_name)
|
|
passwd = self._read_guest_file(disk, "/etc/passwd", temporary / "passwd")
|
|
shadow = self._read_guest_file(disk, "/etc/shadow", temporary / "shadow")
|
|
sshd_config = self._read_guest_file(
|
|
disk,
|
|
"/etc/ssh/sshd_config",
|
|
temporary / "sshd_config",
|
|
)
|
|
authorized_keys = self._read_guest_file(
|
|
disk,
|
|
"/root/.ssh/authorized_keys",
|
|
temporary / "authorized_keys",
|
|
required=False,
|
|
)
|
|
assert passwd is not None
|
|
assert shadow is not None
|
|
assert sshd_config is not None
|
|
|
|
account_names = {
|
|
line.split(":", 1)[0]
|
|
for line in passwd.path.read_text(encoding="utf-8").splitlines()
|
|
if ":" in line
|
|
}
|
|
if username not in account_names:
|
|
self._create_guest_user(disk, username, passwd, shadow, temporary)
|
|
|
|
shadow.path.write_text(
|
|
_set_shadow_password(
|
|
shadow.path.read_text(encoding="utf-8"),
|
|
username,
|
|
password_hash,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
shadow.path.chmod(shadow.mode)
|
|
if username != "root":
|
|
self._provision_guest_sudo(disk, username, temporary)
|
|
|
|
sshd_config.path.write_text(
|
|
_enable_ssh_password_authentication(
|
|
sshd_config.path.read_text(encoding="utf-8"),
|
|
username,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
sshd_config.path.chmod(sshd_config.mode)
|
|
|
|
self._write_guest_file(disk, "/etc/shadow", shadow)
|
|
self._write_guest_file(disk, "/etc/ssh/sshd_config", sshd_config)
|
|
self._replace_guest_host_keys(disk, temporary)
|
|
if authorized_keys is not None:
|
|
self._remove_firecracker_demo_key(disk, authorized_keys)
|
|
except (OSError, UnicodeError) as error:
|
|
raise UvmError(f"could not provision guest credentials: {error}") from error
|
|
|
|
def _provision_guest_sudo(self, disk: Path, username: str, temporary: Path) -> None:
|
|
sudoers = self._read_guest_file(disk, "/etc/sudoers", temporary / "sudoers", required=False)
|
|
sudo = self._runner.run(("debugfs", "-R", "stat /usr/bin/sudo", disk), capture=True)
|
|
mode = _MODE_PATTERN.search(sudo.stdout)
|
|
owner = _OWNER_PATTERN.search(sudo.stdout)
|
|
if (
|
|
sudoers is None
|
|
or sudoers.uid != 0 or sudoers.mode & 0o022
|
|
or re.search(r"Type:\s+regular\b", sudo.stdout) is None
|
|
or mode is None or not int(mode.group(1), 8) & 0o4000
|
|
or not int(mode.group(1), 8) & 0o111
|
|
or owner is None or int(owner.group(1)) != 0
|
|
):
|
|
raise UvmError(
|
|
"guest image needs setuid-root sudo and /etc/sudoers for a custom administrator; "
|
|
"run uvm install --force to build the complete Ubuntu image"
|
|
)
|
|
contents = sudoers.path.read_text(encoding="utf-8").rstrip()
|
|
sudoers.path.write_text(
|
|
contents + "\n\n# Managed by uvm: guest administrator\n"
|
|
+ f"Defaults:{username} !rootpw, !runaspw, !targetpw\n"
|
|
+ f"{username} ALL=(ALL:ALL) PASSWD: ALL\n",
|
|
encoding="utf-8",
|
|
)
|
|
sudoers.path.chmod(sudoers.mode)
|
|
self._write_guest_file(disk, "/etc/sudoers", sudoers)
|
|
|
|
def _create_guest_user(
|
|
self,
|
|
disk: Path,
|
|
username: str,
|
|
passwd: _GuestFile,
|
|
shadow: _GuestFile,
|
|
temporary: Path,
|
|
) -> None:
|
|
if username == "root":
|
|
raise UvmError("guest rootfs is missing the root account")
|
|
group = self._read_guest_file(disk, "/etc/group", temporary / "group")
|
|
gshadow = self._read_guest_file(
|
|
disk, "/etc/gshadow", temporary / "gshadow", required=False,
|
|
)
|
|
assert group is not None
|
|
passwd_entries = [
|
|
line.split(":")
|
|
for line in passwd.path.read_text(encoding="utf-8").splitlines()
|
|
if ":" in line
|
|
]
|
|
group_entries = [
|
|
line.split(":")
|
|
for line in group.path.read_text(encoding="utf-8").splitlines()
|
|
if ":" in line
|
|
]
|
|
try:
|
|
used_uids = {int(fields[2]) for fields in passwd_entries}
|
|
used_gids = {int(fields[3]) for fields in passwd_entries}
|
|
used_gids.update(int(fields[2]) for fields in group_entries)
|
|
except (IndexError, ValueError) as error:
|
|
raise UvmError("guest rootfs has invalid passwd or group entries") from error
|
|
|
|
uid = _next_guest_id(used_uids)
|
|
existing_group = next((fields for fields in group_entries if fields[0] == username), None)
|
|
if existing_group is not None:
|
|
gid = int(existing_group[2])
|
|
else:
|
|
gid = uid if uid not in used_gids else _next_guest_id(used_gids)
|
|
|
|
home = f"/home/{username}"
|
|
updates = [
|
|
("/etc/passwd", passwd, f"{username}:x:{uid}:{gid}::{home}:/bin/sh"),
|
|
("/etc/shadow", shadow, f"{username}:!:{int(time.time() // 86400)}:0:99999:7:::"),
|
|
]
|
|
if existing_group is None:
|
|
updates.append(("/etc/group", group, f"{username}:x:{gid}:"))
|
|
if gshadow is not None:
|
|
updates.append(("/etc/gshadow", gshadow, f"{username}:!::"))
|
|
for guest_path, source, entry in updates:
|
|
contents = source.path.read_text(encoding="utf-8")
|
|
if any(line.split(":", 1)[0] == username for line in contents.splitlines()):
|
|
raise UvmError(f"guest rootfs already has an entry for {username} in {guest_path}")
|
|
separator = "" if not contents or contents.endswith("\n") else "\n"
|
|
source.path.write_text(f"{contents}{separator}{entry}\n", encoding="utf-8")
|
|
source.path.chmod(source.mode)
|
|
|
|
self._create_guest_directory(disk, "/home", 0o755, 0, 0, exist_ok=True)
|
|
self._create_guest_directory(disk, home, 0o700, uid, gid)
|
|
for guest_path, source, _entry in updates:
|
|
if guest_path != "/etc/shadow": # Published with its password below.
|
|
self._write_guest_file(disk, guest_path, source)
|
|
|
|
def _create_guest_directory(
|
|
self,
|
|
disk: Path,
|
|
guest_path: str,
|
|
mode: int,
|
|
uid: int,
|
|
gid: int,
|
|
*,
|
|
exist_ok: bool = False,
|
|
) -> None:
|
|
command = ("debugfs", "-R", f"stat {guest_path}", disk)
|
|
existing = self._runner.run(command, check=False, capture=True)
|
|
if exist_ok and re.search(r"Type:\s+directory\b", existing.stdout):
|
|
return
|
|
if "file not found" not in f"{existing.stdout}\n{existing.stderr}".lower():
|
|
raise UvmError(f"cannot create guest rootfs directory: {guest_path}")
|
|
|
|
self._runner.run(("debugfs", "-w", "-R", f"mkdir {guest_path}", disk), capture=True)
|
|
for field, value in (
|
|
("mode", f"0{0o40000 | mode:o}"),
|
|
("uid", str(uid)),
|
|
("gid", str(gid)),
|
|
):
|
|
self._runner.run(
|
|
("debugfs", "-w", "-R", f"set_inode_field {guest_path} {field} {value}", disk),
|
|
capture=True,
|
|
)
|
|
|
|
# debugfs can report command failures while still returning exit status zero.
|
|
created = self._runner.run(command, capture=True)
|
|
mode_match = _MODE_PATTERN.search(created.stdout)
|
|
owner_match = _OWNER_PATTERN.search(created.stdout)
|
|
if (
|
|
re.search(r"Type:\s+directory\b", created.stdout) is None
|
|
or mode_match is None
|
|
or owner_match is None
|
|
or int(mode_match.group(1), 8) != mode
|
|
or (int(owner_match.group(1)), int(owner_match.group(2))) != (uid, gid)
|
|
):
|
|
raise UvmError(f"could not verify guest rootfs directory: {guest_path}")
|
|
|
|
def _password_hash(self, password: str) -> str:
|
|
result = self._runner.run(
|
|
("openssl", "passwd", "-6", "-stdin"),
|
|
capture=True,
|
|
input_text=f"{password}\n",
|
|
sensitive=True,
|
|
)
|
|
password_hash = result.stdout.strip()
|
|
if not password_hash.startswith("$6$") or any(
|
|
character in password_hash for character in ("\n", "\r", ":")
|
|
):
|
|
raise UvmError("openssl returned an invalid guest password hash")
|
|
return password_hash
|
|
|
|
def _read_guest_file(
|
|
self,
|
|
disk: Path,
|
|
guest_path: str,
|
|
destination: Path,
|
|
*,
|
|
required: bool = True,
|
|
regular_only: bool = False,
|
|
) -> _GuestFile | None:
|
|
stat_result = self._runner.run(
|
|
("debugfs", "-R", f"stat {guest_path}", disk),
|
|
check=False,
|
|
capture=True,
|
|
)
|
|
file_type = re.search(r"Type:\s+(\S+)", stat_result.stdout)
|
|
if regular_only and file_type is not None and file_type.group(1) != "regular":
|
|
return None
|
|
mode_match = _MODE_PATTERN.search(stat_result.stdout)
|
|
owner_match = _OWNER_PATTERN.search(stat_result.stdout)
|
|
flags_match = _FLAGS_PATTERN.search(stat_result.stdout)
|
|
links_match = _LINKS_PATTERN.search(stat_result.stdout)
|
|
if (
|
|
mode_match is None
|
|
or owner_match is None
|
|
or flags_match is None
|
|
or links_match is None
|
|
):
|
|
output = f"{stat_result.stdout}\n{stat_result.stderr}".lower()
|
|
if not required and "file not found" in output:
|
|
return None
|
|
if required and "file not found" in output:
|
|
raise UvmError(f"guest rootfs is missing required file: {guest_path}")
|
|
raise UvmError(f"could not inspect guest rootfs file: {guest_path}")
|
|
|
|
self._runner.run(
|
|
("debugfs", "-R", f"dump {guest_path} {destination}", disk),
|
|
capture=True,
|
|
)
|
|
if not destination.is_file():
|
|
raise UvmError(f"could not read guest rootfs file: {guest_path}")
|
|
attributes = self._runner.run(
|
|
("debugfs", "-R", f"ea_list {guest_path}", disk),
|
|
check=False,
|
|
capture=True,
|
|
)
|
|
diagnostics = [
|
|
line
|
|
for line in attributes.stderr.splitlines()
|
|
if line and not line.startswith("debugfs ")
|
|
]
|
|
if attributes.returncode != 0 or diagnostics:
|
|
raise UvmError(f"could not inspect guest rootfs metadata: {guest_path}")
|
|
return _GuestFile(
|
|
path=destination,
|
|
mode=int(mode_match.group(1), 8),
|
|
uid=int(owner_match.group(1)),
|
|
gid=int(owner_match.group(2)),
|
|
flags=int(flags_match.group(1), 16),
|
|
links=int(links_match.group(1)),
|
|
has_extended_attributes=bool(attributes.stdout.strip()),
|
|
)
|
|
|
|
def _write_guest_file(self, disk: Path, guest_path: str, source: _GuestFile) -> None:
|
|
if source.links != 1:
|
|
raise UvmError(f"cannot safely replace hard-linked guest file: {guest_path}")
|
|
if source.has_extended_attributes:
|
|
raise UvmError(
|
|
f"cannot safely replace guest file with extended attributes: {guest_path}"
|
|
)
|
|
self._runner.run(("debugfs", "-w", "-R", f"rm {guest_path}", disk), capture=True)
|
|
self._runner.run(
|
|
("debugfs", "-w", "-R", f"write {source.path} {guest_path}", disk),
|
|
capture=True,
|
|
)
|
|
for field, value in (
|
|
("mode", f"0{0o100000 | source.mode:o}"),
|
|
("uid", str(source.uid)),
|
|
("gid", str(source.gid)),
|
|
):
|
|
self._runner.run(
|
|
(
|
|
"debugfs",
|
|
"-w",
|
|
"-R",
|
|
f"set_inode_field {guest_path} {field} {value}",
|
|
disk,
|
|
),
|
|
capture=True,
|
|
)
|
|
if source.flags is not None:
|
|
self._runner.run(
|
|
(
|
|
"debugfs",
|
|
"-w",
|
|
"-R",
|
|
f"set_inode_field {guest_path} flags 0x{source.flags:x}",
|
|
disk,
|
|
),
|
|
capture=True,
|
|
)
|
|
|
|
verification = source.path.with_name(f"{source.path.name}.verify")
|
|
written = self._read_guest_file(disk, guest_path, verification)
|
|
assert written is not None
|
|
if (
|
|
verification.read_bytes() != source.path.read_bytes()
|
|
or written.mode != source.mode
|
|
or written.uid != source.uid
|
|
or written.gid != source.gid
|
|
or (source.flags is not None and written.flags != source.flags)
|
|
):
|
|
raise UvmError(f"could not verify updated guest rootfs file: {guest_path}")
|
|
|
|
def _replace_guest_host_keys(self, disk: Path, temporary: Path) -> None:
|
|
for key_type, bits in _HOST_KEY_TYPES:
|
|
name = f"ssh_host_{key_type}_key"
|
|
private_path = temporary / name
|
|
command: list[str | Path] = [
|
|
"ssh-keygen",
|
|
"-q",
|
|
"-t",
|
|
key_type,
|
|
"-N",
|
|
"",
|
|
"-C",
|
|
"",
|
|
"-f",
|
|
private_path,
|
|
]
|
|
if bits is not None:
|
|
command[4:4] = ["-b", bits]
|
|
self._runner.run(command, capture=True)
|
|
|
|
for suffix, default_mode in (("", 0o600), (".pub", 0o644)):
|
|
guest_path = f"/etc/ssh/{name}{suffix}"
|
|
existing = self._read_guest_file(
|
|
disk,
|
|
guest_path,
|
|
temporary / f"existing-{name}{suffix}",
|
|
required=False,
|
|
)
|
|
generated = private_path.with_name(f"{name}{suffix}")
|
|
source = _GuestFile(
|
|
path=generated,
|
|
mode=existing.mode if existing is not None else default_mode,
|
|
uid=existing.uid if existing is not None else 0,
|
|
gid=existing.gid if existing is not None else 0,
|
|
)
|
|
self._write_guest_file(disk, guest_path, source)
|
|
|
|
def _remove_firecracker_demo_key(self, disk: Path, authorized_keys: _GuestFile) -> None:
|
|
contents = authorized_keys.path.read_text(encoding="utf-8")
|
|
updated = _without_firecracker_demo_key(contents)
|
|
if updated == contents:
|
|
return
|
|
if updated is not None:
|
|
authorized_keys.path.write_text(updated, encoding="utf-8")
|
|
authorized_keys.path.chmod(authorized_keys.mode)
|
|
self._write_guest_file(
|
|
disk,
|
|
"/root/.ssh/authorized_keys",
|
|
authorized_keys,
|
|
)
|
|
return
|
|
|
|
self._runner.run(
|
|
("debugfs", "-w", "-R", "rm /root/.ssh/authorized_keys", disk),
|
|
capture=True,
|
|
)
|
|
if self._read_guest_file(
|
|
disk,
|
|
"/root/.ssh/authorized_keys",
|
|
authorized_keys.path.with_name("authorized_keys.verify"),
|
|
required=False,
|
|
) is not None:
|
|
raise UvmError("could not remove the insecure Firecracker demo SSH key")
|
|
|
|
|
|
def _next_guest_id(used: set[int]) -> int:
|
|
for candidate in range(1000, 60001):
|
|
if candidate not in used:
|
|
return candidate
|
|
raise UvmError("guest rootfs has no available regular user/group IDs")
|
|
|
|
|
|
def _set_shadow_password(contents: str, username: str, password_hash: str) -> str:
|
|
lines = contents.splitlines()
|
|
for index, line in enumerate(lines):
|
|
fields = line.split(":")
|
|
if fields[0] == username and len(fields) >= 2:
|
|
fields[1] = password_hash
|
|
lines[index] = ":".join(fields)
|
|
return "\n".join(lines) + "\n"
|
|
raise UvmError(f"guest rootfs has no shadow entry for user: {username}")
|
|
|
|
|
|
def _enable_ssh_password_authentication(contents: str, username: str) -> str:
|
|
managed_directives = {
|
|
"challengeresponseauthentication",
|
|
"hostkey",
|
|
"kbdinteractiveauthentication",
|
|
"passwordauthentication",
|
|
}
|
|
if username == "root":
|
|
managed_directives.add("permitrootlogin")
|
|
global_lines = []
|
|
match_lines = []
|
|
in_match = False
|
|
for line in contents.splitlines():
|
|
stripped = line.lstrip()
|
|
directive = stripped.split(None, 1)[0].lower() if stripped else ""
|
|
if directive == "match":
|
|
in_match = True
|
|
if stripped.startswith("#") or directive not in managed_directives:
|
|
(match_lines if in_match else global_lines).append(line)
|
|
|
|
directives = [
|
|
"HostKey /etc/ssh/ssh_host_rsa_key",
|
|
"HostKey /etc/ssh/ssh_host_ecdsa_key",
|
|
"HostKey /etc/ssh/ssh_host_ed25519_key",
|
|
"PasswordAuthentication no",
|
|
"KbdInteractiveAuthentication no",
|
|
"ChallengeResponseAuthentication no",
|
|
]
|
|
if username == "root":
|
|
directives.append("PermitRootLogin yes")
|
|
# Global-only directives (e.g. UsePAM and Subsystem) must precede every Match block.
|
|
globals_text = "\n".join(global_lines).strip("\n")
|
|
return (
|
|
"# Managed by uvm\n"
|
|
+ "\n".join(directives)
|
|
+ (f"\n{globals_text}" if globals_text else "")
|
|
+ f"\nMatch User {username}\n"
|
|
+ " PasswordAuthentication yes\n"
|
|
+ "Match all\n"
|
|
+ "\n".join(match_lines).rstrip()
|
|
+ "\n"
|
|
)
|
|
|
|
|
|
def _without_firecracker_demo_key(contents: str) -> str | None:
|
|
lines = contents.splitlines()
|
|
retained = [
|
|
line
|
|
for line in lines
|
|
if " ".join(line.split()[:2]) != _FIRECRACKER_DEMO_PUBLIC_KEY
|
|
]
|
|
if len(retained) == len(lines):
|
|
return contents
|
|
return "\n".join(retained) + "\n" if retained else None
|