From 65f96d5a5498edd0638198c3b55898b8561ca4ac Mon Sep 17 00:00:00 2001
From: Ben Grande <ben.grande.b@gmail.com>
Date: Fri, 22 Aug 2025 18:18:57 +0200
Subject: [PATCH 1/3] Redo netvm setup after unpause

Applying deferred netvm for preloaded disposables is handled separately
to ensure that before a disposable is returned to the user, the
networking is already set up. If the domain-unpaused event of the
NetVMMixin kicks in before the preload is used, it is ignored by the
"is_preload" attribute, if it kicks after, it is ignored by the absent
"deferred-netvm-original" feature.

Fixes: https://github.com/QubesOS/qubes-issues/issues/10173
For: https://github.com/QubesOS/qubes-issues/issues/1512
---
 qubes/app.py                      |   7 +-
 qubes/tests/integ/network.py      | 350 ++++++++++++++++++++++++------
 qubes/tests/integ/network_ipv6.py | 101 +++++++--
 qubes/tests/vm/mix/net.py         | 156 +++++++++++++
 qubes/tests/vm/qubesvm.py         |  13 +-
 qubes/vm/dispvm.py                |  10 +-
 qubes/vm/mix/net.py               | 108 ++++++++-
 qubes/vm/qubesvm.py               |   8 +
 templates/libvirt/devices/net.xml |   2 +-
 9 files changed, 651 insertions(+), 104 deletions(-)

diff --git a/qubes/app.py b/qubes/app.py
index f559fb651..29d42fa92 100644
--- a/qubes/app.py
+++ b/qubes/app.py
@@ -1621,7 +1621,12 @@ def _domain_event_callback(self, _conn, domain, event, _detail, _opaque):
                 )
         elif event == libvirt.VIR_DOMAIN_EVENT_RESUMED:
             try:
-                vm.fire_event("domain-unpaused")
+                if getattr(vm, "skip_unpause_event", False):
+                    vm.skip_unpause_event = False
+                else:
+                    asyncio.ensure_future(
+                        vm.fire_event_async("domain-unpaused")
+                    )
             except Exception:  # pylint: disable=broad-except
                 self.log.exception(
                     "Uncaught exception from domain-unpaused handler "
diff --git a/qubes/tests/integ/network.py b/qubes/tests/integ/network.py
index b7e22b4f9..d114e9e11 100644
--- a/qubes/tests/integ/network.py
+++ b/qubes/tests/integ/network.py
@@ -42,6 +42,15 @@ class VmNetworkingMixin(object):
     ping_cmd = "ping -W 1 -n -c 1 {target}"
     ping_ip = ping_cmd.format(target=test_ip)
     ping_name = ping_cmd.format(target=test_name)
+    ping_deadline_cmd = (
+        "i=0;"
+        "while test $i -le 10; do "
+        "  if ping -w 2 -n -c 1 {target}; then exit 0; fi;"
+        "  i=$((i+1)); sleep 1; "
+        "done; exit 4"
+    )
+    ping_deadline_ip = ping_deadline_cmd.format(target=test_ip)
+    ping_deadline_name = ping_deadline_cmd.format(target=test_name)
 
     # filled by load_tests
     template = None
@@ -59,12 +68,17 @@ def run_cmd(self, vm, cmd, user="root", timeout=1 << 62):
                 asyncio.wait_for(vm.run_for_stdio(cmd, user=user), timeout)
             )
         except subprocess.CalledProcessError as e:
+            self.log.critical(
+                "Command failed on {}: {}: stdout: {}: stderr: {}".format(
+                    vm.name, cmd, e.stdout, e.stderr
+                )
+            )
             return e.returncode
         return 0
 
     def setUp(self):
         """
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         super(VmNetworkingMixin, self).setUp()
         if self.template.startswith("whonix-"):
@@ -76,12 +90,17 @@ def setUp(self):
         self.testnetvm = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("netvm1"), label="red"
         )
-        self.loop.run_until_complete(self.testnetvm.create_on_disk())
-        self.testnetvm.provides_network = True
-        self.testnetvm.netvm = None
-        # avoid races with NetworkManager, self.configure_netvm() configures
-        # everything directly
-        self.testnetvm.features["service.network-manager"] = False
+        self.testnetvm2 = self.app.add_new_vm(
+            qubes.vm.appvm.AppVM, name=self.make_vm_name("netvm2"), label="red"
+        )
+        self.netvms = [self.testnetvm, self.testnetvm2]
+        for qube in self.netvms:
+            self.loop.run_until_complete(qube.create_on_disk())
+            qube.provides_network = True
+            qube.netvm = None
+            # avoid races with NetworkManager, self.configure_netvm() configures
+            # everything directly
+            qube.features["service.network-manager"] = False
         self.testvm1 = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("vm1"), label="red"
         )
@@ -95,17 +114,25 @@ def _run_cmd_and_log_output(self, vm, cmd):
         """Used in tearDown to collect more info"""
         if not vm.is_running():
             return
-        with contextlib.suppress(subprocess.CalledProcessError):
-            output, _ = self.loop.run_until_complete(
-                vm.run_for_stdio(cmd, user="root", stderr=subprocess.STDOUT)
-            )
-            self.log.critical("{}: {}: {}".format(vm.name, cmd, output))
+        try:
+            if vm.klass == "AdminVM":
+                proc = subprocess.run(
+                    cmd, capture_output=True, check=True, shell=True
+                )
+                stdout = proc.stdout
+            else:
+                stdout, _ = self.loop.run_until_complete(
+                    vm.run_for_stdio(cmd, user="root", stderr=subprocess.STDOUT)
+                )
+        except subprocess.CalledProcessError as e:
+            stdout = getattr(e, "stdout", str(e))
+        self.log.critical("{}: {}: {}".format(vm.name, cmd, stdout))
 
     def tearDown(self):
         # collect more info on failure
         if not self.success():
             for vm in (
-                self.testnetvm,
+                *self.netvms,
                 self.testvm1,
                 getattr(self, "proxy", None),
             ):
@@ -128,72 +155,260 @@ def tearDown(self):
                 self._run_cmd_and_log_output(
                     vm, "systemctl --no-pager status xendriverdomain"
                 )
+                self._run_cmd_and_log_output(
+                    vm, "journalctl --no-pager --since '10 seconds ago'"
+                )
                 self._run_cmd_and_log_output(
                     vm, "cat /var/log/xen/xen-hotplug.log"
                 )
-
+            self._run_cmd_and_log_output(self.app.domains[0], "xl list")
+        del self.netvms
         super(VmNetworkingMixin, self).tearDown()
 
     def configure_netvm(self):
         """
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
 
