8dda66d693
gpg2 requires explicit `%no-protection`, otherwise tries to interactively (through gpg-agent) prompt for the passphrase.
398 lines
15 KiB
Python
398 lines
15 KiB
Python
#
|
|
# The Qubes OS Project, http://www.qubes-os.org
|
|
#
|
|
# Copyright (C) 2015 Marek Marczykowski-Górecki <marmarek@invisiblethingslab.com>
|
|
#
|
|
# This library is free software; you can redistribute it and/or
|
|
# modify it under the terms of the GNU Lesser General Public
|
|
# License as published by the Free Software Foundation; either
|
|
# version 2.1 of the License, or (at your option) any later version.
|
|
#
|
|
# This library 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
|
|
# Lesser General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Lesser General Public
|
|
# License along with this library; if not, see <https://www.gnu.org/licenses/>.
|
|
# USA.
|
|
#
|
|
|
|
import asyncio
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
import qubes
|
|
import qubes.tests
|
|
|
|
VM_PREFIX = "test-"
|
|
|
|
@unittest.skipUnless(os.path.exists('/usr/bin/rpmsign') and
|
|
os.path.exists('/usr/bin/rpmbuild'),
|
|
'rpm-sign and/or rpm-build not installed')
|
|
class TC_00_Dom0UpgradeMixin(object):
|
|
"""
|
|
Tests for downloading dom0 updates using VMs based on different templates
|
|
"""
|
|
pkg_name = 'qubes-test-pkg'
|
|
dom0_update_common_opts = ['--disablerepo=*', '--enablerepo=test']
|
|
|
|
@classmethod
|
|
def generate_key(cls, keydir):
|
|
gpg_opts = ['gpg', '--quiet', '--no-default-keyring',
|
|
'--homedir', keydir]
|
|
p = subprocess.Popen(gpg_opts + ['--gen-key', '--batch'],
|
|
stdin=subprocess.PIPE,
|
|
stderr=open(os.devnull, 'w'))
|
|
p.stdin.write('''
|
|
Key-Type: RSA
|
|
Key-Length: 1024
|
|
Key-Usage: sign
|
|
Name-Real: Qubes test
|
|
Expire-Date: 0
|
|
%no-protection
|
|
%commit
|
|
'''.format(keydir=keydir).encode())
|
|
p.stdin.close()
|
|
p.wait()
|
|
|
|
subprocess.check_call(gpg_opts + ['-a', '--export',
|
|
'--output', os.path.join(keydir, 'pubkey.asc')])
|
|
p = subprocess.Popen(gpg_opts + ['--with-colons', '--list-keys'],
|
|
stdout=subprocess.PIPE)
|
|
for line in p.stdout.readlines():
|
|
fields = line.decode().split(':')
|
|
if fields[0] == 'pub':
|
|
return fields[4][-8:].lower()
|
|
raise RuntimeError
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(TC_00_Dom0UpgradeMixin, cls).setUpClass()
|
|
|
|
cls.tmpdir = tempfile.mkdtemp()
|
|
|
|
cls.keyid = cls.generate_key(cls.tmpdir)
|
|
|
|
with open('/etc/yum.repos.d/test.repo', 'w') as repo_file:
|
|
repo_file.write('''
|
|
[test]
|
|
name = Test
|
|
baseurl = http://localhost:8080/
|
|
enabled = 1
|
|
''')
|
|
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
os.unlink('/etc/yum.repos.d/test.repo')
|
|
|
|
shutil.rmtree(cls.tmpdir)
|
|
|
|
def setUp(self):
|
|
super(TC_00_Dom0UpgradeMixin, self).setUp()
|
|
if self.template.startswith('whonix-'):
|
|
# Whonix redirect all the traffic through tor, so repository
|
|
# on http://localhost:8080/ is unavailable
|
|
self.skipTest("Test not supported for this template")
|
|
self.init_default_template(self.template)
|
|
self.updatevm = self.app.add_new_vm(
|
|
qubes.vm.appvm.AppVM,
|
|
name=self.make_vm_name("updatevm"),
|
|
label='red'
|
|
)
|
|
self.loop.run_until_complete(self.updatevm.create_on_disk())
|
|
self.app.updatevm = self.updatevm
|
|
self.app.save()
|
|
subprocess.call(['rpm', '-e', self.pkg_name],
|
|
stderr=subprocess.DEVNULL)
|
|
subprocess.check_call(['rpm', '--import',
|
|
os.path.join(self.tmpdir, 'pubkey.asc')])
|
|
self.loop.run_until_complete(self.updatevm.start())
|
|
self.repo_running = False
|
|
self.repo_proc = None
|
|
|
|
def tearDown(self):
|
|
if self.repo_proc:
|
|
self.repo_proc.terminate()
|
|
self.loop.run_until_complete(self.repo_proc.wait())
|
|
del self.repo_proc
|
|
self.app.updatevm = None
|
|
super(TC_00_Dom0UpgradeMixin, self).tearDown()
|
|
|
|
subprocess.call(['rpm', '-e', self.pkg_name],
|
|
stderr=subprocess.DEVNULL)
|
|
subprocess.call(['rpm', '-e', 'gpg-pubkey-{}'.format(
|
|
self.keyid)], stderr=subprocess.DEVNULL)
|
|
|
|
for pkg in os.listdir(self.tmpdir):
|
|
if pkg.endswith('.rpm'):
|
|
os.unlink(pkg)
|
|
|
|
def create_pkg(self, dir, name, version):
|
|
spec_path = os.path.join(dir, name+'.spec')
|
|
spec = open(spec_path, 'w')
|
|
spec.write(
|
|
'''
|
|
Name: {name}
|
|
Summary: Test Package
|
|
Version: {version}
|
|
Release: 1
|
|
Vendor: Invisible Things Lab
|
|
License: GPL
|
|
Group: Qubes
|
|
URL: http://www.qubes-os.org
|
|
|
|
%description
|
|
Test package
|
|
|
|
%install
|
|
|
|
%files
|
|
'''.format(name=name, version=version)
|
|
)
|
|
spec.close()
|
|
subprocess.check_call(
|
|
['rpmbuild', '--quiet', '-bb', '--define', '_rpmdir {}'.format(dir),
|
|
spec_path])
|
|
pkg_path = os.path.join(dir, 'x86_64',
|
|
'{}-{}-1.x86_64.rpm'.format(name, version))
|
|
subprocess.check_call(['chmod', 'go-rw', '/dev/tty'])
|
|
subprocess.check_call(
|
|
['rpm', '--quiet', '--define=_gpg_path {}'.format(dir),
|
|
'--define=_gpg_name {}'.format("Qubes test"),
|
|
'--addsign', pkg_path],
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.STDOUT)
|
|
subprocess.check_call(['chmod', 'go+rw', '/dev/tty'])
|
|
return pkg_path
|
|
|
|
def send_pkg(self, filename):
|
|
with open(filename, 'rb') as f_pkg:
|
|
self.loop.run_until_complete(self.updatevm.run_for_stdio(
|
|
'mkdir -p /tmp/repo; cat > /tmp/repo/{}'.format(
|
|
os.path.basename(filename)),
|
|
input=f_pkg.read()))
|
|
try:
|
|
self.loop.run_until_complete(
|
|
self.updatevm.run_for_stdio('cd /tmp/repo; createrepo .'))
|
|
except subprocess.CalledProcessError as e:
|
|
if e.returncode == 127:
|
|
self.skipTest('createrepo not installed in template {}'.format(
|
|
self.template))
|
|
else:
|
|
self.skipTest('createrepo failed with code {}, '
|
|
'cannot perform the test'.format(e.returncode))
|
|
self.start_repo()
|
|
|
|
def start_repo(self):
|
|
if self.repo_running:
|
|
return
|
|
self.repo_proc = self.loop.run_until_complete(self.updatevm.run(
|
|
'cd /tmp/repo && python -m SimpleHTTPServer 8080',
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.STDOUT))
|
|
self.repo_running = True
|
|
|
|
def test_000_update(self):
|
|
"""Dom0 update tests
|
|
|
|
Check if package update is:
|
|
- detected
|
|
- installed
|
|
- "updates pending" flag is cleared
|
|
"""
|
|
filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
|
|
subprocess.check_call(['rpm', '-i', filename])
|
|
filename = self.create_pkg(self.tmpdir, self.pkg_name, '2.0')
|
|
self.send_pkg(filename)
|
|
self.app.domains[0].features['updates-available'] = True
|
|
|
|
logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
|
|
with open(logpath, 'w') as f_log:
|
|
proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
|
|
'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
|
|
stdout=f_log,
|
|
stderr=subprocess.STDOUT))
|
|
self.loop.run_until_complete(proc.wait())
|
|
if proc.returncode:
|
|
del proc
|
|
with open(logpath) as f_log:
|
|
self.fail("qubes-dom0-update failed: " + f_log.read())
|
|
del proc
|
|
|
|
retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
|
|
self.pkg_name)], stdout=subprocess.DEVNULL)
|
|
self.assertEqual(retcode, 1, 'Package {}-1.0 still installed after '
|
|
'update'.format(self.pkg_name))
|
|
retcode = subprocess.call(['rpm', '-q', '{}-2.0'.format(
|
|
self.pkg_name)], stdout=subprocess.DEVNULL)
|
|
self.assertEqual(retcode, 0, 'Package {}-2.0 not installed after '
|
|
'update'.format(self.pkg_name))
|
|
self.assertFalse(
|
|
self.app.domains[0].features.get('updates-available', False),
|
|
"'updates pending' flag not cleared")
|
|
|
|
def test_005_update_flag_clear(self):
|
|
"""Check if 'updates pending' flag is cleared"""
|
|
|
|
# create any pkg (but not install it) to initialize repo in the VM
|
|
filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
|
|
self.send_pkg(filename)
|
|
self.app.domains[0].features['updates-available'] = True
|
|
|
|
logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
|
|
with open(logpath, 'w') as f_log:
|
|
proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
|
|
'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
|
|
stdout=f_log,
|
|
stderr=subprocess.STDOUT))
|
|
self.loop.run_until_complete(proc.wait())
|
|
if proc.returncode:
|
|
del proc
|
|
with open(logpath) as f_log:
|
|
self.fail("qubes-dom0-update failed: " + f_log.read())
|
|
del proc
|
|
|
|
with open(logpath) as f:
|
|
dom0_update_output = f.read()
|
|
self.assertFalse('Errno' in dom0_update_output or
|
|
'Couldn\'t' in dom0_update_output,
|
|
"qubes-dom0-update reported an error: {}".
|
|
format(dom0_update_output))
|
|
|
|
self.assertFalse(
|
|
self.app.domains[0].features.get('updates-available', False),
|
|
"'updates pending' flag not cleared")
|
|
|
|
def test_006_update_flag_clear(self):
|
|
"""Check if 'updates pending' flag is cleared, using --clean"""
|
|
|
|
# create any pkg (but not install it) to initialize repo in the VM
|
|
filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
|
|
self.send_pkg(filename)
|
|
self.app.domains[0].features['updates-available'] = True
|
|
|
|
# remove also repodata to test #1685
|
|
if os.path.exists('/var/lib/qubes/updates/repodata'):
|
|
shutil.rmtree('/var/lib/qubes/updates/repodata')
|
|
logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
|
|
with open(logpath, 'w') as f_log:
|
|
proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
|
|
'qubes-dom0-update', '-y', '--clean',
|
|
*self.dom0_update_common_opts,
|
|
stdout=f_log,
|
|
stderr=subprocess.STDOUT))
|
|
self.loop.run_until_complete(proc.wait())
|
|
if proc.returncode:
|
|
del proc
|
|
with open(logpath) as f_log:
|
|
self.fail("qubes-dom0-update failed: " + f_log.read())
|
|
del proc
|
|
|
|
with open(logpath) as f:
|
|
dom0_update_output = f.read()
|
|
self.assertFalse('Errno' in dom0_update_output or
|
|
'Couldn\'t' in dom0_update_output,
|
|
"qubes-dom0-update reported an error: {}".
|
|
format(dom0_update_output))
|
|
|
|
self.assertFalse(
|
|
self.app.domains[0].features.get('updates-available', False),
|
|
"'updates pending' flag not cleared")
|
|
|
|
def test_010_instal(self):
|
|
filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
|
|
self.send_pkg(filename)
|
|
|
|
logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
|
|
with open(logpath, 'w') as f_log:
|
|
proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
|
|
'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
|
|
self.pkg_name,
|
|
stdout=f_log,
|
|
stderr=subprocess.STDOUT))
|
|
self.loop.run_until_complete(proc.wait())
|
|
if proc.returncode:
|
|
del proc
|
|
with open(logpath) as f_log:
|
|
self.fail("qubes-dom0-update failed: " + f_log.read())
|
|
del proc
|
|
|
|
retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
|
|
self.pkg_name)], stdout=open('/dev/null', 'w'))
|
|
self.assertEqual(retcode, 0, 'Package {}-1.0 not installed'.format(
|
|
self.pkg_name))
|
|
|
|
def test_020_install_wrong_sign(self):
|
|
subprocess.call(['rpm', '-e', 'gpg-pubkey-{}'.format(
|
|
self.keyid)])
|
|
filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
|
|
self.send_pkg(filename)
|
|
|
|
logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
|
|
with open(logpath, 'w') as f_log:
|
|
proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
|
|
'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
|
|
self.pkg_name,
|
|
stdout=f_log,
|
|
stderr=subprocess.STDOUT))
|
|
self.loop.run_until_complete(proc.wait())
|
|
if not proc.returncode:
|
|
del proc
|
|
with open(logpath) as f_log:
|
|
self.fail("qubes-dom0-update unexpectedly succeeded: " +
|
|
f_log.read())
|
|
del proc
|
|
|
|
retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
|
|
self.pkg_name)], stdout=subprocess.DEVNULL)
|
|
self.assertEqual(retcode, 1,
|
|
'Package {}-1.0 installed although '
|
|
'signature is invalid'.format(self.pkg_name))
|
|
|
|
def test_030_install_unsigned(self):
|
|
filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
|
|
subprocess.check_call(['rpm', '--delsign', filename],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.STDOUT)
|
|
self.send_pkg(filename)
|
|
|
|
logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
|
|
with open(logpath, 'w') as f_log:
|
|
proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
|
|
'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
|
|
self.pkg_name,
|
|
stdout=f_log,
|
|
stderr=subprocess.STDOUT))
|
|
self.loop.run_until_complete(proc.wait())
|
|
if not proc.returncode:
|
|
del proc
|
|
with open(logpath) as f_log:
|
|
self.fail("qubes-dom0-update unexpectedly succeeded: " +
|
|
f_log.read())
|
|
del proc
|
|
|
|
retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
|
|
self.pkg_name)], stdout=subprocess.DEVNULL)
|
|
self.assertEqual(retcode, 1,
|
|
'UNSIGNED package {}-1.0 installed'.format(self.pkg_name))
|
|
|
|
|
|
def create_testcases_for_templates():
|
|
return qubes.tests.create_testcases_for_templates('TC_00_Dom0Upgrade',
|
|
TC_00_Dom0UpgradeMixin, qubes.tests.SystemTestCase,
|
|
module=sys.modules[__name__])
|
|
|
|
def load_tests(loader, tests, pattern):
|
|
tests.addTests(loader.loadTestsFromNames(
|
|
create_testcases_for_templates()))
|
|
return tests
|
|
|
|
qubes.tests.maybe_create_testcases_on_import(create_testcases_for_templates)
|