dispvm.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292
  1. #
  2. # The Qubes OS Project, http://www.qubes-os.org
  3. #
  4. # Copyright (C) 2016 Marek Marczykowski-Górecki
  5. # <marmarek@invisiblethingslab.com>
  6. #
  7. # This library is free software; you can redistribute it and/or
  8. # modify it under the terms of the GNU Lesser General Public
  9. # License as published by the Free Software Foundation; either
  10. # version 2.1 of the License, or (at your option) any later version.
  11. #
  12. # This library is distributed in the hope that it will be useful,
  13. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  15. # Lesser General Public License for more details.
  16. #
  17. # You should have received a copy of the GNU Lesser General Public
  18. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  19. #
  20. import os
  21. import subprocess
  22. import tempfile
  23. import time
  24. import unittest
  25. from distutils import spawn
  26. import asyncio
  27. import sys
  28. import qubes.tests
  29. class TC_04_DispVM(qubes.tests.SystemTestCase):
  30. def setUp(self):
  31. super(TC_04_DispVM, self).setUp()
  32. self.init_default_template()
  33. self.disp_base = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  34. name=self.make_vm_name('dvm'),
  35. label='red',
  36. )
  37. self.loop.run_until_complete(self.disp_base.create_on_disk())
  38. self.disp_base.template_for_dispvms = True
  39. self.app.default_dispvm = self.disp_base
  40. self.testvm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  41. name=self.make_vm_name('vm'),
  42. label='red',
  43. )
  44. self.loop.run_until_complete(self.testvm.create_on_disk())
  45. self.app.save()
  46. def tearDown(self):
  47. self.app.default_dispvm = None
  48. super(TC_04_DispVM, self).tearDown()
  49. def test_002_cleanup(self):
  50. self.loop.run_until_complete(self.testvm.start())
  51. try:
  52. (stdout, _) = self.loop.run_until_complete(
  53. self.testvm.run_for_stdio("qvm-run-vm --dispvm bash",
  54. input=b"echo test; qubesdb-read /name; echo ERROR\n"))
  55. except subprocess.CalledProcessError as err:
  56. self.fail('qvm-run-vm failed with {} code, stderr: {}'.format(
  57. err.returncode, err.stderr))
  58. lines = stdout.decode('ascii').splitlines()
  59. self.assertEqual(lines[0], "test")
  60. dispvm_name = lines[1]
  61. # wait for actual DispVM destruction
  62. self.loop.run_until_complete(asyncio.sleep(5))
  63. self.assertNotIn(dispvm_name, self.app.domains)
  64. def test_003_cleanup_destroyed(self):
  65. """
  66. Check if DispVM is properly removed even if it terminated itself (#1660)
  67. :return:
  68. """
  69. self.loop.run_until_complete(self.testvm.start())
  70. p = self.loop.run_until_complete(
  71. self.testvm.run("qvm-run-vm --dispvm bash; true",
  72. stdin=subprocess.PIPE, stdout=subprocess.PIPE))
  73. p.stdin.write(b"qubesdb-read /name\n")
  74. p.stdin.write(b"echo ERROR\n")
  75. p.stdin.write(b"sudo poweroff\n")
  76. # do not close p.stdin on purpose - wait to automatic disconnect when
  77. # domain is destroyed
  78. timeout = 70
  79. lines_task = asyncio.ensure_future(p.stdout.read())
  80. self.loop.run_until_complete(asyncio.wait_for(p.wait(), timeout))
  81. self.loop.run_until_complete(lines_task)
  82. lines = lines_task.result().splitlines()
  83. self.assertTrue(lines, 'No output received from DispVM')
  84. dispvm_name = lines[0]
  85. self.assertNotEquals(dispvm_name, b"ERROR")
  86. self.assertNotIn(dispvm_name, self.app.domains)
  87. class TC_20_DispVMMixin(object):
  88. def setUp(self):
  89. super(TC_20_DispVMMixin, self).setUp()
  90. self.init_default_template(self.template)
  91. self.disp_base = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  92. name=self.make_vm_name('dvm'),
  93. label='red', template_for_dispvms=True,
  94. )
  95. self.loop.run_until_complete(self.disp_base.create_on_disk())
  96. self.app.default_dispvm = self.disp_base
  97. self.app.save()
  98. def tearDown(self):
  99. self.app.default_dispvm = None
  100. super(TC_20_DispVMMixin, self).tearDown()
  101. def test_010_simple_dvm_run(self):
  102. dispvm = self.loop.run_until_complete(
  103. qubes.vm.dispvm.DispVM.from_appvm(self.disp_base))
  104. try:
  105. self.loop.run_until_complete(dispvm.start())
  106. (stdout, _) = self.loop.run_until_complete(
  107. dispvm.run_service_for_stdio('qubes.VMShell',
  108. input=b"echo test"))
  109. self.assertEqual(stdout, b"test\n")
  110. finally:
  111. self.loop.run_until_complete(dispvm.cleanup())
  112. @unittest.skipUnless(spawn.find_executable('xdotool'),
  113. "xdotool not installed")
  114. def test_020_gui_app(self):
  115. dispvm = self.loop.run_until_complete(
  116. qubes.vm.dispvm.DispVM.from_appvm(self.disp_base))
  117. try:
  118. self.loop.run_until_complete(dispvm.start())
  119. self.loop.run_until_complete(self.wait_for_session(dispvm))
  120. p = self.loop.run_until_complete(
  121. dispvm.run_service('qubes.VMShell',
  122. stdin=subprocess.PIPE,
  123. stdout=subprocess.PIPE))
  124. # wait for DispVM startup:
  125. p.stdin.write(b"echo test\n")
  126. self.loop.run_until_complete(p.stdin.drain())
  127. l = self.loop.run_until_complete(p.stdout.readline())
  128. self.assertEqual(l, b"test\n")
  129. self.assertTrue(dispvm.is_running())
  130. try:
  131. window_title = 'user@%s' % (dispvm.name,)
  132. # close xterm on Return, but after short delay, to allow
  133. # xdotool to send also keyup event
  134. p.stdin.write("xterm -e "
  135. "\"sh -c 'echo \\\"\033]0;{}\007\\\";read x;"
  136. "sleep 0.1;'\"\n".
  137. format(window_title).encode())
  138. self.loop.run_until_complete(p.stdin.drain())
  139. self.wait_for_window(window_title)
  140. time.sleep(0.5)
  141. self.enter_keys_in_window(window_title, ['Return'])
  142. # Wait for window to close
  143. self.wait_for_window(window_title, show=False)
  144. finally:
  145. p.stdin.close()
  146. del p
  147. finally:
  148. self.loop.run_until_complete(dispvm.cleanup())
  149. dispvm_name = dispvm.name
  150. del dispvm
  151. # give it a time for shutdown + cleanup
  152. self.loop.run_until_complete(asyncio.sleep(5))
  153. self.assertNotIn(dispvm_name, self.app.domains,
  154. "DispVM not removed from qubes.xml")
  155. def _handle_editor(self, winid):
  156. (window_title, _) = subprocess.Popen(
  157. ['xdotool', 'getwindowname', winid], stdout=subprocess.PIPE).\
  158. communicate()
  159. window_title = window_title.decode().strip().\
  160. replace('(', '\(').replace(')', '\)')
  161. time.sleep(1)
  162. if "gedit" in window_title:
  163. subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid,
  164. 'type', 'Test test 2'])
  165. subprocess.check_call(['xdotool', 'key', '--window', winid,
  166. 'key', 'Return'])
  167. time.sleep(0.5)
  168. subprocess.check_call(['xdotool',
  169. 'key', 'ctrl+s', 'ctrl+q'])
  170. elif "LibreOffice" in window_title:
  171. # wait for actual editor (we've got splash screen)
  172. search = subprocess.Popen(['xdotool', 'search', '--sync',
  173. '--onlyvisible', '--all', '--name', '--class', 'disp*|Writer'],
  174. stdout=subprocess.PIPE,
  175. stderr=open(os.path.devnull, 'w'))
  176. retcode = search.wait()
  177. if retcode == 0:
  178. winid = search.stdout.read().strip()
  179. time.sleep(0.5)
  180. subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid,
  181. 'type', 'Test test 2'])
  182. subprocess.check_call(['xdotool', 'key', '--window', winid,
  183. 'key', 'Return'])
  184. time.sleep(0.5)
  185. subprocess.check_call(['xdotool',
  186. 'key', '--delay', '100', 'ctrl+s',
  187. 'Return', 'ctrl+q'])
  188. elif "emacs" in window_title:
  189. subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid,
  190. 'type', 'Test test 2'])
  191. subprocess.check_call(['xdotool', 'key', '--window', winid,
  192. 'key', 'Return'])
  193. time.sleep(0.5)
  194. subprocess.check_call(['xdotool',
  195. 'key', 'ctrl+x', 'ctrl+s'])
  196. subprocess.check_call(['xdotool',
  197. 'key', 'ctrl+x', 'ctrl+c'])
  198. elif "vim" in window_title or "user@" in window_title:
  199. subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid,
  200. 'key', 'i', 'type', 'Test test 2'])
  201. subprocess.check_call(['xdotool', 'key', '--window', winid,
  202. 'key', 'Return'])
  203. subprocess.check_call(
  204. ['xdotool',
  205. 'key', 'Escape', 'colon', 'w', 'q', 'Return'])
  206. else:
  207. self.fail("Unknown editor window: {}".format(window_title))
  208. @unittest.skipUnless(spawn.find_executable('xdotool'),
  209. "xdotool not installed")
  210. def test_030_edit_file(self):
  211. self.testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  212. name=self.make_vm_name('vm1'),
  213. label='red',
  214. template=self.app.domains[self.template])
  215. self.loop.run_until_complete(self.testvm1.create_on_disk())
  216. self.app.save()
  217. self.loop.run_until_complete(self.testvm1.start())
  218. self.loop.run_until_complete(
  219. self.testvm1.run_for_stdio("echo test1 > /home/user/test.txt"))
  220. p = self.loop.run_until_complete(
  221. self.testvm1.run("qvm-open-in-dvm /home/user/test.txt"))
  222. wait_count = 0
  223. winid = None
  224. while True:
  225. search = self.loop.run_until_complete(
  226. asyncio.create_subprocess_exec(
  227. 'xdotool', 'search', '--onlyvisible', '--class',
  228. 'disp[0-9]*',
  229. stdout=subprocess.PIPE,
  230. stderr=subprocess.DEVNULL))
  231. stdout, _ = self.loop.run_until_complete(search.communicate())
  232. if search.returncode == 0:
  233. winid = stdout.strip()
  234. # get window title
  235. (window_title, _) = subprocess.Popen(
  236. ['xdotool', 'getwindowname', winid], stdout=subprocess.PIPE). \
  237. communicate()
  238. window_title = window_title.decode().strip()
  239. # ignore LibreOffice splash screen and window with no title
  240. # set yet
  241. if window_title and not window_title.startswith("LibreOffice")\
  242. and not window_title == 'VMapp command':
  243. break
  244. wait_count += 1
  245. if wait_count > 100:
  246. self.fail("Timeout while waiting for editor window")
  247. self.loop.run_until_complete(asyncio.sleep(0.3))
  248. time.sleep(0.5)
  249. self._handle_editor(winid)
  250. self.loop.run_until_complete(p.wait())
  251. (test_txt_content, _) = self.loop.run_until_complete(
  252. self.testvm1.run_for_stdio("cat /home/user/test.txt"))
  253. # Drop BOM if added by editor
  254. if test_txt_content.startswith(b'\xef\xbb\xbf'):
  255. test_txt_content = test_txt_content[3:]
  256. self.assertEqual(test_txt_content, b"Test test 2\ntest1\n")
  257. def load_tests(loader, tests, pattern):
  258. tests.addTests(loader.loadTestsFromNames(
  259. qubes.tests.create_testcases_for_templates('TC_20_DispVM',
  260. TC_20_DispVMMixin, qubes.tests.SystemTestCase,
  261. module=sys.modules[__name__])))
  262. return tests