tests: convert network tests to use nft

Use custom-* nft chains for tests-related rules. There were left few iptables invocation, but those should be independent of other rules (mostly logging, not filtering).
QubesOS · Apr 20, 2023 · 2b15065 · 2b15065
1 parent f714f57
commit 2b15065
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 6 deletions.
diff --git a/qubes/tests/integ/network.py b/qubes/tests/integ/network.py
@@ -110,6 +110,7 @@ def tearDown(self):
                 self._run_cmd_and_log_output(vm, 'iptables -vnL')
                 self._run_cmd_and_log_output(vm, 'iptables -vnL -t nat')
                 self._run_cmd_and_log_output(vm, 'nft list table qubes-firewall')
+                self._run_cmd_and_log_output(vm, 'nft list ruleset')
                 self._run_cmd_and_log_output(vm, 'systemctl --no-pager status qubes-firewall')
                 self._run_cmd_and_log_output(vm, 'systemctl --no-pager status qubes-iptables')
                 self._run_cmd_and_log_output(vm, 'systemctl --no-pager status xendriverdomain')
@@ -142,7 +143,7 @@ def run_netvm_cmd(cmd):
         run_netvm_cmd("ip link add test0 type dummy")
         run_netvm_cmd("ip link set test0 up")
         run_netvm_cmd("ip addr add {}/24 dev test0".format(self.test_ip))
-        run_netvm_cmd("iptables -I INPUT -d {} -j ACCEPT --wait".format(
+        run_netvm_cmd("nft add ip qubes custom-input ip daddr {} accept".format(
             self.test_ip))
         # ignore failure
         self.run_cmd(self.testnetvm, "while pkill dnsmasq; do sleep 1; done")
@@ -627,7 +628,7 @@ def test_203_fake_ip_inter_vm_allow(self):
         self.loop.run_until_complete(self.start_vm(self.testvm1))
         self.loop.run_until_complete(self.start_vm(self.testvm2))
 
-        cmd = 'iptables -I FORWARD -s {} -d {} -j ACCEPT'.format(
+        cmd = 'nft add ip qubes custom-forward ip saddr {} ip daddr {} accept'.format(
             self.testvm2.ip, self.testvm1.ip)
         try:
             self.loop.run_until_complete(self.proxy.run_for_stdio(
@@ -637,7 +638,8 @@ def test_203_fake_ip_inter_vm_allow(self):
                 '{} failed with: {}'.format(cmd, e.returncode)) from None
 
         try:
-            cmd = 'iptables -I INPUT -s {} -j ACCEPT'.format(self.testvm2.ip)
+            cmd = "nft add ip qubes custom-input ip saddr {} counter accept".format(
+                self.testvm2.ip))
             self.loop.run_until_complete(self.testvm1.run_for_stdio(
                 cmd, user='root'))
         except subprocess.CalledProcessError as e:
@@ -648,13 +650,13 @@ def test_203_fake_ip_inter_vm_allow(self):
             self.ping_cmd.format(target=self.testvm1.ip)), 0)
 
         try:
-            cmd = 'iptables -nvxL INPUT | grep {}'.format(self.testvm2.ip)
+            cmd = 'nft list chain ip qubes custom-input | grep {}'.format(self.testvm2.ip)
             (stdout, _) = self.loop.run_until_complete(
                 self.testvm1.run_for_stdio(cmd, user='root'))
         except subprocess.CalledProcessError as e:
             raise AssertionError(
                 '{} failed with {}'.format(cmd, e.returncode)) from None
-        self.assertNotEqual(stdout.decode().split()[0], '0',
+        self.assertNotEqual(stdout.decode().split()[-2], '0',
             'Packets didn\'t managed to the VM')
 
     def test_204_fake_ip_proxy(self):

diff --git a/qubes/tests/integ/network_ipv6.py b/qubes/tests/integ/network_ipv6.py
@@ -73,7 +73,7 @@ def run_netvm_cmd(cmd):
 
         run_netvm_cmd("ip addr add {}/128 dev test0".format(self.test_ip6))
         run_netvm_cmd(
-            "ip6tables -I INPUT -d {} -j ACCEPT".format(self.test_ip6))
+            "nft add ip6 qubes custom-input ip6 daddr {} accept".format(self.test_ip6))
         # ignore failure
         self.run_cmd(self.testnetvm, "while pkill dnsmasq; do sleep 1; done")
         run_netvm_cmd(