test_firewall.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562
  1. import logging
  2. import operator
  3. from unittest import TestCase
  4. from unittest.mock import patch
  5. import qubesagent.firewall
  6. class DummyIptablesRestore(object):
  7. # pylint: disable=too-few-public-methods
  8. def __init__(self, worker_mock, family):
  9. self._worker_mock = worker_mock
  10. self._family = family
  11. self.returncode = 0
  12. def communicate(self, stdin=None):
  13. self._worker_mock.loaded_iptables[self._family] = stdin
  14. return ("", None)
  15. class DummyQubesDB(object):
  16. def __init__(self, worker_mock):
  17. self._worker_mock = worker_mock
  18. self.entries = {}
  19. self.pending_watches = []
  20. def read(self, key):
  21. try:
  22. return self.entries[key]
  23. except KeyError:
  24. return None
  25. def multiread(self, prefix):
  26. result = {}
  27. for key, value in self.entries.items():
  28. if key.startswith(prefix):
  29. result[key] = value
  30. return result
  31. def list(self, prefix):
  32. result = []
  33. for key in self.entries.keys():
  34. if key.startswith(prefix):
  35. result.append(key)
  36. return result
  37. def watch(self, path):
  38. pass
  39. def read_watch(self):
  40. try:
  41. return self.pending_watches.pop(0)
  42. except IndexError:
  43. return None
  44. class FirewallWorker(qubesagent.firewall.FirewallWorker):
  45. def __init__(self):
  46. # pylint: disable=super-init-not-called
  47. # don't call super on purpose - avoid connecting to QubesDB
  48. # super(FirewallWorker, self).__init__()
  49. self.qdb = DummyQubesDB(self)
  50. self.log = logging.getLogger('qubes.tests')
  51. self.init_called = False
  52. self.cleanup_called = False
  53. self.user_script_called = False
  54. self.rules = {}
  55. def apply_rules(self, source_addr, rules):
  56. self.rules[source_addr] = rules
  57. def cleanup(self):
  58. self.init_called = True
  59. def init(self):
  60. self.cleanup_called = True
  61. def run_user_script(self):
  62. self.user_script_called = True
  63. class IptablesWorker(qubesagent.firewall.IptablesWorker):
  64. '''Override methods actually modifying system state to only log what
  65. would be done'''
  66. def __init__(self):
  67. # pylint: disable=super-init-not-called
  68. # don't call super on purpose - avoid connecting to QubesDB
  69. # super(IptablesWorker, self).__init__()
  70. # copied __init__:
  71. self.qdb = DummyQubesDB(self)
  72. self.log = logging.getLogger('qubes.tests')
  73. self.chains = {
  74. 4: set(),
  75. 6: set(),
  76. }
  77. #: instead of really running `iptables`, log what would be called
  78. self.called_commands = {
  79. 4: [],
  80. 6: [],
  81. }
  82. #: rules that would be loaded with `iptables-restore`
  83. self.loaded_iptables = {
  84. 4: None,
  85. 6: None,
  86. }
  87. def run_ipt(self, family, args, **kwargs):
  88. self.called_commands[family].append(args)
  89. def run_ipt_restore(self, family, args):
  90. return DummyIptablesRestore(self, family)
  91. @staticmethod
  92. def dns_addresses(family=None):
  93. if family == 4:
  94. return ['1.1.1.1', '2.2.2.2']
  95. else:
  96. return ['2001::1', '2001::2']
  97. class NftablesWorker(qubesagent.firewall.NftablesWorker):
  98. '''Override methods actually modifying system state to only log what
  99. would be done'''
  100. def __init__(self):
  101. # pylint: disable=super-init-not-called
  102. # don't call super on purpose - avoid connecting to QubesDB
  103. # super(IptablesWorker, self).__init__()
  104. # copied __init__:
  105. self.qdb = DummyQubesDB(self)
  106. self.log = logging.getLogger('qubes.tests')
  107. self.chains = {
  108. 4: set(),
  109. 6: set(),
  110. }
  111. #: instead of really running `nft`, log what would be loaded
  112. #: rules that would be loaded with `nft`
  113. self.loaded_rules = []
  114. def run_nft(self, nft_input):
  115. self.loaded_rules.append(nft_input)
  116. @staticmethod
  117. def dns_addresses(family=None):
  118. if family == 4:
  119. return ['1.1.1.1', '2.2.2.2']
  120. else:
  121. return ['2001::1', '2001::2']
  122. class TestIptablesWorker(TestCase):
  123. def setUp(self):
  124. super(TestIptablesWorker, self).setUp()
  125. self.obj = IptablesWorker()
  126. self.subprocess_patch = patch('subprocess.call')
  127. self.subprocess_mock = self.subprocess_patch.start()
  128. def tearDown(self):
  129. self.subprocess_patch.stop()
  130. def test_000_chain_for_addr(self):
  131. self.assertEqual(
  132. self.obj.chain_for_addr('10.137.0.1'), 'qbs-10-137-0-1')
  133. self.assertEqual(
  134. self.obj.chain_for_addr('fd09:24ef:4179:0000::3'),
  135. 'qbs-09-24ef-4179-0000--3')
  136. def test_001_create_chain(self):
  137. testdata = [
  138. (4, '10.137.0.1', 'qbs-10-137-0-1'),
  139. (6, 'fd09:24ef:4179:0000::3', 'qbs-fd09-24ef-4179-0000--3')
  140. ]
  141. for family, addr, chain in testdata:
  142. self.obj.create_chain(addr, chain, family)
  143. self.assertEqual(self.obj.called_commands[family],
  144. [['-N', chain],
  145. ['-I', 'QBS-FORWARD', '-s', addr, '-j', chain]])
  146. def test_002_prepare_rules4(self):
  147. rules = [
  148. {'action': 'accept', 'proto': 'tcp',
  149. 'dstports': '80-80', 'dst4': '1.2.3.0/24'},
  150. {'action': 'accept', 'proto': 'udp',
  151. 'dstports': '443-1024', 'dsthost': 'yum.qubes-os.org'},
  152. {'action': 'accept', 'specialtarget': 'dns'},
  153. {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
  154. {'action': 'drop', 'proto': 'icmp'},
  155. {'action': 'drop'},
  156. ]
  157. expected_iptables = (
  158. "*filter\n"
  159. "-A chain -d 1.2.3.0/24 -p tcp --dport 80:80 -j ACCEPT\n"
  160. "-A chain -d 82.94.215.165/32 -p udp --dport 443:1024 -j ACCEPT\n"
  161. "-A chain -d 1.1.1.1/32 -p tcp --dport 53:53 -j ACCEPT\n"
  162. "-A chain -d 2.2.2.2/32 -p tcp --dport 53:53 -j ACCEPT\n"
  163. "-A chain -d 1.1.1.1/32 -p udp --dport 53:53 -j ACCEPT\n"
  164. "-A chain -d 2.2.2.2/32 -p udp --dport 53:53 -j ACCEPT\n"
  165. "-A chain -d 1.1.1.1/32 -p udp --dport 53:53 -j DROP\n"
  166. "-A chain -d 2.2.2.2/32 -p udp --dport 53:53 -j DROP\n"
  167. "-A chain -p icmp -j DROP\n"
  168. "-A chain -j DROP\n"
  169. "COMMIT\n"
  170. )
  171. self.assertEqual(self.obj.prepare_rules('chain', rules, 4),
  172. expected_iptables)
  173. with self.assertRaises(qubesagent.firewall.RuleParseError):
  174. self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4)
  175. with self.assertRaises(qubesagent.firewall.RuleParseError):
  176. self.obj.prepare_rules('chain', [{'dst6': 'a::b'}], 4)
  177. with self.assertRaises(qubesagent.firewall.RuleParseError):
  178. self.obj.prepare_rules('chain', [{'dst4': '3.3.3.3'}], 6)
  179. def test_003_prepare_rules6(self):
  180. rules = [
  181. {'action': 'accept', 'proto': 'tcp',
  182. 'dstports': '80-80', 'dst6': 'a::b/128'},
  183. {'action': 'accept', 'proto': 'tcp',
  184. 'dsthost': 'ripe.net'},
  185. {'action': 'accept', 'specialtarget': 'dns'},
  186. {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
  187. {'action': 'drop', 'proto': 'icmp'},
  188. {'action': 'drop'},
  189. ]
  190. expected_iptables = (
  191. "*filter\n"
  192. "-A chain -d a::b/128 -p tcp --dport 80:80 -j ACCEPT\n"
  193. "-A chain -d 2001:67c:2e8:22::c100:68b/128 -p tcp -j ACCEPT\n"
  194. "-A chain -d 2001::1/128 -p tcp --dport 53:53 -j ACCEPT\n"
  195. "-A chain -d 2001::2/128 -p tcp --dport 53:53 -j ACCEPT\n"
  196. "-A chain -d 2001::1/128 -p udp --dport 53:53 -j ACCEPT\n"
  197. "-A chain -d 2001::2/128 -p udp --dport 53:53 -j ACCEPT\n"
  198. "-A chain -d 2001::1/128 -p udp --dport 53:53 -j DROP\n"
  199. "-A chain -d 2001::2/128 -p udp --dport 53:53 -j DROP\n"
  200. "-A chain -p icmpv6 -j DROP\n"
  201. "-A chain -j DROP\n"
  202. "COMMIT\n"
  203. )
  204. self.assertEqual(self.obj.prepare_rules('chain', rules, 6),
  205. expected_iptables)
  206. def test_004_apply_rules4(self):
  207. rules = [{'action': 'accept'}]
  208. chain = 'qbs-10-137-0-1'
  209. self.obj.apply_rules('10.137.0.1', rules)
  210. self.assertEqual(self.obj.called_commands[4],
  211. [
  212. ['-N', chain],
  213. ['-I', 'QBS-FORWARD', '-s', '10.137.0.1', '-j', chain],
  214. ['-F', chain]])
  215. self.assertEqual(self.obj.loaded_iptables[4],
  216. self.obj.prepare_rules(chain, rules, 4))
  217. self.assertEqual(self.obj.called_commands[6], [])
  218. self.assertIsNone(self.obj.loaded_iptables[6])
  219. def test_005_apply_rules6(self):
  220. rules = [{'action': 'accept'}]
  221. chain = 'qbs-2000--a'
  222. self.obj.apply_rules('2000::a', rules)
  223. self.assertEqual(self.obj.called_commands[6],
  224. [
  225. ['-N', chain],
  226. ['-I', 'QBS-FORWARD', '-s', '2000::a', '-j', chain],
  227. ['-F', chain]])
  228. self.assertEqual(self.obj.loaded_iptables[6],
  229. self.obj.prepare_rules(chain, rules, 6))
  230. self.assertEqual(self.obj.called_commands[4], [])
  231. self.assertIsNone(self.obj.loaded_iptables[4])
  232. def test_006_init(self):
  233. self.obj.init()
  234. self.assertEqual(self.obj.called_commands[4], [
  235. ['-F', 'QBS-FORWARD'],
  236. ['-A', 'QBS-FORWARD', '!', '-i', 'vif+', '-j', 'RETURN'],
  237. ['-A', 'QBS-FORWARD', '-j', 'DROP']])
  238. self.assertEqual(self.obj.called_commands[6], [
  239. ['-F', 'QBS-FORWARD'],
  240. ['-A', 'QBS-FORWARD', '!', '-i', 'vif+', '-j', 'RETURN'],
  241. ['-A', 'QBS-FORWARD', '-j', 'DROP']])
  242. def test_007_cleanup(self):
  243. self.obj.init()
  244. self.obj.create_chain('1.2.3.4', 'chain-ip4-1', 4)
  245. self.obj.create_chain('1.2.3.6', 'chain-ip4-2', 4)
  246. self.obj.create_chain('2000::1', 'chain-ip6-1', 6)
  247. self.obj.create_chain('2000::2', 'chain-ip6-2', 6)
  248. # forget about commands called earlier
  249. self.obj.called_commands[4] = []
  250. self.obj.called_commands[6] = []
  251. self.obj.cleanup()
  252. self.assertEqual([self.obj.called_commands[4][0]] +
  253. sorted(self.obj.called_commands[4][1:], key=operator.itemgetter(1)),
  254. [['-F', 'QBS-FORWARD'],
  255. ['-F', 'chain-ip4-1'],
  256. ['-X', 'chain-ip4-1'],
  257. ['-F', 'chain-ip4-2'],
  258. ['-X', 'chain-ip4-2']])
  259. self.assertEqual([self.obj.called_commands[6][0]] +
  260. sorted(self.obj.called_commands[6][1:], key=operator.itemgetter(1)),
  261. [['-F', 'QBS-FORWARD'],
  262. ['-F', 'chain-ip6-1'],
  263. ['-X', 'chain-ip6-1'],
  264. ['-F', 'chain-ip6-2'],
  265. ['-X', 'chain-ip6-2']])
  266. class TestNftablesWorker(TestCase):
  267. def setUp(self):
  268. super(TestNftablesWorker, self).setUp()
  269. self.obj = NftablesWorker()
  270. self.subprocess_patch = patch('subprocess.call')
  271. self.subprocess_mock = self.subprocess_patch.start()
  272. def tearDown(self):
  273. self.subprocess_patch.stop()
  274. def test_000_chain_for_addr(self):
  275. self.assertEqual(
  276. self.obj.chain_for_addr('10.137.0.1'), 'qbs-10-137-0-1')
  277. self.assertEqual(
  278. self.obj.chain_for_addr('fd09:24ef:4179:0000::3'),
  279. 'qbs-fd09-24ef-4179-0000--3')
  280. def expected_create_chain(self, family, addr, chain):
  281. return (
  282. 'table {family} qubes-firewall {{\n'
  283. ' chain {chain} {{\n'
  284. ' }}\n'
  285. ' chain forward {{\n'
  286. ' {family} saddr {addr} jump {chain}\n'
  287. ' }}\n'
  288. '}}\n'.format(family=family, addr=addr, chain=chain))
  289. def test_001_create_chain(self):
  290. testdata = [
  291. (4, '10.137.0.1', 'qbs-10-137-0-1'),
  292. (6, 'fd09:24ef:4179:0000::3', 'qbs-fd09-24ef-4179-0000--3')
  293. ]
  294. for family, addr, chain in testdata:
  295. self.obj.create_chain(addr, chain, family)
  296. self.assertEqual(self.obj.loaded_rules,
  297. [self.expected_create_chain('ip', '10.137.0.1', 'qbs-10-137-0-1'),
  298. self.expected_create_chain(
  299. 'ip6', 'fd09:24ef:4179:0000::3', 'qbs-fd09-24ef-4179-0000--3'),
  300. ])
  301. def test_002_prepare_rules4(self):
  302. rules = [
  303. {'action': 'accept', 'proto': 'tcp',
  304. 'dstports': '80-80', 'dst4': '1.2.3.0/24'},
  305. {'action': 'accept', 'proto': 'udp',
  306. 'dstports': '443-1024', 'dsthost': 'yum.qubes-os.org'},
  307. {'action': 'accept', 'specialtarget': 'dns'},
  308. {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
  309. {'action': 'drop', 'proto': 'icmp'},
  310. {'action': 'drop'},
  311. ]
  312. expected_nft = (
  313. 'flush chain ip qubes-firewall chain\n'
  314. 'table ip qubes-firewall {\n'
  315. ' chain chain {\n'
  316. ' ip protocol tcp ip daddr 1.2.3.0/24 tcp dport 80 accept\n'
  317. ' ip protocol udp ip daddr { 82.94.215.165/32 } '
  318. 'udp dport 443-1024 accept\n'
  319. ' ip daddr { 1.1.1.1/32, 2.2.2.2/32 } tcp dport 53 accept\n'
  320. ' ip daddr { 1.1.1.1/32, 2.2.2.2/32 } udp dport 53 accept\n'
  321. ' ip protocol udp ip daddr { 1.1.1.1/32, 2.2.2.2/32 } udp dport '
  322. '53 drop\n'
  323. ' ip protocol icmp drop\n'
  324. ' drop\n'
  325. ' }\n'
  326. '}\n'
  327. )
  328. self.assertEqual(self.obj.prepare_rules('chain', rules, 4),
  329. expected_nft)
  330. with self.assertRaises(qubesagent.firewall.RuleParseError):
  331. self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4)
  332. with self.assertRaises(qubesagent.firewall.RuleParseError):
  333. self.obj.prepare_rules('chain', [{'dst6': 'a::b'}], 4)
  334. with self.assertRaises(qubesagent.firewall.RuleParseError):
  335. self.obj.prepare_rules('chain', [{'dst4': '3.3.3.3'}], 6)
  336. def test_003_prepare_rules6(self):
  337. rules = [
  338. {'action': 'accept', 'proto': 'tcp',
  339. 'dstports': '80-80', 'dst6': 'a::b/128'},
  340. {'action': 'accept', 'proto': 'tcp',
  341. 'dsthost': 'ripe.net'},
  342. {'action': 'accept', 'specialtarget': 'dns'},
  343. {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
  344. {'action': 'drop', 'proto': 'icmp', 'icmptype': '128'},
  345. {'action': 'drop'},
  346. ]
  347. expected_nft = (
  348. 'flush chain ip6 qubes-firewall chain\n'
  349. 'table ip6 qubes-firewall {\n'
  350. ' chain chain {\n'
  351. ' ip6 nexthdr tcp ip6 daddr a::b/128 tcp dport 80 accept\n'
  352. ' ip6 nexthdr tcp ip6 daddr { 2001:67c:2e8:22::c100:68b/128 } '
  353. 'accept\n'
  354. ' ip6 daddr { 2001::1/128, 2001::2/128 } tcp dport 53 accept\n'
  355. ' ip6 daddr { 2001::1/128, 2001::2/128 } udp dport 53 accept\n'
  356. ' ip6 nexthdr udp ip6 daddr { 2001::1/128, 2001::2/128 } '
  357. 'udp dport 53 drop\n'
  358. ' ip6 nexthdr icmpv6 icmpv6 type 128 drop\n'
  359. ' drop\n'
  360. ' }\n'
  361. '}\n'
  362. )
  363. self.assertEqual(self.obj.prepare_rules('chain', rules, 6),
  364. expected_nft)
  365. def test_004_apply_rules4(self):
  366. rules = [{'action': 'accept'}]
  367. chain = 'qbs-10-137-0-1'
  368. self.obj.apply_rules('10.137.0.1', rules)
  369. self.assertEqual(self.obj.loaded_rules,
  370. [self.expected_create_chain('ip', '10.137.0.1', chain),
  371. self.obj.prepare_rules(chain, rules, 4),
  372. ])
  373. def test_005_apply_rules6(self):
  374. rules = [{'action': 'accept'}]
  375. chain = 'qbs-2000--a'
  376. self.obj.apply_rules('2000::a', rules)
  377. self.assertEqual(self.obj.loaded_rules,
  378. [self.expected_create_chain('ip6', '2000::a', chain),
  379. self.obj.prepare_rules(chain, rules, 6),
  380. ])
  381. def test_006_init(self):
  382. self.obj.init()
  383. self.assertEqual(self.obj.loaded_rules,
  384. [
  385. 'table ip qubes-firewall {\n'
  386. ' chain forward {\n'
  387. ' type filter hook forward priority 0;\n'
  388. ' policy drop;\n'
  389. ' ct state established,related accept\n'
  390. ' meta iifname != "vif*" accept\n'
  391. ' }\n'
  392. '}\n'
  393. 'table ip6 qubes-firewall {\n'
  394. ' chain forward {\n'
  395. ' type filter hook forward priority 0;\n'
  396. ' policy drop;\n'
  397. ' ct state established,related accept\n'
  398. ' meta iifname != "vif*" accept\n'
  399. ' }\n'
  400. '}\n'
  401. ])
  402. def test_007_cleanup(self):
  403. self.obj.init()
  404. self.obj.create_chain('1.2.3.4', 'chain-ip4-1', 4)
  405. self.obj.create_chain('1.2.3.6', 'chain-ip4-2', 4)
  406. self.obj.create_chain('2000::1', 'chain-ip6-1', 6)
  407. self.obj.create_chain('2000::2', 'chain-ip6-2', 6)
  408. # forget about commands called earlier
  409. self.obj.loaded_rules = []
  410. self.obj.cleanup()
  411. self.assertEqual(self.obj.loaded_rules,
  412. ['delete table ip qubes-firewall\n'
  413. 'delete table ip6 qubes-firewall\n',
  414. ])
  415. class TestFirewallWorker(TestCase):
  416. def setUp(self):
  417. self.obj = FirewallWorker()
  418. rules = {
  419. '10.137.0.1': {
  420. 'policy': 'accept',
  421. '0000': 'proto=tcp dstports=80-80 action=drop',
  422. '0001': 'proto=udp specialtarget=dns action=accept',
  423. '0002': 'proto=udp action=drop',
  424. },
  425. '10.137.0.2': {'policy': 'accept'},
  426. # no policy
  427. '10.137.0.3': {'0000': 'proto=tcp action=accept'},
  428. # no action
  429. '10.137.0.4': {
  430. 'policy': 'drop',
  431. '0000': 'proto=tcp'
  432. },
  433. }
  434. for addr, entries in rules.items():
  435. for key, value in entries.items():
  436. self.obj.qdb.entries[
  437. '/qubes-firewall/{}/{}'.format(addr, key)] = value
  438. self.subprocess_patch = patch('subprocess.call')
  439. self.subprocess_mock = self.subprocess_patch.start()
  440. def tearDown(self):
  441. self.subprocess_patch.stop()
  442. def test_read_rules(self):
  443. expected_rules1 = [
  444. {'proto': 'tcp', 'dstports': '80-80', 'action': 'drop'},
  445. {'proto': 'udp', 'specialtarget': 'dns', 'action': 'accept'},
  446. {'proto': 'udp', 'action': 'drop'},
  447. {'action': 'accept'},
  448. ]
  449. expected_rules2 = [
  450. {'action': 'accept'},
  451. ]
  452. self.assertEqual(self.obj.read_rules('10.137.0.1'), expected_rules1)
  453. self.assertEqual(self.obj.read_rules('10.137.0.2'), expected_rules2)
  454. with self.assertRaises(qubesagent.firewall.RuleParseError):
  455. self.obj.read_rules('10.137.0.3')
  456. with self.assertRaises(qubesagent.firewall.RuleParseError):
  457. self.obj.read_rules('10.137.0.4')
  458. def test_list_targets(self):
  459. self.assertEqual(self.obj.list_targets(), set(['10.137.0.{}'.format(x)
  460. for x in range(1, 5)]))
  461. def test_is_ip6(self):
  462. self.assertTrue(self.obj.is_ip6('2000::abcd'))
  463. self.assertTrue(self.obj.is_ip6('2000:1:2:3:4:5:6:abcd'))
  464. self.assertFalse(self.obj.is_ip6('10.137.0.1'))
  465. def test_handle_addr(self):
  466. self.obj.handle_addr('10.137.0.2')
  467. self.assertEqual(self.obj.rules['10.137.0.2'], [{'action': 'accept'}])
  468. # fallback to block all
  469. self.obj.handle_addr('10.137.0.3')
  470. self.assertEqual(self.obj.rules['10.137.0.3'], [{'action': 'drop'}])
  471. self.obj.handle_addr('10.137.0.4')
  472. self.assertEqual(self.obj.rules['10.137.0.4'], [{'action': 'drop'}])
  473. @patch('os.path.isfile')
  474. @patch('os.access')
  475. @patch('subprocess.call')
  476. def test_run_user_script(self, mock_subprocess, mock_os_access,
  477. mock_os_path_isfile):
  478. mock_os_path_isfile.return_value = False
  479. mock_os_access.return_value = False
  480. super(FirewallWorker, self.obj).run_user_script()
  481. self.assertFalse(mock_subprocess.called)
  482. mock_os_path_isfile.return_value = True
  483. mock_os_access.return_value = False
  484. super(FirewallWorker, self.obj).run_user_script()
  485. self.assertFalse(mock_subprocess.called)
  486. mock_os_path_isfile.return_value = True
  487. mock_os_access.return_value = True
  488. super(FirewallWorker, self.obj).run_user_script()
  489. mock_subprocess.assert_called_once_with(
  490. ['/rw/config/qubes-firewall-user-script'])
  491. def test_main(self):
  492. self.obj.main()
  493. self.assertTrue(self.obj.init_called)
  494. self.assertTrue(self.obj.cleanup_called)
  495. self.assertTrue(self.obj.user_script_called)
  496. self.assertEqual(set(self.obj.rules.keys()), self.obj.list_targets())
  497. # rules content were already tested