firecracker status(): open /dev/kvm O_RDWR; check kernel, dropbear, mke2fs

Two reviewer findings addressed:

1. _kvm_accessible() now opens /dev/kvm with O_RDWR|O_CLOEXEC instead
   of read-only. VM creation requires write access; a read-only
   descriptor can satisfy KVM_GET_API_VERSION but fails at boot time.

2. status() now checks every hard prerequisite that require_firecracker()
   checks at launch: guest kernel image, static dropbear binary, and
   mke2fs. Previously a host with a configured TAP pool but missing
   artifacts could pass status() and then fail during launch.

Regression coverage: TestFirecrackerKvmCheck gains test_kvm_open_rdwr_fails;
TestFirecrackerArtifactCheck covers each missing artifact; existing
TestFirecrackerStatus fixtures updated to stub the new checks.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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
+    VM creation requires write access; opening read-only may satisfy the
-    permission bits but a non-functional KVM subsystem is caught here
+    ioctl but fails at boot time, so O_RDWR is the permission check."""
-    rather than at VM boot time."""
     if not os.path.exists(util._KVM_DEVICE):
         return False
     try:
-        with open(util._KVM_DEVICE, "rb") as kvm:
+        fd = os.open(util._KVM_DEVICE, os.O_RDWR | os.O_CLOEXEC)
-            fcntl.ioctl(kvm, _KVM_GET_API_VERSION)
+        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:

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):
+    def test_kvm_open_rdwr_fails(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", 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=[]))

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)