from __future__ import annotations import configparser import hashlib import shutil import tempfile import unittest from dataclasses import replace from ipaddress import IPv4Address, IPv4Network from pathlib import Path from unittest.mock import Mock, patch from uvm.config import Settings from uvm.errors import UvmError, ValidationError from uvm.images import GUEST_NETWORK_TEMPLATE, ImageStore from uvm.system import CommandResult, CommandRunner PROFILE = "/etc/systemd/network/10-uvm.network" # The shipped v1/v2 profile that retained the IP but lost the proto-boot route. LEGACY_PROFILE = ( "[Match]\nName=eth0\n\n[Network]\n" "KeepConfiguration=static\nDHCP=no\nLinkLocalAddressing=no\nIPv6AcceptRA=no\n" ) class GuestNetworkValidationTests(unittest.TestCase): def test_invalid_guest_addresses_fail_before_disk_access(self) -> None: runner = Mock(spec=CommandRunner) store = ImageStore(Settings(), runner) for address in ("invalid", "::1", "192.0.2.2", "10.42.0.0", "10.42.0.255", "10.42.0.1"): with self.subTest(address=address), self.assertRaises(ValidationError): store.configure_guest_network(Path("/nonexistent/rootfs.ext4"), address) runner.run.assert_not_called() def test_shared_or_linked_disks_are_rejected_before_guest_commands(self) -> None: with tempfile.TemporaryDirectory() as name: settings = Settings(base=Path(name)) settings.images_dir.mkdir() settings.rootfs_image.write_bytes(b"shared template") disk = settings.base / "rootfs.ext4" disk.write_bytes(b"private disk") linked = settings.base / "linked.ext4" linked.hardlink_to(disk) symlink = settings.base / "symlink.ext4" symlink.symlink_to(disk) runner = Mock(spec=CommandRunner) store = ImageStore(settings, runner) for candidate in (settings.rootfs_image, disk, linked, symlink): with self.subTest(candidate=candidate), self.assertRaisesRegex(UvmError, "unsafe"): store.configure_guest_network(candidate, "10.42.0.2") runner.run.assert_not_called() @unittest.skipUnless(all(shutil.which(tool) for tool in ("mkfs.ext4", "debugfs", "e2fsck", "cp")), "e2fsprogs and coreutils are required for offline guest networking checks") class OfflineGuestNetworkTests(unittest.TestCase): def setUp(self) -> None: temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.base = Path(temporary.name) self.settings = Settings(base=self.base / "uvm") self.settings.images_dir.mkdir(parents=True) self.disk = self.base / "VM with spaces" / "rootfs.ext4" self.disk.parent.mkdir() self.runner = CommandRunner(emit=None) self.store = ImageStore(self.settings, self.runner) self.retained = { "/etc/shadow": "adm:$6$original-password-hash:1:0:99999:7:::\n", "/etc/ssh/ssh_host_ed25519_key": "original private host key\n", "/home/adm/data": "retained guest application data\n", } def make_disk(self, *, marker="uvm-ubuntu-jammy-v2\n", profile=LEGACY_PROFILE, symlink=False) -> None: root = self.base / "tree" if root.exists(): shutil.rmtree(root) files = dict(self.retained) if marker is not None: files["/etc/uvm-image"] = marker if profile is not None: files["/etc/alternate.network" if symlink else PROFILE] = profile for path, contents in files.items(): local = root / path.lstrip("/") local.parent.mkdir(parents=True, exist_ok=True) local.write_text(contents, encoding="utf-8") local.chmod(0o644) if symlink: local = root / PROFILE.lstrip("/") local.parent.mkdir(parents=True, exist_ok=True) local.symlink_to("/etc/alternate.network") with self.settings.rootfs_image.open("wb") as image: image.truncate(32 * 1024 * 1024) self.runner.run(("mkfs.ext4", "-q", "-F", "-b", "4096", "-d", root, self.settings.rootfs_image), capture=True) if profile is not None: for field in ("uid", "gid"): self.runner.run(("debugfs", "-w", "-R", f"set_inode_field {PROFILE} {field} 0", self.settings.rootfs_image), capture=True) self.runner.run(("cp", "--sparse=always", self.settings.rootfs_image, self.disk), capture=True) self.disk.chmod(0o600) def guest_file(self, path: str) -> str: return self.runner.run(("debugfs", "-R", f"cat {path}", self.disk), capture=True).stdout def snapshot(self) -> tuple[int, bytes]: return self.disk.stat().st_ino, hashlib.sha256(self.disk.read_bytes()).digest() def test_templates_gain_an_explicit_address_gateway_and_boot_route_preservation(self) -> None: for marker, profile in (("uvm-ubuntu-jammy-v1\n", LEGACY_PROFILE), ("uvm-ubuntu-jammy-v2\n", LEGACY_PROFILE), ("uvm-ubuntu-jammy-v3\n", GUEST_NETWORK_TEMPLATE)): with self.subTest(marker=marker): self.make_disk(marker=marker, profile=profile) original_template = hashlib.sha256(self.settings.rootfs_image.read_bytes()).digest() self.assertTrue(self.store.configure_guest_network(self.disk, "10.42.0.2")) config = configparser.ConfigParser() config.read_string(self.guest_file(PROFILE)) self.assertEqual(config["Match"]["Name"], "eth0") self.assertEqual(config["Network"]["Address"], "10.42.0.2/24") self.assertEqual(config["Network"]["Gateway"], "10.42.0.1") self.assertTrue(config["Network"].getboolean("KeepConfiguration")) self.assertFalse(config["Network"].getboolean("DHCP")) for path, contents in self.retained.items(): self.assertEqual(self.guest_file(path), contents) self.assertEqual(hashlib.sha256(self.settings.rootfs_image.read_bytes()).digest(), original_template) self.assertEqual(self.disk.stat().st_size, 32 * 1024 * 1024) self.assertEqual(self.disk.stat().st_mode & 0o777, 0o600) self.runner.run(("e2fsck", "-f", "-n", self.disk), capture=True) before = self.snapshot() with patch.object(self.store, "_copy_sparse") as copy: self.assertFalse(self.store.configure_guest_network(self.disk, "10.42.0.2")) copy.assert_not_called() self.assertEqual(self.snapshot(), before) def test_custom_subnets_and_gateways_are_used_in_the_guest_profile(self) -> None: settings = replace(self.settings, network=IPv4Network("10.50.0.0/16"), gateway=IPv4Address("10.50.1.1")) self.make_disk() self.assertTrue(ImageStore(settings, self.runner).configure_guest_network(self.disk, "10.50.8.9")) config = configparser.ConfigParser() config.read_string(self.guest_file(PROFILE)) self.assertEqual(config["Network"]["Address"], "10.50.8.9/16") self.assertEqual(config["Network"]["Gateway"], "10.50.1.1") def test_custom_images_missing_profiles_and_guest_edits_are_left_intact(self) -> None: for marker, profile, symlink in ( (None, LEGACY_PROFILE, False), ("custom-image\n", LEGACY_PROFILE, False), ("uvm-ubuntu-jammy-v2\n", None, False), ("uvm-ubuntu-jammy-v2\n", LEGACY_PROFILE + "# guest customization\n", False), ("uvm-ubuntu-jammy-v2\n", LEGACY_PROFILE + "Gateway=10.42.0.9\n", False), ("uvm-ubuntu-jammy-v2\n", LEGACY_PROFILE, True), ): with self.subTest(marker=marker, profile=profile, symlink=symlink): self.make_disk(marker=marker, profile=profile, symlink=symlink) before = self.snapshot() with patch.object(self.store, "_copy_sparse") as copy: self.assertFalse(self.store.configure_guest_network(self.disk, "10.42.0.2")) copy.assert_not_called() self.assertEqual(self.snapshot(), before) def test_a_failed_or_interrupted_profile_write_keeps_the_original_disk(self) -> None: for error in (UvmError("profile readback failed"), KeyboardInterrupt()): with self.subTest(error=type(error).__name__): self.make_disk() before = self.snapshot() write = self.store._write_guest_file def fail(candidate, path, contents): self.assertNotEqual(candidate, self.disk) write(candidate, path, contents) raise error with patch.object(self.store, "_write_guest_file", side_effect=fail), self.assertRaises(type(error)): self.store.configure_guest_network(self.disk, "10.42.0.2") self.assertEqual(self.snapshot(), before) self.assertEqual(list(self.disk.parent.iterdir()), [self.disk]) def test_filesystem_check_failure_prevents_publication(self) -> None: self.make_disk() before = self.snapshot() run = self.runner.run def check_failure(command, **options): if command[0] == "e2fsck": return CommandResult(tuple(map(str, command)), 4, stderr="unrepaired filesystem errors") return run(command, **options) with patch.object(self.runner, "run", side_effect=check_failure): with self.assertRaisesRegex(UvmError, "unrepaired filesystem errors"): self.store.configure_guest_network(self.disk, "10.42.0.2") self.assertEqual(self.snapshot(), before) self.assertEqual(list(self.disk.parent.iterdir()), [self.disk]) def test_guest_edits_revealed_after_journal_replay_prevent_migration(self) -> None: self.make_disk() before = self.snapshot() edited = self.base / "edited.network" edited.write_text(LEGACY_PROFILE + "Gateway=10.42.0.9\n", encoding="utf-8") run = self.runner.run def replay(command, **options): result = run(command, **options) if command[:3] == ("e2fsck", "-f", "-p"): run(("debugfs", "-w", "-R", f"rm {PROFILE}", command[-1]), capture=True) run(("debugfs", "-w", "-R", f"write {edited} {PROFILE}", command[-1]), capture=True) for field in ("uid", "gid"): run(("debugfs", "-w", "-R", f"set_inode_field {PROFILE} {field} 0", command[-1]), capture=True) return result with patch.object(self.runner, "run", side_effect=replay): self.assertFalse(self.store.configure_guest_network(self.disk, "10.42.0.2")) self.assertEqual(self.snapshot(), before) self.assertEqual(list(self.disk.parent.iterdir()), [self.disk])