__init__.py 42 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895
  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 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 General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU General Public License along
  19. # with this program; if not, see <http://www.gnu.org/licenses/>.
  20. import os
  21. import socket
  22. import unittest.mock
  23. import shutil
  24. import qubes.tests
  25. import qubespolicy
  26. tmp_policy_dir = '/tmp/policy'
  27. system_info = {
  28. 'domains': {
  29. 'dom0': {
  30. 'tags': ['dom0-tag'],
  31. 'type': 'AdminVM',
  32. 'default_dispvm': 'default-dvm',
  33. 'template_for_dispvms': False,
  34. },
  35. 'test-vm1': {
  36. 'tags': ['tag1', 'tag2'],
  37. 'type': 'AppVM',
  38. 'default_dispvm': 'default-dvm',
  39. 'template_for_dispvms': False,
  40. },
  41. 'test-vm2': {
  42. 'tags': ['tag2'],
  43. 'type': 'AppVM',
  44. 'default_dispvm': 'default-dvm',
  45. 'template_for_dispvms': False,
  46. },
  47. 'test-vm3': {
  48. 'tags': ['tag3'],
  49. 'type': 'AppVM',
  50. 'default_dispvm': 'default-dvm',
  51. 'template_for_dispvms': True,
  52. },
  53. 'default-dvm': {
  54. 'tags': [],
  55. 'type': 'AppVM',
  56. 'default_dispvm': 'default-dvm',
  57. 'template_for_dispvms': True,
  58. },
  59. 'test-invalid-dvm': {
  60. 'tags': ['tag1', 'tag2'],
  61. 'type': 'AppVM',
  62. 'default_dispvm': 'test-vm1',
  63. 'template_for_dispvms': False,
  64. },
  65. 'test-no-dvm': {
  66. 'tags': ['tag1', 'tag2'],
  67. 'type': 'AppVM',
  68. 'default_dispvm': None,
  69. 'template_for_dispvms': False,
  70. },
  71. 'test-template': {
  72. 'tags': ['tag1', 'tag2'],
  73. 'type': 'TemplateVM',
  74. 'default_dispvm': 'default-dvm',
  75. 'template_for_dispvms': False,
  76. },
  77. 'test-standalone': {
  78. 'tags': ['tag1', 'tag2'],
  79. 'type': 'StandaloneVM',
  80. 'default_dispvm': 'default-dvm',
  81. 'template_for_dispvms': False,
  82. },
  83. }
  84. }
  85. class TC_00_PolicyRule(qubes.tests.QubesTestCase):
  86. def test_000_verify_target_value(self):
  87. self.assertTrue(
  88. qubespolicy.verify_target_value(system_info, 'test-vm1'))
  89. self.assertTrue(
  90. qubespolicy.verify_target_value(system_info, 'default-dvm'))
  91. self.assertTrue(
  92. qubespolicy.verify_target_value(system_info, '$dispvm'))
  93. self.assertTrue(
  94. qubespolicy.verify_target_value(system_info, '$dispvm:default-dvm'))
  95. self.assertTrue(
  96. qubespolicy.verify_target_value(system_info, 'test-template'))
  97. self.assertTrue(
  98. qubespolicy.verify_target_value(system_info, 'test-standalone'))
  99. self.assertTrue(
  100. qubespolicy.verify_target_value(system_info, '$adminvm'))
  101. self.assertFalse(
  102. qubespolicy.verify_target_value(system_info, 'no-such-vm'))
  103. self.assertFalse(
  104. qubespolicy.verify_target_value(system_info,
  105. '$dispvm:test-invalid-dvm'))
  106. self.assertFalse(
  107. qubespolicy.verify_target_value(system_info, '$dispvm:test-vm1'))
  108. self.assertFalse(
  109. qubespolicy.verify_target_value(system_info, ''))
  110. self.assertFalse(
  111. qubespolicy.verify_target_value(system_info, '$default'))
  112. self.assertFalse(
  113. qubespolicy.verify_target_value(system_info, '$anyvm'))
  114. self.assertFalse(
  115. qubespolicy.verify_target_value(system_info, '$tag:tag1'))
  116. self.assertFalse(
  117. qubespolicy.verify_target_value(system_info, '$dispvm:$tag:tag1'))
  118. self.assertFalse(
  119. qubespolicy.verify_target_value(system_info, '$invalid'))
  120. def test_010_verify_special_value(self):
  121. self.assertTrue(qubespolicy.verify_special_value('$tag:tag',
  122. for_target=False))
  123. self.assertTrue(qubespolicy.verify_special_value('$tag:other-tag',
  124. for_target=False))
  125. self.assertTrue(qubespolicy.verify_special_value('$type:AppVM',
  126. for_target=False))
  127. self.assertTrue(qubespolicy.verify_special_value('$adminvm',
  128. for_target=False))
  129. self.assertTrue(qubespolicy.verify_special_value('$dispvm:some-vm',
  130. for_target=True))
  131. self.assertTrue(qubespolicy.verify_special_value('$dispvm:$tag:tag1',
  132. for_target=True))
  133. self.assertFalse(qubespolicy.verify_special_value('$default',
  134. for_target=False))
  135. self.assertFalse(qubespolicy.verify_special_value('$dispvm',
  136. for_target=False))
  137. self.assertFalse(qubespolicy.verify_special_value('$dispvm:some-vm',
  138. for_target=False))
  139. self.assertFalse(qubespolicy.verify_special_value('$dispvm:$tag:tag1',
  140. for_target=False))
  141. self.assertFalse(qubespolicy.verify_special_value('$invalid',
  142. for_target=False))
  143. self.assertFalse(qubespolicy.verify_special_value('vm-name',
  144. for_target=False))
  145. self.assertFalse(qubespolicy.verify_special_value('$tag:',
  146. for_target=False))
  147. self.assertFalse(qubespolicy.verify_special_value('$type:',
  148. for_target=False))
  149. def test_020_line_simple(self):
  150. line = qubespolicy.PolicyRule('$anyvm $anyvm ask', 'filename', 12)
  151. self.assertEqual(line.filename, 'filename')
  152. self.assertEqual(line.lineno, 12)
  153. self.assertEqual(line.action, qubespolicy.Action.ask)
  154. self.assertEqual(line.source, '$anyvm')
  155. self.assertEqual(line.target, '$anyvm')
  156. self.assertEqual(line.full_action, 'ask')
  157. self.assertIsNone(line.override_target)
  158. self.assertIsNone(line.override_user)
  159. self.assertIsNone(line.default_target)
  160. def test_021_line_simple(self):
  161. # also check spaces in action field
  162. line = qubespolicy.PolicyRule(
  163. '$tag:tag1 $type:AppVM ask, target=test-vm2, user=user',
  164. 'filename', 12)
  165. self.assertEqual(line.filename, 'filename')
  166. self.assertEqual(line.lineno, 12)
  167. self.assertEqual(line.action, qubespolicy.Action.ask)
  168. self.assertEqual(line.source, '$tag:tag1')
  169. self.assertEqual(line.target, '$type:AppVM')
  170. self.assertEqual(line.full_action, 'ask, target=test-vm2, user=user')
  171. self.assertEqual(line.override_target, 'test-vm2')
  172. self.assertEqual(line.override_user, 'user')
  173. self.assertIsNone(line.default_target)
  174. def test_022_line_simple(self):
  175. line = qubespolicy.PolicyRule(
  176. '$anyvm $default allow,target=$dispvm:test-vm2',
  177. 'filename', 12)
  178. self.assertEqual(line.filename, 'filename')
  179. self.assertEqual(line.lineno, 12)
  180. self.assertEqual(line.action, qubespolicy.Action.allow)
  181. self.assertEqual(line.source, '$anyvm')
  182. self.assertEqual(line.target, '$default')
  183. self.assertEqual(line.full_action, 'allow,target=$dispvm:test-vm2')
  184. self.assertEqual(line.override_target, '$dispvm:test-vm2')
  185. self.assertIsNone(line.override_user)
  186. self.assertIsNone(line.default_target)
  187. def test_023_line_simple(self):
  188. line = qubespolicy.PolicyRule(
  189. '$anyvm $default ask,default_target=test-vm1',
  190. 'filename', 12)
  191. self.assertEqual(line.filename, 'filename')
  192. self.assertEqual(line.lineno, 12)
  193. self.assertEqual(line.action, qubespolicy.Action.ask)
  194. self.assertEqual(line.source, '$anyvm')
  195. self.assertEqual(line.target, '$default')
  196. self.assertEqual(line.full_action, 'ask,default_target=test-vm1')
  197. self.assertIsNone(line.override_target)
  198. self.assertIsNone(line.override_user)
  199. self.assertEqual(line.default_target, 'test-vm1')
  200. def test_024_line_simple(self):
  201. line = qubespolicy.PolicyRule(
  202. '$anyvm $adminvm ask,default_target=$adminvm',
  203. 'filename', 12)
  204. self.assertEqual(line.filename, 'filename')
  205. self.assertEqual(line.lineno, 12)
  206. self.assertEqual(line.action, qubespolicy.Action.ask)
  207. self.assertEqual(line.source, '$anyvm')
  208. self.assertEqual(line.target, '$adminvm')
  209. self.assertEqual(line.full_action, 'ask,default_target=$adminvm')
  210. self.assertIsNone(line.override_target)
  211. self.assertIsNone(line.override_user)
  212. self.assertEqual(line.default_target, '$adminvm')
  213. def test_030_line_invalid(self):
  214. invalid_lines = [
  215. '$dispvm $default allow', # $dispvm can't be a source
  216. '$default $default allow', # $default can't be a source
  217. '$anyvm $default allow,target=$dispvm:$tag:tag1', # $dispvm:$tag
  218. # as override target
  219. '$anyvm $default allow,target=$tag:tag1', # $tag as override target
  220. '$anyvm $default deny,target=test-vm1', # target= used with deny
  221. '$anyvm $anyvm deny,default_target=test-vm1', # default_target=
  222. # with deny
  223. '$anyvm $anyvm deny,user=user', # user= with deny
  224. '$anyvm $anyvm invalid', # invalid action
  225. '$anyvm $anyvm allow,invalid=xx', # invalid option
  226. '$anyvm $anyvm', # missing action
  227. '$anyvm $anyvm allow,default_target=test-vm1', # default_target=
  228. # with allow
  229. '$invalid $anyvm allow', # invalid source
  230. '$anyvm $invalid deny', # invalid target
  231. '', # empty line
  232. '$anyvm $anyvm allow extra', # trailing words
  233. '$anyvm $default allow', # $default allow without target=
  234. ]
  235. for line in invalid_lines:
  236. with self.subTest(line):
  237. with self.assertRaises(qubespolicy.PolicySyntaxError):
  238. qubespolicy.PolicyRule(line, 'filename', 12)
  239. def test_040_match_single(self):
  240. is_match_single = qubespolicy.PolicyRule.is_match_single
  241. self.assertTrue(is_match_single(system_info, '$anyvm', 'test-vm1'))
  242. self.assertTrue(is_match_single(system_info, '$anyvm', '$default'))
  243. self.assertTrue(is_match_single(system_info, '$anyvm', ''))
  244. self.assertTrue(is_match_single(system_info, '$default', ''))
  245. self.assertTrue(is_match_single(system_info, '$default', '$default'))
  246. self.assertTrue(is_match_single(system_info, '$tag:tag1', 'test-vm1'))
  247. self.assertTrue(is_match_single(system_info, '$type:AppVM', 'test-vm1'))
  248. self.assertTrue(is_match_single(system_info,
  249. '$type:TemplateVM', 'test-template'))
  250. self.assertTrue(is_match_single(system_info, '$anyvm', '$dispvm'))
  251. self.assertTrue(is_match_single(system_info,
  252. '$anyvm', '$dispvm:default-dvm'))
  253. self.assertTrue(is_match_single(system_info, '$dispvm', '$dispvm'))
  254. self.assertTrue(is_match_single(system_info,
  255. '$dispvm:$tag:tag3', '$dispvm:test-vm3'))
  256. self.assertTrue(is_match_single(system_info, '$adminvm', '$adminvm'))
  257. self.assertTrue(is_match_single(system_info, '$adminvm', 'dom0'))
  258. self.assertTrue(is_match_single(system_info, 'dom0', '$adminvm'))
  259. self.assertTrue(is_match_single(system_info, 'dom0', 'dom0'))
  260. self.assertTrue(is_match_single(system_info,
  261. '$dispvm:default-dvm', '$dispvm:default-dvm'))
  262. self.assertTrue(is_match_single(system_info, '$anyvm', '$dispvm'))
  263. self.assertTrue(is_match_single(system_info, '$anyvm', 'test-vm1'))
  264. self.assertTrue(is_match_single(system_info, '$anyvm', 'test-vm1'))
  265. self.assertTrue(is_match_single(system_info, '$anyvm', 'test-vm1'))
  266. self.assertFalse(is_match_single(system_info, '$default', 'test-vm1'))
  267. self.assertFalse(is_match_single(system_info, '$tag:tag1', 'test-vm3'))
  268. self.assertFalse(is_match_single(system_info, '$anyvm', 'no-such-vm'))
  269. # test-vm1.template_for_dispvms=False
  270. self.assertFalse(is_match_single(system_info,
  271. '$anyvm', '$dispvm:test-vm1'))
  272. # test-vm1.template_for_dispvms=False
  273. self.assertFalse(is_match_single(system_info,
  274. '$dispvm:test-vm1', '$dispvm:test-vm1'))
  275. self.assertFalse(is_match_single(system_info,
  276. '$dispvm:$tag:tag1', '$dispvm:test-vm1'))
  277. # test-vm3 has not tag1
  278. self.assertFalse(is_match_single(system_info,
  279. '$dispvm:$tag:tag1', '$dispvm:test-vm3'))
  280. # default-dvm has no tag3
  281. self.assertFalse(is_match_single(system_info,
  282. '$dispvm:$tag:tag3', '$dispvm:default-dvm'))
  283. self.assertFalse(is_match_single(system_info, '$anyvm', 'dom0'))
  284. self.assertFalse(is_match_single(system_info, '$anyvm', '$adminvm'))
  285. self.assertFalse(is_match_single(system_info,
  286. '$tag:dom0-tag', '$adminvm'))
  287. self.assertFalse(is_match_single(system_info,
  288. '$type:AdminVM', '$adminvm'))
  289. self.assertFalse(is_match_single(system_info,
  290. '$tag:dom0-tag', 'dom0'))
  291. self.assertFalse(is_match_single(system_info,
  292. '$type:AdminVM', 'dom0'))
  293. self.assertFalse(is_match_single(system_info, '$tag:tag1', 'dom0'))
  294. self.assertFalse(is_match_single(system_info, '$anyvm', '$tag:tag1'))
  295. self.assertFalse(is_match_single(system_info, '$anyvm', '$type:AppVM'))
  296. self.assertFalse(is_match_single(system_info, '$anyvm', '$invalid'))
  297. self.assertFalse(is_match_single(system_info, '$invalid', '$invalid'))
  298. self.assertFalse(is_match_single(system_info, '$anyvm', 'no-such-vm'))
  299. self.assertFalse(is_match_single(system_info,
  300. 'no-such-vm', 'no-such-vm'))
  301. self.assertFalse(is_match_single(system_info, '$dispvm', 'test-vm1'))
  302. self.assertFalse(is_match_single(system_info, '$dispvm', 'default-dvm'))
  303. self.assertFalse(is_match_single(system_info,
  304. '$dispvm:default-dvm', 'default-dvm'))
  305. self.assertFalse(is_match_single(system_info, '$anyvm', 'test-vm1\n'))
  306. self.assertFalse(is_match_single(system_info, '$anyvm', 'test-vm1 '))
  307. def test_050_match(self):
  308. line = qubespolicy.PolicyRule('$anyvm $anyvm allow')
  309. self.assertTrue(line.is_match(system_info, 'test-vm1', 'test-vm2'))
  310. line = qubespolicy.PolicyRule('$anyvm $anyvm allow')
  311. self.assertFalse(line.is_match(system_info, 'no-such-vm', 'test-vm2'))
  312. line = qubespolicy.PolicyRule('$anyvm $anyvm allow')
  313. self.assertFalse(line.is_match(system_info, 'test-vm1', 'no-such-vm'))
  314. line = qubespolicy.PolicyRule('$anyvm $dispvm allow')
  315. self.assertTrue(line.is_match(system_info, 'test-vm1', '$dispvm'))
  316. line = qubespolicy.PolicyRule('$anyvm $dispvm allow')
  317. self.assertFalse(line.is_match(system_info,
  318. 'test-vm1', '$dispvm:default-dvm'))
  319. line = qubespolicy.PolicyRule('$anyvm $dispvm:default-dvm allow')
  320. self.assertTrue(line.is_match(system_info, 'test-vm1', '$dispvm'))
  321. line = qubespolicy.PolicyRule('$anyvm $dispvm:default-dvm allow')
  322. self.assertTrue(line.is_match(system_info,
  323. 'test-vm1', '$dispvm:default-dvm'))
  324. line = qubespolicy.PolicyRule('$anyvm $dispvm:$tag:tag3 allow')
  325. self.assertTrue(line.is_match(system_info,
  326. 'test-vm1', '$dispvm:test-vm3'))
  327. def test_060_expand_target(self):
  328. lines = {
  329. '$anyvm $anyvm allow': ['test-vm1', 'test-vm2', 'test-vm3',
  330. '$dispvm:test-vm3',
  331. 'default-dvm', '$dispvm:default-dvm', 'test-invalid-dvm',
  332. 'test-no-dvm', 'test-template', 'test-standalone', '$dispvm'],
  333. '$anyvm $dispvm allow': ['$dispvm'],
  334. '$anyvm $dispvm:default-dvm allow': ['$dispvm:default-dvm'],
  335. # no DispVM from test-vm1 allowed
  336. '$anyvm $dispvm:test-vm1 allow': [],
  337. '$anyvm $dispvm:test-vm3 allow': ['$dispvm:test-vm3'],
  338. '$anyvm $dispvm:$tag:tag1 allow': [],
  339. '$anyvm $dispvm:$tag:tag3 allow': ['$dispvm:test-vm3'],
  340. '$anyvm test-vm1 allow': ['test-vm1'],
  341. '$anyvm $type:AppVM allow': ['test-vm1', 'test-vm2', 'test-vm3',
  342. 'default-dvm', 'test-invalid-dvm', 'test-no-dvm'],
  343. '$anyvm $type:TemplateVM allow': ['test-template'],
  344. '$anyvm $tag:tag1 allow': ['test-vm1', 'test-invalid-dvm',
  345. 'test-template', 'test-standalone', 'test-no-dvm'],
  346. '$anyvm $tag:tag2 allow': ['test-vm1', 'test-vm2',
  347. 'test-invalid-dvm', 'test-template', 'test-standalone',
  348. 'test-no-dvm'],
  349. '$anyvm $tag:no-such-tag allow': [],
  350. }
  351. for line in lines:
  352. with self.subTest(line):
  353. policy_line = qubespolicy.PolicyRule(line)
  354. self.assertCountEqual(list(policy_line.expand_target(system_info)),
  355. lines[line])
  356. def test_070_expand_override_target(self):
  357. line = qubespolicy.PolicyRule(
  358. '$anyvm $anyvm allow,target=test-vm2')
  359. self.assertEqual(
  360. line.expand_override_target(system_info, 'test-vm1'),
  361. 'test-vm2')
  362. def test_071_expand_override_target_dispvm(self):
  363. line = qubespolicy.PolicyRule(
  364. '$anyvm $anyvm allow,target=$dispvm')
  365. self.assertEqual(
  366. line.expand_override_target(system_info, 'test-vm1'),
  367. '$dispvm:default-dvm')
  368. def test_072_expand_override_target_dispvm_specific(self):
  369. line = qubespolicy.PolicyRule(
  370. '$anyvm $anyvm allow,target=$dispvm:test-vm3')
  371. self.assertEqual(
  372. line.expand_override_target(system_info, 'test-vm1'),
  373. '$dispvm:test-vm3')
  374. def test_073_expand_override_target_dispvm_none(self):
  375. line = qubespolicy.PolicyRule(
  376. '$anyvm $anyvm allow,target=$dispvm')
  377. self.assertEqual(
  378. line.expand_override_target(system_info, 'test-no-dvm'),
  379. None)
  380. def test_074_expand_override_target_dom0(self):
  381. line = qubespolicy.PolicyRule(
  382. '$anyvm $anyvm allow,target=dom0')
  383. self.assertEqual(
  384. line.expand_override_target(system_info, 'test-no-dvm'),
  385. 'dom0')
  386. def test_075_expand_override_target_dom0(self):
  387. line = qubespolicy.PolicyRule(
  388. '$anyvm $anyvm allow,target=$adminvm')
  389. self.assertEqual(
  390. line.expand_override_target(system_info, 'test-no-dvm'),
  391. '$adminvm')
  392. class TC_10_PolicyAction(qubes.tests.QubesTestCase):
  393. def test_000_init(self):
  394. rule = qubespolicy.PolicyRule('$anyvm $anyvm deny')
  395. with self.assertRaises(qubespolicy.AccessDenied):
  396. qubespolicy.PolicyAction('test.service', 'test-vm1', 'test-vm2',
  397. rule, 'test-vm2')
  398. def test_001_init(self):
  399. rule = qubespolicy.PolicyRule('$anyvm $anyvm ask')
  400. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  401. None, rule, 'test-vm2', ['test-vm2', 'test-vm3'])
  402. self.assertEqual(action.service, 'test.service')
  403. self.assertEqual(action.source, 'test-vm1')
  404. self.assertIsNone(action.target)
  405. self.assertEqual(action.original_target, 'test-vm2')
  406. self.assertEqual(action.targets_for_ask, ['test-vm2', 'test-vm3'])
  407. self.assertEqual(action.rule, rule)
  408. self.assertEqual(action.action, qubespolicy.Action.ask)
  409. def test_002_init_invalid(self):
  410. rule_ask = qubespolicy.PolicyRule('$anyvm $anyvm ask')
  411. rule_allow = qubespolicy.PolicyRule('$anyvm $anyvm allow')
  412. with self.assertRaises(AssertionError):
  413. qubespolicy.PolicyAction('test.service', 'test-vm1',
  414. None, rule_allow, 'test-vm2', None)
  415. with self.assertRaises(AssertionError):
  416. qubespolicy.PolicyAction('test.service', 'test-vm1',
  417. 'test-vm2', rule_allow, 'test-vm2', ['test-vm2', 'test-vm3'])
  418. with self.assertRaises(AssertionError):
  419. qubespolicy.PolicyAction('test.service', 'test-vm1',
  420. None, rule_ask, 'test-vm2', None)
  421. def test_003_init_default_target(self):
  422. rule_ask = qubespolicy.PolicyRule('$anyvm $anyvm ask')
  423. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  424. 'test-vm1', rule_ask, 'test-vm2', ['test-vm2'])
  425. self.assertIsNone(action.target)
  426. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  427. 'test-vm2', rule_ask, 'test-vm2', ['test-vm2'])
  428. self.assertEqual(action.target, 'test-vm2')
  429. def test_010_handle_user_response(self):
  430. rule = qubespolicy.PolicyRule('$anyvm $anyvm ask')
  431. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  432. None, rule, 'test-vm2', ['test-vm2', 'test-vm3'])
  433. action.handle_user_response(True, 'test-vm2')
  434. self.assertEqual(action.action, qubespolicy.Action.allow)
  435. self.assertEqual(action.target, 'test-vm2')
  436. def test_011_handle_user_response(self):
  437. rule = qubespolicy.PolicyRule('$anyvm $anyvm ask')
  438. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  439. None, rule, 'test-vm2', ['test-vm2', 'test-vm3'])
  440. with self.assertRaises(AssertionError):
  441. action.handle_user_response(True, 'test-no-dvm')
  442. def test_012_handle_user_response(self):
  443. rule = qubespolicy.PolicyRule('$anyvm $anyvm ask')
  444. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  445. None, rule, 'test-vm2', ['test-vm2', 'test-vm3'])
  446. with self.assertRaises(qubespolicy.AccessDenied):
  447. action.handle_user_response(False, None)
  448. self.assertEqual(action.action, qubespolicy.Action.deny)
  449. def test_013_handle_user_response_with_default_target(self):
  450. rule = qubespolicy.PolicyRule(
  451. '$anyvm $anyvm ask,default_target=test-vm2')
  452. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  453. None, rule, 'test-vm2', ['test-vm2', 'test-vm3'])
  454. action.handle_user_response(True, 'test-vm2')
  455. self.assertEqual(action.action, qubespolicy.Action.allow)
  456. self.assertEqual(action.target, 'test-vm2')
  457. @unittest.mock.patch('qubespolicy.qubesd_call')
  458. @unittest.mock.patch('subprocess.call')
  459. def test_020_execute(self, mock_subprocess, mock_qubesd_call):
  460. rule = qubespolicy.PolicyRule('$anyvm $anyvm allow')
  461. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  462. 'test-vm2', rule, 'test-vm2')
  463. action.execute('some-ident')
  464. self.assertEqual(mock_qubesd_call.mock_calls,
  465. [unittest.mock.call('test-vm2', 'admin.vm.Start')])
  466. self.assertEqual(mock_subprocess.mock_calls,
  467. [unittest.mock.call([qubespolicy.QREXEC_CLIENT, '-d', 'test-vm2',
  468. '-c', 'some-ident', 'DEFAULT:QUBESRPC test.service test-vm1'])])
  469. @unittest.mock.patch('qubespolicy.qubesd_call')
  470. @unittest.mock.patch('subprocess.call')
  471. def test_021_execute_dom0(self, mock_subprocess, mock_qubesd_call):
  472. rule = qubespolicy.PolicyRule('$anyvm dom0 allow')
  473. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  474. 'dom0', rule, 'dom0')
  475. action.execute('some-ident')
  476. self.assertEqual(mock_qubesd_call.mock_calls, [])
  477. self.assertEqual(mock_subprocess.mock_calls,
  478. [unittest.mock.call([qubespolicy.QREXEC_CLIENT, '-d', 'dom0',
  479. '-c', 'some-ident',
  480. qubespolicy.QUBES_RPC_MULTIPLEXER_PATH +
  481. ' test.service test-vm1 dom0'])])
  482. @unittest.mock.patch('qubespolicy.qubesd_call')
  483. @unittest.mock.patch('subprocess.call')
  484. def test_022_execute_dispvm(self, mock_subprocess, mock_qubesd_call):
  485. rule = qubespolicy.PolicyRule('$anyvm $dispvm:default-dvm allow')
  486. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  487. '$dispvm:default-dvm', rule, '$dispvm:default-dvm')
  488. mock_qubesd_call.side_effect = (lambda target, call:
  489. b'dispvm-name' if call == 'admin.vm.CreateDisposable' else
  490. unittest.mock.DEFAULT)
  491. action.execute('some-ident')
  492. self.assertEqual(mock_qubesd_call.mock_calls,
  493. [unittest.mock.call('default-dvm', 'admin.vm.CreateDisposable'),
  494. unittest.mock.call('dispvm-name', 'admin.vm.Start'),
  495. unittest.mock.call('dispvm-name', 'admin.vm.Kill')])
  496. self.assertEqual(mock_subprocess.mock_calls,
  497. [unittest.mock.call([qubespolicy.QREXEC_CLIENT, '-d', 'dispvm-name',
  498. '-c', 'some-ident', '-W',
  499. 'DEFAULT:QUBESRPC test.service test-vm1'])])
  500. @unittest.mock.patch('qubespolicy.qubesd_call')
  501. @unittest.mock.patch('subprocess.call')
  502. def test_023_execute_already_running(self, mock_subprocess,
  503. mock_qubesd_call):
  504. rule = qubespolicy.PolicyRule('$anyvm $anyvm allow')
  505. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  506. 'test-vm2', rule, 'test-vm2')
  507. mock_qubesd_call.side_effect = \
  508. qubespolicy.QubesMgmtException('QubesVMNotHaltedError')
  509. action.execute('some-ident')
  510. self.assertEqual(mock_qubesd_call.mock_calls,
  511. [unittest.mock.call('test-vm2', 'admin.vm.Start')])
  512. self.assertEqual(mock_subprocess.mock_calls,
  513. [unittest.mock.call([qubespolicy.QREXEC_CLIENT, '-d', 'test-vm2',
  514. '-c', 'some-ident', 'DEFAULT:QUBESRPC test.service test-vm1'])])
  515. @unittest.mock.patch('qubespolicy.qubesd_call')
  516. @unittest.mock.patch('subprocess.call')
  517. def test_024_execute_startup_error(self, mock_subprocess,
  518. mock_qubesd_call):
  519. rule = qubespolicy.PolicyRule('$anyvm $anyvm allow')
  520. action = qubespolicy.PolicyAction('test.service', 'test-vm1',
  521. 'test-vm2', rule, 'test-vm2')
  522. mock_qubesd_call.side_effect = \
  523. qubespolicy.QubesMgmtException('QubesVMError')
  524. with self.assertRaises(qubespolicy.QubesMgmtException):
  525. action.execute('some-ident')
  526. self.assertEqual(mock_qubesd_call.mock_calls,
  527. [unittest.mock.call('test-vm2', 'admin.vm.Start')])
  528. self.assertEqual(mock_subprocess.mock_calls, [])
  529. class TC_20_Policy(qubes.tests.QubesTestCase):
  530. def setUp(self):
  531. super(TC_20_Policy, self).setUp()
  532. if not os.path.exists(tmp_policy_dir):
  533. os.mkdir(tmp_policy_dir)
  534. def tearDown(self):
  535. shutil.rmtree(tmp_policy_dir)
  536. super(TC_20_Policy, self).tearDown()
  537. def test_000_load(self):
  538. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  539. f.write('test-vm1 test-vm2 allow\n')
  540. f.write('\n')
  541. f.write('# comment\n')
  542. f.write('test-vm2 test-vm3 ask\n')
  543. f.write(' # comment \n')
  544. f.write('$anyvm $anyvm ask\n')
  545. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  546. self.assertEqual(policy.service, 'test.service')
  547. self.assertEqual(len(policy.policy_rules), 3)
  548. self.assertEqual(policy.policy_rules[0].source, 'test-vm1')
  549. self.assertEqual(policy.policy_rules[0].target, 'test-vm2')
  550. self.assertEqual(policy.policy_rules[0].action,
  551. qubespolicy.Action.allow)
  552. def test_001_not_existent(self):
  553. with self.assertRaises(qubespolicy.AccessDenied):
  554. qubespolicy.Policy('no-such.service', tmp_policy_dir)
  555. def test_002_include(self):
  556. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  557. f.write('test-vm1 test-vm2 allow\n')
  558. f.write('$include:test.service2\n')
  559. f.write('$anyvm $anyvm deny\n')
  560. with open(os.path.join(tmp_policy_dir, 'test.service2'), 'w') as f:
  561. f.write('test-vm3 $default allow,target=test-vm2\n')
  562. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  563. self.assertEqual(policy.service, 'test.service')
  564. self.assertEqual(len(policy.policy_rules), 3)
  565. self.assertEqual(policy.policy_rules[0].source, 'test-vm1')
  566. self.assertEqual(policy.policy_rules[0].target, 'test-vm2')
  567. self.assertEqual(policy.policy_rules[0].action,
  568. qubespolicy.Action.allow)
  569. self.assertEqual(policy.policy_rules[0].filename,
  570. tmp_policy_dir + '/test.service')
  571. self.assertEqual(policy.policy_rules[0].lineno, 1)
  572. self.assertEqual(policy.policy_rules[1].source, 'test-vm3')
  573. self.assertEqual(policy.policy_rules[1].target, '$default')
  574. self.assertEqual(policy.policy_rules[1].action,
  575. qubespolicy.Action.allow)
  576. self.assertEqual(policy.policy_rules[1].filename,
  577. tmp_policy_dir + '/test.service2')
  578. self.assertEqual(policy.policy_rules[1].lineno, 1)
  579. self.assertEqual(policy.policy_rules[2].source, '$anyvm')
  580. self.assertEqual(policy.policy_rules[2].target, '$anyvm')
  581. self.assertEqual(policy.policy_rules[2].action,
  582. qubespolicy.Action.deny)
  583. self.assertEqual(policy.policy_rules[2].filename,
  584. tmp_policy_dir + '/test.service')
  585. self.assertEqual(policy.policy_rules[2].lineno, 3)
  586. def test_010_find_rule(self):
  587. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  588. f.write('test-vm1 test-vm2 allow\n')
  589. f.write('test-vm1 $anyvm ask\n')
  590. f.write('test-vm2 $tag:tag1 deny\n')
  591. f.write('test-vm2 $tag:tag2 allow\n')
  592. f.write('test-vm2 $dispvm:$tag:tag3 allow\n')
  593. f.write('test-vm2 $dispvm:$tag:tag2 allow\n')
  594. f.write('test-vm2 $dispvm:default-dvm allow\n')
  595. f.write('$type:AppVM $default allow,target=test-vm3\n')
  596. f.write('$tag:tag1 $type:AppVM allow\n')
  597. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  598. self.assertEqual(policy.find_matching_rule(
  599. system_info, 'test-vm1', 'test-vm2'), policy.policy_rules[0])
  600. self.assertEqual(policy.find_matching_rule(
  601. system_info, 'test-vm1', 'test-vm3'), policy.policy_rules[1])
  602. self.assertEqual(policy.find_matching_rule(
  603. system_info, 'test-vm2', 'test-vm2'), policy.policy_rules[3])
  604. self.assertEqual(policy.find_matching_rule(
  605. system_info, 'test-vm2', 'test-no-dvm'), policy.policy_rules[2])
  606. # $anyvm matches $default too
  607. self.assertEqual(policy.find_matching_rule(
  608. system_info, 'test-vm1', ''), policy.policy_rules[1])
  609. self.assertEqual(policy.find_matching_rule(
  610. system_info, 'test-vm2', ''), policy.policy_rules[7])
  611. self.assertEqual(policy.find_matching_rule(
  612. system_info, 'test-vm2', '$default'), policy.policy_rules[7])
  613. self.assertEqual(policy.find_matching_rule(
  614. system_info, 'test-no-dvm', 'test-vm3'), policy.policy_rules[8])
  615. self.assertEqual(policy.find_matching_rule(
  616. system_info, 'test-vm2', '$dispvm:test-vm3'),
  617. policy.policy_rules[4])
  618. self.assertEqual(policy.find_matching_rule(
  619. system_info, 'test-vm2', '$dispvm'),
  620. policy.policy_rules[6])
  621. with self.assertRaises(qubespolicy.AccessDenied):
  622. policy.find_matching_rule(
  623. system_info, 'test-no-dvm', 'test-standalone')
  624. with self.assertRaises(qubespolicy.AccessDenied):
  625. policy.find_matching_rule(system_info, 'test-no-dvm', '$dispvm')
  626. with self.assertRaises(qubespolicy.AccessDenied):
  627. policy.find_matching_rule(
  628. system_info, 'test-standalone', '$default')
  629. def test_020_collect_targets_for_ask(self):
  630. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  631. f.write('test-vm1 test-vm2 allow\n')
  632. f.write('test-vm1 $anyvm ask\n')
  633. f.write('test-vm2 $tag:tag1 deny\n')
  634. f.write('test-vm2 $tag:tag2 allow\n')
  635. f.write('test-no-dvm $type:AppVM deny\n')
  636. f.write('$type:AppVM $default allow,target=test-vm3\n')
  637. f.write('$tag:tag1 $type:AppVM allow\n')
  638. f.write('test-no-dvm $dispvm allow\n')
  639. f.write('test-standalone $dispvm allow\n')
  640. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  641. self.assertCountEqual(policy.collect_targets_for_ask(system_info,
  642. 'test-vm1'), ['test-vm1', 'test-vm2', 'test-vm3',
  643. '$dispvm:test-vm3',
  644. 'default-dvm', '$dispvm:default-dvm', 'test-invalid-dvm',
  645. 'test-no-dvm', 'test-template', 'test-standalone'])
  646. self.assertCountEqual(policy.collect_targets_for_ask(system_info,
  647. 'test-vm2'), ['test-vm2', 'test-vm3'])
  648. self.assertCountEqual(policy.collect_targets_for_ask(system_info,
  649. 'test-vm3'), ['test-vm3'])
  650. self.assertCountEqual(policy.collect_targets_for_ask(system_info,
  651. 'test-standalone'), ['test-vm1', 'test-vm2', 'test-vm3',
  652. 'default-dvm', 'test-no-dvm', 'test-invalid-dvm',
  653. '$dispvm:default-dvm'])
  654. self.assertCountEqual(policy.collect_targets_for_ask(system_info,
  655. 'test-no-dvm'), [])
  656. def test_030_eval_simple(self):
  657. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  658. f.write('test-vm1 test-vm2 allow\n')
  659. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  660. action = policy.evaluate(system_info, 'test-vm1', 'test-vm2')
  661. self.assertEqual(action.rule, policy.policy_rules[0])
  662. self.assertEqual(action.action, qubespolicy.Action.allow)
  663. self.assertEqual(action.target, 'test-vm2')
  664. self.assertEqual(action.original_target, 'test-vm2')
  665. self.assertEqual(action.service, 'test.service')
  666. self.assertIsNone(action.targets_for_ask)
  667. with self.assertRaises(qubespolicy.AccessDenied):
  668. policy.evaluate(system_info, 'test-vm2', '$default')
  669. def test_031_eval_default(self):
  670. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  671. f.write('test-vm1 test-vm2 allow\n')
  672. f.write('test-vm1 $default allow,target=test-vm2\n')
  673. f.write('$tag:tag1 test-vm2 ask\n')
  674. f.write('$tag:tag2 $anyvm allow\n')
  675. f.write('test-vm3 $anyvm deny\n')
  676. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  677. action = policy.evaluate(system_info, 'test-vm1', '$default')
  678. self.assertEqual(action.rule, policy.policy_rules[1])
  679. self.assertEqual(action.action, qubespolicy.Action.allow)
  680. self.assertEqual(action.target, 'test-vm2')
  681. self.assertEqual(action.original_target, '$default')
  682. self.assertEqual(action.service, 'test.service')
  683. self.assertIsNone(action.targets_for_ask)
  684. with self.assertRaises(qubespolicy.AccessDenied):
  685. # action allow should hit, but no target specified (either by
  686. # caller or policy)
  687. policy.evaluate(system_info, 'test-standalone', '$default')
  688. def test_032_eval_ask(self):
  689. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  690. f.write('test-vm1 test-vm2 allow\n')
  691. f.write('test-vm1 $default allow,target=test-vm2\n')
  692. f.write('$tag:tag1 test-vm2 ask\n')
  693. f.write('$tag:tag1 test-vm3 ask,default_target=test-vm3\n')
  694. f.write('$tag:tag2 $anyvm allow\n')
  695. f.write('test-vm3 $anyvm deny\n')
  696. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  697. action = policy.evaluate(system_info, 'test-standalone', 'test-vm2')
  698. self.assertEqual(action.rule, policy.policy_rules[2])
  699. self.assertEqual(action.action, qubespolicy.Action.ask)
  700. self.assertIsNone(action.target)
  701. self.assertEqual(action.original_target, 'test-vm2')
  702. self.assertEqual(action.service, 'test.service')
  703. self.assertCountEqual(action.targets_for_ask,
  704. ['test-vm1', 'test-vm2', 'test-vm3', '$dispvm:test-vm3',
  705. 'default-dvm', '$dispvm:default-dvm', 'test-invalid-dvm',
  706. 'test-no-dvm', 'test-template', 'test-standalone'])
  707. def test_033_eval_ask(self):
  708. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  709. f.write('test-vm1 test-vm2 allow\n')
  710. f.write('test-vm1 $default allow,target=test-vm2\n')
  711. f.write('$tag:tag1 test-vm2 ask\n')
  712. f.write('$tag:tag1 test-vm3 ask,default_target=test-vm3\n')
  713. f.write('$tag:tag2 $anyvm allow\n')
  714. f.write('test-vm3 $anyvm deny\n')
  715. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  716. action = policy.evaluate(system_info, 'test-standalone', 'test-vm3')
  717. self.assertEqual(action.rule, policy.policy_rules[3])
  718. self.assertEqual(action.action, qubespolicy.Action.ask)
  719. self.assertEqual(action.target, 'test-vm3')
  720. self.assertEqual(action.original_target, 'test-vm3')
  721. self.assertEqual(action.service, 'test.service')
  722. self.assertCountEqual(action.targets_for_ask,
  723. ['test-vm1', 'test-vm2', 'test-vm3', '$dispvm:test-vm3',
  724. 'default-dvm', '$dispvm:default-dvm', 'test-invalid-dvm',
  725. 'test-no-dvm', 'test-template', 'test-standalone'])
  726. def test_034_eval_resolve_dispvm(self):
  727. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  728. f.write('test-vm3 $dispvm allow\n')
  729. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  730. action = policy.evaluate(system_info, 'test-vm3', '$dispvm')
  731. self.assertEqual(action.rule, policy.policy_rules[0])
  732. self.assertEqual(action.action, qubespolicy.Action.allow)
  733. self.assertEqual(action.target, '$dispvm:default-dvm')
  734. self.assertEqual(action.original_target, '$dispvm')
  735. self.assertEqual(action.service, 'test.service')
  736. self.assertIsNone(action.targets_for_ask)
  737. def test_035_eval_resolve_dispvm_fail(self):
  738. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  739. f.write('test-no-dvm $dispvm allow\n')
  740. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  741. with self.assertRaises(qubespolicy.AccessDenied):
  742. policy.evaluate(system_info, 'test-no-dvm', '$dispvm')
  743. def test_036_eval_invalid_override_target(self):
  744. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  745. f.write('test-vm3 $anyvm allow,target=no-such-vm\n')
  746. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  747. with self.assertRaises(qubespolicy.AccessDenied):
  748. policy.evaluate(system_info, 'test-vm3', '$default')
  749. def test_037_eval_ask_no_targets(self):
  750. with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
  751. f.write('test-vm3 $default ask\n')
  752. policy = qubespolicy.Policy('test.service', tmp_policy_dir)
  753. with self.assertRaises(qubespolicy.AccessDenied):
  754. policy.evaluate(system_info, 'test-vm3', '$default')
  755. class TC_30_Misc(qubes.tests.QubesTestCase):
  756. @unittest.mock.patch('socket.socket')
  757. def test_000_qubesd_call(self, mock_socket):
  758. mock_config = {
  759. 'return_value.makefile.return_value.read.return_value': b'0\x00data'
  760. }
  761. mock_socket.configure_mock(**mock_config)
  762. result = qubespolicy.qubesd_call('test', 'internal.method')
  763. self.assertEqual(result, b'data')
  764. self.assertEqual(mock_socket.mock_calls, [
  765. unittest.mock.call(socket.AF_UNIX, socket.SOCK_STREAM),
  766. unittest.mock.call().connect(qubespolicy.QUBESD_INTERNAL_SOCK),
  767. unittest.mock.call().sendall(b'dom0'),
  768. unittest.mock.call().sendall(b'\x00'),
  769. unittest.mock.call().sendall(b'internal.method'),
  770. unittest.mock.call().sendall(b'\x00'),
  771. unittest.mock.call().sendall(b'test'),
  772. unittest.mock.call().sendall(b'\x00'),
  773. unittest.mock.call().sendall(b'\x00'),
  774. unittest.mock.call().shutdown(socket.SHUT_WR),
  775. unittest.mock.call().makefile('rb'),
  776. unittest.mock.call().makefile().read(),
  777. ])
  778. @unittest.mock.patch('socket.socket')
  779. def test_001_qubesd_call_arg_payload(self, mock_socket):
  780. mock_config = {
  781. 'return_value.makefile.return_value.read.return_value': b'0\x00data'
  782. }
  783. mock_socket.configure_mock(**mock_config)
  784. result = qubespolicy.qubesd_call('test', 'internal.method', 'arg',
  785. b'payload')
  786. self.assertEqual(result, b'data')
  787. self.assertEqual(mock_socket.mock_calls, [
  788. unittest.mock.call(socket.AF_UNIX, socket.SOCK_STREAM),
  789. unittest.mock.call().connect(qubespolicy.QUBESD_INTERNAL_SOCK),
  790. unittest.mock.call().sendall(b'dom0'),
  791. unittest.mock.call().sendall(b'\x00'),
  792. unittest.mock.call().sendall(b'internal.method'),
  793. unittest.mock.call().sendall(b'\x00'),
  794. unittest.mock.call().sendall(b'test'),
  795. unittest.mock.call().sendall(b'\x00'),
  796. unittest.mock.call().sendall(b'arg'),
  797. unittest.mock.call().sendall(b'\x00'),
  798. unittest.mock.call().sendall(b'payload'),
  799. unittest.mock.call().shutdown(socket.SHUT_WR),
  800. unittest.mock.call().makefile('rb'),
  801. unittest.mock.call().makefile().read(),
  802. ])
  803. @unittest.mock.patch('socket.socket')
  804. def test_002_qubesd_call_exception(self, mock_socket):
  805. mock_config = {
  806. 'return_value.makefile.return_value.read.return_value':
  807. b'2\x00SomeError\x00traceback\x00message\x00'
  808. }
  809. mock_socket.configure_mock(**mock_config)
  810. with self.assertRaises(qubespolicy.QubesMgmtException) as e:
  811. qubespolicy.qubesd_call('test', 'internal.method')
  812. self.assertEqual(e.exception.exc_type, 'SomeError')
  813. self.assertEqual(mock_socket.mock_calls, [
  814. unittest.mock.call(socket.AF_UNIX, socket.SOCK_STREAM),
  815. unittest.mock.call().connect(qubespolicy.QUBESD_INTERNAL_SOCK),
  816. unittest.mock.call().sendall(b'dom0'),
  817. unittest.mock.call().sendall(b'\x00'),
  818. unittest.mock.call().sendall(b'internal.method'),
  819. unittest.mock.call().sendall(b'\x00'),
  820. unittest.mock.call().sendall(b'test'),
  821. unittest.mock.call().sendall(b'\x00'),
  822. unittest.mock.call().sendall(b'\x00'),
  823. unittest.mock.call().shutdown(socket.SHUT_WR),
  824. unittest.mock.call().makefile('rb'),
  825. unittest.mock.call().makefile().read(),
  826. ])