-        def run_netvm_cmd(cmd):
+        def run_netvm_cmd(qube, cmd):
             try:
                 self.loop.run_until_complete(
-                    self.testnetvm.run_for_stdio(cmd, user="root")
+                    qube.run_for_stdio(cmd, user="root")
                 )
             except subprocess.CalledProcessError as e:
                 self.fail(
-                    "Command '%s' failed: %s%s"
-                    % (cmd, e.stdout.decode(), e.stderr.decode())
+                    "Command failed on %s: '%s': stdout: %s: stderr: %s"
+                    % (qube, cmd, e.stdout.decode(), e.stderr.decode())
                 )
 
-        if not self.testnetvm.is_running():
-            self.loop.run_until_complete(self.testnetvm.start())
-        # Ensure that dnsmasq is installed:
-        try:
-            self.loop.run_until_complete(
-                self.testnetvm.run_for_stdio("dnsmasq --version", user="root")
+        for qube in self.netvms:
+            if not qube.is_running():
+                self.loop.run_until_complete(self.start_vm(qube))
+            # Ensure that dnsmasq is installed:
+            try:
+                self.loop.run_until_complete(
+                    qube.run_for_stdio("dnsmasq --version", user="root")
+                )
+            except subprocess.CalledProcessError:
+                self.skipTest("dnsmasq not installed")
+
+            run_netvm_cmd(qube, "ip link add test0 type dummy")
+            run_netvm_cmd(qube, "ip link set test0 up")
+            run_netvm_cmd(
+                qube, "ip addr add {}/24 dev test0".format(self.test_ip)
             )
-        except subprocess.CalledProcessError:
-            self.skipTest("dnsmasq not installed")
-
-        run_netvm_cmd("ip link add test0 type dummy")
-        run_netvm_cmd("ip link set test0 up")
-        run_netvm_cmd("ip addr add {}/24 dev test0".format(self.test_ip))
-        run_netvm_cmd(
-            "nft add ip qubes custom-input ip daddr {} accept".format(
-                self.test_ip
+            run_netvm_cmd(
+                qube,
+                "nft add ip qubes custom-input ip daddr {} accept".format(
+                    self.test_ip
+                ),
             )
-        )
-        # ignore failure
-        self.run_cmd(self.testnetvm, "while pkill dnsmasq; do sleep 1; done")
-        run_netvm_cmd(
-            "dnsmasq -a {ip} -A /{name}/{ip} -i test0 -z".format(
-                ip=self.test_ip, name=self.test_name
+            # ignore failure
+            self.run_cmd(qube, "while pkill dnsmasq; do sleep 1; done")
+            run_netvm_cmd(
+                qube,
+                "dnsmasq -a {ip} -A /{name}/{ip} -i test0 -z".format(
+                    ip=self.test_ip, name=self.test_name
+                ),
             )
-        )
-        run_netvm_cmd(
-            "rm -f /etc/resolv.conf && echo nameserver {} > /etc/resolv.conf".format(
-                self.test_ip
+            run_netvm_cmd(
+                qube,
+                "rm -f /etc/resolv.conf && echo nameserver {} > /etc/resolv.conf".format(
+                    self.test_ip
+                ),
             )
-        )
-        run_netvm_cmd("systemctl try-restart systemd-resolved || :")
-        run_netvm_cmd("/usr/lib/qubes/qubes-setup-dnat-to-ns")
+            run_netvm_cmd(qube, "systemctl try-restart systemd-resolved || :")
+            run_netvm_cmd(qube, "/usr/lib/qubes/qubes-setup-dnat-to-ns")
 
     def test_000_simple_networking(self):
         """
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.loop.run_until_complete(self.start_vm(self.testvm1))
         self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
         self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0)
 
+    def _networking_paused_from_none_to_existent(
+        self, ip, name, ip_deadline, name_deadline
+    ):
+        test = "netvm=none -> netvm=something"
+        self.log.critical(test)
+        print(test)
+        self.testvm1.netvm = None
+        self.loop.run_until_complete(self.start_vm(self.testvm1))
+        self.loop.run_until_complete(self.testvm1.pause())
+        self.testvm1.netvm = self.testnetvm
+        self.assertEqual(
+            self.testvm1.features.get("deferred-netvm-original", None), ""
+        )
+        self.loop.run_until_complete(self.testvm1.unpause())
+        self._run_cmd_and_log_output(self.testvm1, ip_deadline)
+        self._run_cmd_and_log_output(self.testvm1, name_deadline)
+        self.assertEqual(
+            self.run_cmd(self.testvm1, ip),
+            0,
+            "Ping by IP on " + test + " failed",
+        )
+        self.assertEqual(
+            self.run_cmd(self.testvm1, name),
+            0,
+            "Ping by name on " + test + " failed",
+        )
+        self.assertEqual(
+            self.testvm1.features.get("deferred-netvm-original", None), None
+        )
+        self.shutdown_and_wait(self.testvm1)
+
+    def _networking_paused_from_existent_to_none(
+        self, ip, name, ip_deadline, name_deadline
+    ):
+        # pylint: disable=unused-argument
+        test = "netvm=something -> netvm=none"
+        self.log.critical(test)
+        print(test)
+        self.loop.run_until_complete(self.start_vm(self.testvm1))
+        self.loop.run_until_complete(self.testvm1.pause())
+        self.testvm1.netvm = None
+        self.assertEqual(
+            self.testvm1.features.get("deferred-netvm-original", None),
+            self.testnetvm.name,
+        )
+        self.loop.run_until_complete(self.testvm1.unpause())
+        self.assertNotEqual(
+            self.run_cmd(self.testvm1, ip),
+            0,
+            "Ping by IP on " + test + " succeeded but should have failed",
+        )
+        self.assertNotEqual(
+            self.run_cmd(self.testvm1, name),
+            0,
+            "Ping by name on " + test + " succeeded but should have failed",
+        )
+        self.assertEqual(
+            self.testvm1.features.get("deferred-netvm-original", None), None
+        )
+        self.shutdown_and_wait(self.testvm1)
+
+    def _networking_paused_change_shutdown_old(
+        self, ip, name, ip_deadline, name_deadline
+    ):
+        test = "netvm=something -> netvm=something (shutdown old)"
+        self.log.critical(test)
+        print(test)
+        self.testvm1.netvm = self.testnetvm2
+        self.loop.run_until_complete(self.start_vm(self.testvm1))
+        self.loop.run_until_complete(self.testvm1.pause())
+        self.testvm1.netvm = self.testnetvm
+        self.assertEqual(
+            self.testvm1.features.get("deferred-netvm-original", None),
+            self.testnetvm2.name,
+        )
+        self.shutdown_and_wait(self.testnetvm2)
+        self.loop.run_until_complete(self.testvm1.unpause())
+        self._run_cmd_and_log_output(self.testvm1, ip_deadline)
+        self._run_cmd_and_log_output(self.testvm1, name_deadline)
+        self.assertEqual(
+            self.run_cmd(self.testvm1, ip),
+            0,
+            "Ping by IP on " + test + " failed",
+        )
+        self.assertEqual(
+            self.run_cmd(self.testvm1, name),
+            0,
+            "Ping by name on " + test + " failed",
+        )
+        self.assertEqual(
+            self.testvm1.features.get("deferred-netvm-original", None), None
+        )
+        self.shutdown_and_wait(self.testvm1)
+
+    def _networking_paused_change_purge_old(
+        self, ip, name, ip_deadline, name_deadline
+    ):
+        test = "netvm=something -> netvm=something (purge old)"
+        self.log.critical(test)
+        print(test)
+        self.testvm1.netvm = self.testnetvm2
+        self.loop.run_until_complete(self.start_vm(self.testvm1))
+        self.loop.run_until_complete(self.start_vm(self.testnetvm2))
+        self.loop.run_until_complete(self.testvm1.pause())
+        self.testvm1.netvm = self.testnetvm
+        self.assertEqual(
+            self.testvm1.features.get("deferred-netvm-original", None),
+            self.testnetvm2.name,
+        )
+        self.loop.run_until_complete(self.testnetvm2.kill())
+        del self.app.domains[self.testnetvm2]
+        self.loop.run_until_complete(self.testnetvm2.remove_from_disk())
+        self.app.save()
+        self.testnetvm2 = None
+        self.loop.run_until_complete(self.testvm1.unpause())
+        self._run_cmd_and_log_output(self.testvm1, ip_deadline)
+        self._run_cmd_and_log_output(self.testvm1, name_deadline)
+        self.assertEqual(
+            self.run_cmd(self.testvm1, ip),
+            0,
+            "Ping by IP on " + test + " failed",
+        )
+        self.assertEqual(
+            self.run_cmd(self.testvm1, name),
+            0,
+            "Ping by name on " + test + " failed",
+        )
+        self.assertEqual(
+            self.testvm1.features.get("deferred-netvm-original", None), None
+        )
+        self.shutdown_and_wait(self.testvm1)
+
+    def test_001_simple_networking_paused_from_none_to_existent(self):
+        """
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
+        """
+        self._networking_paused_from_none_to_existent(
+            self.ping_ip,
+            self.ping_name,
+            self.ping_deadline_ip,
+            self.ping_deadline_name,
+        )
+
+    def test_001_simple_networking_paused_from_existent_to_none(self):
+        """
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
+        """
+        self._networking_paused_from_existent_to_none(
+            self.ping_ip,
+            self.ping_name,
+            self.ping_deadline_ip,
+            self.ping_deadline_name,
+        )
+
+    @unittest.skip("kernel issue")
+    def test_001_simple_networking_paused_change_shutdown_old(self):
+        """
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
+        """
+        self._networking_paused_change_shutdown_old(
+            self.ping_ip,
+            self.ping_name,
+            self.ping_deadline_ip,
+            self.ping_deadline_name,
+        )
+
+    @unittest.skip("kernel issue")
+    def test_001_simple_networking_paused_change_purge_old(self):
+        """
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
+        """
+        self._networking_paused_change_purge_old(
+            self.ping_ip,
+            self.ping_name,
+            self.ping_deadline_ip,
+            self.ping_deadline_name,
+        )
+
     def test_010_simple_proxyvm(self):
         """
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.proxy = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("proxy"), label="red"
@@ -228,7 +443,8 @@ def test_010_simple_proxyvm(self):
         )
 
         self.proxy.netvm = None
-        self.loop.run_until_complete(self.testnetvm.shutdown(wait=True))
+        for qube in self.netvms:
+            self.shutdown_and_wait(qube)
         # change IP to test if all info is updated, especially DNS redirect
         self.test_ip = "192.168.45.123"
         self.ping_ip = self.ping_cmd.format(target=self.test_ip)
@@ -263,7 +479,7 @@ def test_010_simple_proxyvm(self):
     )
     def test_020_simple_proxyvm_nm(self):
         """
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.proxy = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("proxy"), label="red"
@@ -340,7 +556,7 @@ def test_020_simple_proxyvm_nm(self):
 
     def test_030_firewallvm_firewall(self):
         """
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.proxy = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("proxy"), label="red"
@@ -389,7 +605,7 @@ def test_030_firewallvm_firewall(self):
             # block all except ICMP
 
             self.testvm1.firewall.rules = [
-                (qubes.firewall.Rule(None, action="accept", proto="icmp"))
+                qubes.firewall.Rule(None, action="accept", proto="icmp")
             ]
             self.testvm1.firewall.save()
             # Ugly hack b/c there is no feedback when the rules are actually
@@ -477,7 +693,7 @@ def test_030_firewallvm_firewall(self):
 
     def test_031_firewall_dynamic_block(self):
         """
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.proxy = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("proxy"), label="red"
@@ -576,7 +792,7 @@ def test_031_firewall_dynamic_block(self):
 
     def test_040_inter_vm(self):
         """
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.proxy = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("proxy"), label="red"
@@ -639,7 +855,7 @@ def test_040_inter_vm(self):
     def test_050_spoof_ip(self):
         """Test if VM IP spoofing is blocked
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.loop.run_until_complete(self.start_vm(self.testvm1))
 
@@ -707,17 +923,17 @@ def test_050_spoof_ip(self):
     def test_100_late_xldevd_startup(self):
         """Regression test for #1990
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         # Simulater late xl devd startup
         cmd = "systemctl stop xendriverdomain"
         if self.run_cmd(self.testnetvm, cmd) != 0:
-            self.fail("Command '%s' failed" % cmd)
+            self.fail("Command failed on '%s': '%s'" % (self.testnetvm, cmd))
         self.loop.run_until_complete(self.start_vm(self.testvm1))
 
         cmd = "systemctl start xendriverdomain"
         if self.run_cmd(self.testnetvm, cmd) != 0:
-            self.fail("Command '%s' failed" % cmd)
+            self.fail("Command failed on '%s': '%s'" % (self.testnetvm, cmd))
 
         # let it initialize the interface(s)
         time.sleep(1)
@@ -798,7 +1014,7 @@ def test_114_reattach_after_provider_crash(self):
     def test_200_fake_ip_simple(self):
         """Test hiding VM real IP
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.testvm1.features["net.fake-ip"] = "192.168.1.128"
         self.testvm1.features["net.fake-gateway"] = "192.168.1.1"
@@ -833,7 +1049,7 @@ def test_200_fake_ip_simple(self):
     def test_201_fake_ip_without_gw(self):
         """Test hiding VM real IP
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.testvm1.features["net.fake-ip"] = "192.168.1.128"
         self.app.save()
@@ -855,7 +1071,7 @@ def test_201_fake_ip_without_gw(self):
     def test_202_fake_ip_firewall(self):
         """Test hiding VM real IP, firewall
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.testvm1.features["net.fake-ip"] = "192.168.1.128"
         self.testvm1.features["net.fake-gateway"] = "192.168.1.1"
@@ -918,7 +1134,7 @@ def test_202_fake_ip_firewall(self):
     def test_203_fake_ip_inter_vm_allow(self):
         """Access VM with "fake IP" from other VM (when firewall allows)
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.proxy = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("proxy"), label="red"
@@ -990,7 +1206,7 @@ def test_203_fake_ip_inter_vm_allow(self):
     def test_204_fake_ip_proxy(self):
         """Test hiding VM real IP
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.proxy = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("proxy"), label="red"
@@ -1054,7 +1270,7 @@ def test_204_fake_ip_proxy(self):
     def test_210_custom_ip_simple(self):
         """Custom AppVM IP
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.testvm1.ip = "192.168.1.1"
         self.app.save()
@@ -1065,7 +1281,7 @@ def test_210_custom_ip_simple(self):
     def test_211_custom_ip_proxy(self):
         """Custom ProxyVM IP
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.proxy = self.app.add_new_vm(
             qubes.vm.appvm.AppVM, name=self.make_vm_name("proxy"), label="red"
@@ -1085,7 +1301,7 @@ def test_211_custom_ip_proxy(self):
     def test_212_custom_ip_firewall(self):
         """Custom VM IP and firewall
 
-        :type self: qubes.tests.SystemTestCase | VMNetworkingMixin
+        :type self: qubes.tests.SystemTestCase | VmNetworkingMixin
         """
         self.testvm1.ip = "192.168.1.1"
 
diff --git a/qubes/tests/integ/network_ipv6.py b/qubes/tests/integ/network_ipv6.py
index 9ce462595..646638acb 100644
--- a/qubes/tests/integ/network_ipv6.py
+++ b/qubes/tests/integ/network_ipv6.py
@@ -37,17 +37,30 @@ class VmIPv6NetworkingMixin(VmNetworkingMixin):
     test_ip6 = "2000:abcd::1"
 
     ping6_cmd = "ping -6 -W 1 -n -c 1 {target}"
+    ping6_deadline_cmd = (
+        "i=0;"
+        "while test $i -le 10; do "
+        "  if ping -6 -w 2 -n -c 1 {target}; then exit 0; fi;"
+        "  i=$((i+1)); sleep 1; "
+        "done; exit 4"
+    )
 
     def setUp(self):
         super(VmIPv6NetworkingMixin, self).setUp()
         self.ping6_ip = self.ping6_cmd.format(target=self.test_ip6)
         self.ping6_name = self.ping6_cmd.format(target=self.test_name)
+        self.ping6_deadline_ip = self.ping6_deadline_cmd.format(
+            target=self.test_ip6
+        )
+        self.ping6_deadline_name = self.ping6_deadline_cmd.format(
+            target=self.test_name
+        )
 
     def tearDown(self):
         # collect more info on failure (ipv4 info collected in parent)
         if self._outcome and not self._outcome.success:
             for vm in (
-                self.testnetvm,
+                *self.netvms,
                 self.testvm1,
                 getattr(self, "proxy", None),
             ):
@@ -59,20 +72,17 @@ def tearDown(self):
                 self._run_cmd_and_log_output(
                     vm, "nft list table ip6 qubes-firewall"
                 )
-
         super().tearDown()
 
     def configure_netvm(self):
         """
         :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
         """
-        self.testnetvm.features["ipv6"] = True
-        super(VmIPv6NetworkingMixin, self).configure_netvm()
 
-        def run_netvm_cmd(cmd):
+        def run_netvm_cmd(qube, cmd):
             try:
                 self.loop.run_until_complete(
-                    self.testnetvm.run_for_stdio(cmd, user="root")
+                    qube.run_for_stdio(cmd, user="root")
                 )
             except subprocess.CalledProcessError as e:
                 self.fail(
@@ -80,19 +90,28 @@ def run_netvm_cmd(cmd):
                     % (cmd, e.stdout.decode(), e.stderr.decode())
                 )
 
-        run_netvm_cmd("ip addr add {}/128 dev test0".format(self.test_ip6))
-        run_netvm_cmd(
-            "nft add ip6 qubes custom-input ip6 daddr {} accept".format(
-                self.test_ip6
+        for qube in self.netvms:
+            qube.features["ipv6"] = True
+        super(VmIPv6NetworkingMixin, self).configure_netvm()
+
+        for qube in self.netvms:
+            run_netvm_cmd(
+                qube, "ip addr add {}/128 dev test0".format(self.test_ip6)
             )
-        )
-        # ignore failure
-        self.run_cmd(self.testnetvm, "while pkill dnsmasq; do sleep 1; done")
-        run_netvm_cmd(
-            "dnsmasq -a {ip} -A /{name}/{ip} -A /{name}/{ip6} -i test0 -z".format(
-                ip=self.test_ip, ip6=self.test_ip6, name=self.test_name
+            run_netvm_cmd(
+                qube,
+                "nft add ip6 qubes custom-input ip6 daddr {} accept".format(
+                    self.test_ip6
+                ),
+            )
+            # ignore failure
+            self.run_cmd(qube, "while pkill dnsmasq; do sleep 1; done")
+            run_netvm_cmd(
+                qube,
+                "dnsmasq -a {ip} -A /{name}/{ip} -A /{name}/{ip6} -i test0 -z".format(
+                    ip=self.test_ip, ip6=self.test_ip6, name=self.test_name
+                ),
             )
-        )
 
     def test_500_ipv6_simple_networking(self):
         """
@@ -102,6 +121,52 @@ def test_500_ipv6_simple_networking(self):
         self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0)
         self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0)
 
+    def test_501_simple_networking_paused_from_none_to_existent(self):
+        """
+        :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
+        """
+        self._networking_paused_from_none_to_existent(
+            self.ping6_ip,
+            self.ping6_name,
+            self.ping6_deadline_ip,
+            self.ping6_deadline_name,
+        )
+
+    def test_501_simple_networking_paused_from_existent_to_none(self):
+        """
+        :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
+        """
+        self._networking_paused_from_existent_to_none(
+            self.ping6_ip,
+            self.ping6_name,
+            self.ping6_deadline_ip,
+            self.ping6_deadline_name,
+        )
+
+    @unittest.skip("kernel issue")
+    def test_501_simple_networking_paused_change_shutdown_old(self):
+        """
+        :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
+        """
+        self._networking_paused_change_shutdown_old(
+            self.ping6_ip,
+            self.ping6_name,
+            self.ping6_deadline_ip,
+            self.ping6_deadline_name,
+        )
+
+    @unittest.skip("kernel issue")
+    def test_501_simple_networking_paused_change_purge_old(self):
+        """
+        :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
+        """
+        self._networking_paused_change_purge_old(
+            self.ping6_ip,
+            self.ping6_name,
+            self.ping6_deadline_ip,
+            self.ping6_deadline_name,
+        )
+
     def test_510_ipv6_simple_proxyvm(self):
         """
         :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
@@ -277,7 +342,7 @@ def test_530_ipv6_firewallvm_firewall(self):
             # block all except ICMP
 
             self.testvm1.firewall.rules = [
-                (qubes.firewall.Rule(None, action="accept", proto="icmp"))
+                qubes.firewall.Rule(None, action="accept", proto="icmp")
             ]
             self.testvm1.firewall.save()
             # Ugly hack b/c there is no feedback when the rules are actually
diff --git a/qubes/tests/vm/mix/net.py b/qubes/tests/vm/mix/net.py
index 7c32a6647..0d61d1033 100644
--- a/qubes/tests/vm/mix/net.py
+++ b/qubes/tests/vm/mix/net.py
@@ -159,6 +159,162 @@ def test_145_netvm_change(self):
                 mock_detach.reset_mock()
                 mock_attach.reset_mock()
 
+    def test_146_netvm_defer(self):
+        vm = self.get_vm()
+        self.setup_netvms(vm)
+        with (
+            patch("qubes.vm.qubesvm.QubesVM.is_running", lambda x: True),
+            patch("qubes.vm.qubesvm.QubesVM.is_paused", lambda x: True),
+            patch("qubes.vm.mix.net.NetVMMixin.attach_network") as mock_attach,
+            patch("qubes.vm.mix.net.NetVMMixin.detach_network") as mock_detach,
+            patch("qubes.vm.qubesvm.QubesVM.create_qdb_entries"),
+            patch("qubes.vm.qubesvm.QubesVM.run_service_for_stdio"),
+        ):
+
+            with self.subTest("try to apply deferred netvm when not set"):
+                with patch(
+                    "qubes.vm.qubesvm.QubesVM.is_paused", lambda x: False
+                ):
+                    self.loop.run_until_complete(vm.apply_deferred_netvm())
+                    mock_detach.assert_not_called()
+                    mock_attach.assert_not_called()
+
+            mock_detach.reset_mock()
+            mock_attach.reset_mock()
+            with self.subTest("changing netvm and restoring original netvm"):
+                original_netvm = vm.netvm.name
+                vm.netvm = self.netvm2
+                mock_detach.assert_not_called()
+                mock_attach.assert_not_called()
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None),
+                    original_netvm,
+                )
+                vm.netvm = original_netvm
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None), None
+                )
+                mock_detach.assert_not_called()
+                mock_attach.assert_not_called()
+
+            mock_detach.reset_mock()
+            mock_attach.reset_mock()
+            with self.subTest(
+                "changing netvm and restoring original netvm from none"
+            ):
+                original_netvm = vm.netvm.name
+                with patch(
+                    "qubes.vm.qubesvm.QubesVM.is_paused", lambda x: False
+                ):
+                    vm.netvm = None
+                mock_detach.reset_mock()
+                mock_attach.reset_mock()
+                vm.netvm = original_netvm
+                mock_detach.assert_not_called()
+                mock_attach.assert_not_called()
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None),
+                    "",
+                )
+                vm.netvm = None
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None), None
+                )
+                mock_detach.assert_not_called()
+                mock_attach.assert_not_called()
+                with patch(
+                    "qubes.vm.qubesvm.QubesVM.is_paused", lambda x: False
+                ):
+                    vm.netvm = original_netvm
+
+            mock_detach.reset_mock()
+            mock_attach.reset_mock()
+            with self.subTest("changing netvm"):
+                original_netvm = vm.netvm.name
+                vm.netvm = self.netvm2
+                mock_detach.assert_not_called()
+                mock_attach.assert_not_called()
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None),
+                    original_netvm,
+                )
+                with patch(
+                    "qubes.vm.qubesvm.QubesVM.is_paused", lambda x: False
+                ):
+                    self.loop.run_until_complete(vm.apply_deferred_netvm())
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None), None
+                )
+                mock_detach.assert_called()
+                mock_attach.assert_called()
+
+            mock_detach.reset_mock()
+            mock_attach.reset_mock()
+            with self.subTest("setting netvm to none"):
+                original_netvm = vm.netvm.name
+                vm.netvm = None
+                mock_detach.assert_not_called()
+                mock_attach.assert_not_called()
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None),
+                    original_netvm,
+                )
+                with patch(
+                    "qubes.vm.qubesvm.QubesVM.is_paused", lambda x: False
+                ):
+                    self.loop.run_until_complete(vm.apply_deferred_netvm())
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None), None
+                )
+                mock_detach.assert_called()
+                mock_attach.assert_not_called()
+
+            mock_detach.reset_mock()
+            mock_attach.reset_mock()
+            with self.subTest("resetting netvm to default"):
+                original_netvm = vm.netvm.name if vm.netvm else ""
+                del vm.netvm
+                mock_detach.assert_not_called()
+                mock_attach.assert_not_called()
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None),
+                    original_netvm,
+                )
+                with patch(
+                    "qubes.vm.qubesvm.QubesVM.is_paused", lambda x: False
+                ):
+                    self.loop.run_until_complete(vm.apply_deferred_netvm())
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None), None
+                )
+                mock_detach.assert_called()
+                mock_attach.assert_called_once()
+
+            mock_detach.reset_mock()
+            mock_attach.reset_mock()
+            with self.subTest("skip apply of preload"):
+                original_netvm = vm.netvm.name
+                vm.is_preload = True
+                vm.netvm = self.netvm2
+                mock_detach.assert_not_called()
+                mock_attach.assert_not_called()
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None),
+                    original_netvm,
+                )
+                with patch(
+                    "qubes.vm.qubesvm.QubesVM.is_paused", lambda x: False
+                ):
+                    self.loop.run_until_complete(vm.apply_deferred_netvm())
+                self.assertEqual(
+                    vm.features.get("deferred-netvm-original", None),
+                    original_netvm,
+                )
+                mock_detach.assert_not_called()
+                mock_attach.assert_not_called()
+            vm.is_preload = False
+            del vm.features["deferred-netvm-original"]
+
     def test_150_ip(self):
         vm = self.get_vm()
         self.setup_netvms(vm)
diff --git a/qubes/tests/vm/qubesvm.py b/qubes/tests/vm/qubesvm.py
index f3e1b6822..50bfa1821 100644
--- a/qubes/tests/vm/qubesvm.py
+++ b/qubes/tests/vm/qubesvm.py
@@ -1691,7 +1691,12 @@ def test_600_libvirt_xml_hvm_pcidev(self):
         )
         dom0 = self.get_vm(name="dom0", qid=0)
         vm = self.get_vm(uuid=my_uuid)
-        vm.netvm = None
+        vm.paused = lambda x: False
+        with unittest.mock.patch(
+            "qubes.vm.qubesvm.QubesVM.is_paused"
+        ) as mock_is_paused:
+            mock_is_paused.return_value = False
+            vm.netvm = None
         vm.virt_mode = "hvm"
         vm.kernel = None
         # even with meminfo-writer enabled, should have memory==maxmem
@@ -1803,7 +1808,11 @@ def test_600_libvirt_xml_hvm_pcidev_s0ix(self):
         dom0 = self.get_vm(name="dom0", qid=0)
         dom0.features["suspend-s0ix"] = True
         vm = self.get_vm(uuid=my_uuid)
-        vm.netvm = None
+        with unittest.mock.patch(
+            "qubes.vm.qubesvm.QubesVM.is_paused"
+        ) as mock_is_paused:
+            mock_is_paused.return_value = False
+            vm.netvm = None
         vm.virt_mode = "hvm"
         vm.kernel = None
         # even with meminfo-writer enabled, should have memory==maxmem
diff --git a/qubes/vm/dispvm.py b/qubes/vm/dispvm.py
index 6ecb0493a..c22ab9e3d 100644
--- a/qubes/vm/dispvm.py
+++ b/qubes/vm/dispvm.py
@@ -419,14 +419,14 @@ def on_domain_paused(
             self.log.info("Paused preloaded qube")
 
     @qubes.events.handler("domain-unpaused")
-    def on_domain_unpaused(
+    async def on_domain_unpaused(
         self, event, **kwargs
     ):  # pylint: disable=unused-argument
         """Mark preloaded disposables as used."""
         # Qube start triggers unpause via 'libvirt_domain.resume()'.
         if self.is_preload and self.is_fully_usable():
             self.log.info("Unpaused preloaded qube will be marked as used")
-            self.use_preload()
+            await self.use_preload()
 
     @qubes.events.handler("domain-shutdown")
     async def on_domain_shutdown(self, _event, **_kwargs):
@@ -547,7 +547,7 @@ async def from_appvm(cls, appvm, preload=False, **kwargs):
                     if dispvm.is_paused():
                         await dispvm.unpause()
                     else:
-                        dispvm.use_preload()
+                        await dispvm.use_preload()
                     app.save()
                     return dispvm
                 except asyncio.TimeoutError:
@@ -585,7 +585,7 @@ async def from_appvm(cls, appvm, preload=False, **kwargs):
         app.save()
         return dispvm
 
-    def use_preload(self):
+    async def use_preload(self):
         """
         Marks preloaded DispVM as used (tainted).
 
