dispvm.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323
  1. # -*- encoding: utf-8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2017 Marek Marczykowski-Górecki
  6. # <marmarek@invisiblethingslab.com>
  7. #
  8. # This library is free software; you can redistribute it and/or
  9. # modify it under the terms of the GNU Lesser General Public
  10. # License as published by the Free Software Foundation; either
  11. # version 2.1 of the License, or (at your option) any later version.
  12. #
  13. # This library is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16. # Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public
  19. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  20. import unittest.mock as mock
  21. import asyncio
  22. import qubes.vm.dispvm
  23. import qubes.vm.appvm
  24. import qubes.vm.templatevm
  25. import qubes.tests
  26. import qubes.tests.vm
  27. import qubes.tests.vm.appvm
  28. class TestApp(qubes.tests.vm.TestApp):
  29. def __init__(self):
  30. super(TestApp, self).__init__()
  31. self.qid_counter = 1
  32. def add_new_vm(self, cls, **kwargs):
  33. qid = self.qid_counter
  34. self.qid_counter += 1
  35. vm = cls(self, None, qid=qid, **kwargs)
  36. self.domains[vm.name] = vm
  37. self.domains[vm] = vm
  38. return vm
  39. class TC_00_DispVM(qubes.tests.QubesTestCase):
  40. def setUp(self):
  41. super(TC_00_DispVM, self).setUp()
  42. self.app = TestApp()
  43. self.app.save = mock.Mock()
  44. self.app.pools['default'] = qubes.tests.vm.appvm.TestPool(name='default')
  45. self.app.pools['linux-kernel'] = mock.Mock(**{
  46. 'init_volume.return_value.pool': 'linux-kernel'})
  47. self.app.vmm.offline_mode = True
  48. self.template = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM,
  49. name='test-template', label='red')
  50. self.appvm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  51. name='test-vm', template=self.template, label='red')
  52. self.app.domains[self.appvm.name] = self.appvm
  53. self.app.domains[self.appvm] = self.appvm
  54. self.addCleanup(self.cleanup_dispvm)
  55. def cleanup_dispvm(self):
  56. if hasattr(self, 'dispvm'):
  57. self.dispvm.close()
  58. del self.dispvm
  59. self.template.close()
  60. self.appvm.close()
  61. del self.template
  62. del self.appvm
  63. self.app.domains.clear()
  64. self.app.pools.clear()
  65. @asyncio.coroutine
  66. def mock_coro(self, *args, **kwargs):
  67. pass
  68. @mock.patch('os.symlink')
  69. @mock.patch('os.makedirs')
  70. @mock.patch('qubes.storage.Storage')
  71. def test_000_from_appvm(self, mock_storage, mock_makedirs, mock_symlink):
  72. mock_storage.return_value.create.side_effect = self.mock_coro
  73. self.appvm.template_for_dispvms = True
  74. orig_getitem = self.app.domains.__getitem__
  75. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  76. as mock_domains:
  77. mock_domains.configure_mock(**{
  78. 'get_new_unused_dispid': mock.Mock(return_value=42),
  79. '__getitem__.side_effect': orig_getitem
  80. })
  81. dispvm = self.loop.run_until_complete(
  82. qubes.vm.dispvm.DispVM.from_appvm(self.appvm))
  83. mock_domains.get_new_unused_dispid.assert_called_once_with()
  84. self.assertEqual(dispvm.name, 'disp42')
  85. self.assertEqual(dispvm.template, self.appvm)
  86. self.assertEqual(dispvm.label, self.appvm.label)
  87. self.assertEqual(dispvm.label, self.appvm.label)
  88. self.assertEqual(dispvm.auto_cleanup, True)
  89. mock_makedirs.assert_called_once_with(
  90. '/var/lib/qubes/appvms/' + dispvm.name, mode=0o775, exist_ok=True)
  91. mock_symlink.assert_not_called()
  92. def test_001_from_appvm_reject_not_allowed(self):
  93. with self.assertRaises(qubes.exc.QubesException):
  94. dispvm = self.loop.run_until_complete(
  95. qubes.vm.dispvm.DispVM.from_appvm(self.appvm))
  96. def test_002_template_change(self):
  97. self.appvm.template_for_dispvms = True
  98. orig_getitem = self.app.domains.__getitem__
  99. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  100. as mock_domains:
  101. mock_domains.configure_mock(**{
  102. 'get_new_unused_dispid': mock.Mock(return_value=42),
  103. '__getitem__.side_effect': orig_getitem
  104. })
  105. self.dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
  106. name='test-dispvm', template=self.appvm)
  107. self.dispvm.template = self.appvm
  108. self.dispvm.start()
  109. if not self.app.vmm.offline_mode:
  110. assert not dispvm.is_halted()
  111. with self.assertRaises(qubes.exc.QubesVMNotHaltedError):
  112. self.dispvm.template = self.appvm
  113. with self.assertRaises(qubes.exc.QubesValueError):
  114. self.dispvm.template = qubes.property.DEFAULT
  115. self.dispvm.kill()
  116. self.dispvm.template = self.appvm
  117. def test_003_dvmtemplate_template_change(self):
  118. self.appvm.template_for_dispvms = True
  119. orig_domains = self.app.domains
  120. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  121. as mock_domains:
  122. mock_domains.configure_mock(**{
  123. 'get_new_unused_dispid': mock.Mock(return_value=42),
  124. '__getitem__.side_effect': orig_domains.__getitem__,
  125. '__iter__.side_effect': orig_domains.__iter__,
  126. '__setitem__.side_effect': orig_domains.__setitem__,
  127. })
  128. self.dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
  129. name='test-dispvm', template=self.appvm)
  130. self.appvm.template = self.template
  131. with self.assertRaises(qubes.exc.QubesValueError):
  132. self.appvm.template = qubes.property.DEFAULT
  133. def test_004_dvmtemplate_allowed_change(self):
  134. self.appvm.template_for_dispvms = True
  135. orig_domains = self.app.domains
  136. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  137. as mock_domains:
  138. mock_domains.configure_mock(**{
  139. 'get_new_unused_dispid': mock.Mock(return_value=42),
  140. '__getitem__.side_effect': orig_domains.__getitem__,
  141. '__iter__.side_effect': orig_domains.__iter__,
  142. '__setitem__.side_effect': orig_domains.__setitem__,
  143. })
  144. self.dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
  145. name='test-dispvm', template=self.appvm)
  146. with self.assertRaises(qubes.exc.QubesVMInUseError):
  147. self.appvm.template_for_dispvms = False
  148. def test_010_create_direct(self):
  149. self.appvm.template_for_dispvms = True
  150. orig_getitem = self.app.domains.__getitem__
  151. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  152. as mock_domains:
  153. mock_domains.configure_mock(**{
  154. 'get_new_unused_dispid': mock.Mock(return_value=42),
  155. '__getitem__.side_effect': orig_getitem
  156. })
  157. self.dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
  158. name='test-dispvm', template=self.appvm)
  159. mock_domains.get_new_unused_dispid.assert_called_once_with()
  160. dispvm=self.dispvm
  161. self.assertEqual(dispvm.name, 'test-dispvm')
  162. self.assertEqual(dispvm.template, self.appvm)
  163. self.assertEqual(dispvm.label, self.appvm.label)
  164. self.assertEqual(dispvm.label, self.appvm.label)
  165. self.assertEqual(dispvm.auto_cleanup, False)
  166. def test_011_create_direct_generate_name(self):
  167. self.appvm.template_for_dispvms = True
  168. orig_getitem = self.app.domains.__getitem__
  169. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  170. as mock_domains:
  171. mock_domains.configure_mock(**{
  172. 'get_new_unused_dispid': mock.Mock(return_value=42),
  173. '__getitem__.side_effect': orig_getitem
  174. })
  175. dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
  176. template=self.appvm)
  177. mock_domains.get_new_unused_dispid.assert_called_once_with()
  178. self.assertEqual(dispvm.name, 'disp42')
  179. self.assertEqual(dispvm.template, self.appvm)
  180. self.assertEqual(dispvm.label, self.appvm.label)
  181. self.assertEqual(dispvm.auto_cleanup, False)
  182. def test_011_create_direct_reject(self):
  183. orig_getitem = self.app.domains.__getitem__
  184. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  185. as mock_domains:
  186. mock_domains.configure_mock(**{
  187. 'get_new_unused_dispid': mock.Mock(return_value=42),
  188. '__getitem__.side_effect': orig_getitem
  189. })
  190. with self.assertRaises(qubes.exc.QubesException):
  191. self.app.add_new_vm(qubes.vm.dispvm.DispVM,
  192. name='test-dispvm', template=self.appvm)
  193. self.assertFalse(mock_domains.get_new_unused_dispid.called)
  194. @mock.patch('os.symlink')
  195. @mock.patch('os.makedirs')
  196. def test_020_copy_storage_pool(self, mock_makedirs, mock_symlink):
  197. self.app.pools['alternative'] = qubes.tests.vm.appvm.TestPool(name='alternative')
  198. self.appvm.template_for_dispvms = True
  199. self.loop.run_until_complete(self.template.create_on_disk())
  200. self.loop.run_until_complete(self.appvm.create_on_disk(pool='alternative'))
  201. orig_getitem = self.app.domains.__getitem__
  202. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  203. as mock_domains:
  204. mock_domains.configure_mock(**{
  205. 'get_new_unused_dispid': mock.Mock(return_value=42),
  206. '__getitem__.side_effect': orig_getitem
  207. })
  208. dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
  209. name='test-dispvm', template=self.appvm)
  210. self.loop.run_until_complete(dispvm.create_on_disk())
  211. self.assertIs(dispvm.template, self.appvm)
  212. self.assertIs(dispvm.volumes['private'].pool,
  213. self.appvm.volumes['private'].pool)
  214. self.assertIs(dispvm.volumes['root'].pool,
  215. self.appvm.volumes['root'].pool)
  216. self.assertIs(dispvm.volumes['volatile'].pool,
  217. self.appvm.volumes['volatile'].pool)
  218. def test_021_storage_template_change(self):
  219. self.appvm.template_for_dispvms = True
  220. orig_domains = self.app.domains
  221. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  222. as mock_domains:
  223. mock_domains.configure_mock(**{
  224. 'get_new_unused_dispid': mock.Mock(return_value=42),
  225. '__getitem__.side_effect': orig_domains.__getitem__,
  226. '__iter__.side_effect': orig_domains.__iter__,
  227. '__setitem__.side_effect': orig_domains.__setitem__,
  228. })
  229. vm = self.dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
  230. name='test-dispvm', template=self.appvm)
  231. self.loop.run_until_complete(vm.create_on_disk())
  232. self.assertIs(vm.volume_config['root']['source'],
  233. self.template.volumes['root'])
  234. # create new mock, so new template will get different volumes
  235. self.app.pools['default'] = mock.Mock(**{
  236. 'init_volume.return_value.pool': 'default'})
  237. template2 = qubes.vm.templatevm.TemplateVM(self.app, None,
  238. qid=3, name=qubes.tests.VMPREFIX + 'template2')
  239. self.app.domains[template2.name] = template2
  240. self.app.domains[template2] = template2
  241. self.appvm.template = template2
  242. self.assertFalse(vm.volume_config['root']['save_on_stop'])
  243. self.assertTrue(vm.volume_config['root']['snap_on_start'])
  244. self.assertNotEqual(vm.volume_config['root']['source'],
  245. self.template.volumes['root'])
  246. self.assertIs(vm.volume_config['root']['source'],
  247. template2.volumes['root'])
  248. self.assertIs(vm.volume_config['root']['source'],
  249. self.appvm.volume_config['root']['source'])
  250. self.assertIs(vm.volume_config['private']['source'],
  251. self.appvm.volumes['private'])
  252. def test_022_storage_app_change(self):
  253. self.appvm.template_for_dispvms = True
  254. self.assertTrue(self.appvm.events_enabled)
  255. orig_domains = self.app.domains
  256. with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
  257. as mock_domains:
  258. mock_domains.configure_mock(**{
  259. 'get_new_unused_dispid': mock.Mock(return_value=42),
  260. '__getitem__.side_effect': orig_domains.__getitem__,
  261. '__iter__.side_effect': orig_domains.__iter__,
  262. '__setitem__.side_effect': orig_domains.__setitem__,
  263. })
  264. vm = self.dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
  265. name='test-dispvm', template=self.appvm)
  266. self.assertTrue(vm.events_enabled)
  267. self.loop.run_until_complete(self.dispvm.create_on_disk())
  268. # create new mock, so new template will get different volumes
  269. self.app.pools['default'] = mock.Mock(**{
  270. 'init_volume.return_value.pool': 'default'})
  271. template2 = qubes.vm.templatevm.TemplateVM(self.app, None,
  272. qid=3, name=qubes.tests.VMPREFIX + 'template2')
  273. self.assertTrue(template2.events_enabled)
  274. self.app.domains[template2.name] = template2
  275. self.app.domains[template2] = template2
  276. app2 = qubes.vm.appvm.AppVM(self.app, None,
  277. qid=4, name=qubes.tests.VMPREFIX + 'app2', template=template2)
  278. self.assertTrue(app2.events_enabled)
  279. app2.template_for_dispvms = True
  280. self.app.domains[app2.name] = app2
  281. self.app.domains[app2] = app2
  282. self.dispvm.template = app2
  283. self.assertIs(vm, self.dispvm)
  284. self.assertFalse(vm.volume_config['root']['save_on_stop'])
  285. self.assertTrue(vm.volume_config['root']['snap_on_start'])
  286. self.assertFalse(vm.volume_config['private']['save_on_stop'])
  287. self.assertTrue(vm.volume_config['private']['snap_on_start'])
  288. self.assertNotEqual(vm.volume_config['root']['source'],
  289. self.template.volumes['root'])
  290. self.assertNotEqual(vm.volume_config['root']['source'],
  291. self.appvm.volumes['root'].source)
  292. self.assertNotEqual(vm.volume_config['private']['source'],
  293. self.appvm.volumes['private'])
  294. self.assertIs(vm.volume_config['root']['source'],
  295. template2.volumes['root'])
  296. self.assertIs(app2.volume_config['root']['source'],
  297. template2.volumes['root'])
  298. self.assertIs(vm.volume_config['private']['source'],
  299. app2.volumes['private'])