T7390: VPP CGNAT implementation (#30)

CLI:
```
set vpp nat cgnat interface outside <interface>        # multi
set vpp nat cgnat interface inside <interface>         # multi

set vpp nat cgnat rule <rule> outside-prefix <prefix>
set vpp nat cgnat rule <rule> inside-prefix <prefix>

set vpp nat cgnat timeout udp <sec>                # default 300
set vpp nat cgnat timeout tcp-established <sec>    # default 7440
set vpp nat cgnat timeout tcp-transitory <sec>     # default 240
set vpp nat cgnat timeout icmp <sec>               # default 60
```
OP mode:
```
show vpp nat cgnat interfaces
show vpp nat cgnat mappings
show vpp nat cgnat sessions

clear vpp cgnat inside-address <address> port <port> external-address <address> port <port>
```

4 files changed, 363 insertions, 2 deletions

diff --git a/src/conf_mode/vpp.py b/src/conf_mode/vpp.py
index 7f9adc68b..3718f3897 100755
--- a/src/conf_mode/vpp.py
+++ b/src/conf_mode/vpp.py
@@ -138,6 +138,8 @@ def get_config(config=None):
     # NAT dependency
     if conf.exists(['vpp', 'nat44']):
         set_dependents('vpp_nat', conf)
+    if conf.exists(['vpp', 'nat', 'cgnat']):
+        set_dependents('vpp_nat_cgnat', conf)
 
     if not conf.exists(base):
         return {
diff --git a/src/conf_mode/vpp_nat.py b/src/conf_mode/vpp_nat.py
index d38599388..a6bc70032 100644
--- a/src/conf_mode/vpp_nat.py
+++ b/src/conf_mode/vpp_nat.py
@@ -186,7 +186,11 @@ def verify(config):
                     raise ConfigError(
                         f'{interface} must be a VPP interface for "address-pool translation interface"'
                     )
-                iface_address = get_interface_address(interface).get('addr_info', [])[0].get('local')
+                iface_address = (
+                    get_interface_address(interface)
+                    .get('addr_info', [])[0]
+                    .get('local')
+                )
                 addresses_translation.append(iface_address)
 
         if 'twice_nat' in address_pool:
@@ -211,7 +215,11 @@ def verify(config):
                     raise ConfigError(
                         f'{interface} must be a VPP interface for "address-pool twice-nat interface"'
                     )
-                iface_address = get_interface_address(interface).get('addr_info', [])[0].get('local')
+                iface_address = (
+                    get_interface_address(interface)
+                    .get('addr_info', [])[0]
+                    .get('local')
+                )
                 addresses_twice_nat.append(iface_address)
 
     if 'static' in config:
diff --git a/src/conf_mode/vpp_nat_cgnat.py b/src/conf_mode/vpp_nat_cgnat.py
new file mode 100644
index 000000000..05260c476
--- /dev/null
+++ b/src/conf_mode/vpp_nat_cgnat.py
@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2025 VyOS Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# 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, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+
+from vyos import ConfigError
+from vyos.config import Config, config_dict_merge
+from vyos.configdict import node_changed
+from vyos.configdiff import Diff
+from vyos.vpp.utils import cli_ifaces_list
+
+from vyos.vpp.nat.det44 import Det44
+
+
+def get_config(config=None) -> dict:
+    if config:
+        conf = config
+    else:
+        conf = Config()
+
+    base = ['vpp', 'nat', 'cgnat']
+
+    # Get config_dict with default values
+    config = conf.get_config_dict(
+        base,
+        key_mangling=('-', '_'),
+        get_first_key=True,
+        no_tag_node_value_mangle=True,
+    )
+
+    # Get effective config as we need full dictionary to delete
+    effective_config = conf.get_config_dict(
+        base,
+        key_mangling=('-', '_'),
+        effective=True,
+        get_first_key=True,
+        no_tag_node_value_mangle=True,
+    )
+
+    if not config:
+        config['remove'] = True
+
+    # Get default values which we need to conditionally update into the
+    # dictionary retrieved.
+    default_values = conf.get_config_defaults(**config.kwargs, recursive=True)
+    config = config_dict_merge(default_values, config)
+
+    config_changed = node_changed(
+        conf,
+        base,
+        key_mangling=('-', '_'),
+        recursive=True,
+        expand_nodes=Diff.DELETE | Diff.ADD,
+    )
+
+    changed_rules = node_changed(
+        conf,
+        base + ['rule'],
+        key_mangling=('-', '_'),
+        recursive=True,
+        expand_nodes=Diff.DELETE | Diff.ADD,
+    )
+
+    if not config_changed:
+        changed_rules = list(config.get('rule', {}).keys())
+
+    config.update(
+        {
+            'changed_rules': changed_rules,
+            'vpp_ifaces': cli_ifaces_list(conf),
+        }
+    )
+
+    if effective_config:
+        config.update({'effective': effective_config})
+
+    return config
+
+
+def verify(config):
+    if 'remove' in config:
+        return None
+
+    if 'interface' not in config:
+        raise ConfigError('Interfaces must be configured for CGNAT')
+    if 'rule' not in config:
+        raise ConfigError('Rules must be configured for CGNAT')
+
+    required_keys = {'inside', 'outside'}
+    missing_keys = required_keys - set(config['interface'].keys())
+    if missing_keys:
+        raise ConfigError(
+            f'Both inside and outside interfaces must be configured. '
+            f'Please add: {", ".join(missing_keys)}'
+        )
+
+    conflict_ifaces = set(config['interface']['inside']).intersection(
+        set(config['interface']['outside'])
+    )
+    if conflict_ifaces:
+        raise ConfigError(
+            f'Interface cannot be both inside and outside. '
+            f'Please choose a side for: {", ".join(conflict_ifaces)} '
+        )
+
+    for interface in config['interface']['inside']:
+        if interface not in config['vpp_ifaces']:
+            raise ConfigError(
+                f'{interface} must be a VPP interface for inside CGNAT interface'
+            )
+    for interface in config['interface']['outside']:
+        if interface not in config['vpp_ifaces']:
+            raise ConfigError(
+                f'{interface} must be a VPP interface for outside CGNAT interface'
+            )
+
+    required_keys = {'outside_prefix', 'inside_prefix'}
+    for rule in config['rule']:
+        missing_keys = required_keys - set(config['rule'][rule].keys())
+        if missing_keys:
+            raise ConfigError(
+                f'Both inside-prefix and outside-prefix must be configured in rule {rule}. '
+                f'Please add: {", ".join(missing_keys).replace("_", "-")}'
+            )
+
+
+def generate(config):
+    pass
+
+
+def apply(config):
+    cgnat = Det44()
+
+    if 'remove' in config:
+        cgnat.disable_det44_plugin()
+        return None
+
+    if 'effective' in config:
+        remove_config = config.get('effective')
+        # Delete inside interfaces
+        for interface in remove_config['interface']['inside']:
+            if interface not in config.get('interface', {}).get('inside', []):
+                cgnat.delete_det44_interface_inside(interface)
+        # Delete outside interfaces
+        for interface in remove_config['interface']['outside']:
+            if interface not in config.get('interface', {}).get('outside', []):
+                cgnat.delete_det44_interface_outside(interface)
+        # Delete CGNAT rules
+        for rule in config['changed_rules']:
+            if rule in remove_config.get('rule', {}):
+                rule_config = remove_config['rule'][rule]
+                in_addr, in_plen = rule_config['inside_prefix'].split('/')
+                out_addr, out_plen = rule_config['outside_prefix'].split('/')
+                cgnat.delete_det44_mapping(
+                    in_addr=in_addr,
+                    in_plen=int(in_plen),
+                    out_addr=out_addr,
+                    out_plen=int(out_plen),
+                )
+
+    # Add DET44
+    cgnat.enable_det44_plugin()
+    # Add inside interfaces
+    for interface in config['interface']['inside']:
+        cgnat.add_det44_interface_inside(interface)
+    # Add outside interfaces
+    for interface in config['interface']['outside']:
+        cgnat.add_det44_interface_outside(interface)
+    # Add CGNAT rules
+    for rule in config['changed_rules']:
+        if rule in config.get('rule', {}):
+            rule_config = config['rule'][rule]
+            in_addr, in_plen = rule_config['inside_prefix'].split('/')
+            out_addr, out_plen = rule_config['outside_prefix'].split('/')
+            cgnat.add_det44_mapping(
+                in_addr=in_addr,
+                in_plen=int(in_plen),
+                out_addr=out_addr,
+                out_plen=int(out_plen),
+            )
+    # Set CGNAT timeouts
+    cgnat.set_det44_timeouts(
+        icmp=int(config['timeout']['icmp']),
+        udp=int(config['timeout']['udp']),
+        tcp_established=int(config['timeout']['tcp_established']),
+        tcp_transitory=int(config['timeout']['tcp_transitory']),
+    )
+
+
+if __name__ == '__main__':
+    try:
+        c = get_config()
+        verify(c)
+        generate(c)
+        apply(c)
+    except ConfigError as e:
+        print(e)
+        exit(1)
diff --git a/src/op_mode/vpp_nat_cgnat.py b/src/op_mode/vpp_nat_cgnat.py
new file mode 100644
index 000000000..5112c70be
--- /dev/null
+++ b/src/op_mode/vpp_nat_cgnat.py
@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2025 VyOS Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# 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, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+
+import json
+import sys
+from tabulate import tabulate
+
+import vyos.opmode
+from vyos.configquery import ConfigTreeQuery
+
+from vyos.vpp import VPPControl
+
+
+def _verify(func):
+    """Decorator checks if config for VPP NAT CGNAT exists"""
+    from functools import wraps
+
+    @wraps(func)
+    def _wrapper(*args, **kwargs):
+        config = ConfigTreeQuery()
+        base = 'vpp nat cgnat'
+        if not config.exists(base):
+            raise vyos.opmode.UnconfiguredSubsystem(f'{base} is not configured')
+
+        return func(*args, **kwargs)
+
+    return _wrapper
+
+
+def _get_raw_output(data_dump):
+    data = [json.loads(json.dumps(d._asdict(), default=str)) for d in data_dump]
+    return data
+
+
+def _get_formatted_output_interfaces(vpp, interfaces):
+    print('CGNAT interfaces:')
+    for interface in interfaces:
+        name = vpp.get_interface_name(interface['sw_if_index'])
+        iface_type = 'in' if interface['is_inside'] else 'out'
+        print(f'  {name} {iface_type}')
+
+
+def _get_formatted_output_mappings(rules_list):
+    data_entries = []
+    for rule in rules_list:
+        in_addr = rule.get('in_addr')
+        in_plen = str(rule.get('in_plen'))
+        out_addr = rule.get('out_addr')
+        out_plen = str(rule.get('out_plen'))
+        sharing_ratio = rule.get('sharing_ratio')
+        ports_per_host = rule.get('ports_per_host')
+        ses_num = rule.get('ses_num')
+
+        values = [
+            f'{in_addr}/{in_plen}',
+            f'{out_addr}/{out_plen}',
+            sharing_ratio,
+            ports_per_host,
+            ses_num,
+        ]
+        data_entries.append(values)
+    headers = [
+        'Inside',
+        'Outside',
+        'Sharing ratio',
+        'Ports per host',
+        'Sessions',
+    ]
+    out = sorted(data_entries, key=lambda x: x[0])
+    return tabulate(out, headers=headers, tablefmt='simple')
+
+
+@_verify
+def show_sessions(raw: bool):
+    vpp = VPPControl()
+    out = vpp.cli_cmd('show det44 sessions').reply
+    out = out.replace('NAT44 deterministic', 'CGNAT')
+    return out
+
+
+@_verify
+def show_mappings(raw: bool):
+    vpp = VPPControl()
+    nat_static_dump = vpp.api.det44_map_dump()
+    rules_list: list[dict] = _get_raw_output(nat_static_dump)
+
+    if raw:
+        return rules_list
+
+    else:
+        return _get_formatted_output_mappings(rules_list)
+
+
+@_verify
+def show_interfaces(raw: bool):
+    vpp = VPPControl()
+    interfaces_dump = vpp.api.det44_interface_dump()
+    interfaces: list[dict] = _get_raw_output(interfaces_dump)
+
+    if raw:
+        return interfaces
+
+    else:
+        return _get_formatted_output_interfaces(vpp, interfaces)
+
+
+@_verify
+def clear_session(address: str, port: str, ext_address: str, ext_port: str):
+    vpp = VPPControl()
+    vpp.api.det44_close_session_in(
+        in_addr=address,
+        in_port=int(port),
+        ext_addr=ext_address,
+        ext_port=int(ext_port),
+    )
+
+
+if __name__ == '__main__':
+    try:
+        res = vyos.opmode.run(sys.modules[__name__])
+        if res:
+            print(res)
+    except (ValueError, vyos.opmode.Error) as e:
+        print(e)
+        sys.exit(1)