@@ -598,6 +598,7 @@ def use_preload(self):
             self.log.info("Using preloaded qube")
             if not appvm.features.get("internal", None):
                 del self.features["internal"]
+            await self.apply_deferred_netvm()
             self.preload_requested = None
             del self.features["preload-dispvm-in-progress"]
         else:
@@ -605,6 +606,7 @@ def use_preload(self):
             self.log.warning("Using a preloaded qube before requesting it")
             if not appvm.features.get("internal", None):
                 del self.features["internal"]
+            await self.apply_deferred_netvm()
             appvm.remove_preload_from_list([self.name])
             self.features["preload-dispvm-in-progress"] = False
         self.app.save()
diff --git a/qubes/vm/mix/net.py b/qubes/vm/mix/net.py
index 383e96589..b2532a225 100644
--- a/qubes/vm/mix/net.py
+++ b/qubes/vm/mix/net.py
@@ -20,12 +20,14 @@
 # License along with this library; if not, see <https://www.gnu.org/licenses/>.
 #
 
-""" This module contains the NetVMMixin """
+"""This module contains the NetVMMixin"""
+import asyncio
 import ipaddress
 import os
 import re
 
 import libvirt  # pylint: disable=import-error
+import lxml.etree
 import qubes
 import qubes.config
 import qubes.events
