tests: update for new firewall API

QubesOS/qubes-issues#1815

qubes/tests/firewall.py

@@ -53,6 +53,8 @@ class TestVM(object):
         self.dir_path = '/tmp'
         self.app = TestApp()
 
+    def fire_event(self, event):
+        pass
 
 # noinspection PyPep8Naming
 class TC_00_RuleChoice(qubes.tests.QubesTestCase):

qubes/tests/int/basic.py

@@ -31,9 +31,11 @@ import time
 import unittest
 
 import qubes
+import qubes.firewall
 import qubes.tests
 import qubes.vm.appvm
 import qubes.vm.qubesvm
+import qubes.vm.standalonevm
 import qubes.vm.templatevm
 
 import libvirt  # pylint: disable=import-error
@@ -85,12 +87,16 @@ class TC_01_Properties(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
         newname = self.make_vm_name('newname')
 
         self.assertEqual(self.vm.name, self.vmname)
-        self.vm.write_firewall_conf({'allow': False, 'allowDns': False})
+        self.vm.firewall.policy = 'drop'
+        self.vm.firewall.rules = [
+            qubes.firewall.Rule(None, action='accept', specialtarget='dns')
+        ]
+        self.vm.firewall.save()
         self.vm.autostart = True
         self.addCleanup(os.system,
                         'sudo systemctl -q disable qubes-vm@{}.service || :'.
                         format(self.vmname))
-        pre_rename_firewall = self.vm.get_firewall_conf()
+        pre_rename_firewall = self.vm.firewall.rules
 
         with self.assertNotRaises(
                 (OSError, libvirt.libvirtError, qubes.exc.QubesException)):
@@ -117,9 +123,10 @@ class TC_01_Properties(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
         self.assertFalse(os.path.exists(
             os.path.join(os.getenv("HOME"), ".local/share/applications",
                 self.vmname + "-firefox.desktop")))
-        self.assertEquals(pre_rename_firewall, self.vm.get_firewall_conf())
+        self.vm.firewall.load()
+        self.assertEquals(pre_rename_firewall, self.vm.firewall.rules)
         with self.assertNotRaises((qubes.exc.QubesException, OSError)):
-            self.vm.write_firewall_conf({'allow': False})
+            self.vm.firewall.save()
         self.assertTrue(self.vm.autostart)
         self.assertTrue(os.path.exists(
             '/etc/systemd/system/multi-user.target.wants/'
@@ -178,24 +185,19 @@ class TC_01_Properties(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
                           testvm2.include_in_backups)
         self.assertEquals(testvm1.default_user, testvm2.default_user)
         self.assertEquals(testvm1.features, testvm2.features)
-        # TODO
+        self.assertEquals(testvm1.firewall.rules,
-        # self.assertEquals(testvm1.get_firewall_conf(),
+                          testvm2.firewall.rules)
-        #                   testvm2.get_firewall_conf())
 
         # now some non-default values
         testvm1.netvm = None
         testvm1.label = 'orange'
         testvm1.memory = 512
-        firewall = testvm1.get_firewall_conf()
+        firewall = testvm1.firewall
-        firewall['allowDns'] = False
+        firewall.policy = 'drop'
-        firewall['allowYumProxy'] = False
+        firewall.rules = [
-        firewall['rules'] = [{'address': '1.2.3.4',
+            qubes.firewall.Rule(None, action='accept', dsthost='1.2.3.0/24',
-                              'netmask': 24,
+                proto='tcp', dstports=22)]
-                              'proto': 'tcp',
+        firewall.save()
-                              'portBegin': 22,
-                              'portEnd': 22,
-                              }]
-        testvm1.write_firewall_conf(firewall)
 
         testvm3 = self.app.add_new_vm(testvm1.__class__,
                                      name=self.make_vm_name("clone2"),
@@ -226,9 +228,8 @@ class TC_01_Properties(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
                           testvm3.include_in_backups)
         self.assertEquals(testvm1.default_user, testvm3.default_user)
         self.assertEquals(testvm1.features, testvm3.features)
-        # TODO
+        self.assertEquals(testvm1.firewall.rules,
-        # self.assertEquals(testvm1.get_firewall_conf(),
+                          testvm2.firewall.rules)
-        #                   testvm3.get_firewall_conf())
 
     def test_020_name_conflict_app(self):
         # TODO decide what exception should be here

qubes/tests/int/network.py

@@ -196,11 +196,8 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 
         # block all for first
 
-        self.testvm1.write_firewall_conf({
+        self.testvm1.firewall.policy = 'drop'
-            'allow': False,
+        self.testvm1.firewall.save()
-            'allowDns': False,
-            'allowIcmp': False,
-        })
         self.testvm1.start()
         self.assertTrue(self.proxy.is_running())
 
@@ -225,11 +222,10 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 
         # block all except ICMP
 
-        self.testvm1.write_firewall_conf({
+        self.testvm1.firewall.rules = [(
-            'allow': False,
+            qubes.firewall.Rule(None, action='accept', proto='icmp')
-            'allowDns': False,
+        )]
-            'allowIcmp': True,
+        self.testvm1.firewall.save()
-        })
         # Ugly hack b/c there is no feedback when the rules are actually applied
         time.sleep(3)
         self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0,
@@ -239,11 +235,11 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 
         # all TCP still blocked
 
-        self.testvm1.write_firewall_conf({
+        self.testvm1.firewall.rules = [
-            'allow': False,
+            qubes.firewall.Rule(None, action='accept', proto='icmp'),
-            'allowDns': True,
+            qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
-            'allowIcmp': True,
+        ]
-        })
+        self.testvm1.firewall.save()
         # Ugly hack b/c there is no feedback when the rules are actually applied
         time.sleep(3)
         self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0,
@@ -253,15 +249,13 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 
         # block all except target
 
-        self.testvm1.write_firewall_conf({
+        self.testvm1.firewall.policy = 'drop'
-            'allow': False,
+        self.testvm1.firewall.rules = [
-            'allowDns': True,
+            qubes.firewall.Rule(None, action='accept', dsthost=self.test_ip,
-            'allowIcmp': True,
+                proto='tcp', dstports=1234),
-            'rules': [{'address': self.test_ip,
+        ]
-                       'netmask': 32,
+        self.testvm1.firewall.save()
-                       'proto': 'tcp',
+
-                       'portBegin': 1234
-                      }] })
         # Ugly hack b/c there is no feedback when the rules are actually applied
         time.sleep(3)
         self.assertEqual(self.run_cmd(self.testvm1, nc_cmd), 0,
@@ -269,16 +263,13 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 
         # allow all except target
 
-        self.testvm1.write_firewall_conf({
+        self.testvm1.firewall.policy = 'accept'
-            'allow': True,
+        self.testvm1.firewall.rules = [
-            'allowDns': True,
+            qubes.firewall.Rule(None, action='drop', dsthost=self.test_ip,
-            'allowIcmp': True,
+                proto='tcp', dstports=1234),
-            'rules': [{'address': self.test_ip,
+        ]
-                       'netmask': 32,
+        self.testvm1.firewall.save()
-                       'proto': 'tcp',
+
-                       'portBegin': 1234
-                      }]
-        })
         # Ugly hack b/c there is no feedback when the rules are actually applied
         time.sleep(3)
         self.assertNotEqual(self.run_cmd(self.testvm1, nc_cmd), 0,