diff --git a/bot_bottle/backend/firecracker/setup.py b/bot_bottle/backend/firecracker/setup.py index 0d5d00d..d63dbb5 100644 --- a/bot_bottle/backend/firecracker/setup.py +++ b/bot_bottle/backend/firecracker/setup.py @@ -237,16 +237,18 @@ def _firecracker_binary_ok() -> bool: def _kvm_accessible() -> bool: - """True iff /dev/kvm can be opened and responds to KVM_GET_API_VERSION. + """True iff /dev/kvm can be opened read-write and responds to KVM_GET_API_VERSION. - Uses an ioctl rather than os.access() so that a device with correct - permission bits but a non-functional KVM subsystem is caught here - rather than at VM boot time.""" + VM creation requires write access; opening read-only may satisfy the + ioctl but fails at boot time, so O_RDWR is the permission check.""" if not os.path.exists(util._KVM_DEVICE): return False try: - with open(util._KVM_DEVICE, "rb") as kvm: - fcntl.ioctl(kvm, _KVM_GET_API_VERSION) + fd = os.open(util._KVM_DEVICE, os.O_RDWR | os.O_CLOEXEC) + try: + fcntl.ioctl(fd, _KVM_GET_API_VERSION) + finally: + os.close(fd) return True except OSError: return False @@ -281,6 +283,30 @@ def status() -> int: f"KVM: {util._KVM_DEVICE} not accessible (open/ioctl failed)\n" ) ok = False + kernel = util.kernel_path() + if kernel.is_file(): + sys.stderr.write(f"guest kernel: {kernel}\n") + else: + sys.stderr.write( + f"guest kernel: NOT found at {kernel} " + f"(set BOT_BOTTLE_FC_KERNEL or cache a vmlinux there)\n" + ) + ok = False + dropbear = util.dropbear_path() + if dropbear.is_file(): + sys.stderr.write(f"dropbear: {dropbear}\n") + else: + sys.stderr.write( + f"dropbear: NOT found at {dropbear} " + f"(set BOT_BOTTLE_FC_DROPBEAR or cache a static binary)\n" + ) + ok = False + mke2fs = shutil.which("mke2fs") + if mke2fs is not None: + sys.stderr.write(f"mke2fs: {mke2fs}\n") + else: + sys.stderr.write("mke2fs: NOT found on PATH (install e2fsprogs)\n") + ok = False missing = netpool.missing_taps() total = netpool.pool_size() if missing: diff --git a/tests/unit/test_backend_setup.py b/tests/unit/test_backend_setup.py index 2b9b173..f738032 100644 --- a/tests/unit/test_backend_setup.py +++ b/tests/unit/test_backend_setup.py @@ -357,28 +357,86 @@ class TestFirecrackerKvmCheck(unittest.TestCase): self.assertFalse(fc._kvm_accessible()) def test_kvm_accessible_when_ioctl_succeeds(self): - m = MagicMock() - m.__enter__ = MagicMock(return_value=m) - m.__exit__ = MagicMock(return_value=False) with patch.object(fc.os.path, "exists", return_value=True), \ - patch("builtins.open", return_value=m), \ + patch.object(fc.os, "open", return_value=42), \ + patch.object(fc.os, "close"), \ patch.object(fc.fcntl, "ioctl", return_value=12): self.assertTrue(fc._kvm_accessible()) - def test_kvm_present_but_ioctl_fails(self): - m = MagicMock() - m.__enter__ = MagicMock(return_value=m) - m.__exit__ = MagicMock(return_value=False) + def test_kvm_open_rdwr_fails(self): with patch.object(fc.os.path, "exists", return_value=True), \ - patch("builtins.open", return_value=m), \ + patch.object(fc.os, "open", side_effect=OSError("Permission denied")): + self.assertFalse(fc._kvm_accessible()) + + def test_kvm_present_but_ioctl_fails(self): + with patch.object(fc.os.path, "exists", return_value=True), \ + patch.object(fc.os, "open", return_value=42), \ + patch.object(fc.os, "close"), \ patch.object(fc.fcntl, "ioctl", side_effect=OSError("permission denied")): self.assertFalse(fc._kvm_accessible()) +class TestFirecrackerArtifactCheck(unittest.TestCase): + """status() reports missing guest kernel, dropbear binary, and mke2fs.""" + + def _apply_all_ok(self, stack: contextlib.ExitStack) -> None: + """Stub every status() check to pass except what the test overrides.""" + stack.enter_context(patch.object(fc, "_firecracker_binary_ok", return_value=True)) + stack.enter_context(patch.object(fc, "_kvm_accessible", return_value=True)) + k: MagicMock = MagicMock() + k.is_file.return_value = True + stack.enter_context(patch.object(fc.util, "kernel_path", return_value=k)) + d: MagicMock = MagicMock() + d.is_file.return_value = True + stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=d)) + stack.enter_context(patch.object(fc.shutil, "which", return_value="/usr/bin/x")) + stack.enter_context(patch.object(netpool, "missing_taps", return_value=[])) + stack.enter_context(patch.object(netpool, "pool_size", return_value=8)) + stack.enter_context(patch.object(netpool, "overlapping_routes", return_value=[])) + stack.enter_context(patch.object(fc, "_report_persistence", lambda: None)) + + def test_status_fails_when_kernel_missing(self): + with contextlib.ExitStack() as stack: + self._apply_all_ok(stack) + k: MagicMock = MagicMock() + k.is_file.return_value = False + stack.enter_context(patch.object(fc.util, "kernel_path", return_value=k)) + rc, out = _cap(fc.status) + self.assertEqual(1, rc) + self.assertIn("guest kernel: NOT found", out) + + def test_status_fails_when_dropbear_missing(self): + with contextlib.ExitStack() as stack: + self._apply_all_ok(stack) + d: MagicMock = MagicMock() + d.is_file.return_value = False + stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=d)) + rc, out = _cap(fc.status) + self.assertEqual(1, rc) + self.assertIn("dropbear: NOT found", out) + + def test_status_fails_when_mke2fs_missing(self): + with contextlib.ExitStack() as stack: + self._apply_all_ok(stack) + stack.enter_context(patch.object( + fc.shutil, "which", + side_effect=lambda cmd: (None if cmd == "mke2fs" else "/usr/bin/x"), # type: ignore[misc] + )) + rc, out = _cap(fc.status) + self.assertEqual(1, rc) + self.assertIn("mke2fs: NOT found", out) + + class TestFirecrackerStatusRuntime(unittest.TestCase): """status() reports binary and KVM problems and returns non-zero.""" def _apply_pool_ok(self, stack: contextlib.ExitStack) -> None: + kernel_mock = MagicMock() + kernel_mock.is_file.return_value = True + dropbear_mock = MagicMock() + dropbear_mock.is_file.return_value = True + stack.enter_context(patch.object(fc.util, "kernel_path", return_value=kernel_mock)) + stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=dropbear_mock)) stack.enter_context(patch.object(netpool, "missing_taps", return_value=[])) stack.enter_context(patch.object(netpool, "pool_size", return_value=8)) stack.enter_context(patch.object(netpool, "overlapping_routes", return_value=[])) diff --git a/tests/unit/test_firecracker_backend.py b/tests/unit/test_firecracker_backend.py index 7d4ccd9..6ec358e 100644 --- a/tests/unit/test_firecracker_backend.py +++ b/tests/unit/test_firecracker_backend.py @@ -132,32 +132,56 @@ class TestFirecrackerStatus(unittest.TestCase): rc = fc_setup.status() return rc, buf.getvalue() + def _stub_artifacts(self, fc_setup: object) -> tuple[MagicMock, MagicMock]: + k: MagicMock = MagicMock() + k.is_file.return_value = True + d: MagicMock = MagicMock() + d.is_file.return_value = True + return k, d + def test_ready_when_taps_present_even_if_nft_unverifiable(self): from bot_bottle.backend.firecracker import setup as fc_setup + + def _which(cmd: str) -> str | None: + return None if cmd == "nft" else f"/usr/bin/{cmd}" + + k, d = self._stub_artifacts(fc_setup) with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \ patch.object(fc_setup.netpool, "overlapping_routes", return_value=[]), \ - patch.object(fc_setup.shutil, "which", return_value=None), \ + patch.object(fc_setup.shutil, "which", side_effect=_which), \ patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \ - patch.object(fc_setup, "_kvm_accessible", return_value=True): + patch.object(fc_setup, "_kvm_accessible", return_value=True), \ + patch.object(fc_setup.util, "kernel_path", return_value=k), \ + patch.object(fc_setup.util, "dropbear_path", return_value=d): rc, out = self._run() self.assertEqual(0, rc) self.assertIn("unverified", out) def test_not_ready_when_taps_missing(self): from bot_bottle.backend.firecracker import setup as fc_setup + k, d = self._stub_artifacts(fc_setup) with patch.object(fc_setup.netpool, "missing_taps", return_value=["bbfc0"]), \ patch.object(fc_setup.netpool, "overlapping_routes", return_value=[]), \ - patch.object(fc_setup.shutil, "which", return_value=None): + patch.object(fc_setup.shutil, "which", return_value=None), \ + patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \ + patch.object(fc_setup, "_kvm_accessible", return_value=True), \ + patch.object(fc_setup.util, "kernel_path", return_value=k), \ + patch.object(fc_setup.util, "dropbear_path", return_value=d): rc, _ = self._run() self.assertEqual(1, rc) def test_not_ready_on_range_overlap(self): from bot_bottle.backend.firecracker import netpool from bot_bottle.backend.firecracker import setup as fc_setup + k, d = self._stub_artifacts(fc_setup) conflict = netpool.RouteConflict(dst="10.243.0.0/24", dev="eth0") with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \ patch.object(fc_setup.netpool, "overlapping_routes", return_value=[conflict]), \ - patch.object(fc_setup.shutil, "which", return_value=None): + patch.object(fc_setup.shutil, "which", return_value=None), \ + patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \ + patch.object(fc_setup, "_kvm_accessible", return_value=True), \ + patch.object(fc_setup.util, "kernel_path", return_value=k), \ + patch.object(fc_setup.util, "dropbear_path", return_value=d): rc, out = self._run() self.assertEqual(1, rc) self.assertIn("CLASHES", out)