qvm_firewall.py 8.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. # encoding=utf-8
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2016 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 Lesser General Public License as published by
  10. # the Free Software Foundation; either version 2.1 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 Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public License along
  19. # with this program; if not, write to the Free Software Foundation, Inc.,
  20. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  21. #
  22. import argparse
  23. import qubesadmin.firewall
  24. import qubesadmin.tests
  25. import qubesadmin.tests.tools
  26. import qubesadmin.tools.qvm_firewall
  27. class TC_00_RuleAction(qubesadmin.tests.QubesTestCase):
  28. def setUp(self):
  29. super(TC_00_RuleAction, self).setUp()
  30. self.action = qubesadmin.tools.qvm_firewall.RuleAction(
  31. None, dest='rule')
  32. def test_000_named_opts(self):
  33. ns = argparse.Namespace()
  34. self.action(None, ns, ['dsthost=127.0.0.1', 'action=accept'])
  35. self.assertEqual(ns.rule,
  36. qubesadmin.firewall.Rule(
  37. None, action='accept', dsthost='127.0.0.1/32'))
  38. def test_001_unnamed_opts(self):
  39. ns = argparse.Namespace()
  40. self.action(None, ns, ['accept', '127.0.0.1', 'tcp', '80'])
  41. self.assertEqual(ns.rule,
  42. qubesadmin.firewall.Rule(
  43. None, action='accept', dsthost='127.0.0.1/32',
  44. proto='tcp', dstports=80))
  45. def test_002_unnamed_opts(self):
  46. ns = argparse.Namespace()
  47. self.action(None, ns, ['accept', '127.0.0.1', 'icmp', '8'])
  48. self.assertEqual(ns.rule,
  49. qubesadmin.firewall.Rule(
  50. None, action='accept', dsthost='127.0.0.1/32',
  51. proto='icmp', icmptype=8))
  52. def test_003_mixed_opts(self):
  53. ns = argparse.Namespace()
  54. self.action(None, ns, ['dsthost=127.0.0.1', 'accept',
  55. 'dstports=443', 'tcp'])
  56. self.assertEqual(ns.rule,
  57. qubesadmin.firewall.Rule(
  58. None, action='accept', dsthost='127.0.0.1/32',
  59. proto='tcp', dstports=443))
  60. class TC_10_qvm_firewall(qubesadmin.tests.QubesTestCase):
  61. def setUp(self):
  62. super(TC_10_qvm_firewall, self).setUp()
  63. self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
  64. b'0\0test-vm class=AppVM state=Halted\n'
  65. def test_000_list(self):
  66. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Get',
  67. None, None)] = \
  68. b'0\0action=accept dsthost=qubes-os.org\n' \
  69. b'action=drop proto=icmp\n'
  70. with qubesadmin.tests.tools.StdoutBuffer() as stdout:
  71. qubesadmin.tools.qvm_firewall.main(['test-vm', 'list'], app=self.app)
  72. self.assertEqual(
  73. [l.strip() for l in stdout.getvalue().splitlines()],
  74. ['NO ACTION HOST PROTOCOL PORT(S) SPECIAL '
  75. 'TARGET ICMP TYPE COMMENT',
  76. '0 accept qubes-os.org - - - '
  77. ' - -',
  78. '1 drop - icmp - - '
  79. ' - -',
  80. ])
  81. def test_001_list(self):
  82. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Get',
  83. None, None)] = \
  84. b'0\0action=accept dsthost=qubes-os.org proto=tcp ' \
  85. b'dstports=443-443\n' \
  86. b'action=drop proto=icmp icmptype=8\n' \
  87. b'action=accept specialtarget=dns comment=Allow DNS\n'
  88. with qubesadmin.tests.tools.StdoutBuffer() as stdout:
  89. qubesadmin.tools.qvm_firewall.main(['test-vm', 'list'], app=self.app)
  90. self.assertEqual(
  91. [l.strip() for l in stdout.getvalue().splitlines()],
  92. ['NO ACTION HOST PROTOCOL PORT(S) SPECIAL '
  93. 'TARGET ICMP TYPE COMMENT',
  94. '0 accept qubes-os.org tcp 443 - '
  95. ' - -',
  96. '1 drop - icmp - - '
  97. ' 8 -',
  98. '2 accept - - - dns '
  99. ' - Allow DNS',
  100. ])
  101. def test_002_list_raw(self):
  102. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Get',
  103. None, None)] = \
  104. b'0\0action=accept dsthost=qubes-os.org\n' \
  105. b'action=drop proto=icmp\n'
  106. with qubesadmin.tests.tools.StdoutBuffer() as stdout:
  107. qubesadmin.tools.qvm_firewall.main(['test-vm', '--raw', 'list'],
  108. app=self.app)
  109. self.assertEqual(
  110. [l.strip() for l in stdout.getvalue().splitlines()],
  111. ['action=accept dsthost=qubes-os.org',
  112. 'action=drop proto=icmp',
  113. ])
  114. def test_003_list_raw_reload(self):
  115. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Get',
  116. None, None)] = \
  117. b'0\0action=accept dsthost=qubes-os.org\n' \
  118. b'action=drop proto=icmp\n'
  119. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Reload',
  120. None, None)] = b'0\0'
  121. with qubesadmin.tests.tools.StdoutBuffer() as stdout:
  122. qubesadmin.tools.qvm_firewall.main(
  123. ['test-vm', '--raw', '--reload', 'list'],
  124. app=self.app)
  125. self.assertEqual(
  126. [l.strip() for l in stdout.getvalue().splitlines()],
  127. ['action=accept dsthost=qubes-os.org',
  128. 'action=drop proto=icmp',
  129. ])
  130. def test_010_add_after(self):
  131. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Get',
  132. None, None)] = \
  133. b'0\0action=accept dsthost=qubes-os.org\n' \
  134. b'action=drop proto=icmp\n'
  135. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Set', None,
  136. b'action=accept dsthost=qubes-os.org\n'
  137. b'action=drop proto=icmp\n'
  138. b'action=accept dst4=192.168.0.0/24 comment=Allow LAN\n')] = \
  139. b'0\0'
  140. qubesadmin.tools.qvm_firewall.main(
  141. ['test-vm', 'add', 'accept', '192.168.0.0/24', 'comment=Allow LAN'],
  142. app=self.app
  143. )
  144. def test_011_add_before(self):
  145. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Get',
  146. None, None)] = \
  147. b'0\0action=accept dsthost=qubes-os.org\n' \
  148. b'action=drop proto=icmp\n'
  149. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Set', None,
  150. b'action=accept dsthost=qubes-os.org\n'
  151. b'action=accept dst4=192.168.0.0/24 comment=Allow LAN\n'
  152. b'action=drop proto=icmp\n')] = b'0\0'
  153. qubesadmin.tools.qvm_firewall.main(
  154. ['test-vm', 'add', '--before', '1', 'accept', '192.168.0.0/24',
  155. 'comment=Allow LAN'],
  156. app=self.app
  157. )
  158. def test_020_del_number(self):
  159. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Get',
  160. None, None)] = \
  161. b'0\0action=accept dsthost=qubes-os.org\n' \
  162. b'action=drop proto=icmp\n'
  163. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Set', None,
  164. b'action=accept dsthost=qubes-os.org\n')] = b'0\0'
  165. qubesadmin.tools.qvm_firewall.main(
  166. ['test-vm', 'del', '--rule-no', '1'],
  167. app=self.app
  168. )
  169. def test_021_del_rule(self):
  170. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Get',
  171. None, None)] = \
  172. b'0\0action=accept dsthost=qubes-os.org\n' \
  173. b'action=drop proto=icmp\n'
  174. self.app.expected_calls[('test-vm', 'admin.vm.firewall.Set', None,
  175. b'action=accept dsthost=qubes-os.org\n')] = b'0\0'
  176. qubesadmin.tools.qvm_firewall.main(
  177. ['test-vm', 'del', 'drop', 'proto=icmp'],
  178. app=self.app
  179. )