nat66: T4598: Add exclude options in nat66

4 files changed, 53 insertions, 34 deletions

diff --git a/data/templates/firewall/nftables-nat66.j2 b/data/templates/firewall/nftables-nat66.j2
index ca19506f2..2fe04b4ff 100644
--- a/data/templates/firewall/nftables-nat66.j2
+++ b/data/templates/firewall/nftables-nat66.j2
@@ -63,6 +63,10 @@
 {% if dest_address is vyos_defined %}
 {%     set output = output ~ ' ' ~ dest_address %}
 {% endif %}
+{% if config.exclude is vyos_defined %}
+{#     rule has been marked as 'exclude' thus we simply return here #}
+{%     set trns_address = 'return' %}
+{% endif %}
 {% if trns_address is vyos_defined %}
 {%     set output = output ~ ' ' ~ trns_address %}
 {% endif %}
diff --git a/interface-definitions/nat66.xml.in b/interface-definitions/nat66.xml.in
index ac3198f45..b50e11c49 100644
--- a/interface-definitions/nat66.xml.in
+++ b/interface-definitions/nat66.xml.in
@@ -35,6 +35,12 @@
                   <valueless/>
                 </properties>
               </leafNode>
+              <leafNode name="exclude">
+                <properties>
+                  <help>Exclude packets matching this rule from NAT</help>
+                  <valueless/>
+                </properties>
+              </leafNode>
               <leafNode name="log">
                 <properties>
                   <help>NAT66 rule logging</help>
@@ -162,6 +168,12 @@
                   <valueless/>
                 </properties>
               </leafNode>
+              <leafNode name="exclude">
+                <properties>
+                  <help>Exclude packets matching this rule from NAT</help>
+                  <valueless/>
+                </properties>
+              </leafNode>
               <leafNode name="log">
                 <properties>
                   <help>NAT66 rule logging</help>
diff --git a/smoketest/scripts/cli/test_nat66.py b/smoketest/scripts/cli/test_nat66.py
index aac6a30f9..4b5625569 100755
--- a/smoketest/scripts/cli/test_nat66.py
+++ b/smoketest/scripts/cli/test_nat66.py
@@ -42,6 +42,17 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
         self.cli_delete(base_path)
         self.cli_commit()
 
+    def verify_nftables(self, nftables_search, table, inverse=False, args=''):
+        nftables_output = cmd(f'sudo nft {args} list table {table}')
+
+        for search in nftables_search:
+            matched = False
+            for line in nftables_output.split("\n"):
+                if all(item in line for item in search):
+                    matched = True
+                    break
+            self.assertTrue(not matched if inverse else matched, msg=search)
+
     def test_source_nat66(self):
         source_prefix = 'fc00::/64'
         translation_prefix = 'fc01::/64'
@@ -49,29 +60,23 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
         self.cli_set(src_path + ['rule', '1', 'source', 'prefix', source_prefix])
         self.cli_set(src_path + ['rule', '1', 'translation', 'address', translation_prefix])
 
-        # check validate() - outbound-interface must be defined
-        self.cli_commit()
+        self.cli_set(src_path + ['rule', '2', 'outbound-interface', 'eth1'])
+        self.cli_set(src_path + ['rule', '2', 'source', 'prefix', source_prefix])
+        self.cli_set(src_path + ['rule', '2', 'translation', 'address', 'masquerade'])
 
-        tmp = cmd('sudo nft -j list table ip6 nat')
-        data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp))
+        self.cli_set(src_path + ['rule', '3', 'outbound-interface', 'eth1'])
+        self.cli_set(src_path + ['rule', '3', 'source', 'prefix', source_prefix])
+        self.cli_set(src_path + ['rule', '3', 'exclude'])
 
-        for idx in range(0, len(data_json)):
-            data = data_json[idx]
+        self.cli_commit()
 
-            self.assertEqual(data['chain'], 'POSTROUTING')
-            self.assertEqual(data['family'], 'ip6')
-            self.assertEqual(data['table'], 'nat')
+        nftables_search = [
+            ['oifname "eth1"', 'ip6 saddr fc00::/64', 'snat prefix to fc01::/64'],
+            ['oifname "eth1"', 'ip6 saddr fc00::/64', 'masquerade'],
+            ['oifname "eth1"', 'ip6 saddr fc00::/64', 'return']
+        ]
 
-            iface = dict_search('match.right', data['expr'][0])
-            address = dict_search('match.right.prefix.addr', data['expr'][2])
-            mask = dict_search('match.right.prefix.len', data['expr'][2])
-            translation_address = dict_search('snat.addr.prefix.addr', data['expr'][3])
-            translation_mask = dict_search('snat.addr.prefix.len', data['expr'][3])
-
-            self.assertEqual(iface, 'eth1')
-            # check for translation address
-            self.assertEqual(f'{translation_address}/{translation_mask}', translation_prefix)
-            self.assertEqual(f'{address}/{mask}', source_prefix)
+        self.verify_nftables(nftables_search, 'ip6 nat')
 
     def test_source_nat66_address(self):
         source_prefix = 'fc00::/64'
@@ -106,28 +111,25 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
     def test_destination_nat66(self):
         destination_address = 'fc00::1'
         translation_address = 'fc01::1'
+        source_address = 'fc02::1'
         self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'eth1'])
         self.cli_set(dst_path + ['rule', '1', 'destination', 'address', destination_address])
         self.cli_set(dst_path + ['rule', '1', 'translation', 'address', translation_address])
 
+        self.cli_set(dst_path + ['rule', '2', 'inbound-interface', 'eth1'])
+        self.cli_set(dst_path + ['rule', '2', 'destination', 'address', destination_address])
+        self.cli_set(dst_path + ['rule', '2', 'source', 'address', source_address])
+        self.cli_set(dst_path + ['rule', '2', 'exclude'])
+
         # check validate() - outbound-interface must be defined
         self.cli_commit()
 
-        tmp = cmd('sudo nft -j list table ip6 nat')
-        data_json = jmespath.search('nftables[?rule].rule[?chain]', json.loads(tmp))
-
-        for idx in range(0, len(data_json)):
-            data = data_json[idx]
-
-            self.assertEqual(data['chain'], 'PREROUTING')
-            self.assertEqual(data['family'], 'ip6')
-            self.assertEqual(data['table'], 'nat')
-
-            iface = dict_search('match.right', data['expr'][0])
-            dnat_addr = dict_search('dnat.addr', data['expr'][3])
+        nftables_search = [
+            ['iifname "eth1"', 'ip6 daddr fc00::1', 'dnat to fc01::1'],
+            ['iifname "eth1"', 'ip6 saddr fc02::1', 'ip6 daddr fc00::1', 'return']
+        ]
 
-            self.assertEqual(dnat_addr, translation_address)
-            self.assertEqual(iface, 'eth1')
+        self.verify_nftables(nftables_search, 'ip6 nat')
 
     def test_destination_nat66_prefix(self):
         destination_prefix = 'fc00::/64'
diff --git a/src/conf_mode/nat66.py b/src/conf_mode/nat66.py
index 0972151a0..f64102d88 100755
--- a/src/conf_mode/nat66.py
+++ b/src/conf_mode/nat66.py
@@ -125,7 +125,8 @@ def verify(nat):
                 if addr != 'masquerade' and not is_ipv6(addr):
                     raise ConfigError(f'IPv6 address {addr} is not a valid address')
             else:
-                raise ConfigError(f'{err_msg} translation address not specified')
+                if 'exclude' not in config:
+                    raise ConfigError(f'{err_msg} translation address not specified')
 
             prefix = dict_search('source.prefix', config)
             if prefix != None: