basic.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500
  1. # pylint: disable=invalid-name
  2. #
  3. # The Qubes OS Project, https://www.qubes-os.org/
  4. #
  5. # Copyright (C) 2014-2015
  6. # Marek Marczykowski-Górecki <marmarek@invisiblethingslab.com>
  7. # Copyright (C) 2015 Wojtek Porczyk <woju@invisiblethingslab.com>
  8. #
  9. # This program is free software; you can redistribute it and/or modify
  10. # it under the terms of the GNU General Public License as published by
  11. # the Free Software Foundation; either version 2 of the License, or
  12. # (at your option) any later version.
  13. #
  14. # This program is distributed in the hope that it will be useful,
  15. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  16. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  17. # GNU General Public License for more details.
  18. #
  19. # You should have received a copy of the GNU General Public License along
  20. # with this program; if not, write to the Free Software Foundation, Inc.,
  21. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  22. #
  23. from distutils import spawn
  24. import asyncio
  25. import os
  26. import subprocess
  27. import tempfile
  28. import time
  29. import unittest
  30. import qubes
  31. import qubes.firewall
  32. import qubes.tests
  33. import qubes.vm.appvm
  34. import qubes.vm.qubesvm
  35. import qubes.vm.standalonevm
  36. import qubes.vm.templatevm
  37. import libvirt # pylint: disable=import-error
  38. class TC_00_Basic(qubes.tests.SystemTestCase):
  39. def setUp(self):
  40. super(TC_00_Basic, self).setUp()
  41. self.init_default_template()
  42. def test_000_qubes_create(self):
  43. self.assertIsInstance(self.app, qubes.Qubes)
  44. def test_100_qvm_create(self):
  45. vmname = self.make_vm_name('appvm')
  46. vm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  47. name=vmname, template=self.app.default_template,
  48. label='red')
  49. self.assertIsNotNone(vm)
  50. self.assertEqual(vm.name, vmname)
  51. self.assertEqual(vm.template, self.app.default_template)
  52. self.loop.run_until_complete(vm.create_on_disk())
  53. with self.assertNotRaises(qubes.exc.QubesException):
  54. self.loop.run_until_complete(vm.storage.verify())
  55. def test_040_qdb_watch(self):
  56. flag = set()
  57. def handler(vm, event, path):
  58. if path == '/test-watch-path':
  59. flag.add(True)
  60. vm = self.app.domains[0]
  61. vm.watch_qdb_path('/test-watch-path')
  62. vm.add_handler('domain-qdb-change:/test-watch-path', handler)
  63. self.assertFalse(flag)
  64. vm.untrusted_qdb.write('/test-watch-path', 'test-value')
  65. self.loop.run_until_complete(asyncio.sleep(0.1))
  66. self.assertTrue(flag)
  67. class TC_01_Properties(qubes.tests.SystemTestCase):
  68. # pylint: disable=attribute-defined-outside-init
  69. def setUp(self):
  70. super(TC_01_Properties, self).setUp()
  71. self.init_default_template()
  72. self.vmname = self.make_vm_name('appvm')
  73. self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=self.vmname,
  74. template=self.app.default_template,
  75. label='red')
  76. self.loop.run_until_complete(self.vm.create_on_disk())
  77. @unittest.expectedFailure
  78. def test_030_clone(self):
  79. testvm1 = self.app.add_new_vm(
  80. qubes.vm.appvm.AppVM,
  81. name=self.make_vm_name("vm"),
  82. template=self.app.default_template,
  83. label='red')
  84. self.loop.run_until_complete(testvm1.create_on_disk())
  85. testvm2 = self.app.add_new_vm(testvm1.__class__,
  86. name=self.make_vm_name("clone"),
  87. template=testvm1.template,
  88. label='red')
  89. testvm2.clone_properties(testvm1)
  90. self.loop.run_until_complete(testvm2.clone_disk_files(testvm1))
  91. self.assertTrue(self.loop.run_until_complete(testvm1.storage.verify()))
  92. self.assertIn('source', testvm1.volumes['root'].config)
  93. self.assertNotEquals(testvm2, None)
  94. self.assertNotEquals(testvm2.volumes, {})
  95. self.assertIn('source', testvm2.volumes['root'].config)
  96. # qubes.xml reload
  97. self.app.save()
  98. testvm1 = self.app.domains[testvm1.qid]
  99. testvm2 = self.app.domains[testvm2.qid]
  100. self.assertEqual(testvm1.label, testvm2.label)
  101. self.assertEqual(testvm1.netvm, testvm2.netvm)
  102. self.assertEqual(testvm1.property_is_default('netvm'),
  103. testvm2.property_is_default('netvm'))
  104. self.assertEqual(testvm1.kernel, testvm2.kernel)
  105. self.assertEqual(testvm1.kernelopts, testvm2.kernelopts)
  106. self.assertEqual(testvm1.property_is_default('kernel'),
  107. testvm2.property_is_default('kernel'))
  108. self.assertEqual(testvm1.property_is_default('kernelopts'),
  109. testvm2.property_is_default('kernelopts'))
  110. self.assertEqual(testvm1.memory, testvm2.memory)
  111. self.assertEqual(testvm1.maxmem, testvm2.maxmem)
  112. self.assertEqual(testvm1.devices, testvm2.devices)
  113. self.assertEqual(testvm1.include_in_backups,
  114. testvm2.include_in_backups)
  115. self.assertEqual(testvm1.default_user, testvm2.default_user)
  116. self.assertEqual(testvm1.features, testvm2.features)
  117. self.assertEqual(testvm1.firewall.rules,
  118. testvm2.firewall.rules)
  119. # now some non-default values
  120. testvm1.netvm = None
  121. testvm1.label = 'orange'
  122. testvm1.memory = 512
  123. firewall = testvm1.firewall
  124. firewall.rules = [
  125. qubes.firewall.Rule(None, action='accept', dsthost='1.2.3.0/24',
  126. proto='tcp', dstports=22)]
  127. firewall.save()
  128. testvm3 = self.app.add_new_vm(testvm1.__class__,
  129. name=self.make_vm_name("clone2"),
  130. template=testvm1.template,
  131. label='red',)
  132. testvm3.clone_properties(testvm1)
  133. self.loop.run_until_complete(testvm3.clone_disk_files(testvm1))
  134. # qubes.xml reload
  135. self.app.save()
  136. testvm1 = self.app.domains[testvm1.qid]
  137. testvm3 = self.app.domains[testvm3.qid]
  138. self.assertEqual(testvm1.label, testvm3.label)
  139. self.assertEqual(testvm1.netvm, testvm3.netvm)
  140. self.assertEqual(testvm1.property_is_default('netvm'),
  141. testvm3.property_is_default('netvm'))
  142. self.assertEqual(testvm1.kernel, testvm3.kernel)
  143. self.assertEqual(testvm1.kernelopts, testvm3.kernelopts)
  144. self.assertEqual(testvm1.property_is_default('kernel'),
  145. testvm3.property_is_default('kernel'))
  146. self.assertEqual(testvm1.property_is_default('kernelopts'),
  147. testvm3.property_is_default('kernelopts'))
  148. self.assertEqual(testvm1.memory, testvm3.memory)
  149. self.assertEqual(testvm1.maxmem, testvm3.maxmem)
  150. self.assertEqual(testvm1.devices, testvm3.devices)
  151. self.assertEqual(testvm1.include_in_backups,
  152. testvm3.include_in_backups)
  153. self.assertEqual(testvm1.default_user, testvm3.default_user)
  154. self.assertEqual(testvm1.features, testvm3.features)
  155. self.assertEqual(testvm1.firewall.rules,
  156. testvm2.firewall.rules)
  157. def test_020_name_conflict_app(self):
  158. # TODO decide what exception should be here
  159. with self.assertRaises((qubes.exc.QubesException, ValueError)):
  160. self.vm2 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  161. name=self.vmname, template=self.app.default_template,
  162. label='red')
  163. self.loop.run_until_complete(self.vm2.create_on_disk())
  164. def test_021_name_conflict_template(self):
  165. # TODO decide what exception should be here
  166. with self.assertRaises((qubes.exc.QubesException, ValueError)):
  167. self.vm2 = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM,
  168. name=self.vmname, label='red')
  169. self.loop.run_until_complete(self.vm2.create_on_disk())
  170. class TC_02_QvmPrefs(qubes.tests.SystemTestCase):
  171. # pylint: disable=attribute-defined-outside-init
  172. def setUp(self):
  173. super(TC_02_QvmPrefs, self).setUp()
  174. self.init_default_template()
  175. self.sharedopts = ['--qubesxml', qubes.tests.XMLPATH]
  176. def setup_appvm(self):
  177. self.testvm = self.app.add_new_vm(
  178. qubes.vm.appvm.AppVM,
  179. name=self.make_vm_name("vm"),
  180. label='red')
  181. self.loop.run_until_complete(self.testvm.create_on_disk())
  182. self.app.save()
  183. def setup_hvm(self):
  184. self.testvm = self.app.add_new_vm(
  185. qubes.vm.appvm.AppVM,
  186. name=self.make_vm_name("hvm"),
  187. label='red')
  188. self.testvm.virt_mode = 'hvm'
  189. self.loop.run_until_complete(self.testvm.create_on_disk())
  190. self.app.save()
  191. def pref_set(self, name, value, valid=True):
  192. self.loop.run_until_complete(self._pref_set(name, value, valid))
  193. @asyncio.coroutine
  194. def _pref_set(self, name, value, valid=True):
  195. cmd = ['qvm-prefs']
  196. if value != '-D':
  197. cmd.append('--')
  198. cmd.extend((self.testvm.name, name, value))
  199. p = yield from asyncio.create_subprocess_exec(*cmd,
  200. stdout=subprocess.PIPE,
  201. stderr=subprocess.PIPE)
  202. (stdout, stderr) = yield from p.communicate()
  203. if valid:
  204. self.assertEqual(p.returncode, 0,
  205. "qvm-prefs .. '{}' '{}' failed: {}{}".format(
  206. name, value, stdout, stderr
  207. ))
  208. else:
  209. self.assertNotEquals(p.returncode, 0,
  210. "qvm-prefs should reject value '{}' for "
  211. "property '{}'".format(value, name))
  212. def pref_get(self, name):
  213. self.loop.run_until_complete(self._pref_get(name))
  214. @asyncio.coroutine
  215. def _pref_get(self, name):
  216. p = yield from asyncio.create_subprocess_exec(
  217. 'qvm-prefs', *self.sharedopts, '--', self.testvm.name, name,
  218. stdout=subprocess.PIPE)
  219. (stdout, _) = yield from p.communicate()
  220. self.assertEqual(p.returncode, 0)
  221. return stdout.strip()
  222. bool_test_values = [
  223. ('true', 'True', True),
  224. ('False', 'False', True),
  225. ('0', 'False', True),
  226. ('1', 'True', True),
  227. ('invalid', '', False)
  228. ]
  229. def execute_tests(self, name, values):
  230. """
  231. Helper function, which executes tests for given property.
  232. :param values: list of tuples (value, expected, valid),
  233. where 'value' is what should be set and 'expected' is what should
  234. qvm-prefs returns as a property value and 'valid' marks valid and
  235. invalid values - if it's False, qvm-prefs should reject the value
  236. :return: None
  237. """
  238. for (value, expected, valid) in values:
  239. self.pref_set(name, value, valid)
  240. if valid:
  241. self.assertEqual(self.pref_get(name), expected)
  242. @unittest.skip('test not converted to core3 API')
  243. def test_006_template(self):
  244. templates = [tpl for tpl in self.app.domains.values() if
  245. isinstance(tpl, qubes.vm.templatevm.TemplateVM)]
  246. if not templates:
  247. self.skipTest("No templates installed")
  248. some_template = templates[0].name
  249. self.setup_appvm()
  250. self.execute_tests('template', [
  251. (some_template, some_template, True),
  252. ('invalid', '', False),
  253. ])
  254. @unittest.skip('test not converted to core3 API')
  255. def test_014_pcidevs(self):
  256. self.setup_appvm()
  257. self.execute_tests('pcidevs', [
  258. ('[]', '[]', True),
  259. ('[ "00:00.0" ]', "['00:00.0']", True),
  260. ('invalid', '', False),
  261. ('[invalid]', '', False),
  262. # TODO:
  263. # ('["12:12.0"]', '', False)
  264. ])
  265. @unittest.skip('test not converted to core3 API')
  266. def test_024_pv_reject_hvm_props(self):
  267. self.setup_appvm()
  268. self.execute_tests('guiagent_installed', [('False', '', False)])
  269. self.execute_tests('qrexec_installed', [('False', '', False)])
  270. self.execute_tests('drive', [('/tmp/drive.img', '', False)])
  271. self.execute_tests('timezone', [('localtime', '', False)])
  272. @unittest.skip('test not converted to core3 API')
  273. def test_025_hvm_reject_pv_props(self):
  274. self.setup_hvm()
  275. self.execute_tests('kernel', [('default', '', False)])
  276. self.execute_tests('kernelopts', [('default', '', False)])
  277. class TC_03_QvmRevertTemplateChanges(qubes.tests.SystemTestCase):
  278. # pylint: disable=attribute-defined-outside-init
  279. def setUp(self):
  280. super(TC_03_QvmRevertTemplateChanges, self).setUp()
  281. self.init_default_template()
  282. def setup_pv_template(self):
  283. self.test_template = self.app.add_new_vm(
  284. qubes.vm.templatevm.TemplateVM,
  285. name=self.make_vm_name("pv-clone"),
  286. label='red'
  287. )
  288. self.test_template.clone_properties(self.app.default_template)
  289. self.loop.run_until_complete(
  290. self.test_template.clone_disk_files(self.app.default_template))
  291. self.app.save()
  292. def setup_hvm_template(self):
  293. self.test_template = self.app.add_new_vm(
  294. qubes.vm.templatevm.TemplateVM,
  295. name=self.make_vm_name("hvm"),
  296. label='red',
  297. virt_mode='hvm',
  298. )
  299. self.loop.run_until_complete(self.test_template.create_on_disk())
  300. self.app.save()
  301. def get_rootimg_checksum(self):
  302. p = subprocess.Popen(
  303. ['sha1sum', self.test_template.volumes['root'].path],
  304. stdout=subprocess.PIPE)
  305. return p.communicate()[0]
  306. def _do_test(self):
  307. checksum_before = self.get_rootimg_checksum()
  308. self.loop.run_until_complete(self.test_template.start())
  309. self.shutdown_and_wait(self.test_template)
  310. checksum_changed = self.get_rootimg_checksum()
  311. if checksum_before == checksum_changed:
  312. self.log.warning("template not modified, test result will be "
  313. "unreliable")
  314. self.assertNotEqual(self.test_template.volumes['root'].revisions, {})
  315. pool_vid = repr(self.test_template.volumes['root']).strip("'")
  316. revert_cmd = ['qvm-block', 'revert', pool_vid]
  317. p = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  318. *revert_cmd))
  319. self.loop.run_until_complete(p.wait())
  320. self.assertEqual(p.returncode, 0)
  321. checksum_after = self.get_rootimg_checksum()
  322. self.assertEqual(checksum_before, checksum_after)
  323. @unittest.expectedFailure
  324. def test_000_revert_pv(self):
  325. """
  326. Test qvm-revert-template-changes for PV template
  327. """
  328. self.setup_pv_template()
  329. self._do_test()
  330. @unittest.skip('HVM not yet implemented')
  331. def test_000_revert_hvm(self):
  332. """
  333. Test qvm-revert-template-changes for HVM template
  334. """
  335. # TODO: have some system there, so the root.img will get modified
  336. self.setup_hvm_template()
  337. self._do_test()
  338. class TC_30_Gui_daemon(qubes.tests.SystemTestCase):
  339. def setUp(self):
  340. super(TC_30_Gui_daemon, self).setUp()
  341. self.init_default_template()
  342. @unittest.skipUnless(
  343. spawn.find_executable('xdotool'),
  344. "xdotool not installed")
  345. def test_000_clipboard(self):
  346. testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  347. name=self.make_vm_name('vm1'), label='red')
  348. self.loop.run_until_complete(testvm1.create_on_disk())
  349. testvm2 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  350. name=self.make_vm_name('vm2'), label='red')
  351. self.loop.run_until_complete(testvm2.create_on_disk())
  352. self.app.save()
  353. self.loop.run_until_complete(asyncio.wait([
  354. testvm1.start(),
  355. testvm2.start()]))
  356. window_title = 'user@{}'.format(testvm1.name)
  357. self.loop.run_until_complete(testvm1.run(
  358. 'zenity --text-info --editable --title={}'.format(window_title)))
  359. self.wait_for_window(window_title)
  360. time.sleep(0.5)
  361. test_string = "test{}".format(testvm1.xid)
  362. # Type and copy some text
  363. subprocess.check_call(['xdotool', 'search', '--name', window_title,
  364. 'windowactivate', '--sync',
  365. 'type', test_string])
  366. # second xdotool call because type --terminator do not work (SEGV)
  367. # additionally do not use search here, so window stack will be empty
  368. # and xdotool will use XTEST instead of generating events manually -
  369. # this will be much better - at least because events will have
  370. # correct timestamp (so gui-daemon would not drop the copy request)
  371. subprocess.check_call(['xdotool',
  372. 'key', 'ctrl+a', 'ctrl+c', 'ctrl+shift+c',
  373. 'Escape'])
  374. clipboard_content = \
  375. open('/var/run/qubes/qubes-clipboard.bin', 'r').read().strip()
  376. self.assertEqual(clipboard_content, test_string,
  377. "Clipboard copy operation failed - content")
  378. clipboard_source = \
  379. open('/var/run/qubes/qubes-clipboard.bin.source',
  380. 'r').read().strip()
  381. self.assertEqual(clipboard_source, testvm1.name,
  382. "Clipboard copy operation failed - owner")
  383. # Then paste it to the other window
  384. window_title = 'user@{}'.format(testvm2.name)
  385. p = self.loop.run_until_complete(testvm2.run(
  386. 'zenity --entry --title={} > /tmp/test.txt'.format(window_title)))
  387. self.wait_for_window(window_title)
  388. subprocess.check_call(['xdotool', 'key', '--delay', '100',
  389. 'ctrl+shift+v', 'ctrl+v', 'Return'])
  390. self.loop.run_until_complete(p.wait())
  391. # And compare the result
  392. (test_output, _) = self.loop.run_until_complete(
  393. testvm2.run_for_stdio('cat /tmp/test.txt'))
  394. self.assertEqual(test_string, test_output.strip().decode('ascii'))
  395. clipboard_content = \
  396. open('/var/run/qubes/qubes-clipboard.bin', 'r').read().strip()
  397. self.assertEqual(clipboard_content, "",
  398. "Clipboard not wiped after paste - content")
  399. clipboard_source = \
  400. open('/var/run/qubes/qubes-clipboard.bin.source', 'r').\
  401. read().strip()
  402. self.assertEqual(clipboard_source, "",
  403. "Clipboard not wiped after paste - owner")
  404. class TC_05_StandaloneVM(qubes.tests.SystemTestCase):
  405. def setUp(self):
  406. super(TC_05_StandaloneVM, self).setUp()
  407. self.init_default_template()
  408. def test_000_create_start(self):
  409. testvm1 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
  410. name=self.make_vm_name('vm1'), label='red')
  411. testvm1.features['qrexec'] = True
  412. self.loop.run_until_complete(
  413. testvm1.clone_disk_files(self.app.default_template))
  414. self.app.save()
  415. self.loop.run_until_complete(testvm1.start())
  416. self.assertEqual(testvm1.get_power_state(), "Running")
  417. def test_100_resize_root_img(self):
  418. testvm1 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
  419. name=self.make_vm_name('vm1'), label='red')
  420. testvm1.features['qrexec'] = True
  421. self.loop.run_until_complete(
  422. testvm1.clone_disk_files(self.app.default_template))
  423. self.app.save()
  424. self.loop.run_until_complete(
  425. testvm1.storage.resize(testvm1.volumes['root'], 20 * 1024 ** 3))
  426. self.assertEqual(testvm1.volumes['root'].size, 20 * 1024 ** 3)
  427. self.loop.run_until_complete(testvm1.start())
  428. # new_size in 1k-blocks
  429. (new_size, _) = self.loop.run_until_complete(
  430. testvm1.run_for_stdio('df --output=size /|tail -n 1'))
  431. # some safety margin for FS metadata
  432. self.assertGreater(int(new_size.strip()), 19 * 1024 ** 2)
  433. # vim: ts=4 sw=4 et