Fix changing default_netvm

qubes/app.py

@@ -1700,31 +1700,26 @@ def on_property_pre_set_default_netvm(
         self, event, name, newvalue, oldvalue=None
     ):
         # pylint: disable=unused-argument,invalid-name
-        if (
-            newvalue is not None
-            and oldvalue is not None
-            and oldvalue.is_running()
-            and not newvalue.is_running()
-            and self.domains.get_vms_connected_to(oldvalue)
-        ):
-            raise qubes.exc.QubesVMNotRunningError(
-                newvalue,
-                "Cannot change {!r} to domain that "
-                "is not running ({!r}).".format(name, newvalue.name),
-            )
+        for vm in self.domains:
+            if hasattr(vm, "netvm") and vm.property_is_default("netvm"):
+                # Use pre-set event instead of pre-reset, so it can get both
+                # old and new values, and perform necessary actions.
+                # Especially, it needs to detach old netvm.
+                vm.fire_event(
+                    "property-pre-set:netvm",
+                    pre_event=True,
+                    name="netvm",
+                    newvalue=newvalue,
+                    oldvalue=oldvalue,
+                )
 
     @qubes.events.handler("property-set:default_netvm")
     def on_property_set_default_netvm(
         self, event, name, newvalue, oldvalue=None
     ):
         # pylint: disable=unused-argument
         for vm in self.domains:
