From 07a3b0f5ae87a2ab400390c9fa0ca632d7815e15 Mon Sep 17 00:00:00 2001 From: "Nataliia S." <81954790+natali-rs1985@users.noreply.github.com> Date: Tue, 1 Apr 2025 17:30:12 +0300 Subject: T7283: VPP add static NAT support (#24) * T7283: VPP add static NAT support Add static mapping NAT implementation ``` set vpp nat44 static rule 10 outbound-interface 'eth0' set vpp nat44 static rule 10 inbound-interface 'eth1' set vpp nat44 static rule 10 destination address 192.168.122.10 # optional, if not set outbound interface ip address is used set vpp nat44 static rule 10 destination port 6545 # optional set vpp nat44 static rule 10 protocol tcp|udp|icmp|all # optional, defaults to "all" set vpp nat44 static rule 10 translation address 100.64.0.10 set vpp nat44 static rule 10 translation port 64010 # optional ``` * Improve help strings (Daniil Baturin) --------- Co-authored-by: Daniil Baturin --- python/vyos/vpp/control_vpp.py | 15 ++++++++ python/vyos/vpp/nat/nat44.py | 81 +++++++++++++++++++++++++++++++++++++----- 2 files changed, 88 insertions(+), 8 deletions(-) (limited to 'python') diff --git a/python/vyos/vpp/control_vpp.py b/python/vyos/vpp/control_vpp.py index 2c81b2db2..9d45bbd91 100644 --- a/python/vyos/vpp/control_vpp.py +++ b/python/vyos/vpp/control_vpp.py @@ -160,6 +160,21 @@ class VPPControl: return iface.sw_if_index return None + @_Decorators.api_call + def get_interface_name(self, index: int) -> str | None: + """Find interface name by interface index in VPP + + Args: + index (int): interface index inside VPP + + Returns: + str | None: Interface name or None (if was not found) + """ + for iface in self.__vpp_api_client.api.sw_interface_dump(): + if iface.sw_if_index == index: + return iface.interface_name + return None + @_Decorators.check_retval @_Decorators.api_call def lcp_pair_add( diff --git a/python/vyos/vpp/nat/nat44.py b/python/vyos/vpp/nat/nat44.py index 47d0ced1d..985e5a583 100644 --- a/python/vyos/vpp/nat/nat44.py +++ b/python/vyos/vpp/nat/nat44.py @@ -86,38 +86,34 @@ class Nat44: 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, + flags=0x20, 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, + flags=0x20, 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, + flags=0x10, 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, + flags=0x10, sw_if_index=self.vpp.get_sw_if_index(self.interface_out), is_add=False, ) @@ -149,3 +145,72 @@ class Nat44: def enable_ipfix(self): """Enable NAT44 IPFIX logging""" self.vpp.api.nat44_ei_ipfix_enable_disable(enable=True) + + +class Nat44Static(Nat44): + def __init__(self): + self.vpp = VPPControl() + + def add_inbound_interface(self, interface_in): + self.interface_in = interface_in + self.add_nat44_interface_inside() + + def delete_inbound_interface(self, interface_in): + self.interface_in = interface_in + self.delete_nat44_interface_inside() + + def add_outbound_interface(self, interface_out): + self.interface_out = interface_out + self.add_nat44_interface_outside() + + def delete_outbound_interface(self, interface_out): + self.interface_out = interface_out + self.delete_nat44_interface_outside() + + def add_nat44_static_mapping( + self, + iface_out, + local_ip, + external_ip, + local_port, + external_port, + protocol, + use_iface, + ): + """Add NAT44 static mapping""" + self.vpp.api.nat44_add_del_static_mapping_v2( + local_ip_address=local_ip, + external_ip_address=external_ip, + protocol=protocol, + local_port=local_port, + external_port=external_port, + flags=0x08 if not (protocol or local_port) else 0x00, + external_sw_if_index=( + self.vpp.get_sw_if_index(iface_out) if use_iface else 0xFFFFFFFF + ), + is_add=True, + ) + + def delete_nat44_static_mapping( + self, + iface_out, + local_ip, + external_ip, + local_port, + external_port, + protocol, + use_iface, + ): + """Delete NAT44 static mapping""" + self.vpp.api.nat44_add_del_static_mapping_v2( + local_ip_address=local_ip, + external_ip_address=external_ip, + protocol=protocol, + local_port=local_port, + external_port=external_port, + flags=0x08 if not (protocol or local_port) else 0x00, + external_sw_if_index=( + self.vpp.get_sw_if_index(iface_out) if use_iface else 0xFFFFFFFF + ), + is_add=False, + ) -- cgit v1.2.3