aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/test_nat.py (renamed from test/test_snat.py)874
-rw-r--r--test/vpp_papi_provider.py187
2 files changed, 525 insertions, 536 deletions
diff --git a/test/test_snat.py b/test/test_nat.py
index eb47bbb81d2..0d622b085e6 100644
--- a/test/test_snat.py
+++ b/test/test_nat.py
@@ -18,7 +18,7 @@ from time import sleep
class MethodHolder(VppTestCase):
- """ SNAT create capture and verify method holder """
+ """ NAT create capture and verify method holder """
@classmethod
def setUpClass(cls):
@@ -241,11 +241,11 @@ class MethodHolder(VppTestCase):
Create packet stream for outside network
:param out_if: Outside interface
- :param dst_ip: Destination IP address (Default use global SNAT address)
+ :param dst_ip: Destination IP address (Default use global NAT address)
:param ttl: TTL of generated packets
"""
if dst_ip is None:
- dst_ip = self.snat_addr
+ dst_ip = self.nat_addr
pkts = []
# TCP
p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
@@ -273,13 +273,13 @@ class MethodHolder(VppTestCase):
Verify captured packets on outside network
:param capture: Captured packets
- :param nat_ip: Translated IP address (Default use global SNAT address)
+ :param nat_ip: Translated IP address (Default use global NAT address)
:param same_port: Sorce port number is not translated (Default False)
:param packet_num: Expected number of packets (Default 3)
:param dst_ip: Destination IP address (Default do not verify)
"""
if nat_ip is None:
- nat_ip = self.snat_addr
+ nat_ip = self.nat_addr
self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
@@ -399,13 +399,13 @@ class MethodHolder(VppTestCase):
:param capture: Captured packets
:param src_ip: Translated IP address or IP address of VPP
- (Default use global SNAT address)
+ (Default use global NAT address)
:param packet_num: Expected number of packets (Default 3)
:param icmp_type: Type of error ICMP packet
we are expecting (Default 11)
"""
if src_ip is None:
- src_ip = self.snat_addr
+ src_ip = self.nat_addr
self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
@@ -480,7 +480,7 @@ class MethodHolder(VppTestCase):
# sourceIPv4Address
self.assertEqual(self.pg0.remote_ip4n, record[8])
# postNATSourceIPv4Address
- self.assertEqual(socket.inet_pton(socket.AF_INET, self.snat_addr),
+ self.assertEqual(socket.inet_pton(socket.AF_INET, self.nat_addr),
record[225])
# ingressVRFID
self.assertEqual(struct.pack("!I", 0), record[234])
@@ -518,12 +518,12 @@ class MethodHolder(VppTestCase):
self.assertEqual(struct.pack("!I", 0), record[283])
-class TestSNAT(MethodHolder):
- """ SNAT Test Cases """
+class TestNAT44(MethodHolder):
+ """ NAT44 Test Cases """
@classmethod
def setUpClass(cls):
- super(TestSNAT, cls).setUpClass()
+ super(TestNAT44, cls).setUpClass()
try:
cls.tcp_port_in = 6303
@@ -532,7 +532,7 @@ class TestSNAT(MethodHolder):
cls.udp_port_out = 6304
cls.icmp_id_in = 6305
cls.icmp_id_out = 6305
- cls.snat_addr = '10.0.0.3'
+ cls.nat_addr = '10.0.0.3'
cls.ipfix_src_port = 4739
cls.ipfix_domain_id = 1
@@ -570,12 +570,12 @@ class TestSNAT(MethodHolder):
cls.pg8.admin_up()
except Exception:
- super(TestSNAT, cls).tearDownClass()
+ super(TestNAT44, cls).tearDownClass()
raise
- def clear_snat(self):
+ def clear_nat44(self):
"""
- Clear SNAT configuration.
+ Clear NAT44 configuration.
"""
# I found no elegant way to do this
self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
@@ -600,50 +600,51 @@ class TestSNAT(MethodHolder):
if self.pg7.has_ip4_config:
self.pg7.unconfig_ip4()
- interfaces = self.vapi.snat_interface_addr_dump()
+ interfaces = self.vapi.nat44_interface_addr_dump()
for intf in interfaces:
- self.vapi.snat_add_interface_addr(intf.sw_if_index, is_add=0)
+ self.vapi.nat44_add_interface_addr(intf.sw_if_index, is_add=0)
- self.vapi.snat_ipfix(enable=0, src_port=self.ipfix_src_port,
- domain_id=self.ipfix_domain_id)
+ self.vapi.nat_ipfix(enable=0, src_port=self.ipfix_src_port,
+ domain_id=self.ipfix_domain_id)
self.ipfix_src_port = 4739
self.ipfix_domain_id = 1
- interfaces = self.vapi.snat_interface_dump()
+ interfaces = self.vapi.nat44_interface_dump()
for intf in interfaces:
- self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
- intf.is_inside,
- is_add=0)
+ self.vapi.nat44_interface_add_del_feature(intf.sw_if_index,
+ intf.is_inside,
+ is_add=0)
- interfaces = self.vapi.snat_interface_output_feature_dump()
+ interfaces = self.vapi.nat44_interface_output_feature_dump()
for intf in interfaces:
- self.vapi.snat_interface_add_del_output_feature(intf.sw_if_index,
- intf.is_inside,
- is_add=0)
+ self.vapi.nat44_interface_add_del_output_feature(intf.sw_if_index,
+ intf.is_inside,
+ is_add=0)
- static_mappings = self.vapi.snat_static_mapping_dump()
+ static_mappings = self.vapi.nat44_static_mapping_dump()
for sm in static_mappings:
- self.vapi.snat_add_static_mapping(sm.local_ip_address,
- sm.external_ip_address,
- local_port=sm.local_port,
- external_port=sm.external_port,
- addr_only=sm.addr_only,
- vrf_id=sm.vrf_id,
- protocol=sm.protocol,
- is_add=0)
-
- adresses = self.vapi.snat_address_dump()
+ self.vapi.nat44_add_del_static_mapping(
+ sm.local_ip_address,
+ sm.external_ip_address,
+ local_port=sm.local_port,
+ external_port=sm.external_port,
+ addr_only=sm.addr_only,
+ vrf_id=sm.vrf_id,
+ protocol=sm.protocol,
+ is_add=0)
+
+ adresses = self.vapi.nat44_address_dump()
for addr in adresses:
- self.vapi.snat_add_address_range(addr.ip_address,
- addr.ip_address,
- is_add=0)
-
- def snat_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
- local_port=0, external_port=0, vrf_id=0,
- is_add=1, external_sw_if_index=0xFFFFFFFF,
- proto=0):
+ self.vapi.nat44_add_del_address_range(addr.ip_address,
+ addr.ip_address,
+ is_add=0)
+
+ def nat44_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
+ local_port=0, external_port=0, vrf_id=0,
+ is_add=1, external_sw_if_index=0xFFFFFFFF,
+ proto=0):
"""
- Add/delete S-NAT static mapping
+ Add/delete NAT44 static mapping
:param local_ip: Local IP address
:param external_ip: External IP address
@@ -659,7 +660,7 @@ class TestSNAT(MethodHolder):
addr_only = 0
l_ip = socket.inet_pton(socket.AF_INET, local_ip)
e_ip = socket.inet_pton(socket.AF_INET, external_ip)
- self.vapi.snat_add_static_mapping(
+ self.vapi.nat44_add_del_static_mapping(
l_ip,
e_ip,
external_sw_if_index,
@@ -670,24 +671,24 @@ class TestSNAT(MethodHolder):
proto,
is_add)
- def snat_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF):
+ def nat44_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF):
"""
- Add/delete S-NAT address
+ Add/delete NAT44 address
:param ip: IP address
:param is_add: 1 if add, 0 if delete (Default add)
"""
- snat_addr = socket.inet_pton(socket.AF_INET, ip)
- self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add,
- vrf_id=vrf_id)
+ nat_addr = socket.inet_pton(socket.AF_INET, ip)
+ self.vapi.nat44_add_del_address_range(nat_addr, nat_addr, is_add,
+ vrf_id=vrf_id)
def test_dynamic(self):
- """ SNAT dynamic translation test """
+ """ NAT44 dynamic translation test """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# in2out
pkts = self.create_stream_in(self.pg0, self.pg1)
@@ -706,12 +707,12 @@ class TestSNAT(MethodHolder):
self.verify_capture_in(capture, self.pg0)
def test_dynamic_icmp_errors_in2out_ttl_1(self):
- """ SNAT handling of client packets with TTL=1 """
+ """ NAT44 handling of client packets with TTL=1 """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# Client side - generate traffic
pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
@@ -724,12 +725,12 @@ class TestSNAT(MethodHolder):
self.verify_capture_in_with_icmp_errors(capture, self.pg0)
def test_dynamic_icmp_errors_out2in_ttl_1(self):
- """ SNAT handling of server packets with TTL=1 """
+ """ NAT44 handling of server packets with TTL=1 """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# Client side - create sessions
pkts = self.create_stream_in(self.pg0, self.pg1)
@@ -751,12 +752,12 @@ class TestSNAT(MethodHolder):
src_ip=self.pg1.local_ip4)
def test_dynamic_icmp_errors_in2out_ttl_2(self):
- """ SNAT handling of error responses to client packets with TTL=2 """
+ """ NAT44 handling of error responses to client packets with TTL=2 """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# Client side - generate traffic
pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
@@ -767,7 +768,7 @@ class TestSNAT(MethodHolder):
# Server side - simulate ICMP type 11 response
capture = self.pg1.get_capture(len(pkts))
pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
ICMP(type=11) / packet[IP] for packet in capture]
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
@@ -778,12 +779,12 @@ class TestSNAT(MethodHolder):
self.verify_capture_in_with_icmp_errors(capture, self.pg0)
def test_dynamic_icmp_errors_out2in_ttl_2(self):
- """ SNAT handling of error responses to server packets with TTL=2 """
+ """ NAT44 handling of error responses to server packets with TTL=2 """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# Client side - create sessions
pkts = self.create_stream_in(self.pg0, self.pg1)
@@ -813,12 +814,12 @@ class TestSNAT(MethodHolder):
self.verify_capture_out_with_icmp_errors(capture)
def test_ping_out_interface_from_outside(self):
- """ Ping SNAT out interface from outside network """
+ """ Ping NAT44 out interface from outside network """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) /
@@ -843,14 +844,14 @@ class TestSNAT(MethodHolder):
def test_ping_internal_host_from_outside(self):
""" Ping internal host from outside network """
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# out2in
pkt = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr, ttl=64) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr, ttl=64) /
ICMP(id=self.icmp_id_out, type='echo-request'))
self.pg1.add_stream(pkt)
self.pg_enable_capture(self.pg_interfaces)
@@ -871,17 +872,17 @@ class TestSNAT(MethodHolder):
self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
def test_static_in(self):
- """ SNAT 1:1 NAT initialized from inside network """
+ """ 1:1 NAT initialized from inside network """
nat_ip = "10.0.0.10"
self.tcp_port_out = 6303
self.udp_port_out = 6304
self.icmp_id_out = 6305
- self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# in2out
pkts = self.create_stream_in(self.pg0, self.pg1)
@@ -900,17 +901,17 @@ class TestSNAT(MethodHolder):
self.verify_capture_in(capture, self.pg0)
def test_static_out(self):
- """ SNAT 1:1 NAT initialized from outside network """
+ """ 1:1 NAT initialized from outside network """
nat_ip = "10.0.0.20"
self.tcp_port_out = 6303
self.udp_port_out = 6304
self.icmp_id_out = 6305
- self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# out2in
pkts = self.create_stream_out(self.pg1, nat_ip)
@@ -929,25 +930,25 @@ class TestSNAT(MethodHolder):
self.verify_capture_out(capture, nat_ip, True)
def test_static_with_port_in(self):
- """ SNAT 1:1 NAT with port initialized from inside network """
+ """ 1:1 NAPT initialized from inside network """
self.tcp_port_out = 3606
self.udp_port_out = 3607
self.icmp_id_out = 3608
- self.snat_add_address(self.snat_addr)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.tcp_port_in, self.tcp_port_out,
- proto=IP_PROTOS.tcp)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.udp_port_in, self.udp_port_out,
- proto=IP_PROTOS.udp)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.icmp_id_in, self.icmp_id_out,
- proto=IP_PROTOS.icmp)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.tcp_port_in, self.tcp_port_out,
+ proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.udp_port_in, self.udp_port_out,
+ proto=IP_PROTOS.udp)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.icmp_id_in, self.icmp_id_out,
+ proto=IP_PROTOS.icmp)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# in2out
pkts = self.create_stream_in(self.pg0, self.pg1)
@@ -966,25 +967,25 @@ class TestSNAT(MethodHolder):
self.verify_capture_in(capture, self.pg0)
def test_static_with_port_out(self):
- """ SNAT 1:1 NAT with port initialized from outside network """
+ """ 1:1 NAPT initialized from outside network """
self.tcp_port_out = 30606
self.udp_port_out = 30607
self.icmp_id_out = 30608
- self.snat_add_address(self.snat_addr)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.tcp_port_in, self.tcp_port_out,
- proto=IP_PROTOS.tcp)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.udp_port_in, self.udp_port_out,
- proto=IP_PROTOS.udp)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.icmp_id_in, self.icmp_id_out,
- proto=IP_PROTOS.icmp)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.tcp_port_in, self.tcp_port_out,
+ proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.udp_port_in, self.udp_port_out,
+ proto=IP_PROTOS.udp)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
+ self.icmp_id_in, self.icmp_id_out,
+ proto=IP_PROTOS.icmp)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# out2in
pkts = self.create_stream_out(self.pg1)
@@ -1003,7 +1004,7 @@ class TestSNAT(MethodHolder):
self.verify_capture_out(capture)
def test_static_vrf_aware(self):
- """ SNAT 1:1 NAT VRF awareness """
+ """ 1:1 NAT VRF awareness """
nat_ip1 = "10.0.0.30"
nat_ip2 = "10.0.0.40"
@@ -1011,16 +1012,16 @@ class TestSNAT(MethodHolder):
self.udp_port_out = 6304
self.icmp_id_out = 6305
- self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
- vrf_id=10)
- self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
- vrf_id=10)
- self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
- is_inside=0)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
+ self.nat44_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
+ vrf_id=10)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
+ vrf_id=10)
+ self.vapi.nat44_interface_add_del_feature(self.pg3.sw_if_index,
+ is_inside=0)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg4.sw_if_index)
- # inside interface VRF match SNAT static mapping VRF
+ # inside interface VRF match NAT44 static mapping VRF
pkts = self.create_stream_in(self.pg4, self.pg3)
self.pg4.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
@@ -1028,7 +1029,7 @@ class TestSNAT(MethodHolder):
capture = self.pg3.get_capture(len(pkts))
self.verify_capture_out(capture, nat_ip1, True)
- # inside interface VRF don't match SNAT static mapping VRF (packets
+ # inside interface VRF don't match NAT44 static mapping VRF (packets
# are dropped)
pkts = self.create_stream_in(self.pg0, self.pg3)
self.pg0.add_stream(pkts)
@@ -1037,15 +1038,15 @@ class TestSNAT(MethodHolder):
self.pg3.assert_nothing_captured()
def test_multiple_inside_interfaces(self):
- """ SNAT multiple inside interfaces (non-overlapping address space) """
+ """ NAT44 multiple non-overlapping address space inside interfaces """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg3.sw_if_index,
+ is_inside=0)
- # between two S-NAT inside interfaces (no translation)
+ # between two NAT44 inside interfaces (no translation)
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
@@ -1053,7 +1054,7 @@ class TestSNAT(MethodHolder):
capture = self.pg1.get_capture(len(pkts))
self.verify_capture_no_translation(capture, self.pg0, self.pg1)
- # from S-NAT inside to interface without S-NAT feature (no translation)
+ # from NAT44 inside to interface without NAT44 feature (no translation)
pkts = self.create_stream_in(self.pg0, self.pg2)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
@@ -1094,19 +1095,19 @@ class TestSNAT(MethodHolder):
self.verify_capture_in(capture, self.pg1)
def test_inside_overlapping_interfaces(self):
- """ SNAT multiple inside interfaces with overlapping address space """
+ """ NAT44 multiple inside interfaces with overlapping address space """
static_nat_ip = "10.0.0.10"
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
- is_inside=0)
- self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
- self.snat_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
- vrf_id=20)
-
- # between S-NAT inside interfaces with same VRF (no translation)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg3.sw_if_index,
+ is_inside=0)
+ self.vapi.nat44_interface_add_del_feature(self.pg4.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg5.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg6.sw_if_index)
+ self.nat44_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
+ vrf_id=20)
+
+ # between NAT44 inside interfaces with same VRF (no translation)
pkts = self.create_stream_in(self.pg4, self.pg5)
self.pg4.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
@@ -1114,7 +1115,7 @@ class TestSNAT(MethodHolder):
capture = self.pg5.get_capture(len(pkts))
self.verify_capture_no_translation(capture, self.pg4, self.pg5)
- # between S-NAT inside interfaces with different VRF (hairpinning)
+ # between NAT44 inside interfaces with different VRF (hairpinning)
p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
TCP(sport=1234, dport=5678))
@@ -1126,7 +1127,7 @@ class TestSNAT(MethodHolder):
try:
ip = p[IP]
tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
+ self.assertEqual(ip.src, self.nat_addr)
self.assertEqual(ip.dst, self.pg6.remote_ip4)
self.assertNotEqual(tcp.sport, 1234)
self.assertEqual(tcp.dport, 5678)
@@ -1167,9 +1168,9 @@ class TestSNAT(MethodHolder):
self.verify_capture_in(capture, self.pg5)
# pg5 session dump
- addresses = self.vapi.snat_address_dump()
+ addresses = self.vapi.nat44_address_dump()
self.assertEqual(len(addresses), 1)
- sessions = self.vapi.snat_user_session_dump(self.pg5.remote_ip4n, 10)
+ sessions = self.vapi.nat44_user_session_dump(self.pg5.remote_ip4n, 10)
self.assertEqual(len(sessions), 3)
for session in sessions:
self.assertFalse(session.is_static)
@@ -1204,13 +1205,13 @@ class TestSNAT(MethodHolder):
self.verify_capture_in(capture, self.pg6)
# general user and session dump verifications
- users = self.vapi.snat_user_dump()
+ users = self.vapi.nat44_user_dump()
self.assertTrue(len(users) >= 3)
- addresses = self.vapi.snat_address_dump()
+ addresses = self.vapi.nat44_address_dump()
self.assertEqual(len(addresses), 1)
for user in users:
- sessions = self.vapi.snat_user_session_dump(user.ip_address,
- user.vrf_id)
+ sessions = self.vapi.nat44_user_session_dump(user.ip_address,
+ user.vrf_id)
for session in sessions:
self.assertEqual(user.ip_address, session.inside_ip_address)
self.assertTrue(session.total_bytes > session.total_pkts > 0)
@@ -1219,7 +1220,7 @@ class TestSNAT(MethodHolder):
IP_PROTOS.icmp])
# pg4 session dump
- sessions = self.vapi.snat_user_session_dump(self.pg4.remote_ip4n, 10)
+ sessions = self.vapi.nat44_user_session_dump(self.pg4.remote_ip4n, 10)
self.assertTrue(len(sessions) >= 4)
for session in sessions:
self.assertFalse(session.is_static)
@@ -1229,7 +1230,7 @@ class TestSNAT(MethodHolder):
addresses[0].ip_address)
# pg6 session dump
- sessions = self.vapi.snat_user_session_dump(self.pg6.remote_ip4n, 20)
+ sessions = self.vapi.nat44_user_session_dump(self.pg6.remote_ip4n, 20)
self.assertTrue(len(sessions) >= 3)
for session in sessions:
self.assertTrue(session.is_static)
@@ -1242,7 +1243,7 @@ class TestSNAT(MethodHolder):
self.icmp_id_in])
def test_hairpinning(self):
- """ SNAT hairpinning - 1:1 NAT with port"""
+ """ NAT44 hairpinning - 1:1 NAPT """
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
@@ -1251,18 +1252,18 @@ class TestSNAT(MethodHolder):
server_in_port = 5678
server_out_port = 8765
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# add static mapping for server
- self.snat_add_static_mapping(server.ip4, self.snat_addr,
- server_in_port, server_out_port,
- proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(server.ip4, self.nat_addr,
+ server_in_port, server_out_port,
+ proto=IP_PROTOS.tcp)
# send packet from host to server
p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=self.snat_addr) /
+ IP(src=host.ip4, dst=self.nat_addr) /
TCP(sport=host_in_port, dport=server_out_port))
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
@@ -1272,7 +1273,7 @@ class TestSNAT(MethodHolder):
try:
ip = p[IP]
tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
+ self.assertEqual(ip.src, self.nat_addr)
self.assertEqual(ip.dst, server.ip4)
self.assertNotEqual(tcp.sport, host_in_port)
self.assertEqual(tcp.dport, server_in_port)
@@ -1284,7 +1285,7 @@ class TestSNAT(MethodHolder):
# send reply from server to host
p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
- IP(src=server.ip4, dst=self.snat_addr) /
+ IP(src=server.ip4, dst=self.nat_addr) /
TCP(sport=server_in_port, dport=host_out_port))
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
@@ -1294,7 +1295,7 @@ class TestSNAT(MethodHolder):
try:
ip = p[IP]
tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
+ self.assertEqual(ip.src, self.nat_addr)
self.assertEqual(ip.dst, host.ip4)
self.assertEqual(tcp.sport, server_out_port)
self.assertEqual(tcp.dport, host_in_port)
@@ -1304,7 +1305,7 @@ class TestSNAT(MethodHolder):
raise
def test_hairpinning2(self):
- """ SNAT hairpinning - 1:1 NAT"""
+ """ NAT44 hairpinning - 1:1 NAT"""
server1_nat_ip = "10.0.0.10"
server2_nat_ip = "10.0.0.11"
@@ -1314,14 +1315,14 @@ class TestSNAT(MethodHolder):
server_tcp_port = 22
server_udp_port = 20
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# add static mapping for servers
- self.snat_add_static_mapping(server1.ip4, server1_nat_ip)
- self.snat_add_static_mapping(server2.ip4, server2_nat_ip)
+ self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
+ self.nat44_add_static_mapping(server2.ip4, server2_nat_ip)
# host to server1
pkts = []
@@ -1343,7 +1344,7 @@ class TestSNAT(MethodHolder):
capture = self.pg0.get_capture(len(pkts))
for packet in capture:
try:
- self.assertEqual(packet[IP].src, self.snat_addr)
+ self.assertEqual(packet[IP].src, self.nat_addr)
self.assertEqual(packet[IP].dst, server1.ip4)
if packet.haslayer(TCP):
self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
@@ -1364,15 +1365,15 @@ class TestSNAT(MethodHolder):
# server1 to host
pkts = []
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.snat_addr) /
+ IP(src=server1.ip4, dst=self.nat_addr) /
TCP(sport=server_tcp_port, dport=self.tcp_port_out))
pkts.append(p)
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.snat_addr) /
+ IP(src=server1.ip4, dst=self.nat_addr) /
UDP(sport=server_udp_port, dport=self.udp_port_out))
pkts.append(p)
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.snat_addr) /
+ IP(src=server1.ip4, dst=self.nat_addr) /
ICMP(id=self.icmp_id_out, type='echo-reply'))
pkts.append(p)
self.pg0.add_stream(pkts)
@@ -1472,16 +1473,16 @@ class TestSNAT(MethodHolder):
def test_max_translations_per_user(self):
""" MAX translations per user - recycle the least recently used """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# get maximum number of translations per user
- snat_config = self.vapi.snat_show_config()
+ nat44_config = self.vapi.nat_show_config()
# send more than maximum number of translations per user packets
- pkts_num = snat_config.max_translations_per_user + 5
+ pkts_num = nat44_config.max_translations_per_user + 5
pkts = []
for port in range(0, pkts_num):
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
@@ -1496,39 +1497,40 @@ class TestSNAT(MethodHolder):
self.pg1.get_capture(pkts_num)
def test_interface_addr(self):
- """ Acquire SNAT addresses from interface """
- self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
+ """ Acquire NAT44 addresses from interface """
+ self.vapi.nat44_add_interface_addr(self.pg7.sw_if_index)
# no address in NAT pool
- adresses = self.vapi.snat_address_dump()
+ adresses = self.vapi.nat44_address_dump()
self.assertEqual(0, len(adresses))
# configure interface address and check NAT address pool
self.pg7.config_ip4()
- adresses = self.vapi.snat_address_dump()
+ adresses = self.vapi.nat44_address_dump()
self.assertEqual(1, len(adresses))
self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n)
# remove interface address and check NAT address pool
self.pg7.unconfig_ip4()
- adresses = self.vapi.snat_address_dump()
+ adresses = self.vapi.nat44_address_dump()
self.assertEqual(0, len(adresses))
def test_interface_addr_static_mapping(self):
""" Static mapping with addresses from interface """
- self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
- self.snat_add_static_mapping('1.2.3.4',
- external_sw_if_index=self.pg7.sw_if_index)
+ self.vapi.nat44_add_interface_addr(self.pg7.sw_if_index)
+ self.nat44_add_static_mapping(
+ '1.2.3.4',
+ external_sw_if_index=self.pg7.sw_if_index)
# static mappings with external interface
- static_mappings = self.vapi.snat_static_mapping_dump()
+ static_mappings = self.vapi.nat44_static_mapping_dump()
self.assertEqual(1, len(static_mappings))
self.assertEqual(self.pg7.sw_if_index,
static_mappings[0].external_sw_if_index)
# configure interface address and check static mappings
self.pg7.config_ip4()
- static_mappings = self.vapi.snat_static_mapping_dump()
+ static_mappings = self.vapi.nat44_static_mapping_dump()
self.assertEqual(1, len(static_mappings))
self.assertEqual(static_mappings[0].external_ip_address[0:4],
self.pg7.local_ip4n)
@@ -1536,26 +1538,26 @@ class TestSNAT(MethodHolder):
# remove interface address and check static mappings
self.pg7.unconfig_ip4()
- static_mappings = self.vapi.snat_static_mapping_dump()
+ static_mappings = self.vapi.nat44_static_mapping_dump()
self.assertEqual(0, len(static_mappings))
def test_ipfix_nat44_sess(self):
- """ S-NAT IPFIX logging NAT44 session created/delted """
+ """ IPFIX logging NAT44 session created/delted """
self.ipfix_domain_id = 10
self.ipfix_src_port = 20202
colector_port = 30303
bind_layers(UDP, IPFIX, dport=30303)
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
src_address=self.pg3.local_ip4n,
path_mtu=512,
template_interval=10,
collector_port=colector_port)
- self.vapi.snat_ipfix(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port)
+ self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port)
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
@@ -1563,7 +1565,7 @@ class TestSNAT(MethodHolder):
self.pg_start()
capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture)
- self.snat_add_address(self.snat_addr, is_add=0)
+ self.nat44_add_address(self.nat_addr, is_add=0)
self.vapi.cli("ipfix flush") # FIXME this should be an API call
capture = self.pg3.get_capture(3)
ipfix = IPFIXDecoder()
@@ -1585,16 +1587,16 @@ class TestSNAT(MethodHolder):
self.verify_ipfix_nat44_ses(data)
def test_ipfix_addr_exhausted(self):
- """ S-NAT IPFIX logging NAT addresses exhausted """
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ """ IPFIX logging NAT addresses exhausted """
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
src_address=self.pg3.local_ip4n,
path_mtu=512,
template_interval=10)
- self.vapi.snat_ipfix(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port)
+ self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port)
p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
@@ -1624,17 +1626,17 @@ class TestSNAT(MethodHolder):
self.verify_ipfix_addr_exhausted(data)
def test_pool_addr_fib(self):
- """ S-NAT add pool addresses to FIB """
+ """ NAT44 add pool addresses to FIB """
static_addr = '10.0.0.10'
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
- self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr)
- # SNAT address
+ # NAT44 address
p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=self.snat_addr,
+ ARP(op=ARP.who_has, pdst=self.nat_addr,
psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
@@ -1654,9 +1656,9 @@ class TestSNAT(MethodHolder):
self.assertTrue(capture[0].haslayer(ARP))
self.assertTrue(capture[0][ARP].op, ARP.is_at)
- # send ARP to non-SNAT interface
+ # send ARP to non-NAT44 interface
p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=self.snat_addr,
+ ARP(op=ARP.who_has, pdst=self.nat_addr,
psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
self.pg2.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
@@ -1664,12 +1666,12 @@ class TestSNAT(MethodHolder):
capture = self.pg1.get_capture(0)
# remove addresses and verify
- self.snat_add_address(self.snat_addr, is_add=0)
- self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr,
- is_add=0)
+ self.nat44_add_address(self.nat_addr, is_add=0)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr,
+ is_add=0)
p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=self.snat_addr,
+ ARP(op=ARP.who_has, pdst=self.nat_addr,
psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
@@ -1685,7 +1687,7 @@ class TestSNAT(MethodHolder):
capture = self.pg1.get_capture(0)
def test_vrf_mode(self):
- """ S-NAT tenant VRF aware address pool mode """
+ """ NAT44 tenant VRF aware address pool mode """
vrf_id1 = 1
vrf_id2 = 2
@@ -1699,12 +1701,12 @@ class TestSNAT(MethodHolder):
self.pg0.config_ip4()
self.pg1.config_ip4()
- self.snat_add_address(nat_ip1, vrf_id=vrf_id1)
- self.snat_add_address(nat_ip2, vrf_id=vrf_id2)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(nat_ip1, vrf_id=vrf_id1)
+ self.nat44_add_address(nat_ip2, vrf_id=vrf_id2)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg2.sw_if_index,
+ is_inside=0)
# first VRF
pkts = self.create_stream_in(self.pg0, self.pg2)
@@ -1723,17 +1725,17 @@ class TestSNAT(MethodHolder):
self.verify_capture_out(capture, nat_ip2)
def test_vrf_feature_independent(self):
- """ S-NAT tenant VRF independent address pool mode """
+ """ NAT44 tenant VRF independent address pool mode """
nat_ip1 = "10.0.0.10"
nat_ip2 = "10.0.0.11"
- self.snat_add_address(nat_ip1)
- self.snat_add_address(nat_ip2)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(nat_ip1)
+ self.nat44_add_address(nat_ip2)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg2.sw_if_index,
+ is_inside=0)
# first VRF
pkts = self.create_stream_in(self.pg0, self.pg2)
@@ -1752,7 +1754,7 @@ class TestSNAT(MethodHolder):
self.verify_capture_out(capture, nat_ip1)
def test_dynamic_ipless_interfaces(self):
- """ SNAT interfaces without configured ip dynamic map """
+ """ NAT44 interfaces without configured IP address """
self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
self.pg7.remote_mac,
@@ -1772,10 +1774,10 @@ class TestSNAT(MethodHolder):
next_hop_address=self.pg8.remote_ip4n,
next_hop_sw_if_index=self.pg8.sw_if_index)
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg7.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg8.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index,
+ is_inside=0)
# in2out
pkts = self.create_stream_in(self.pg7, self.pg8)
@@ -1786,7 +1788,7 @@ class TestSNAT(MethodHolder):
self.verify_capture_out(capture)
# out2in
- pkts = self.create_stream_out(self.pg8, self.snat_addr)
+ pkts = self.create_stream_out(self.pg8, self.nat_addr)
self.pg8.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
@@ -1794,7 +1796,7 @@ class TestSNAT(MethodHolder):
self.verify_capture_in(capture, self.pg7)
def test_static_ipless_interfaces(self):
- """ SNAT 1:1 NAT interfaces without configured ip """
+ """ NAT44 interfaces without configured IP address - 1:1 NAT """
self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
self.pg7.remote_mac,
@@ -1814,10 +1816,10 @@ class TestSNAT(MethodHolder):
next_hop_address=self.pg8.remote_ip4n,
next_hop_sw_if_index=self.pg8.sw_if_index)
- self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg7.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg8.sw_if_index,
- is_inside=0)
+ self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index,
+ is_inside=0)
# out2in
pkts = self.create_stream_out(self.pg8)
@@ -1833,10 +1835,10 @@ class TestSNAT(MethodHolder):
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg8.get_capture(len(pkts))
- self.verify_capture_out(capture, self.snat_addr, True)
+ self.verify_capture_out(capture, self.nat_addr, True)
def test_static_with_port_ipless_interfaces(self):
- """ SNAT 1:1 NAT with port interfaces without configured ip """
+ """ NAT44 interfaces without configured IP address - 1:1 NAPT """
self.tcp_port_out = 30606
self.udp_port_out = 30607
@@ -1860,19 +1862,19 @@ class TestSNAT(MethodHolder):
next_hop_address=self.pg8.remote_ip4n,
next_hop_sw_if_index=self.pg8.sw_if_index)
- self.snat_add_address(self.snat_addr)
- self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr,
- self.tcp_port_in, self.tcp_port_out,
- proto=IP_PROTOS.tcp)
- self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr,
- self.udp_port_in, self.udp_port_out,
- proto=IP_PROTOS.udp)
- self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr,
- self.icmp_id_in, self.icmp_id_out,
- proto=IP_PROTOS.icmp)
- self.vapi.snat_interface_add_del_feature(self.pg7.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg8.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
+ self.tcp_port_in, self.tcp_port_out,
+ proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
+ self.udp_port_in, self.udp_port_out,
+ proto=IP_PROTOS.udp)
+ self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
+ self.icmp_id_in, self.icmp_id_out,
+ proto=IP_PROTOS.icmp)
+ self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index,
+ is_inside=0)
# out2in
pkts = self.create_stream_out(self.pg8)
@@ -1893,10 +1895,10 @@ class TestSNAT(MethodHolder):
def test_static_unknown_proto(self):
""" 1:1 NAT translate packet with unknown protocol """
nat_ip = "10.0.0.10"
- self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# in2out
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
@@ -1947,11 +1949,11 @@ class TestSNAT(MethodHolder):
host_nat_ip = "10.0.0.10"
server_nat_ip = "10.0.0.11"
- self.snat_add_static_mapping(host.ip4, host_nat_ip)
- self.snat_add_static_mapping(server.ip4, server_nat_ip)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_static_mapping(host.ip4, host_nat_ip)
+ self.nat44_add_static_mapping(server.ip4, server_nat_ip)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# host to server
p = (Ether(dst=self.pg0.local_mac, src=host.mac) /
@@ -1994,11 +1996,11 @@ class TestSNAT(MethodHolder):
raise
def test_unknown_proto(self):
- """ SNAT translate packet with unknown protocol """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ """ NAT44 translate packet with unknown protocol """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# in2out
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
@@ -2020,7 +2022,7 @@ class TestSNAT(MethodHolder):
p = self.pg1.get_capture(1)
packet = p[0]
try:
- self.assertEqual(packet[IP].src, self.snat_addr)
+ self.assertEqual(packet[IP].src, self.nat_addr)
self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
self.assertTrue(packet.haslayer(GRE))
self.check_ip_checksum(packet)
@@ -2030,7 +2032,7 @@ class TestSNAT(MethodHolder):
# out2in
p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
GRE() /
IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
TCP(sport=1234, dport=1234))
@@ -2049,7 +2051,7 @@ class TestSNAT(MethodHolder):
raise
def test_hairpinning_unknown_proto(self):
- """ SNAT translate packet with unknown protocol - hairpinning """
+ """ NAT44 translate packet with unknown protocol - hairpinning """
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
host_in_port = 1234
@@ -2058,13 +2060,13 @@ class TestSNAT(MethodHolder):
server_out_port = 8765
server_nat_ip = "10.0.0.11"
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# add static mapping for server
- self.snat_add_static_mapping(server.ip4, server_nat_ip)
+ self.nat44_add_static_mapping(server.ip4, server_nat_ip)
# host to server
p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
@@ -2086,7 +2088,7 @@ class TestSNAT(MethodHolder):
p = self.pg0.get_capture(1)
packet = p[0]
try:
- self.assertEqual(packet[IP].src, self.snat_addr)
+ self.assertEqual(packet[IP].src, self.nat_addr)
self.assertEqual(packet[IP].dst, server.ip4)
self.assertTrue(packet.haslayer(GRE))
self.check_ip_checksum(packet)
@@ -2096,7 +2098,7 @@ class TestSNAT(MethodHolder):
# server to host
p = (Ether(dst=self.pg0.local_mac, src=server.mac) /
- IP(src=server.ip4, dst=self.snat_addr) /
+ IP(src=server.ip4, dst=self.nat_addr) /
GRE() /
IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
TCP(sport=1234, dport=1234))
@@ -2115,11 +2117,11 @@ class TestSNAT(MethodHolder):
raise
def test_output_feature(self):
- """ S-NAT interface output feature (in2out postrouting) """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_output_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_output_feature(self.pg1.sw_if_index,
- is_inside=0)
+ """ NAT44 interface output feature (in2out postrouting) """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_output_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index,
+ is_inside=0)
# in2out
pkts = self.create_stream_in(self.pg0, self.pg1)
@@ -2138,7 +2140,7 @@ class TestSNAT(MethodHolder):
self.verify_capture_in(capture, self.pg0)
def test_output_feature_vrf_aware(self):
- """ S-NAT interface output feature VRF aware (in2out postrouting) """
+ """ NAT44 interface output feature VRF aware (in2out postrouting) """
nat_ip_vrf10 = "10.0.0.10"
nat_ip_vrf20 = "10.0.0.20"
@@ -2153,12 +2155,12 @@ class TestSNAT(MethodHolder):
next_hop_sw_if_index=self.pg3.sw_if_index,
table_id=20)
- self.snat_add_address(nat_ip_vrf10, vrf_id=10)
- self.snat_add_address(nat_ip_vrf20, vrf_id=20)
- self.vapi.snat_interface_add_del_output_feature(self.pg4.sw_if_index)
- self.vapi.snat_interface_add_del_output_feature(self.pg6.sw_if_index)
- self.vapi.snat_interface_add_del_output_feature(self.pg3.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(nat_ip_vrf10, vrf_id=10)
+ self.nat44_add_address(nat_ip_vrf20, vrf_id=20)
+ self.vapi.nat44_interface_add_del_output_feature(self.pg4.sw_if_index)
+ self.vapi.nat44_interface_add_del_output_feature(self.pg6.sw_if_index)
+ self.vapi.nat44_interface_add_del_output_feature(self.pg3.sw_if_index,
+ is_inside=0)
# in2out VRF 10
pkts = self.create_stream_in(self.pg4, self.pg3)
@@ -2193,7 +2195,7 @@ class TestSNAT(MethodHolder):
self.verify_capture_in(capture, self.pg6)
def test_output_feature_hairpinning(self):
- """ S-NAT interface output feature hairpinning (in2out postrouting) """
+ """ NAT44 interface output feature hairpinning (in2out postrouting) """
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
host_in_port = 1234
@@ -2201,19 +2203,19 @@ class TestSNAT(MethodHolder):
server_in_port = 5678
server_out_port = 8765
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_output_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_output_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_output_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index,
+ is_inside=0)
# add static mapping for server
- self.snat_add_static_mapping(server.ip4, self.snat_addr,
- server_in_port, server_out_port,
- proto=IP_PROTOS.tcp)
+ self.nat44_add_static_mapping(server.ip4, self.nat_addr,
+ server_in_port, server_out_port,
+ proto=IP_PROTOS.tcp)
# send packet from host to server
p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=self.snat_addr) /
+ IP(src=host.ip4, dst=self.nat_addr) /
TCP(sport=host_in_port, dport=server_out_port))
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
@@ -2223,7 +2225,7 @@ class TestSNAT(MethodHolder):
try:
ip = p[IP]
tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
+ self.assertEqual(ip.src, self.nat_addr)
self.assertEqual(ip.dst, server.ip4)
self.assertNotEqual(tcp.sport, host_in_port)
self.assertEqual(tcp.dport, server_in_port)
@@ -2235,7 +2237,7 @@ class TestSNAT(MethodHolder):
# send reply from server to host
p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
- IP(src=server.ip4, dst=self.snat_addr) /
+ IP(src=server.ip4, dst=self.nat_addr) /
TCP(sport=server_in_port, dport=host_out_port))
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
@@ -2245,7 +2247,7 @@ class TestSNAT(MethodHolder):
try:
ip = p[IP]
tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
+ self.assertEqual(ip.src, self.nat_addr)
self.assertEqual(ip.dst, host.ip4)
self.assertEqual(tcp.sport, server_out_port)
self.assertEqual(tcp.dport, host_in_port)
@@ -2255,10 +2257,10 @@ class TestSNAT(MethodHolder):
raise
def tearDown(self):
- super(TestSNAT, self).tearDown()
+ super(TestNAT44, self).tearDown()
if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show snat verbose"))
- self.clear_snat()
+ self.logger.info(self.vapi.cli("show nat44 verbose"))
+ self.clear_nat44()
class TestDeterministicNAT(MethodHolder):
@@ -2267,7 +2269,7 @@ class TestDeterministicNAT(MethodHolder):
@classmethod
def setUpConstants(cls):
super(TestDeterministicNAT, cls).setUpConstants()
- cls.vpp_cmdline.extend(["snat", "{", "deterministic", "}"])
+ cls.vpp_cmdline.extend(["nat", "{", "deterministic", "}"])
@classmethod
def setUpClass(cls):
@@ -2279,7 +2281,7 @@ class TestDeterministicNAT(MethodHolder):
cls.udp_port_in = 6304
cls.udp_external_port = 6304
cls.icmp_id_in = 6305
- cls.snat_addr = '10.0.0.3'
+ cls.nat_addr = '10.0.0.3'
cls.create_pg_interfaces(range(3))
cls.interfaces = list(cls.pg_interfaces)
@@ -2330,11 +2332,11 @@ class TestDeterministicNAT(MethodHolder):
Create packet stream for outside network
:param out_if: Outside interface
- :param dst_ip: Destination IP address (Default use global SNAT address)
+ :param dst_ip: Destination IP address (Default use global NAT address)
:param ttl: TTL of generated packets
"""
if dst_ip is None:
- dst_ip = self.snat_addr
+ dst_ip = self.nat_addr
pkts = []
# TCP
p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
@@ -2361,12 +2363,12 @@ class TestDeterministicNAT(MethodHolder):
Verify captured packets on outside network
:param capture: Captured packets
- :param nat_ip: Translated IP address (Default use global SNAT address)
+ :param nat_ip: Translated IP address (Default use global NAT address)
:param same_port: Sorce port number is not translated (Default False)
:param packet_num: Expected number of packets (Default 3)
"""
if nat_ip is None:
- nat_ip = self.snat_addr
+ nat_ip = self.nat_addr
self.assertEqual(packet_num, len(capture))
for packet in capture:
try:
@@ -2404,7 +2406,7 @@ class TestDeterministicNAT(MethodHolder):
# SYN + ACK packet out->in
p = (Ether(src=out_if.remote_mac, dst=out_if.local_mac) /
- IP(src=out_if.remote_ip4, dst=self.snat_addr) /
+ IP(src=out_if.remote_ip4, dst=self.nat_addr) /
TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
flags="SA"))
out_if.add_stream(p)
@@ -2442,7 +2444,7 @@ class TestDeterministicNAT(MethodHolder):
self.assertEqual(self.pg0.remote_ip4n, record[8])
def test_deterministic_mode(self):
- """ S-NAT run deterministic mode """
+ """ NAT plugin run deterministic mode """
in_addr = '172.16.255.0'
out_addr = '172.17.255.50'
in_addr_t = '172.16.255.20'
@@ -2452,17 +2454,17 @@ class TestDeterministicNAT(MethodHolder):
in_plen = 24
out_plen = 32
- snat_config = self.vapi.snat_show_config()
- self.assertEqual(1, snat_config.deterministic)
+ nat_config = self.vapi.nat_show_config()
+ self.assertEqual(1, nat_config.deterministic)
- self.vapi.snat_add_det_map(in_addr_n, in_plen, out_addr_n, out_plen)
+ self.vapi.nat_det_add_del_map(in_addr_n, in_plen, out_addr_n, out_plen)
- rep1 = self.vapi.snat_det_forward(in_addr_t_n)
+ rep1 = self.vapi.nat_det_forward(in_addr_t_n)
self.assertEqual(rep1.out_addr[:4], out_addr_n)
- rep2 = self.vapi.snat_det_reverse(out_addr_n, rep1.out_port_hi)
+ rep2 = self.vapi.nat_det_reverse(out_addr_n, rep1.out_port_hi)
self.assertEqual(rep2.in_addr[:4], in_addr_t_n)
- deterministic_mappings = self.vapi.snat_det_map_dump()
+ deterministic_mappings = self.vapi.nat_det_map_dump()
self.assertEqual(len(deterministic_mappings), 1)
dsm = deterministic_mappings[0]
self.assertEqual(in_addr_n, dsm.in_addr[:4])
@@ -2470,20 +2472,20 @@ class TestDeterministicNAT(MethodHolder):
self.assertEqual(out_addr_n, dsm.out_addr[:4])
self.assertEqual(out_plen, dsm.out_plen)
- self.clear_snat()
- deterministic_mappings = self.vapi.snat_det_map_dump()
+ self.clear_nat_det()
+ deterministic_mappings = self.vapi.nat_det_map_dump()
self.assertEqual(len(deterministic_mappings), 0)
def test_set_timeouts(self):
""" Set deterministic NAT timeouts """
- timeouts_before = self.vapi.snat_det_get_timeouts()
+ timeouts_before = self.vapi.nat_det_get_timeouts()
- self.vapi.snat_det_set_timeouts(timeouts_before.udp + 10,
- timeouts_before.tcp_established + 10,
- timeouts_before.tcp_transitory + 10,
- timeouts_before.icmp + 10)
+ self.vapi.nat_det_set_timeouts(timeouts_before.udp + 10,
+ timeouts_before.tcp_established + 10,
+ timeouts_before.tcp_transitory + 10,
+ timeouts_before.icmp + 10)
- timeouts_after = self.vapi.snat_det_get_timeouts()
+ timeouts_after = self.vapi.nat_det_get_timeouts()
self.assertNotEqual(timeouts_before.udp, timeouts_after.udp)
self.assertNotEqual(timeouts_before.icmp, timeouts_after.icmp)
@@ -2493,17 +2495,17 @@ class TestDeterministicNAT(MethodHolder):
timeouts_after.tcp_transitory)
def test_det_in(self):
- """ CGNAT translation test (TCP, UDP, ICMP) """
+ """ Deterministic NAT translation test (TCP, UDP, ICMP) """
nat_ip = "10.0.0.10"
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(nat_ip),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
+ 32,
+ socket.inet_aton(nat_ip),
+ 32)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# in2out
pkts = self.create_stream_in(self.pg0, self.pg1)
@@ -2522,7 +2524,7 @@ class TestDeterministicNAT(MethodHolder):
self.verify_capture_in(capture, self.pg0)
# session dump test
- sessions = self.vapi.snat_det_session_dump(self.pg0.remote_ip4n)
+ sessions = self.vapi.nat_det_session_dump(self.pg0.remote_ip4n)
self.assertEqual(len(sessions), 3)
# TCP session
@@ -2546,7 +2548,7 @@ class TestDeterministicNAT(MethodHolder):
self.assertEqual(s.out_port, self.icmp_external_id)
def test_multiple_users(self):
- """ CGNAT multiple users """
+ """ Deterministic NAT multiple users """
nat_ip = "10.0.0.10"
port_in = 80
@@ -2555,13 +2557,13 @@ class TestDeterministicNAT(MethodHolder):
host0 = self.pg0.remote_hosts[0]
host1 = self.pg0.remote_hosts[1]
- self.vapi.snat_add_det_map(host0.ip4n,
- 24,
- socket.inet_aton(nat_ip),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ self.vapi.nat_det_add_del_map(host0.ip4n,
+ 24,
+ socket.inet_aton(nat_ip),
+ 32)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
# host0 to out
p = (Ether(src=host0.mac, dst=self.pg0.local_mac) /
@@ -2603,7 +2605,7 @@ class TestDeterministicNAT(MethodHolder):
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- dms = self.vapi.snat_det_map_dump()
+ dms = self.vapi.nat_det_map_dump()
self.assertEqual(1, len(dms))
self.assertEqual(2, dms[0].ses_num)
@@ -2648,29 +2650,29 @@ class TestDeterministicNAT(MethodHolder):
raise
# session close api test
- self.vapi.snat_det_close_session_out(socket.inet_aton(nat_ip),
- port_out1,
- self.pg1.remote_ip4n,
- external_port)
- dms = self.vapi.snat_det_map_dump()
- self.assertEqual(dms[0].ses_num, 1)
-
- self.vapi.snat_det_close_session_in(host0.ip4n,
- port_in,
+ self.vapi.nat_det_close_session_out(socket.inet_aton(nat_ip),
+ port_out1,
self.pg1.remote_ip4n,
external_port)
- dms = self.vapi.snat_det_map_dump()
+ dms = self.vapi.nat_det_map_dump()
+ self.assertEqual(dms[0].ses_num, 1)
+
+ self.vapi.nat_det_close_session_in(host0.ip4n,
+ port_in,
+ self.pg1.remote_ip4n,
+ external_port)
+ dms = self.vapi.nat_det_map_dump()
self.assertEqual(dms[0].ses_num, 0)
def test_tcp_session_close_detection_in(self):
- """ CGNAT TCP session close initiated from inside network """
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(self.snat_addr),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ """ Deterministic NAT TCP session close from inside network """
+ self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
+ 32,
+ socket.inet_aton(self.nat_addr),
+ 32)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
self.initiate_tcp_session(self.pg0, self.pg1)
@@ -2690,14 +2692,14 @@ class TestDeterministicNAT(MethodHolder):
# ACK packet out -> in
p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
flags="A"))
pkts.append(p)
# FIN packet out -> in
p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
flags="F"))
pkts.append(p)
@@ -2717,22 +2719,22 @@ class TestDeterministicNAT(MethodHolder):
self.pg_start()
self.pg1.get_capture(1)
- # Check if snat closed the session
- dms = self.vapi.snat_det_map_dump()
+ # Check if deterministic NAT44 closed the session
+ dms = self.vapi.nat_det_map_dump()
self.assertEqual(0, dms[0].ses_num)
except:
self.logger.error("TCP session termination failed")
raise
def test_tcp_session_close_detection_out(self):
- """ CGNAT TCP session close initiated from outside network """
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(self.snat_addr),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ """ Deterministic NAT TCP session close from outside network """
+ self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
+ 32,
+ socket.inet_aton(self.nat_addr),
+ 32)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
self.initiate_tcp_session(self.pg0, self.pg1)
@@ -2740,7 +2742,7 @@ class TestDeterministicNAT(MethodHolder):
try:
# FIN packet out -> in
p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
flags="F"))
self.pg1.add_stream(p)
@@ -2771,7 +2773,7 @@ class TestDeterministicNAT(MethodHolder):
# ACK packet out -> in
p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
flags="A"))
self.pg1.add_stream(p)
@@ -2779,8 +2781,8 @@ class TestDeterministicNAT(MethodHolder):
self.pg_start()
self.pg0.get_capture(1)
- # Check if snat closed the session
- dms = self.vapi.snat_det_map_dump()
+ # Check if deterministic NAT44 closed the session
+ dms = self.vapi.nat_det_map_dump()
self.assertEqual(0, dms[0].ses_num)
except:
self.logger.error("TCP session termination failed")
@@ -2788,17 +2790,17 @@ class TestDeterministicNAT(MethodHolder):
@unittest.skipUnless(running_extended_tests(), "part of extended tests")
def test_session_timeout(self):
- """ CGNAT session timeouts """
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(self.snat_addr),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ """ Deterministic NAT session timeouts """
+ self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
+ 32,
+ socket.inet_aton(self.nat_addr),
+ 32)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
self.initiate_tcp_session(self.pg0, self.pg1)
- self.vapi.snat_det_set_timeouts(5, 5, 5, 5)
+ self.vapi.nat_det_set_timeouts(5, 5, 5, 5)
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
@@ -2806,24 +2808,24 @@ class TestDeterministicNAT(MethodHolder):
capture = self.pg1.get_capture(len(pkts))
sleep(15)
- dms = self.vapi.snat_det_map_dump()
+ dms = self.vapi.nat_det_map_dump()
self.assertEqual(0, dms[0].ses_num)
@unittest.skipUnless(running_extended_tests(), "part of extended tests")
def test_session_limit_per_user(self):
- """ CGNAT maximum 1000 sessions per user should be created """
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(self.snat_addr),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
+ """ Deterministic NAT maximum sessions per user limit """
+ self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
+ 32,
+ socket.inet_aton(self.nat_addr),
+ 32)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=0)
self.vapi.set_ipfix_exporter(collector_address=self.pg2.remote_ip4n,
src_address=self.pg2.local_ip4n,
path_mtu=512,
template_interval=10)
- self.vapi.snat_ipfix()
+ self.vapi.nat_ipfix()
pkts = []
for port in range(1025, 2025):
@@ -2857,7 +2859,7 @@ class TestDeterministicNAT(MethodHolder):
self.assertEqual(inner_ip[UDPerror].sport, 3001)
self.assertEqual(inner_ip[UDPerror].dport, 3002)
- dms = self.vapi.snat_det_map_dump()
+ dms = self.vapi.nat_det_map_dump()
self.assertEqual(1000, dms[0].ses_num)
@@ -2877,31 +2879,31 @@ class TestDeterministicNAT(MethodHolder):
data = ipfix.decode_data_set(p.getlayer(Set))
self.verify_ipfix_max_entries_per_user(data)
- def clear_snat(self):
+ def clear_nat_det(self):
"""
- Clear SNAT configuration.
+ Clear deterministic NAT configuration.
"""
- self.vapi.snat_ipfix(enable=0)
- self.vapi.snat_det_set_timeouts()
- deterministic_mappings = self.vapi.snat_det_map_dump()
+ self.vapi.nat_ipfix(enable=0)
+ self.vapi.nat_det_set_timeouts()
+ deterministic_mappings = self.vapi.nat_det_map_dump()
for dsm in deterministic_mappings:
- self.vapi.snat_add_det_map(dsm.in_addr,
- dsm.in_plen,
- dsm.out_addr,
- dsm.out_plen,
- is_add=0)
+ self.vapi.nat_det_add_del_map(dsm.in_addr,
+ dsm.in_plen,
+ dsm.out_addr,
+ dsm.out_plen,
+ is_add=0)
- interfaces = self.vapi.snat_interface_dump()
+ interfaces = self.vapi.nat44_interface_dump()
for intf in interfaces:
- self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
- intf.is_inside,
- is_add=0)
+ self.vapi.nat44_interface_add_del_feature(intf.sw_if_index,
+ intf.is_inside,
+ is_add=0)
def tearDown(self):
super(TestDeterministicNAT, self).tearDown()
if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show snat detail"))
- self.clear_snat()
+ self.logger.info(self.vapi.cli("show nat44 detail"))
+ self.clear_nat_det()
class TestNAT64(MethodHolder):
diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py
index 95de0be6f2c..61db4d6b14b 100644
--- a/test/vpp_papi_provider.py
+++ b/test/vpp_papi_provider.py
@@ -1061,41 +1061,41 @@ class VppPapiProvider(object):
'mt_next_hop_table_id': next_hop_table_id,
'mt_next_hop_out_label_stack': next_hop_out_label_stack})
- def snat_interface_add_del_feature(
+ def nat44_interface_add_del_feature(
self,
sw_if_index,
is_inside=1,
is_add=1):
- """Enable/disable S-NAT feature on the interface
+ """Enable/disable NAT44 feature on the interface
:param sw_if_index: Software index of the interface
:param is_inside: 1 if inside, 0 if outside (Default value = 1)
:param is_add: 1 if add, 0 if delete (Default value = 1)
"""
return self.api(
- self.papi.snat_interface_add_del_feature,
+ self.papi.nat44_interface_add_del_feature,
{'is_add': is_add,
'is_inside': is_inside,
'sw_if_index': sw_if_index})
- def snat_interface_add_del_output_feature(
+ def nat44_interface_add_del_output_feature(
self,
sw_if_index,
is_inside=1,
is_add=1):
- """Enable/disable S-NAT output feature on the interface
+ """Enable/disable NAT44 output feature on the interface
:param sw_if_index: Software index of the interface
:param is_inside: 1 if inside, 0 if outside (Default value = 1)
:param is_add: 1 if add, 0 if delete (Default value = 1)
"""
return self.api(
- self.papi.snat_interface_add_del_output_feature,
+ self.papi.nat44_interface_add_del_output_feature,
{'is_add': is_add,
'is_inside': is_inside,
'sw_if_index': sw_if_index})
- def snat_add_static_mapping(
+ def nat44_add_del_static_mapping(
self,
local_ip,
external_ip=0,
@@ -1105,9 +1105,8 @@ class VppPapiProvider(object):
addr_only=1,
vrf_id=0,
protocol=0,
- is_add=1,
- is_ip4=1):
- """Add/delete S-NAT static mapping
+ is_add=1):
+ """Add/delete NAT44 static mapping
:param local_ip: Local IP address
:param external_ip: External IP address
@@ -1118,12 +1117,10 @@ class VppPapiProvider(object):
:param vrf_id: VRF ID
:param protocol: IP protocol (Default value = 0)
:param is_add: 1 if add, 0 if delete (Default value = 1)
- :param is_ip4: 1 if address type is IPv4 (Default value = 1)
"""
return self.api(
- self.papi.snat_add_static_mapping,
+ self.papi.nat44_add_del_static_mapping,
{'is_add': is_add,
- 'is_ip4': is_ip4,
'addr_only': addr_only,
'local_ip_address': local_ip,
'external_ip_address': external_ip,
@@ -1133,100 +1130,96 @@ class VppPapiProvider(object):
'vrf_id': vrf_id,
'protocol': protocol})
- def snat_add_address_range(
+ def nat44_add_del_address_range(
self,
first_ip_address,
last_ip_address,
is_add=1,
- is_ip4=1,
vrf_id=0xFFFFFFFF):
- """Add/del S-NAT address range
+ """Add/del NAT44 address range
:param first_ip_address: First IP address
:param last_ip_address: Last IP address
:param vrf_id: VRF id for the address range
:param is_add: 1 if add, 0 if delete (Default value = 1)
- :param is_ip4: 1 if address type is IPv4 (Default value = 1)
"""
return self.api(
- self.papi.snat_add_address_range,
- {'is_ip4': is_ip4,
- 'first_ip_address': first_ip_address,
+ self.papi.nat44_add_del_address_range,
+ {'first_ip_address': first_ip_address,
'last_ip_address': last_ip_address,
'vrf_id': vrf_id,
'is_add': is_add})
- def snat_address_dump(self):
- """Dump S-NAT addresses
- :return: Dictionary of S-NAT addresses
+ def nat44_address_dump(self):
+ """Dump NAT44 addresses
+ :return: Dictionary of NAT44 addresses
"""
- return self.api(self.papi.snat_address_dump, {})
+ return self.api(self.papi.nat44_address_dump, {})
- def snat_interface_dump(self):
- """Dump interfaces with S-NAT feature
- :return: Dictionary of interfaces with S-NAT feature
+ def nat44_interface_dump(self):
+ """Dump interfaces with NAT44 feature
+ :return: Dictionary of interfaces with NAT44 feature
"""
- return self.api(self.papi.snat_interface_dump, {})
+ return self.api(self.papi.nat44_interface_dump, {})
- def snat_interface_output_feature_dump(self):
- """Dump interfaces with S-NAT output feature
- :return: Dictionary of interfaces with S-NAT output feature
+ def nat44_interface_output_feature_dump(self):
+ """Dump interfaces with NAT44 output feature
+ :return: Dictionary of interfaces with NAT44 output feature
"""
- return self.api(self.papi.snat_interface_output_feature_dump, {})
+ return self.api(self.papi.nat44_interface_output_feature_dump, {})
- def snat_static_mapping_dump(self):
- """Dump S-NAT static mappings
- :return: Dictionary of S-NAT static mappings
+ def nat44_static_mapping_dump(self):
+ """Dump NAT44 static mappings
+ :return: Dictionary of NAT44 static mappings
"""
- return self.api(self.papi.snat_static_mapping_dump, {})
+ return self.api(self.papi.nat44_static_mapping_dump, {})
- def snat_show_config(self):
- """Show S-NAT config
- :return: S-NAT config parameters
+ def nat_show_config(self):
+ """Show NAT plugin config
+ :return: NAT plugin config parameters
"""
- return self.api(self.papi.snat_show_config, {})
+ return self.api(self.papi.nat_show_config, {})
- def snat_add_interface_addr(
+ def nat44_add_interface_addr(
self,
sw_if_index,
is_add=1):
- """Add/del S-NAT address from interface
+ """Add/del NAT44 address from interface
:param sw_if_index: Software index of the interface
:param is_add: 1 if add, 0 if delete (Default value = 1)
"""
- return self.api(self.papi.snat_add_del_interface_addr,
+ return self.api(self.papi.nat44_add_del_interface_addr,
{'is_add': is_add, 'sw_if_index': sw_if_index})
- def snat_interface_addr_dump(self):
- """Dump S-NAT addresses interfaces
- :return: Dictionary of S-NAT addresses interfaces
+ def nat44_interface_addr_dump(self):
+ """Dump NAT44 addresses interfaces
+ :return: Dictionary of NAT44 addresses interfaces
"""
- return self.api(self.papi.snat_interface_addr_dump, {})
+ return self.api(self.papi.nat44_interface_addr_dump, {})
- def snat_ipfix(
+ def nat_ipfix(
self,
domain_id=1,
src_port=4739,
enable=1):
- """Enable/disable S-NAT IPFIX logging
+ """Enable/disable NAT IPFIX logging
:param domain_id: Observation domain ID (Default value = 1)
:param src_port: Source port number (Default value = 4739)
:param enable: 1 if enable, 0 if disable (Default value = 1)
"""
return self.api(
- self.papi.snat_ipfix_enable_disable,
+ self.papi.nat_ipfix_enable_disable,
{'domain_id': domain_id,
'src_port': src_port,
'enable': enable})
- def snat_user_session_dump(
+ def nat44_user_session_dump(
self,
ip_address,
- vrf_id,
- is_ip4=1):
- """Dump S-NAT user's sessions
+ vrf_id):
+ """Dump NAT44 user's sessions
:param ip_address: ip adress of the user to be dumped
:param cpu_index: cpu_index on which the user is
@@ -1234,26 +1227,25 @@ class VppPapiProvider(object):
:return: Dictionary of S-NAT sessions
"""
return self.api(
- self.papi.snat_user_session_dump,
+ self.papi.nat44_user_session_dump,
{'ip_address': ip_address,
- 'vrf_id': vrf_id,
- 'is_ip4': is_ip4})
+ 'vrf_id': vrf_id})
- def snat_user_dump(self):
- """Dump S-NAT users
+ def nat44_user_dump(self):
+ """Dump NAT44 users
- :return: Dictionary of S-NAT users
+ :return: Dictionary of NAT44 users
"""
- return self.api(self.papi.snat_user_dump, {})
+ return self.api(self.papi.nat44_user_dump, {})
- def snat_add_det_map(
+ def nat_det_add_del_map(
self,
in_addr,
in_plen,
out_addr,
out_plen,
is_add=1):
- """Add/delete S-NAT deterministic mapping
+ """Add/delete deterministic NAT mapping
:param is_add - 1 if add, 0 if delete
:param in_addr - inside IP address
@@ -1262,14 +1254,15 @@ class VppPapiProvider(object):
:param out_plen - outside IP address prefix length
"""
return self.api(
- self.papi.snat_add_det_map,
+ self.papi.nat_det_add_del_map,
{'is_add': is_add,
+ 'is_nat44': 1,
'in_addr': in_addr,
'in_plen': in_plen,
'out_addr': out_addr,
'out_plen': out_plen})
- def snat_det_forward(
+ def nat_det_forward(
self,
in_addr):
"""Get outside address and port range from inside address
@@ -1277,10 +1270,11 @@ class VppPapiProvider(object):
:param in_addr - inside IP address
"""
return self.api(
- self.papi.snat_det_forward,
- {'in_addr': in_addr})
+ self.papi.nat_det_forward,
+ {'in_addr': in_addr,
+ 'is_nat44': 1})
- def snat_det_reverse(
+ def nat_det_reverse(
self,
out_addr,
out_port):
@@ -1290,18 +1284,18 @@ class VppPapiProvider(object):
:param out_port - outside port
"""
return self.api(
- self.papi.snat_det_reverse,
+ self.papi.nat_det_reverse,
{'out_addr': out_addr,
'out_port': out_port})
- def snat_det_map_dump(self):
- """Dump S-NAT deterministic mappings
+ def nat_det_map_dump(self):
+ """Dump deterministic NAT mappings
- :return: Dictionary of S-NAT deterministic mappings
+ :return: Dictionary of deterministic NAT mappings
"""
- return self.api(self.papi.snat_det_map_dump, {})
+ return self.api(self.papi.nat_det_map_dump, {})
- def snat_det_set_timeouts(
+ def nat_det_set_timeouts(
self,
udp=300,
tcp_established=7440,
@@ -1315,78 +1309,71 @@ class VppPapiProvider(object):
:param icmp - ICMP timeout (Default value = 60)
"""
return self.api(
- self.papi.snat_det_set_timeouts,
+ self.papi.nat_det_set_timeouts,
{'udp': udp,
'tcp_established': tcp_established,
'tcp_transitory': tcp_transitory,
'icmp': icmp})
- def snat_det_get_timeouts(self):
+ def nat_det_get_timeouts(self):
"""Get values of timeouts for deterministic NAT
:return: Timeouts for deterministic NAT (in seconds)
"""
- return self.api(self.papi.snat_det_get_timeouts, {})
+ return self.api(self.papi.nat_det_get_timeouts, {})
- def snat_det_close_session_out(
+ def nat_det_close_session_out(
self,
out_addr,
out_port,
ext_addr,
- ext_port,
- is_ip4=1):
- """Close CGN session using outside address and port
+ ext_port):
+ """Close deterministic NAT session using outside address and port
:param out_addr - outside IP address
:param out_port - outside port
:param ext_addr - external host IP address
:param ext_port - external host port
- :param is_ip4: 1 if address type is IPv4 (Default value = 1)
"""
return self.api(
- self.papi.snat_det_close_session_out,
+ self.papi.nat_det_close_session_out,
{'out_addr': out_addr,
'out_port': out_port,
'ext_addr': ext_addr,
- 'ext_port': ext_port,
- 'is_ip4': is_ip4})
+ 'ext_port': ext_port})
- def snat_det_close_session_in(
+ def nat_det_close_session_in(
self,
in_addr,
in_port,
ext_addr,
- ext_port,
- is_ip4=1):
- """Close CGN session using inside address and port
+ ext_port):
+ """Close deterministic NAT session using inside address and port
:param in_addr - inside IP address
:param in_port - inside port
:param ext_addr - external host IP address
:param ext_port - external host port
- :param is_ip4: 1 if address type is IPv4 (Default value = 1)
"""
return self.api(
- self.papi.snat_det_close_session_in,
+ self.papi.nat_det_close_session_in,
{'in_addr': in_addr,
'in_port': in_port,
'ext_addr': ext_addr,
'ext_port': ext_port,
- 'is_ip4': is_ip4})
+ 'is_nat44': 1})
- def snat_det_session_dump(
+ def nat_det_session_dump(
self,
- user_addr,
- is_ip4=1):
- """Dump S-NAT deterministic sessions belonging to a user
+ user_addr):
+ """Dump deterministic NAT sessions belonging to a user
:param user_addr - inside IP address of the user
- :param is_ip4: - 1 if address type is IPv4 (Default value = 1)
- :return: Dictionary of S-NAT deterministic sessions
+ :return: Dictionary of deterministic NAT sessions
"""
return self.api(
- self.papi.snat_det_session_dump,
- {'is_ip4': is_ip4,
+ self.papi.nat_det_session_dump,
+ {'is_nat44': 1,
'user_addr': user_addr})
def nat64_add_del_pool_addr_range(