Merge remote-tracking branch 'origin/pr/807'

* origin/pr/807:
  Avoid hanging qubesd on unresponsive domU shutdown
  Raise appropriate exception on unresponsive guest
  Cleanup libvirt state query
  Helper to detect hypervisor via libvirt

Pull request description:

For: QubesOS/qubes-issues#10835

Author: marmarek

qubes/app.py

@@ -181,6 +181,9 @@ def __init__(self, offline_mode=None, libvirt_reconnect_cb=None):
         self._libvirt_conn = None
         self._xs = None
         self._xc = None
+        self._libvirt_conn_type = None
+        self._libvirt_conn_uri = None
+        self._is_xen = False
 
     @property
     def offline_mode(self):
@@ -203,14 +206,18 @@ def init_vmm_connection(self):
                 "VMM operations disabled in offline mode"
             )
 
-        if "xen.lowlevel.xs" in sys.modules:
-            self._xs = xen.lowlevel.xs.xs()
-        if "xen.lowlevel.xc" in sys.modules:
-            self._xc = xen.lowlevel.xc.xc()
         self._libvirt_conn = VirConnectWrapper(
             qubes.config.defaults["libvirt_uri"],
             reconnect_cb=self._libvirt_reconnect_cb,
         )
+        self._libvirt_conn_type = self._libvirt_conn.getType()
+        self._libvirt_conn_uri = self._libvirt_conn.getURI()
+        if self._libvirt_conn_type == "Xen":
+            self._is_xen = True
+            if "xen.lowlevel.xs" in sys.modules:
+                self._xs = xen.lowlevel.xs.xs()
+            if "xen.lowlevel.xc" in sys.modules:
+                self._xc = xen.lowlevel.xc.xc()
         libvirt.registerErrorHandler(self._libvirt_error_handler, None)
 
     @property
@@ -219,21 +226,33 @@ def libvirt_conn(self):
         self.init_vmm_connection()
         return self._libvirt_conn
 
+    @property
+    def libvirt_conn_type(self):
+        """Connection type to libvirt"""
+        self.init_vmm_connection()
+        return self._libvirt_conn_type
+
+    @property
+    def libvirt_conn_uri(self):
+        """Connection URI to libvirt"""
+        self.init_vmm_connection()
+        return self._libvirt_conn_uri
+
+    @property
+    def is_xen(self):
+        self.init_vmm_connection()
+        return self._is_xen
+
     @property
     def xs(self):
         """Connection to Xen Store
 
         This property is available only when running on Xen.
         """
-
-        # XXX what about the case when we run under KVM,
-        # but xen modules are importable?
-        if "xen.lowlevel.xs" not in sys.modules:
+        if not self.is_xen:
             raise AttributeError(
                 "xs object is available under Xen hypervisor only"
             )
-
-        self.init_vmm_connection()
         return self._xs
 
     @property
@@ -242,15 +261,10 @@ def xc(self):
 
         This property is available only when running on Xen.
         """
-
-        # XXX what about the case when we run under KVM,
-        # but xen modules are importable?
-        if "xen.lowlevel.xc" not in sys.modules:
+        if not self.is_xen:
             raise AttributeError(
                 "xc object is available under Xen hypervisor only"
             )
-
-        self.init_vmm_connection()
         return self._xc
 
     def close(self):
@@ -347,21 +361,19 @@ def get_free_xen_memory(self):
 
         :raises NotImplementedError: when not under Xen
         """
-        try:
-            self._physinfo = self.app.vmm.xc.physinfo()
-        except AttributeError:
+        if not self.app.vmm.is_xen:
             raise NotImplementedError("This function requires Xen hypervisor")
+        self._physinfo = self.app.vmm.xc.physinfo()
         return int(self._physinfo["free_memory"])
 
     def is_iommu_supported(self):
         """Check if IOMMU is supported on this platform"""
         if self._physinfo is None:
-            try:
-                self._physinfo = self.app.vmm.xc.physinfo()
-            except AttributeError:
+            if not self.app.vmm.is_xen:
                 raise NotImplementedError(
                     "This function requires Xen hypervisor"
                 )
+            self._physinfo = self.app.vmm.xc.physinfo()
         return "hvm_directio" in self._physinfo["virt_caps"]
 
     def get_vm_stats(self, previous_time=None, previous=None, only_vm=None):
@@ -401,18 +413,17 @@ def get_vm_stats(self, previous_time=None, previous=None, only_vm=None):
 
         current_time = time.time()
         current = {}
