app.py 39 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905
  1. # -*- encoding: utf8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2017 Marek Marczykowski-Górecki
  6. # <marmarek@invisiblethingslab.com>
  7. #
  8. # This program is free software; you can redistribute it and/or modify
  9. # it under the terms of the GNU Lesser General Public License as published by
  10. # the Free Software Foundation; either version 2 of the License, or
  11. # (at your option) any later version.
  12. #
  13. # This program is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  16. # GNU Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public License along
  19. # with this program; if not, see <http://www.gnu.org/licenses/>.
  20. import os
  21. import shutil
  22. import socket
  23. import subprocess
  24. import unittest
  25. import multiprocessing
  26. try:
  27. import unittest.mock as mock
  28. except ImportError:
  29. import mock
  30. import tempfile
  31. import qubesadmin.exc
  32. import qubesadmin.tests
  33. class TC_00_VMCollection(qubesadmin.tests.QubesTestCase):
  34. def test_000_list(self):
  35. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  36. b'0\x00test-vm class=AppVM state=Running\n'
  37. self.assertEqual(
  38. list(self.app.domains.keys()),
  39. ['test-vm'])
  40. self.assertAllCalled()
  41. def test_001_getitem(self):
  42. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  43. b'0\x00test-vm class=AppVM state=Running\n'
  44. try:
  45. vm = self.app.domains['test-vm']
  46. self.assertEqual(vm.name, 'test-vm')
  47. except KeyError:
  48. self.fail('VM not found in collection')
  49. self.assertAllCalled()
  50. with self.assertRaises(KeyError):
  51. vm = self.app.domains['test-non-existent']
  52. self.assertAllCalled()
  53. def test_002_in(self):
  54. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  55. b'0\x00test-vm class=AppVM state=Running\n'
  56. self.assertIn('test-vm', self.app.domains)
  57. self.assertAllCalled()
  58. self.assertNotIn('test-non-existent', self.app.domains)
  59. self.assertAllCalled()
  60. def test_003_iter(self):
  61. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  62. b'0\x00test-vm class=AppVM state=Running\n'
  63. self.assertEqual([vm.name for vm in self.app.domains], ['test-vm'])
  64. self.assertAllCalled()
  65. def test_004_delitem(self):
  66. self.app.expected_calls[('test-vm', 'admin.vm.Remove', None, None)] = \
  67. b'0\x00'
  68. del self.app.domains['test-vm']
  69. self.assertAllCalled()
  70. def test_005_keys(self):
  71. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  72. b'0\x00test-vm class=AppVM state=Running\n' \
  73. b'test-vm2 class=AppVM state=Running\n'
  74. self.assertEqual(set(self.app.domains.keys()),
  75. set(['test-vm', 'test-vm2']))
  76. self.assertAllCalled()
  77. def test_006_values(self):
  78. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  79. b'0\x00test-vm class=AppVM state=Running\n' \
  80. b'test-vm2 class=AppVM state=Running\n'
  81. values = self.app.domains.values()
  82. for obj in values:
  83. self.assertIsInstance(obj, qubesadmin.vm.QubesVM)
  84. self.assertEqual(set([vm.name for vm in values]),
  85. set(['test-vm', 'test-vm2']))
  86. self.assertAllCalled()
  87. def test_007_getitem_blind_mode(self):
  88. self.app.blind_mode = True
  89. try:
  90. vm = self.app.domains['test-vm']
  91. self.assertEqual(vm.name, 'test-vm')
  92. except KeyError:
  93. self.fail('VM not found in collection')
  94. self.assertAllCalled()
  95. with self.assertNotRaises(KeyError):
  96. vm = self.app.domains['test-non-existent']
  97. self.assertAllCalled()
  98. def test_008_in_blind_mode(self):
  99. self.app.blind_mode = True
  100. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  101. b'0\x00test-vm class=AppVM state=Running\n'
  102. self.assertIn('test-vm', self.app.domains)
  103. self.assertAllCalled()
  104. self.assertNotIn('test-non-existent', self.app.domains)
  105. self.assertAllCalled()
  106. def test_009_getitem_cache_class(self):
  107. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  108. b'0\x00test-vm class=AppVM state=Running\n'
  109. try:
  110. vm = self.app.domains['test-vm']
  111. self.assertEqual(vm.name, 'test-vm')
  112. self.assertEqual(vm.klass, 'AppVM')
  113. except KeyError:
  114. self.fail('VM not found in collection')
  115. self.assertAllCalled()
  116. class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
  117. def setUp(self):
  118. super(TC_10_QubesBase, self).setUp()
  119. self.check_output_patch = mock.patch(
  120. 'subprocess.check_output')
  121. self.check_output_mock = self.check_output_patch.start()
  122. def tearDown(self):
  123. self.check_output_patch.stop()
  124. super(TC_10_QubesBase, self).tearDown()
  125. def test_010_new_simple(self):
  126. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM', None,
  127. b'name=new-vm label=red')] = b'0\x00'
  128. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  129. b'0\x00new-vm class=AppVM state=Running\n'
  130. vm = self.app.add_new_vm('AppVM', 'new-vm', 'red')
  131. self.assertEqual(vm.name, 'new-vm')
  132. self.assertEqual(vm.klass, 'AppVM')
  133. self.assertAllCalled()
  134. def test_011_new_template(self):
  135. self.app.expected_calls[('dom0', 'admin.vm.Create.TemplateVM', None,
  136. b'name=new-template label=red')] = b'0\x00'
  137. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  138. b'0\x00new-template class=TemplateVM state=Running\n'
  139. vm = self.app.add_new_vm('TemplateVM', 'new-template', 'red')
  140. self.assertEqual(vm.name, 'new-template')
  141. self.assertEqual(vm.klass, 'TemplateVM')
  142. self.assertAllCalled()
  143. def test_012_new_template_based(self):
  144. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  145. 'some-template', b'name=new-vm label=red')] = b'0\x00'
  146. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  147. b'0\x00new-vm class=AppVM state=Running\n'
  148. vm = self.app.add_new_vm('AppVM', 'new-vm', 'red', 'some-template')
  149. self.assertEqual(vm.name, 'new-vm')
  150. self.assertEqual(vm.klass, 'AppVM')
  151. self.assertAllCalled()
  152. def test_013_new_objects_params(self):
  153. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  154. 'some-template', b'name=new-vm label=red')] = b'0\x00'
  155. self.app.expected_calls[('dom0', 'admin.label.List', None, None)] = \
  156. b'0\x00red\nblue\n'
  157. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  158. b'0\x00new-vm class=AppVM state=Running\n' \
  159. b'some-template class=TemplateVM state=Running\n'
  160. vm = self.app.add_new_vm(self.app.get_vm_class('AppVM'), 'new-vm',
  161. self.app.get_label('red'), self.app.domains['some-template'])
  162. self.assertEqual(vm.name, 'new-vm')
  163. self.assertEqual(vm.klass, 'AppVM')
  164. self.assertAllCalled()
  165. def test_014_new_pool(self):
  166. self.app.expected_calls[('dom0', 'admin.vm.CreateInPool.AppVM', None,
  167. b'name=new-vm label=red pool=some-pool')] = b'0\x00'
  168. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  169. b'0\x00new-vm class=AppVM state=Running\n'
  170. vm = self.app.add_new_vm('AppVM', 'new-vm', 'red', pool='some-pool')
  171. self.assertEqual(vm.name, 'new-vm')
  172. self.assertEqual(vm.klass, 'AppVM')
  173. self.assertAllCalled()
  174. def test_015_new_pools(self):
  175. self.app.expected_calls[('dom0', 'admin.vm.CreateInPool.AppVM', None,
  176. b'name=new-vm label=red pool:private=some-pool '
  177. b'pool:volatile=other-pool')] = b'0\x00'
  178. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  179. b'0\x00new-vm class=AppVM state=Running\n'
  180. vm = self.app.add_new_vm('AppVM', 'new-vm', 'red',
  181. pools={'private': 'some-pool', 'volatile': 'other-pool'})
  182. self.assertEqual(vm.name, 'new-vm')
  183. self.assertEqual(vm.klass, 'AppVM')
  184. self.assertAllCalled()
  185. def test_016_new_template_based_default(self):
  186. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  187. None, b'name=new-vm label=red')] = b'0\x00'
  188. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  189. b'0\x00new-vm class=AppVM state=Running\n'
  190. vm = self.app.add_new_vm('AppVM', 'new-vm', 'red',
  191. template=qubesadmin.DEFAULT)
  192. self.assertEqual(vm.name, 'new-vm')
  193. self.assertEqual(vm.klass, 'AppVM')
  194. self.assertAllCalled()
  195. def test_020_get_label(self):
  196. self.app.expected_calls[('dom0', 'admin.label.List', None, None)] = \
  197. b'0\x00red\nblue\n'
  198. label = self.app.get_label('red')
  199. self.assertEqual(label.name, 'red')
  200. self.assertAllCalled()
  201. def clone_setup_common_calls(self, src, dst):
  202. # have each property type with default=no, each special-cased,
  203. # and some with default=yes
  204. properties = {
  205. 'label': 'default=False type=label red',
  206. 'template': 'default=False type=vm test-template',
  207. 'memory': 'default=False type=int 400',
  208. 'kernel': 'default=False type=str 4.9.31',
  209. 'netvm': 'default=False type=vm test-net',
  210. 'virt_mode': 'default=False type=str hvm',
  211. 'default_user': 'default=True type=str user',
  212. }
  213. self.app.expected_calls[
  214. (src, 'admin.vm.property.List', None, None)] = \
  215. b'0\0qid\nname\n' + \
  216. b'\n'.join(prop.encode() for prop in properties.keys()) + \
  217. b'\n'
  218. for prop, value in properties.items():
  219. self.app.expected_calls[
  220. (src, 'admin.vm.property.Get', prop, None)] = \
  221. b'0\0' + value.encode()
  222. # special cases handled by admin.vm.Create call
  223. if prop in ('label', 'template'):
  224. continue
  225. # default properties should not be set
  226. if 'default=True' in value:
  227. continue
  228. self.app.expected_calls[
  229. (dst, 'admin.vm.property.Set', prop,
  230. value.split()[-1].encode())] = b'0\0'
  231. # tags
  232. self.app.expected_calls[
  233. (src, 'admin.vm.tag.List', None, None)] = \
  234. b'0\0tag1\ntag2\n'
  235. self.app.expected_calls[
  236. (dst, 'admin.vm.tag.Set', 'tag1', None)] = b'0\0'
  237. self.app.expected_calls[
  238. (dst, 'admin.vm.tag.Set', 'tag2', None)] = b'0\0'
  239. # features
  240. self.app.expected_calls[
  241. (src, 'admin.vm.feature.List', None, None)] = \
  242. b'0\0feat1\nfeat2\n'
  243. self.app.expected_calls[
  244. (src, 'admin.vm.feature.Get', 'feat1', None)] = \
  245. b'0\0feat1-value with spaces'
  246. self.app.expected_calls[
  247. (src, 'admin.vm.feature.Get', 'feat2', None)] = \
  248. b'0\x001'
  249. self.app.expected_calls[
  250. (dst, 'admin.vm.feature.Set', 'feat1',
  251. b'feat1-value with spaces')] = b'0\0'
  252. self.app.expected_calls[
  253. (dst, 'admin.vm.feature.Set', 'feat2', b'1')] = b'0\0'
  254. # firewall
  255. rules = (
  256. b'action=drop dst4=192.168.0.0/24\n'
  257. b'action=accept\n'
  258. )
  259. self.app.expected_calls[
  260. (src, 'admin.vm.firewall.Get', None, None)] = \
  261. b'0\x00' + rules
  262. self.app.expected_calls[
  263. (dst, 'admin.vm.firewall.Set', None, rules)] = \
  264. b'0\x00'
  265. # storage
  266. for vm in (src, dst):
  267. self.app.expected_calls[
  268. (vm, 'admin.vm.volume.Info', 'root', None)] = \
  269. b'0\x00pool=lvm\n' \
  270. b'vid=vm-' + vm.encode() + b'/root\n' \
  271. b'size=10737418240\n' \
  272. b'usage=2147483648\n' \
  273. b'rw=False\n' \
  274. b'internal=True\n' \
  275. b'source=vm-test-template/root\n' \
  276. b'save_on_stop=False\n' \
  277. b'snap_on_start=True\n'
  278. self.app.expected_calls[
  279. (vm, 'admin.vm.volume.Info', 'private', None)] = \
  280. b'0\x00pool=lvm\n' \
  281. b'vid=vm-' + vm.encode() + b'/private\n' \
  282. b'size=2147483648\n' \
  283. b'usage=214748364\n' \
  284. b'rw=True\n' \
  285. b'internal=True\n' \
  286. b'save_on_stop=True\n' \
  287. b'snap_on_start=False\n'
  288. self.app.expected_calls[
  289. (vm, 'admin.vm.volume.Info', 'volatile', None)] = \
  290. b'0\x00pool=lvm\n' \
  291. b'vid=vm-' + vm.encode() + b'/volatile\n' \
  292. b'size=10737418240\n' \
  293. b'usage=0\n' \
  294. b'rw=True\n' \
  295. b'internal=True\n' \
  296. b'source=None\n' \
  297. b'save_on_stop=False\n' \
  298. b'snap_on_start=False\n'
  299. self.app.expected_calls[
  300. (vm, 'admin.vm.volume.Info', 'kernel', None)] = \
  301. b'0\x00pool=linux-kernel\n' \
  302. b'vid=\n' \
  303. b'size=0\n' \
  304. b'usage=0\n' \
  305. b'rw=False\n' \
  306. b'internal=True\n' \
  307. b'source=None\n' \
  308. b'save_on_stop=False\n' \
  309. b'snap_on_start=False\n'
  310. self.app.expected_calls[
  311. (vm, 'admin.vm.volume.List', None, None)] = \
  312. b'0\x00root\nprivate\nvolatile\nkernel\n'
  313. self.app.expected_calls[
  314. (src, 'admin.vm.volume.CloneFrom', 'private', None)] = \
  315. b'0\x00token-private'
  316. self.app.expected_calls[
  317. (dst, 'admin.vm.volume.CloneTo', 'private', b'token-private')] = \
  318. b'0\x00'
  319. self.app.expected_calls[
  320. ('dom0', 'admin.property.Get', 'default_pool_private', None)] = \
  321. b'0\0default=True type=str lvm'
  322. def test_030_clone(self):
  323. self.clone_setup_common_calls('test-vm', 'new-name')
  324. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  325. b'0\x00new-name class=AppVM state=Halted\n' \
  326. b'test-vm class=AppVM state=Halted\n' \
  327. b'test-template class=TemplateVM state=Halted\n' \
  328. b'test-net class=AppVM state=Halted\n'
  329. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  330. 'test-template', b'name=new-name label=red')] = b'0\x00'
  331. new_vm = self.app.clone_vm('test-vm', 'new-name')
  332. self.assertEqual(new_vm.name, 'new-name')
  333. self.check_output_mock.assert_called_once_with(
  334. ['qvm-appmenus', '--init', '--update',
  335. '--source', 'test-vm', 'new-name'],
  336. stderr=subprocess.STDOUT
  337. )
  338. self.assertAllCalled()
  339. def test_031_clone_object(self):
  340. self.clone_setup_common_calls('test-vm', 'new-name')
  341. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  342. b'0\x00new-name class=AppVM state=Halted\n' \
  343. b'test-vm class=AppVM state=Halted\n' \
  344. b'test-template class=TemplateVM state=Halted\n' \
  345. b'test-net class=AppVM state=Halted\n'
  346. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  347. 'test-template', b'name=new-name label=red')] = b'0\x00'
  348. new_vm = self.app.clone_vm(self.app.domains['test-vm'], 'new-name')
  349. self.assertEqual(new_vm.name, 'new-name')
  350. self.assertAllCalled()
  351. def test_032_clone_pool(self):
  352. self.clone_setup_common_calls('test-vm', 'new-name')
  353. for volume in ('root', 'private', 'volatile', 'kernel'):
  354. del self.app.expected_calls[
  355. 'test-vm', 'admin.vm.volume.Info', volume, None]
  356. del self.app.expected_calls[
  357. 'dom0', 'admin.property.Get', 'default_pool_private', None]
  358. self.app.expected_calls[('dom0', 'admin.vm.CreateInPool.AppVM',
  359. 'test-template',
  360. b'name=new-name label=red pool=some-pool')] = b'0\x00'
  361. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  362. b'0\x00new-name class=AppVM state=Halted\n' \
  363. b'test-vm class=AppVM state=Halted\n' \
  364. b'test-template class=TemplateVM state=Halted\n' \
  365. b'test-net class=AppVM state=Halted\n'
  366. new_vm = self.app.clone_vm('test-vm', 'new-name', pool='some-pool')
  367. self.assertEqual(new_vm.name, 'new-name')
  368. self.assertAllCalled()
  369. def test_033_clone_pools(self):
  370. self.clone_setup_common_calls('test-vm', 'new-name')
  371. for volume in ('root', 'private', 'volatile', 'kernel'):
  372. del self.app.expected_calls[
  373. 'test-vm', 'admin.vm.volume.Info', volume, None]
  374. del self.app.expected_calls[
  375. 'dom0', 'admin.property.Get', 'default_pool_private', None]
  376. self.app.expected_calls[('dom0', 'admin.vm.CreateInPool.AppVM',
  377. 'test-template',
  378. b'name=new-name label=red pool:private=some-pool '
  379. b'pool:volatile=other-pool')] = b'0\x00'
  380. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  381. b'0\x00new-name class=AppVM state=Halted\n' \
  382. b'test-vm class=AppVM state=Halted\n' \
  383. b'test-template class=TemplateVM state=Halted\n' \
  384. b'test-net class=AppVM state=Halted\n'
  385. new_vm = self.app.clone_vm('test-vm', 'new-name',
  386. pools={'private': 'some-pool', 'volatile': 'other-pool'})
  387. self.assertEqual(new_vm.name, 'new-name')
  388. self.assertAllCalled()
  389. def test_034_clone_class_change(self):
  390. self.clone_setup_common_calls('test-vm', 'new-name')
  391. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  392. b'0\x00new-name class=StandaloneVM state=Halted\n' \
  393. b'test-vm class=AppVM state=Halted\n' \
  394. b'test-template class=TemplateVM state=Halted\n' \
  395. b'test-net class=AppVM state=Halted\n'
  396. self.app.expected_calls[('dom0', 'admin.vm.Create.StandaloneVM',
  397. 'test-template', b'name=new-name label=red')] = b'0\x00'
  398. self.app.expected_calls[
  399. ('new-name', 'admin.vm.volume.Info', 'root', None)] = \
  400. b'0\x00pool=lvm\n' \
  401. b'vid=vm-new-name/root\n' \
  402. b'size=10737418240\n' \
  403. b'usage=2147483648\n' \
  404. b'rw=True\n' \
  405. b'internal=True\n' \
  406. b'source=None\n' \
  407. b'save_on_stop=True\n' \
  408. b'snap_on_start=False\n'
  409. self.app.expected_calls[
  410. ('test-vm', 'admin.vm.volume.CloneFrom', 'root', None)] = \
  411. b'0\x00token-root'
  412. self.app.expected_calls[
  413. ('new-name', 'admin.vm.volume.CloneTo', 'root', b'token-root')] = \
  414. b'0\x00'
  415. new_vm = self.app.clone_vm('test-vm', 'new-name',
  416. new_cls='StandaloneVM')
  417. self.assertEqual(new_vm.name, 'new-name')
  418. self.assertEqual(new_vm.klass, 'StandaloneVM')
  419. self.assertAllCalled()
  420. def test_035_clone_fail(self):
  421. self.app.expected_calls[
  422. ('test-vm', 'admin.vm.property.List', None, None)] = \
  423. b'0\0qid\nname\ntemplate\nlabel\nmemory\n'
  424. # simplify it a little, shouldn't get this far anyway
  425. self.app.expected_calls[
  426. ('test-vm', 'admin.vm.volume.List', None, None)] = \
  427. b'0\x00'
  428. self.app.expected_calls[
  429. ('test-vm', 'admin.vm.property.Get', 'label', None)] = \
  430. b'0\0default=False type=label red'
  431. self.app.expected_calls[
  432. ('test-vm', 'admin.vm.property.Get', 'template', None)] = \
  433. b'0\0default=False type=vm test-template'
  434. self.app.expected_calls[
  435. ('test-vm', 'admin.vm.property.Get', 'memory', None)] = \
  436. b'0\0default=False type=int 400'
  437. self.app.expected_calls[
  438. ('new-name', 'admin.vm.property.Set', 'memory', b'400')] = \
  439. b'2\0QubesException\0\0something happened\0'
  440. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  441. b'0\x00new-name class=AppVM state=Halted\n' \
  442. b'test-vm class=AppVM state=Halted\n' \
  443. b'test-template class=TemplateVM state=Halted\n' \
  444. b'test-net class=AppVM state=Halted\n'
  445. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  446. 'test-template', b'name=new-name label=red')] = b'0\x00'
  447. self.app.expected_calls[('new-name', 'admin.vm.Remove', None, None)] = \
  448. b'0\x00'
  449. with self.assertRaises(qubesadmin.exc.QubesException):
  450. self.app.clone_vm('test-vm', 'new-name')
  451. self.assertAllCalled()
  452. def test_036_clone_ignore_errors_prop(self):
  453. self.clone_setup_common_calls('test-vm', 'new-name')
  454. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  455. b'0\x00new-name class=AppVM state=Halted\n' \
  456. b'test-vm class=AppVM state=Halted\n' \
  457. b'test-template class=TemplateVM state=Halted\n' \
  458. b'test-net class=AppVM state=Halted\n'
  459. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  460. 'test-template', b'name=new-name label=red')] = b'0\x00'
  461. self.app.expected_calls[
  462. ('new-name', 'admin.vm.property.Set', 'memory', b'400')] = \
  463. b'2\0QubesException\0\0something happened\0'
  464. new_vm = self.app.clone_vm('test-vm', 'new-name', ignore_errors=True)
  465. self.assertEqual(new_vm.name, 'new-name')
  466. self.assertAllCalled()
  467. def test_037_clone_ignore_errors_feature(self):
  468. self.clone_setup_common_calls('test-vm', 'new-name')
  469. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  470. b'0\x00new-name class=AppVM state=Halted\n' \
  471. b'test-vm class=AppVM state=Halted\n' \
  472. b'test-template class=TemplateVM state=Halted\n' \
  473. b'test-net class=AppVM state=Halted\n'
  474. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  475. 'test-template', b'name=new-name label=red')] = b'0\x00'
  476. self.app.expected_calls[
  477. ('new-name', 'admin.vm.feature.Set', 'feat2', b'1')] = \
  478. b'2\0QubesException\0\0something happened\0'
  479. new_vm = self.app.clone_vm('test-vm', 'new-name', ignore_errors=True)
  480. self.assertEqual(new_vm.name, 'new-name')
  481. self.assertAllCalled()
  482. def test_038_clone_ignore_errors_tag(self):
  483. self.clone_setup_common_calls('test-vm', 'new-name')
  484. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  485. b'0\x00new-name class=AppVM state=Halted\n' \
  486. b'test-vm class=AppVM state=Halted\n' \
  487. b'test-template class=TemplateVM state=Halted\n' \
  488. b'test-net class=AppVM state=Halted\n'
  489. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  490. 'test-template', b'name=new-name label=red')] = b'0\x00'
  491. self.app.expected_calls[
  492. ('new-name', 'admin.vm.tag.Set', 'tag1', None)] = \
  493. b'2\0QubesException\0\0something happened\0'
  494. new_vm = self.app.clone_vm('test-vm', 'new-name', ignore_errors=True)
  495. self.assertEqual(new_vm.name, 'new-name')
  496. self.assertAllCalled()
  497. def test_039_clone_ignore_errors_firewall(self):
  498. self.clone_setup_common_calls('test-vm', 'new-name')
  499. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  500. b'0\x00new-name class=AppVM state=Halted\n' \
  501. b'test-vm class=AppVM state=Halted\n' \
  502. b'test-template class=TemplateVM state=Halted\n' \
  503. b'test-net class=AppVM state=Halted\n'
  504. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  505. 'test-template', b'name=new-name label=red')] = b'0\x00'
  506. self.app.expected_calls[
  507. ('new-name', 'admin.vm.firewall.Set', None,
  508. b'action=drop dst4=192.168.0.0/24\naction=accept\n')] = \
  509. b'2\0QubesException\0\0something happened\0'
  510. new_vm = self.app.clone_vm('test-vm', 'new-name', ignore_errors=True)
  511. self.assertEqual(new_vm.name, 'new-name')
  512. self.assertAllCalled()
  513. def test_040_clone_ignore_errors_storage(self):
  514. self.clone_setup_common_calls('test-vm', 'new-name')
  515. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  516. b'0\x00new-name class=AppVM state=Halted\n' \
  517. b'test-vm class=AppVM state=Halted\n' \
  518. b'test-template class=TemplateVM state=Halted\n' \
  519. b'test-net class=AppVM state=Halted\n'
  520. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  521. 'test-template', b'name=new-name label=red')] = b'0\x00'
  522. self.app.expected_calls[
  523. ('new-name', 'admin.vm.volume.CloneTo', 'private',
  524. b'token-private')] = \
  525. b'2\0QubesException\0\0something happened\0'
  526. self.app.expected_calls[('new-name', 'admin.vm.Remove', None, None)] = \
  527. b'0\x00'
  528. del self.app.expected_calls[
  529. ('new-name', 'admin.vm.volume.Info', 'root', None)]
  530. del self.app.expected_calls[
  531. ('new-name', 'admin.vm.volume.Info', 'volatile', None)]
  532. with self.assertRaises(qubesadmin.exc.QubesException):
  533. self.app.clone_vm('test-vm', 'new-name', ignore_errors=True)
  534. self.assertAllCalled()
  535. def test_041_clone_fail_storage(self):
  536. self.clone_setup_common_calls('test-vm', 'new-name')
  537. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  538. b'0\x00new-name class=AppVM state=Halted\n' \
  539. b'test-vm class=AppVM state=Halted\n' \
  540. b'test-template class=TemplateVM state=Halted\n' \
  541. b'test-net class=AppVM state=Halted\n'
  542. self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
  543. 'test-template', b'name=new-name label=red')] = b'0\x00'
  544. self.app.expected_calls[
  545. ('new-name', 'admin.vm.volume.CloneTo', 'private',
  546. b'token-private')] = \
  547. b'2\0QubesException\0\0something happened\0'
  548. self.app.expected_calls[('new-name', 'admin.vm.Remove', None, None)] = \
  549. b'0\x00'
  550. del self.app.expected_calls[
  551. ('new-name', 'admin.vm.volume.Info', 'root', None)]
  552. del self.app.expected_calls[
  553. ('new-name', 'admin.vm.volume.Info', 'volatile', None)]
  554. with self.assertRaises(qubesadmin.exc.QubesException):
  555. self.app.clone_vm('test-vm', 'new-name')
  556. self.assertAllCalled()
  557. def test_042_clone_nondefault_pool(self):
  558. self.clone_setup_common_calls('test-vm', 'new-name')
  559. self.app.expected_calls[
  560. ('test-vm', 'admin.vm.volume.Info', 'private', None)] = \
  561. b'0\x00pool=another\n' \
  562. b'vid=vm-test-vm/private\n' \
  563. b'size=2147483648\n' \
  564. b'usage=214748364\n' \
  565. b'rw=True\n' \
  566. b'internal=True\n' \
  567. b'save_on_stop=True\n' \
  568. b'snap_on_start=False\n'
  569. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  570. b'0\x00new-name class=AppVM state=Halted\n' \
  571. b'test-vm class=AppVM state=Halted\n' \
  572. b'test-template class=TemplateVM state=Halted\n' \
  573. b'test-net class=AppVM state=Halted\n'
  574. self.app.expected_calls[('dom0', 'admin.vm.CreateInPool.AppVM',
  575. 'test-template', b'name=new-name label=red pool:private=another')]\
  576. = b'0\x00'
  577. new_vm = self.app.clone_vm('test-vm', 'new-name')
  578. self.assertEqual(new_vm.name, 'new-name')
  579. self.check_output_mock.assert_called_once_with(
  580. ['qvm-appmenus', '--init', '--update',
  581. '--source', 'test-vm', 'new-name'],
  582. stderr=subprocess.STDOUT
  583. )
  584. self.assertAllCalled()
  585. class TC_20_QubesLocal(unittest.TestCase):
  586. def setUp(self):
  587. super(TC_20_QubesLocal, self).setUp()
  588. self.socket_dir = tempfile.mkdtemp()
  589. self.orig_sock = qubesadmin.config.QUBESD_SOCKET
  590. qubesadmin.config.QUBESD_SOCKET = os.path.join(self.socket_dir, 'sock')
  591. self.proc = None
  592. self.app = qubesadmin.app.QubesLocal()
  593. def listen_and_send(self, send_data):
  594. '''Listen on socket and send data in response.
  595. :param bytes send_data: data to send
  596. '''
  597. self.socket_pipe, child_pipe = multiprocessing.Pipe()
  598. self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  599. self.socket.bind(os.path.join(self.socket_dir, 'sock'))
  600. self.socket.listen(1)
  601. def worker(sock, pipe, send_data_):
  602. conn, addr = sock.accept()
  603. pipe.send(conn.makefile('rb').read())
  604. conn.sendall(send_data_)
  605. conn.close()
  606. self.proc = multiprocessing.Process(target=worker,
  607. args=(self.socket, child_pipe, send_data))
  608. self.proc.start()
  609. self.socket.close()
  610. def get_request(self):
  611. '''Get request sent to "qubesd" mock'''
  612. return self.socket_pipe.recv()
  613. def tearDown(self):
  614. qubesadmin.config.QUBESD_SOCKET = self.orig_sock
  615. if self.proc is not None:
  616. try:
  617. self.proc.terminate()
  618. except OSError:
  619. pass
  620. shutil.rmtree(self.socket_dir)
  621. super(TC_20_QubesLocal, self).tearDown()
  622. def test_000_qubesd_call(self):
  623. self.listen_and_send(b'0\0')
  624. self.app.qubesd_call('test-vm', 'some.method', 'arg1', b'payload')
  625. self.assertEqual(self.get_request(),
  626. b'dom0\0some.method\0test-vm\0arg1\0payload')
  627. def test_001_qubesd_call_none_arg(self):
  628. self.listen_and_send(b'0\0')
  629. self.app.qubesd_call('test-vm', 'some.method', None, b'payload')
  630. self.assertEqual(self.get_request(),
  631. b'dom0\0some.method\0test-vm\0\0payload')
  632. def test_002_qubesd_call_none_payload(self):
  633. self.listen_and_send(b'0\0')
  634. self.app.qubesd_call('test-vm', 'some.method', None, None)
  635. self.assertEqual(self.get_request(),
  636. b'dom0\0some.method\0test-vm\0\0')
  637. def test_003_qubesd_call_payload_stream(self):
  638. # this should really be in setUp()...
  639. tmpdir = tempfile.mkdtemp()
  640. self.addCleanup(shutil.rmtree, tmpdir)
  641. service_path = os.path.join(tmpdir, 'test.service')
  642. payload_input = os.path.join(tmpdir, 'payload-input')
  643. with open(service_path, 'w') as f:
  644. f.write('#!/bin/bash\n'
  645. 'env > {dir}/env\n'
  646. 'echo "$@" > {dir}/args\n'
  647. 'cat > {dir}/payload\n'
  648. 'echo -en \'0\\0return-value\'\n'.format(dir=tmpdir))
  649. os.chmod(service_path, 0o755)
  650. with open(payload_input, 'w+') as payload_file:
  651. payload_file.write('some payload\n')
  652. payload_file.seek(0)
  653. with mock.patch('qubesadmin.config.QREXEC_SERVICES_DIR',
  654. tmpdir):
  655. value = self.app.qubesd_call('test-vm', 'test.service',
  656. 'some-arg', payload_stream=payload_file)
  657. self.assertEqual(value, b'return-value')
  658. self.assertTrue(os.path.exists(tmpdir + '/env'))
  659. with open(tmpdir + '/env') as env:
  660. self.assertIn('QREXEC_REMOTE_DOMAIN=dom0\n', env)
  661. self.assertIn('QREXEC_REQUESTED_TARGET=test-vm\n', env)
  662. self.assertTrue(os.path.exists(tmpdir + '/args'))
  663. with open(tmpdir + '/args') as args:
  664. self.assertEqual(args.read(), 'some-arg\n')
  665. self.assertTrue(os.path.exists(tmpdir + '/payload'))
  666. with open(tmpdir + '/payload') as payload:
  667. self.assertEqual(payload.read(), 'some payload\n')
  668. def test_004_qubesd_call_payload_stream_proc(self):
  669. # this should really be in setUp()...
  670. tmpdir = tempfile.mkdtemp()
  671. self.addCleanup(shutil.rmtree, tmpdir)
  672. service_path = os.path.join(tmpdir, 'test.service')
  673. echo = subprocess.Popen(['echo', 'some payload'],
  674. stdout=subprocess.PIPE)
  675. with open(service_path, 'w') as f:
  676. f.write('#!/bin/bash\n'
  677. 'env > {dir}/env\n'
  678. 'echo "$@" > {dir}/args\n'
  679. 'cat > {dir}/payload\n'
  680. 'echo -en \'0\\0return-value\'\n'.format(dir=tmpdir))
  681. os.chmod(service_path, 0o755)
  682. with mock.patch('qubesadmin.config.QREXEC_SERVICES_DIR',
  683. tmpdir):
  684. value = self.app.qubesd_call('test-vm', 'test.service',
  685. 'some-arg', payload_stream=echo.stdout)
  686. echo.stdout.close()
  687. self.assertEqual(value, b'return-value')
  688. self.assertTrue(os.path.exists(tmpdir + '/env'))
  689. with open(tmpdir + '/env') as env:
  690. self.assertIn('QREXEC_REMOTE_DOMAIN=dom0\n', env)
  691. self.assertIn('QREXEC_REQUESTED_TARGET=test-vm\n', env)
  692. self.assertTrue(os.path.exists(tmpdir + '/args'))
  693. with open(tmpdir + '/args') as args:
  694. self.assertEqual(args.read(), 'some-arg\n')
  695. self.assertTrue(os.path.exists(tmpdir + '/payload'))
  696. with open(tmpdir + '/payload') as payload:
  697. self.assertEqual(payload.read(), 'some payload\n')
  698. @mock.patch('os.isatty', lambda fd: fd == 2)
  699. def test_010_run_service(self):
  700. self.listen_and_send(b'0\0')
  701. with mock.patch('subprocess.Popen') as mock_proc:
  702. p = self.app.run_service('some-vm', 'service.name')
  703. mock_proc.assert_called_once_with([
  704. qubesadmin.config.QREXEC_CLIENT,
  705. '-d', 'some-vm', '-T', 'DEFAULT:QUBESRPC service.name dom0'],
  706. stdin=subprocess.PIPE, stdout=subprocess.PIPE,
  707. stderr=subprocess.PIPE)
  708. self.assertEqual(self.get_request(),
  709. b'dom0\0admin.vm.Start\0some-vm\0\0')
  710. def test_011_run_service_filter_esc(self):
  711. self.listen_and_send(b'0\0')
  712. with mock.patch('subprocess.Popen') as mock_proc:
  713. p = self.app.run_service('some-vm', 'service.name', filter_esc=True)
  714. mock_proc.assert_called_once_with([
  715. qubesadmin.config.QREXEC_CLIENT,
  716. '-d', 'some-vm', '-t', '-T',
  717. 'DEFAULT:QUBESRPC service.name dom0'],
  718. stdin=subprocess.PIPE, stdout=subprocess.PIPE,
  719. stderr=subprocess.PIPE)
  720. self.assertEqual(self.get_request(),
  721. b'dom0\0admin.vm.Start\0some-vm\0\0')
  722. @mock.patch('os.isatty', lambda fd: fd == 2)
  723. def test_012_run_service_user(self):
  724. self.listen_and_send(b'0\0')
  725. with mock.patch('subprocess.Popen') as mock_proc:
  726. p = self.app.run_service('some-vm', 'service.name', user='user')
  727. mock_proc.assert_called_once_with([
  728. qubesadmin.config.QREXEC_CLIENT,
  729. '-d', 'some-vm', '-T',
  730. 'user:QUBESRPC service.name dom0'],
  731. stdin=subprocess.PIPE, stdout=subprocess.PIPE,
  732. stderr=subprocess.PIPE)
  733. self.assertEqual(self.get_request(),
  734. b'dom0\0admin.vm.Start\0some-vm\0\0')
  735. def test_013_run_service_default_target(self):
  736. with self.assertRaises(ValueError):
  737. self.app.run_service('', 'service.name')
  738. class TC_30_QubesRemote(unittest.TestCase):
  739. def setUp(self):
  740. super(TC_30_QubesRemote, self).setUp()
  741. self.proc_mock = mock.Mock()
  742. self.proc_mock.configure_mock(**{
  743. 'return_value.returncode': 0
  744. })
  745. self.proc_patch = mock.patch('subprocess.Popen', self.proc_mock)
  746. self.proc_patch.start()
  747. self.app = qubesadmin.app.QubesRemote()
  748. def set_proc_stdout(self, send_data):
  749. self.proc_mock.configure_mock(**{
  750. 'return_value.communicate.return_value': (send_data, None)
  751. })
  752. def tearDown(self):
  753. self.proc_patch.stop()
  754. super(TC_30_QubesRemote, self).tearDown()
  755. def test_000_qubesd_call(self):
  756. self.set_proc_stdout(b'0\0')
  757. self.app.qubesd_call('test-vm', 'some.method', 'arg1', b'payload')
  758. self.assertEqual(self.proc_mock.mock_calls, [
  759. mock.call([qubesadmin.config.QREXEC_CLIENT_VM, 'test-vm',
  760. 'some.method+arg1'],
  761. stdin=subprocess.PIPE, stdout=subprocess.PIPE,
  762. stderr=subprocess.PIPE),
  763. mock.call().communicate(b'payload')
  764. ])
  765. def test_001_qubesd_call_none_arg(self):
  766. self.set_proc_stdout(b'0\0')
  767. self.app.qubesd_call('test-vm', 'some.method', None, b'payload')
  768. self.assertEqual(self.proc_mock.mock_calls, [
  769. mock.call([qubesadmin.config.QREXEC_CLIENT_VM, 'test-vm',
  770. 'some.method'],
  771. stdin=subprocess.PIPE, stdout=subprocess.PIPE,
  772. stderr=subprocess.PIPE),
  773. mock.call().communicate(b'payload')
  774. ])
  775. def test_002_qubesd_call_none_payload(self):
  776. self.set_proc_stdout(b'0\0')
  777. self.app.qubesd_call('test-vm', 'some.method', None, None)
  778. self.assertEqual(self.proc_mock.mock_calls, [
  779. mock.call([qubesadmin.config.QREXEC_CLIENT_VM, 'test-vm',
  780. 'some.method'],
  781. stdin=subprocess.PIPE, stdout=subprocess.PIPE,
  782. stderr=subprocess.PIPE),
  783. mock.call().communicate(None)
  784. ])
  785. def test_003_qubesd_call_payload_stream(self):
  786. self.set_proc_stdout(b'0\0return-value')
  787. tmpdir = tempfile.mkdtemp()
  788. self.addCleanup(shutil.rmtree, tmpdir)
  789. payload_input = os.path.join(tmpdir, 'payload-input')
  790. with open(payload_input, 'w+') as payload_file:
  791. payload_file.write('some payload\n')
  792. payload_file.seek(0)
  793. value = self.app.qubesd_call('test-vm', 'some.method',
  794. 'some-arg', payload_stream=payload_file)
  795. self.assertEqual(self.proc_mock.mock_calls, [
  796. mock.call([qubesadmin.config.QREXEC_CLIENT_VM, 'test-vm',
  797. 'some.method+some-arg'],
  798. stdin=payload_file, stdout=subprocess.PIPE,
  799. stderr=subprocess.PIPE),
  800. mock.call().communicate(None)
  801. ])
  802. self.assertEqual(value, b'return-value')
  803. def test_010_run_service(self):
  804. self.app.run_service('some-vm', 'service.name')
  805. self.proc_mock.assert_called_once_with([
  806. qubesadmin.config.QREXEC_CLIENT_VM,
  807. 'some-vm', 'service.name'],
  808. stdin=subprocess.PIPE, stdout=subprocess.PIPE,
  809. stderr=subprocess.PIPE)
  810. def test_011_run_service_filter_esc(self):
  811. with self.assertRaises(NotImplementedError):
  812. p = self.app.run_service('some-vm', 'service.name', filter_esc=True)
  813. def test_012_run_service_user(self):
  814. with self.assertRaises(ValueError):
  815. p = self.app.run_service('some-vm', 'service.name', user='user')
  816. def test_013_run_service_default_target(self):
  817. self.app.run_service('', 'service.name')
  818. self.proc_mock.assert_called_once_with([
  819. qubesadmin.config.QREXEC_CLIENT_VM,
  820. '', 'service.name'],
  821. stdin=subprocess.PIPE, stdout=subprocess.PIPE,
  822. stderr=subprocess.PIPE)