diff --git a/qubes/app.py b/qubes/app.py
index a4e38da38..632519a04 100644
--- a/qubes/app.py
+++ b/qubes/app.py
@@ -735,6 +735,45 @@ def validate_kernel(obj, property_name: str, kernel: str) -> None:
)
+def get_qube_prop_deps(
+ qube,
+ system_properties: list | None = None,
+ qube_properties: list | None = None,
+):
+ if not system_properties:
+ system_properties = []
+ if not qube_properties:
+ qube_properties = []
+ system_deps: list = []
+ qube_deps: list = []
+ for obj in itertools.chain(qube.app.domains, (qube.app,)):
+ if obj is qube:
+ continue
+ for prop in obj.property_list():
+ if not (
+ isinstance(prop, qubes.vm.VMProperty)
+ and getattr(obj, prop.__name__, None) == qube
+ ):
+ continue
+ if getattr(obj, "is_preload", False) and (
+ prop.__name__ == "template"
+ or (
+ prop.__name__ in ["default_dispvm", "management_dispvm"]
+ and getattr(obj, "template", None) == qube
+ )
+ ):
+ continue
+ if isinstance(obj, qubes.app.Qubes):
+ if system_properties and prop.__name__ not in system_properties:
+ continue
+ system_deps.append(prop.__name__)
+ elif not obj.property_is_default(prop):
+ if qube_properties and prop.__name__ not in qube_properties:
+ continue
+ qube_deps.append((obj.name, prop.__name__))
+ return system_deps, qube_deps
+
+
class Qubes(qubes.PropertyHolder):
"""Main Qubes application
@@ -890,16 +929,20 @@ class Qubes(qubes.PropertyHolder):
"default_dispvm",
load_stage=3,
default=None,
- doc="Default DispVM base for service calls",
+ setter=qubes.vm.setter_disposable_template,
allow_none=True,
+ doc="""Default disposable template to be used for spawning disposable
+ qubes for service calls in this system.""",
)
management_dispvm = qubes.VMProperty(
"management_dispvm",
load_stage=3,
default=None,
- doc="Default DispVM base for managing VMs",
allow_none=True,
+ setter=qubes.vm.setter_disposable_template,
+ doc="""Default disposable template to be used for spawning disposable
+ qubes for managing a qube in this system.""",
)
default_pool = qubes.property(
@@ -1046,7 +1089,10 @@ def store(self):
return self._store
def _migrate_global_properties(self):
- """Migrate renamed/dropped properties"""
+ """Migrate renamed/dropped properties or properties that had weak or no
+ setter to a stricter setter, that would make current value invalid,
+ requiring setting it to None, to avoid failure to load XML and qubesd.
+ """
if self.xml is None:
return
@@ -1091,6 +1137,21 @@ def _migrate_global_properties(self):
pass
node_default_fw_netvm.getparent().remove(node_default_fw_netvm)
+ # Drop invalid setting.
+ for prop in ["default_dispvm", "management_dispvm"]:
+ for obj in [self] + list(self.domains):
+ if obj.xml is None:
+ continue
+ node_dispvm = obj.xml.find(
+ f"./properties/property[@name='{prop}']"
+ )
+ if node_dispvm is None or node_dispvm.text is None:
+ continue
+ dispvm = self.domains[node_dispvm.text]
+ if getattr(dispvm, "template_for_dispvms", None):
+ continue
+ node_dispvm.text = ""
+
def _migrate_labels(self):
"""Migrate changed labels"""
if self.xml is None:
@@ -1649,28 +1710,11 @@ def on_domain_pre_deleted(self, event, vm):
"""
# pylint: disable=unused-argument
dependencies = []
- for obj in itertools.chain(self.domains, (self,)):
- if obj is vm:
- # allow removed VM to reference itself
- continue
- for prop in obj.property_list():
- with suppress(AttributeError):
- if (
- isinstance(prop, qubes.vm.VMProperty)
- and getattr(obj, prop.__name__) == vm
- ):
- if getattr(obj, "is_preload", False) and (
- prop.__name__ == "template"
- or (
- prop.__name__ == "default_dispvm"
- and getattr(obj, "template", None) == vm
- )
- ):
- continue
- if isinstance(obj, qubes.app.Qubes):
- dependencies.insert(0, ("@GLOBAL", prop.__name__))
- elif not obj.property_is_default(prop):
- dependencies.append((obj.name, prop.__name__))
+ system_deps, qube_deps = get_qube_prop_deps(qube=vm)
+ if system_deps:
+ dependencies = [("@GLOBAL", dep) for dep in system_deps]
+ if qube_deps:
+ dependencies.extend(qube_deps)
if dependencies:
self.log.error(
"Cannot remove %s as it is used by %s",
diff --git a/qubes/tests/app.py b/qubes/tests/app.py
index 09341606e..489a52781 100644
--- a/qubes/tests/app.py
+++ b/qubes/tests/app.py
@@ -637,7 +637,169 @@ def test_100_property_migrate_default_fw_netvm(self):
app.close()
del app
- def test_101_property_migrate_label(self):
+ def test_101_property_migrate_disposable_template(self):
+ xml_template = """
+
+
+
+ {app_default_dispvm}
+ {app_management_dispvm}
+
+
+
+
+
+
+
+
+
+
+
+ {adminvm_default_dispvm}
+
+
+
+
+
+ 1
+ work
+
+ 2fcfc1f4-b2fe-4361-931a-c5294b35edfa
+ {work_management_dispvm}
+
+
+
+
+
+
+
+ 3
+ disp-template
+
+ 2ccfc1f4-b2fe-4361-931a-c5294b35edfa
+ True
+
+
+
+
+
+ """
+
+ with self.subTest("default"):
+ with open("/tmp/qubestest.xml", "w", encoding="ascii") as xml_file:
+ xml_file.write(
+ xml_template.format(
+ app_default_dispvm="disp-template",
+ app_management_dispvm="disp-template",
+ adminvm_default_dispvm="disp-template",
+ work_management_dispvm="disp-template",
+ )
+ )
+ app = qubes.Qubes("/tmp/qubestest.xml", offline_mode=True)
+ adminvm = app.domains["dom0"]
+ work = app.domains["work"]
+ disp_template = app.domains["disp-template"]
+
+ self.assertFalse(app.property_is_default("default_dispvm"))
+ self.assertEqual(app.default_dispvm, disp_template)
+ self.assertFalse(app.property_is_default("management_dispvm"))
+ self.assertEqual(app.management_dispvm, disp_template)
+
+ self.assertFalse(adminvm.property_is_default("default_dispvm"))
+ self.assertEqual(adminvm.default_dispvm, disp_template)
+
+ self.assertTrue(work.property_is_default("default_dispvm"))
+ self.assertEqual(work.default_dispvm, disp_template)
+ self.assertFalse(work.property_is_default("management_dispvm"))
+ self.assertEqual(work.management_dispvm, disp_template)
+
+ self.assertTrue(disp_template.property_is_default("default_dispvm"))
+ self.assertEqual(disp_template.default_dispvm, disp_template)
+ self.assertTrue(
+ disp_template.property_is_default("management_dispvm")
+ )
+ self.assertEqual(disp_template.management_dispvm, disp_template)
+
+ app.close()
+ del app
+
+ with self.subTest("invalid-system"):
+ with open("/tmp/qubestest.xml", "w", encoding="ascii") as xml_file:
+ xml_file.write(
+ xml_template.format(
+ app_default_dispvm="work",
+ app_management_dispvm="work",
+ adminvm_default_dispvm="disp-template",
+ work_management_dispvm="disp-template",
+ )
+ )
+ app = qubes.Qubes("/tmp/qubestest.xml", offline_mode=True)
+ adminvm = app.domains["dom0"]
+ work = app.domains["work"]
+ disp_template = app.domains["disp-template"]
+
+ self.assertFalse(app.property_is_default("default_dispvm"))
+ self.assertEqual(app.default_dispvm, None)
+ self.assertFalse(app.property_is_default("management_dispvm"))
+ self.assertEqual(app.management_dispvm, None)
+
+ self.assertFalse(adminvm.property_is_default("default_dispvm"))
+ self.assertEqual(adminvm.default_dispvm, disp_template)
+
+ self.assertTrue(work.property_is_default("default_dispvm"))
+ self.assertEqual(work.default_dispvm, None)
+ self.assertFalse(work.property_is_default("management_dispvm"))
+ self.assertEqual(work.management_dispvm, disp_template)
+
+ self.assertTrue(disp_template.property_is_default("default_dispvm"))
+ self.assertEqual(disp_template.default_dispvm, None)
+ self.assertTrue(
+ disp_template.property_is_default("management_dispvm")
+ )
+ self.assertEqual(disp_template.management_dispvm, None)
+
+ app.close()
+ del app
+
+ with self.subTest("invalid-per-qube"):
+ with open("/tmp/qubestest.xml", "w", encoding="ascii") as xml_file:
+ xml_file.write(
+ xml_template.format(
+ app_default_dispvm="disp-template",
+ app_management_dispvm="disp-template",
+ adminvm_default_dispvm="work",
+ work_management_dispvm="work",
+ )
+ )
+ app = qubes.Qubes("/tmp/qubestest.xml", offline_mode=True)
+ adminvm = app.domains["dom0"]
+ work = app.domains["work"]
+ disp_template = app.domains["disp-template"]
+
+ self.assertFalse(app.property_is_default("default_dispvm"))
+ self.assertEqual(app.default_dispvm, disp_template)
+ self.assertFalse(app.property_is_default("management_dispvm"))
+ self.assertEqual(app.management_dispvm, disp_template)
+
+ self.assertFalse(adminvm.property_is_default("default_dispvm"))
+ self.assertEqual(adminvm.default_dispvm, None)
+
+ self.assertTrue(work.property_is_default("default_dispvm"))
+ self.assertEqual(work.default_dispvm, disp_template)
+ self.assertFalse(work.property_is_default("management_dispvm"))
+ self.assertEqual(work.management_dispvm, None)
+
+ self.assertTrue(disp_template.property_is_default("default_dispvm"))
+ self.assertEqual(disp_template.default_dispvm, disp_template)
+ self.assertTrue(
+ disp_template.property_is_default("management_dispvm")
+ )
+ self.assertEqual(disp_template.management_dispvm, disp_template)
+
+ app.close()
+ del app
+
+ def test_102_property_migrate_label(self):
xml_template = """
@@ -1053,27 +1215,32 @@ def test_203_remove_default_dispvm(self):
appvm = self.app.add_new_vm(
"AppVM", name="test-appvm", template=self.template, label="red"
)
+ appvm.template_for_dispvms = True
self.app.default_dispvm = appvm
with mock.patch.object(self.app, "vmm"):
with self.assertRaises(qubes.exc.QubesVMInUseError):
del self.app.domains[appvm]
- def test_204_remove_appvm_dispvm(self):
- dispvm = self.app.add_new_vm(
- "AppVM", name="test-appvm", template=self.template, label="red"
+ def test_204_remove_appvm_used_as_default_dispvm(self):
+ appvm = self.app.add_new_vm(
+ "AppVM",
+ name="test-appvm",
+ template=self.template,
+ label="red",
+ template_for_dispvms=True,
)
self.app.add_new_vm(
"AppVM",
name="test-appvm2",
template=self.template,
- default_dispvm=dispvm,
+ default_dispvm=appvm,
label="red",
)
with mock.patch.object(self.app, "vmm"):
with self.assertRaises(qubes.exc.QubesVMInUseError):
- del self.app.domains[dispvm]
+ del self.app.domains[appvm]
- def test_205_remove_appvm_dispvm(self):
+ def test_205_remove_appvm_used_as_template_of_dispvm(self):
appvm = self.app.add_new_vm(
"AppVM",
name="test-appvm",
@@ -1257,6 +1424,20 @@ def test_303_preload_default_dispvm_change_partial_noop(self):
"domain-preload-dispvm-start", reason=mock.ANY
)
+ def test_304_event_default_dispvm(self):
+ self.appvm.template_for_dispvms = False
+ with self.assertRaises(qubes.exc.QubesPropertyValueError):
+ self.app.default_dispvm = self.appvm
+ self.appvm.template_for_dispvms = True
+ self.app.default_dispvm = self.appvm
+
+ def test_305_event_management_dispvm(self):
+ self.appvm.template_for_dispvms = False
+ with self.assertRaises(qubes.exc.QubesPropertyValueError):
+ self.app.management_dispvm = self.appvm
+ self.appvm.template_for_dispvms = True
+ self.app.management_dispvm = self.appvm
+
@qubes.tests.skipUnlessGit
def test_900_example_xml_in_doc(self):
path = os.path.join(qubes.tests.in_git, "doc/example.xml")
diff --git a/qubes/tests/vm/adminvm.py b/qubes/tests/vm/adminvm.py
index 3ce26080f..ec269cd11 100644
--- a/qubes/tests/vm/adminvm.py
+++ b/qubes/tests/vm/adminvm.py
@@ -131,7 +131,7 @@ def test_700_run_service(self, mock_subprocess):
"test.service",
"dom0",
"name",
- "dom0"
+ "dom0",
)
mock_subprocess.reset_mock()
@@ -162,7 +162,7 @@ def test_700_run_service(self, mock_subprocess):
"test.service",
self.appvm.name,
"name",
- "dom0"
+ "dom0",
)
@unittest.mock.patch("qubes.vm.adminvm.AdminVM.run_service")
@@ -297,3 +297,13 @@ def test_803_preload_set_delay(self):
for value in cases_valid:
with self.subTest(value=value):
self.vm.features["preload-dispvm-delay"] = value
+
+ def test_901_prop_event_disposable_template(self):
+ appvm = self.app.add_new_vm(
+ "AppVM",
+ name="test-1",
+ template=self.template,
+ label="red",
+ )
+ with self.assertRaises(qubes.exc.QubesPropertyValueError):
+ self.vm.default_dispvm = appvm
diff --git a/qubes/tests/vm/dispvm.py b/qubes/tests/vm/dispvm.py
index 54b3a1f00..f06282437 100644
--- a/qubes/tests/vm/dispvm.py
+++ b/qubes/tests/vm/dispvm.py
@@ -404,6 +404,9 @@ def test_009_dvmtemplate_allowed_change(self):
with mock.patch.object(
self.app, "domains", wraps=self.app.domains
) as mock_domains:
+ # pylint: disable=attribute-defined-outside-init
+ self.app.property_list = mock.Mock()
+ self.app.property_list.return_value = []
mock_domains.configure_mock(
**{
"get_new_unused_dispid": mock.Mock(return_value=42),
@@ -725,3 +728,29 @@ def test_024_is_preload_outdated(self, _mock_makedirs, _mock_symlink):
sorted(dispvm.is_preload_outdated()["properties"]),
sorted(["netvm", "dns", "visible_netmask"]),
)
+
+ def test_030_set_disposable_template(self):
+ self.appvm.template_for_dispvms = True
+ self.appvm_alt.template_for_dispvms = False
+ orig_getitem = self.app.domains.__getitem__
+ with mock.patch.object(
+ self.app, "domains", wraps=self.app.domains
+ ) as mock_domains:
+ mock_domains.configure_mock(
+ **{
+ "get_new_unused_dispid": mock.Mock(return_value=42),
+ "__getitem__.side_effect": orig_getitem,
+ }
+ )
+ dispvm = self.app.add_new_vm(
+ qubes.vm.dispvm.DispVM,
+ name="test-dispvm",
+ template=self.appvm,
+ )
+ with self.assertRaises(qubes.exc.QubesPropertyValueError):
+ dispvm.default_dispvm = self.appvm_alt
+ with self.assertRaises(qubes.exc.QubesPropertyValueError):
+ dispvm.management_dispvm = self.appvm_alt
+ self.appvm_alt.template_for_dispvms = True
+ dispvm.default_dispvm = self.appvm_alt
+ dispvm.management_dispvm = self.appvm_alt
diff --git a/qubes/tests/vm/mix/dvmtemplate.py b/qubes/tests/vm/mix/dvmtemplate.py
index cebc5ab72..7c581c693 100755
--- a/qubes/tests/vm/mix/dvmtemplate.py
+++ b/qubes/tests/vm/mix/dvmtemplate.py
@@ -19,6 +19,8 @@
# License along with this library; if not, see .
#
+import os
+import shutil
from unittest import mock
import qubes
@@ -55,17 +57,41 @@ class TC_00_DVMTemplateMixin(
):
def setUp(self):
super().setUp()
- self.app = TestApp()
- self.app.save = mock.Mock()
- self.app.pools["default"] = qubes.tests.vm.appvm.TestPool(
- name="default"
+ self.test_base_dir = "/tmp/qubes-test-dir"
+ self.base_dir_patch = mock.patch.dict(
+ qubes.config.system_path, {"qubes_base_dir": self.test_base_dir}
+ )
+ self.base_dir_patch2 = mock.patch(
+ "qubes.config.qubes_base_dir", self.test_base_dir
+ )
+ self.base_dir_patch3 = mock.patch.dict(
+ qubes.config.defaults["pool_configs"]["varlibqubes"],
+ {"dir_path": self.test_base_dir},
)
- self.app.pools["linux-kernel"] = qubes.tests.vm.appvm.TestPool(
- name="linux-kernel"
+ self.skip_kernel_validation_patch = mock.patch(
+ "qubes.app.validate_kernel", lambda obj, key, value: None
)
+ self.base_dir_patch.start()
+ self.base_dir_patch2.start()
+ self.base_dir_patch3.start()
+ self.skip_kernel_validation_patch.start()
+ self.app = qubes.Qubes("/tmp/qubes-test.xml", load=False)
+ self.app.vmm = mock.Mock(spec=qubes.app.VMMConnection)
+ self.app.load_initial_values()
+ self.loop.run_until_complete(self.app.setup_pools())
+ self.app.default_kernel = "1.0"
+ self.app.default_netvm = None
+ with qubes.tests.substitute_entry_points(
+ "qubes.storage", "qubes.tests.storage"
+ ):
+ self.loop.run_until_complete(
+ self.app.add_pool("test", driver="test")
+ )
+ self.app.default_pool = "varlibqubes"
+
+ self.app.save = mock.Mock()
self.app.vmm.offline_mode = True
- self.adminvm = self.app.add_new_vm(qubes.vm.adminvm.AdminVM)
- self.addCleanup(self.cleanup_adminvm)
+ self.adminvm = self.app.domains[0]
self.template = self.app.add_new_vm(
qubes.vm.templatevm.TemplateVM, name="test-template", label="red"
)
@@ -85,36 +111,36 @@ def setUp(self):
self.appvm.features["gui"] = False
self.appvm.features["supported-rpc.qubes.WaitForRunningSystem"] = True
self.appvm.features["supported-rpc.qubes.WaitForSession"] = True
- self.app.domains[self.appvm.name] = self.appvm
- self.app.domains[self.appvm] = self.appvm
self.app.default_dispvm = self.appvm
self.addCleanup(self.cleanup_dispvm)
self.emitter = qubes.tests.TestEmitter()
def tearDown(self):
+ try:
+ os.unlink("/tmp/qubestest.xml")
+ except: # pylint: disable=bare-except
+ pass
+ self.base_dir_patch3.stop()
+ self.base_dir_patch2.stop()
+ self.base_dir_patch.stop()
+ self.skip_kernel_validation_patch.stop()
+ if os.path.exists(self.test_base_dir):
+ shutil.rmtree(self.test_base_dir)
self.app.default_dispvm = None
+ del self.appvm
+ del self.template
+ del self.template_alt
+ del self.adminvm
+ self.app.close()
+ del self.app
del self.emitter
super().tearDown()
- def cleanup_adminvm(self):
- self.adminvm.close()
- del self.adminvm
-
def cleanup_dispvm(self):
if hasattr(self, "dispvm"):
- self.dispvm.close()
del self.dispvm
if hasattr(self, "dispvm_alt"):
- self.dispvm_alt.close()
del self.dispvm_alt
- self.template.close()
- self.template_alt.close()
- self.appvm.close()
- del self.template
- del self.template_alt
- del self.appvm
- self.app.domains.clear()
- self.app.pools.clear()
async def mock_coro(self, *args, **kwargs):
pass
@@ -290,6 +316,7 @@ def test_012_dvm_preload_set_max(self, mock_events):
)
mock_events.reset_mock()
+ self.app.default_dispvm = None
self.appvm.template_for_dispvms = False
self.appvm.features["preload-dispvm-max"] = "2"
mock_events.assert_not_called()
@@ -388,6 +415,8 @@ def test_040_dvm_preload_set_template_for_dispvms(
self, mock_remove, mock_events
):
# Remove preloads when disabling property.
+ self.app.default_dispvm = None
+
mock_events.side_effect = self.mock_coro
self.appvm.template_for_dispvms = False
mock_events.assert_not_called()
@@ -456,10 +485,59 @@ def test_040_dvm_preload_set_template_for_dispvms(
)
mock_events.reset_mock()
mock_remove.reset_mock()
+ self.dispvm.default_dispvm = None
+ self.dispvm_alt.default_dispvm = None
del self.appvm.template_for_dispvms
mock_remove.assert_called_once_with(0, reason=mock.ANY)
mock_events.assert_not_called()
+ def test_050_dvm_del_template_for_dispvms_used_by_system(self):
+ for prop in ["default_dispvm", "management_dispvm"]:
+ with self.subTest(prop=prop):
+ self.appvm.template_for_dispvms = True
+ setattr(self.app, prop, self.appvm)
+ with self.assertRaises(qubes.exc.QubesVMInUseError):
+ self.appvm.template_for_dispvms = False
+ setattr(self.app, prop, None)
+ self.appvm.template_for_dispvms = False
+
+ def test_051_dvm_del_template_for_dispvms_from_self(self):
+ self.app.default_dispvm = None
+ self.app.management_dispvm = None
+ for prop in ["default_dispvm", "management_dispvm"]:
+ with self.subTest(prop=prop):
+ self.appvm.template_for_dispvms = True
+ setattr(self.appvm, prop, self.appvm)
+ self.appvm.template_for_dispvms = False
+ setattr(self.appvm, prop, None)
+
+ def test_052_dvm_del_template_for_dispvms_used_by_others(self):
+ self.app.default_dispvm = None
+ self.app.management_dispvm = None
+ for prop in ["default_dispvm", "management_dispvm"]:
+ with self.subTest(prop=prop):
+ self.appvm.template_for_dispvms = True
+ setattr(self.template, prop, self.appvm)
+ with self.assertRaises(qubes.exc.QubesVMInUseError):
+ self.appvm.template_for_dispvms = False
+ setattr(self.template, prop, None)
+ self.appvm.template_for_dispvms = False
+ with self.subTest(prop="template"):
+ self.appvm.template_for_dispvms = True
+ self.dispvm = self.app.add_new_vm(
+ qubes.vm.dispvm.DispVM,
+ name="test-dispvm",
+ template=self.appvm,
+ label="red",
+ dispid=42,
+ )
+ for remove_prop in ["default_dispvm", "management_dispvm"]:
+ setattr(self.dispvm, remove_prop, None)
+ with self.assertRaises(qubes.exc.QubesVMInUseError):
+ self.appvm.template_for_dispvms = False
+ del self.app.domains[self.dispvm]
+ self.appvm.template_for_dispvms = False
+
def test_100_get_preload_templates(self):
print(qubes.vm.dispvm.get_preload_templates(self.app))
self.appvm.features["supported-rpc.qubes.WaitForRunningSystem"] = True
diff --git a/qubes/tests/vm/qubesvm.py b/qubes/tests/vm/qubesvm.py
index 816139488..834665b0c 100644
--- a/qubes/tests/vm/qubesvm.py
+++ b/qubes/tests/vm/qubesvm.py
@@ -794,7 +794,9 @@ def test_281_autostart_systemd(self):
def test_290_management_dispvm(self):
vm = self.get_vm()
- vm2 = self.get_vm("test2", qid=2)
+ vm2 = self.get_vm(
+ "test2", qid=2, cls=qubes.vm.appvm.AppVM, template_for_dispvms=True
+ )
self.app.management_dispvm = None
self.assertPropertyDefaultValue(vm, "management_dispvm", None)
self.app.management_dispvm = vm
@@ -809,7 +811,9 @@ def test_290_management_dispvm(self):
def test_291_management_dispvm_template_based(self):
tpl = self.get_vm(name="tpl", cls=qubes.vm.templatevm.TemplateVM)
vm = self.get_vm(cls=qubes.vm.appvm.AppVM, template=tpl, qid=2)
- vm2 = self.get_vm("test2", qid=3)
+ vm2 = self.get_vm(
+ "test2", qid=3, cls=qubes.vm.appvm.AppVM, template_for_dispvms=True
+ )
del vm.volumes
self.app.management_dispvm = None
try:
@@ -824,6 +828,22 @@ def test_291_management_dispvm_template_based(self):
finally:
self.app.management_dispvm = None
+ def test_292_management_dispvm_setter_without_template_for_dispvms(self):
+ vm = self.get_vm()
+ kwargs = {"name": "test2", "qid": 2, "management_dispvm": vm}
+ with self.assertRaises(qubes.exc.QubesPropertyValueError):
+ self.get_vm(**kwargs)
+ vm.template_for_dispvms = True
+ self.get_vm(**kwargs)
+
+ def test_293_default_dispvm_setter_without_template_for_dispvms(self):
+ vm = self.get_vm()
+ kwargs = {"name": "test2", "qid": 2, "default_dispvm": vm}
+ with self.assertRaises(qubes.exc.QubesPropertyValueError):
+ self.get_vm(**kwargs)
+ vm.template_for_dispvms = True
+ self.get_vm(**kwargs)
+
@unittest.skip("TODO")
def test_320_seamless_gui_mode(self):
vm = self.get_vm()
diff --git a/qubes/vm/__init__.py b/qubes/vm/__init__.py
index adfa97929..56992d47a 100644
--- a/qubes/vm/__init__.py
+++ b/qubes/vm/__init__.py
@@ -117,6 +117,36 @@ def _setter_qid(self, prop, value):
return value
+def validate_disposable_template(obj, prop, value) -> None:
+ """Helper function to validate change of disposable template settings,
+ such as default_dispvm and management_dispvm."""
+ if not value:
+ return
+ assert isinstance(value, qubes.vm.qubesvm.QubesVM)
+ if not getattr(value, "template_for_dispvms", None):
+ if isinstance(obj, qubes.app.Qubes):
+ target = "system"
+ else:
+ target = "qube {!r}".format(obj.name)
+ msg = (
+ "Cannot set invalid property to {!s}, {!s} {!r} has "
+ "template_for_dispvms set to False".format(
+ target, str(prop), value.name
+ )
+ )
+ raise qubes.exc.QubesPropertyValueError(
+ obj,
+ obj.property_get_def(prop),
+ value,
+ msg,
+ )
+
+
+def setter_disposable_template(obj, prop, value):
+ validate_disposable_template(obj, prop, value)
+ return value
+
+
class Tags(set):
"""Manager of the tags.
diff --git a/qubes/vm/adminvm.py b/qubes/vm/adminvm.py
index 518b916a6..30b5005b4 100644
--- a/qubes/vm/adminvm.py
+++ b/qubes/vm/adminvm.py
@@ -58,7 +58,9 @@ class AdminVM(LocalVM):
load_stage=4,
allow_none=True,
default=(lambda self: self.app.default_dispvm),
- doc="Default VM to be used as Disposable VM for service calls.",
+ setter=qubes.vm.setter_disposable_template,
+ doc="""Default disposable template to be used for spawning disposable
+ qubes for service calls.""",
)
include_in_backups = qubes.property(
diff --git a/qubes/vm/dispvm.py b/qubes/vm/dispvm.py
index a0e4e3df5..14b9d5f15 100644
--- a/qubes/vm/dispvm.py
+++ b/qubes/vm/dispvm.py
@@ -281,7 +281,9 @@ class DispVM(qubes.vm.qubesvm.QubesVM):
load_stage=4,
allow_none=True,
default=(lambda self: self.template),
- doc="Default disposable template to be used for service calls.",
+ setter=qubes.vm.setter_disposable_template,
+ doc="""Default disposable template to be used for spawning disposable
+ qubes for service calls.""",
)
default_volume_config = {
diff --git a/qubes/vm/mix/dvmtemplate.py b/qubes/vm/mix/dvmtemplate.py
index 629bfe213..e5abfdc55 100644
--- a/qubes/vm/mix/dvmtemplate.py
+++ b/qubes/vm/mix/dvmtemplate.py
@@ -343,13 +343,17 @@ def on_feature_set_preload_dispvm(
qube = self.app.domains[qube]
qube.fire_event("property-reset:is_preload", name="is_preload")
- @qubes.events.handler("property-pre-set:template_for_dispvms")
+ @qubes.events.handler(
+ "property-pre-set:template_for_dispvms",
+ "property-pre-reset:template_for_dispvms",
+ )
def __on_pre_set_dvmtemplate(
- self, event, name, newvalue, oldvalue=None
+ self, event, name, newvalue=None, oldvalue=None
) -> None:
"""
Forbid disabling ``template_for_dispvms`` while there are disposables
- running.
+ running or it is set a system or per qube disposable template property,
+ normally ``default_dispvm`` or ``management_dispvm``.
:param str event: Event which was fired.
:param str name: Property name.
@@ -362,32 +366,35 @@ def __on_pre_set_dvmtemplate(
return
if not newvalue and not oldvalue:
return
- dependencies = [
- disp.name for disp in self.dispvms if not disp.is_preload
- ]
- if dependencies:
+ system_props = ["default_dispvm", "management_dispvm"]
+ qube_props = ["default_dispvm", "management_dispvm", "template"]
+ system_deps, qube_deps = qubes.app.get_qube_prop_deps(
+ qube=self,
+ system_properties=system_props,
+ qube_properties=qube_props,
+ )
+ if system_deps:
msg = (
- "Cannot change template_for_dispvms to False while there are "
- "some disposables based on this disposable template",
+ "Cannot change template_for_dispvms to False while it is in use"
+ " by the system by any of these properties: %s"
+ % (", ".join(system_props))
+ )
+ self.log.error("%s", msg)
+ raise qubes.exc.QubesVMInUseError(self, msg)
+ if qube_deps:
+ msg = (
+ "Cannot change template_for_dispvms to False while it is the "
+ "disposable template of a qube by any of these properties: %s"
+ % (", ".join(qube_props))
+ )
+ self.log.error(
+ "%s: %s", msg, ", ".join(":".join(i) for i in qube_deps)
)
- self.log.error("%s: %s", msg, ", ".join(dependencies))
raise qubes.exc.QubesVMInUseError(self, msg)
self.remove_preload_excess(
0, reason="template_for_dispvms was set to False"
)
- @qubes.events.handler("property-pre-del:template_for_dispvms")
- def __on_pre_del_dvmtemplate(self, event, name, oldvalue=None) -> None:
- """
- Forbid disabling ``template_for_dispvms`` while there are disposables
- running.
-
- :param str event: Event which was fired.
- :param str name: Property name.
- :param bool oldvalue: Old value of the property.
- """
- self.__on_pre_set_dvmtemplate(event, name, False, oldvalue)
-
@qubes.events.handler("property-set:template_for_dispvms")
def __on_set_dvmtemplate(self, event, name, newvalue, oldvalue=None):
# pylint: disable=unused-argument
diff --git a/qubes/vm/qubesvm.py b/qubes/vm/qubesvm.py
index ee0124120..23b9a404c 100644
--- a/qubes/vm/qubesvm.py
+++ b/qubes/vm/qubesvm.py
@@ -913,7 +913,9 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.LocalVM):
load_stage=4,
allow_none=True,
default=(lambda self: self.app.default_dispvm),
- doc="Default VM to be used as Disposable VM for service calls.",
+ setter=qubes.vm.setter_disposable_template,
+ doc="""Default disposable template to be used for spawning disposable
+ qubes for service calls.""",
)
management_dispvm = qubes.VMProperty(
@@ -923,7 +925,9 @@ class QubesVM(qubes.vm.mix.net.NetVMMixin, qubes.vm.LocalVM):
default=_default_with_template(
"management_dispvm", (lambda self: self.app.management_dispvm)
),
- doc="Default DVM template for Disposable VM for managing this VM.",
+ setter=qubes.vm.setter_disposable_template,
+ doc="""Default disposable template to be used for spawning disposable
+ qubes for managing this qube.""",
)
updateable = qubes.property(