Allow VM boot modes to change a VM's default user

qubes/ext/core_features.py

@@ -108,9 +108,11 @@ async def qubes_features_request(self, vm, event, untrusted_features):
         # handle boot mode advertisement
         old_bootmode_info = {}
         for feature_key, feature_val in vm.features.items():
-            if feature_key.startswith(
-                "boot-mode.kernelopts."
-            ) or feature_key.startswith("boot-mode.name."):
+            if (
+                feature_key.startswith("boot-mode.kernelopts.")
+                or feature_key.startswith("boot-mode.name.")
+                or feature_key.startswith("boot-mode.default-user.")
+            ):
                 old_bootmode_info[feature_key] = feature_val
         new_bootmode_info = {}
         new_bootmode_names = []
@@ -136,18 +138,26 @@ async def qubes_features_request(self, vm, event, untrusted_features):
             untrusted_feature_key,
             untrusted_feature_value,
         ) in untrusted_features.items():
-            if untrusted_feature_key.startswith("boot-mode.name."):
-                bootmode_key_parts = untrusted_feature_key.split(".")
-                if len(bootmode_key_parts) != 3:
-                    # Boot mode key contains unexpected data, reject it
-                    continue
-                bootmode_name = bootmode_key_parts[2]
-                if bootmode_name == "":
-                    continue
+            if not untrusted_feature_key.startswith("boot-mode."):
+                continue
+            bootmode_key_parts = untrusted_feature_key.split(".")
+            if len(bootmode_key_parts) != 3:
+                # Boot mode key contains unexpected data, reject it
+                continue
+            bootmode_name = bootmode_key_parts[2]
+            if bootmode_name == "":
+                continue
+            if (
+                f"boot-mode.kernelopts.{bootmode_name}" not in new_bootmode_info
+            ) and bootmode_name != "default":
+                continue
+            if untrusted_feature_key.startswith(
+                "boot-mode.name."
+            ) or untrusted_feature_key.startswith("boot-mode.default-user."):
                 if (
-                    f"boot-mode.kernelopts.{bootmode_name}"
-                    not in new_bootmode_info
-                ) and bootmode_name != "default":
+                    untrusted_feature_key.startswith("boot-mode.default-user.")
+                    and bootmode_name == "default"
+                ):
                     continue
                 bootmode_feature = untrusted_feature_key
                 bootmode_value = untrusted_feature_value

qubes/tests/ext.py

@@ -1587,6 +1587,95 @@ def test_055_bootmode_preserve_oldvals(self):
             ],
         )
 
+    def test_056_bootmode_default_user(self):
+        del self.vm.template
+        self.loop.run_until_complete(
+            self.ext.qubes_features_request(
+                self.vm,
+                "features-request",
+                untrusted_features={
+                    "boot-mode.name.vmreq": "VMReq",
+                    "boot-mode.kernelopts.vmreq": "vmreq1 vmreq2",
+                    "boot-mode.default-user.vmreq": "altuser",
+                },
+            )
+        )
+        self.assertListEqual(
+            self.vm.mock_calls,
+            [
+                ("features.items", (), {}),
+                (
+                    "features.__setitem__",
+                    ("boot-mode.kernelopts.vmreq", "vmreq1 vmreq2"),
+                    {},
+                ),
+                (
+                    "features.__setitem__",
+                    ("boot-mode.name.vmreq", "VMReq"),
+                    {},
+                ),
+                (
+                    "features.__setitem__",
+                    ("boot-mode.default-user.vmreq", "altuser"),
+                    {},
+                ),
+                ("features.get", ("qrexec", False), {}),
+                ("features.get", ("qrexec", False), {}),
+            ],
+        )
+
+    def test_056_bootmode_default_user_mismatch(self):
+        del self.vm.template
+        self.loop.run_until_complete(
+            self.ext.qubes_features_request(
+                self.vm,
+                "features-request",
+                untrusted_features={
+                    "boot-mode.name.vmreq": "VMReq",
+                    "boot-mode.kernelopts.vmreq": "vmreq1 vmreq2",
+                    "boot-mode.default-user.nope": "altuser",
+                },
+            )
+        )
+        self.assertListEqual(
+            self.vm.mock_calls,
+            [
+                ("features.items", (), {}),
+                (
+                    "features.__setitem__",
+                    ("boot-mode.kernelopts.vmreq", "vmreq1 vmreq2"),
+                    {},
+                ),
+                (
+                    "features.__setitem__",
+                    ("boot-mode.name.vmreq", "VMReq"),
+                    {},
+                ),
+                ("features.get", ("qrexec", False), {}),
+                ("features.get", ("qrexec", False), {}),
+            ],
+        )
+
+    def test_057_bootmode_default_user_default_bootmode(self):
+        del self.vm.template
+        self.loop.run_until_complete(
+            self.ext.qubes_features_request(
+                self.vm,
+                "features-request",
+                untrusted_features={
+                    "boot-mode.default-user.default": "altuser",
+                },
+            )
+        )
+        self.assertListEqual(
+            self.vm.mock_calls,
+            [
+                ("features.items", (), {}),
+                ("features.get", ("qrexec", False), {}),
+                ("features.get", ("qrexec", False), {}),
+            ],
+        )
+
     def test_100_servicevm_feature(self):
         self.vm.provides_network = True
         self.ext.set_servicevm_feature(self.vm)

qubes/tests/integ/basic.py

@@ -513,6 +513,30 @@ async def _test_bootmode(self, tpl, vm):
         else:
             self.assertIn("opt1=val1", cmdline)
 