-        try:
-            if only_vm:
-                xid = only_vm.xid
-                if xid < 0:
-                    raise qubes.exc.QubesVMNotRunningError(only_vm)
-                info = self.app.vmm.xc.domain_getinfo(xid, 1)
-                if info[0]["domid"] != xid:
-                    raise qubes.exc.QubesVMNotRunningError(only_vm)
-            else:
-                info = self.app.vmm.xc.domain_getinfo(0, 1024)
-        except AttributeError:
+        if not self.app.vmm.is_xen:
             raise NotImplementedError("This function requires Xen hypervisor")
+        if only_vm:
+            xid = only_vm.xid
+            if xid < 0:
+                raise qubes.exc.QubesVMNotRunningError(only_vm)
+            info = self.app.vmm.xc.domain_getinfo(xid, 1)
+            if info[0]["domid"] != xid:
+                raise qubes.exc.QubesVMNotRunningError(only_vm)
+        else:
+            info = self.app.vmm.xc.domain_getinfo(0, 1024)
         # TODO: add stubdomain stats to actual VMs
         for vm in info:
             domid = vm["domid"]

qubes/tests/vm/mix/net.py

@@ -20,6 +20,7 @@
 # License along with this library; if not, see <https://www.gnu.org/licenses/>.
 #
 import ipaddress
+from unittest import mock
 from unittest.mock import patch
 
 import qubes
@@ -395,7 +396,16 @@ def test_180_shutdown(self, mock_halted, mock_shutdown):
             with self.assertRaises(qubes.exc.QubesVMInUseError):
                 self.loop.run_until_complete(vm.netvm.shutdown())
             vm.is_preload = True
