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