-            if (
-                hasattr(vm, "provides_network")
-                and not vm.provides_network
-                and hasattr(vm, "netvm")
-                and vm.property_is_default("netvm")
-            ):
+            if hasattr(vm, "netvm") and vm.property_is_default("netvm"):
                 # fire property-reset:netvm as it is responsible for resetting
                 # netvm to its default value
                 vm.fire_event(

qubes/tests/app.py

@@ -1116,6 +1116,41 @@ def test_207_default_kernel(self):
                 existence.side_effect = [True, False]
                 self.app.default_kernel = "unittest_GNU_Hurd_1.0.0"
 
+    def test_208_default_netvm_change(self):
+        netvm1 = self.app.add_new_vm(
+            "AppVM",
+            name="netvm1",
+            template=self.template,
+            label="red",
+            provides_network=True,
+            netvm=None,
+        )
+        netvm2 = self.app.add_new_vm(
+            "AppVM",
+            name="netvm2",
+            template=self.template,
+            label="red",
+            provides_network=True,
+            netvm=None,
+        )
+        self.app.default_netvm = netvm1
+        with (
+            mock.patch("qubes.vm.qubesvm.QubesVM.is_running", lambda x: True),
+            mock.patch(
+                "qubes.vm.mix.net.NetVMMixin.attach_network"
+            ) as mock_attach,
+            mock.patch(
+                "qubes.vm.mix.net.NetVMMixin.detach_network"
+            ) as mock_detach,
+            mock.patch("qubes.vm.qubesvm.QubesVM.create_qdb_entries"),
+        ):
+
+            self.app.default_netvm = netvm2
+            mock_detach.assert_called()
+            mock_attach.assert_called()
+
+        self.app.default_netvm = None
+
     def test_300_preload_default_dispvm(self):
         """Fire event for new setting from no previous one."""
         self.appvm.features["preload-dispvm-max"] = "1"
@@ -1138,11 +1173,10 @@ def test_301_preload_default_dispvm_change(self):
         self.app.default_dispvm = self.appvm
         self.appvm_alt.features["preload-dispvm-max"] = "1"
         # Global is not set and thus there are no events.
-        with mock.patch.object(
-            self.appvm, "fire_event_async"
-        ) as mock_old, mock.patch.object(
-            self.appvm_alt, "fire_event_async"
-        ) as mock_new:
+        with (
+            mock.patch.object(self.appvm, "fire_event_async") as mock_old,
+            mock.patch.object(self.appvm_alt, "fire_event_async") as mock_new,
+        ):
             self.app.default_dispvm = self.appvm_alt
             mock_old.assert_not_called()
             mock_new.assert_not_called()
@@ -1151,11 +1185,10 @@ def test_301_preload_default_dispvm_change(self):
         self.app.default_dispvm = self.appvm
         self.appvm.features["preload-dispvm-max"] = "2"
         self.appvm_alt.features["preload-dispvm-max"] = "2"
-        with mock.patch.object(
-            self.appvm, "fire_event_async"
-        ) as mock_old, mock.patch.object(
-            self.appvm_alt, "fire_event_async"
-        ) as mock_new:
+        with (
+            mock.patch.object(self.appvm, "fire_event_async") as mock_old,
+            mock.patch.object(self.appvm_alt, "fire_event_async") as mock_new,
+        ):
             self.app.default_dispvm = self.appvm_alt
             mock_old.assert_called_once_with(
                 "domain-preload-dispvm-start", reason=mock.ANY
@@ -1172,11 +1205,10 @@ def test_302_preload_default_dispvm_change_noop(self):
         self.appvm_alt.features["preload-dispvm-max"] = "1"
         self.app.domains["dom0"].features["preload-dispvm-max"] = "1"
         self.app.default_dispvm = self.appvm
-        with mock.patch.object(
-            self.appvm, "fire_event_async"
-        ) as mock_old, mock.patch.object(
-            self.appvm_alt, "fire_event_async"
-        ) as mock_new:
+        with (
+            mock.patch.object(self.appvm, "fire_event_async") as mock_old,
+            mock.patch.object(self.appvm_alt, "fire_event_async") as mock_new,
+        ):
             self.app.default_dispvm = self.appvm_alt
             mock_old.assert_not_called()
             mock_new.assert_not_called()
@@ -1189,11 +1221,10 @@ def test_303_preload_default_dispvm_change_partial_noop(self):
         self.appvm_alt.features["preload-dispvm-max"] = "1"
         self.app.domains["dom0"].features["preload-dispvm-max"] = "2"
         self.app.default_dispvm = self.appvm
-        with mock.patch.object(
-            self.appvm, "fire_event_async"
-        ) as mock_old, mock.patch.object(
-            self.appvm_alt, "fire_event_async"
-        ) as mock_new:
+        with (
+            mock.patch.object(self.appvm, "fire_event_async") as mock_old,
+            mock.patch.object(self.appvm_alt, "fire_event_async") as mock_new,
+        ):
             self.app.default_dispvm = self.appvm_alt
             mock_old.assert_not_called()
             mock_new.assert_called_once_with(

qubes/tests/vm/mix/net.py

@@ -21,6 +21,7 @@
 #
 import ipaddress
 import unittest
+from unittest.mock import patch
 
 import qubes
 import qubes.vm.qubesvm
@@ -122,6 +123,42 @@ def test_144_netvm_loopback2(self):
         self.netvm1.netvm = vm
         self.assertPropertyInvalidValue(vm, "netvm", self.netvm2)
 
+    def test_145_netvm_change(self):
+        vm = self.get_vm()
+        self.setup_netvms(vm)
+        with (
+            patch("qubes.vm.qubesvm.QubesVM.is_running", 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"),
+        ):
+
+            with self.subTest("setting netvm to none"):
+                vm.netvm = None
+                mock_detach.assert_called_once()
+                mock_attach.assert_not_called()
+                mock_detach.reset_mock()
+
+            with self.subTest("connecting netvm again"):
+                vm.netvm = self.netvm1
+                mock_detach.assert_not_called()
+                mock_attach.assert_called_once()
+                mock_attach.reset_mock()
+
+            with self.subTest("changing netvm"):
+                vm.netvm = self.netvm2
+                mock_detach.assert_called_once()
+                mock_attach.assert_called_once()
+                mock_detach.reset_mock()
+                mock_attach.reset_mock()
+
+            with self.subTest("resetting netvm to default"):
+                del vm.netvm
+                mock_detach.assert_called_once()
+                mock_attach.assert_called_once()
+                mock_detach.reset_mock()
+                mock_attach.reset_mock()
+
     def test_150_ip(self):
         vm = self.get_vm()
         self.setup_netvms(vm)

qubes/tests/vm/qubesvm.py

@@ -1230,7 +1230,7 @@ def test_600_libvirt_xml_hvm_with_qemu_args(self):
         vm.netvm = None
         vm.virt_mode = "hvm"
         vm.debug = True
-        vm.features['qemu-extra-args'] = '-some-option'
+        vm.features["qemu-extra-args"] = "-some-option"
         libvirt_xml = vm.create_config_file()
         self.assertXMLEqual(
             lxml.etree.XML(libvirt_xml), lxml.etree.XML(expected)

qubes/vm/mix/net.py

@@ -322,7 +322,7 @@ def on_domain_shutdown_net(self, event, **kwargs):
                 pass
 
     @qubes.events.handler("domain-start")
-    def on_domain_started(self, event, **kwargs):
+    def on_domain_started_net(self, event, **kwargs):
         """Connect this domain to its downstream domains. Also reload firewall
         in its netvm.
 
@@ -467,7 +467,7 @@ def reload_connected_ips(self):
 
     @qubes.events.handler("property-pre-reset:netvm")
     def on_property_pre_reset_netvm(self, event, name, oldvalue=None):
-        """Sets the the NetVM to default NetVM"""
+        """Sets the NetVM to default NetVM"""
         # pylint: disable=unused-argument
         # we are changing to default netvm
         newvalue = type(self).netvm.get_default(self)
@@ -485,7 +485,7 @@ def on_property_pre_reset_netvm(self, event, name, oldvalue=None):
 
     @qubes.events.handler("property-reset:netvm")
     def on_property_reset_netvm(self, event, name, oldvalue=None):
-        """Sets the the NetVM to default NetVM"""
+        """Sets the NetVM to default NetVM"""
         # pylint: disable=unused-argument
         # we are changing to default netvm
         newvalue = self.netvm

test-packages/qubesdb.py

@@ -1,4 +1,7 @@
 class QubesDB:
+    def __init__(self, name):
+        pass
+
     def read(self, key):
         return b'testvm'