admin: add admin.vm.volume.Clone

QubesOS/qubes-issues#2622

qubes/api/admin.py

@@ -291,6 +291,24 @@ def vm_volume_revert(self, untrusted_payload):
         self.dest.storage.get_pool(volume).revert(revision)
         self.app.save()
 
+    @qubes.api.method('admin.vm.volume.Clone')
+    @asyncio.coroutine
+    def vm_volume_clone(self, untrusted_payload):
+        assert self.arg in self.dest.volumes.keys()
+        untrusted_target = untrusted_payload.decode('ascii').strip()
+        del untrusted_payload
+        qubes.vm.validate_name(None, None, untrusted_target)
+        target_vm = self.app.domains[untrusted_target]
+        del untrusted_target
+        assert self.arg in target_vm.volumes.keys()
+
+        volume = self.dest.volumes[self.arg]
+
+        self.fire_event_for_permission(target_vm=target_vm, volume=volume)
+
+        yield from target_vm.storage.clone_volume(self.dest, self.arg)
+        self.app.save()
+
     @qubes.api.method('admin.vm.volume.Resize')
     @asyncio.coroutine
     def vm_volume_resize(self, untrusted_payload):

qubes/tests/api_admin.py

@@ -1598,6 +1598,95 @@ def test_511_vm_volume_import_running(self):
                 self.call_mgmt_func(b'admin.vm.volume.Import', b'test-vm1',
                     b'private')
 
+    def test_520_vm_volume_clone(self):
+        self.vm2 = self.app.add_new_vm('AppVM', label='red', name='test-vm2',
+            template='test-template')
+        self.vm.volumes = unittest.mock.MagicMock()
+        self.vm2.volumes = unittest.mock.MagicMock()
+        volumes_conf = {
+            'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
+        }
+        self.vm.volumes.configure_mock(**volumes_conf)
+        self.vm2.volumes.configure_mock(**volumes_conf)
+        self.vm2.storage = unittest.mock.Mock()
+        func_mock = unittest.mock.Mock()
+
+        @asyncio.coroutine
+        def coroutine_mock(*args, **kwargs):
+            return func_mock(*args, **kwargs)
+        self.vm2.storage.clone_volume = coroutine_mock
+
+        self.call_mgmt_func(b'admin.vm.volume.Clone',
+                b'test-vm1', b'private', b'test-vm2')
+        self.assertEqual(self.vm.volumes.mock_calls,
+            [('keys', (), {}),
+             ('__getitem__', ('private', ), {}),
+             ('__getitem__().__hash__', (), {}),
+            ])
+        self.assertEqual(self.vm2.volumes.mock_calls,
+            [unittest.mock.call.keys()])
+        self.assertEqual(self.vm2.storage.mock_calls, [])
+        self.assertEqual(func_mock.mock_calls, [
+            unittest.mock.call(self.vm, 'private')
+        ])
+        self.app.save.assert_called_once_with()
+
+    def test_521_vm_volume_clone_invalid_volume(self):
+        self.vm2 = self.app.add_new_vm('AppVM', label='red', name='test-vm2',
+            template='test-template')
+        self.vm.volumes = unittest.mock.MagicMock()
+        self.vm2.volumes = unittest.mock.MagicMock()
+        volumes_conf = {
+            'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
+        }
+        self.vm.volumes.configure_mock(**volumes_conf)
+        self.vm2.volumes.configure_mock(**volumes_conf)
+        self.vm2.storage = unittest.mock.Mock()
+        func_mock = unittest.mock.Mock()
+
+        @asyncio.coroutine
+        def coroutine_mock(*args, **kwargs):
+            return func_mock(*args, **kwargs)
+        self.vm2.storage.clone_volume = coroutine_mock
+
+        with self.assertRaises(AssertionError):
+            self.call_mgmt_func(b'admin.vm.volume.Clone',
+                    b'test-vm1', b'private123', b'test-vm2')
+        self.assertEqual(self.vm.volumes.mock_calls,
+            [('keys', (), {})])
+        self.assertEqual(self.vm2.volumes.mock_calls, [])
+        self.assertEqual(self.vm2.storage.mock_calls, [])
+        self.assertEqual(func_mock.mock_calls, [])
+        self.assertFalse(self.app.save.called)
+
+    def test_522_vm_volume_clone_invalid_vm(self):
+        self.vm2 = self.app.add_new_vm('AppVM', label='red', name='test-vm2',
+            template='test-template')
+        self.vm.volumes = unittest.mock.MagicMock()
+        self.vm2.volumes = unittest.mock.MagicMock()
+        volumes_conf = {
+            'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
+        }
+        self.vm.volumes.configure_mock(**volumes_conf)
+        self.vm2.volumes.configure_mock(**volumes_conf)
+        self.vm2.storage = unittest.mock.Mock()
+        func_mock = unittest.mock.Mock()
+
+        @asyncio.coroutine
+        def coroutine_mock(*args, **kwargs):
+            return func_mock(*args, **kwargs)
+        self.vm2.storage.clone_volume = coroutine_mock
+
+        with self.assertRaises(AssertionError):
+            self.call_mgmt_func(b'admin.vm.volume.Clone',
+                    b'test-vm1', b'private123', b'no-such-vm')
+        self.assertEqual(self.vm.volumes.mock_calls,
+            [('keys', (), {})])
+        self.assertEqual(self.vm2.volumes.mock_calls, [])
+        self.assertEqual(self.vm2.storage.mock_calls, [])
+        self.assertEqual(func_mock.mock_calls, [])
+        self.assertFalse(self.app.save.called)
+
     def test_990_vm_unexpected_payload(self):
         methods_with_no_payload = [
             b'admin.vm.List',