@@ -341,6 +343,53 @@ def on_domain_started_net(self, event, **kwargs):
             except (qubes.exc.QubesException, libvirt.libvirtError):
                 vm.log.warning("Cannot attach network", exc_info=1)
 
+    async def apply_deferred_netvm(self):
+        """Apply deferred netvm changes in case qube could not apply it at the
+        time it was requested."""
+        if getattr(self, "is_preload", False):
+            # Networking of preloaded disposable must be done when it is being
+            # marked as used, so we guarantee that it finishes before the user
+            # can use the disposable.
+            return
+        deferred_from = self.features.get("deferred-netvm-original", None)
+        if deferred_from is None:
+            return
+        oldvalue = None
+        if deferred_from in self.app.domains:
+            oldvalue = self.app.domains[deferred_from]
+        self.fire_event(
+            "property-pre-set:netvm",
+            pre_event=True,
+            name="netvm",
+            newvalue=self.netvm,
+            oldvalue=oldvalue,
+        )
+        if self.netvm:
+            self.fire_event(
+                "property-set:netvm",
+                name="netvm",
+                newvalue=self.netvm,
+                oldvalue=oldvalue,
+            )
+        del self.features["deferred-netvm-original"]
+        # Guarantee that uplink has been concluded (established or absent).
+        await asyncio.wait_for(
+            self.run_service_for_stdio(
+                "qubes.WaitForNetworkUplink", user="root"
+            ),
+            timeout=10,
+        )
+
+    @qubes.events.handler("domain-unpaused")
+    async def on_domain_unpaused_net(
+        self, event, **kwargs
+    ):  # pylint: disable=unused-argument
+        """Check for deferred netvm changes in case qube was paused while
+        changes happened."""
+        if not self.is_fully_usable():
+            return
+        await self.apply_deferred_netvm()
+
     @qubes.events.handler("domain-pre-shutdown")
     def on_domain_pre_shutdown(self, event, force=False):
         """Checks before NetVM shutdown if any connected domains are running.
@@ -373,6 +422,12 @@ def attach_network(self):
             self.netvm.start()
 
         self.netvm.set_mapped_ip_info_for_vm(self)
+
+        if self.is_paused():
+            self.log.warning(
+                "Deferred attaching libvirt net device because qube is paused"
+            )
+            return
         self.libvirt_domain.attachDevice(
             self.app.env.get_template("libvirt/devices/net.xml").render(vm=self)
         )
@@ -383,13 +438,24 @@ def detach_network(self):
         if not self.is_running():
             raise qubes.exc.QubesVMNotRunningError(self)
         if self.netvm is None:
-            raise qubes.exc.QubesVMError(
-                self, "netvm should not be {}".format(self.netvm)
+            deferred_from = self.features.get("deferred-netvm", None)
+            if deferred_from is not None:
+                raise qubes.exc.QubesVMError(
+                    self, "netvm should not be {}".format(self.netvm)
+                )
+
+        if self.is_paused():
+            self.log.warning(
+                "Deferred detaching libvirt net device because qube is paused"
             )
+            return
 
-        self.libvirt_domain.detachDevice(
-            self.app.env.get_template("libvirt/devices/net.xml").render(vm=self)
-        )
+        # Properties extracted from libvirt_domain to support deferred netvm.
+        root = lxml.etree.fromstring(self.libvirt_domain.XMLDesc())
+        eth = root.find(".//interface[@type='ethernet']")
+        if eth is None:
+            return
+        self.libvirt_domain.detachDevice(lxml.etree.tostring(eth).decode())
 
     def is_networked(self):
         """Check whether this VM can reach network (firewall notwithstanding).
