#!/usr/bin/python # -*- coding: utf-8 -*- # # The Qubes OS Project, http://www.qubes-os.org # # Copyright (C) 2016 Marek Marczykowski-Górecki # # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. # # from distutils import spawn import qubes.tests import subprocess import tempfile import unittest import os import time class TC_04_DispVM(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase): def setUp(self): super(TC_04_DispVM, self).setUp() self.init_default_template() self.disp_base = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=self.make_vm_name('dvm'), label='red', ) self.disp_base.create_on_disk() self.app.default_dispvm = self.disp_base self.testvm = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=self.make_vm_name('vm'), label='red', ) self.testvm.create_on_disk() self.app.save() def test_002_cleanup(self): self.testvm.start() p = self.testvm.run("qvm-run --dispvm bash", passio_popen=True) (stdout, _) = p.communicate(input="echo test; qubesdb-read /name; " "echo ERROR\n") self.assertEquals(p.returncode, 0) lines = stdout.splitlines() self.assertEqual(lines[0], "test") dispvm_name = lines[1] # wait for actual DispVM destruction time.sleep(1) self.reload_db() self.assertNotIn(dispvm_name, self.app.domains) def test_003_cleanup_destroyed(self): """ Check if DispVM is properly removed even if it terminated itself (#1660) :return: """ self.testvm.start() p = self.testvm.run("qvm-run --dispvm bash; true", passio_popen=True) p.stdin.write("qubesdb-read /name\n") p.stdin.write("echo ERROR\n") p.stdin.write("sudo poweroff\n") # do not close p.stdin on purpose - wait to automatic disconnect when # domain is destroyed timeout = 30 while timeout > 0: if p.poll(): break time.sleep(1) timeout -= 1 # includes check for None - timeout self.assertEquals(p.returncode, 0) lines = p.stdout.read().splitlines() self.assertTrue(lines, 'No output received from DispVM') dispvm_name = lines[0] self.assertNotEquals(dispvm_name, "ERROR") self.reload_db() self.assertNotIn(dispvm_name, self.app.domains) class TC_20_DispVMMixin(qubes.tests.SystemTestsMixin): def setUp(self): super(TC_20_DispVMMixin, self).setUp() self.init_default_template(self.template) self.disp_base = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=self.make_vm_name('dvm'), label='red', ) self.disp_base.create_on_disk() self.app.default_dispvm = self.disp_base self.app.save() def test_010_simple_dvm_run(self): dispvm = qubes.vm.dispvm.DispVM.from_appvm(self.disp_base) try: dispvm.start() p = dispvm.run_service('qubes.VMShell', passio_popen=True) (stdout, _) = p.communicate(input="echo test") self.assertEqual(stdout, "test\n") finally: dispvm.cleanup() @unittest.skipUnless(spawn.find_executable('xdotool'), "xdotool not installed") def test_020_gui_app(self): dispvm = qubes.vm.dispvm.DispVM.from_appvm(self.disp_base) try: dispvm.start() p = dispvm.run_service('qubes.VMShell', passio_popen=True) # wait for DispVM startup: p.stdin.write("echo test\n") p.stdin.flush() l = p.stdout.readline() self.assertEqual(l, "test\n") self.assertTrue(dispvm.is_running()) try: window_title = 'user@%s' % (dispvm.name,) p.stdin.write("xterm -e " "\"sh -c 'echo \\\"\033]0;{}\007\\\";read x;'\"\n". format(window_title)) self.wait_for_window(window_title) time.sleep(0.5) self.enter_keys_in_window(window_title, ['Return']) # Wait for window to close self.wait_for_window(window_title, show=False) finally: p.stdin.close() finally: dispvm.cleanup() self.reload_db() self.assertNotIn(dispvm.name, self.app.domains, "DispVM not removed from qubes.xml") def _handle_editor(self, winid): (window_title, _) = subprocess.Popen( ['xdotool', 'getwindowname', winid], stdout=subprocess.PIPE).\ communicate() window_title = window_title.strip().\ replace('(', '\(').replace(')', '\)') time.sleep(1) if "gedit" in window_title: subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid, 'type', 'Test test 2']) subprocess.check_call(['xdotool', 'key', '--window', winid, 'key', 'Return']) time.sleep(0.5) subprocess.check_call(['xdotool', 'key', 'ctrl+s', 'ctrl+q']) elif "LibreOffice" in window_title: # wait for actual editor (we've got splash screen) search = subprocess.Popen(['xdotool', 'search', '--sync', '--onlyvisible', '--all', '--name', '--class', 'disp*|Writer'], stdout=subprocess.PIPE, stderr=open(os.path.devnull, 'w')) retcode = search.wait() if retcode == 0: winid = search.stdout.read().strip() time.sleep(0.5) subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid, 'type', 'Test test 2']) subprocess.check_call(['xdotool', 'key', '--window', winid, 'key', 'Return']) time.sleep(0.5) subprocess.check_call(['xdotool', 'key', '--delay', '100', 'ctrl+s', 'Return', 'ctrl+q']) elif "emacs" in window_title: subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid, 'type', 'Test test 2']) subprocess.check_call(['xdotool', 'key', '--window', winid, 'key', 'Return']) time.sleep(0.5) subprocess.check_call(['xdotool', 'key', 'ctrl+x', 'ctrl+s']) subprocess.check_call(['xdotool', 'key', 'ctrl+x', 'ctrl+c']) elif "vim" in window_title or "user@" in window_title: subprocess.check_call(['xdotool', 'windowactivate', '--sync', winid, 'key', 'i', 'type', 'Test test 2']) subprocess.check_call(['xdotool', 'key', '--window', winid, 'key', 'Return']) subprocess.check_call( ['xdotool', 'key', 'Escape', 'colon', 'w', 'q', 'Return']) else: self.fail("Unknown editor window: {}".format(window_title)) @unittest.skipUnless(spawn.find_executable('xdotool'), "xdotool not installed") def test_030_edit_file(self): testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM, name=self.make_vm_name('vm1'), label='red', template=self.app.domains[self.template]) testvm1.create_on_disk() self.app.save() testvm1.start() testvm1.run("echo test1 > /home/user/test.txt", wait=True) p = testvm1.run("qvm-open-in-dvm /home/user/test.txt", passio_popen=True) wait_count = 0 winid = None while True: search = subprocess.Popen(['xdotool', 'search', '--onlyvisible', '--class', 'disp*'], stdout=subprocess.PIPE, stderr=open(os.path.devnull, 'w')) retcode = search.wait() if retcode == 0: winid = search.stdout.read().strip() # get window title (window_title, _) = subprocess.Popen( ['xdotool', 'getwindowname', winid], stdout=subprocess.PIPE). \ communicate() window_title = window_title.strip() # ignore LibreOffice splash screen and window with no title # set yet if window_title and not window_title.startswith("LibreOffice")\ and not window_title == 'VMapp command': break wait_count += 1 if wait_count > 100: self.fail("Timeout while waiting for editor window") time.sleep(0.3) time.sleep(0.5) self._handle_editor(winid) p.wait() p = testvm1.run("cat /home/user/test.txt", passio_popen=True) (test_txt_content, _) = p.communicate() # Drop BOM if added by editor if test_txt_content.startswith('\xef\xbb\xbf'): test_txt_content = test_txt_content[3:] self.assertEqual(test_txt_content, "Test test 2\ntest1\n") def load_tests(loader, tests, pattern): try: app = qubes.Qubes() templates = [vm.name for vm in app.domains if isinstance(vm, qubes.vm.templatevm.TemplateVM)] except OSError: templates = [] for template in templates: tests.addTests(loader.loadTestsFromTestCase( type( 'TC_20_DispVM_' + template, (TC_20_DispVMMixin, qubes.tests.QubesTestCase), {'template': template}))) return tests