Add qvm-restart

doc/manpages/qvm-restart.rst

@@ -0,0 +1,54 @@
+.. program:: qvm-restart
+
+:program:`qvm-restart` -- Restart selected or currently running qubes
+=====================================================================
+
+Synopsis
+--------
+
+:command:`qvm-restart` [-h] [--verbose] [--quiet] [--all] [--exclude *EXCLUDE*] [--timeout *TIMEOUT*] [*VMNAME*]
+
+Options
+-------
+
+.. option:: --help, -h
+
+   show the help message and exit
+
+.. option:: --verbose, -v
+
+   increase verbosity
+
+.. option:: --quiet, -q
+
+   decrease verbosity
+
+.. option:: --all
+
+   perform the action on all running qubes except dom0 & unnamed DispVMs
+
+.. option:: --exclude=EXCLUDE
+
+   exclude the qube from :option:`--all`
+
+.. option:: --timeout
+
+   timeout after which domains are killed. The default is decided by global
+   `default_shutdown_timeout` property and qube `shutdown_timeout` property
+
+.. option:: --version
+
+   Show program's version number and exit
+
+
+Authors
+-------
+
+| Joanna Rutkowska <joanna at invisiblethingslab dot com>
+| Rafal Wojtczuk <rafal at invisiblethingslab dot com>
+| Marek Marczykowski <marmarek at invisiblethingslab dot com>
+| Wojtek Porczyk <woju at invisiblethingslab dot com>
+
+| For complete author list see: https://github.com/QubesOS/qubes-core-admin-client.git
+
+.. vim: ts=3 sw=3 et tw=80

doc/manpages/qvm-shutdown.rst

@@ -6,7 +6,7 @@
 Synopsis
 --------
 
-:command:`qvm-shutdown` [-h] [--verbose] [--quiet] [--all] [--exclude *EXCLUDE*] [--wait] [--timeout *TIMEOUT*] [*VMNAME*]
+:command:`qvm-shutdown` [-h] [--verbose] [--quiet] [--all] [--force] [--exclude *EXCLUDE*] [--wait] [--timeout *TIMEOUT*] [*VMNAME*]
 
 Options
 -------

qubesadmin/tests/tools/qvm_restart.py

