Qubes/core-admin
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

tests: separate test for qrexec "loopback" call

Browse Source
This commit is contained in:
Marek Marczykowski-Górecki 2014-10-29 21:38:54 +01:00
parent 91425d2e0b
commit c3b0b4f75f
1 changed files with 49 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
tests/vm_qrexec_gui.py
View File

@ -32,10 +32,14 @@ class VmRunningTests(unittest.TestCase):
self.qc = QubesVmCollection() self.qc = QubesVmCollection()
self.qc.lock_db_for_writing() self.qc.lock_db_for_writing()
self.qc.load() self.qc.load()
self.testvm = self.qc.add_new_vm("QubesAppVm", self.testvm1 = self.qc.add_new_vm("QubesAppVm",
name="%stestvm" % VM_PREFIX, name="%svm1" % VM_PREFIX,
template=self.qc.get_default_template()) template=self.qc.get_default_template())
self.testvm.create_on_disk(verbose=False) self.testvm1.create_on_disk(verbose=False)
self.testvm2 = self.qc.add_new_vm("QubesAppVm",
name="%svm2" % VM_PREFIX,
template=self.qc.get_default_template())
self.testvm2.create_on_disk(verbose=False)
self.qc.save() self.qc.save()
self.qc.unlock_db() self.qc.unlock_db()
@ -67,37 +71,39 @@ class VmRunningTests(unittest.TestCase):
self.remove_vms(vmlist) self.remove_vms(vmlist)
def test_000_start_shutdown(self): def test_000_start_shutdown(self):
self.testvm.start() self.testvm1.start()
self.assertEquals(self.testvm.get_power_state(), "Running") self.assertEquals(self.testvm1.get_power_state(), "Running")
self.testvm.shutdown() self.testvm1.shutdown()
shutdown_counter = 0 shutdown_counter = 0
while self.testvm.is_running(): while self.testvm1.is_running():
if shutdown_counter > defaults["shutdown_counter_max"]: if shutdown_counter > defaults["shutdown_counter_max"]:
self.fail("VM hanged during shutdown") self.fail("VM hanged during shutdown")
shutdown_counter += 1 shutdown_counter += 1
time.sleep(1) time.sleep(1)
time.sleep(1) time.sleep(1)
self.assertEquals(self.testvm.get_power_state(), "Halted") self.assertEquals(self.testvm1.get_power_state(), "Halted")
def test_010_run_gui_app(self): def test_010_run_gui_app(self):
self.testvm.start() self.testvm1.start()
self.assertEquals(self.testvm.get_power_state(), "Running") self.assertEquals(self.testvm1.get_power_state(), "Running")
self.testvm.run("gnome-terminal") self.testvm1.run("gnome-terminal")
wait_count = 0 wait_count = 0
while subprocess.call(['xdotool', 'search', '--name', 'user@%s' % while subprocess.call(['xdotool', 'search', '--name', 'user@%s' %
self.testvm.name]) > 0: self.testvm1.name], stdout=open(os.path.devnull, 'w'),
stderr=subprocess.STDOUT) > 0:
wait_count += 1 wait_count += 1
if wait_count > 100: if wait_count > 100:
self.fail("Timeout while waiting for gnome-terminal window") self.fail("Timeout while waiting for gnome-terminal window")
time.sleep(0.1) time.sleep(0.1)
time.sleep(0.5)
subprocess.check_call(['xdotool', 'search', '--name', 'user@%s' % subprocess.check_call(['xdotool', 'search', '--name', 'user@%s' %
self.testvm.name, 'windowactivate', 'type', 'exit\n']) self.testvm1.name, 'windowactivate', 'type', 'exit\n'])
wait_count = 0 wait_count = 0
while subprocess.call(['xdotool', 'search', '--name', 'user@%s' % while subprocess.call(['xdotool', 'search', '--name', 'user@%s' %
self.testvm.name], stdout=open(os.path.devnull, 'w'), self.testvm1.name], stdout=open(os.path.devnull, 'w'),
stderr=subprocess.STDOUT) == 0: stderr=subprocess.STDOUT) == 0:
wait_count += 1 wait_count += 1
if wait_count > 100: if wait_count > 100:
@ -106,9 +112,10 @@ class VmRunningTests(unittest.TestCase):
time.sleep(0.1) time.sleep(0.1)
def test_100_qrexec_filecopy(self): def test_100_qrexec_filecopy(self):
self.testvm.start() self.testvm1.start()
p = self.testvm.run("qvm-copy-to-vm %s /etc/passwd" % self.testvm2.start()
self.testvm.name, passio_popen=True, p = self.testvm1.run("qvm-copy-to-vm %s /etc/passwd" %
self.testvm2.name, passio_popen=True,
passio_stderr=True) passio_stderr=True)
# Confirm transfer # Confirm transfer
subprocess.check_call(['xdotool', 'search', '--sync', '--name', 'Question', subprocess.check_call(['xdotool', 'search', '--sync', '--name', 'Question',
@ -116,22 +123,39 @@ class VmRunningTests(unittest.TestCase):
p.wait() p.wait()
self.assertEqual(p.returncode, 0, "qvm-copy-to-vm failed: %s" % self.assertEqual(p.returncode, 0, "qvm-copy-to-vm failed: %s" %
p.stderr.read()) p.stderr.read())
retcode = self.testvm.run("diff /etc/passwd " retcode = self.testvm2.run("diff /etc/passwd "
"/home/user/QubesIncoming/%s/passwd" % self.testvm.name, wait=True) "/home/user/QubesIncoming/%s/passwd" % self.testvm1.name, wait=True)
self.assertEqual(retcode, 0, "file differs") self.assertEqual(retcode, 0, "file differs")
def test_110_qrexec_filecopy_deny(self): def test_110_qrexec_filecopy_deny(self):
self.testvm.start() self.testvm1.start()
p = self.testvm.run("qvm-copy-to-vm %s /etc/passwd" % self.testvm2.start()
self.testvm.name, passio_popen=True) p = self.testvm1.run("qvm-copy-to-vm %s /etc/passwd" %
self.testvm2.name, passio_popen=True)
# Deny transfer # Deny transfer
subprocess.check_call(['xdotool', 'search', '--sync', '--name', 'Question', subprocess.check_call(['xdotool', 'search', '--sync', '--name', 'Question',
'key', 'n']) 'key', 'n'])
p.wait() p.wait()
self.assertEqual(p.returncode, 1, "qvm-copy-to-vm unexpectedly " self.assertEqual(p.returncode, 1, "qvm-copy-to-vm unexpectedly "
"succeeded") "succeeded")
retcode = self.testvm.run("ls /home/user/QubesIncoming/%s" % retcode = self.testvm1.run("ls /home/user/QubesIncoming/%s" %
self.testvm.name, wait=True, self.testvm1.name, wait=True,
ignore_stderr=True) ignore_stderr=True)
self.assertEqual(retcode, 2, "QubesIncoming exists although file copy was " self.assertEqual(retcode, 2, "QubesIncoming exists although file copy was "
"denied") "denied")
def test_120_qrexec_filecopy_self(self):
self.testvm1.start()
p = self.testvm1.run("qvm-copy-to-vm %s /etc/passwd" %
self.testvm1.name, passio_popen=True,
passio_stderr=True)
# Confirm transfer
subprocess.check_call(['xdotool', 'search', '--sync', '--name', 'Question',
'key', 'y'])
p.wait()
self.assertEqual(p.returncode, 0, "qvm-copy-to-vm failed: %s" %
p.stderr.read())
retcode = self.testvm1.run("diff /etc/passwd "
"/home/user/QubesIncoming/%s/passwd" % self.testvm1.name, wait=True)
self.assertEqual(retcode, 0, "file differs")