api_misc.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278
  1. # -*- encoding: utf8 -*-
  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 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. from unittest import mock
  22. import qubes.tests
  23. import qubes.api.misc
  24. class TC_00_API_Misc(qubes.tests.QubesTestCase):
  25. def setUp(self):
  26. super(TC_00_API_Misc, self).setUp()
  27. self.src = mock.NonCallableMagicMock()
  28. self.app = mock.NonCallableMock()
  29. self.dest = mock.NonCallableMock()
  30. self.dest.name = 'dom0'
  31. self.app.configure_mock(domains={
  32. 'dom0': self.dest,
  33. 'test-vm': self.src,
  34. })
  35. def configure_qdb(self, entries):
  36. self.src.configure_mock(**{
  37. 'untrusted_qdb.read.side_effect': (
  38. lambda path: entries.get(path, None)),
  39. 'untrusted_qdb.list.side_effect': (
  40. lambda path: sorted(entries.keys())),
  41. })
  42. def call_mgmt_func(self, method, arg=b'', payload=b''):
  43. mgmt_obj = qubes.api.misc.QubesMiscAPI(self.app,
  44. b'test-vm', method, b'dom0', arg)
  45. loop = asyncio.get_event_loop()
  46. response = loop.run_until_complete(
  47. mgmt_obj.execute(untrusted_payload=payload))
  48. return response
  49. def test_000_features_request(self):
  50. qdb_entries = {
  51. '/features-request/feature1': b'1',
  52. '/features-request/feature2': b'',
  53. '/features-request/feature3': b'other',
  54. }
  55. self.configure_qdb(qdb_entries)
  56. response = self.call_mgmt_func(b'qubes.FeaturesRequest')
  57. self.assertIsNone(response)
  58. self.assertEqual(self.app.mock_calls, [
  59. mock.call.save()
  60. ])
  61. self.assertEqual(self.src.mock_calls, [
  62. mock.call.untrusted_qdb.list('/features-request/'),
  63. mock.call.untrusted_qdb.read('/features-request/feature1'),
  64. mock.call.untrusted_qdb.read('/features-request/feature2'),
  65. mock.call.untrusted_qdb.read('/features-request/feature3'),
  66. mock.call.fire_event('features-request', untrusted_features={
  67. 'feature1': '1', 'feature2': '', 'feature3': 'other'})
  68. ])
  69. def test_001_features_request_empty(self):
  70. self.configure_qdb({})
  71. response = self.call_mgmt_func(b'qubes.FeaturesRequest')
  72. self.assertIsNone(response)
  73. self.assertEqual(self.app.mock_calls, [
  74. mock.call.save()
  75. ])
  76. self.assertEqual(self.src.mock_calls, [
  77. mock.call.untrusted_qdb.list('/features-request/'),
  78. mock.call.fire_event('features-request', untrusted_features={})
  79. ])
  80. def test_002_features_request_invalid1(self):
  81. qdb_entries = {
  82. '/features-request/feature1': b'test spaces',
  83. }
  84. self.configure_qdb(qdb_entries)
  85. with self.assertRaises(AssertionError):
  86. self.call_mgmt_func(b'qubes.FeaturesRequest')
  87. self.assertEqual(self.app.mock_calls, [])
  88. self.assertEqual(self.src.mock_calls, [
  89. mock.call.untrusted_qdb.list('/features-request/'),
  90. mock.call.untrusted_qdb.read('/features-request/feature1'),
  91. ])
  92. def test_003_features_request_invalid2(self):
  93. qdb_entries = {
  94. '/features-request/feature1': b'\xfe\x01',
  95. }
  96. self.configure_qdb(qdb_entries)
  97. with self.assertRaises(UnicodeDecodeError):
  98. self.call_mgmt_func(b'qubes.FeaturesRequest')
  99. self.assertEqual(self.app.mock_calls, [])
  100. self.assertEqual(self.src.mock_calls, [
  101. mock.call.untrusted_qdb.list('/features-request/'),
  102. mock.call.untrusted_qdb.read('/features-request/feature1'),
  103. ])
  104. def test_010_notify_tools(self):
  105. qdb_entries = {
  106. '/qubes-tools/version': b'1',
  107. '/qubes-tools/qrexec': b'1',
  108. '/qubes-tools/gui': b'1',
  109. '/qubes-tools/os': b'Linux',
  110. '/qubes-tools/default-user': b'user',
  111. }
  112. self.configure_qdb(qdb_entries)
  113. response = self.call_mgmt_func(b'qubes.NotifyTools')
  114. self.assertIsNone(response)
  115. self.assertEqual(self.app.mock_calls, [
  116. mock.call.save()
  117. ])
  118. self.assertEqual(self.src.mock_calls, [
  119. mock.call.untrusted_qdb.read('/qubes-tools/qrexec'),
  120. mock.call.untrusted_qdb.read('/qubes-tools/gui'),
  121. mock.call.untrusted_qdb.read('/qubes-tools/default-user'),
  122. mock.call.fire_event('features-request', untrusted_features={
  123. 'gui': '1',
  124. 'default-user': 'user',
  125. 'qrexec': '1'}),
  126. ])
  127. self.assertEqual(self.app.mock_calls, [mock.call.save()])
  128. def test_013_notify_tools_no_version(self):
  129. qdb_entries = {
  130. '/qubes-tools/qrexec': b'1',
  131. '/qubes-tools/gui': b'1',
  132. '/qubes-tools/os': b'Linux',
  133. '/qubes-tools/default-user': b'user',
  134. }
  135. self.configure_qdb(qdb_entries)
  136. response = self.call_mgmt_func(b'qubes.NotifyTools')
  137. self.assertIsNone(response)
  138. self.assertEqual(self.src.mock_calls, [
  139. mock.call.untrusted_qdb.read('/qubes-tools/qrexec'),
  140. mock.call.untrusted_qdb.read('/qubes-tools/gui'),
  141. mock.call.untrusted_qdb.read('/qubes-tools/default-user'),
  142. mock.call.fire_event('features-request', untrusted_features={
  143. 'gui': '1',
  144. 'default-user': 'user',
  145. 'qrexec': '1'}),
  146. ])
  147. self.assertEqual(self.app.mock_calls, [mock.call.save()])
  148. def test_015_notify_tools_invalid_value_qrexec(self):
  149. qdb_entries = {
  150. '/qubes-tools/version': b'1',
  151. '/qubes-tools/qrexec': b'invalid value',
  152. '/qubes-tools/gui': b'0',
  153. '/qubes-tools/os': b'Linux',
  154. '/qubes-tools/default-user': b'user',
  155. }
  156. self.configure_qdb(qdb_entries)
  157. with self.assertRaises(AssertionError):
  158. self.call_mgmt_func(b'qubes.NotifyTools')
  159. self.assertEqual(self.app.mock_calls, [])
  160. self.assertEqual(self.src.mock_calls, [
  161. mock.call.untrusted_qdb.read('/qubes-tools/qrexec'),
  162. ])
  163. def test_016_notify_tools_invalid_value_gui(self):
  164. qdb_entries = {
  165. '/qubes-tools/version': b'1',
  166. '/qubes-tools/qrexec': b'1',
  167. '/qubes-tools/gui': b'invalid value',
  168. '/qubes-tools/os': b'Linux',
  169. '/qubes-tools/default-user': b'user',
  170. }
  171. self.configure_qdb(qdb_entries)
  172. with self.assertRaises(AssertionError):
  173. self.call_mgmt_func(b'qubes.NotifyTools')
  174. self.assertEqual(self.app.mock_calls, [])
  175. self.assertEqual(self.src.mock_calls, [
  176. mock.call.untrusted_qdb.read('/qubes-tools/qrexec'),
  177. mock.call.untrusted_qdb.read('/qubes-tools/gui'),
  178. ])
  179. def test_020_notify_updates_standalone(self):
  180. del self.src.template
  181. response = self.call_mgmt_func(b'qubes.NotifyUpdates', payload=b'1\n')
  182. self.assertIsNone(response)
  183. self.assertEqual(self.src.mock_calls, [
  184. mock.call.updateable.__bool__(),
  185. mock.call.features.__setitem__('updates-available', True),
  186. ])
  187. self.assertEqual(self.app.mock_calls, [mock.call.save()])
  188. def test_021_notify_updates_standalone2(self):
  189. del self.src.template
  190. response = self.call_mgmt_func(b'qubes.NotifyUpdates', payload=b'0\n')
  191. self.assertIsNone(response)
  192. self.assertEqual(self.src.mock_calls, [
  193. mock.call.updateable.__bool__(),
  194. mock.call.features.__setitem__('updates-available', False),
  195. ])
  196. self.assertEqual(self.app.mock_calls, [
  197. mock.call.save()
  198. ])
  199. def test_022_notify_updates_invalid(self):
  200. del self.src.template
  201. with self.assertRaises(AssertionError):
  202. self.call_mgmt_func(b'qubes.NotifyUpdates', payload=b'')
  203. self.assertEqual(self.src.mock_calls, [])
  204. self.assertEqual(self.app.mock_calls, [])
  205. def test_023_notify_updates_invalid2(self):
  206. del self.src.template
  207. with self.assertRaises(AssertionError):
  208. self.call_mgmt_func(b'qubes.NotifyUpdates', payload=b'no updates')
  209. self.assertEqual(self.src.mock_calls, [])
  210. self.assertEqual(self.app.mock_calls, [])
  211. def test_024_notify_updates_template_based_no_updates(self):
  212. '''No updates on template-based VM, should not reset state'''
  213. self.src.updateable = False
  214. self.src.template.is_running.return_value = False
  215. response = self.call_mgmt_func(b'qubes.NotifyUpdates', payload=b'0\n')
  216. self.assertIsNone(response)
  217. self.assertEqual(self.src.mock_calls, [
  218. mock.call.template.is_running(),
  219. ])
  220. self.assertEqual(self.app.mock_calls, [])
  221. def test_025_notify_updates_template_based(self):
  222. '''Some updates on template-based VM, should save flag'''
  223. self.src.updateable = False
  224. self.src.template.is_running.return_value = False
  225. self.src.storage.outdated_volumes = []
  226. response = self.call_mgmt_func(b'qubes.NotifyUpdates', payload=b'1\n')
  227. self.assertIsNone(response)
  228. self.assertEqual(self.src.mock_calls, [
  229. mock.call.template.is_running(),
  230. mock.call.template.features.__setitem__('updates-available', True),
  231. ])
  232. self.assertEqual(self.app.mock_calls, [
  233. mock.call.save()
  234. ])
  235. def test_026_notify_updates_template_based_outdated(self):
  236. self.src.updateable = False
  237. self.src.template.is_running.return_value = False
  238. self.src.storage.outdated_volumes = ['root']
  239. response = self.call_mgmt_func(b'qubes.NotifyUpdates', payload=b'1\n')
  240. self.assertIsNone(response)
  241. self.assertEqual(self.src.mock_calls, [
  242. mock.call.template.is_running(),
  243. ])
  244. self.assertIsInstance(self.src.template.updates_available, mock.Mock)
  245. self.assertEqual(self.app.mock_calls, [])
  246. def test_027_notify_updates_template_based_template_running(self):
  247. self.src.updateable = False
  248. self.src.template.is_running.return_value = True
  249. self.src.storage.outdated_volumes = []
  250. response = self.call_mgmt_func(b'qubes.NotifyUpdates', payload=b'1\n')
  251. self.assertIsNone(response)
  252. self.assertEqual(self.src.mock_calls, [
  253. mock.call.template.is_running(),
  254. ])
  255. self.assertIsInstance(self.src.updates_available, mock.Mock)
  256. self.assertEqual(self.app.mock_calls, [])