api_internal.py 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184
  1. # -*- encoding: utf-8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2019 Marek Marczykowski-Górecki
  6. # <marmarek@invisiblethingslab.com>
  7. #
  8. # This program is free software; you can redistribute it and/or modify
  9. # it under the terms of the GNU General Public License as published by
  10. # the Free Software Foundation; either version 2 of the License, or
  11. # (at your option) any later version.
  12. #
  13. # This program 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
  16. # GNU General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU General Public License along
  19. # with this program; if not, see <http://www.gnu.org/licenses/>.
  20. import asyncio
  21. import qubes.api.internal
  22. import qubes.tests
  23. import qubes.vm.adminvm
  24. from unittest import mock
  25. import json
  26. def mock_coro(f):
  27. @asyncio.coroutine
  28. def coro_f(*args, **kwargs):
  29. return f(*args, **kwargs)
  30. return coro_f
  31. class TC_00_API_Misc(qubes.tests.QubesTestCase):
  32. def setUp(self):
  33. super().setUp()
  34. self.app = mock.NonCallableMock()
  35. self.dom0 = mock.NonCallableMock(spec=qubes.vm.adminvm.AdminVM)
  36. self.dom0.name = 'dom0'
  37. self.domains = {
  38. 'dom0': self.dom0,
  39. }
  40. self.app.domains = mock.MagicMock(**{
  41. '__iter__.side_effect': lambda: iter(self.domains.values()),
  42. '__getitem__.side_effect': self.domains.get,
  43. })
  44. def tearDown(self):
  45. self.domains.clear()
  46. self.dom0 = None
  47. super().tearDown()
  48. def create_mockvm(self, features=None):
  49. if features is None:
  50. features = {}
  51. vm = mock.Mock()
  52. vm.features.check_with_template.side_effect = features.get
  53. vm.run_service.return_value.wait = mock_coro(
  54. vm.run_service.return_value.wait)
  55. vm.run_service = mock_coro(vm.run_service)
  56. vm.suspend = mock_coro(vm.suspend)
  57. vm.resume = mock_coro(vm.resume)
  58. return vm
  59. def call_mgmt_func(self, method, arg=b'', payload=b''):
  60. mgmt_obj = qubes.api.internal.QubesInternalAPI(self.app,
  61. b'dom0', method, b'dom0', arg)
  62. loop = asyncio.get_event_loop()
  63. response = loop.run_until_complete(
  64. mgmt_obj.execute(untrusted_payload=payload))
  65. return response
  66. def test_000_suspend_pre(self):
  67. running_vm = self.create_mockvm(features={'qrexec': True})
  68. running_vm.is_running.return_value = True
  69. not_running_vm = self.create_mockvm(features={'qrexec': True})
  70. not_running_vm.is_running.return_value = False
  71. no_qrexec_vm = self.create_mockvm()
  72. no_qrexec_vm.is_running.return_value = True
  73. self.domains.update({
  74. 'running': running_vm,
  75. 'not-running': not_running_vm,
  76. 'no-qrexec': no_qrexec_vm,
  77. })
  78. ret = self.call_mgmt_func(b'internal.SuspendPre')
  79. self.assertIsNone(ret)
  80. self.assertFalse(self.dom0.called)
  81. self.assertNotIn(('run_service', ('qubes.SuspendPreAll',), mock.ANY),
  82. not_running_vm.mock_calls)
  83. self.assertNotIn(('suspend', (), {}),
  84. not_running_vm.mock_calls)
  85. self.assertIn(('run_service', ('qubes.SuspendPreAll',), mock.ANY),
  86. running_vm.mock_calls)
  87. self.assertIn(('suspend', (), {}),
  88. running_vm.mock_calls)
  89. self.assertNotIn(('run_service', ('qubes.SuspendPreAll',), mock.ANY),
  90. no_qrexec_vm.mock_calls)
  91. self.assertIn(('suspend', (), {}),
  92. no_qrexec_vm.mock_calls)
  93. def test_001_suspend_post(self):
  94. running_vm = self.create_mockvm(features={'qrexec': True})
  95. running_vm.is_running.return_value = True
  96. running_vm.get_power_state.return_value = 'Suspended'
  97. not_running_vm = self.create_mockvm(features={'qrexec': True})
  98. not_running_vm.is_running.return_value = False
  99. not_running_vm.get_power_state.return_value = 'Halted'
  100. no_qrexec_vm = self.create_mockvm()
  101. no_qrexec_vm.is_running.return_value = True
  102. no_qrexec_vm.get_power_state.return_value = 'Suspended'
  103. self.domains.update({
  104. 'running': running_vm,
  105. 'not-running': not_running_vm,
  106. 'no-qrexec': no_qrexec_vm,
  107. })
  108. ret = self.call_mgmt_func(b'internal.SuspendPost')
  109. self.assertIsNone(ret)
  110. self.assertFalse(self.dom0.called)
  111. self.assertNotIn(('run_service', ('qubes.SuspendPostAll',), mock.ANY),
  112. not_running_vm.mock_calls)
  113. self.assertNotIn(('resume', (), {}),
  114. not_running_vm.mock_calls)
  115. self.assertIn(('run_service', ('qubes.SuspendPostAll',), mock.ANY),
  116. running_vm.mock_calls)
  117. self.assertIn(('resume', (), {}),
  118. running_vm.mock_calls)
  119. self.assertNotIn(('run_service', ('qubes.SuspendPostAll',), mock.ANY),
  120. no_qrexec_vm.mock_calls)
  121. self.assertIn(('resume', (), {}),
  122. no_qrexec_vm.mock_calls)
  123. def test_010_get_system_info(self):
  124. self.dom0.name = 'dom0'
  125. self.dom0.tags = ['tag1', 'tag2']
  126. self.dom0.default_dispvm = None
  127. self.dom0.template_for_dispvms = False
  128. self.dom0.label.icon = 'icon-dom0'
  129. del self.dom0.guivm
  130. vm = mock.NonCallableMock(spec=qubes.vm.qubesvm.QubesVM)
  131. vm.name = 'vm'
  132. vm.tags = ['tag3', 'tag4']
  133. vm.default_dispvm = vm
  134. vm.template_for_dispvms = True
  135. vm.label.icon = 'icon-vm'
  136. vm.guivm = vm
  137. self.domains['vm'] = vm
  138. ret = json.loads(self.call_mgmt_func(b'internal.GetSystemInfo'))
  139. self.assertEqual(ret, {
  140. 'domains': {
  141. 'dom0': {
  142. 'tags': ['tag1', 'tag2'],
  143. 'type': 'AdminVM',
  144. 'default_dispvm': None,
  145. 'template_for_dispvms': False,
  146. 'icon': 'icon-dom0',
  147. 'guivm': None,
  148. },
  149. 'vm': {
  150. 'tags': ['tag3', 'tag4'],
  151. 'type': 'QubesVM',
  152. 'default_dispvm': 'vm',
  153. 'template_for_dispvms': True,
  154. 'icon': 'icon-vm',
  155. 'guivm': 'vm',
  156. }
  157. }
  158. })