From 448d4f6db9cf6dfceffccf988301e5f4d04c9afa Mon Sep 17 00:00:00 2001
From: sarthurdev <965089+sarthurdev@users.noreply.github.com>
Date: Tue, 20 Sep 2022 14:19:23 +0200
Subject: nat: T4605: Refactor NAT to use python module for parsing rules

* Rename table to vyos_nat
* Refactor tests to use `verify_nftables` format
---
 smoketest/scripts/cli/test_load_balancning_wan.py |   1 +
 smoketest/scripts/cli/test_nat.py                 | 141 ++++++++++------------
 2 files changed, 68 insertions(+), 74 deletions(-)

(limited to 'smoketest')

diff --git a/smoketest/scripts/cli/test_load_balancning_wan.py b/smoketest/scripts/cli/test_load_balancning_wan.py
index 303dece86..23020b9b1 100755
--- a/smoketest/scripts/cli/test_load_balancning_wan.py
+++ b/smoketest/scripts/cli/test_load_balancning_wan.py
@@ -177,6 +177,7 @@ class TestLoadBalancingWan(VyOSUnitTestSHIM.TestCase):
 }"""
         nat_vyos_pre_snat_hook = """table ip nat {
 	chain VYOS_PRE_SNAT_HOOK {
+		type nat hook postrouting priority srcnat - 1; policy accept;
 		counter jump WANLOADBALANCE
 		return
 	}
diff --git a/smoketest/scripts/cli/test_nat.py b/smoketest/scripts/cli/test_nat.py
index 408facfb3..5863409be 100755
--- a/smoketest/scripts/cli/test_nat.py
+++ b/smoketest/scripts/cli/test_nat.py
@@ -26,6 +26,7 @@ from vyos.util import dict_search
 base_path = ['nat']
 src_path = base_path + ['source']
 dst_path = base_path + ['destination']
+static_path = base_path + ['static']
 
 class TestNAT(VyOSUnitTestSHIM.TestCase):
     @classmethod
@@ -40,10 +41,24 @@ class TestNAT(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_snat(self):
         rules = ['100', '110', '120', '130', '200', '210', '220', '230']
         outbound_iface_100 = 'eth0'
         outbound_iface_200 = 'eth1'
+
+        nftables_search = ['jump VYOS_PRE_SNAT_HOOK']
+
         for rule in rules:
             network = f'192.168.{rule}.0/24'
             # depending of rule order we check either for source address for NAT
@@ -52,51 +67,16 @@ class TestNAT(VyOSUnitTestSHIM.TestCase):
                 self.cli_set(src_path + ['rule', rule, 'source', 'address', network])
                 self.cli_set(src_path + ['rule', rule, 'outbound-interface', outbound_iface_100])
                 self.cli_set(src_path + ['rule', rule, 'translation', 'address', 'masquerade'])
+                nftables_search.append([f'saddr {network}', f'oifname "{outbound_iface_100}"', 'masquerade'])
             else:
                 self.cli_set(src_path + ['rule', rule, 'destination', 'address', network])
                 self.cli_set(src_path + ['rule', rule, 'outbound-interface', outbound_iface_200])
                 self.cli_set(src_path + ['rule', rule, 'exclude'])
+                nftables_search.append([f'daddr {network}', f'oifname "{outbound_iface_200}"', 'return'])
 
         self.cli_commit()
 
-        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)):
-            data = data_json[idx]
-            if idx == 0:
-                self.assertEqual(data['chain'], 'POSTROUTING')
-                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_SNAT_HOOK')
-            else:
-                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)
+        self.verify_nftables(nftables_search, 'ip vyos_nat')
 
     def test_dnat(self):
         rules = ['100', '110', '120', '130', '200', '210', '220', '230']
@@ -105,56 +85,29 @@ class TestNAT(VyOSUnitTestSHIM.TestCase):
         inbound_proto_100 = 'udp'
         inbound_proto_200 = 'tcp'
 
+        nftables_search = ['jump VYOS_PRE_DNAT_HOOK']
+
         for rule in rules:
             port = f'10{rule}'
             self.cli_set(dst_path + ['rule', rule, 'source', 'port', port])
             self.cli_set(dst_path + ['rule', rule, 'translation', 'address', '192.0.2.1'])
             self.cli_set(dst_path + ['rule', rule, 'translation', 'port', port])
+            rule_search = [f'dnat to 192.0.2.1:{port}']
             if int(rule) < 200:
                 self.cli_set(dst_path + ['rule', rule, 'protocol', inbound_proto_100])
                 self.cli_set(dst_path + ['rule', rule, 'inbound-interface', inbound_iface_100])
+                rule_search.append(f'{inbound_proto_100} sport {port}')
+                rule_search.append(f'iifname "{inbound_iface_100}"')
             else:
                 self.cli_set(dst_path + ['rule', rule, 'protocol', inbound_proto_200])
                 self.cli_set(dst_path + ['rule', rule, 'inbound-interface', inbound_iface_200])
+                rule_search.append(f'iifname "{inbound_iface_200}"')
 
-        self.cli_commit()
-
-        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)):
-            data = data_json[idx]
-            if idx == 0:
-                self.assertEqual(data['chain'], 'PREROUTING')
-                self.assertEqual(data['family'], 'ip')
-                self.assertEqual(data['table'], 'nat')
+            nftables_search.append(rule_search)
 
