From cc35b80379b6e4c824c14c0dfc900d8048ed0b5a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Fri, 14 Feb 2025 14:04:12 +0100 Subject: [PATCH 1/2] tests: move TC_06_AppVMMixin from basic to misc 'basic' is ran on every storage pool and as such is a bit time sensitive. Since tests in TC_06_AppVMMixin are getting longer (and not really relevant for storage testing), move it to another file. --- qubes/tests/integ/basic.py | 93 -------------------------- qubes/tests/integ/misc.py | 132 +++++++++++++++++++++++++++++++++++++ rpm_spec/core-dom0.spec.in | 1 + 3 files changed, 133 insertions(+), 93 deletions(-) create mode 100644 qubes/tests/integ/misc.py diff --git a/qubes/tests/integ/basic.py b/qubes/tests/integ/basic.py index b798fd5c5..e8921373d 100644 --- a/qubes/tests/integ/basic.py +++ b/qubes/tests/integ/basic.py @@ -851,93 +851,6 @@ def test_101_resize_root_img_online(self): self.assertGreater(int(new_size.strip()), 19 * 1024**2) -class TC_06_AppVMMixin(object): - template = None - - def setUp(self): - super(TC_06_AppVMMixin, self).setUp() - self.init_default_template(self.template) - - def test_010_os_metadata(self): - tpl = self.app.default_template - if self.template.startswith("fedora-"): - self.assertEqual(tpl.features.get("os-distribution"), "fedora") - version = self.template.split("-")[1] - self.assertEqual(tpl.features.get("os-version"), version) - self.assertIsNotNone(tpl.features.get("os-eol")) - elif self.template.startswith("debian-"): - self.assertEqual(tpl.features.get("os-distribution"), "debian") - version = self.template.split("-")[1] - self.assertEqual(tpl.features.get("os-version"), version) - self.assertIsNotNone(tpl.features.get("os-eol")) - elif self.template.startswith("whonix-"): - self.assertEqual(tpl.features.get("os-distribution"), "whonix") - self.assertEqual(tpl.features.get("os-distribution-like"), "debian") - version = self.template.split("-")[2] - self.assertEqual(tpl.features.get("os-version"), version) - elif self.template.startswith("kali-core"): - self.assertEqual(tpl.features.get("os-distribution"), "kali") - self.assertEqual(tpl.features.get("os-distribution-like"), "debian") - - @unittest.skipUnless( - spawn.find_executable("xdotool"), "xdotool not installed" - ) - def test_121_start_standalone_with_cdrom_vm(self): - cdrom_vmname = self.make_vm_name("cdrom") - self.cdrom_vm = self.app.add_new_vm( - "AppVM", label="red", name=cdrom_vmname - ) - self.loop.run_until_complete(self.cdrom_vm.create_on_disk()) - self.loop.run_until_complete(self.cdrom_vm.start()) - iso_path = self.create_bootable_iso() - with open(iso_path, "rb") as iso_f: - self.loop.run_until_complete( - self.cdrom_vm.run_for_stdio( - "cat > /home/user/boot.iso", stdin=iso_f - ) - ) - - vmname = self.make_vm_name("appvm") - self.vm = self.app.add_new_vm("StandaloneVM", label="red", name=vmname) - self.loop.run_until_complete(self.vm.create_on_disk()) - self.vm.kernel = None - self.vm.virt_mode = "hvm" - - # start the VM using qvm-start tool, to test --cdrom option there - p = self.loop.run_until_complete( - asyncio.create_subprocess_exec( - "qvm-start", - "--cdrom=" + cdrom_vmname + ":/home/user/boot.iso", - self.vm.name, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - ) - ) - (stdout, _) = self.loop.run_until_complete(p.communicate()) - self.assertEqual(p.returncode, 0, stdout) - # check if VM do not crash instantly - self.loop.run_until_complete(asyncio.sleep(50)) - self.assertTrue(self.vm.is_running()) - # Type 'halt' - subprocess.check_call( - [ - "xdotool", - "search", - "--name", - self.vm.name, - "type", - "--window", - "%1", - "halt\r", - ] - ) - for _ in range(10): - if not self.vm.is_running(): - break - self.loop.run_until_complete(asyncio.sleep(1)) - self.assertFalse(self.vm.is_running()) - - def create_testcases_for_templates(): yield from qubes.tests.create_testcases_for_templates( "TC_05_StandaloneVM", @@ -945,12 +858,6 @@ def create_testcases_for_templates(): qubes.tests.SystemTestCase, module=sys.modules[__name__], ) - yield from qubes.tests.create_testcases_for_templates( - "TC_06_AppVM", - TC_06_AppVMMixin, - qubes.tests.SystemTestCase, - module=sys.modules[__name__], - ) def load_tests(loader, tests, pattern): diff --git a/qubes/tests/integ/misc.py b/qubes/tests/integ/misc.py new file mode 100644 index 000000000..a4ae9c1ef --- /dev/null +++ b/qubes/tests/integ/misc.py @@ -0,0 +1,132 @@ +# +# The Qubes OS Project, https://www.qubes-os.org/ +# +# Copyright (C) 2025 Marek Marczykowski-Górecki +# +# +# This library 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 library 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 library; if not, see . +import asyncio +import contextlib +import subprocess +import sys +import unittest +from distutils import spawn + +import qubes + + +class TC_06_AppVMMixin(object): + template = None + + def setUp(self): + super(TC_06_AppVMMixin, self).setUp() + self.init_default_template(self.template) + + def test_010_os_metadata(self): + tpl = self.app.default_template + if self.template.startswith("fedora-"): + self.assertEqual(tpl.features.get("os-distribution"), "fedora") + version = self.template.split("-")[1] + self.assertEqual(tpl.features.get("os-version"), version) + self.assertIsNotNone(tpl.features.get("os-eol")) + elif self.template.startswith("debian-"): + self.assertEqual(tpl.features.get("os-distribution"), "debian") + version = self.template.split("-")[1] + self.assertEqual(tpl.features.get("os-version"), version) + self.assertIsNotNone(tpl.features.get("os-eol")) + elif self.template.startswith("whonix-"): + self.assertEqual(tpl.features.get("os-distribution"), "whonix") + self.assertEqual(tpl.features.get("os-distribution-like"), "debian") + version = self.template.split("-")[2] + self.assertEqual(tpl.features.get("os-version"), version) + elif self.template.startswith("kali-core"): + self.assertEqual(tpl.features.get("os-distribution"), "kali") + self.assertEqual(tpl.features.get("os-distribution-like"), "debian") + + @unittest.skipUnless( + spawn.find_executable("xdotool"), "xdotool not installed" + ) + def test_121_start_standalone_with_cdrom_vm(self): + cdrom_vmname = self.make_vm_name("cdrom") + self.cdrom_vm = self.app.add_new_vm( + "AppVM", label="red", name=cdrom_vmname + ) + self.loop.run_until_complete(self.cdrom_vm.create_on_disk()) + self.loop.run_until_complete(self.cdrom_vm.start()) + iso_path = self.create_bootable_iso() + with open(iso_path, "rb") as iso_f: + self.loop.run_until_complete( + self.cdrom_vm.run_for_stdio( + "cat > /home/user/boot.iso", stdin=iso_f + ) + ) + + vmname = self.make_vm_name("appvm") + self.vm = self.app.add_new_vm("StandaloneVM", label="red", name=vmname) + self.loop.run_until_complete(self.vm.create_on_disk()) + self.vm.kernel = None + self.vm.virt_mode = "hvm" + + # start the VM using qvm-start tool, to test --cdrom option there + p = self.loop.run_until_complete( + asyncio.create_subprocess_exec( + "qvm-start", + "--cdrom=" + cdrom_vmname + ":/home/user/boot.iso", + self.vm.name, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + ) + (stdout, _) = self.loop.run_until_complete(p.communicate()) + self.assertEqual(p.returncode, 0, stdout) + # check if VM do not crash instantly + self.loop.run_until_complete(asyncio.sleep(50)) + self.assertTrue(self.vm.is_running()) + # Type 'halt' + subprocess.check_call( + [ + "xdotool", + "search", + "--name", + self.vm.name, + "type", + "--window", + "%1", + "halt\r", + ] + ) + for _ in range(10): + if not self.vm.is_running(): + break + self.loop.run_until_complete(asyncio.sleep(1)) + self.assertFalse(self.vm.is_running()) + + +def create_testcases_for_templates(): + yield from qubes.tests.create_testcases_for_templates( + "TC_06_AppVM", + TC_06_AppVMMixin, + qubes.tests.SystemTestCase, + module=sys.modules[__name__], + ) + +def load_tests(loader, tests, pattern): + tests.addTests(loader.loadTestsFromNames(create_testcases_for_templates())) + + return tests + + +qubes.tests.maybe_create_testcases_on_import(create_testcases_for_templates) + +# vim: ts=4 sw=4 et diff --git a/rpm_spec/core-dom0.spec.in b/rpm_spec/core-dom0.spec.in index 2ea3d2eb1..532602e92 100644 --- a/rpm_spec/core-dom0.spec.in +++ b/rpm_spec/core-dom0.spec.in @@ -521,6 +521,7 @@ done %{python3_sitelib}/qubes/tests/integ/dom0_update.py %{python3_sitelib}/qubes/tests/integ/vm_update.py %{python3_sitelib}/qubes/tests/integ/mime.py +%{python3_sitelib}/qubes/tests/integ/misc.py %{python3_sitelib}/qubes/tests/integ/network.py %{python3_sitelib}/qubes/tests/integ/network_ipv6.py %{python3_sitelib}/qubes/tests/integ/grub.py From 796ca601bf600a9af85ec269d366ebe4fa468121 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Thu, 13 Feb 2025 16:37:50 +0100 Subject: [PATCH 2/2] tests: add test for password-less emergency console It shouldn't prompt for the password, as the root account is locked. Related to https://github.com/QubesOS/qubes-core-agent-linux/pull/526 --- qubes/tests/integ/misc.py | 122 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 122 insertions(+) diff --git a/qubes/tests/integ/misc.py b/qubes/tests/integ/misc.py index a4ae9c1ef..cbc1f9d02 100644 --- a/qubes/tests/integ/misc.py +++ b/qubes/tests/integ/misc.py @@ -54,6 +54,128 @@ def test_010_os_metadata(self): self.assertEqual(tpl.features.get("os-distribution"), "kali") self.assertEqual(tpl.features.get("os-distribution-like"), "debian") + def test_110_rescue_console(self): + self.loop.run_until_complete(self._test_110_rescue_console()) + + async def _test_110_rescue_console(self): + self.testvm = self.app.add_new_vm( + "AppVM", label="red", name=self.make_vm_name("vm") + ) + await self.testvm.create_on_disk() + self.testvm.kernelopts = "emergency" + # avoid qrexec timeout + self.testvm.features["qrexec"] = "" + self.app.save() + await self.testvm.start() + # call admin.vm.Console via qrexec-client so it sets all the variables + console_proc = await asyncio.create_subprocess_exec( + "qrexec-client", + "-d", + "dom0", + f"DEFAULT:QUBESRPC admin.vm.Console dom0 name {self.testvm.name}", + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + ) + try: + await asyncio.wait_for( + self._interact_emergency_console(console_proc), 120 + ) + finally: + with contextlib.suppress(ProcessLookupError): + console_proc.terminate() + await console_proc.communicate() + + async def _interact_emergency_console( + self, console_proc: asyncio.subprocess.Process + ): + emergency_mode_found = False + whoami_typed = False + while True: + try: + line = await asyncio.wait_for( + console_proc.stdout.readline(), 30 + ) + except TimeoutError: + break + if b"emergency mode" in line: + emergency_mode_found = True + if emergency_mode_found and b"Press Enter" in line: + console_proc.stdin.write(b"\n") + await console_proc.stdin.drain() + # shell prompt doesn't include newline, so the top loop won't + # progress on it + while True: + try: + line2 = await asyncio.wait_for( + console_proc.stdout.read(128), 5 + ) + except TimeoutError: + break + if b"bash" in line2 or b"root#" in line2: + break + console_proc.stdin.write(b"echo $USER\n") + await console_proc.stdin.drain() + whoami_typed = True + if whoami_typed and b"root" in line: + return + if whoami_typed: + self.fail("Calling whoami failed, but emergency console started") + if emergency_mode_found: + self.fail("Emergency mode started, but didn't got shell") + self.fail("Emergency mode not found") + + def test_111_rescue_console_initrd(self): + if "minimal" in self.template: + self.skipTest( + "Test not relevant for minimal template - booting " + "in-vm kernel not supported" + ) + self.loop.run_until_complete(self._test_111_rescue_console_initrd()) + + async def _test_111_rescue_console_initrd(self): + self.testvm = self.app.add_new_vm( + qubes.vm.standalonevm.StandaloneVM, + name=self.make_vm_name("vm"), + label="red", + ) + self.testvm.kernel = None + self.testvm.features.update(self.app.default_template.features) + await self.testvm.clone_disk_files(self.app.default_template) + self.app.save() + + await self.testvm.start() + await self.testvm.run_for_stdio( + "echo 'GRUB_CMDLINE_LINUX=\"$GRUB_CMDLINE_LINUX rd.emergency\"' >> " + "/etc/default/grub", + user="root", + ) + await self.testvm.run_for_stdio( + "update-grub2 || grub2-mkconfig -o /boot/grub2/grub.cfg", + user="root", + ) + await self.testvm.shutdown(wait=True) + + # avoid qrexec timeout + self.testvm.features["qrexec"] = "" + await self.testvm.start() + # call admin.vm.Console via qrexec-client so it sets all the variables + console_proc = await asyncio.create_subprocess_exec( + "qrexec-client", + "-d", + "dom0", + f"DEFAULT:QUBESRPC admin.vm.Console dom0 name {self.testvm.name}", + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + ) + try: + await asyncio.wait_for( + self._interact_emergency_console(console_proc), 60 + ) + finally: + with contextlib.suppress(ProcessLookupError): + console_proc.terminate() + await console_proc.communicate() + @unittest.skipUnless( spawn.find_executable("xdotool"), "xdotool not installed" )