-            self.loop.run_until_complete(vm.netvm.shutdown())
+            mock_proc = mock.AsyncMock()
+            mock_proc.communicate.return_value = (b"", None)
+            mock_proc.wait.return_value = 0
+            mock_proc.returncode = 0
+            with (
+                patch.object(vm.app, "vmm") as mock_vmm,
+                patch("asyncio.create_subprocess_exec", return_value=mock_proc),
+            ):
+                mock_vmm.is_xen = True
+                self.loop.run_until_complete(vm.netvm.shutdown())
 
     def test_200_vmid_to_ipv4(self):
         testcases = (

qubes/tests/vm/qubesvm.py

@@ -2950,6 +2950,74 @@ def test_626_qdb_keyboard_layout_change(
     async def coroutine_mock(self, mock, *args, **kwargs):
         return mock(*args, **kwargs)
 
+    @unittest.mock.patch("asyncio.create_subprocess_exec")
+    @unittest.mock.patch(
+        "qubes.vm.qubesvm.QubesVM.is_running", return_value=True
+    )
+    @unittest.mock.patch("qubes.vm.qubesvm.QubesVM.libvirt_domain")
+    @unittest.mock.patch(
+        "qubes.vm.qubesvm.QubesVM.is_halted", return_value=False
+    )
+    def test_650_shutdown(
+        self, mock_halted, mock_shutdown, mock_running, mock_async_sub
+    ):
+        # pylint: disable=unused-argument
+        vm = self.get_vm()
+        mock_proc = unittest.mock.AsyncMock()
+        mock_proc.communicate.return_value = (b"", None)
+        mock_proc.returncode = 0
+        mock_proc.wait.return_value = mock_proc.returncode
+        mock_async_sub.return_value = mock_proc
+        with unittest.mock.patch.object(vm.app, "vmm") as mock_vmm:
+            with self.subTest("xen"):
+                mock_vmm.is_xen = True
+                self.loop.run_until_complete(vm.shutdown())
+                mock_async_sub.assert_called_once_with(
+                    "xl",
+                    "shutdown",
+                    "-F",
+                    vm.name,
+                    stdin=unittest.mock.ANY,
+                    stdout=unittest.mock.ANY,
+                    stderr=unittest.mock.ANY,
+                )
+
+            mock_async_sub.reset_mock()
+            with self.subTest("xen-fail"):
+                mock_proc.communicate.return_value = (b"Oh no (xen)", None)
+                mock_proc.returncode = 1
+                mock_vmm.is_xen = True
+                with self.assertRaises(qubes.exc.QubesVMShutdownTimeoutError):
+                    self.loop.run_until_complete(vm.shutdown())
+
+            mock_async_sub.reset_mock()
+            with self.subTest("not-xen"):
+                mock_proc.returncode = 0
+                uri = "kvm:///"
+                mock_vmm.is_xen = False
+                mock_vmm.libvirt_conn_uri = uri
+                self.loop.run_until_complete(vm.shutdown())
+                mock_async_sub.assert_called_once_with(
+                    "virsh",
+                    "-c",
+                    uri,
+                    "shutdown",
+                    vm.name,
+                    stdin=unittest.mock.ANY,
+                    stdout=unittest.mock.ANY,
+                    stderr=unittest.mock.ANY,
+                )
+
+            mock_async_sub.reset_mock()
+            with self.subTest("not-xen-fail"):
+                mock_proc.communicate.return_value = (b"Oh no (not-xen)", None)
+                mock_proc.returncode = 1
+                uri = "kvm:///"
+                mock_vmm.is_xen = False
+                mock_vmm.libvirt_conn_uri = uri
+                with self.assertRaises(qubes.exc.QubesVMShutdownTimeoutError):
+                    self.loop.run_until_complete(vm.shutdown())
+
     @unittest.mock.patch("asyncio.create_subprocess_exec")
     def test_700_run_service(self, mock_subprocess):
         start_mock = unittest.mock.AsyncMock()

qubes/vm/qubesvm.py

@@ -1,3 +1,4 @@
+# pylint: disable=too-many-lines
 #
 # The Qubes OS Project, https://www.qubes-os.org/
 #
@@ -1661,7 +1662,39 @@ async def shutdown(self, force=False, wait=False, timeout=None):
             if self.is_paused():
                 self.libvirt_domain.destroy()
             else:
-                self.libvirt_domain.shutdown()
+                # Some libvirt actions have a global lock on a domain, blocking
+                # a lot of libvirt operations and even qubesd. When possible to
+                # act without it, do so to avoid the whole qubesd hanging.
+                if self.app.vmm.is_xen:
+                    command = ["xl", "shutdown", "-F", self.name]
+                else:
+                    uri = self.app.vmm.libvirt_conn_uri
+                    command = ["virsh", "-c", uri, "shutdown", self.name]
+                try:
+                    proc = await asyncio.create_subprocess_exec(
+                        *command,
+                        stdin=asyncio.subprocess.DEVNULL,
+                        stdout=asyncio.subprocess.PIPE,
+                        stderr=asyncio.subprocess.STDOUT,
+                    )
+                    stdout, _ = await proc.communicate()
+                    await proc.wait()
+                    if proc.returncode:
+                        raise subprocess.CalledProcessError(
+                            proc.returncode,
+                            command,
+                            output=stdout,
+                        )
+                except subprocess.CalledProcessError as e:
+                    self.log.error(
+                        "Attempted {!s} with subprocess but exited with "
+                        "error code: {!s}: {!r}".format(
+                            "shutdown",
+                            e.returncode,
+                            qubes.utils.sanitize_stderr_for_log(e.output),
+                        )
+                    )
+                    raise qubes.exc.QubesVMShutdownTimeoutError(self)
 
             if wait:
                 if timeout is None:
@@ -2453,22 +2486,19 @@ def get_power_state(self):
 
         try:
             if libvirt_domain.isActive():
-                # pylint: disable=line-too-long
-                if libvirt_domain.state()[0] == libvirt.VIR_DOMAIN_PAUSED:
-                    return "Paused"
-                if libvirt_domain.state()[0] == libvirt.VIR_DOMAIN_CRASHED:
-                    return "Crashed"
-                if libvirt_domain.state()[0] == libvirt.VIR_DOMAIN_SHUTDOWN:
-                    return "Halting"
-                if libvirt_domain.state()[0] == libvirt.VIR_DOMAIN_SHUTOFF:
-                    return "Dying"
-                if (
-                    libvirt_domain.state()[0] == libvirt.VIR_DOMAIN_PMSUSPENDED
-                ):  # nopep8
-                    return "Suspended"
+                state_dict = {
+                    libvirt.VIR_DOMAIN_PAUSED: "Paused",  # 0x3
+                    libvirt.VIR_DOMAIN_SHUTDOWN: "Halting",  # 0x4
+                    libvirt.VIR_DOMAIN_SHUTOFF: "Dying",  # 0x5
+                    libvirt.VIR_DOMAIN_CRASHED: "Crashed",  # 0x6
+                    libvirt.VIR_DOMAIN_PMSUSPENDED: "Suspended",  # 0x7
+                }
+                state = libvirt_domain.state()[0]
+                if state in state_dict:
+                    return state_dict[state]
                 if not self.is_fully_usable():
                     return "Transient"
-                return "Running"
+                return "Running"  # 0x1
 
             return "Halted"
         except libvirt.libvirtError as e: