nat: nat66: T4650: Rewrite op-mode nat translation

Rewrite op-moe "show nat|nat66 translation" to vyos.opmode format
Ability to get machine-readable format "raw"

1 files changed, 101 insertions, 0 deletions

diff --git a/src/op_mode/nat.py b/src/op_mode/nat.py
index dec04aa48..1339d5b92 100755
--- a/src/op_mode/nat.py
+++ b/src/op_mode/nat.py
@@ -17,6 +17,7 @@
 import jmespath
 import json
 import sys
+import xmltodict
 
 from sys import exit
 from tabulate import tabulate
@@ -27,6 +28,29 @@ from vyos.util import dict_search
 import vyos.opmode
 
 
+def _get_xml_translation(direction, family):
+    """
+    Get conntrack XML output --src-nat|--dst-nat
+    """
+    if direction == 'source':
+        opt = '--src-nat'
+    if direction == 'destination':
+        opt = '--dst-nat'
+    return cmd(f'sudo conntrack --dump --family {family} {opt} --output xml')
+
+
+def _xml_to_dict(xml):
+    """
+    Convert XML to dictionary
+    Return: dictionary
+    """
+    parse = xmltodict.parse(xml, attr_prefix='')
+    # If only one conntrack entry we must change dict
+    if 'meta' in parse['conntrack']['flow']:
+        return dict(conntrack={'flow': [parse['conntrack']['flow']]})
+    return parse
+
+
 def _get_json_data(direction, family):
     """
     Get NAT format JSON
@@ -52,6 +76,22 @@ def _get_raw_data_rules(direction, family):
     return rules
 
 
+def _get_raw_translation(direction, family):
+    """
+    Return: dictionary
+    """
+    xml = _get_xml_translation(direction, family)
+    if len(xml) == 0:
+        output = {'conntrack':
+            {
+                'error': True,
+                'reason': 'entries not found'
+            }
+        }
+        return output
+    return _xml_to_dict(xml)
+
+
 def _get_formatted_output_rules(data, direction, family):
     # Add default values before loop
     sport, dport, proto = 'any', 'any', 'any'
@@ -180,6 +220,58 @@ def _get_formatted_output_statistics(data, direction):
     return output
 
 
+def _get_formatted_translation(dict_data, nat_direction, family):
+    data_entries = []
+    if 'error' in dict_data['conntrack']:
+        return 'Entries not found'
+    for entry in dict_data['conntrack']['flow']:
+        orig_src, orig_dst, orig_sport, orig_dport = {}, {}, {}, {}
+        reply_src, reply_dst, reply_sport, reply_dport = {}, {}, {}, {}
+        proto = {}
+        for meta in entry['meta']:
+            direction = meta['direction']
+            if direction in ['original']:
+                if 'layer3' in meta:
+                    orig_src = meta['layer3']['src']
+                    orig_dst = meta['layer3']['dst']
+                if 'layer4' in meta:
+                    if meta.get('layer4').get('sport'):
+                        orig_sport = meta['layer4']['sport']
+                    if meta.get('layer4').get('dport'):
+                        orig_dport = meta['layer4']['dport']
+                    proto = meta['layer4']['protoname']
+            if direction in ['reply']:
+                if 'layer3' in meta:
+                    reply_src = meta['layer3']['src']
+                    reply_dst = meta['layer3']['dst']
+                if 'layer4' in meta:
+                    if meta.get('layer4').get('sport'):
+                        reply_sport = meta['layer4']['sport']
+                    if meta.get('layer4').get('dport'):
+                        reply_dport = meta['layer4']['dport']
+                    proto = meta['layer4']['protoname']
+            if direction == 'independent':
+                conn_id = meta['id']
+                timeout = meta['timeout']
+                orig_src = f'{orig_src}:{orig_sport}' if orig_sport else orig_src
+                orig_dst = f'{orig_dst}:{orig_dport}' if orig_dport else orig_dst
+                reply_src = f'{reply_src}:{reply_sport}' if reply_sport else reply_src
+                reply_dst = f'{reply_dst}:{reply_dport}' if reply_dport else reply_dst
+                state = meta['state'] if 'state' in meta else ''
+                mark = meta['mark']
+                zone = meta['zone'] if 'zone' in meta else ''
+                if nat_direction == 'source':
+                    data_entries.append(
+                        [orig_src, reply_dst, proto, timeout, mark, zone])
+                elif nat_direction == 'destination':
+                    data_entries.append(
+                        [orig_dst, reply_src, proto, timeout, mark, zone])
+
+    headers = ["Pre-NAT", "Post-NAT", "Proto", "Timeout", "Mark", "Zone"]
+    output = tabulate(data_entries, headers, numalign="left")
+    return output
+
+
 def show_rules(raw: bool, direction: str, family: str):
     nat_rules = _get_raw_data_rules(direction, family)
     if raw:
@@ -196,6 +288,15 @@ def show_statistics(raw: bool, direction: str, family: str):
         return _get_formatted_output_statistics(nat_statistics, direction)
 
 
+def show_translations(raw: bool, direction: str, family: str):
+    family = 'ipv6' if family == 'inet6' else 'ipv4'
+    nat_translation = _get_raw_translation(direction, family)
+    if raw:
+        return nat_translation
+    else:
+        return _get_formatted_translation(nat_translation, direction, family)
+
+
 if __name__ == '__main__':
     try:
         res = vyos.opmode.run(sys.modules[__name__])