Merge remote-tracking branch 'origin/pr/308'

* origin/pr/308:
  Move devices check to on_domain_pre_deleted
  Prevent removing VM if it provides devices in persistent mode

qubes/app.py

@@ -1462,6 +1462,15 @@ class Qubes(qubes.PropertyHolder):
                 except AttributeError:
                     pass
 
+        assignments = vm.get_provided_assignments()
+        if assignments:
+            desc = ', '.join(
+                assignment.ident for assignment in assignments)
+            raise qubes.exc.QubesVMInUseError(
+                vm,
+                'VM has devices attached persistently to other VMs: ' +
+                desc)
+
     @qubes.events.handler('domain-delete')
     def on_domain_deleted(self, event, vm):
         # pylint: disable=unused-argument

qubes/tests/api_admin.py

@@ -1,4 +1,4 @@
-# -*- encoding: utf8 -*-
+# -*- encoding: utf-8 -*-
 #
 # The Qubes OS Project, http://www.qubes-os.org
 #
@@ -1747,6 +1747,22 @@ netvm default=True type=vm
         self.assertFalse(mock_remove.called)
         self.assertFalse(self.app.save.called)
 
+    @unittest.mock.patch('qubes.storage.Storage.remove')
+    @unittest.mock.patch('shutil.rmtree')
+    def test_502_vm_remove_attached(self, mock_rmtree, mock_remove):
+        self.setup_for_clone()
+        assignment = qubes.devices.DeviceAssignment(
+            self.vm, '1234', persistent=True)
+        self.loop.run_until_complete(
+            self.vm2.devices['testclass'].attach(assignment))
+
+        mock_remove.side_effect = self.dummy_coro
+        with self.assertRaises(qubes.exc.QubesVMInUseError):
+            self.call_mgmt_func(b'admin.vm.Remove', b'test-vm1')
+        self.assertFalse(mock_rmtree.called)
+        self.assertFalse(mock_remove.called)
+        self.assertFalse(self.app.save.called)
+
     def test_510_vm_volume_import(self):
         value = self.call_mgmt_func(b'admin.vm.volume.Import', b'test-vm1',
             b'private')

qubes/tests/app.py

@@ -673,6 +673,15 @@ class TC_90_Qubes(qubes.tests.QubesTestCase):
             with self.assertRaises(qubes.exc.QubesVMInUseError):
                 del self.app.domains[appvm]
 
+    def test_206_remove_attached(self):
+        # See also qubes.tests.api_admin.
+        vm = self.app.add_new_vm(
+            'AppVM', name='test-vm', template=self.template, label='red')
+        assignment = mock.Mock(ident='1234')
+        vm.get_provided_assignments = lambda: [assignment]
+        with self.assertRaises(qubes.exc.QubesVMInUseError):
+            del self.app.domains[vm]
+
     @qubes.tests.skipUnlessGit
     def test_900_example_xml_in_doc(self):
         self.assertXMLIsValid(

qubes/vm/__init__.py

@@ -289,6 +289,19 @@ class BaseVM(qubes.PropertyHolder):
 
         # SEE:1815 firewall, policy.
 
+    def get_provided_assignments(self):
+        '''List of persistent device assignments from this VM.'''
+
+        assignments = []
+        for domain in self.app.domains:
+            if domain == self:
+                continue
+            for device_collection in domain.devices.values():
+                for assignment in device_collection.persistent():
+                    if assignment.backend_domain == self:
+                        assignments.append(assignment)
+        return assignments
+
     def init_log(self):
         '''Initialise logger for this domain.'''
         self.log = qubes.log.get_vm_logger(self.name)