Merge pull request #5006 from ritika0313/T7513-CGNAT-exclude-rule-vyos-1x

T7513: vyos-1x: CGNAT Exclude Rule CLI support

2 files changed, 115 insertions, 0 deletions

diff --git a/src/conf_mode/vpp_nat_cgnat.py b/src/conf_mode/vpp_nat_cgnat.py
index cd38d8697..48d0339b1 100644
--- a/src/conf_mode/vpp_nat_cgnat.py
+++ b/src/conf_mode/vpp_nat_cgnat.py
@@ -26,6 +26,13 @@ from vyos.vpp.utils import vpp_iface_name_transform
 from vyos.vpp.nat.det44 import Det44
 from vyos.vpp.control_vpp import VPPControl
 
+protocol_map = {
+    'all': 0,
+    'icmp': 1,
+    'tcp': 6,
+    'udp': 17,
+}
+
 
 def get_config(config=None) -> dict:
     if config:
@@ -80,12 +87,22 @@ def get_config(config=None) -> dict:
         expand_nodes=Diff.DELETE | Diff.ADD,
     )
 
+    changed_exclude_rules = node_changed(
+        conf,
+        base + ['exclude', 'rule'],
+        key_mangling=('-', '_'),
+        recursive=True,
+        expand_nodes=Diff.DELETE | Diff.ADD,
+    )
+
     if not config_changed:
         changed_rules = list(config.get('rule', {}).keys())
+        changed_exclude_rules = list(config.get('exclude', {}).get('rule', {}).keys())
 
     config.update(
         {
             'changed_rules': changed_rules,
+            'changed_exclude_rules': changed_exclude_rules,
             'vpp_ifaces': cli_ifaces_list(conf),
         }
     )
@@ -140,6 +157,46 @@ def verify(config):
                 f'Please add: {", ".join(missing_keys).replace("_", "-")}'
             )
 
+    # Verify exclude rules (identity mappings)
+    if 'exclude' in config:
+        # Track identity mappings to detect duplicates
+        seen_mappings = {}
+
+        for rule, rule_config in config['exclude'].get('rule', {}).items():
+            error_msg = f'Exclude rule {rule}:'
+
+            if 'local_address' not in rule_config:
+                raise ConfigError(f'{error_msg} local-address must be specified')
+
+            has_protocol = (
+                'protocol' in rule_config and rule_config.get('protocol') != 'all'
+            )
+            has_port = 'local_port' in rule_config
+
+            # Either both protocol and local-port are set, or neither
+            if has_protocol != has_port:
+                raise ConfigError(
+                    f'{error_msg} protocol and local-port must either both be specified or both omitted'
+                )
+
+            # Check for duplicate identity mappings
+            # VPP identifies identity mappings by (address, protocol, port) tuple
+            local_addr = rule_config['local_address']
+            protocol = rule_config.get('protocol', 'all')
+            port = rule_config.get('local_port', 0)
+
+            mapping_key = (local_addr, protocol, port)
+
+            if mapping_key in seen_mappings:
+                duplicate_rule = seen_mappings[mapping_key]
+                raise ConfigError(
+                    f'{error_msg} duplicate identity mapping - '
+                    f'address {local_addr}, protocol {protocol}, port {port} '
+                    f'already configured in exclude rule {duplicate_rule}'
+                )
+
+            seen_mappings[mapping_key] = rule
+
 
 def generate(config):
     pass
@@ -175,6 +232,16 @@ def apply(config):
                     out_addr=out_addr,
                     out_plen=int(out_plen),
                 )
+        # Delete CGNAT exclude rules
+        for rule in config['changed_exclude_rules']:
+            if rule in remove_config.get('exclude', {}).get('rule', {}):
+                rule_config = remove_config['exclude']['rule'][rule]
+                cgnat.delete_det44_identity_mapping(
+                    ip_address=rule_config.get('local_address'),
+                    protocol=protocol_map[rule_config.get('protocol', 'all')],
+                    port=int(rule_config.get('local_port', 0)),
+                    tag=rule_config.get('description', ''),
+                )
 
     # Add DET44
     cgnat.enable_det44_plugin()
@@ -198,6 +265,16 @@ def apply(config):
                 out_addr=out_addr,
                 out_plen=int(out_plen),
             )
+    # Add CGNAT exclude rules
+    for rule in config['changed_exclude_rules']:
+        if rule in config.get('exclude', {}).get('rule', {}):
+            rule_config = config['exclude']['rule'][rule]
+            cgnat.add_det44_identity_mapping(
+                ip_address=rule_config.get('local_address'),
+                protocol=protocol_map[rule_config.get('protocol', 'all')],
+                port=int(rule_config.get('local_port', 0)),
+                tag=rule_config.get('description', ''),
+            )
     # Set CGNAT timeouts
     cgnat.set_det44_timeouts(
         icmp=int(config['timeout']['icmp']),
diff --git a/src/op_mode/vpp_nat_cgnat.py b/src/op_mode/vpp_nat_cgnat.py
index 9a7dd2086..0ef0ca410 100644
--- a/src/op_mode/vpp_nat_cgnat.py
+++ b/src/op_mode/vpp_nat_cgnat.py
@@ -120,6 +120,44 @@ def show_interfaces(raw: bool):
 
 
 @_verify
+def show_exclude_rules(raw: bool):
+    """Show CGNAT exclude rules (identity mappings)"""
+    vpp = VPPControl()
+    identity_mappings_dump = vpp.api.det44_identity_mapping_dump()
+    mappings: list[dict] = _get_raw_output(identity_mappings_dump)
+
+    if raw:
+        return mappings
+
+    if not mappings:
+        return "No CGNAT exclude rules configured"
+
+    data_entries = []
+    for m in mappings:
+        proto_map = {0: 'all', 1: 'icmp', 6: 'tcp', 17: 'udp'}
+        proto_name = proto_map.get(m.get('protocol'), str(m.get('protocol')))
+        port_str = str(m.get('port')) if m.get('port') else 'any'
+
+        # Check if address-only (flag & 1)
+        if m.get('flags', 0) & 1:
+            proto_name = 'all'
+            port_str = 'any'
+
+        tag_raw = m.get('tag')
+        if isinstance(tag_raw, bytes):
+            tag = tag_raw.decode('utf-8', errors='replace').rstrip('\x00')
+        else:
+            tag = str(tag_raw) if tag_raw else ''
+
+        values = [m.get('addr'), proto_name, port_str, m.get('vrf_id', 0), tag]
+        data_entries.append(values)
+
+    headers = ['Address', 'Protocol', 'Port', 'VRF', 'Description']
+    out = sorted(data_entries, key=lambda x: x[0])
+    return tabulate(out, headers=headers, tablefmt='simple')
+
+
+@_verify
 def clear_session(address: str, port: str, ext_address: str, ext_port: str):
     vpp = VPPControl()
     vpp.api.det44_close_session_in(