@@ -515,10 +581,30 @@ def on_property_pre_set_netvm(self, event, name, newvalue, oldvalue=None):
                     ),
                 )
 
+        deferred_from = self.features.get("deferred-netvm-original", None)
+        if deferred_from is not None and (
+            (not deferred_from and not newvalue)
+            or (newvalue and deferred_from == newvalue.name)
+        ):
+            # No deferred netvm as original value is restored.
+            del self.features["deferred-netvm-original"]
+            return
+
+        if self.is_paused():
+            if deferred_from is None:
+                self.features["deferred-netvm-original"] = getattr(
+                    oldvalue, "name", ""
+                )
+            return
+
         # don't check oldvalue, because it's missing if it was default
-        if self.netvm is not None:
-            if self.is_running() and self.netvm.is_running():
-                self.detach_network()
+        if deferred_from is None:
+            if self.netvm is None:
+                return
+            if not (self.is_running() and self.netvm.is_running()):
+                return
+
+        self.detach_network()
 
     @qubes.events.handler("property-set:netvm")
     def on_property_set_netvm(self, event, name, newvalue, oldvalue=None):
@@ -526,7 +612,6 @@ def on_property_set_netvm(self, event, name, newvalue, oldvalue=None):
         net-domain-connect event
         """
         # pylint: disable=unused-argument
-
         if oldvalue is not None and oldvalue.is_running():
             oldvalue.reload_connected_ips()
 
@@ -539,7 +624,8 @@ def on_property_set_netvm(self, event, name, newvalue, oldvalue=None):
         if self.is_running():
             # refresh IP, DNS etc
             self.create_qdb_entries()
-            self.attach_network()
+            if not self.is_paused():
+                self.attach_network()
 
             newvalue.fire_event("net-domain-connect", vm=self)
 
diff --git a/qubes/vm/qubesvm.py b/qubes/vm/qubesvm.py
index e783932bc..c25923b94 100644
--- a/qubes/vm/qubesvm.py
+++ b/qubes/vm/qubesvm.py
@@ -439,6 +439,8 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.LocalVM):
 
             Fired when the domain has been unpaused.
 
+            Handler for this event may be asynchronous.
+
             :param subject: Event emitter (the qube object)
             :param event: Event name (``'domain-unpaused'``)
 
@@ -1161,6 +1163,8 @@ def __init__(self, app, xml, volume_config=None, **kwargs):
         # start(). This should not be accessed anywhere else.
         self._domain_stopped_lock = asyncio.Lock()
 
+        self.skip_unpause_event = None
+
         if xml is None:
             # we are creating new VM and attributes came through kwargs
             assert hasattr(self, "qid")
@@ -1531,7 +1535,9 @@ async def start(
                 await self.fire_event_async(
                     "domain-pre-unpaused", pre_event=True
                 )
+                self.skip_unpause_event = True
                 self.libvirt_domain.resume()
+                await self.fire_event_async("domain-unpaused")
 
                 if (
                     self.virt_mode == "hvm"
@@ -1790,7 +1796,9 @@ async def unpause(self):
             raise qubes.exc.QubesVMNotPausedError(self)
 
         await self.fire_event_async("domain-pre-unpaused", pre_event=True)
+        self.skip_unpause_event = True
         self.libvirt_domain.resume()
+        await self.fire_event_async("domain-unpaused")
 
         return self
 
diff --git a/templates/libvirt/devices/net.xml b/templates/libvirt/devices/net.xml
index 8fd4f95ae..4c3824440 100644
--- a/templates/libvirt/devices/net.xml
+++ b/templates/libvirt/devices/net.xml
@@ -8,4 +8,4 @@
     <script path="vif-route-qubes" />
 </interface>
 
-{# vim : set ft=jinja ts=4 sts=4 sw=4 et : #}
+{# vim: set ft=jinja ts=4 sts=4 sw=4 et : #}

From 61a74126f23e549892ff8e8ccec5f2e699c608c3 Mon Sep 17 00:00:00 2001
From: Ben Grande <ben.grande.b@gmail.com>
Date: Mon, 1 Sep 2025 14:24:23 +0200
Subject: [PATCH 2/3] Prohibit Admin API from modifying system features

Fixes: https://github.com/QubesOS/qubes-issues/issues/10173
For: https://github.com/QubesOS/qubes-issues/issues/1512
---
 qubes/ext/admin.py       | 25 ++++++++++++++++++++++++-
 qubes/tests/api_admin.py | 20 ++++++++++++++++++++
 qubes/tests/ext.py       | 31 +++++++++++++++++++++++++++++++
 3 files changed, 75 insertions(+), 1 deletion(-)

diff --git a/qubes/ext/admin.py b/qubes/ext/admin.py
index adf652798..d217e51c2 100644
--- a/qubes/ext/admin.py
+++ b/qubes/ext/admin.py
@@ -28,6 +28,14 @@
 from qubes.device_protocol import DeviceInterface
 
 
+PROHIBITED_FEATURES = [
+    "deferred-netvm-original",
+    "preload-dispvm",
+    "preload-dispvm-completed",
+    "preload-dispvm-in-progress",
+]
+
+
 class JustEvaluateAskResolution(parser.AskResolution):
     async def execute(self):
         pass
@@ -39,6 +47,8 @@ async def execute(self):
 
 
 class AdminExtension(qubes.ext.Extension):
+    # pylint: disable=too-few-public-methods
+
     def __init__(self):
         super().__init__()
         # during tests, __init__() of the extension can be called multiple
@@ -47,7 +57,20 @@ def __init__(self):
             self.policy_cache = utils.PolicyCache(lazy_load=True)
             self.policy_cache.initialize_watcher()
 
-    # pylint: disable=too-few-public-methods
+    @qubes.ext.handler(
+        "admin-permission:admin.vm.feature.Set",
+        "admin-permission:admin.vm.feature.Remove",
+    )
+    def on_feature_set_or_remove(self, vm, event, arg, **kwargs):
+        """Forbid changing specific features"""
+        # pylint: disable=unused-argument
+        if arg in PROHIBITED_FEATURES:
+            raise qubes.exc.PermissionDenied(
+                "changing this feature is prohibited by {}.{}".format(
+                    __name__, type(self).__name__
+                )
+            )
+
     @qubes.ext.handler(
         "admin-permission:admin.vm.tag.Set",
         "admin-permission:admin.vm.tag.Remove",
diff --git a/qubes/tests/api_admin.py b/qubes/tests/api_admin.py
index b31dc4abe..418f765c5 100644
--- a/qubes/tests/api_admin.py
+++ b/qubes/tests/api_admin.py
@@ -43,6 +43,7 @@
 import qubes.tests
 import qubes.storage
 
+
 from qubes.device_protocol import (
     DeviceInfo,
     VirtualDevice,
@@ -1667,6 +1668,25 @@ def test_301_feature_remove_none(self):
             )
         self.assertFalse(self.app.save.called)
 
+    def test_303_feature_prohibited(self):
+        del self.app.domains[0].fire_event
+        feature = qubes.ext.admin.PROHIBITED_FEATURES[0]
+        with self.assertRaises(qubes.exc.PermissionDenied):
+            self.call_mgmt_func(
+                b"admin.vm.feature.Set",
+                b"test-vm1",
+                str(feature).encode(),
+                b"some-value",
+            )
+
+        self.vm.features[feature] = False
+        with self.assertRaises(qubes.exc.PermissionDenied):
+            self.call_mgmt_func(
+                b"admin.vm.feature.Remove",
+                b"test-vm1",
+                str(feature).encode(),
+            )
+
     def test_310_feature_checkwithtemplate(self):
         self.vm.features["test-feature"] = "some-value"
         value = self.call_mgmt_func(
diff --git a/qubes/tests/ext.py b/qubes/tests/ext.py
index a72801491..570be73d8 100644
--- a/qubes/tests/ext.py
+++ b/qubes/tests/ext.py
@@ -21,6 +21,7 @@
 import os
 import unittest.mock
 
+import qubes.ext.admin
 import qubes.ext.core_features
 import qubes.ext.custom_persist
 import qubes.ext.services
@@ -2579,3 +2580,33 @@ def test_015_feature_set_path_with_colon_without_options(self):
         self.vm.untrusted_qdb.write.assert_called_with(
             "/persist/test", "/var/test:dir:with:colon"
         )
+
+
+class TC_50_Admin(qubes.tests.QubesTestCase):
+    maxDiff = None
+
+    def setUp(self):
+        super().setUp()
+        self.ext = qubes.ext.admin.AdminExtension()
+
+    def tearDown(self):
+        self.ext.on_qubes_close("app", "qubes-close")
+        super().tearDown()
+
+    def test_000_features_permission(self):
+        feature = qubes.ext.admin.PROHIBITED_FEATURES[0]
+        with self.assertRaises(qubes.exc.PermissionDenied):
+            self.ext.on_feature_set_or_remove(
+                "test-vm1",
+                "admin-permission:admin.vm.feature.Set",
+                feature,
+            )
+
+    def test_000_tags_permission(self):
+        tag = "created-by-test"
+        with self.assertRaises(qubes.exc.PermissionDenied):
+            self.ext.on_tag_set_or_remove(
+                "test-vm1",
+                "admin-permission:admin.vm.tag.Set",
+                tag,
+            )

From fe202649529418d3f376982fc5893712e53cc084 Mon Sep 17 00:00:00 2001
From: Ben Grande <ben.grande.b@gmail.com>
Date: Thu, 18 Sep 2025 08:24:00 +0200
Subject: [PATCH 3/3] Print string representation of preload feature

Concatenating empty ('') value is lost on translation.

For: https://github.com/QubesOS/qubes-issues/issues/1512
---
 qubes/vm/adminvm.py         | 2 +-
 qubes/vm/mix/dvmtemplate.py | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/qubes/vm/adminvm.py b/qubes/vm/adminvm.py
index 20362c743..bdd61b724 100644
--- a/qubes/vm/adminvm.py
+++ b/qubes/vm/adminvm.py
@@ -393,7 +393,7 @@ def on_feature_set_preload_dispvm_max(
             return
         if not (appvm := getattr(self.app, "default_dispvm", None)):
             return
-        reason = "global feature was set to " + str(value)
+        reason = "global feature was set to " + repr(value)
         asyncio.ensure_future(
             appvm.fire_event_async("domain-preload-dispvm-start", reason=reason)
         )
diff --git a/qubes/vm/mix/dvmtemplate.py b/qubes/vm/mix/dvmtemplate.py
index 9a177a37c..69f680370 100644
--- a/qubes/vm/mix/dvmtemplate.py
+++ b/qubes/vm/mix/dvmtemplate.py
@@ -143,7 +143,7 @@ def on_feature_set_preload_dispvm_max(
             return
         if self.is_global_preload_set():
             return
-        reason = "local feature was set to " + str(value)
+        reason = "local feature was set to " + repr(value)
         asyncio.ensure_future(
             self.fire_event_async("domain-preload-dispvm-start", reason=reason)
         )