test_firewall.py 20 KB

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