dom0_update.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396
  1. #
  2. # The Qubes OS Project, http://www.qubes-os.org
  3. #
  4. # Copyright (C) 2015 Marek Marczykowski-Górecki <marmarek@invisiblethingslab.com>
  5. #
  6. # This library is free software; you can redistribute it and/or
  7. # modify it under the terms of the GNU Lesser General Public
  8. # License as published by the Free Software Foundation; either
  9. # version 2.1 of the License, or (at your option) any later version.
  10. #
  11. # This library is distributed in the hope that it will be useful,
  12. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  14. # Lesser General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU Lesser General Public
  17. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  18. # USA.
  19. #
  20. import asyncio
  21. import os
  22. import shutil
  23. import subprocess
  24. import sys
  25. import tempfile
  26. import unittest
  27. import qubes
  28. import qubes.tests
  29. VM_PREFIX = "test-"
  30. @unittest.skipUnless(os.path.exists('/usr/bin/rpmsign') and
  31. os.path.exists('/usr/bin/rpmbuild'),
  32. 'rpm-sign and/or rpm-build not installed')
  33. class TC_00_Dom0UpgradeMixin(object):
  34. """
  35. Tests for downloading dom0 updates using VMs based on different templates
  36. """
  37. pkg_name = 'qubes-test-pkg'
  38. dom0_update_common_opts = ['--disablerepo=*', '--enablerepo=test']
  39. @classmethod
  40. def generate_key(cls, keydir):
  41. gpg_opts = ['gpg', '--quiet', '--no-default-keyring',
  42. '--homedir', keydir]
  43. p = subprocess.Popen(gpg_opts + ['--gen-key', '--batch'],
  44. stdin=subprocess.PIPE,
  45. stderr=open(os.devnull, 'w'))
  46. p.stdin.write('''
  47. Key-Type: RSA
  48. Key-Length: 1024
  49. Key-Usage: sign
  50. Name-Real: Qubes test
  51. Expire-Date: 0
  52. %commit
  53. '''.format(keydir=keydir).encode())
  54. p.stdin.close()
  55. p.wait()
  56. subprocess.check_call(gpg_opts + ['-a', '--export',
  57. '--output', os.path.join(keydir, 'pubkey.asc')])
  58. p = subprocess.Popen(gpg_opts + ['--with-colons', '--list-keys'],
  59. stdout=subprocess.PIPE)
  60. for line in p.stdout.readlines():
  61. fields = line.decode().split(':')
  62. if fields[0] == 'pub':
  63. return fields[4][-8:].lower()
  64. raise RuntimeError
  65. @classmethod
  66. def setUpClass(cls):
  67. super(TC_00_Dom0UpgradeMixin, cls).setUpClass()
  68. cls.tmpdir = tempfile.mkdtemp()
  69. cls.keyid = cls.generate_key(cls.tmpdir)
  70. with open('/etc/yum.repos.d/test.repo', 'w') as repo_file:
  71. repo_file.write('''
  72. [test]
  73. name = Test
  74. baseurl = http://localhost:8080/
  75. enabled = 1
  76. ''')
  77. @classmethod
  78. def tearDownClass(cls):
  79. os.unlink('/etc/yum.repos.d/test.repo')
  80. shutil.rmtree(cls.tmpdir)
  81. def setUp(self):
  82. super(TC_00_Dom0UpgradeMixin, self).setUp()
  83. if self.template.startswith('whonix-'):
  84. # Whonix redirect all the traffic through tor, so repository
  85. # on http://localhost:8080/ is unavailable
  86. self.skipTest("Test not supported for this template")
  87. self.init_default_template(self.template)
  88. self.updatevm = self.app.add_new_vm(
  89. qubes.vm.appvm.AppVM,
  90. name=self.make_vm_name("updatevm"),
  91. label='red'
  92. )
  93. self.loop.run_until_complete(self.updatevm.create_on_disk())
  94. self.app.updatevm = self.updatevm
  95. self.app.save()
  96. subprocess.call(['rpm', '-e', self.pkg_name],
  97. stderr=subprocess.DEVNULL)
  98. subprocess.check_call(['rpm', '--import',
  99. os.path.join(self.tmpdir, 'pubkey.asc')])
  100. self.loop.run_until_complete(self.updatevm.start())
  101. self.repo_running = False
  102. self.repo_proc = None
  103. def tearDown(self):
  104. if self.repo_proc:
  105. self.repo_proc.terminate()
  106. self.loop.run_until_complete(self.repo_proc.wait())
  107. del self.repo_proc
  108. self.app.updatevm = None
  109. super(TC_00_Dom0UpgradeMixin, self).tearDown()
  110. subprocess.call(['rpm', '-e', self.pkg_name],
  111. stderr=subprocess.DEVNULL)
  112. subprocess.call(['rpm', '-e', 'gpg-pubkey-{}'.format(
  113. self.keyid)], stderr=subprocess.DEVNULL)
  114. for pkg in os.listdir(self.tmpdir):
  115. if pkg.endswith('.rpm'):
  116. os.unlink(pkg)
  117. def create_pkg(self, dir, name, version):
  118. spec_path = os.path.join(dir, name+'.spec')
  119. spec = open(spec_path, 'w')
  120. spec.write(
  121. '''
  122. Name: {name}
  123. Summary: Test Package
  124. Version: {version}
  125. Release: 1
  126. Vendor: Invisible Things Lab
  127. License: GPL
  128. Group: Qubes
  129. URL: http://www.qubes-os.org
  130. %description
  131. Test package
  132. %install
  133. %files
  134. '''.format(name=name, version=version)
  135. )
  136. spec.close()
  137. subprocess.check_call(
  138. ['rpmbuild', '--quiet', '-bb', '--define', '_rpmdir {}'.format(dir),
  139. spec_path])
  140. pkg_path = os.path.join(dir, 'x86_64',
  141. '{}-{}-1.x86_64.rpm'.format(name, version))
  142. subprocess.check_call(['chmod', 'go-rw', '/dev/tty'])
  143. subprocess.check_call(
  144. ['rpm', '--quiet', '--define=_gpg_path {}'.format(dir),
  145. '--define=_gpg_name {}'.format("Qubes test"),
  146. '--addsign', pkg_path],
  147. stdin=subprocess.DEVNULL,
  148. stdout=subprocess.DEVNULL,
  149. stderr=subprocess.STDOUT)
  150. subprocess.check_call(['chmod', 'go+rw', '/dev/tty'])
  151. return pkg_path
  152. def send_pkg(self, filename):
  153. with open(filename, 'rb') as f_pkg:
  154. self.loop.run_until_complete(self.updatevm.run_for_stdio(
  155. 'mkdir -p /tmp/repo; cat > /tmp/repo/{}'.format(
  156. os.path.basename(filename)),
  157. input=f_pkg.read()))
  158. try:
  159. self.loop.run_until_complete(
  160. self.updatevm.run_for_stdio('cd /tmp/repo; createrepo .'))
  161. except subprocess.CalledProcessError as e:
  162. if e.returncode == 127:
  163. self.skipTest('createrepo not installed in template {}'.format(
  164. self.template))
  165. else:
  166. self.skipTest('createrepo failed with code {}, '
  167. 'cannot perform the test'.format(e.returncode))
  168. self.start_repo()
  169. def start_repo(self):
  170. if self.repo_running:
  171. return
  172. self.repo_proc = self.loop.run_until_complete(self.updatevm.run(
  173. 'cd /tmp/repo && python -m SimpleHTTPServer 8080',
  174. stdout=subprocess.DEVNULL,
  175. stderr=subprocess.STDOUT))
  176. self.repo_running = True
  177. def test_000_update(self):
  178. """Dom0 update tests
  179. Check if package update is:
  180. - detected
  181. - installed
  182. - "updates pending" flag is cleared
  183. """
  184. filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
  185. subprocess.check_call(['rpm', '-i', filename])
  186. filename = self.create_pkg(self.tmpdir, self.pkg_name, '2.0')
  187. self.send_pkg(filename)
  188. self.app.domains[0].features['updates-available'] = True
  189. logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
  190. with open(logpath, 'w') as f_log:
  191. proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  192. 'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
  193. stdout=f_log,
  194. stderr=subprocess.STDOUT))
  195. self.loop.run_until_complete(proc.wait())
  196. if proc.returncode:
  197. del proc
  198. with open(logpath) as f_log:
  199. self.fail("qubes-dom0-update failed: " + f_log.read())
  200. del proc
  201. retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
  202. self.pkg_name)], stdout=subprocess.DEVNULL)
  203. self.assertEqual(retcode, 1, 'Package {}-1.0 still installed after '
  204. 'update'.format(self.pkg_name))
  205. retcode = subprocess.call(['rpm', '-q', '{}-2.0'.format(
  206. self.pkg_name)], stdout=subprocess.DEVNULL)
  207. self.assertEqual(retcode, 0, 'Package {}-2.0 not installed after '
  208. 'update'.format(self.pkg_name))
  209. self.assertFalse(
  210. self.app.domains[0].features.get('updates-available', False),
  211. "'updates pending' flag not cleared")
  212. def test_005_update_flag_clear(self):
  213. """Check if 'updates pending' flag is cleared"""
  214. # create any pkg (but not install it) to initialize repo in the VM
  215. filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
  216. self.send_pkg(filename)
  217. self.app.domains[0].features['updates-available'] = True
  218. logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
  219. with open(logpath, 'w') as f_log:
  220. proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  221. 'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
  222. stdout=f_log,
  223. stderr=subprocess.STDOUT))
  224. self.loop.run_until_complete(proc.wait())
  225. if proc.returncode:
  226. del proc
  227. with open(logpath) as f_log:
  228. self.fail("qubes-dom0-update failed: " + f_log.read())
  229. del proc
  230. with open(logpath) as f:
  231. dom0_update_output = f.read()
  232. self.assertFalse('Errno' in dom0_update_output or
  233. 'Couldn\'t' in dom0_update_output,
  234. "qubes-dom0-update reported an error: {}".
  235. format(dom0_update_output))
  236. self.assertFalse(
  237. self.app.domains[0].features.get('updates-available', False),
  238. "'updates pending' flag not cleared")
  239. def test_006_update_flag_clear(self):
  240. """Check if 'updates pending' flag is cleared, using --clean"""
  241. # create any pkg (but not install it) to initialize repo in the VM
  242. filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
  243. self.send_pkg(filename)
  244. self.app.domains[0].features['updates-available'] = True
  245. # remove also repodata to test #1685
  246. if os.path.exists('/var/lib/qubes/updates/repodata'):
  247. shutil.rmtree('/var/lib/qubes/updates/repodata')
  248. logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
  249. with open(logpath, 'w') as f_log:
  250. proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  251. 'qubes-dom0-update', '-y', '--clean',
  252. *self.dom0_update_common_opts,
  253. stdout=f_log,
  254. stderr=subprocess.STDOUT))
  255. self.loop.run_until_complete(proc.wait())
  256. if proc.returncode:
  257. del proc
  258. with open(logpath) as f_log:
  259. self.fail("qubes-dom0-update failed: " + f_log.read())
  260. del proc
  261. with open(logpath) as f:
  262. dom0_update_output = f.read()
  263. self.assertFalse('Errno' in dom0_update_output or
  264. 'Couldn\'t' in dom0_update_output,
  265. "qubes-dom0-update reported an error: {}".
  266. format(dom0_update_output))
  267. self.assertFalse(
  268. self.app.domains[0].features.get('updates-available', False),
  269. "'updates pending' flag not cleared")
  270. def test_010_instal(self):
  271. filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
  272. self.send_pkg(filename)
  273. logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
  274. with open(logpath, 'w') as f_log:
  275. proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  276. 'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
  277. self.pkg_name,
  278. stdout=f_log,
  279. stderr=subprocess.STDOUT))
  280. self.loop.run_until_complete(proc.wait())
  281. if proc.returncode:
  282. del proc
  283. with open(logpath) as f_log:
  284. self.fail("qubes-dom0-update failed: " + f_log.read())
  285. del proc
  286. retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
  287. self.pkg_name)], stdout=open('/dev/null', 'w'))
  288. self.assertEqual(retcode, 0, 'Package {}-1.0 not installed'.format(
  289. self.pkg_name))
  290. def test_020_install_wrong_sign(self):
  291. subprocess.call(['rpm', '-e', 'gpg-pubkey-{}'.format(
  292. self.keyid)])
  293. filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
  294. self.send_pkg(filename)
  295. logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
  296. with open(logpath, 'w') as f_log:
  297. proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  298. 'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
  299. self.pkg_name,
  300. stdout=f_log,
  301. stderr=subprocess.STDOUT))
  302. self.loop.run_until_complete(proc.wait())
  303. if not proc.returncode:
  304. del proc
  305. with open(logpath) as f_log:
  306. self.fail("qubes-dom0-update unexpectedly succeeded: " +
  307. f_log.read())
  308. del proc
  309. retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
  310. self.pkg_name)], stdout=subprocess.DEVNULL)
  311. self.assertEqual(retcode, 1,
  312. 'Package {}-1.0 installed although '
  313. 'signature is invalid'.format(self.pkg_name))
  314. def test_030_install_unsigned(self):
  315. filename = self.create_pkg(self.tmpdir, self.pkg_name, '1.0')
  316. subprocess.check_call(['rpm', '--delsign', filename],
  317. stdout=subprocess.DEVNULL,
  318. stderr=subprocess.STDOUT)
  319. self.send_pkg(filename)
  320. logpath = os.path.join(self.tmpdir, 'dom0-update-output.txt')
  321. with open(logpath, 'w') as f_log:
  322. proc = self.loop.run_until_complete(asyncio.create_subprocess_exec(
  323. 'qubes-dom0-update', '-y', *self.dom0_update_common_opts,
  324. self.pkg_name,
  325. stdout=f_log,
  326. stderr=subprocess.STDOUT))
  327. self.loop.run_until_complete(proc.wait())
  328. if not proc.returncode:
  329. del proc
  330. with open(logpath) as f_log:
  331. self.fail("qubes-dom0-update unexpectedly succeeded: " +
  332. f_log.read())
  333. del proc
  334. retcode = subprocess.call(['rpm', '-q', '{}-1.0'.format(
  335. self.pkg_name)], stdout=subprocess.DEVNULL)
  336. self.assertEqual(retcode, 1,
  337. 'UNSIGNED package {}-1.0 installed'.format(self.pkg_name))
  338. def create_testcases_for_templates():
  339. return qubes.tests.create_testcases_for_templates('TC_00_Dom0Upgrade',
  340. TC_00_Dom0UpgradeMixin, qubes.tests.SystemTestCase,
  341. module=sys.modules[__name__])
  342. def load_tests(loader, tests, pattern):
  343. tests.addTests(loader.loadTestsFromNames(
  344. create_testcases_for_templates()))
  345. return tests
  346. qubes.tests.maybe_create_testcases_on_import(create_testcases_for_templates)