basic.py 33 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809
  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 library is free software; you can redistribute it and/or
  10. # modify it under the terms of the GNU Lesser General Public
  11. # License as published by the Free Software Foundation; either
  12. # version 2.1 of the License, or (at your option) any later version.
  13. #
  14. # This library 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 GNU
  17. # Lesser General Public License for more details.
  18. #
  19. # You should have received a copy of the GNU Lesser General Public
  20. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  21. #
  22. from distutils import spawn
  23. import asyncio
  24. import os
  25. import subprocess
  26. import tempfile
  27. import time
  28. import unittest
  29. import collections
  30. import pkg_resources
  31. import shutil
  32. import sys
  33. import qubes
  34. import qubes.firewall
  35. import qubes.tests
  36. import qubes.storage
  37. import qubes.vm.appvm
  38. import qubes.vm.qubesvm
  39. import qubes.vm.standalonevm
  40. import qubes.vm.templatevm
  41. import libvirt # pylint: disable=import-error
  42. class TC_00_Basic(qubes.tests.SystemTestCase):
  43. def setUp(self):
  44. super(TC_00_Basic, self).setUp()
  45. self.init_default_template()
  46. def test_000_qubes_create(self):
  47. self.assertIsInstance(self.app, qubes.Qubes)
  48. def test_100_qvm_create(self):
  49. vmname = self.make_vm_name('appvm')
  50. vm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  51. name=vmname, template=self.app.default_template,
  52. label='red')
  53. self.assertIsNotNone(vm)
  54. self.assertEqual(vm.name, vmname)
  55. self.assertEqual(vm.template, self.app.default_template)
  56. self.loop.run_until_complete(vm.create_on_disk())
  57. with self.assertNotRaises(qubes.exc.QubesException):
  58. self.loop.run_until_complete(vm.storage.verify())
  59. def test_040_qdb_watch(self):
  60. flag = set()
  61. def handler(vm, event, path):
  62. if path == '/test-watch-path':
  63. flag.add(True)
  64. vm = self.app.domains[0]
  65. vm.watch_qdb_path('/test-watch-path')
  66. vm.add_handler('domain-qdb-change:/test-watch-path', handler)
  67. self.assertFalse(flag)
  68. vm.untrusted_qdb.write('/test-watch-path', 'test-value')
  69. self.loop.run_until_complete(asyncio.sleep(0.1))
  70. self.assertTrue(flag)
  71. @unittest.skipUnless(
  72. spawn.find_executable('xdotool'), "xdotool not installed")
  73. def test_120_start_standalone_with_cdrom_dom0(self):
  74. vmname = self.make_vm_name('appvm')
  75. self.vm = self.app.add_new_vm('StandaloneVM', label='red', name=vmname)
  76. self.loop.run_until_complete(self.vm.create_on_disk())
  77. self.vm.kernel = None
  78. self.vm.virt_mode = 'hvm'
  79. iso_path = self.create_bootable_iso()
  80. # start the VM using qvm-start tool, to test --cdrom option there
  81. p = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  82. 'qvm-start', '--cdrom=dom0:' + iso_path, self.vm.name,
  83. stdout=subprocess.PIPE, stderr=subprocess.STDOUT))
  84. (stdout, _) = self.loop.run_until_complete(p.communicate())
  85. self.assertEqual(p.returncode, 0, stdout)
  86. # check if VM do not crash instantly
  87. self.loop.run_until_complete(asyncio.sleep(5))
  88. self.assertTrue(self.vm.is_running())
  89. # Type 'poweroff'
  90. subprocess.check_call(['xdotool', 'search', '--name', self.vm.name,
  91. 'type', 'poweroff\r'])
  92. self.loop.run_until_complete(asyncio.sleep(1))
  93. self.assertFalse(self.vm.is_running())
  94. def _test_200_on_domain_start(self, vm, event, **_kwargs):
  95. '''Simulate domain crash just after startup'''
  96. vm.libvirt_domain.destroy()
  97. def test_200_shutdown_event_race(self):
  98. '''Regression test for 3164'''
  99. vmname = self.make_vm_name('appvm')
  100. self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  101. name=vmname, template=self.app.default_template,
  102. label='red')
  103. # help the luck a little - don't wait for qrexec to easier win the race
  104. self.vm.features['qrexec'] = False
  105. self.loop.run_until_complete(self.vm.create_on_disk())
  106. # another way to help the luck a little - make sure the private
  107. # volume is first in (normally unordered) dict - this way if any
  108. # volume action fails, it will be at or after private volume - not
  109. # before (preventing private volume action)
  110. old_volumes = self.vm.volumes
  111. self.vm.volumes = collections.OrderedDict()
  112. self.vm.volumes['private'] = old_volumes.pop('private')
  113. self.vm.volumes.update(old_volumes.items())
  114. del old_volumes
  115. self.loop.run_until_complete(self.vm.start())
  116. # kill it the way it does not give a chance for domain-shutdown it
  117. # execute
  118. self.vm.libvirt_domain.destroy()
  119. # now, lets try to start the VM again, before domain-shutdown event
  120. # got handled (#3164), and immediately trigger second domain-shutdown
  121. self.vm.add_handler('domain-start', self._test_200_on_domain_start)
  122. self.loop.run_until_complete(self.vm.start())
  123. # and give a chance for both domain-shutdown handlers to execute
  124. self.loop.run_until_complete(asyncio.sleep(1))
  125. with self.assertNotRaises(qubes.exc.QubesException):
  126. # if the above caused two domain-shutdown handlers being called
  127. # one after another, private volume is gone
  128. self.loop.run_until_complete(self.vm.storage.verify())
  129. def _test_201_on_domain_pre_start(self, vm, event, **_kwargs):
  130. '''Simulate domain crash just after startup'''
  131. if not self.domain_shutdown_handled and not self.test_failure_reason:
  132. self.test_failure_reason = \
  133. 'domain-shutdown event was not dispatched before subsequent ' \
  134. 'start'
  135. self.domain_shutdown_handled = False
  136. def _test_201_domain_shutdown_handler(self, vm, event, **kwargs):
  137. if self.domain_shutdown_handled and not self.test_failure_reason:
  138. self.test_failure_reason = 'domain-shutdown event received twice'
  139. self.domain_shutdown_handled = True
  140. def test_201_shutdown_event_race(self):
  141. '''Regression test for 3164 - pure events edition'''
  142. vmname = self.make_vm_name('appvm')
  143. self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  144. name=vmname, template=self.app.default_template,
  145. label='red')
  146. # help the luck a little - don't wait for qrexec to easier win the race
  147. self.vm.features['qrexec'] = False
  148. self.loop.run_until_complete(self.vm.create_on_disk())
  149. # do not throw exception from inside event handler - test framework
  150. # will not recover from it (various objects leaks)
  151. self.test_failure_reason = None
  152. self.domain_shutdown_handled = False
  153. self.vm.add_handler('domain-shutdown',
  154. self._test_201_domain_shutdown_handler)
  155. self.loop.run_until_complete(self.vm.start())
  156. if self.test_failure_reason:
  157. self.fail(self.test_failure_reason)
  158. self.vm.add_handler('domain-pre-start',
  159. self._test_201_on_domain_pre_start)
  160. # kill it the way it does not give a chance for domain-shutdown it
  161. # execute
  162. self.vm.libvirt_domain.destroy()
  163. # now, lets try to start the VM again, before domain-shutdown event
  164. # got handled (#3164), and immediately trigger second domain-shutdown
  165. self.vm.add_handler('domain-start', self._test_200_on_domain_start)
  166. self.loop.run_until_complete(self.vm.start())
  167. if self.test_failure_reason:
  168. self.fail(self.test_failure_reason)
  169. # and give a chance for both domain-shutdown handlers to execute
  170. self.loop.run_until_complete(asyncio.sleep(1))
  171. if self.test_failure_reason:
  172. self.fail(self.test_failure_reason)
  173. self.assertTrue(self.domain_shutdown_handled,
  174. 'second domain-shutdown event was not dispatched after domain '
  175. 'shutdown')
  176. def _check_udev_for_uuid(self, uuid_value):
  177. udev_data_path = '/run/udev/data'
  178. for udev_item in os.listdir(udev_data_path):
  179. # check only block devices
  180. if not udev_item.startswith('b'):
  181. continue
  182. with open(os.path.join(udev_data_path, udev_item)) as udev_file:
  183. self.assertNotIn(uuid_value, udev_file.read(),
  184. 'udev parsed filesystem UUID! ' + udev_item)
  185. def assertVolumesExcludedFromUdev(self, vm):
  186. try:
  187. # first boot, mkfs private volume
  188. self.loop.run_until_complete(vm.start())
  189. # get private volume UUID
  190. private_uuid, _ = self.loop.run_until_complete(
  191. vm.run_for_stdio('blkid -o value /dev/xvdb', user='root'))
  192. private_uuid = private_uuid.decode().splitlines()[0]
  193. # now check if dom0 udev know about it - it shouldn't
  194. self._check_udev_for_uuid(private_uuid)
  195. # now restart the VM and check again
  196. self.loop.run_until_complete(vm.shutdown(wait=True))
  197. self.loop.run_until_complete(vm.start())
  198. self._check_udev_for_uuid(private_uuid)
  199. finally:
  200. del vm
  201. def test_202_udev_block_exclude_default(self):
  202. '''Check if VM images are excluded from udev parsing -
  203. default volume pool'''
  204. vmname = self.make_vm_name('appvm')
  205. self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  206. name=vmname, template=self.app.default_template,
  207. label='red')
  208. self.loop.run_until_complete(self.vm.create_on_disk())
  209. self.assertVolumesExcludedFromUdev(self.vm)
  210. def test_203_udev_block_exclude_varlibqubes(self):
  211. '''Check if VM images are excluded from udev parsing -
  212. varlibqubes pool'''
  213. vmname = self.make_vm_name('appvm')
  214. self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  215. name=vmname, template=self.app.default_template,
  216. label='red')
  217. self.loop.run_until_complete(self.vm.create_on_disk(
  218. pool=self.app.pools['varlibqubes']))
  219. self.assertVolumesExcludedFromUdev(self.vm)
  220. def test_204_udev_block_exclude_custom_file(self):
  221. '''Check if VM images are excluded from udev parsing -
  222. custom file pool'''
  223. vmname = self.make_vm_name('appvm')
  224. pool_path = tempfile.mkdtemp(
  225. prefix='qubes-pool-', dir='/var/tmp')
  226. self.addCleanup(shutil.rmtree, pool_path)
  227. pool = self.app.add_pool('test-filep', dir_path=pool_path,
  228. driver='file')
  229. self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  230. name=vmname, template=self.app.default_template,
  231. label='red')
  232. self.loop.run_until_complete(self.vm.create_on_disk(
  233. pool=pool))
  234. self.assertVolumesExcludedFromUdev(self.vm)
  235. class TC_01_Properties(qubes.tests.SystemTestCase):
  236. # pylint: disable=attribute-defined-outside-init
  237. def setUp(self):
  238. super(TC_01_Properties, self).setUp()
  239. self.init_default_template()
  240. self.vmname = self.make_vm_name('appvm')
  241. self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=self.vmname,
  242. template=self.app.default_template,
  243. label='red')
  244. self.loop.run_until_complete(self.vm.create_on_disk())
  245. self.addCleanup(self.cleanup_props)
  246. def cleanup_props(self):
  247. del self.vm
  248. def test_030_clone(self):
  249. try:
  250. testvm1 = self.app.add_new_vm(
  251. qubes.vm.appvm.AppVM,
  252. name=self.make_vm_name("vm"),
  253. template=self.app.default_template,
  254. label='red')
  255. self.loop.run_until_complete(testvm1.create_on_disk())
  256. testvm2 = self.app.add_new_vm(testvm1.__class__,
  257. name=self.make_vm_name("clone"),
  258. template=testvm1.template,
  259. label='red')
  260. testvm2.clone_properties(testvm1)
  261. testvm2.firewall.clone(testvm1.firewall)
  262. self.loop.run_until_complete(testvm2.clone_disk_files(testvm1))
  263. self.assertTrue(self.loop.run_until_complete(testvm1.storage.verify()))
  264. self.assertIn('source', testvm1.volumes['root'].config)
  265. self.assertNotEquals(testvm2, None)
  266. self.assertNotEquals(testvm2.volumes, {})
  267. self.assertIn('source', testvm2.volumes['root'].config)
  268. # qubes.xml reload
  269. self.app.save()
  270. testvm1 = self.app.domains[testvm1.qid]
  271. testvm2 = self.app.domains[testvm2.qid]
  272. self.assertEqual(testvm1.label, testvm2.label)
  273. self.assertEqual(testvm1.netvm, testvm2.netvm)
  274. self.assertEqual(testvm1.property_is_default('netvm'),
  275. testvm2.property_is_default('netvm'))
  276. self.assertEqual(testvm1.kernel, testvm2.kernel)
  277. self.assertEqual(testvm1.kernelopts, testvm2.kernelopts)
  278. self.assertEqual(testvm1.property_is_default('kernel'),
  279. testvm2.property_is_default('kernel'))
  280. self.assertEqual(testvm1.property_is_default('kernelopts'),
  281. testvm2.property_is_default('kernelopts'))
  282. self.assertEqual(testvm1.memory, testvm2.memory)
  283. self.assertEqual(testvm1.maxmem, testvm2.maxmem)
  284. self.assertEqual(testvm1.devices, testvm2.devices)
  285. self.assertEqual(testvm1.include_in_backups,
  286. testvm2.include_in_backups)
  287. self.assertEqual(testvm1.default_user, testvm2.default_user)
  288. self.assertEqual(testvm1.features, testvm2.features)
  289. self.assertEqual(testvm1.firewall.rules,
  290. testvm2.firewall.rules)
  291. # now some non-default values
  292. testvm1.netvm = None
  293. testvm1.label = 'orange'
  294. testvm1.memory = 512
  295. firewall = testvm1.firewall
  296. firewall.rules = [
  297. qubes.firewall.Rule(None, action='accept', dsthost='1.2.3.0/24',
  298. proto='tcp', dstports=22)]
  299. firewall.save()
  300. testvm3 = self.app.add_new_vm(testvm1.__class__,
  301. name=self.make_vm_name("clone2"),
  302. template=testvm1.template,
  303. label='red',)
  304. testvm3.clone_properties(testvm1)
  305. testvm3.firewall.clone(testvm1.firewall)
  306. self.loop.run_until_complete(testvm3.clone_disk_files(testvm1))
  307. # qubes.xml reload
  308. self.app.save()
  309. testvm1 = self.app.domains[testvm1.qid]
  310. testvm3 = self.app.domains[testvm3.qid]
  311. self.assertEqual(testvm1.label, testvm3.label)
  312. self.assertEqual(testvm1.netvm, testvm3.netvm)
  313. self.assertEqual(testvm1.property_is_default('netvm'),
  314. testvm3.property_is_default('netvm'))
  315. self.assertEqual(testvm1.kernel, testvm3.kernel)
  316. self.assertEqual(testvm1.kernelopts, testvm3.kernelopts)
  317. self.assertEqual(testvm1.property_is_default('kernel'),
  318. testvm3.property_is_default('kernel'))
  319. self.assertEqual(testvm1.property_is_default('kernelopts'),
  320. testvm3.property_is_default('kernelopts'))
  321. self.assertEqual(testvm1.memory, testvm3.memory)
  322. self.assertEqual(testvm1.maxmem, testvm3.maxmem)
  323. self.assertEqual(testvm1.devices, testvm3.devices)
  324. self.assertEqual(testvm1.include_in_backups,
  325. testvm3.include_in_backups)
  326. self.assertEqual(testvm1.default_user, testvm3.default_user)
  327. self.assertEqual(testvm1.features, testvm3.features)
  328. self.assertEqual(testvm1.firewall.rules,
  329. testvm3.firewall.rules)
  330. finally:
  331. try:
  332. del firewall
  333. except NameError:
  334. pass
  335. try:
  336. del testvm1
  337. except NameError:
  338. pass
  339. try:
  340. del testvm2
  341. except NameError:
  342. pass
  343. try:
  344. del testvm3
  345. except NameError:
  346. pass
  347. def test_020_name_conflict_app(self):
  348. # TODO decide what exception should be here
  349. with self.assertRaises((qubes.exc.QubesException, ValueError)):
  350. self.vm2 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  351. name=self.vmname, template=self.app.default_template,
  352. label='red')
  353. self.loop.run_until_complete(self.vm2.create_on_disk())
  354. def test_021_name_conflict_template(self):
  355. # TODO decide what exception should be here
  356. with self.assertRaises((qubes.exc.QubesException, ValueError)):
  357. self.vm2 = self.app.add_new_vm(qubes.vm.templatevm.TemplateVM,
  358. name=self.vmname, label='red')
  359. self.loop.run_until_complete(self.vm2.create_on_disk())
  360. class TC_02_QvmPrefs(qubes.tests.SystemTestCase):
  361. # pylint: disable=attribute-defined-outside-init
  362. def setUp(self):
  363. super(TC_02_QvmPrefs, self).setUp()
  364. self.init_default_template()
  365. self.sharedopts = ['--qubesxml', qubes.tests.XMLPATH]
  366. def setup_appvm(self):
  367. self.testvm = self.app.add_new_vm(
  368. qubes.vm.appvm.AppVM,
  369. name=self.make_vm_name("vm"),
  370. label='red')
  371. self.loop.run_until_complete(self.testvm.create_on_disk())
  372. self.app.save()
  373. def setup_hvm(self):
  374. self.testvm = self.app.add_new_vm(
  375. qubes.vm.appvm.AppVM,
  376. name=self.make_vm_name("hvm"),
  377. label='red')
  378. self.testvm.virt_mode = 'hvm'
  379. self.loop.run_until_complete(self.testvm.create_on_disk())
  380. self.app.save()
  381. def pref_set(self, name, value, valid=True):
  382. self.loop.run_until_complete(self._pref_set(name, value, valid))
  383. @asyncio.coroutine
  384. def _pref_set(self, name, value, valid=True):
  385. cmd = ['qvm-prefs']
  386. if value != '-D':
  387. cmd.append('--')
  388. cmd.extend((self.testvm.name, name, value))
  389. p = yield from asyncio.create_subprocess_exec(*cmd,
  390. stdout=subprocess.PIPE,
  391. stderr=subprocess.PIPE)
  392. (stdout, stderr) = yield from p.communicate()
  393. if valid:
  394. self.assertEqual(p.returncode, 0,
  395. "qvm-prefs .. '{}' '{}' failed: {}{}".format(
  396. name, value, stdout, stderr
  397. ))
  398. else:
  399. self.assertNotEquals(p.returncode, 0,
  400. "qvm-prefs should reject value '{}' for "
  401. "property '{}'".format(value, name))
  402. def pref_get(self, name):
  403. self.loop.run_until_complete(self._pref_get(name))
  404. @asyncio.coroutine
  405. def _pref_get(self, name):
  406. p = yield from asyncio.create_subprocess_exec(
  407. 'qvm-prefs', *self.sharedopts, '--', self.testvm.name, name,
  408. stdout=subprocess.PIPE)
  409. (stdout, _) = yield from p.communicate()
  410. self.assertEqual(p.returncode, 0)
  411. return stdout.strip()
  412. bool_test_values = [
  413. ('true', 'True', True),
  414. ('False', 'False', True),
  415. ('0', 'False', True),
  416. ('1', 'True', True),
  417. ('invalid', '', False)
  418. ]
  419. def execute_tests(self, name, values):
  420. """
  421. Helper function, which executes tests for given property.
  422. :param values: list of tuples (value, expected, valid),
  423. where 'value' is what should be set and 'expected' is what should
  424. qvm-prefs returns as a property value and 'valid' marks valid and
  425. invalid values - if it's False, qvm-prefs should reject the value
  426. :return: None
  427. """
  428. for (value, expected, valid) in values:
  429. self.pref_set(name, value, valid)
  430. if valid:
  431. self.assertEqual(self.pref_get(name), expected)
  432. @unittest.skip('test not converted to core3 API')
  433. def test_006_template(self):
  434. templates = [tpl for tpl in self.app.domains.values() if
  435. isinstance(tpl, qubes.vm.templatevm.TemplateVM)]
  436. if not templates:
  437. self.skipTest("No templates installed")
  438. some_template = templates[0].name
  439. self.setup_appvm()
  440. self.execute_tests('template', [
  441. (some_template, some_template, True),
  442. ('invalid', '', False),
  443. ])
  444. @unittest.skip('test not converted to core3 API')
  445. def test_014_pcidevs(self):
  446. self.setup_appvm()
  447. self.execute_tests('pcidevs', [
  448. ('[]', '[]', True),
  449. ('[ "00:00.0" ]', "['00:00.0']", True),
  450. ('invalid', '', False),
  451. ('[invalid]', '', False),
  452. # TODO:
  453. # ('["12:12.0"]', '', False)
  454. ])
  455. @unittest.skip('test not converted to core3 API')
  456. def test_024_pv_reject_hvm_props(self):
  457. self.setup_appvm()
  458. self.execute_tests('guiagent_installed', [('False', '', False)])
  459. self.execute_tests('qrexec_installed', [('False', '', False)])
  460. self.execute_tests('drive', [('/tmp/drive.img', '', False)])
  461. self.execute_tests('timezone', [('localtime', '', False)])
  462. @unittest.skip('test not converted to core3 API')
  463. def test_025_hvm_reject_pv_props(self):
  464. self.setup_hvm()
  465. self.execute_tests('kernel', [('default', '', False)])
  466. self.execute_tests('kernelopts', [('default', '', False)])
  467. class TC_03_QvmRevertTemplateChanges(qubes.tests.SystemTestCase):
  468. # pylint: disable=attribute-defined-outside-init
  469. def setUp(self):
  470. super(TC_03_QvmRevertTemplateChanges, self).setUp()
  471. self.init_default_template()
  472. def cleanup_template(self):
  473. del self.test_template
  474. def setup_template(self):
  475. self.test_template = self.app.add_new_vm(
  476. qubes.vm.templatevm.TemplateVM,
  477. name=self.make_vm_name("pv-clone"),
  478. label='red'
  479. )
  480. self.addCleanup(self.cleanup_template)
  481. self.test_template.clone_properties(self.app.default_template)
  482. self.test_template.features.update(self.app.default_template.features)
  483. self.test_template.tags.update(self.app.default_template.tags)
  484. self.loop.run_until_complete(
  485. self.test_template.clone_disk_files(self.app.default_template))
  486. self.test_template.volumes['root'].revisions_to_keep = 3
  487. self.app.save()
  488. def get_rootimg_checksum(self):
  489. return subprocess.check_output(
  490. ['sha1sum', self.test_template.volumes['root'].path])
  491. def _do_test(self):
  492. checksum_before = self.get_rootimg_checksum()
  493. self.loop.run_until_complete(self.test_template.start())
  494. self.shutdown_and_wait(self.test_template)
  495. checksum_changed = self.get_rootimg_checksum()
  496. if checksum_before == checksum_changed:
  497. self.log.warning("template not modified, test result will be "
  498. "unreliable")
  499. self.assertNotEqual(self.test_template.volumes['root'].revisions, {})
  500. revert_cmd = ['qvm-volume', 'revert', self.test_template.name + ':root']
  501. p = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  502. *revert_cmd))
  503. self.loop.run_until_complete(p.wait())
  504. self.assertEqual(p.returncode, 0)
  505. del p
  506. checksum_after = self.get_rootimg_checksum()
  507. self.assertEqual(checksum_before, checksum_after)
  508. def test_000_revert_linux(self):
  509. """
  510. Test qvm-revert-template-changes for PV template
  511. """
  512. self.setup_template()
  513. self._do_test()
  514. @unittest.skip('TODO: some non-linux system')
  515. def test_001_revert_non_linux(self):
  516. """
  517. Test qvm-revert-template-changes for HVM template
  518. """
  519. # TODO: have some system there, so the root.img will get modified
  520. self.setup_template()
  521. self._do_test()
  522. class TC_30_Gui_daemon(qubes.tests.SystemTestCase):
  523. def setUp(self):
  524. super(TC_30_Gui_daemon, self).setUp()
  525. self.init_default_template()
  526. @unittest.skipUnless(
  527. spawn.find_executable('xdotool'),
  528. "xdotool not installed")
  529. def test_000_clipboard(self):
  530. testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  531. name=self.make_vm_name('vm1'), label='red')
  532. self.loop.run_until_complete(testvm1.create_on_disk())
  533. testvm2 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  534. name=self.make_vm_name('vm2'), label='red')
  535. self.loop.run_until_complete(testvm2.create_on_disk())
  536. self.app.save()
  537. self.loop.run_until_complete(asyncio.wait([
  538. testvm1.start(),
  539. testvm2.start()]))
  540. self.loop.run_until_complete(asyncio.wait([
  541. self.wait_for_session(testvm1),
  542. self.wait_for_session(testvm2)]))
  543. window_title = 'user@{}'.format(testvm1.name)
  544. self.loop.run_until_complete(testvm1.run(
  545. 'zenity --text-info --editable --title={}'.format(window_title)))
  546. self.wait_for_window(window_title)
  547. time.sleep(0.5)
  548. test_string = "test{}".format(testvm1.xid)
  549. # Type and copy some text
  550. subprocess.check_call(['xdotool', 'search', '--name', window_title,
  551. 'windowactivate', '--sync',
  552. 'type', test_string])
  553. # second xdotool call because type --terminator do not work (SEGV)
  554. # additionally do not use search here, so window stack will be empty
  555. # and xdotool will use XTEST instead of generating events manually -
  556. # this will be much better - at least because events will have
  557. # correct timestamp (so gui-daemon would not drop the copy request)
  558. subprocess.check_call(['xdotool',
  559. 'key', 'ctrl+a', 'ctrl+c', 'ctrl+shift+c',
  560. 'Escape'])
  561. clipboard_content = \
  562. open('/var/run/qubes/qubes-clipboard.bin', 'r').read().strip()
  563. self.assertEqual(clipboard_content, test_string,
  564. "Clipboard copy operation failed - content")
  565. clipboard_source = \
  566. open('/var/run/qubes/qubes-clipboard.bin.source',
  567. 'r').read().strip()
  568. self.assertEqual(clipboard_source, testvm1.name,
  569. "Clipboard copy operation failed - owner")
  570. # Then paste it to the other window
  571. window_title = 'user@{}'.format(testvm2.name)
  572. p = self.loop.run_until_complete(testvm2.run(
  573. 'zenity --entry --title={} > /tmp/test.txt'.format(window_title)))
  574. self.wait_for_window(window_title)
  575. subprocess.check_call(['xdotool', 'key', '--delay', '100',
  576. 'ctrl+shift+v', 'ctrl+v', 'Return'])
  577. self.loop.run_until_complete(p.wait())
  578. # And compare the result
  579. (test_output, _) = self.loop.run_until_complete(
  580. testvm2.run_for_stdio('cat /tmp/test.txt'))
  581. self.assertEqual(test_string, test_output.strip().decode('ascii'))
  582. clipboard_content = \
  583. open('/var/run/qubes/qubes-clipboard.bin', 'r').read().strip()
  584. self.assertEqual(clipboard_content, "",
  585. "Clipboard not wiped after paste - content")
  586. clipboard_source = \
  587. open('/var/run/qubes/qubes-clipboard.bin.source', 'r').\
  588. read().strip()
  589. self.assertEqual(clipboard_source, "",
  590. "Clipboard not wiped after paste - owner")
  591. class TC_05_StandaloneVMMixin(object):
  592. def setUp(self):
  593. super(TC_05_StandaloneVMMixin, self).setUp()
  594. self.init_default_template(self.template)
  595. def test_000_create_start(self):
  596. self.testvm1 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
  597. name=self.make_vm_name('vm1'), label='red')
  598. self.testvm1.features.update(self.app.default_template.features)
  599. self.loop.run_until_complete(
  600. self.testvm1.clone_disk_files(self.app.default_template))
  601. self.app.save()
  602. self.loop.run_until_complete(self.testvm1.start())
  603. self.assertEqual(self.testvm1.get_power_state(), "Running")
  604. def test_100_resize_root_img(self):
  605. self.testvm1 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
  606. name=self.make_vm_name('vm1'), label='red')
  607. self.testvm1.features.update(self.app.default_template.features)
  608. self.loop.run_until_complete(
  609. self.testvm1.clone_disk_files(self.app.default_template))
  610. self.app.save()
  611. try:
  612. self.loop.run_until_complete(
  613. self.testvm1.storage.resize(self.testvm1.volumes['root'],
  614. 20 * 1024 ** 3))
  615. except (subprocess.CalledProcessError,
  616. qubes.storage.StoragePoolException) as e:
  617. # exception object would leak VM reference
  618. self.fail(str(e))
  619. self.assertEqual(self.testvm1.volumes['root'].size, 20 * 1024 ** 3)
  620. self.loop.run_until_complete(self.testvm1.start())
  621. # new_size in 1k-blocks
  622. (new_size, _) = self.loop.run_until_complete(
  623. self.testvm1.run_for_stdio('df --output=size /|tail -n 1'))
  624. # some safety margin for FS metadata
  625. self.assertGreater(int(new_size.strip()), 19 * 1024 ** 2)
  626. def test_101_resize_root_img_online(self):
  627. self.testvm1 = self.app.add_new_vm(qubes.vm.standalonevm.StandaloneVM,
  628. name=self.make_vm_name('vm1'), label='red')
  629. self.testvm1.features['qrexec'] = True
  630. self.loop.run_until_complete(
  631. self.testvm1.clone_disk_files(self.app.default_template))
  632. self.testvm1.features.update(self.app.default_template.features)
  633. self.app.save()
  634. self.loop.run_until_complete(self.testvm1.start())
  635. try:
  636. self.loop.run_until_complete(
  637. self.testvm1.storage.resize(self.testvm1.volumes['root'],
  638. 20 * 1024 ** 3))
  639. except (subprocess.CalledProcessError,
  640. qubes.storage.StoragePoolException) as e:
  641. # exception object would leak VM reference
  642. self.fail(str(e))
  643. self.assertEqual(self.testvm1.volumes['root'].size, 20 * 1024 ** 3)
  644. # new_size in 1k-blocks
  645. (new_size, _) = self.loop.run_until_complete(
  646. self.testvm1.run_for_stdio('df --output=size /|tail -n 1'))
  647. # some safety margin for FS metadata
  648. self.assertGreater(int(new_size.strip()), 19 * 1024 ** 2)
  649. class TC_06_AppVMMixin(object):
  650. template = None
  651. def setUp(self):
  652. super(TC_06_AppVMMixin, self).setUp()
  653. self.init_default_template(self.template)
  654. @unittest.skipUnless(
  655. spawn.find_executable('xdotool'), "xdotool not installed")
  656. def test_121_start_standalone_with_cdrom_vm(self):
  657. cdrom_vmname = self.make_vm_name('cdrom')
  658. self.cdrom_vm = self.app.add_new_vm('AppVM', label='red',
  659. name=cdrom_vmname)
  660. self.loop.run_until_complete(self.cdrom_vm.create_on_disk())
  661. self.loop.run_until_complete(self.cdrom_vm.start())
  662. iso_path = self.create_bootable_iso()
  663. with open(iso_path, 'rb') as iso_f:
  664. self.loop.run_until_complete(
  665. self.cdrom_vm.run_for_stdio('cat > /home/user/boot.iso',
  666. stdin=iso_f))
  667. vmname = self.make_vm_name('appvm')
  668. self.vm = self.app.add_new_vm('StandaloneVM', label='red', name=vmname)
  669. self.loop.run_until_complete(self.vm.create_on_disk())
  670. self.vm.kernel = None
  671. self.vm.virt_mode = 'hvm'
  672. # start the VM using qvm-start tool, to test --cdrom option there
  673. p = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  674. 'qvm-start', '--cdrom=' + cdrom_vmname + ':/home/user/boot.iso',
  675. self.vm.name,
  676. stdout=subprocess.PIPE, stderr=subprocess.STDOUT))
  677. (stdout, _) = self.loop.run_until_complete(p.communicate())
  678. self.assertEqual(p.returncode, 0, stdout)
  679. # check if VM do not crash instantly
  680. self.loop.run_until_complete(asyncio.sleep(5))
  681. self.assertTrue(self.vm.is_running())
  682. # Type 'poweroff'
  683. subprocess.check_call(['xdotool', 'search', '--name', self.vm.name,
  684. 'type', 'poweroff\r'])
  685. self.loop.run_until_complete(asyncio.sleep(1))
  686. self.assertFalse(self.vm.is_running())
  687. def load_tests(loader, tests, pattern):
  688. tests.addTests(loader.loadTestsFromNames(
  689. qubes.tests.create_testcases_for_templates('TC_05_StandaloneVM',
  690. TC_05_StandaloneVMMixin, qubes.tests.SystemTestCase,
  691. module=sys.modules[__name__])))
  692. tests.addTests(loader.loadTestsFromNames(
  693. qubes.tests.create_testcases_for_templates('TC_06_AppVM',
  694. TC_06_AppVMMixin, qubes.tests.SystemTestCase,
  695. module=sys.modules[__name__])))
  696. return tests
  697. # vim: ts=4 sw=4 et