Merge pull request #16 from sever-sever/T7181

T7181: VPP add initial source NAT implentation

2 files changed, 154 insertions, 0 deletions

diff --git a/python/vyos/vpp/nat/__init__.py b/python/vyos/vpp/nat/__init__.py
index e69de29bb..bbd011435 100644
--- a/python/vyos/vpp/nat/__init__.py
+++ b/python/vyos/vpp/nat/__init__.py
@@ -0,0 +1,3 @@
+from .nat44 import Nat44
+
+__all__ = ['Nat44']
diff --git a/python/vyos/vpp/nat/nat44.py b/python/vyos/vpp/nat/nat44.py
new file mode 100644
index 000000000..47d0ced1d
--- /dev/null
+++ b/python/vyos/vpp/nat/nat44.py
@@ -0,0 +1,151 @@
+#
+# 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.vpp import VPPControl
+
+
+class Nat44:
+    def __init__(
+        self,
+        interface_in: str,
+        interface_out: str,
+        translation_pool: str,
+    ):
+        self.interface_in = interface_in
+        self.interface_out = interface_out
+        self.translation_pool = translation_pool
+        self.vpp = VPPControl()
+
+    def enable_nat44_ed(self):
+        """Enable NAT44 endpoint dependent plugin
+        Example:
+            from vyos.vpp.nat import Nat44
+            nat44 = Nat44()
+            nat44.enable_nat44_ed()
+        https://github.com/FDio/vpp/blob/stable/2410/src/plugins/nat/nat44-ed/nat44_ed.api
+        """
+        self.vpp.api.nat44_ed_plugin_enable_disable(enable=True)
+
+    def enable_nat44_ei(self):
+        """Enable NAT44 endpoint independent plugin
+        Example:
+            from vyos.vpp.nat import Nat44
+            nat44 = Nat44()
+            nat44.enable_nat44_ei()
+        """
+        self.vpp.api.nat44_ei_plugin_enable_disable(enable=True)
+
+    def enable_nat44_forwarding(self):
+        """Enable NAT44 forwarding
+        Example:
+            from vyos.vpp.nat import Nat44
+            nat44 = Nat44()
+            nat44.enable_nat44_forwarding()
+        """
+        self.vpp.api.nat44_forwarding_enable_disable(enable=True)
+
+    def disable_nat44_forwarding(self):
+        """Disable NAT44 forwarding
+        Example:
+            from vyos.vpp.nat import Nat44
+            nat44 = Nat44()
+            nat44.disable_nat44_forwarding()
+        """
+        self.vpp.api.nat44_forwarding_enable_disable(enable=False)
+
+    def add_nat44_out_interface(self):
+        """Add NAT44 output interface
+        Example:
+            from vyos.vpp.nat import Nat44
+            nat44 = Nat44('eth0')
+            nat44.add_nat44_out_interface()
+        """
+        self.vpp.api.nat44_ed_add_del_output_interface(
+            sw_if_index=self.vpp.get_sw_if_index(self.interface_out),
+            is_add=True,
+        )
+
+    def delete_nat44_out_interface(self):
+        """Delete NAT44 output interface"""
+        self.vpp.api.nat44_ed_add_del_output_interface(
+            sw_if_index=self.vpp.get_sw_if_index(self.interface_out),
+            is_add=False,
+        )
+
+    # needs to check
+    def add_nat44_interface_inside(self):
+        """Add NAT44 interface"""
+        self.vpp.api.nat44_interface_add_del_feature(
+            flags=1,
+            sw_if_index=self.vpp.get_sw_if_index(self.interface_in),
+            is_add=True,
+        )
+
+    # needs to check
+    def delete_nat44_interface_inside(self):
+        """Delete NAT44 interface"""
+        self.vpp.api.nat44_interface_add_del_feature(
+            flags=1,
+            sw_if_index=self.vpp.get_sw_if_index(self.interface_in),
+            is_add=False,
+        )
+
+    # needs to check
+    def add_nat44_interface_outside(self):
+        """Add NAT44 interface"""
+        self.vpp.api.nat44_interface_add_del_feature(
+            flags=0,
+            sw_if_index=self.vpp.get_sw_if_index(self.interface_out),
+            is_add=True,
+        )
+
+    # needs to check
+    def delete_nat44_interface_outside(self):
+        """Delete NAT44 interface"""
+        self.vpp.api.nat44_interface_add_del_feature(
+            flags=0,
+            sw_if_index=self.vpp.get_sw_if_index(self.interface_out),
+            is_add=False,
+        )
+
+    def add_nat44_address_range(self):
+        """Add NAT44 address range"""
+        if '-' not in self.translation_pool and self.translation_pool != 'masquerade':
+            first_ip_address = last_ip_address = self.translation_pool
+        else:
+            first_ip_address, last_ip_address = self.translation_pool.split('-')
+        self.vpp.api.nat44_add_del_address_range(
+            first_ip_address=first_ip_address,
+            last_ip_address=last_ip_address,
+            is_add=True,
+        )
+
+    def delete_nat44_address_range(self):
+        """Delete NAT44 address range"""
+        if '-' not in self.translation_pool and self.translation_pool != 'masquerade':
+            first_ip_address = last_ip_address = self.translation_pool
+        else:
+            first_ip_address, last_ip_address = self.translation_pool.split('-')
+        self.vpp.api.nat44_add_del_address_range(
+            first_ip_address=first_ip_address,
+            last_ip_address=last_ip_address,
+            is_add=False,
+        )
+
+    def enable_ipfix(self):
+        """Enable NAT44 IPFIX logging"""
+        self.vpp.api.nat44_ei_ipfix_enable_disable(enable=True)