dispvm.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270
  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 program is free software; you can redistribute it and/or
  8. # modify it under the terms of the GNU General Public License
  9. # as published by the Free Software Foundation; either version 2
  10. # of the License, or (at your option) any later version.
  11. #
  12. # This program 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
  15. # GNU General Public License for more details.
  16. #
  17. # You should have received a copy of the GNU General Public License
  18. # along with this program; if not, write to the Free Software
  19. # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
  20. #
  21. import os
  22. import subprocess
  23. import tempfile
  24. import time
  25. import unittest
  26. from distutils import spawn
  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.disp_base.create_on_disk()
  37. self.app.default_dispvm = self.disp_base
  38. self.testvm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  39. name=self.make_vm_name('vm'),
  40. label='red',
  41. )
  42. self.testvm.create_on_disk()
  43. self.app.save()
  44. @unittest.expectedFailure
  45. def test_002_cleanup(self):
  46. self.testvm.start()
  47. p = self.testvm.run("qvm-run --dispvm bash", passio_popen=True)
  48. (stdout, _) = p.communicate(input=b"echo test; qubesdb-read /name; "
  49. b"echo ERROR\n")
  50. self.assertEqual(p.returncode, 0)
  51. lines = stdout.decode('ascii').splitlines()
  52. self.assertEqual(lines[0], "test")
  53. dispvm_name = lines[1]
  54. # wait for actual DispVM destruction
  55. time.sleep(1)
  56. self.assertNotIn(dispvm_name, self.app.domains)
  57. @unittest.expectedFailure
  58. def test_003_cleanup_destroyed(self):
  59. """
  60. Check if DispVM is properly removed even if it terminated itself (#1660)
  61. :return:
  62. """
  63. self.testvm.start()
  64. p = self.testvm.run("qvm-run --dispvm bash; true", passio_popen=True)
  65. p.stdin.write(b"qubesdb-read /name\n")
  66. p.stdin.write(b"echo ERROR\n")
  67. p.stdin.write(b"sudo poweroff\n")
  68. # do not close p.stdin on purpose - wait to automatic disconnect when
  69. # domain is destroyed
  70. timeout = 30
  71. while timeout > 0:
  72. if p.poll():
  73. break
  74. time.sleep(1)
  75. timeout -= 1
  76. # includes check for None - timeout
  77. self.assertEqual(p.returncode, 0)
  78. lines = p.stdout.read().splitlines()
  79. self.assertTrue(lines, 'No output received from DispVM')
  80. dispvm_name = lines[0]
  81. self.assertNotEquals(dispvm_name, b"ERROR")
  82. self.assertNotIn(dispvm_name, self.app.domains)
  83. class TC_20_DispVMMixin(object):
  84. def setUp(self):
  85. super(TC_20_DispVMMixin, self).setUp()
  86. self.init_default_template(self.template)
  87. self.disp_base = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  88. name=self.make_vm_name('dvm'),
  89. label='red', template_for_dispvms=True,
  90. )
  91. self.loop.run_until_complete(self.disp_base.create_on_disk())
  92. self.app.default_dispvm = self.disp_base
  93. self.app.save()
  94. def test_010_simple_dvm_run(self):
  95. dispvm = self.loop.run_until_complete(
  96. qubes.vm.dispvm.DispVM.from_appvm(self.disp_base))
  97. try:
  98. self.loop.run_until_complete(dispvm.start())
  99. p = self.loop.run_until_complete(
  100. dispvm.run_service('qubes.VMShell', passio_popen=True))
  101. (stdout, _) = p.communicate(input=b"echo test")
  102. self.assertEqual(stdout, b"test\n")
  103. finally:
  104. self.loop.run_until_complete(dispvm.cleanup())
  105. @unittest.skipUnless(spawn.find_executable('xdotool'),
  106. "xdotool not installed")
  107. def test_020_gui_app(self):
  108. dispvm = self.loop.run_until_complete(
  109. qubes.vm.dispvm.DispVM.from_appvm(self.disp_base))
  110. try:
  111. self.loop.run_until_complete(dispvm.start())
  112. p = self.loop.run_until_complete(
  113. dispvm.run_service('qubes.VMShell', passio_popen=True))
  114. # wait for DispVM startup:
  115. p.stdin.write(b"echo test\n")
  116. p.stdin.flush()
  117. l = self.loop.run_until_complete(p.stdout.readline())
  118. self.assertEqual(l, b"test\n")
  119. self.assertTrue(dispvm.is_running())
  120. try:
  121. window_title = 'user@%s' % (dispvm.name,)
  122. p.stdin.write("xterm -e "
  123. "\"sh -c 'echo \\\"\033]0;{}\007\\\";read x;'\"\n".
  124. format(window_title).encode())
  125. self.wait_for_window(window_title)
  126. time.sleep(0.5)
  127. self.enter_keys_in_window(window_title, ['Return'])
  128. # Wait for window to close
  129. self.wait_for_window(window_title, show=False)
  130. finally:
  131. p.stdin.close()
  132. finally:
  133. self.loop.run_until_complete(dispvm.cleanup())
  134. self.assertNotIn(dispvm.name, self.app.domains,
  135. "DispVM not removed from qubes.xml")
  136. def _handle_editor(self, winid):
  137. (window_title, _) = subprocess.Popen(
  138. ['xdotool', 'getwindowname', winid], stdout=subprocess.PIPE).\
  139. communicate()
  140. window_title = window_title.decode().strip().\
  141. replace('(', '\(').replace(')', '\)')
  142. time.sleep(1)
  143. if "gedit" in window_title:
  144. subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid,
  145. 'type', 'Test test 2'])
  146. subprocess.check_call(['xdotool', 'key', '--window', winid,
  147. 'key', 'Return'])
  148. time.sleep(0.5)
  149. subprocess.check_call(['xdotool',
  150. 'key', 'ctrl+s', 'ctrl+q'])
  151. elif "LibreOffice" in window_title:
  152. # wait for actual editor (we've got splash screen)
  153. search = subprocess.Popen(['xdotool', 'search', '--sync',
  154. '--onlyvisible', '--all', '--name', '--class', 'disp*|Writer'],
  155. stdout=subprocess.PIPE,
  156. stderr=open(os.path.devnull, 'w'))
  157. retcode = search.wait()
  158. if retcode == 0:
  159. winid = search.stdout.read().strip()
  160. time.sleep(0.5)
  161. subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid,
  162. 'type', 'Test test 2'])
  163. subprocess.check_call(['xdotool', 'key', '--window', winid,
  164. 'key', 'Return'])
  165. time.sleep(0.5)
  166. subprocess.check_call(['xdotool',
  167. 'key', '--delay', '100', 'ctrl+s',
  168. 'Return', 'ctrl+q'])
  169. elif "emacs" in window_title:
  170. subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid,
  171. 'type', 'Test test 2'])
  172. subprocess.check_call(['xdotool', 'key', '--window', winid,
  173. 'key', 'Return'])
  174. time.sleep(0.5)
  175. subprocess.check_call(['xdotool',
  176. 'key', 'ctrl+x', 'ctrl+s'])
  177. subprocess.check_call(['xdotool',
  178. 'key', 'ctrl+x', 'ctrl+c'])
  179. elif "vim" in window_title or "user@" in window_title:
  180. subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid,
  181. 'key', 'i', 'type', 'Test test 2'])
  182. subprocess.check_call(['xdotool', 'key', '--window', winid,
  183. 'key', 'Return'])
  184. subprocess.check_call(
  185. ['xdotool',
  186. 'key', 'Escape', 'colon', 'w', 'q', 'Return'])
  187. else:
  188. self.fail("Unknown editor window: {}".format(window_title))
  189. @unittest.skipUnless(spawn.find_executable('xdotool'),
  190. "xdotool not installed")
  191. @unittest.expectedFailure
  192. def test_030_edit_file(self):
  193. testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  194. name=self.make_vm_name('vm1'),
  195. label='red',
  196. template=self.app.domains[self.template])
  197. self.loop.run_until_complete(testvm1.create_on_disk())
  198. self.app.save()
  199. self.loop.run_until_complete(testvm1.start())
  200. self.loop.run_until_complete(
  201. testvm1.run_for_stdio("echo test1 > /home/user/test.txt"))
  202. p = self.loop.run_until_complete(
  203. testvm1.run("qvm-open-in-dvm /home/user/test.txt"))
  204. wait_count = 0
  205. winid = None
  206. while True:
  207. search = subprocess.Popen(['xdotool', 'search',
  208. '--onlyvisible', '--class', 'disp*'],
  209. stdout=subprocess.PIPE,
  210. stderr=open(os.path.devnull, 'w'))
  211. retcode = search.wait()
  212. if retcode == 0:
  213. winid = search.stdout.read().strip()
  214. # get window title
  215. (window_title, _) = subprocess.Popen(
  216. ['xdotool', 'getwindowname', winid], stdout=subprocess.PIPE). \
  217. communicate()
  218. window_title = window_title.decode().strip()
  219. # ignore LibreOffice splash screen and window with no title
  220. # set yet
  221. if window_title and not window_title.startswith("LibreOffice")\
  222. and not window_title == 'VMapp command':
  223. break
  224. wait_count += 1
  225. if wait_count > 100:
  226. self.fail("Timeout while waiting for editor window")
  227. time.sleep(0.3)
  228. time.sleep(0.5)
  229. self._handle_editor(winid)
  230. self.loop.run_until_complete(p.wait())
  231. (test_txt_content, _) = self.loop.run_until_complete(
  232. testvm1.run_for_stdio("cat /home/user/test.txt"))
  233. # Drop BOM if added by editor
  234. if test_txt_content.startswith(b'\xef\xbb\xbf'):
  235. test_txt_content = test_txt_content[3:]
  236. self.assertEqual(test_txt_content, b"Test test 2\ntest1\n")
  237. def load_tests(loader, tests, pattern):
  238. for template in qubes.tests.list_templates():
  239. tests.addTests(loader.loadTestsFromTestCase(
  240. type(
  241. 'TC_20_DispVM_' + template,
  242. (TC_20_DispVMMixin, qubes.tests.SystemTestCase),
  243. {'template': template})))
  244. return tests