diff --git a/qubes/api/admin.py b/qubes/api/admin.py index be596696..9172d8e4 100644 --- a/qubes/api/admin.py +++ b/qubes/api/admin.py @@ -291,6 +291,24 @@ class QubesAdminAPI(qubes.api.AbstractQubesAPI): 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): diff --git a/qubes/tests/api_admin.py b/qubes/tests/api_admin.py index 36c3d8ac..6a18e221 100644 --- a/qubes/tests/api_admin.py +++ b/qubes/tests/api_admin.py @@ -1598,6 +1598,95 @@ class TC_00_VMs(AdminAPITestCase): 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',