NAT: T4365: Fix for nat tables manipulation on netfilter

2 files changed, 68 insertions, 48 deletions

diff --git a/data/templates/firewall/nftables-nat.tmpl b/data/templates/firewall/nftables-nat.tmpl
index 922f3dcb4..7a925b264 100644
--- a/data/templates/firewall/nftables-nat.tmpl
+++ b/data/templates/firewall/nftables-nat.tmpl
@@ -138,8 +138,9 @@
 {%   endif %}
 {% endmacro %}
 
-# Start with clean NAT table
-flush table ip nat
+# Start with clean SNAT and DNAT chains
+flush chain ip nat PREROUTING
+flush chain ip nat POSTROUTING
 {% if helper_functions is vyos_defined('remove') %}
 {# NAT if going to be disabled - remove rules and targets from nftables #}
 {%   set base_command = 'delete rule ip raw' %}
@@ -164,6 +165,7 @@ add rule ip raw NAT_CONNTRACK counter accept
 #
 # Destination NAT rules build up here
 #
+add rule ip nat PREROUTING counter jump VYOS_PRE_DNAT_HOOK
 {% if destination.rule is vyos_defined %}
 {%   for rule, config in destination.rule.items() if config.disable is not vyos_defined %}
 {{ nat_rule(rule, config, 'PREROUTING') }}
@@ -172,6 +174,7 @@ add rule ip raw NAT_CONNTRACK counter accept
 #
 # Source NAT rules build up here
 #
+add rule ip nat POSTROUTING counter jump VYOS_PRE_SNAT_HOOK
 {% if source.rule is vyos_defined %}
 {%   for rule, config in source.rule.items() if config.disable is not vyos_defined %}
 {{ nat_rule(rule, config, 'POSTROUTING') }}
diff --git a/smoketest/scripts/cli/test_nat.py b/smoketest/scripts/cli/test_nat.py
index 2e1b8d431..1f8e4c307 100755
--- a/smoketest/scripts/cli/test_nat.py
+++ b/smoketest/scripts/cli/test_nat.py
@@ -59,36 +59,44 @@ class TestNAT(VyOSUnitTestSHIM.TestCase):
 
         self.cli_commit()
 
-        tmp = cmd('sudo nft -j list table nat')
+        tmp = cmd('sudo nft -j list chain ip nat POSTROUTING')
         data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp))
 
         for idx in range(0, len(data_json)):
-            rule = str(rules[idx])
             data = data_json[idx]
-            network = f'192.168.{rule}.0/24'
-
-            self.assertEqual(data['chain'], 'POSTROUTING')
-            self.assertEqual(data['comment'], f'SRC-NAT-{rule}')
-            self.assertEqual(data['family'], 'ip')
-            self.assertEqual(data['table'], 'nat')
+            if idx == 0:
+                self.assertEqual(data['chain'], 'POSTROUTING')
+                self.assertEqual(data['family'], 'ip')
+                self.assertEqual(data['table'], 'nat')
 
-            iface = dict_search('match.right', data['expr'][0])
-            direction = dict_search('match.left.payload.field', data['expr'][1])
-            address = dict_search('match.right.prefix.addr', data['expr'][1])
-            mask = dict_search('match.right.prefix.len', data['expr'][1])
-
-            if int(rule) < 200:
-                self.assertEqual(direction, 'saddr')
-                self.assertEqual(iface, outbound_iface_100)
-                # check for masquerade keyword
-                self.assertIn('masquerade', data['expr'][3])
+                jump_target = dict_search('jump.target', data['expr'][1])
+                self.assertEqual(jump_target,'VYOS_PRE_SNAT_HOOK')
             else:
-                self.assertEqual(direction, 'daddr')
-                self.assertEqual(iface, outbound_iface_200)
-                # check for return keyword due to 'exclude'
-                self.assertIn('return', data['expr'][3])
-
-            self.assertEqual(f'{address}/{mask}', network)
+                rule = str(rules[idx - 1])
+                network = f'192.168.{rule}.0/24'
+
+                self.assertEqual(data['chain'], 'POSTROUTING')
+                self.assertEqual(data['comment'], f'SRC-NAT-{rule}')
+                self.assertEqual(data['family'], 'ip')
+                self.assertEqual(data['table'], 'nat')
+
+                iface = dict_search('match.right', data['expr'][0])
+                direction = dict_search('match.left.payload.field', data['expr'][1])
+                address = dict_search('match.right.prefix.addr', data['expr'][1])
+                mask = dict_search('match.right.prefix.len', data['expr'][1])
+
+                if int(rule) < 200:
+                    self.assertEqual(direction, 'saddr')
+                    self.assertEqual(iface, outbound_iface_100)
+                    # check for masquerade keyword
+                    self.assertIn('masquerade', data['expr'][3])
+                else:
+                    self.assertEqual(direction, 'daddr')
+                    self.assertEqual(iface, outbound_iface_200)
+                    # check for return keyword due to 'exclude'
+                    self.assertIn('return', data['expr'][3])
+
+                self.assertEqual(f'{address}/{mask}', network)
 
     def test_dnat(self):
         rules = ['100', '110', '120', '130', '200', '210', '220', '230']
@@ -111,33 +119,42 @@ class TestNAT(VyOSUnitTestSHIM.TestCase):
 
         self.cli_commit()
 
-        tmp = cmd('sudo nft -j list table nat')
+        tmp = cmd('sudo nft -j list chain ip nat PREROUTING')
         data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp))
 
         for idx in range(0, len(data_json)):
-            rule = str(rules[idx])
             data = data_json[idx]
-            port = int(f'10{rule}')
-
-            self.assertEqual(data['chain'], 'PREROUTING')
-            self.assertEqual(data['comment'].split()[0], f'DST-NAT-{rule}')
-            self.assertEqual(data['family'], 'ip')
-            self.assertEqual(data['table'], 'nat')
-
-            iface = dict_search('match.right', data['expr'][0])
-            direction = dict_search('match.left.payload.field', data['expr'][1])
-            protocol = dict_search('match.left.payload.protocol', data['expr'][1])
-            dnat_addr = dict_search('dnat.addr', data['expr'][3])
-            dnat_port = dict_search('dnat.port', data['expr'][3])
-
-            self.assertEqual(direction, 'sport')
-            self.assertEqual(dnat_addr, '192.0.2.1')
-            self.assertEqual(dnat_port, port)
-            if int(rule) < 200:
-                self.assertEqual(iface, inbound_iface_100)
-                self.assertEqual(protocol, inbound_proto_100)
+            if idx == 0:
+                self.assertEqual(data['chain'], 'PREROUTING')
+                self.assertEqual(data['family'], 'ip')
+                self.assertEqual(data['table'], 'nat')
+
+                jump_target = dict_search('jump.target', data['expr'][1])
+                self.assertEqual(jump_target,'VYOS_PRE_DNAT_HOOK')
             else:
-                self.assertEqual(iface, inbound_iface_200)
+
+                rule = str(rules[idx - 1])
+                port = int(f'10{rule}')
+
+                self.assertEqual(data['chain'], 'PREROUTING')
+                self.assertEqual(data['comment'].split()[0], f'DST-NAT-{rule}')
+                self.assertEqual(data['family'], 'ip')
+                self.assertEqual(data['table'], 'nat')
+
+                iface = dict_search('match.right', data['expr'][0])
+                direction = dict_search('match.left.payload.field', data['expr'][1])
+                protocol = dict_search('match.left.payload.protocol', data['expr'][1])
+                dnat_addr = dict_search('dnat.addr', data['expr'][3])
+                dnat_port = dict_search('dnat.port', data['expr'][3])
+
+                self.assertEqual(direction, 'sport')
+                self.assertEqual(dnat_addr, '192.0.2.1')
+                self.assertEqual(dnat_port, port)
+                if int(rule) < 200:
+                    self.assertEqual(iface, inbound_iface_100)
+                    self.assertEqual(protocol, inbound_proto_100)
+                else:
+                    self.assertEqual(iface, inbound_iface_200)
 
     def test_snat_required_translation_address(self):
         # T2813: Ensure translation address is specified