firewall.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585
  1. #
  2. # The Qubes OS Project, https://www.qubes-os.org/
  3. #
  4. # Copyright (C) 2016
  5. # Marek Marczykowski-Górecki <marmarek@invisiblethingslab.com>
  6. #
  7. # This library is free software; you can redistribute it and/or
  8. # modify it under the terms of the GNU Lesser General Public
  9. # License as published by the Free Software Foundation; either
  10. # version 2.1 of the License, or (at your option) any later version.
  11. #
  12. # This library is distributed in the hope that it will be useful,
  13. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  15. # Lesser General Public License for more details.
  16. #
  17. # You should have received a copy of the GNU Lesser General Public
  18. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  19. #
  20. import datetime
  21. import os
  22. import lxml.etree
  23. import unittest
  24. import qubes.firewall
  25. import qubes.tests
  26. class TestOption(qubes.firewall.RuleChoice):
  27. opt1 = 'opt1'
  28. opt2 = 'opt2'
  29. another = 'another'
  30. class TestVMM(object):
  31. def __init__(self):
  32. self.offline_mode = True
  33. class TestApp(object):
  34. def __init__(self):
  35. self.vmm = TestVMM()
  36. class TestVM(object):
  37. def __init__(self):
  38. self.firewall_conf = 'test-firewall.xml'
  39. self.dir_path = '/tmp'
  40. self.app = TestApp()
  41. def fire_event(self, event):
  42. pass
  43. # noinspection PyPep8Naming
  44. class TC_00_RuleChoice(qubes.tests.QubesTestCase):
  45. def test_000_accept_allowed(self):
  46. with self.assertNotRaises(ValueError):
  47. TestOption('opt1')
  48. TestOption('opt2')
  49. TestOption('another')
  50. def test_001_value_list(self):
  51. instance = TestOption('opt1')
  52. self.assertEqual(
  53. set(instance.allowed_values), {'opt1', 'opt2', 'another'})
  54. def test_010_reject_others(self):
  55. self.assertRaises(ValueError, lambda: TestOption('invalid'))
  56. class TC_01_Action(qubes.tests.QubesTestCase):
  57. def test_000_allowed_values(self):
  58. with self.assertNotRaises(ValueError):
  59. instance = qubes.firewall.Action('accept')
  60. self.assertEqual(
  61. set(instance.allowed_values), {'accept', 'drop'})
  62. def test_001_rule(self):
  63. instance = qubes.firewall.Action('accept')
  64. self.assertEqual(instance.rule, 'action=accept')
  65. self.assertEqual(instance.api_rule, 'action=accept')
  66. # noinspection PyPep8Naming
  67. class TC_02_Proto(qubes.tests.QubesTestCase):
  68. def test_000_allowed_values(self):
  69. with self.assertNotRaises(ValueError):
  70. instance = qubes.firewall.Proto('tcp')
  71. self.assertEqual(
  72. set(instance.allowed_values), {'tcp', 'udp', 'icmp'})
  73. def test_001_rule(self):
  74. instance = qubes.firewall.Proto('tcp')
  75. self.assertEqual(instance.rule, 'proto=tcp')
  76. self.assertEqual(instance.api_rule, 'proto=tcp')
  77. # noinspection PyPep8Naming
  78. class TC_02_DstHost(qubes.tests.QubesTestCase):
  79. def test_000_hostname(self):
  80. with self.assertNotRaises(ValueError):
  81. instance = qubes.firewall.DstHost('qubes-os.org')
  82. self.assertEqual(instance.type, 'dsthost')
  83. def test_001_ipv4(self):
  84. with self.assertNotRaises(ValueError):
  85. instance = qubes.firewall.DstHost('127.0.0.1')
  86. self.assertEqual(instance.type, 'dst4')
  87. self.assertEqual(instance.prefixlen, 32)
  88. self.assertEqual(str(instance), '127.0.0.1/32')
  89. self.assertEqual(instance.rule, 'dst4=127.0.0.1/32')
  90. def test_002_ipv4_prefixlen(self):
  91. with self.assertNotRaises(ValueError):
  92. instance = qubes.firewall.DstHost('127.0.0.0', 8)
  93. self.assertEqual(instance.type, 'dst4')
  94. self.assertEqual(instance.prefixlen, 8)
  95. self.assertEqual(str(instance), '127.0.0.0/8')
  96. self.assertEqual(instance.rule, 'dst4=127.0.0.0/8')
  97. def test_003_ipv4_parse_prefixlen(self):
  98. with self.assertNotRaises(ValueError):
  99. instance = qubes.firewall.DstHost('127.0.0.0/8')
  100. self.assertEqual(instance.type, 'dst4')
  101. self.assertEqual(instance.prefixlen, 8)
  102. self.assertEqual(str(instance), '127.0.0.0/8')
  103. self.assertEqual(instance.rule, 'dst4=127.0.0.0/8')
  104. def test_004_ipv4_invalid_prefix(self):
  105. with self.assertRaises(ValueError):
  106. qubes.firewall.DstHost('127.0.0.0/33')
  107. with self.assertRaises(ValueError):
  108. qubes.firewall.DstHost('127.0.0.0', 33)
  109. with self.assertRaises(ValueError):
  110. qubes.firewall.DstHost('127.0.0.0/-1')
  111. def test_005_ipv4_reject_shortened(self):
  112. # not strictly required, but ppl are used to it
  113. with self.assertRaises(ValueError):
  114. qubes.firewall.DstHost('127/8')
  115. def test_006_ipv4_invalid_addr(self):
  116. with self.assertRaises(ValueError):
  117. qubes.firewall.DstHost('137.327.0.0/16')
  118. with self.assertRaises(ValueError):
  119. qubes.firewall.DstHost('1.2.3.4.5/32')
  120. @unittest.expectedFailure
  121. def test_007_ipv4_invalid_network(self):
  122. with self.assertRaises(ValueError):
  123. qubes.firewall.DstHost('127.0.0.1/32')
  124. def test_010_ipv6(self):
  125. with self.assertNotRaises(ValueError):
  126. instance = qubes.firewall.DstHost('2001:abcd:efab::3')
  127. self.assertEqual(instance.type, 'dst6')
  128. self.assertEqual(instance.prefixlen, 128)
  129. self.assertEqual(str(instance), '2001:abcd:efab::3/128')
  130. self.assertEqual(instance.rule, 'dst6=2001:abcd:efab::3/128')
  131. self.assertEqual(instance.api_rule, 'dst6=2001:abcd:efab::3/128')
  132. def test_011_ipv6_prefixlen(self):
  133. with self.assertNotRaises(ValueError):
  134. instance = qubes.firewall.DstHost('2001:abcd:efab::', 64)
  135. self.assertEqual(instance.type, 'dst6')
  136. self.assertEqual(instance.prefixlen, 64)
  137. self.assertEqual(str(instance), '2001:abcd:efab::/64')
  138. self.assertEqual(instance.rule, 'dst6=2001:abcd:efab::/64')
  139. self.assertEqual(instance.api_rule, 'dst6=2001:abcd:efab::/64')
  140. def test_012_ipv6_parse_prefixlen(self):
  141. with self.assertNotRaises(ValueError):
  142. instance = qubes.firewall.DstHost('2001:abcd:efab::/64')
  143. self.assertEqual(instance.type, 'dst6')
  144. self.assertEqual(instance.prefixlen, 64)
  145. self.assertEqual(str(instance), '2001:abcd:efab::/64')
  146. self.assertEqual(instance.rule, 'dst6=2001:abcd:efab::/64')
  147. self.assertEqual(instance.api_rule, 'dst6=2001:abcd:efab::/64')
  148. def test_013_ipv6_invalid_prefix(self):
  149. with self.assertRaises(ValueError):
  150. qubes.firewall.DstHost('2001:abcd:efab::3/129')
  151. with self.assertRaises(ValueError):
  152. qubes.firewall.DstHost('2001:abcd:efab::3', 129)
  153. with self.assertRaises(ValueError):
  154. qubes.firewall.DstHost('2001:abcd:efab::3/-1')
  155. def test_014_ipv6_invalid_addr(self):
  156. with self.assertRaises(ValueError):
  157. qubes.firewall.DstHost('2001:abcd:efab0123::3/128')
  158. with self.assertRaises(ValueError):
  159. qubes.firewall.DstHost('2001:abcd:efab:3/128')
  160. with self.assertRaises(ValueError):
  161. qubes.firewall.DstHost('2001:abcd:efab:a:a:a:a:a:a:3/128')
  162. with self.assertRaises(ValueError):
  163. qubes.firewall.DstHost('2001:abcd:efgh::3/128')
  164. @unittest.expectedFailure
  165. def test_015_ipv6_invalid_network(self):
  166. with self.assertRaises(ValueError):
  167. qubes.firewall.DstHost('2001:abcd:efab::3/64')
  168. def test_020_invalid_hostname(self):
  169. with self.assertRaises(ValueError):
  170. qubes.firewall.DstHost('www qubes-os.org')
  171. with self.assertRaises(ValueError):
  172. qubes.firewall.DstHost('https://qubes-os.org')
  173. class TC_03_DstPorts(qubes.tests.QubesTestCase):
  174. def test_000_single_str(self):
  175. with self.assertNotRaises(ValueError):
  176. instance = qubes.firewall.DstPorts('80')
  177. self.assertEqual(str(instance), '80')
  178. self.assertEqual(instance.range, [80, 80])
  179. self.assertEqual(instance.rule, 'dstports=80-80')
  180. self.assertEqual(instance.api_rule, 'dstports=80-80')
  181. def test_001_single_int(self):
  182. with self.assertNotRaises(ValueError):
  183. instance = qubes.firewall.DstPorts(80)
  184. self.assertEqual(str(instance), '80')
  185. self.assertEqual(instance.range, [80, 80])
  186. self.assertEqual(instance.rule, 'dstports=80-80')
  187. self.assertEqual(instance.api_rule, 'dstports=80-80')
  188. def test_002_range(self):
  189. with self.assertNotRaises(ValueError):
  190. instance = qubes.firewall.DstPorts('80-90')
  191. self.assertEqual(str(instance), '80-90')
  192. self.assertEqual(instance.range, [80, 90])
  193. self.assertEqual(instance.rule, 'dstports=80-90')
  194. def test_003_invalid(self):
  195. with self.assertRaises(ValueError):
  196. qubes.firewall.DstPorts('80-90-100')
  197. with self.assertRaises(ValueError):
  198. qubes.firewall.DstPorts('abcdef')
  199. with self.assertRaises(ValueError):
  200. qubes.firewall.DstPorts('80 90')
  201. with self.assertRaises(ValueError):
  202. qubes.firewall.DstPorts('')
  203. def test_004_reversed_range(self):
  204. with self.assertRaises(ValueError):
  205. qubes.firewall.DstPorts('100-20')
  206. def test_005_out_of_range(self):
  207. with self.assertRaises(ValueError):
  208. qubes.firewall.DstPorts('1000000000000')
  209. with self.assertRaises(ValueError):
  210. qubes.firewall.DstPorts(1000000000000)
  211. with self.assertRaises(ValueError):
  212. qubes.firewall.DstPorts('1-1000000000000')
  213. class TC_04_IcmpType(qubes.tests.QubesTestCase):
  214. def test_000_number(self):
  215. with self.assertNotRaises(ValueError):
  216. instance = qubes.firewall.IcmpType(8)
  217. self.assertEqual(str(instance), '8')
  218. self.assertEqual(instance.rule, 'icmptype=8')
  219. def test_001_str(self):
  220. with self.assertNotRaises(ValueError):
  221. instance = qubes.firewall.IcmpType('8')
  222. self.assertEqual(str(instance), '8')
  223. self.assertEqual(instance.rule, 'icmptype=8')
  224. self.assertEqual(instance.api_rule, 'icmptype=8')
  225. def test_002_invalid(self):
  226. with self.assertRaises(ValueError):
  227. qubes.firewall.IcmpType(600)
  228. with self.assertRaises(ValueError):
  229. qubes.firewall.IcmpType(-1)
  230. with self.assertRaises(ValueError):
  231. qubes.firewall.IcmpType('abcde')
  232. with self.assertRaises(ValueError):
  233. qubes.firewall.IcmpType('')
  234. class TC_05_SpecialTarget(qubes.tests.QubesTestCase):
  235. def test_000_allowed_values(self):
  236. with self.assertNotRaises(ValueError):
  237. instance = qubes.firewall.SpecialTarget('dns')
  238. self.assertEqual(
  239. set(instance.allowed_values), {'dns'})
  240. def test_001_rule(self):
  241. instance = qubes.firewall.SpecialTarget('dns')
  242. self.assertEqual(instance.rule, 'specialtarget=dns')
  243. self.assertEqual(instance.api_rule, 'specialtarget=dns')
  244. class TC_06_Expire(qubes.tests.QubesTestCase):
  245. def test_000_number(self):
  246. with self.assertNotRaises(ValueError):
  247. instance = qubes.firewall.Expire(1463292452)
  248. self.assertEqual(str(instance), '1463292452')
  249. self.assertEqual(instance.api_rule, 'expire=1463292452')
  250. self.assertEqual(instance.datetime,
  251. datetime.datetime(2016, 5, 15, 6, 7, 32))
  252. self.assertIsNone(instance.rule)
  253. def test_001_str(self):
  254. with self.assertNotRaises(ValueError):
  255. instance = qubes.firewall.Expire('1463292452')
  256. self.assertEqual(str(instance), '1463292452')
  257. self.assertEqual(instance.datetime,
  258. datetime.datetime(2016, 5, 15, 6, 7, 32))
  259. self.assertIsNone(instance.rule)
  260. def test_002_invalid(self):
  261. with self.assertRaises(ValueError):
  262. qubes.firewall.Expire('abcdef')
  263. with self.assertRaises(ValueError):
  264. qubes.firewall.Expire('')
  265. def test_003_expired(self):
  266. with self.assertNotRaises(ValueError):
  267. instance = qubes.firewall.Expire('1463292452')
  268. self.assertTrue(instance.expired)
  269. with self.assertNotRaises(ValueError):
  270. instance = qubes.firewall.Expire('1583292452')
  271. self.assertFalse(instance.expired)
  272. class TC_07_Comment(qubes.tests.QubesTestCase):
  273. def test_000_str(self):
  274. with self.assertNotRaises(ValueError):
  275. instance = qubes.firewall.Comment('Some comment')
  276. self.assertEqual(str(instance), 'Some comment')
  277. self.assertEqual(instance.api_rule, 'comment=Some comment')
  278. self.assertIsNone(instance.rule)
  279. class TC_08_Rule(qubes.tests.QubesTestCase):
  280. def test_000_simple(self):
  281. with self.assertNotRaises(ValueError):
  282. rule = qubes.firewall.Rule(None, action='accept', proto='icmp')
  283. self.assertEqual(rule.rule, 'action=accept proto=icmp')
  284. self.assertIsNone(rule.dsthost)
  285. self.assertIsNone(rule.dstports)
  286. self.assertIsNone(rule.icmptype)
  287. self.assertIsNone(rule.comment)
  288. self.assertIsNone(rule.expire)
  289. self.assertEqual(str(rule.action), 'accept')
  290. self.assertEqual(str(rule.proto), 'icmp')
  291. def test_001_expire(self):
  292. with self.assertNotRaises(ValueError):
  293. rule = qubes.firewall.Rule(None, action='accept', proto='icmp',
  294. expire='1463292452')
  295. self.assertIsNone(rule.rule)
  296. with self.assertNotRaises(ValueError):
  297. rule = qubes.firewall.Rule(None, action='accept', proto='icmp',
  298. expire='1663292452')
  299. self.assertIsNotNone(rule.rule)
  300. def test_002_dstports(self):
  301. with self.assertNotRaises(ValueError):
  302. rule = qubes.firewall.Rule(None, action='accept', proto='tcp',
  303. dstports=80)
  304. self.assertEqual(str(rule.dstports), '80')
  305. with self.assertNotRaises(ValueError):
  306. rule = qubes.firewall.Rule(None, action='accept', proto='udp',
  307. dstports=80)
  308. self.assertEqual(str(rule.dstports), '80')
  309. def test_003_reject_invalid(self):
  310. with self.assertRaises((ValueError, AssertionError)):
  311. # missing action
  312. qubes.firewall.Rule(None, proto='icmp')
  313. with self.assertRaises(ValueError):
  314. # not proto=tcp or proto=udp for dstports
  315. qubes.firewall.Rule(None, action='accept', proto='icmp',
  316. dstports=80)
  317. with self.assertRaises(ValueError):
  318. # not proto=tcp or proto=udp for dstports
  319. qubes.firewall.Rule(None, action='accept', dstports=80)
  320. with self.assertRaises(ValueError):
  321. # not proto=icmp for icmptype
  322. qubes.firewall.Rule(None, action='accept', proto='tcp',
  323. icmptype=8)
  324. with self.assertRaises(ValueError):
  325. # not proto=icmp for icmptype
  326. qubes.firewall.Rule(None, action='accept', icmptype=8)
  327. def test_004_proto_change(self):
  328. rule = qubes.firewall.Rule(None, action='accept', proto='tcp')
  329. with self.assertNotRaises(ValueError):
  330. rule.proto = 'udp'
  331. self.assertEqual(rule.rule, 'action=accept proto=udp')
  332. rule = qubes.firewall.Rule(None, action='accept', proto='tcp',
  333. dstports=80)
  334. with self.assertNotRaises(ValueError):
  335. rule.proto = 'udp'
  336. self.assertEqual(rule.rule, 'action=accept proto=udp dstports=80-80')
  337. rule = qubes.firewall.Rule(None, action='accept')
  338. with self.assertNotRaises(ValueError):
  339. rule.proto = 'udp'
  340. self.assertEqual(rule.rule, 'action=accept proto=udp')
  341. with self.assertNotRaises(ValueError):
  342. rule.dstports = 80
  343. self.assertEqual(rule.rule, 'action=accept proto=udp dstports=80-80')
  344. with self.assertNotRaises(ValueError):
  345. rule.proto = 'icmp'
  346. self.assertEqual(rule.rule, 'action=accept proto=icmp')
  347. self.assertIsNone(rule.dstports)
  348. rule.icmptype = 8
  349. self.assertEqual(rule.rule, 'action=accept proto=icmp icmptype=8')
  350. with self.assertNotRaises(ValueError):
  351. rule.proto = qubes.property.DEFAULT
  352. self.assertEqual(rule.rule, 'action=accept')
  353. self.assertIsNone(rule.dstports)
  354. def test_005_from_xml_v1(self):
  355. xml_txt = \
  356. '<rule address="192.168.0.0" proto="tcp" netmask="24" port="443"/>'
  357. with self.assertNotRaises(ValueError):
  358. rule = qubes.firewall.Rule.from_xml_v1(
  359. lxml.etree.fromstring(xml_txt), 'accept')
  360. self.assertEqual(rule.dsthost, '192.168.0.0/24')
  361. self.assertEqual(rule.proto, 'tcp')
  362. self.assertEqual(rule.dstports, '443')
  363. self.assertIsNone(rule.expire)
  364. self.assertIsNone(rule.comment)
  365. def test_006_from_xml_v1(self):
  366. xml_txt = \
  367. '<rule address="qubes-os.org" proto="tcp" ' \
  368. 'port="443" toport="1024"/>'
  369. with self.assertNotRaises(ValueError):
  370. rule = qubes.firewall.Rule.from_xml_v1(
  371. lxml.etree.fromstring(xml_txt), 'drop')
  372. self.assertEqual(rule.dsthost, 'qubes-os.org')
  373. self.assertEqual(rule.proto, 'tcp')
  374. self.assertEqual(rule.dstports, '443-1024')
  375. self.assertEqual(rule.action, 'drop')
  376. self.assertIsNone(rule.expire)
  377. self.assertIsNone(rule.comment)
  378. def test_007_from_xml_v1(self):
  379. xml_txt = \
  380. '<rule address="192.168.0.0" netmask="24" expire="1463292452"/>'
  381. with self.assertNotRaises(ValueError):
  382. rule = qubes.firewall.Rule.from_xml_v1(
  383. lxml.etree.fromstring(xml_txt), 'accept')
  384. self.assertEqual(rule.dsthost, '192.168.0.0/24')
  385. self.assertEqual(rule.expire, '1463292452')
  386. self.assertEqual(rule.action, 'accept')
  387. self.assertIsNone(rule.proto)
  388. self.assertIsNone(rule.dstports)
  389. def test_008_from_api_string(self):
  390. rule_txt = 'action=drop proto=tcp dstports=80-80'
  391. with self.assertNotRaises(ValueError):
  392. rule = qubes.firewall.Rule.from_api_string(
  393. rule_txt)
  394. self.assertEqual(rule.dstports.range, [80, 80])
  395. self.assertEqual(rule.proto, 'tcp')
  396. self.assertEqual(rule.action, 'drop')
  397. self.assertIsNone(rule.dsthost)
  398. self.assertIsNone(rule.expire)
  399. self.assertIsNone(rule.comment)
  400. self.assertEqual(rule.api_rule, rule_txt)
  401. def test_009_from_api_string(self):
  402. rule_txt = 'action=accept expire=2063292452 proto=tcp ' \
  403. 'comment=Some comment, with spaces'
  404. with self.assertNotRaises(ValueError):
  405. rule = qubes.firewall.Rule.from_api_string(
  406. rule_txt)
  407. self.assertEqual(rule.comment, 'Some comment, with spaces')
  408. self.assertEqual(rule.proto, 'tcp')
  409. self.assertEqual(rule.action, 'accept')
  410. self.assertEqual(rule.expire, '2063292452')
  411. self.assertIsNone(rule.dstports)
  412. self.assertIsNone(rule.dsthost)
  413. self.assertEqual(rule.api_rule, rule_txt)
  414. class TC_10_Firewall(qubes.tests.QubesTestCase):
  415. def setUp(self):
  416. super(TC_10_Firewall, self).setUp()
  417. self.vm = TestVM()
  418. firewall_path = os.path.join('/tmp', self.vm.firewall_conf)
  419. if os.path.exists(firewall_path):
  420. os.unlink(firewall_path)
  421. def tearDown(self):
  422. firewall_path = os.path.join('/tmp', self.vm.firewall_conf)
  423. if os.path.exists(firewall_path):
  424. os.unlink(firewall_path)
  425. return super(TC_10_Firewall, self).tearDown()
  426. def test_000_defaults(self):
  427. fw = qubes.firewall.Firewall(self.vm, False)
  428. fw.load_defaults()
  429. self.assertEqual(fw.policy, 'drop')
  430. self.assertEqual(fw.rules, [qubes.firewall.Rule(None, action='accept')])
  431. def test_001_save_load_empty(self):
  432. fw = qubes.firewall.Firewall(self.vm, True)
  433. self.assertEqual(fw.policy, 'drop')
  434. self.assertEqual(fw.rules, [qubes.firewall.Rule(None, action='accept')])
  435. fw.save()
  436. fw.load()
  437. self.assertEqual(fw.policy, 'drop')
  438. self.assertEqual(fw.rules, [qubes.firewall.Rule(None, action='accept')])
  439. def test_002_save_load_rules(self):
  440. fw = qubes.firewall.Firewall(self.vm, True)
  441. rules = [
  442. qubes.firewall.Rule(None, action='drop', proto='icmp'),
  443. qubes.firewall.Rule(None, action='drop', proto='tcp', dstports=80),
  444. qubes.firewall.Rule(None, action='accept', proto='udp',
  445. dstports=67),
  446. qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
  447. ]
  448. fw.rules.extend(rules)
  449. fw.save()
  450. self.assertTrue(os.path.exists(os.path.join(
  451. self.vm.dir_path, self.vm.firewall_conf)))
  452. fw = qubes.firewall.Firewall(TestVM(), True)
  453. self.assertEqual(fw.policy, qubes.firewall.Action.drop)
  454. self.assertEqual(fw.rules,
  455. [qubes.firewall.Rule(None, action='accept')] + rules)
  456. def test_003_load_v1(self):
  457. xml_txt = """<QubesFirewallRules dns="allow" icmp="allow"
  458. policy="deny" yumProxy="allow">
  459. <rule address="192.168.0.0" proto="tcp" netmask="24" port="80"/>
  460. <rule address="qubes-os.org" proto="tcp" port="443"/>
  461. </QubesFirewallRules>
  462. """
  463. with open(os.path.join('/tmp', self.vm.firewall_conf), 'w') as f:
  464. f.write(xml_txt)
  465. with self.assertNotRaises(ValueError):
  466. fw = qubes.firewall.Firewall(self.vm)
  467. self.assertEqual(str(fw.policy), 'drop')
  468. rules = [
  469. qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
  470. qubes.firewall.Rule(None, action='accept', proto='icmp'),
  471. qubes.firewall.Rule(None, action='accept', proto='tcp',
  472. dsthost='192.168.0.0/24', dstports='80'),
  473. qubes.firewall.Rule(None, action='accept', proto='tcp',
  474. dsthost='qubes-os.org', dstports='443')
  475. ]
  476. self.assertEqual(fw.rules, rules)
  477. def test_004_save_skip_expired(self):
  478. fw = qubes.firewall.Firewall(self.vm, True)
  479. rules = [
  480. qubes.firewall.Rule(None, action='drop', proto='icmp'),
  481. qubes.firewall.Rule(None, action='drop', proto='tcp', dstports=80),
  482. qubes.firewall.Rule(None, action='accept', proto='udp',
  483. dstports=67, expire=1373300257),
  484. qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
  485. ]
  486. fw.rules = rules
  487. fw.save()
  488. rules.pop(2)
  489. fw = qubes.firewall.Firewall(self.vm, True)
  490. self.assertEqual(fw.rules, rules)
  491. def test_005_qdb_entries(self):
  492. fw = qubes.firewall.Firewall(self.vm, True)
  493. rules = [
  494. qubes.firewall.Rule(None, action='drop', proto='icmp'),
  495. qubes.firewall.Rule(None, action='drop', proto='tcp', dstports=80),
  496. qubes.firewall.Rule(None, action='accept', proto='udp'),
  497. qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
  498. ]
  499. fw.rules = rules
  500. expected_qdb_entries = {
  501. 'policy': 'drop',
  502. '0000': 'action=drop proto=icmp',
  503. '0001': 'action=drop proto=tcp dstports=80-80',
  504. '0002': 'action=accept proto=udp',
  505. '0003': 'action=accept specialtarget=dns',
  506. }
  507. self.assertEqual(fw.qdb_entries(), expected_qdb_entries)