test_firewall.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571
  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 REJECT "
  166. "--reject-with icmp-admin-prohibited\n"
  167. "-A chain -d 2.2.2.2/32 -p udp --dport 53:53 -j REJECT "
  168. "--reject-with icmp-admin-prohibited\n"
  169. "-A chain -p icmp -j REJECT "
  170. "--reject-with icmp-admin-prohibited\n"
  171. "-A chain -j REJECT "
  172. "--reject-with icmp-admin-prohibited\n"
  173. "COMMIT\n"
  174. )
  175. self.assertEqual(self.obj.prepare_rules('chain', rules, 4),
  176. expected_iptables)
  177. with self.assertRaises(qubesagent.firewall.RuleParseError):
  178. self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4)
  179. with self.assertRaises(qubesagent.firewall.RuleParseError):
  180. self.obj.prepare_rules('chain', [{'dst6': 'a::b'}], 4)
  181. with self.assertRaises(qubesagent.firewall.RuleParseError):
  182. self.obj.prepare_rules('chain', [{'dst4': '3.3.3.3'}], 6)
  183. def test_003_prepare_rules6(self):
  184. rules = [
  185. {'action': 'accept', 'proto': 'tcp',
  186. 'dstports': '80-80', 'dst6': 'a::b/128'},
  187. {'action': 'accept', 'proto': 'tcp',
  188. 'dsthost': 'ripe.net'},
  189. {'action': 'accept', 'specialtarget': 'dns'},
  190. {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
  191. {'action': 'drop', 'proto': 'icmp'},
  192. {'action': 'drop'},
  193. ]
  194. expected_iptables = (
  195. "*filter\n"
  196. "-A chain -d a::b/128 -p tcp --dport 80:80 -j ACCEPT\n"
  197. "-A chain -d 2001:67c:2e8:22::c100:68b/128 -p tcp -j ACCEPT\n"
  198. "-A chain -d 2001::1/128 -p tcp --dport 53:53 -j ACCEPT\n"
  199. "-A chain -d 2001::2/128 -p tcp --dport 53:53 -j ACCEPT\n"
  200. "-A chain -d 2001::1/128 -p udp --dport 53:53 -j ACCEPT\n"
  201. "-A chain -d 2001::2/128 -p udp --dport 53:53 -j ACCEPT\n"
  202. "-A chain -d 2001::1/128 -p udp --dport 53:53 -j REJECT "
  203. "--reject-with icmp6-adm-prohibited\n"
  204. "-A chain -d 2001::2/128 -p udp --dport 53:53 -j REJECT "
  205. "--reject-with icmp6-adm-prohibited\n"
  206. "-A chain -p icmpv6 -j REJECT "
  207. "--reject-with icmp6-adm-prohibited\n"
  208. "-A chain -j REJECT "
  209. "--reject-with icmp6-adm-prohibited\n"
  210. "COMMIT\n"
  211. )
  212. self.assertEqual(self.obj.prepare_rules('chain', rules, 6),
  213. expected_iptables)
  214. def test_004_apply_rules4(self):
  215. rules = [{'action': 'accept'}]
  216. chain = 'qbs-10-137-0-1'
  217. self.obj.apply_rules('10.137.0.1', rules)
  218. self.assertEqual(self.obj.called_commands[4],
  219. [
  220. ['-N', chain],
  221. ['-I', 'QBS-FORWARD', '-s', '10.137.0.1', '-j', chain],
  222. ['-F', chain]])
  223. self.assertEqual(self.obj.loaded_iptables[4],
  224. self.obj.prepare_rules(chain, rules, 4))
  225. self.assertEqual(self.obj.called_commands[6], [])
  226. self.assertIsNone(self.obj.loaded_iptables[6])
  227. def test_005_apply_rules6(self):
  228. rules = [{'action': 'accept'}]
  229. chain = 'qbs-2000--a'
  230. self.obj.apply_rules('2000::a', rules)
  231. self.assertEqual(self.obj.called_commands[6],
  232. [
  233. ['-N', chain],
  234. ['-I', 'QBS-FORWARD', '-s', '2000::a', '-j', chain],
  235. ['-F', chain]])
  236. self.assertEqual(self.obj.loaded_iptables[6],
  237. self.obj.prepare_rules(chain, rules, 6))
  238. self.assertEqual(self.obj.called_commands[4], [])
  239. self.assertIsNone(self.obj.loaded_iptables[4])
  240. def test_006_init(self):
  241. self.obj.init()
  242. self.assertEqual(self.obj.called_commands[4], [
  243. ['-F', 'QBS-FORWARD'],
  244. ['-A', 'QBS-FORWARD', '!', '-i', 'vif+', '-j', 'RETURN'],
  245. ['-A', 'QBS-FORWARD', '-j', 'DROP']])
  246. self.assertEqual(self.obj.called_commands[6], [
  247. ['-F', 'QBS-FORWARD'],
  248. ['-A', 'QBS-FORWARD', '!', '-i', 'vif+', '-j', 'RETURN'],
  249. ['-A', 'QBS-FORWARD', '-j', 'DROP']])
  250. def test_007_cleanup(self):
  251. self.obj.init()
  252. self.obj.create_chain('1.2.3.4', 'chain-ip4-1', 4)
  253. self.obj.create_chain('1.2.3.6', 'chain-ip4-2', 4)
  254. self.obj.create_chain('2000::1', 'chain-ip6-1', 6)
  255. self.obj.create_chain('2000::2', 'chain-ip6-2', 6)
  256. # forget about commands called earlier
  257. self.obj.called_commands[4] = []
  258. self.obj.called_commands[6] = []
  259. self.obj.cleanup()
  260. self.assertEqual([self.obj.called_commands[4][0]] +
  261. sorted(self.obj.called_commands[4][1:], key=operator.itemgetter(1)),
  262. [['-F', 'QBS-FORWARD'],
  263. ['-F', 'chain-ip4-1'],
  264. ['-X', 'chain-ip4-1'],
  265. ['-F', 'chain-ip4-2'],
  266. ['-X', 'chain-ip4-2']])
  267. self.assertEqual([self.obj.called_commands[6][0]] +
  268. sorted(self.obj.called_commands[6][1:], key=operator.itemgetter(1)),
  269. [['-F', 'QBS-FORWARD'],
  270. ['-F', 'chain-ip6-1'],
  271. ['-X', 'chain-ip6-1'],
  272. ['-F', 'chain-ip6-2'],
  273. ['-X', 'chain-ip6-2']])
  274. class TestNftablesWorker(TestCase):
  275. def setUp(self):
  276. super(TestNftablesWorker, self).setUp()
  277. self.obj = NftablesWorker()
  278. self.subprocess_patch = patch('subprocess.call')
  279. self.subprocess_mock = self.subprocess_patch.start()
  280. def tearDown(self):
  281. self.subprocess_patch.stop()
  282. def test_000_chain_for_addr(self):
  283. self.assertEqual(
  284. self.obj.chain_for_addr('10.137.0.1'), 'qbs-10-137-0-1')
  285. self.assertEqual(
  286. self.obj.chain_for_addr('fd09:24ef:4179:0000::3'),
  287. 'qbs-fd09-24ef-4179-0000--3')
  288. def expected_create_chain(self, family, addr, chain):
  289. return (
  290. 'table {family} qubes-firewall {{\n'
  291. ' chain {chain} {{\n'
  292. ' }}\n'
  293. ' chain forward {{\n'
  294. ' {family} saddr {addr} jump {chain}\n'
  295. ' }}\n'
  296. '}}\n'.format(family=family, addr=addr, chain=chain))
  297. def test_001_create_chain(self):
  298. testdata = [
  299. (4, '10.137.0.1', 'qbs-10-137-0-1'),
  300. (6, 'fd09:24ef:4179:0000::3', 'qbs-fd09-24ef-4179-0000--3')
  301. ]
  302. for family, addr, chain in testdata:
  303. self.obj.create_chain(addr, chain, family)
  304. self.assertEqual(self.obj.loaded_rules,
  305. [self.expected_create_chain('ip', '10.137.0.1', 'qbs-10-137-0-1'),
  306. self.expected_create_chain(
  307. 'ip6', 'fd09:24ef:4179:0000::3', 'qbs-fd09-24ef-4179-0000--3'),
  308. ])
  309. def test_002_prepare_rules4(self):
  310. rules = [
  311. {'action': 'accept', 'proto': 'tcp',
  312. 'dstports': '80-80', 'dst4': '1.2.3.0/24'},
  313. {'action': 'accept', 'proto': 'udp',
  314. 'dstports': '443-1024', 'dsthost': 'yum.qubes-os.org'},
  315. {'action': 'accept', 'specialtarget': 'dns'},
  316. {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
  317. {'action': 'drop', 'proto': 'icmp'},
  318. {'action': 'drop'},
  319. ]
  320. expected_nft = (
  321. 'flush chain ip qubes-firewall chain\n'
  322. 'table ip qubes-firewall {\n'
  323. ' chain chain {\n'
  324. ' ip protocol tcp ip daddr 1.2.3.0/24 tcp dport 80 accept\n'
  325. ' ip protocol udp ip daddr { 82.94.215.165/32 } '
  326. 'udp dport 443-1024 accept\n'
  327. ' ip daddr { 1.1.1.1/32, 2.2.2.2/32 } tcp dport 53 accept\n'
  328. ' ip daddr { 1.1.1.1/32, 2.2.2.2/32 } udp dport 53 accept\n'
  329. ' ip protocol udp ip daddr { 1.1.1.1/32, 2.2.2.2/32 } udp dport '
  330. '53 reject with icmp type admin-prohibited\n'
  331. ' ip protocol icmp reject with icmp type admin-prohibited\n'
  332. ' reject with icmp type admin-prohibited\n'
  333. ' }\n'
  334. '}\n'
  335. )
  336. self.assertEqual(self.obj.prepare_rules('chain', rules, 4),
  337. expected_nft)
  338. with self.assertRaises(qubesagent.firewall.RuleParseError):
  339. self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4)
  340. with self.assertRaises(qubesagent.firewall.RuleParseError):
  341. self.obj.prepare_rules('chain', [{'dst6': 'a::b'}], 4)
  342. with self.assertRaises(qubesagent.firewall.RuleParseError):
  343. self.obj.prepare_rules('chain', [{'dst4': '3.3.3.3'}], 6)
  344. def test_003_prepare_rules6(self):
  345. rules = [
  346. {'action': 'accept', 'proto': 'tcp',
  347. 'dstports': '80-80', 'dst6': 'a::b/128'},
  348. {'action': 'accept', 'proto': 'tcp',
  349. 'dsthost': 'ripe.net'},
  350. {'action': 'accept', 'specialtarget': 'dns'},
  351. {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
  352. {'action': 'drop', 'proto': 'icmp', 'icmptype': '128'},
  353. {'action': 'drop'},
  354. ]
  355. expected_nft = (
  356. 'flush chain ip6 qubes-firewall chain\n'
  357. 'table ip6 qubes-firewall {\n'
  358. ' chain chain {\n'
  359. ' ip6 nexthdr tcp ip6 daddr a::b/128 tcp dport 80 accept\n'
  360. ' ip6 nexthdr tcp ip6 daddr { 2001:67c:2e8:22::c100:68b/128 } '
  361. 'accept\n'
  362. ' ip6 daddr { 2001::1/128, 2001::2/128 } tcp dport 53 accept\n'
  363. ' ip6 daddr { 2001::1/128, 2001::2/128 } udp dport 53 accept\n'
  364. ' ip6 nexthdr udp ip6 daddr { 2001::1/128, 2001::2/128 } '
  365. 'udp dport 53 reject with icmpv6 type admin-prohibited\n'
  366. ' ip6 nexthdr icmpv6 icmpv6 type 128 reject with icmpv6 type '
  367. 'admin-prohibited\n'
  368. ' reject with icmpv6 type admin-prohibited\n'
  369. ' }\n'
  370. '}\n'
  371. )
  372. self.assertEqual(self.obj.prepare_rules('chain', rules, 6),
  373. expected_nft)
  374. def test_004_apply_rules4(self):
  375. rules = [{'action': 'accept'}]
  376. chain = 'qbs-10-137-0-1'
  377. self.obj.apply_rules('10.137.0.1', rules)
  378. self.assertEqual(self.obj.loaded_rules,
  379. [self.expected_create_chain('ip', '10.137.0.1', chain),
  380. self.obj.prepare_rules(chain, rules, 4),
  381. ])
  382. def test_005_apply_rules6(self):
  383. rules = [{'action': 'accept'}]
  384. chain = 'qbs-2000--a'
  385. self.obj.apply_rules('2000::a', rules)
  386. self.assertEqual(self.obj.loaded_rules,
  387. [self.expected_create_chain('ip6', '2000::a', chain),
  388. self.obj.prepare_rules(chain, rules, 6),
  389. ])
  390. def test_006_init(self):
  391. self.obj.init()
  392. self.assertEqual(self.obj.loaded_rules,
  393. [
  394. 'table ip qubes-firewall {\n'
  395. ' chain forward {\n'
  396. ' type filter hook forward priority 0;\n'
  397. ' policy drop;\n'
  398. ' ct state established,related accept\n'
  399. ' meta iifname != "vif*" accept\n'
  400. ' }\n'
  401. '}\n'
  402. 'table ip6 qubes-firewall {\n'
  403. ' chain forward {\n'
  404. ' type filter hook forward priority 0;\n'
  405. ' policy drop;\n'
  406. ' ct state established,related accept\n'
  407. ' meta iifname != "vif*" accept\n'
  408. ' }\n'
  409. '}\n'
  410. ])
  411. def test_007_cleanup(self):
  412. self.obj.init()
  413. self.obj.create_chain('1.2.3.4', 'chain-ip4-1', 4)
  414. self.obj.create_chain('1.2.3.6', 'chain-ip4-2', 4)
  415. self.obj.create_chain('2000::1', 'chain-ip6-1', 6)
  416. self.obj.create_chain('2000::2', 'chain-ip6-2', 6)
  417. # forget about commands called earlier
  418. self.obj.loaded_rules = []
  419. self.obj.cleanup()
  420. self.assertEqual(self.obj.loaded_rules,
  421. ['delete table ip qubes-firewall\n'
  422. 'delete table ip6 qubes-firewall\n',
  423. ])
  424. class TestFirewallWorker(TestCase):
  425. def setUp(self):
  426. self.obj = FirewallWorker()
  427. rules = {
  428. '10.137.0.1': {
  429. 'policy': 'accept',
  430. '0000': 'proto=tcp dstports=80-80 action=drop',
  431. '0001': 'proto=udp specialtarget=dns action=accept',
  432. '0002': 'proto=udp action=drop',
  433. },
  434. '10.137.0.2': {'policy': 'accept'},
  435. # no policy
  436. '10.137.0.3': {'0000': 'proto=tcp action=accept'},
  437. # no action
  438. '10.137.0.4': {
  439. 'policy': 'drop',
  440. '0000': 'proto=tcp'
  441. },
  442. }
  443. for addr, entries in rules.items():
  444. for key, value in entries.items():
  445. self.obj.qdb.entries[
  446. '/qubes-firewall/{}/{}'.format(addr, key)] = value
  447. self.subprocess_patch = patch('subprocess.call')
  448. self.subprocess_mock = self.subprocess_patch.start()
  449. def tearDown(self):
  450. self.subprocess_patch.stop()
  451. def test_read_rules(self):
  452. expected_rules1 = [
  453. {'proto': 'tcp', 'dstports': '80-80', 'action': 'drop'},
  454. {'proto': 'udp', 'specialtarget': 'dns', 'action': 'accept'},
  455. {'proto': 'udp', 'action': 'drop'},
  456. {'action': 'accept'},
  457. ]
  458. expected_rules2 = [
  459. {'action': 'accept'},
  460. ]
  461. self.assertEqual(self.obj.read_rules('10.137.0.1'), expected_rules1)
  462. self.assertEqual(self.obj.read_rules('10.137.0.2'), expected_rules2)
  463. with self.assertRaises(qubesagent.firewall.RuleParseError):
  464. self.obj.read_rules('10.137.0.3')
  465. with self.assertRaises(qubesagent.firewall.RuleParseError):
  466. self.obj.read_rules('10.137.0.4')
  467. def test_list_targets(self):
  468. self.assertEqual(self.obj.list_targets(), set(['10.137.0.{}'.format(x)
  469. for x in range(1, 5)]))
  470. def test_is_ip6(self):
  471. self.assertTrue(self.obj.is_ip6('2000::abcd'))
  472. self.assertTrue(self.obj.is_ip6('2000:1:2:3:4:5:6:abcd'))
  473. self.assertFalse(self.obj.is_ip6('10.137.0.1'))
  474. def test_handle_addr(self):
  475. self.obj.handle_addr('10.137.0.2')
  476. self.assertEqual(self.obj.rules['10.137.0.2'], [{'action': 'accept'}])
  477. # fallback to block all
  478. self.obj.handle_addr('10.137.0.3')
  479. self.assertEqual(self.obj.rules['10.137.0.3'], [{'action': 'drop'}])
  480. self.obj.handle_addr('10.137.0.4')
  481. self.assertEqual(self.obj.rules['10.137.0.4'], [{'action': 'drop'}])
  482. @patch('os.path.isfile')
  483. @patch('os.access')
  484. @patch('subprocess.call')
  485. def test_run_user_script(self, mock_subprocess, mock_os_access,
  486. mock_os_path_isfile):
  487. mock_os_path_isfile.return_value = False
  488. mock_os_access.return_value = False
  489. super(FirewallWorker, self.obj).run_user_script()
  490. self.assertFalse(mock_subprocess.called)
  491. mock_os_path_isfile.return_value = True
  492. mock_os_access.return_value = False
  493. super(FirewallWorker, self.obj).run_user_script()
  494. self.assertFalse(mock_subprocess.called)
  495. mock_os_path_isfile.return_value = True
  496. mock_os_access.return_value = True
  497. super(FirewallWorker, self.obj).run_user_script()
  498. mock_subprocess.assert_called_once_with(
  499. ['/rw/config/qubes-firewall-user-script'])
  500. def test_main(self):
  501. self.obj.main()
  502. self.assertTrue(self.obj.init_called)
  503. self.assertTrue(self.obj.cleanup_called)
  504. self.assertTrue(self.obj.user_script_called)
  505. self.assertEqual(set(self.obj.rules.keys()), self.obj.list_targets())
  506. # rules content were already tested