-                jump_target = dict_search('jump.target', data['expr'][1])
-                self.assertEqual(jump_target,'VYOS_PRE_DNAT_HOOK')
-            else:
+        self.cli_commit()
 
-                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)
+        self.verify_nftables(nftables_search, 'ip vyos_nat')
 
     def test_snat_required_translation_address(self):
         # T2813: Ensure translation address is specified
@@ -193,8 +146,48 @@ class TestNAT(VyOSUnitTestSHIM.TestCase):
         # without any rule
         self.cli_set(src_path)
         self.cli_set(dst_path)
+        self.cli_set(static_path)
+        self.cli_commit()
+
+    def test_dnat_without_translation_address(self):
+        self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'eth1'])
+        self.cli_set(dst_path + ['rule', '1', 'destination', 'port', '443'])
+        self.cli_set(dst_path + ['rule', '1', 'protocol', 'tcp'])
+        self.cli_set(dst_path + ['rule', '1', 'translation', 'port', '443'])
+
+        self.cli_commit()
+
+        nftables_search = [
+            ['iifname "eth1"', 'tcp dport 443', 'dnat to :443']
+        ]
+
+        self.verify_nftables(nftables_search, 'ip vyos_nat')
+
+    def test_static_nat(self):
+        dst_addr_1 = '10.0.1.1'
+        translate_addr_1 = '192.168.1.1'
+        dst_addr_2 = '203.0.113.0/24'
+        translate_addr_2 = '192.0.2.0/24'
+        ifname = 'eth0'
+
+        self.cli_set(static_path + ['rule', '10', 'destination', 'address', dst_addr_1])
+        self.cli_set(static_path + ['rule', '10', 'inbound-interface', ifname])
+        self.cli_set(static_path + ['rule', '10', 'translation', 'address', translate_addr_1])
+
+        self.cli_set(static_path + ['rule', '20', 'destination', 'address', dst_addr_2])
+        self.cli_set(static_path + ['rule', '20', 'inbound-interface', ifname])
+        self.cli_set(static_path + ['rule', '20', 'translation', 'address', translate_addr_2])
+
         self.cli_commit()
 
+        nftables_search = [
+            [f'iifname "{ifname}"', f'ip daddr {dst_addr_1}', f'dnat to {translate_addr_1}'],
+            [f'oifname "{ifname}"', f'ip saddr {translate_addr_1}', f'snat to {dst_addr_1}'],
+            [f'iifname "{ifname}"', f'dnat ip prefix to ip daddr map {{ {dst_addr_2} : {translate_addr_2} }}'],
+            [f'oifname "{ifname}"', f'snat ip prefix to ip daddr map {{ {translate_addr_2} : {dst_addr_2} }}']
+        ]
+
+        self.verify_nftables(nftables_search, 'ip vyos_nat')
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)
-- 
cgit v1.2.3


From e6ba98a85ca72abc7e7e2001d208bcd1806c2c13 Mon Sep 17 00:00:00 2001
From: sarthurdev <965089+sarthurdev@users.noreply.github.com>
Date: Tue, 20 Sep 2022 17:16:43 +0200
Subject: nat66: T4605: Refactor NAT66 to use python module for parsing rules

* Rename table to vyos_nat
* Refactor tests to use `verify_nftables` format
---
 data/templates/firewall/nftables-nat66.j2 | 158 ++++++------------------------
 data/vyos-firewall-init.conf              |  20 ----
 python/vyos/nat.py                        |  27 +++--
 smoketest/scripts/cli/test_nat66.py       |  63 +++++-------
 src/conf_mode/nat66.py                    |  11 ++-
 src/op_mode/show_nat66_statistics.py      |   2 +-
 6 files changed, 85 insertions(+), 196 deletions(-)

(limited to 'smoketest')

diff --git a/data/templates/firewall/nftables-nat66.j2 b/data/templates/firewall/nftables-nat66.j2
index 28714c7a7..27b3eec88 100644
--- a/data/templates/firewall/nftables-nat66.j2
+++ b/data/templates/firewall/nftables-nat66.j2
@@ -1,125 +1,5 @@
 #!/usr/sbin/nft -f
 
