Merge pull request #1426 from sever-sever/T4545-nat

nat: T4545: Rewrite show nat source rules script

1 files changed, 162 insertions, 0 deletions

diff --git a/src/op_mode/nat.py b/src/op_mode/nat.py
new file mode 100755
index 000000000..f3f79f863
--- /dev/null
+++ b/src/op_mode/nat.py
@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2022 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import jmespath
+import json
+import sys
+
+from sys import exit
+from tabulate import tabulate
+
+from vyos.util import cmd
+from vyos.util import dict_search
+
+import vyos.opmode
+
+
+def _get_json_data(direction):
+    """
+    Get NAT format JSON
+    """
+    if direction == 'source':
+        chain = 'POSTROUTING'
+    if direction == 'destination':
+        chain = 'PREROUTING'
+    return cmd(f'sudo nft --json list chain ip nat {chain}')
+
+
+def _get_raw_data_rules(direction):
+    """Get interested rules
+    :returns dict
+    """
+    data = _get_json_data(direction)
+    data_dict = json.loads(data)
+    rules = []
+    for rule in data_dict['nftables']:
+        if 'rule' in rule and 'comment' in rule['rule']:
+            rules.append(rule)
+    return rules
+
+
+def _get_formatted_output_rules(data, direction):
+    data_entries = []
+    for rule in data:
+        if 'comment' in rule['rule']:
+            comment = rule.get('rule').get('comment')
+            rule_number = comment.split('-')[-1]
+            rule_number = rule_number.split(' ')[0]
+        if 'expr' in rule['rule']:
+            interface = rule.get('rule').get('expr')[0].get('match').get('right') \
+                if jmespath.search('rule.expr[*].match.left.meta', rule) else 'any'
+        for index, match in enumerate(jmespath.search('rule.expr[*].match', rule)):
+            if 'payload' in match['left']:
+                if 'prefix' in match['right'] or 'set' in match['right']:
+                    # Merge dict src/dst l3_l4 parameters
+                    my_dict = {**match['left']['payload'], **match['right']}
+                    proto = my_dict.get('protocol').upper()
+                    if my_dict['field'] == 'saddr':
+                        saddr = f'{my_dict["prefix"]["addr"]}/{my_dict["prefix"]["len"]}'
+                    elif my_dict['field'] == 'daddr':
+                        daddr = f'{my_dict["prefix"]["addr"]}/{my_dict["prefix"]["len"]}'
+                    elif my_dict['field'] == 'sport':
+                        # Port range or single port
+                        if jmespath.search('set[*].range', my_dict):
+                            sport = my_dict['set'][0]['range']
+                            sport = '-'.join(map(str, sport))
+                        else:
+                            sport = my_dict.get('set')
+                            sport = ','.join(map(str, sport))
+                    elif my_dict['field'] == 'dport':
+                        # Port range or single port
+                        if jmespath.search('set[*].range', my_dict):
+                            dport = my_dict["set"][0]["range"]
+                            dport = '-'.join(map(str, dport))
+                        else:
+                            dport = my_dict.get('set')
+                            dport = ','.join(map(str, dport))
+                else:
+                    if jmespath.search('left.payload.field', match) == 'saddr':
+                        saddr = match.get('right')
+                    if jmespath.search('left.payload.field', match) == 'daddr':
+                        daddr = match.get('right')
+            else:
+                saddr = '0.0.0.0/0'
+                daddr = '0.0.0.0/0'
+                sport = 'any'
+                dport = 'any'
+                proto = 'any'
+
+            source = f'''{saddr}
+sport {sport}'''
+            destination = f'''{daddr}
+dport {dport}'''
+
+            if jmespath.search('left.payload.field', match) == 'protocol':
+                field_proto = match.get('right').upper()
+
+            for expr in rule.get('rule').get('expr'):
+                if 'snat' in expr:
+                    translation = dict_search('snat.addr', expr)
+                    if expr['snat'] and 'port' in expr['snat']:
+                        if jmespath.search('snat.port.range', expr):
+                            port = dict_search('snat.port.range', expr)
+                            port = '-'.join(map(str, port))
+                        else:
+                            port = expr['snat']['port']
+                        translation = f'''{translation}
+port {port}'''
+
+                elif 'masquerade' in expr:
+                    translation = 'masquerade'
+                    if expr['masquerade'] and 'port' in expr['masquerade']:
+                        if jmespath.search('masquerade.port.range', expr):
+                            port = dict_search('masquerade.port.range', expr)
+                            port = '-'.join(map(str, port))
+                        else:
+                            port = expr['masquerade']['port']
+
+                        translation = f'''{translation}
+port {port}'''
+                else:
+                    translation = 'exclude'
+        # Overwrite match loop 'proto' if specified filed 'protocol' exist
+        if 'protocol' in jmespath.search('rule.expr[*].match.left.payload.field', rule):
+            proto = jmespath.search('rule.expr[0].match.right', rule).upper()
+
+        data_entries.append([rule_number, source, destination, proto, interface, translation])
+
+    interface_header = 'Out-Int' if direction == 'source' else 'In-Int'
+    headers = ["Rule", "Source", "Destination", "Proto", interface_header, "Translation"]
+    output = tabulate(data_entries, headers, numalign="left")
+    return output
+
+
+def show_rules(raw: bool, direction: str):
+    nat_rules = _get_raw_data_rules(direction)
+    if raw:
+        return nat_rules
+    else:
+        return _get_formatted_output_rules(nat_rules, direction)
+
+
+if __name__ == '__main__':
+    try:
+        res = vyos.opmode.run(sys.modules[__name__])
+        if res:
+            print(res)
+    except ValueError as e:
+        print(e)
+        sys.exit(1)