@@ -0,0 +1,188 @@
+# -*- encoding: utf-8 -*-
+#
+# The Qubes OS Project, http://www.qubes-os.org
+#
+# Copyright (C) 2025 Marek Marczykowski-Górecki
+#                               <[email protected]>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation; either version 2.1 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with this program; if not, see <http://www.gnu.org/licenses/>.
+
+# pylint: disable=missing-docstring
+
+import asyncio
+import unittest.mock
+
+import qubesadmin.tests
+import qubesadmin.tests.tools
+import qubesadmin.tools.qvm_restart
+
+
+class TC_00_qvm_restart(qubesadmin.tests.QubesTestCase):
+    @unittest.skipUnless(
+        qubesadmin.tools.qvm_restart.have_events, "Events not present"
+    )
+    def test_000_restart_running(self):
+        """Restarting just one already running qube"""
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        mock_events = unittest.mock.AsyncMock()
+        patch = unittest.mock.patch(
+            "qubesadmin.events.EventsDispatcher._get_events_reader", mock_events
+        )
+        patch.start()
+        self.addCleanup(patch.stop)
+        mock_events.side_effect = qubesadmin.tests.tools.MockEventsReader(
+            [
+                b"1\0\0connection-established\0\0",
+                b"1\0some-vm\0domain-shutdown\0\0",
+            ]
+        )
+
+        self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = (
+            b"0\x00some-vm class=AppVM state=Running\n"
+        )
+        self.app.expected_calls[
+            ("some-vm", "admin.vm.Shutdown", "force", None)
+        ] = b"0\x00"
+        self.app.expected_calls[
+            ("some-vm", "admin.vm.CurrentState", None, None)
+        ] = (
+            [b"0\x00power_state=Running"]
+            + [b"0\x00power_state=Halted"]
+            + [b"0\x00power_state=Halted"]
+        )
+        self.app.expected_calls[("some-vm", "admin.vm.Start", None, None)] = (
+            b"0\x00"
+        )
+        qubesadmin.tools.qvm_restart.main(["some-vm"], app=self.app)
+        self.assertAllCalled()
+
+    @unittest.skipUnless(
+        qubesadmin.tools.qvm_restart.have_events, "Events not present"
+    )
+    def test_001_restart_halted(self):
+        """Trying restart on a already halted qube"""
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        mock_events = unittest.mock.AsyncMock()
+        patch = unittest.mock.patch(
+            "qubesadmin.events.EventsDispatcher._get_events_reader", mock_events
+        )
+        patch.start()
+        self.addCleanup(patch.stop)
+        mock_events.side_effect = qubesadmin.tests.tools.MockEventsReader(
+            [
+                b"1\0\0connection-established\0\0",
+                b"1\0some-vm\0domain-shutdown\0\0",
+            ]
+        )
+
+        self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = (
+            b"0\x00some-vm class=AppVM state=Halted\n"
+        )
+        self.app.expected_calls[
+            ("some-vm", "admin.vm.Shutdown", "force", None)
+        ] = b"0\x00"
+        self.app.expected_calls[
+            ("some-vm", "admin.vm.CurrentState", None, None)
+        ] = (
+            [b"0\x00power_state=Halted"]
+            + [b"0\x00power_state=Halted"]
+            + [b"0\x00power_state=Halted"]
+        )
+        self.app.expected_calls[("some-vm", "admin.vm.Start", None, None)] = (
+            b"0\x00"
+        )
+        qubesadmin.tools.qvm_restart.main(["some-vm"], app=self.app)
+        self.assertAllCalled()
+
+    @unittest.skipUnless(
+        qubesadmin.tools.qvm_restart.have_events, "Events not present"
+    )
+    def test_002_restart_all(self):
+        """Restarting all running qubes (and skipping unnamed DispVMs)"""
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        mock_events = unittest.mock.AsyncMock()
+        patch = unittest.mock.patch(
+            "qubesadmin.events.EventsDispatcher._get_events_reader", mock_events
+        )
+        patch.start()
+        self.addCleanup(patch.stop)
+        mock_events.side_effect = qubesadmin.tests.tools.MockEventsReader(
+            [
+                b"1\0\0connection-established\0\0",
+                b"1\0some-vm\0domain-shutdown\0\0",
+                b"1\0sys-usb\0domain-shutdown\0\0",
+            ]
+        )
+
+        self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = (
+            b"0\x00some-vm class=AppVM state=Running\n"
+            b"dom0 class=AdminVM state=Running\n"
+            b"sys-usb class=DispVM state=Running\n"
+            b"disp007 class=DispVM state=Running\n"
+            b"dormant-vm class=DispVM state=Halted\n"
+        )
+        self.app.expected_calls[
+            ("sys-usb", "admin.vm.CurrentState", None, None)
+        ] = [b"0\x00power_state=Running"]
+        self.app.expected_calls[
+            ("sys-usb", "admin.vm.property.Get", "auto_cleanup", None)
+        ] = b"0\x00default=True type=bool False"
+        self.app.expected_calls[
+            ("disp007", "admin.vm.CurrentState", None, None)
+        ] = [b"0\x00power_state=Running"]
+        self.app.expected_calls[
+            ("disp007", "admin.vm.property.Get", "auto_cleanup", None)
+        ] = b"0\x00default=True type=bool True"
+        self.app.expected_calls[
+            ("dormant-vm", "admin.vm.CurrentState", None, None)
+        ] = [b"0\x00power_state=Halted"]
+        for vm in ["some-vm", "sys-usb"]:
+            self.app.expected_calls[
+                (vm, "admin.vm.Shutdown", "force", None)
+            ] = b"0\x00"
+            self.app.expected_calls[
+                (vm, "admin.vm.CurrentState", None, None)
+            ] = (
+                [b"0\x00power_state=Running"]
+                + [b"0\x00power_state=Running"]
+                + [b"0\x00power_state=Halted"]
+                + [b"0\x00power_state=Halted"]
+            )
+            self.app.expected_calls[(vm, "admin.vm.Start", None, None)] = (
+                b"0\x00"
+            )
+        qubesadmin.tools.qvm_restart.main(["--all"], app=self.app)
+        self.assertAllCalled()
+
+    def test_003_restart_dispvm(self):
+        """Trying to restart a unnamed DispVM"""
+        self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = (
+            b"0\x00some-vm class=AppVM state=Running\n"
+            b"dom0 class=AdminVM state=Running\n"
+            b"sys-usb class=DispVM state=Running\n"
+            b"disp007 class=DispVM state=Running\n"
+            b"dormant-vm class=DispVM state=Halted\n"
+        )
+        self.app.expected_calls[
+            ("disp007", "admin.vm.property.Get", "auto_cleanup", None)
+        ] = b"0\x00default=True type=bool True"
+        with self.assertRaises(SystemExit):
+            qubesadmin.tools.qvm_restart.main(["disp007"], app=self.app)
+        self.assertAllCalled()

qubesadmin/tools/qvm_restart.py

@@ -0,0 +1,88 @@
+# encoding=utf-8
+#
+# The Qubes OS Project, http://www.qubes-os.org
+#
+# Copyright (C) 2010-2016  Joanna Rutkowska <[email protected]>
+# Copyright (C) 2011-2025  Marek Marczykowski-Górecki
+#                                              <[email protected]>
+# Copyright (C) 2016       Wojtek Porczyk <[email protected]>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation; either version 2.1 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with this program; if not, see <http://www.gnu.org/licenses/>.
+
+"""Restart a qube / qubes"""
+
+import sys
+
+try:
+    import qubesadmin.events.utils
+
+    have_events = True
+except ImportError:
+    have_events = False
+import qubesadmin.tools
+from qubesadmin.tools import qvm_start, qvm_shutdown
+
+parser = qubesadmin.tools.QubesArgumentParser(
+    description=__doc__, vmname_nargs="+"
+)
+
+parser.add_argument(
+    "--timeout",
+    action="store",
+    type=float,
+    help="timeout after which domains are killed before being restarted",
+)
+
+
+def main(args=None, app=None):  # pylint: disable=missing-docstring
+    args = parser.parse_args(args, app=app)
+
+    if not args.all_domains:
+        # Check if user explicitly specified dom0 or unnamed DispVMs
+        invalid_domains = [
+            vm
+            for vm in args.domains
+            if vm.klass == "AdminVM"
+            or (vm.klass == "DispVM" and vm.auto_cleanup)
+        ]
+        if invalid_domains:
+            parser.error_runtime(
+                "Can not restart: "
+                + ", ".join(vm.name for vm in invalid_domains),
+                "dom0 or unnamed DispVMs could not be restarted",
+            )
+        target_domains = args.domains
+    else:
+        # Only restart running, non-DispVM and not dom0 with --all option
+        target_domains = [
+            vm
+            for vm in args.domains
+            if vm.get_power_state() == "Running"
+            and vm.klass != "AdminVM"
+            and not (vm.klass == "DispVM" and vm.auto_cleanup)
+        ]
+
+    # Forcing shutdown to allow graceful restart of ServiceVMs
+    shutdown_cmd = [vm.name for vm in target_domains] + ["--wait", "--force"]
+    shutdown_cmd += ["--timeout", str(args.timeout)] if args.timeout else []
+    shutdown_cmd += ["--quiet"] if args.quiet else []
+    qvm_shutdown.main(shutdown_cmd, app=args.app)
+
+    start_cmd = [vm.name for vm in target_domains]
+    start_cmd += ["--quiet"] if args.quiet else []
+    qvm_start.main(start_cmd, app=args.app)
+
+
+if __name__ == "__main__":
+    sys.exit(main())