-{% macro nptv6_rule(rule,config, chain) %}
-{% set comment  = '' %}
-{% set base_log = '' %}
-{% set dst_prefix  = 'ip6 daddr ' ~ config.destination.prefix.replace('!','!= ') if config.destination.prefix is vyos_defined %}
-{% set src_prefix  = 'ip6 saddr ' ~ config.source.prefix.replace('!','!= ') if config.source.prefix is vyos_defined %}
-{% set source_address  = 'ip6 saddr ' ~ config.source.address.replace('!','!= ') if config.source.address is vyos_defined %}
-{% set dest_address  = 'ip6 daddr ' ~ config.destination.address.replace('!','!= ') if config.destination.address is vyos_defined %}
-{# Port #}
-{% if config.source.port is vyos_defined and config.source.port.startswith('!') %}
-{%     set src_port  = 'sport != { ' ~ config.source.port.replace('!','') ~ ' }' %}
-{% else %}
-{%     set src_port  = 'sport { ' ~ config.source.port ~ ' }' if config.source.port is vyos_defined %}
-{% endif %}
-{% if config.destination.port is vyos_defined and config.destination.port.startswith('!') %}
-{%     set dst_port  = 'dport != { ' ~ config.destination.port.replace('!','') ~ ' }' %}
-{% else %}
-{%     set dst_port  = 'dport { ' ~ config.destination.port ~ ' }' if config.destination.port is vyos_defined %}
-{% endif %}
-{% if chain is vyos_defined('PREROUTING') %}
-{%     set comment   = 'DST-NAT66-' ~ rule %}
-{%     set base_log  = '[NAT66-DST-' ~ rule %}
-{%     set interface = ' iifname "' ~ config.inbound_interface ~ '"' if config.inbound_interface is vyos_defined and config.inbound_interface is not vyos_defined('any') else '' %}
-{%     if config.translation.address | is_ip_network %}
-{#         support 1:1 network translation #}
-{%         set dnat_type = 'dnat prefix to ' %}
-{%     else   %}
-{%         set dnat_type = 'dnat to ' %}
-{%     endif %}
-{%     set trns_address = dnat_type ~ config.translation.address if config.translation.address is vyos_defined %}
-{% elif chain is vyos_defined('POSTROUTING') %}
-{%     set comment   = 'SRC-NAT66-' ~ rule %}
-{%     set base_log  = '[NAT66-SRC-' ~ rule %}
-{%     if config.translation.address is vyos_defined %}
-{%         if config.translation.address is vyos_defined('masquerade') %}
-{%             set trns_address = config.translation.address %}
-{%         else %}
-{%             if config.translation.address | is_ip_network %}
-{#                 support 1:1 network translation #}
-{%                 set snat_type = 'snat prefix to ' %}
-{%             else   %}
-{%                 set snat_type = 'snat to ' %}
-{%             endif %}
-{%             set trns_address = snat_type ~ config.translation.address %}
-{%         endif %}
-{%     endif   %}
-{%     set interface = ' oifname "' ~ config.outbound_interface ~ '"' if config.outbound_interface is vyos_defined else '' %}
-{% endif %}
-{% set trns_port = ':' ~ config.translation.port if config.translation.port is vyos_defined %}
-{# protocol has a default value thus it is always present #}
-{% if config.protocol is vyos_defined('tcp_udp') %}
-{%     set protocol  = 'tcp' %}
-{%     set comment   = comment ~ ' tcp_udp' %}
-{% else %}
-{%     set protocol  = config.protocol %}
-{% endif %}
-{% if config.log is vyos_defined %}
-{%     if config.translation.address is vyos_defined('masquerade') %}
-{%         set log = base_log ~ '-MASQ]' %}
-{%     else %}
-{%         set log = base_log ~ ']' %}
-{%     endif %}
-{% endif %}
-{% if config.exclude is vyos_defined %}
-{#     rule has been marked as 'exclude' thus we simply return here #}
-{%     set trns_addr = 'return' %}
-{%     set trns_port = '' %}
-{% endif %}
-{% set output = 'add rule ip6 nat ' ~ chain ~ interface %}
-{# Count packets #}
-{% set output = output ~ ' counter' %}
-{# Special handling of log option, we must repeat the entire rule before the #}
-{# NAT translation options are added, this is essential                      #}
-{% if log is vyos_defined %}
-{%     set log_output = output ~ ' log prefix "' ~ log ~ '" comment "' ~ comment ~ '"' %}
-{% endif %}
-{% if src_prefix is vyos_defined %}
-{%     set output = output ~ ' ' ~ src_prefix %}
-{% endif %}
-{% if dst_port is vyos_defined %}
-{%     set output = output ~ ' ' ~ protocol ~ ' ' ~ dst_port %}
-{% endif %}
-{% if dst_prefix is vyos_defined %}
-{%     set output = output ~ ' ' ~ dst_prefix %}
-{% endif %}
-{% if source_address is vyos_defined %}
-{%     set output = output ~ ' ' ~ source_address %}
-{% endif %}
-{% if src_port is vyos_defined %}
-{%     set output = output ~ ' ' ~ protocol ~ ' ' ~ src_port %}
-{% endif %}
-{% 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 %}
-{% if trns_port is vyos_defined %}
-{#     Do not add a whitespace here, translation port must be directly added after IP address #}
-{#     e.g. 2001:db8::1:3389                                                                   #}
-{%     set output = output ~ trns_port %}
-{% endif %}
-{% if comment is vyos_defined %}
-{%     set output = output ~ ' comment "' ~ comment ~ '"' %}
-{% endif %}
-{{ log_output if log_output is vyos_defined }}
-{{ output }}
-{# Special handling if protocol is tcp_udp, we must repeat the entire rule with udp as protocol #}
-{% if config.protocol is vyos_defined('tcp_udp') %}
-{#     Beware of trailing whitespace, without it the comment tcp_udp will be changed to udp_udp #}
-{{ log_output | replace('tcp ', 'udp ') if log_output is vyos_defined }}
-{{ output | replace('tcp ', 'udp ') }}
-{% endif %}
-{% endmacro %}
-
-# Start with clean NAT table
-flush table ip6 nat
 {% if helper_functions is vyos_defined('remove') %}
 {# NAT if going to be disabled - remove rules and targets from nftables #}
 {%     set base_command = 'delete rule ip6 raw' %}
@@ -137,19 +17,41 @@ add rule ip6 raw NAT_CONNTRACK counter accept
 {{ base_command }} OUTPUT     position {{ out_ct_conntrack }} counter jump NAT_CONNTRACK
 {% endif %}
 
-#
-# Destination NAT66 rules build up here
-#
+{% if first_install is not vyos_defined %}
+delete table ip6 vyos_nat
+{% endif %}
+table ip6 vyos_nat {
+    #
+    # Destination NAT66 rules build up here
+    #
+    chain PREROUTING {
+        type nat hook prerouting priority -100; policy accept;
+        counter jump VYOS_DNPT_HOOK
 {% if destination.rule is vyos_defined %}
 {%     for rule, config in destination.rule.items() if config.disable is not vyos_defined %}
-{{ nptv6_rule(rule, config, 'PREROUTING') }}
+        {{ config | nat_rule(rule, 'destination', ipv6=True) }}
 {%     endfor %}
 {% endif %}
-#
-# Source NAT66 rules build up here
-#
+    }
+
+    #
+    # Source NAT66 rules build up here
+    #
+    chain POSTROUTING {
+        type nat hook postrouting priority 100; policy accept;
+        counter jump VYOS_SNPT_HOOK
 {% if source.rule is vyos_defined %}
 {%     for rule, config in source.rule.items() if config.disable is not vyos_defined %}
-{{ nptv6_rule(rule, config, 'POSTROUTING') }}
+        {{ config | nat_rule(rule, 'source', ipv6=True) }}
 {%     endfor %}
 {% endif %}
+    }
+
+    chain VYOS_DNPT_HOOK {
+        return
+    }
+
+    chain VYOS_SNPT_HOOK {
+        return
+    }
+}
diff --git a/data/vyos-firewall-init.conf b/data/vyos-firewall-init.conf
index 348299462..3a0e1ee48 100644
--- a/data/vyos-firewall-init.conf
+++ b/data/vyos-firewall-init.conf
@@ -28,26 +28,6 @@ table ip nat {
     }
 }
 
-table ip6 nat {
-    chain PREROUTING {
-        type nat hook prerouting priority -100; policy accept;
-        counter jump VYOS_DNPT_HOOK
-    }
-
-    chain POSTROUTING {
-        type nat hook postrouting priority 100; policy accept;
-        counter jump VYOS_SNPT_HOOK
-    }
-
-    chain VYOS_DNPT_HOOK {
-        return
-    }
-
-    chain VYOS_SNPT_HOOK {
-        return
-    }
-}
-
 table inet mangle {
     chain FORWARD {
         type filter hook forward priority -150; policy accept;
diff --git a/python/vyos/nat.py b/python/vyos/nat.py
index 654afa424..44dd65372 100644
--- a/python/vyos/nat.py
+++ b/python/vyos/nat.py
@@ -17,11 +17,15 @@
 from vyos.template import is_ip_network
 from vyos.util import dict_search_args
 
-def parse_nat_rule(rule_conf, rule_id, nat_type):
+def parse_nat_rule(rule_conf, rule_id, nat_type, ipv6=False):
     output = []
+    ip_prefix = 'ip6' if ipv6 else 'ip'
     log_prefix = ('DST' if nat_type == 'destination' else 'SRC') + f'-NAT-{rule_id}'
     log_suffix = ''
 
+    if ipv6:
+        log_prefix = log_prefix.replace("NAT-", "NAT66-")
+
     ignore_type_addr = False
     translation_str = ''
 
@@ -39,7 +43,7 @@ def parse_nat_rule(rule_conf, rule_id, nat_type):
         protocol = rule_conf['protocol']
         if protocol == 'tcp_udp':
             protocol = '{ tcp, udp }'
-        output.append(f'ip protocol {protocol}')
+        output.append(f'meta l4proto {protocol}')
 
     if 'exclude' in rule_conf:
         translation_str = 'return'
@@ -51,9 +55,12 @@ def parse_nat_rule(rule_conf, rule_id, nat_type):
         port = dict_search_args(rule_conf, 'translation', 'port')
 
         if addr and is_ip_network(addr):
-            map_addr =  dict_search_args(rule_conf, nat_type, 'address')
-            translation_output.append(f'ip prefix to ip {translation_prefix}addr map {{ {map_addr} : {addr} }}')
-            ignore_type_addr = True
+            if not ipv6:
+                map_addr =  dict_search_args(rule_conf, nat_type, 'address')
+                translation_output.append(f'{ip_prefix} prefix to {ip_prefix} {translation_prefix}addr map {{ {map_addr} : {addr} }}')
+                ignore_type_addr = True
+            else:
+                translation_output.append(f'prefix to {addr}')
         elif addr == 'masquerade':
             if port:
                 addr = f'{addr} to '
@@ -85,7 +92,15 @@ def parse_nat_rule(rule_conf, rule_id, nat_type):
             if addr[:1] == '!':
                 operator = '!='
                 addr = addr[1:]
-            output.append(f'ip {prefix}addr {operator} {addr}')
+            output.append(f'{ip_prefix} {prefix}addr {operator} {addr}')
+
+        addr_prefix = dict_search_args(rule_conf, target, 'prefix')
+        if addr_prefix and ipv6:
+            operator = ''
+            if addr_prefix[:1] == '!':
+                operator = '!='
+                addr_prefix = addr[1:]
+            output.append(f'ip6 {prefix}addr {operator} {addr_prefix}')
 
         port = dict_search_args(rule_conf, target, 'port')
         if port:
diff --git a/smoketest/scripts/cli/test_nat66.py b/smoketest/scripts/cli/test_nat66.py
index 537b094a4..6cf7ca0a1 100755
--- a/smoketest/scripts/cli/test_nat66.py
+++ b/smoketest/scripts/cli/test_nat66.py
@@ -71,12 +71,12 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
         self.cli_commit()
 
         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']
+            ['oifname "eth1"', f'ip6 saddr {source_prefix}', f'snat prefix to {translation_prefix}'],
+            ['oifname "eth1"', f'ip6 saddr {source_prefix}', 'masquerade'],
+            ['oifname "eth1"', f'ip6 saddr {source_prefix}', 'return']
         ]
 
-        self.verify_nftables(nftables_search, 'ip6 nat')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_source_nat66_address(self):
         source_prefix = 'fc00::/64'
@@ -88,25 +88,11 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
         # 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'], 'POSTROUTING')
-            self.assertEqual(data['family'], 'ip6')
-            self.assertEqual(data['table'], 'nat')
-
-            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])
-            snat_address = dict_search('snat.addr', data['expr'][3])
+        nftables_search = [
+            ['oifname "eth1"', f'ip6 saddr {source_prefix}', f'snat to {translation_address}']
+        ]
 
-            self.assertEqual(iface, 'eth1')
-            # check for translation address
-            self.assertEqual(snat_address, translation_address)
-            self.assertEqual(f'{address}/{mask}', source_prefix)
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_destination_nat66(self):
         destination_address = 'fc00::1'
@@ -129,7 +115,7 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
             ['iifname "eth1"', 'ip6 saddr fc02::1', 'ip6 daddr fc00::1', 'return']
         ]
 
-        self.verify_nftables(nftables_search, 'ip6 nat')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_destination_nat66_protocol(self):
         translation_address = '2001:db8:1111::1'
@@ -153,7 +139,7 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
             ['iifname "eth1"', 'tcp dport 4545', 'ip6 saddr 2001:db8:2222::/64', 'tcp sport 8080', 'dnat to 2001:db8:1111::1:5555']
         ]
 
-        self.verify_nftables(nftables_search, 'ip6 nat')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_destination_nat66_prefix(self):
         destination_prefix = 'fc00::/64'
@@ -165,22 +151,25 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
         # 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))
+        nftables_search = [
+            ['iifname "eth1"', f'ip6 daddr {destination_prefix}', f'dnat prefix to {translation_prefix}']
+        ]
+
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
-        for idx in range(0, len(data_json)):
-            data = data_json[idx]
+    def test_destination_nat66_without_translation_address(self):
+        self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'eth1'])
+        self.cli_set(dst_path + ['rule', '1', 'destination', 'port', '443'])
+        self.cli_set(dst_path + ['rule', '1', 'protocol', 'tcp'])
+        self.cli_set(dst_path + ['rule', '1', 'translation', 'port', '443'])
 
-            self.assertEqual(data['chain'], 'PREROUTING')
-            self.assertEqual(data['family'], 'ip6')
-            self.assertEqual(data['table'], 'nat')
+        self.cli_commit()
 
-            iface = dict_search('match.right', data['expr'][0])
-            translation_address = dict_search('dnat.addr.prefix.addr', data['expr'][3])
-            translation_mask = dict_search('dnat.addr.prefix.len', data['expr'][3])
+        nftables_search = [
+            ['iifname "eth1"', 'tcp dport 443', 'dnat to :443']
+        ]
 
-            self.assertEqual(f'{translation_address}/{translation_mask}', translation_prefix)
-            self.assertEqual(iface, 'eth1')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_source_nat66_required_translation_prefix(self):
         # T2813: Ensure translation address is specified
@@ -222,7 +211,7 @@ class TestNAT66(VyOSUnitTestSHIM.TestCase):
             ['oifname "eth1"', 'ip6 saddr 2001:db8:2222::/64', 'tcp dport 9999', 'tcp sport 8080', 'snat to 2001:db8:1111::1:80']
         ]
 
-        self.verify_nftables(nftables_search, 'ip6 nat')
+        self.verify_nftables(nftables_search, 'ip6 vyos_nat')
 
     def test_nat66_no_rules(self):
         # T3206: deleting all rules but keep the direction 'destination' or
diff --git a/src/conf_mode/nat66.py b/src/conf_mode/nat66.py
index f64102d88..d8f913b0c 100755
--- a/src/conf_mode/nat66.py
+++ b/src/conf_mode/nat66.py
@@ -36,7 +36,7 @@ airbag.enable()
 
 k_mod = ['nft_nat', 'nft_chain_nat']
 
-nftables_nat66_config = '/tmp/vyos-nat66-rules.nft'
+nftables_nat66_config = '/run/nftables_nat66.nft'
 ndppd_config = '/run/ndppd/ndppd.conf'
 
 def get_handler(json, chain, target):
@@ -147,6 +147,9 @@ def verify(nat):
     return None
 
 def generate(nat):
+    if not os.path.exists(nftables_nat66_config):
+        nat['first_install'] = True
+
     render(nftables_nat66_config, 'firewall/nftables-nat66.j2', nat, permission=0o755)
     render(ndppd_config, 'ndppd/ndppd.conf.j2', nat, permission=0o755)
     return None
@@ -154,15 +157,15 @@ def generate(nat):
 def apply(nat):
     if not nat:
         return None
-    cmd(f'{nftables_nat66_config}')
+
+    cmd(f'nft -f {nftables_nat66_config}')
+
     if 'deleted' in nat or not dict_search('source.rule', nat):
         cmd('systemctl stop ndppd')
         if os.path.isfile(ndppd_config):
             os.unlink(ndppd_config)
     else:
         cmd('systemctl restart ndppd')
-    if os.path.isfile(nftables_nat66_config):
-        os.unlink(nftables_nat66_config)
 
     return None
 
diff --git a/src/op_mode/show_nat66_statistics.py b/src/op_mode/show_nat66_statistics.py
index bc81692ae..cb10aed9f 100755
--- a/src/op_mode/show_nat66_statistics.py
+++ b/src/op_mode/show_nat66_statistics.py
@@ -44,7 +44,7 @@ group.add_argument("--destination", help="Show statistics for configured destina
 args = parser.parse_args()
 
 if args.source or args.destination:
-    tmp = cmd('sudo nft -j list table ip6 nat')
+    tmp = cmd('sudo nft -j list table ip6 vyos_nat')
     tmp = json.loads(tmp)
 
     source = r"nftables[?rule.chain=='POSTROUTING'].rule.{chain: chain, handle: handle, comment: comment, counter: expr[].counter | [0], interface: expr[].match.right | [0] }"
-- 
cgit v1.2.3


From c6bbe051574acf5ca1501e631d73ac06bdb17b30 Mon Sep 17 00:00:00 2001
From: sarthurdev <965089+sarthurdev@users.noreply.github.com>
Date: Wed, 21 Sep 2022 02:05:30 +0200
Subject: nat: T4605: Refactor static NAT to use python module for parsing
 rules

* Rename table to vyos_nat
* Add static NAT smoketest
---
 data/templates/firewall/nftables-static-nat.j2 | 122 ++++---------------------
 data/vyos-firewall-init.conf                   |  20 ----
 python/vyos/nat.py                             |  62 +++++++++++++
 python/vyos/template.py                        |   5 +
 smoketest/scripts/cli/test_nat.py              |   4 +-
 src/conf_mode/nat.py                           |   4 +-
 6 files changed, 91 insertions(+), 126 deletions(-)

(limited to 'smoketest')

diff --git a/data/templates/firewall/nftables-static-nat.j2 b/data/templates/firewall/nftables-static-nat.j2
index d3c43858f..790c33ce9 100644
--- a/data/templates/firewall/nftables-static-nat.j2
+++ b/data/templates/firewall/nftables-static-nat.j2
@@ -1,115 +1,31 @@
 #!/usr/sbin/nft -f
 
-{% macro nat_rule(rule, config, chain) %}
-{% set comment  = '' %}
-{% set base_log = '' %}
-
-{% if chain is vyos_defined('PREROUTING') %}
-{%     set comment   = 'STATIC-NAT-' ~ rule %}
-{%     set base_log  = '[NAT-DST-' ~ rule %}
-{%     set interface = ' iifname "' ~ config.inbound_interface ~ '"' if config.inbound_interface is vyos_defined and config.inbound_interface is not vyos_defined('any') else '' %}
-{%     if config.translation.address is vyos_defined %}
-{#         support 1:1 network translation #}
-{%         if config.translation.address | is_ip_network %}
-{%             set trns_addr = 'dnat ip prefix to ip daddr map { ' ~ config.destination.address ~ ' : ' ~ config.translation.address ~ ' }' %}
-{#             we can now clear out the dst_addr part as it's already covered in aboves map #}
-{%         else %}
-{%             set dst_addr  = 'ip daddr ' ~ config.destination.address if config.destination.address is vyos_defined %}
-{%             set trns_addr = 'dnat to ' ~ config.translation.address %}
-{%         endif %}
-{%     endif %}
-{% elif chain is vyos_defined('POSTROUTING') %}
-{%     set comment   = 'STATIC-NAT-' ~ rule %}
-{%     set base_log  = '[NAT-SRC-' ~ rule %}
-{%     set interface = ' oifname "' ~ config.inbound_interface ~ '"' if config.inbound_interface is vyos_defined and config.inbound_interface is not vyos_defined('any') else '' %}
-{%     if config.translation.address is vyos_defined %}
-{#         support 1:1 network translation #}
-{%         if config.translation.address | is_ip_network %}
-{%             set trns_addr = 'snat ip prefix to ip saddr map { ' ~ config.translation.address ~ ' : ' ~ config.destination.address ~ ' }' %}
-{#             we can now clear out the src_addr part as it's already covered in aboves map #}
-{%         else %}
-{%             set src_addr  = 'ip saddr ' ~ config.translation.address if config.translation.address is vyos_defined %}
-{%             set trns_addr = 'snat to ' ~ config.destination.address %}
-{%         endif %}
-{%     endif %}
-{% endif %}
-
-{% if config.exclude is vyos_defined %}
-{#     rule has been marked as 'exclude' thus we simply return here #}
-{%     set trns_addr = 'return' %}
-{%     set trns_port = '' %}
-{% endif %}
-
-{% if config.translation.options is vyos_defined %}
-{%     if config.translation.options.address_mapping is vyos_defined('persistent') %}
-{%         set trns_opts_addr  = 'persistent' %}
-{%     endif %}
-{%     if config.translation.options.port_mapping is vyos_defined('random') %}
-{%         set trns_opts_port  = 'random' %}
-{%     elif config.translation.options.port_mapping is vyos_defined('fully-random') %}
-{%         set trns_opts_port  = 'fully-random' %}
-{%     endif %}
-{% endif %}
-
-{% if trns_opts_addr is vyos_defined and trns_opts_port is vyos_defined %}
-{%     set trns_opts  = trns_opts_addr ~ ',' ~ trns_opts_port %}
-{% elif trns_opts_addr is vyos_defined %}
-{%     set trns_opts  = trns_opts_addr %}
-{% elif trns_opts_port is vyos_defined %}
-{%     set trns_opts  = trns_opts_port %}
-{% endif %}
-
-{% set output = 'add rule ip vyos_static_nat ' ~ chain ~ interface %}
-
-{% if dst_addr is vyos_defined %}
-{%     set output = output ~ ' ' ~ dst_addr %}
+{% if first_install is not vyos_defined %}
+delete table ip vyos_static_nat
 {% endif %}
-{% if src_addr is vyos_defined %}
-{%     set output = output ~ ' ' ~ src_addr %}
-{% endif %}
-
-{# Count packets #}
-{% set output = output ~ ' counter' %}
-{# Special handling of log option, we must repeat the entire rule before the #}
-{# NAT translation options are added, this is essential                      #}
-{% if log is vyos_defined %}
-{%     set log_output = output ~ ' log prefix "' ~ log ~ '" comment "' ~ comment ~ '"' %}
-{% endif %}
-{% if trns_addr is vyos_defined %}
-{%     set output = output ~ ' ' ~ trns_addr %}
-{% endif %}
-
-{% if trns_opts is vyos_defined %}
-{%     set output = output ~ ' ' ~ trns_opts %}
-{% endif %}
-{% if comment is vyos_defined %}
-{%     set output = output ~ ' comment "' ~ comment ~ '"' %}
-{% endif %}
-{{ log_output if log_output is vyos_defined }}
-{{ output }}
-{% endmacro %}
-
-# Start with clean STATIC NAT chains
-flush chain ip vyos_static_nat PREROUTING
-flush chain ip vyos_static_nat POSTROUTING
+table ip vyos_static_nat {
+    #
+    # Destination NAT rules build up here
+    #
 
-{# NAT if enabled - add targets to nftables #}
-
-#
-# Destination NAT rules build up here
-#
-add rule ip vyos_static_nat PREROUTING counter jump VYOS_PRE_DNAT_HOOK
+    chain PREROUTING {
+        type nat hook prerouting priority -100; policy accept;
 {% if static.rule is vyos_defined %}
 {%     for rule, config in static.rule.items() if config.disable is not vyos_defined %}
-{{ nat_rule(rule, config, 'PREROUTING') }}
+    {{ config | nat_static_rule(rule, 'destination') }}
 {%     endfor %}
 {% endif %}
-#
-# Source NAT rules build up here
-#
-add rule ip vyos_static_nat POSTROUTING counter jump VYOS_PRE_SNAT_HOOK
+    }
+
+    #
+    # Source NAT rules build up here
+    #
+    chain POSTROUTING {
+        type nat hook postrouting priority 100; policy accept;
 {% if static.rule is vyos_defined %}
 {%     for rule, config in static.rule.items() if config.disable is not vyos_defined %}
-{{ nat_rule(rule, config, 'POSTROUTING') }}
+    {{ config | nat_static_rule(rule, 'source') }}
 {%     endfor %}
 {% endif %}
+    }
+}
diff --git a/data/vyos-firewall-init.conf b/data/vyos-firewall-init.conf
index 3a0e1ee48..11a5bc7bf 100644
--- a/data/vyos-firewall-init.conf
+++ b/data/vyos-firewall-init.conf
@@ -1,25 +1,5 @@
 #!/usr/sbin/nft -f
 
-table ip vyos_static_nat {
-    chain PREROUTING {
-        type nat hook prerouting priority -100; policy accept;
-        counter jump VYOS_PRE_DNAT_HOOK
-    }
-
-    chain POSTROUTING {
-        type nat hook postrouting priority 100; policy accept;
-        counter jump VYOS_PRE_SNAT_HOOK
-    }
-
-    chain VYOS_PRE_DNAT_HOOK {
-        return
-    }
-
-    chain VYOS_PRE_SNAT_HOOK {
-        return
-    }
-}
-
 # Required by wanloadbalance
 table ip nat {
     chain VYOS_PRE_SNAT_HOOK {
diff --git a/python/vyos/nat.py b/python/vyos/nat.py
index 44dd65372..31bbdc386 100644
--- a/python/vyos/nat.py
+++ b/python/vyos/nat.py
@@ -124,3 +124,65 @@ def parse_nat_rule(rule_conf, rule_id, nat_type, ipv6=False):
     output.append(f'comment "{log_prefix}"')
 
     return " ".join(output)
+
+def parse_nat_static_rule(rule_conf, rule_id, nat_type):
+    output = []
+    log_prefix = ('STATIC-DST' if nat_type == 'destination' else 'STATIC-SRC') + f'-NAT-{rule_id}'
+    log_suffix = ''
+
+    ignore_type_addr = False
+    translation_str = ''
+
+    if 'inbound_interface' in rule_conf:
+        ifname = rule_conf['inbound_interface']
+        ifprefix = 'i' if nat_type == 'destination' else 'o'
+        if ifname != 'any':
+            output.append(f'{ifprefix}ifname "{ifname}"')
+
+    if 'exclude' in rule_conf:
+        translation_str = 'return'
+        log_suffix = '-EXCL'
+    elif 'translation' in rule_conf:
+        translation_prefix = nat_type[:1]
+        translation_output = [f'{translation_prefix}nat']
+        addr = dict_search_args(rule_conf, 'translation', 'address')
+        map_addr =  dict_search_args(rule_conf, 'destination', 'address')
+
+        if nat_type == 'source':
+            addr, map_addr = map_addr, addr # Swap
+
+        if addr and is_ip_network(addr):
+            translation_output.append(f'ip prefix to ip {translation_prefix}addr map {{ {map_addr} : {addr} }}')
+            ignore_type_addr = True
+        elif addr:
+            translation_output.append(f'to {addr}')
+
+        options = []
+        addr_mapping = dict_search_args(rule_conf, 'translation', 'options', 'address_mapping')
+        port_mapping = dict_search_args(rule_conf, 'translation', 'options', 'port_mapping')
+        if addr_mapping == 'persistent':
+            options.append('persistent')
+        if port_mapping and port_mapping != 'none':
+            options.append(port_mapping)
+
+        if options:
+            translation_output.append(",".join(options))
+
+        translation_str = " ".join(translation_output)
+
+    prefix = nat_type[:1]
+    addr = dict_search_args(rule_conf, 'translation' if nat_type == 'source' else nat_type, 'address')
+    if addr and not ignore_type_addr:
+        output.append(f'ip {prefix}addr {addr}')
+
+    output.append('counter')
+
+    if translation_str:
+        output.append(translation_str)
+
+    if 'log' in rule_conf:
+        output.append(f'log prefix "[{log_prefix}{log_suffix}]"')
+
+    output.append(f'comment "{log_prefix}"')
+
+    return " ".join(output)
diff --git a/python/vyos/template.py b/python/vyos/template.py
index d9ff98d2e..0870a0523 100644
--- a/python/vyos/template.py
+++ b/python/vyos/template.py
@@ -621,6 +621,11 @@ def nat_rule(rule_conf, rule_id, nat_type, ipv6=False):
     from vyos.nat import parse_nat_rule
     return parse_nat_rule(rule_conf, rule_id, nat_type, ipv6)
 
+@register_filter('nat_static_rule')
+def nat_static_rule(rule_conf, rule_id, nat_type):
+    from vyos.nat import parse_nat_static_rule
+    return parse_nat_static_rule(rule_conf, rule_id, nat_type)
+
 @register_filter('range_to_regex')
 def range_to_regex(num_range):
     from vyos.range_regex import range_to_regex
diff --git a/smoketest/scripts/cli/test_nat.py b/smoketest/scripts/cli/test_nat.py
index 5863409be..f824838c0 100755
--- a/smoketest/scripts/cli/test_nat.py
+++ b/smoketest/scripts/cli/test_nat.py
@@ -184,10 +184,10 @@ class TestNAT(VyOSUnitTestSHIM.TestCase):
             [f'iifname "{ifname}"', f'ip daddr {dst_addr_1}', f'dnat to {translate_addr_1}'],
             [f'oifname "{ifname}"', f'ip saddr {translate_addr_1}', f'snat to {dst_addr_1}'],
             [f'iifname "{ifname}"', f'dnat ip prefix to ip daddr map {{ {dst_addr_2} : {translate_addr_2} }}'],
-            [f'oifname "{ifname}"', f'snat ip prefix to ip daddr map {{ {translate_addr_2} : {dst_addr_2} }}']
+            [f'oifname "{ifname}"', f'snat ip prefix to ip saddr map {{ {translate_addr_2} : {dst_addr_2} }}']
         ]
 
-        self.verify_nftables(nftables_search, 'ip vyos_nat')
+        self.verify_nftables(nftables_search, 'ip vyos_static_nat')
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)
diff --git a/src/conf_mode/nat.py b/src/conf_mode/nat.py
index 3f52d7c1f..8b1a5a720 100755
--- a/src/conf_mode/nat.py
+++ b/src/conf_mode/nat.py
@@ -194,7 +194,9 @@ def generate(nat):
     if tmp > 0:
         raise ConfigError('Configuration file errors encountered!')
 
-    tmp = run(f'nft -c -f {nftables_nat_config}')
+    tmp = run(f'nft -c -f {nftables_static_nat_conf}')
+    if tmp > 0:
+        raise ConfigError('Configuration file errors encountered!')
 
     return None
 
-- 
cgit v1.2.3