+    async def _test_bootmode_default_user(self, vm):
+        await vm.start()
+        await vm.run_for_stdio(
+            "cat > /etc/qubes/post-install.d/50-test.sh",
+            input=b"""#!/bin/sh
+
+qvm-features-request boot-mode.kernelopts.defuser="opt2"
+qvm-features-request boot-mode.name.defuser="Mode with default user"
+qvm-features-request boot-mode.default-user.defuser="altuser"
+qvm-features-request boot-mode.active="defuser"
+         """,
+            user="root",
+        )
+        await vm.run_for_stdio("useradd -m altuser", user="root")
+        await vm.run_for_stdio(
+            "chmod +x " "/etc/qubes/post-install.d/50-test.sh", user="root"
+        )
+        await vm.run_service_for_stdio("qubes.PostInstall", user="root")
+        await vm.shutdown(wait=True)
+
+        await vm.start()
+        user_id = (await vm.run_for_stdio("id -un"))[0].decode()
+        self.assertEqual(user_id.strip(), "altuser")
+
     def test_210_bootmode_template(self):
         self.test_template = self.app.add_new_vm(
             qubes.vm.templatevm.TemplateVM,
@@ -551,6 +575,20 @@ def test_211_bootmode_standalone(self):
         self.app.save()
         self.loop.run_until_complete(self._test_bootmode(self.vm, self.vm))
 
+    def test_212_bootmode_default_user(self):
+        self.vm = self.app.add_new_vm(
+            qubes.vm.standalonevm.StandaloneVM,
+            name=self.make_vm_name("vm"),
+            label="red",
+        )
+        self.vm.clone_properties(self.app.default_template)
+        self.vm.features.update(self.app.default_template.features)
+        self.loop.run_until_complete(
+            self.vm.clone_disk_files(self.app.default_template)
+        )
+        self.app.save()
+        self.loop.run_until_complete(self._test_bootmode_default_user(self.vm))
+
 
 class TC_01_Properties(qubes.tests.SystemTestCase):
     # pylint: disable=attribute-defined-outside-init

qubes/tests/vm/qubesvm.py

@@ -3232,3 +3232,25 @@ def test_811_default_bootmode(self):
         self.assertEqual(vm.bootmode, "testmode3")
         del vm.template.features["boot-mode.kernelopts.testmode3"]
         self.assertEqual(vm.bootmode, "default")
+
+    def test_812_bootmode_default_user(self):
+        vm = self.get_vm(cls=qubes.vm.appvm.AppVM)
+        vm.template = self.get_vm(cls=qubes.vm.templatevm.TemplateVM)
+        vm.bootmode = qubes.property.DEFAULT
+        self.assertEqual(vm.get_default_user(), "user")
+        vm.features["boot-mode.kernelopts.testmode1"] = "abc def"
+        vm.features["boot-mode.default-user.testmode1"] = "altuser"
+        vm.features["boot-mode.active"] = "testmode1"
+        self.assertEqual(vm.get_default_user(), "altuser")
+        del vm.features["boot-mode.default-user.testmode1"]
+        self.assertEqual(vm.get_default_user(), "user")
+        vm.features["boot-mode.default-user.testmode1"] = "altuser"
+        del vm.features["boot-mode.kernelopts.testmode1"]
+        self.assertEqual(vm.get_default_user(), "user")
+        del vm.features["boot-mode.default-user.testmode1"]
+        vm.template.features["boot-mode.kernelopts.testmode2"] = "ghi jkl"
+        vm.template.features["boot-mode.default-user.testmode2"] = "altuser2"
+        vm.features["boot-mode.active"] = "testmode2"
+        self.assertEqual(vm.get_default_user(), "altuser2")
+        del vm.features["boot-mode.active"]
+        self.assertEqual(vm.get_default_user(), "user")

qubes/vm/qubesvm.py

@@ -1807,7 +1807,7 @@
         name = self.name + "-dm" if stubdom else self.name
 
         if user is None:
-            user = self.default_user
+            user = self.get_default_user()
 
         if self.is_paused():
             # XXX what about autostart?
@@ -1876,7 +1876,7 @@
         """  # pylint: disable=redefined-builtin
 
         if user is None:
-            user = self.default_user
+            user = self.get_default_user()
 
         return await asyncio.create_subprocess_exec(
             qubes.config.system_path["qrexec_client_path"],
@@ -2106,7 +2106,7 @@
                 "--",
                 str(self.xid),
                 self.name,
-                self.default_user,
+                self.get_default_user(),
             ]
 
         if not self.debug:
@@ -2274,6 +2274,21 @@
 
     # state of the machine
 
+    def get_default_user(self):
+        """Return a user account name suitable for use as a default user.
+
+        Usually returns the value of the default_user property, but also
+        allows boot modes to override the default user.
+        """
+        if self.bootmode == "default":
+            return self.default_user
+        bootmode_default_user = self.features.check_with_template(
+            f"boot-mode.default-user.{self.bootmode}", None
+        )
+        if bootmode_default_user is None:
+            return self.default_user
+        return bootmode_default_user
+
     def get_power_state(self):
         """Return power state description string.
 
@@ -2726,7 +2741,7 @@
 
         self.untrusted_qdb.write("/name", self.name)
         self.untrusted_qdb.write("/type", self.__class__.__name__)
-        self.untrusted_qdb.write("/default-user", self.default_user)
+        self.untrusted_qdb.write("/default-user", self.get_default_user())
         self.untrusted_qdb.write("/qubes-vm-updateable", str(self.updateable))
         self.untrusted_qdb.write(
             "/qubes-vm-persistence", "full" if self.updateable else "rw-only"