@property
def config_flags(self):
- return VppEnum.vl_api_nat_config_flags_t
-
- @property
- def nat44_config_flags(self):
- return VppEnum.vl_api_nat44_config_flags_t
+ return VppEnum.vl_api_nat44_ei_config_flags_t
@property
def SYSLOG_SEVERITY(self):
"""
if not (local_port and external_port):
- flags |= self.config_flags.NAT_IS_ADDR_ONLY
+ flags |= self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
- self.vapi.nat44_add_del_static_mapping(
+ self.vapi.nat44_ei_add_del_static_mapping(
is_add=is_add,
local_ip_address=local_ip,
external_ip_address=external_ip,
flags=flags,
tag=tag)
- def nat44_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF, twice_nat=0):
+ def nat44_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF):
"""
Add/delete NAT44EI address
:param ip: IP address
:param is_add: 1 if add, 0 if delete (Default add)
- :param twice_nat: twice NAT address for external hosts
"""
- flags = self.config_flags.NAT_IS_TWICE_NAT if twice_nat else 0
- self.vapi.nat44_add_del_address_range(first_ip_address=ip,
- last_ip_address=ip,
- vrf_id=vrf_id,
- is_add=is_add,
- flags=flags)
+ self.vapi.nat44_ei_add_del_address_range(first_ip_address=ip,
+ last_ip_address=ip,
+ vrf_id=vrf_id,
+ is_add=is_add)
def create_routes_and_neigbors(self):
r1 = VppIpRoute(self, self.pg7.remote_ip4, 32,
# natEvent
self.assertEqual(scapy.compat.orb(record[230]), 13)
# natQuotaExceededEvent
- self.assertEqual(struct.pack("I", 1), record[466])
+ self.assertEqual(struct.pack("!I", 1), record[466])
# maxSessionEntries
- self.assertEqual(struct.pack("I", limit), record[471])
+ self.assertEqual(struct.pack("!I", limit), record[471])
def verify_no_nat44_user(self):
""" Verify that there is no NAT44EI user """
- users = self.vapi.nat44_user_dump()
+ users = self.vapi.nat44_ei_user_dump()
self.assertEqual(len(users), 0)
- users = self.statistics.get_counter('/nat44/total-users')
+ users = self.statistics['/nat44-ei/total-users']
self.assertEqual(users[0][0], 0)
- sessions = self.statistics.get_counter('/nat44/total-sessions')
+ sessions = self.statistics['/nat44-ei/total-sessions']
self.assertEqual(sessions[0][0], 0)
def verify_syslog_apmap(self, data, is_add=True):
self.assertEqual(data, p[Raw].load)
-class TestNAT44EIAPI(MethodHolder):
- """ NAT44EI API Test Cases """
-
- fq_nelts = 512
-
- def setUp(self):
- super(TestNAT44EIAPI, self).setUp()
- self.vapi.nat_set_fq_options(frame_queue_nelts=self.fq_nelts)
- self.vapi.nat44_plugin_enable_disable(enable=1)
-
- def tearDown(self):
- super(TestNAT44EIAPI, self).tearDown()
- if not self.vpp_dead:
- self.vapi.nat44_plugin_enable_disable(enable=0)
- self.vapi.cli("clear logging")
-
- def test_show_frame_queue_nelts(self):
- """ API test - worker handoff frame queue elements """
- nat_config = self.vapi.nat_show_fq_options()
- self.assertEqual(self.fq_nelts, nat_config.frame_queue_nelts)
- self.vapi.nat44_plugin_enable_disable(enable=0)
- self.vapi.cli("set nat frame-queue-nelts 256")
- self.vapi.nat44_plugin_enable_disable(enable=1)
- nat_config = self.vapi.nat_show_fq_options()
- self.assertEqual(256, nat_config.frame_queue_nelts)
+def get_nat44_ei_in2out_worker_index(ip, vpp_worker_count):
+ if 0 == vpp_worker_count:
+ return 0
+ numeric = socket.inet_aton(ip)
+ numeric = struct.unpack("!L", numeric)[0]
+ numeric = socket.htonl(numeric)
+ h = numeric + (numeric >> 8) + (numeric >> 16) + (numeric >> 24)
+ return 1 + h % vpp_worker_count
class TestNAT44EI(MethodHolder):
@classmethod
def setUpClass(cls):
super(TestNAT44EI, cls).setUpClass()
- cls.vapi.cli("set log class nat level debug")
+ cls.vapi.cli("set log class nat44-ei level debug")
cls.tcp_port_in = 6303
cls.tcp_port_out = 6303
cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
cls.pg9.resolve_arp()
- def setUp(self):
- super(TestNAT44EI, self).setUp()
- self.vapi.nat44_plugin_enable_disable(
+ def plugin_enable(self):
+ self.vapi.nat44_ei_plugin_enable_disable(
sessions=self.max_translations,
users=self.max_users, enable=1)
+ def setUp(self):
+ super(TestNAT44EI, self).setUp()
+ self.plugin_enable()
+
def tearDown(self):
super(TestNAT44EI, self).tearDown()
if not self.vpp_dead:
- self.vapi.nat_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port,
- enable=0)
+ self.vapi.nat44_ei_ipfix_enable_disable(
+ domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port,
+ enable=0)
self.ipfix_src_port = 4739
self.ipfix_domain_id = 1
- self.vapi.nat44_plugin_enable_disable(enable=0)
+ self.vapi.nat44_ei_plugin_enable_disable(enable=0)
self.vapi.cli("clear logging")
def test_clear_sessions(self):
""" NAT44EI session clearing test """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- nat_config = self.vapi.nat_show_config()
- self.assertEqual(0, nat_config.endpoint_dependent)
-
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture)
- sessions = self.statistics.get_counter('/nat44/total-sessions')
- self.assertTrue(sessions[0][0] > 0)
+ sessions = self.statistics['/nat44-ei/total-sessions']
+ self.assertGreater(sessions[:, 0].sum(), 0, "Session count invalid")
self.logger.info("sessions before clearing: %s" % sessions[0][0])
- self.vapi.cli("clear nat44 sessions")
+ self.vapi.cli("clear nat44 ei sessions")
- sessions = self.statistics.get_counter('/nat44/total-sessions')
- self.assertEqual(sessions[0][0], 0)
+ sessions = self.statistics['/nat44-ei/total-sessions']
+ self.assertEqual(sessions[:, 0].sum(), 0, "Session count invalid")
self.logger.info("sessions after clearing: %s" % sessions[0][0])
def test_dynamic(self):
""" NAT44EI dynamic translation test """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
# in2out
- tcpn = self.statistics.get_counter('/nat44/in2out/slowpath/tcp')[0]
- udpn = self.statistics.get_counter('/nat44/in2out/slowpath/udp')[0]
- icmpn = self.statistics.get_counter('/nat44/in2out/slowpath/icmp')[0]
- drops = self.statistics.get_counter('/nat44/in2out/slowpath/drops')[0]
+ tcpn = self.statistics['/nat44-ei/in2out/slowpath/tcp']
+ udpn = self.statistics['/nat44-ei/in2out/slowpath/udp']
+ icmpn = self.statistics['/nat44-ei/in2out/slowpath/icmp']
+ drops = self.statistics['/nat44-ei/in2out/slowpath/drops']
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.verify_capture_out(capture)
if_idx = self.pg0.sw_if_index
- cnt = self.statistics.get_counter('/nat44/in2out/slowpath/tcp')[0]
- self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
- cnt = self.statistics.get_counter('/nat44/in2out/slowpath/udp')[0]
- self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
- cnt = self.statistics.get_counter('/nat44/in2out/slowpath/icmp')[0]
- self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
- cnt = self.statistics.get_counter('/nat44/in2out/slowpath/drops')[0]
- self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
+ cnt = self.statistics['/nat44-ei/in2out/slowpath/tcp']
+ self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2)
+ cnt = self.statistics['/nat44-ei/in2out/slowpath/udp']
+ self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1)
+ cnt = self.statistics['/nat44-ei/in2out/slowpath/icmp']
+ self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1)
+ cnt = self.statistics['/nat44-ei/in2out/slowpath/drops']
+ self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0)
# out2in
- tcpn = self.statistics.get_counter('/nat44/out2in/slowpath/tcp')[0]
- udpn = self.statistics.get_counter('/nat44/out2in/slowpath/udp')[0]
- icmpn = self.statistics.get_counter('/nat44/out2in/slowpath/icmp')[0]
- drops = self.statistics.get_counter('/nat44/out2in/slowpath/drops')[0]
+ tcpn = self.statistics['/nat44-ei/out2in/slowpath/tcp']
+ udpn = self.statistics['/nat44-ei/out2in/slowpath/udp']
+ icmpn = self.statistics['/nat44-ei/out2in/slowpath/icmp']
+ drops = self.statistics['/nat44-ei/out2in/slowpath/drops']
pkts = self.create_stream_out(self.pg1)
self.pg1.add_stream(pkts)
self.verify_capture_in(capture, self.pg0)
if_idx = self.pg1.sw_if_index
- cnt = self.statistics.get_counter('/nat44/out2in/slowpath/tcp')[0]
- self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
- cnt = self.statistics.get_counter('/nat44/out2in/slowpath/udp')[0]
- self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
- cnt = self.statistics.get_counter('/nat44/out2in/slowpath/icmp')[0]
- self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
- cnt = self.statistics.get_counter('/nat44/out2in/slowpath/drops')[0]
- self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
-
- users = self.statistics.get_counter('/nat44/total-users')
- self.assertEqual(users[0][0], 1)
- sessions = self.statistics.get_counter('/nat44/total-sessions')
- self.assertEqual(sessions[0][0], 3)
+ cnt = self.statistics['/nat44-ei/out2in/slowpath/tcp']
+ self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2)
+ cnt = self.statistics['/nat44-ei/out2in/slowpath/udp']
+ self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1)
+ cnt = self.statistics['/nat44-ei/out2in/slowpath/icmp']
+ self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1)
+ cnt = self.statistics['/nat44-ei/out2in/slowpath/drops']
+ self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0)
+
+ users = self.statistics['/nat44-ei/total-users']
+ self.assertEqual(users[:, 0].sum(), 1)
+ sessions = self.statistics['/nat44-ei/total-sessions']
+ self.assertEqual(sessions[:, 0].sum(), 3)
def test_dynamic_icmp_errors_in2out_ttl_1(self):
""" NAT44EI handling of client packets with TTL=1 """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
""" NAT44EI handling of server packets with TTL=1 """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
"""
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
"""
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
""" NAT44EI ping out interface from outside network """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
""" NAT44EI ping internal host from outside network """
self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
def test_forwarding(self):
""" NAT44EI forwarding test """
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- self.vapi.nat44_forwarding_enable_disable(enable=1)
+ self.vapi.nat44_ei_forwarding_enable_disable(enable=1)
real_ip = self.pg0.remote_ip4
alias_ip = self.nat_addr
- flags = self.config_flags.NAT_IS_ADDR_ONLY
- self.vapi.nat44_add_del_static_mapping(is_add=1,
- local_ip_address=real_ip,
- external_ip_address=alias_ip,
- external_sw_if_index=0xFFFFFFFF,
- flags=flags)
+ flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
+ self.vapi.nat44_ei_add_del_static_mapping(
+ is_add=1, local_ip_address=real_ip,
+ external_ip_address=alias_ip,
+ external_sw_if_index=0xFFFFFFFF,
+ flags=flags)
try:
# static mapping match
self.pg0.remote_hosts[0] = host0
finally:
- self.vapi.nat44_forwarding_enable_disable(enable=0)
- flags = self.config_flags.NAT_IS_ADDR_ONLY
- self.vapi.nat44_add_del_static_mapping(
+ self.vapi.nat44_ei_forwarding_enable_disable(enable=0)
+ flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
+ self.vapi.nat44_ei_add_del_static_mapping(
is_add=0,
local_ip_address=real_ip,
external_ip_address=alias_ip,
self.icmp_id_out = 6305
self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- sm = self.vapi.nat44_static_mapping_dump()
+ sm = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(len(sm), 1)
self.assertEqual(sm[0].tag, '')
self.assertEqual(sm[0].protocol, 0)
tag = "testTAG"
self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip, tag=tag)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- sm = self.vapi.nat44_static_mapping_dump()
+ sm = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(len(sm), 1)
self.assertEqual(sm[0].tag, tag)
self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
self.icmp_id_in, self.icmp_id_out,
proto=IP_PROTOS.icmp)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
self.icmp_id_in, self.icmp_id_out,
proto=IP_PROTOS.icmp)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
vrf_id=10)
self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
vrf_id=10)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg3.sw_if_index,
is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg4.sw_if_index,
flags=flags, is_add=1)
self.icmp_id_out = 6305
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
# 1:1NAT
self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
- sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
+ sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
self.assertEqual(len(sessions), 0)
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
def test_identity_nat(self):
""" NAT44EI Identity NAT """
- flags = self.config_flags.NAT_IS_ADDR_ONLY
- self.vapi.nat44_add_del_identity_mapping(
+ flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
+ self.vapi.nat44_ei_add_del_identity_mapping(
ip_address=self.pg0.remote_ip4, sw_if_index=0xFFFFFFFF,
flags=flags, is_add=1)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
+ sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
self.assertEqual(len(sessions), 0)
- flags = self.config_flags.NAT_IS_ADDR_ONLY
- self.vapi.nat44_add_del_identity_mapping(
+ flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
+ self.vapi.nat44_ei_add_del_identity_mapping(
ip_address=self.pg0.remote_ip4, sw_if_index=0xFFFFFFFF,
flags=flags, vrf_id=1, is_add=1)
- identity_mappings = self.vapi.nat44_identity_mapping_dump()
+ identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
self.assertEqual(len(identity_mappings), 2)
def test_multiple_inside_interfaces(self):
"""
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg3.sw_if_index,
is_add=1)
static_nat_ip = "10.0.0.10"
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg3.sw_if_index,
is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg4.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg5.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg6.sw_if_index,
flags=flags, is_add=1)
self.nat44_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
self.verify_capture_in(capture, self.pg5)
# pg5 session dump
- addresses = self.vapi.nat44_address_dump()
+ addresses = self.vapi.nat44_ei_address_dump()
self.assertEqual(len(addresses), 1)
- sessions = self.vapi.nat44_user_session_dump(self.pg5.remote_ip4, 10)
+ sessions = self.vapi.nat44_ei_user_session_dump(
+ self.pg5.remote_ip4, 10)
self.assertEqual(len(sessions), 3)
for session in sessions:
- self.assertFalse(session.flags & self.config_flags.NAT_IS_STATIC)
+ self.assertFalse(session.flags &
+ self.config_flags.NAT44_EI_STATIC_MAPPING)
self.assertEqual(str(session.inside_ip_address),
self.pg5.remote_ip4)
self.assertEqual(session.outside_ip_address,
self.verify_capture_in(capture, self.pg6)
# general user and session dump verifications
- users = self.vapi.nat44_user_dump()
+ users = self.vapi.nat44_ei_user_dump()
self.assertGreaterEqual(len(users), 3)
- addresses = self.vapi.nat44_address_dump()
+ addresses = self.vapi.nat44_ei_address_dump()
self.assertEqual(len(addresses), 1)
for user in users:
- sessions = self.vapi.nat44_user_session_dump(user.ip_address,
- user.vrf_id)
+ sessions = self.vapi.nat44_ei_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)
self.assertTrue(session.protocol in
[IP_PROTOS.tcp, IP_PROTOS.udp,
IP_PROTOS.icmp])
- self.assertFalse(session.flags &
- self.config_flags.NAT_IS_EXT_HOST_VALID)
# pg4 session dump
- sessions = self.vapi.nat44_user_session_dump(self.pg4.remote_ip4, 10)
+ sessions = self.vapi.nat44_ei_user_session_dump(
+ self.pg4.remote_ip4, 10)
self.assertGreaterEqual(len(sessions), 4)
for session in sessions:
- self.assertFalse(session.flags & self.config_flags.NAT_IS_STATIC)
+ self.assertFalse(
+ session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
self.assertEqual(str(session.inside_ip_address),
self.pg4.remote_ip4)
self.assertEqual(session.outside_ip_address,
addresses[0].ip_address)
# pg6 session dump
- sessions = self.vapi.nat44_user_session_dump(self.pg6.remote_ip4, 20)
+ sessions = self.vapi.nat44_ei_user_session_dump(
+ self.pg6.remote_ip4, 20)
self.assertGreaterEqual(len(sessions), 3)
for session in sessions:
- self.assertTrue(session.flags & self.config_flags.NAT_IS_STATIC)
+ self.assertTrue(
+ session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
self.assertEqual(str(session.inside_ip_address),
self.pg6.remote_ip4)
self.assertEqual(str(session.outside_ip_address),
server_out_port = 8765
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
server_in_port, server_out_port,
proto=IP_PROTOS.tcp)
- cnt = self.statistics.get_counter('/nat44/hairpinning')[0]
+ cnt = self.statistics['/nat44-ei/hairpinning']
# send packet from host to server
p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
IP(src=host.ip4, dst=self.nat_addr) /
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- after = self.statistics.get_counter('/nat44/hairpinning')[0]
+ after = self.statistics['/nat44-ei/hairpinning']
if_idx = self.pg0.sw_if_index
- self.assertEqual(after[if_idx] - cnt[if_idx], 1)
+ self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(), 1)
# send reply from server to host
p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- after = self.statistics.get_counter('/nat44/hairpinning')[0]
+ after = self.statistics['/nat44-ei/hairpinning']
if_idx = self.pg0.sw_if_index
- self.assertEqual(after[if_idx] - cnt[if_idx], 2)
+ self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(),
+ 2+(1 if self.vpp_worker_count > 0 else 0))
def test_hairpinning2(self):
""" NAT44EI hairpinning - 1:1 NAT"""
server_udp_port = 20
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
raise
def test_hairpinning_avoid_inf_loop(self):
- """ NAT44 hairpinning - 1:1 NAPT avoid infinite loop """
+ """ NAT44EI hairpinning - 1:1 NAPT avoid infinite loop """
host = self.pg0.remote_hosts[0]
server = self.pg0.remote_hosts[1]
server_out_port = 8765
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.pg_start()
# Here VPP used to crash due to an infinite loop
- cnt = self.statistics.get_counter('/nat44/hairpinning')[0]
+ cnt = self.statistics['/nat44-ei/hairpinning']
# send packet from host to server
p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
IP(src=host.ip4, dst=self.nat_addr) /
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- after = self.statistics.get_counter('/nat44/hairpinning')[0]
+ after = self.statistics['/nat44-ei/hairpinning']
if_idx = self.pg0.sw_if_index
- self.assertEqual(after[if_idx] - cnt[if_idx], 1)
+ self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(), 1)
# send reply from server to host
p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- after = self.statistics.get_counter('/nat44/hairpinning')[0]
+ after = self.statistics['/nat44-ei/hairpinning']
if_idx = self.pg0.sw_if_index
- self.assertEqual(after[if_idx] - cnt[if_idx], 2)
+ self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(),
+ 2+(1 if self.vpp_worker_count > 0 else 0))
def test_interface_addr(self):
""" NAT44EI acquire addresses from interface """
- self.vapi.nat44_add_del_interface_addr(
+ self.vapi.nat44_ei_add_del_interface_addr(
is_add=1,
sw_if_index=self.pg7.sw_if_index)
# no address in NAT pool
- addresses = self.vapi.nat44_address_dump()
+ addresses = self.vapi.nat44_ei_address_dump()
self.assertEqual(0, len(addresses))
# configure interface address and check NAT address pool
self.pg7.config_ip4()
- addresses = self.vapi.nat44_address_dump()
+ addresses = self.vapi.nat44_ei_address_dump()
self.assertEqual(1, len(addresses))
self.assertEqual(str(addresses[0].ip_address), self.pg7.local_ip4)
# remove interface address and check NAT address pool
self.pg7.unconfig_ip4()
- addresses = self.vapi.nat44_address_dump()
+ addresses = self.vapi.nat44_ei_address_dump()
self.assertEqual(0, len(addresses))
def test_interface_addr_static_mapping(self):
""" NAT44EI Static mapping with addresses from interface """
tag = "testTAG"
- self.vapi.nat44_add_del_interface_addr(
+ self.vapi.nat44_ei_add_del_interface_addr(
is_add=1,
sw_if_index=self.pg7.sw_if_index)
self.nat44_add_static_mapping(
tag=tag)
# static mappings with external interface
- static_mappings = self.vapi.nat44_static_mapping_dump()
+ static_mappings = self.vapi.nat44_ei_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.nat44_static_mapping_dump()
+ static_mappings = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(2, len(static_mappings))
resolved = False
for sm in static_mappings:
# remove interface address and check static mappings
self.pg7.unconfig_ip4()
- static_mappings = self.vapi.nat44_static_mapping_dump()
+ static_mappings = self.vapi.nat44_ei_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 again and check static mappings
self.pg7.config_ip4()
- static_mappings = self.vapi.nat44_static_mapping_dump()
+ static_mappings = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(2, len(static_mappings))
resolved = False
for sm in static_mappings:
external_sw_if_index=self.pg7.sw_if_index,
tag=tag,
is_add=0)
- static_mappings = self.vapi.nat44_static_mapping_dump()
+ static_mappings = self.vapi.nat44_ei_static_mapping_dump()
self.assertEqual(0, len(static_mappings))
def test_interface_addr_identity_nat(self):
""" NAT44EI Identity NAT with addresses from interface """
port = 53053
- self.vapi.nat44_add_del_interface_addr(
+ self.vapi.nat44_ei_add_del_interface_addr(
is_add=1,
sw_if_index=self.pg7.sw_if_index)
- self.vapi.nat44_add_del_identity_mapping(
+ self.vapi.nat44_ei_add_del_identity_mapping(
ip_address=b'0',
sw_if_index=self.pg7.sw_if_index,
port=port,
is_add=1)
# identity mappings with external interface
- identity_mappings = self.vapi.nat44_identity_mapping_dump()
+ identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
self.assertEqual(1, len(identity_mappings))
self.assertEqual(self.pg7.sw_if_index,
identity_mappings[0].sw_if_index)
# configure interface address and check identity mappings
self.pg7.config_ip4()
- identity_mappings = self.vapi.nat44_identity_mapping_dump()
+ identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
resolved = False
self.assertEqual(2, len(identity_mappings))
for sm in identity_mappings:
# remove interface address and check identity mappings
self.pg7.unconfig_ip4()
- identity_mappings = self.vapi.nat44_identity_mapping_dump()
+ identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
self.assertEqual(1, len(identity_mappings))
self.assertEqual(self.pg7.sw_if_index,
identity_mappings[0].sw_if_index)
collector_port = 30303
bind_layers(UDP, IPFIX, dport=30303)
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
path_mtu=512,
template_interval=10,
collector_port=collector_port)
- self.vapi.nat_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port,
- enable=1)
+ self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port,
+ enable=1)
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
def test_ipfix_addr_exhausted(self):
""" NAT44EI IPFIX logging NAT addresses exhausted """
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
src_address=self.pg3.local_ip4,
path_mtu=512,
template_interval=10)
- self.vapi.nat_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port,
- enable=1)
+ self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port,
+ enable=1)
p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
def test_ipfix_max_sessions(self):
""" NAT44EI IPFIX logging maximum session entries exceeded """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- max_sessions = self.max_translations
+ max_sessions_per_thread = self.max_translations
+ max_sessions = max(1, self.vpp_worker_count) * max_sessions_per_thread
pkts = []
for i in range(0, max_sessions):
src_address=self.pg3.local_ip4,
path_mtu=512,
template_interval=10)
- self.vapi.nat_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port,
- enable=1)
+ self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port,
+ enable=1)
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
for p in capture:
if p.haslayer(Data):
data = ipfix.decode_data_set(p.getlayer(Set))
- self.verify_ipfix_max_sessions(data, max_sessions)
+ self.verify_ipfix_max_sessions(data, max_sessions_per_thread)
def test_syslog_apmap(self):
""" NAT44EI syslog address and port mapping creation and deletion """
self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_INFO)
self.vapi.syslog_set_sender(self.pg3.local_ip4, self.pg3.remote_ip4)
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
""" NAT44EI add pool addresses to FIB """
static_addr = '10.0.0.10'
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr)
self.nat44_add_address(nat_ip1, vrf_id=vrf_id1)
self.nat44_add_address(nat_ip2, vrf_id=vrf_id2)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg2.sw_if_index,
is_add=1)
self.nat44_add_address(nat_ip1)
self.nat44_add_address(nat_ip2, vrf_id=99)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg2.sw_if_index,
is_add=1)
""" NAT44EI interfaces without configured IP address """
self.create_routes_and_neigbors()
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg7.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg8.sw_if_index,
is_add=1)
self.create_routes_and_neigbors()
self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg7.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg8.sw_if_index,
is_add=1)
self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
self.icmp_id_in, self.icmp_id_out,
proto=IP_PROTOS.icmp)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg7.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg8.sw_if_index,
is_add=1)
""" NAT44EI 1:1 translate packet with unknown protocol """
nat_ip = "10.0.0.10"
self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.nat44_add_static_mapping(host.ip4, host_nat_ip)
self.nat44_add_static_mapping(server.ip4, server_nat_ip)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
def test_output_feature(self):
""" NAT44EI output feature (in2out postrouting) """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_output_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_output_feature(
is_add=1, flags=flags,
sw_if_index=self.pg0.sw_if_index)
- self.vapi.nat44_interface_add_del_output_feature(
+ self.vapi.nat44_ei_interface_add_del_output_feature(
is_add=1, flags=flags,
sw_if_index=self.pg1.sw_if_index)
- self.vapi.nat44_interface_add_del_output_feature(
+ self.vapi.nat44_ei_interface_add_del_output_feature(
is_add=1,
sw_if_index=self.pg3.sw_if_index)
self.nat44_add_address(nat_ip_vrf10, vrf_id=10)
self.nat44_add_address(nat_ip_vrf20, vrf_id=20)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_output_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_output_feature(
is_add=1, flags=flags,
sw_if_index=self.pg4.sw_if_index)
- self.vapi.nat44_interface_add_del_output_feature(
+ self.vapi.nat44_ei_interface_add_del_output_feature(
is_add=1, flags=flags,
sw_if_index=self.pg6.sw_if_index)
- self.vapi.nat44_interface_add_del_output_feature(
+ self.vapi.nat44_ei_interface_add_del_output_feature(
is_add=1,
sw_if_index=self.pg3.sw_if_index)
server_out_port = 8765
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_output_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_output_feature(
is_add=1, flags=flags,
sw_if_index=self.pg0.sw_if_index)
- self.vapi.nat44_interface_add_del_output_feature(
+ self.vapi.nat44_ei_interface_add_del_output_feature(
is_add=1,
sw_if_index=self.pg1.sw_if_index)
external_port = 0
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg9.sw_if_index,
is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg9.sw_if_index,
flags=flags, is_add=1)
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
- err = self.statistics.get_err_counter(
- '/err/nat44-classify/next in2out')
+ if self.vpp_worker_count > 1:
+ node = "nat44-ei-handoff-classify"
+ else:
+ node = "nat44-ei-classify"
+
+ err = self.statistics.get_err_counter('/err/%s/next in2out' % node)
self.assertEqual(err, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-classify/next out2in')
+ err = self.statistics.get_err_counter('/err/%s/next out2in' % node)
self.assertEqual(err, 1)
def test_del_session(self):
""" NAT44EI delete session """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.pg_start()
self.pg1.get_capture(len(pkts))
- sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
+ sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
nsessions = len(sessions)
- self.vapi.nat44_del_session(address=sessions[0].inside_ip_address,
- port=sessions[0].inside_port,
- protocol=sessions[0].protocol,
- flags=self.config_flags.NAT_IS_INSIDE)
- self.vapi.nat44_del_session(address=sessions[1].outside_ip_address,
- port=sessions[1].outside_port,
- protocol=sessions[1].protocol)
+ self.vapi.nat44_ei_del_session(
+ address=sessions[0].inside_ip_address,
+ port=sessions[0].inside_port,
+ protocol=sessions[0].protocol,
+ flags=self.config_flags.NAT44_EI_IF_INSIDE)
+
+ self.vapi.nat44_ei_del_session(
+ address=sessions[1].outside_ip_address,
+ port=sessions[1].outside_port,
+ protocol=sessions[1].protocol)
- sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4, 0)
+ sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
self.assertEqual(nsessions - len(sessions), 2)
- self.vapi.nat44_del_session(address=sessions[0].inside_ip_address,
- port=sessions[0].inside_port,
- protocol=sessions[0].protocol,
- flags=self.config_flags.NAT_IS_INSIDE)
+ self.vapi.nat44_ei_del_session(
+ address=sessions[0].inside_ip_address,
+ port=sessions[0].inside_port,
+ protocol=sessions[0].protocol,
+ flags=self.config_flags.NAT44_EI_IF_INSIDE)
self.verify_no_nat44_user()
""" NAT44EI translate fragments arriving in order """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
def test_frag_forwarding(self):
""" NAT44EI forwarding fragment test """
- self.vapi.nat44_add_del_interface_addr(
+ self.vapi.nat44_ei_add_del_interface_addr(
is_add=1,
sw_if_index=self.pg1.sw_if_index)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- self.vapi.nat44_forwarding_enable_disable(enable=1)
+ self.vapi.nat44_ei_forwarding_enable_disable(enable=1)
data = b"A" * 16 + b"B" * 16 + b"C" * 3
pkts = self.create_stream_frag(self.pg1,
server_out_port = random.randint(1025, 65535)
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
# add static mapping for server
""" NAT44EI translate fragments arriving out of order """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
def test_port_restricted(self):
""" NAT44EI Port restricted NAT44EI (MAP-E CE) """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- self.vapi.nat_set_addr_and_port_alloc_alg(alg=1,
- psid_offset=6,
- psid_length=6,
- psid=10)
+ self.vapi.nat44_ei_set_addr_and_port_alloc_alg(alg=1,
+ psid_offset=6,
+ psid_length=6,
+ psid=10)
p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
def test_port_range(self):
""" NAT44EI External address port range """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- self.vapi.nat_set_addr_and_port_alloc_alg(alg=2,
- start_port=1025,
- end_port=1027)
+ self.vapi.nat44_ei_set_addr_and_port_alloc_alg(alg=2,
+ start_port=1025,
+ end_port=1027)
pkts = []
for port in range(0, 5):
self.pg2.resolve_arp()
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg2.sw_if_index,
is_add=1)
def test_mss_clamping(self):
""" NAT44EI TCP MSS clamping """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
flags="S", options=[('MSS', 1400)]))
- self.vapi.nat_set_mss_clamping(enable=1, mss_value=1000)
+ self.vapi.nat44_ei_set_mss_clamping(enable=1, mss_value=1000)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
# Negotiated MSS value greater than configured - changed
self.verify_mss_value(capture[0], 1000)
- self.vapi.nat_set_mss_clamping(enable=0, mss_value=1500)
+ self.vapi.nat44_ei_set_mss_clamping(enable=0, mss_value=1500)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
# MSS clamping disabled - negotiated MSS unchanged
self.verify_mss_value(capture[0], 1400)
- self.vapi.nat_set_mss_clamping(enable=1, mss_value=1500)
+ self.vapi.nat44_ei_set_mss_clamping(enable=1, mss_value=1500)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
def test_ha_send(self):
""" NAT44EI Send HA session synchronization events (active) """
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
self.nat44_add_address(self.nat_addr)
- self.vapi.nat_ha_set_listener(ip_address=self.pg3.local_ip4,
- port=12345,
- path_mtu=512)
- self.vapi.nat_ha_set_failover(ip_address=self.pg3.remote_ip4,
- port=12346, session_refresh_interval=10)
+ self.vapi.nat44_ei_ha_set_listener(
+ ip_address=self.pg3.local_ip4, port=12345, path_mtu=512)
+ self.vapi.nat44_ei_ha_set_failover(
+ ip_address=self.pg3.remote_ip4, port=12346,
+ session_refresh_interval=10)
bind_layers(UDP, HANATStateSync, sport=12345)
# create sessions
capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture)
# active send HA events
- self.vapi.nat_ha_flush()
- stats = self.statistics.get_counter('/nat44/ha/add-event-send')
- self.assertEqual(stats[0][0], 3)
+ self.vapi.nat44_ei_ha_flush()
+ stats = self.statistics['/nat44-ei/ha/add-event-send']
+ self.assertEqual(stats[:, 0].sum(), 3)
capture = self.pg3.get_capture(1)
p = capture[0]
self.assert_packet_checksums_valid(p)
self.assertEqual(udp.sport, 12345)
self.assertEqual(udp.dport, 12346)
self.assertEqual(hanat.version, 1)
- self.assertEqual(hanat.thread_index, 0)
+ # self.assertEqual(hanat.thread_index, 0)
self.assertEqual(hanat.count, 3)
seq = hanat.sequence_number
for event in hanat.events:
ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
UDP(sport=12346, dport=12345) /
- HANATStateSync(sequence_number=seq, flags='ACK'))
+ HANATStateSync(sequence_number=seq, flags='ACK',
+ thread_index=hanat.thread_index))
self.pg3.add_stream(ack)
self.pg_start()
- stats = self.statistics.get_counter('/nat44/ha/ack-recv')
- self.assertEqual(stats[0][0], 1)
+ stats = self.statistics['/nat44-ei/ha/ack-recv']
+ self.assertEqual(stats[:, 0].sum(), 1)
# delete one session
self.pg_enable_capture(self.pg_interfaces)
- self.vapi.nat44_del_session(address=self.pg0.remote_ip4,
- port=self.tcp_port_in,
- protocol=IP_PROTOS.tcp,
- flags=self.config_flags.NAT_IS_INSIDE)
- self.vapi.nat_ha_flush()
- stats = self.statistics.get_counter('/nat44/ha/del-event-send')
- self.assertEqual(stats[0][0], 1)
+ self.vapi.nat44_ei_del_session(
+ address=self.pg0.remote_ip4, port=self.tcp_port_in,
+ protocol=IP_PROTOS.tcp, flags=self.config_flags.NAT44_EI_IF_INSIDE)
+ self.vapi.nat44_ei_ha_flush()
+ stats = self.statistics['/nat44-ei/ha/del-event-send']
+ self.assertEqual(stats[:, 0].sum(), 1)
capture = self.pg3.get_capture(1)
p = capture[0]
try:
# do not send ACK, active retry send HA event again
self.pg_enable_capture(self.pg_interfaces)
sleep(12)
- stats = self.statistics.get_counter('/nat44/ha/retry-count')
- self.assertEqual(stats[0][0], 3)
- stats = self.statistics.get_counter('/nat44/ha/missed-count')
- self.assertEqual(stats[0][0], 1)
+ stats = self.statistics['/nat44-ei/ha/retry-count']
+ self.assertEqual(stats[:, 0].sum(), 3)
+ stats = self.statistics['/nat44-ei/ha/missed-count']
+ self.assertEqual(stats[:, 0].sum(), 1)
capture = self.pg3.get_capture(3)
for packet in capture:
self.assertEqual(packet, p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.pg0.get_capture(2)
- self.vapi.nat_ha_flush()
- stats = self.statistics.get_counter('/nat44/ha/refresh-event-send')
- self.assertEqual(stats[0][0], 2)
+ self.vapi.nat44_ei_ha_flush()
+ stats = self.statistics['/nat44-ei/ha/refresh-event-send']
+ self.assertEqual(stats[:, 0].sum(), 2)
capture = self.pg3.get_capture(1)
p = capture[0]
self.assert_packet_checksums_valid(p)
self.assertEqual(event.total_pkts, 2)
self.assertGreater(event.total_bytes, 0)
+ stats = self.statistics['/nat44-ei/ha/ack-recv']
ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
UDP(sport=12346, dport=12345) /
- HANATStateSync(sequence_number=seq, flags='ACK'))
+ HANATStateSync(sequence_number=seq, flags='ACK',
+ thread_index=hanat.thread_index))
self.pg3.add_stream(ack)
self.pg_start()
- stats = self.statistics.get_counter('/nat44/ha/ack-recv')
- self.assertEqual(stats[0][0], 2)
+ stats = self.statistics['/nat44-ei/ha/ack-recv']
+ self.assertEqual(stats[:, 0].sum(), 2)
def test_ha_recv(self):
""" NAT44EI Receive HA session synchronization events (passive) """
self.nat44_add_address(self.nat_addr)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- self.vapi.nat_ha_set_listener(ip_address=self.pg3.local_ip4,
- port=12345,
- path_mtu=512)
+ self.vapi.nat44_ei_ha_set_listener(ip_address=self.pg3.local_ip4,
+ port=12345, path_mtu=512)
bind_layers(UDP, HANATStateSync, sport=12345)
- self.tcp_port_out = random.randint(1025, 65535)
- self.udp_port_out = random.randint(1025, 65535)
+ # this is a bit tricky - HA dictates thread index due to how it's
+ # designed, but once we use HA to create a session, we also want
+ # to pass a packet through said session. so the session must end
+ # up on the correct thread from both directions - in2out (based on
+ # IP address) and out2in (based on outside port)
+
+ # first choose a thread index which is correct for IP
+ thread_index = get_nat44_ei_in2out_worker_index(self.pg0.remote_ip4,
+ self.vpp_worker_count)
+
+ # now pick a port which is correct for given thread
+ port_per_thread = int((0xffff-1024) / max(1, self.vpp_worker_count))
+ self.tcp_port_out = 1024 + random.randint(1, port_per_thread)
+ self.udp_port_out = 1024 + random.randint(1, port_per_thread)
+ if self.vpp_worker_count > 0:
+ self.tcp_port_out += port_per_thread * (thread_index - 1)
+ self.udp_port_out += port_per_thread * (thread_index - 1)
# send HA session add events to failover/passive
p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
eh_addr=self.pg1.remote_ip4,
ehn_addr=self.pg1.remote_ip4,
eh_port=self.udp_external_port,
- ehn_port=self.udp_external_port, fib_index=0)]))
+ ehn_port=self.udp_external_port, fib_index=0)],
+ thread_index=thread_index))
self.pg3.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.assertEqual(hanat.sequence_number, 1)
self.assertEqual(hanat.flags, 'ACK')
self.assertEqual(hanat.version, 1)
- self.assertEqual(hanat.thread_index, 0)
- stats = self.statistics.get_counter('/nat44/ha/ack-send')
- self.assertEqual(stats[0][0], 1)
- stats = self.statistics.get_counter('/nat44/ha/add-event-recv')
- self.assertEqual(stats[0][0], 2)
- users = self.statistics.get_counter('/nat44/total-users')
- self.assertEqual(users[0][0], 1)
- sessions = self.statistics.get_counter('/nat44/total-sessions')
- self.assertEqual(sessions[0][0], 2)
- users = self.vapi.nat44_user_dump()
+ self.assertEqual(hanat.thread_index, thread_index)
+ stats = self.statistics['/nat44-ei/ha/ack-send']
+ self.assertEqual(stats[:, 0].sum(), 1)
+ stats = self.statistics['/nat44-ei/ha/add-event-recv']
+ self.assertEqual(stats[:, 0].sum(), 2)
+ users = self.statistics['/nat44-ei/total-users']
+ self.assertEqual(users[:, 0].sum(), 1)
+ sessions = self.statistics['/nat44-ei/total-sessions']
+ self.assertEqual(sessions[:, 0].sum(), 2)
+ users = self.vapi.nat44_ei_user_dump()
self.assertEqual(len(users), 1)
self.assertEqual(str(users[0].ip_address),
self.pg0.remote_ip4)
# there should be 2 sessions created by HA
- sessions = self.vapi.nat44_user_session_dump(users[0].ip_address,
- users[0].vrf_id)
+ sessions = self.vapi.nat44_ei_user_session_dump(
+ users[0].ip_address, users[0].vrf_id)
self.assertEqual(len(sessions), 2)
for session in sessions:
self.assertEqual(str(session.inside_ip_address),
eh_addr=self.pg1.remote_ip4,
ehn_addr=self.pg1.remote_ip4,
eh_port=self.udp_external_port,
- ehn_port=self.udp_external_port, fib_index=0)]))
+ ehn_port=self.udp_external_port, fib_index=0)],
+ thread_index=thread_index))
self.pg3.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.assertEqual(hanat.sequence_number, 2)
self.assertEqual(hanat.flags, 'ACK')
self.assertEqual(hanat.version, 1)
- users = self.vapi.nat44_user_dump()
+ users = self.vapi.nat44_ei_user_dump()
self.assertEqual(len(users), 1)
self.assertEqual(str(users[0].ip_address),
self.pg0.remote_ip4)
# now we should have only 1 session, 1 deleted by HA
- sessions = self.vapi.nat44_user_session_dump(users[0].ip_address,
- users[0].vrf_id)
+ sessions = self.vapi.nat44_ei_user_session_dump(users[0].ip_address,
+ users[0].vrf_id)
self.assertEqual(len(sessions), 1)
- stats = self.statistics.get_counter('/nat44/ha/del-event-recv')
- self.assertEqual(stats[0][0], 1)
+ stats = self.statistics['/nat44-ei/ha/del-event-recv']
+ self.assertEqual(stats[:, 0].sum(), 1)
- stats = self.statistics.get_err_counter('/err/nat-ha/pkts-processed')
+ stats = self.statistics.get_err_counter(
+ '/err/nat44-ei-ha/pkts-processed')
self.assertEqual(stats, 2)
# send HA session refresh event to failover/passive
ehn_addr=self.pg1.remote_ip4,
eh_port=self.tcp_external_port,
ehn_port=self.tcp_external_port, fib_index=0,
- total_bytes=1024, total_pkts=2)]))
+ total_bytes=1024, total_pkts=2)],
+ thread_index=thread_index))
self.pg3.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertEqual(hanat.sequence_number, 3)
self.assertEqual(hanat.flags, 'ACK')
self.assertEqual(hanat.version, 1)
- users = self.vapi.nat44_user_dump()
+ users = self.vapi.nat44_ei_user_dump()
self.assertEqual(len(users), 1)
self.assertEqual(str(users[0].ip_address),
self.pg0.remote_ip4)
- sessions = self.vapi.nat44_user_session_dump(users[0].ip_address,
- users[0].vrf_id)
+ sessions = self.vapi.nat44_ei_user_session_dump(
+ users[0].ip_address, users[0].vrf_id)
self.assertEqual(len(sessions), 1)
session = sessions[0]
self.assertEqual(session.total_bytes, 1024)
self.assertEqual(session.total_pkts, 2)
- stats = self.statistics.get_counter('/nat44/ha/refresh-event-recv')
- self.assertEqual(stats[0][0], 1)
+ stats = self.statistics['/nat44-ei/ha/refresh-event-recv']
+ self.assertEqual(stats[:, 0].sum(), 1)
- stats = self.statistics.get_err_counter('/err/nat-ha/pkts-processed')
+ stats = self.statistics.get_err_counter(
+ '/err/nat44-ei-ha/pkts-processed')
self.assertEqual(stats, 3)
# send packet to test session created by HA
self.assertEqual(tcp.sport, self.tcp_external_port)
self.assertEqual(tcp.dport, self.tcp_port_in)
+ def reconfigure_frame_queue_nelts(self, frame_queue_nelts):
+ self.vapi.nat44_ei_plugin_enable_disable(enable=0)
+ self.vapi.nat44_ei_set_fq_options(frame_queue_nelts=frame_queue_nelts)
+ # keep plugin configuration persistent
+ self.plugin_enable()
+ return self.vapi.nat44_ei_show_fq_options().frame_queue_nelts
+
+ def test_set_frame_queue_nelts(self):
+ """ NAT44 EI API test - worker handoff frame queue elements """
+ self.assertEqual(self.reconfigure_frame_queue_nelts(512), 512)
+
def show_commands_at_teardown(self):
- self.logger.info(self.vapi.cli("show nat44 addresses"))
- self.logger.info(self.vapi.cli("show nat44 interfaces"))
- self.logger.info(self.vapi.cli("show nat44 static mappings"))
- self.logger.info(self.vapi.cli("show nat44 interface address"))
- self.logger.info(self.vapi.cli("show nat44 sessions detail"))
- self.logger.info(self.vapi.cli("show nat44 hash tables detail"))
- self.logger.info(self.vapi.cli("show nat timeouts"))
+ self.logger.info(self.vapi.cli("show nat44 ei timeouts"))
+ self.logger.info(self.vapi.cli("show nat44 ei addresses"))
+ self.logger.info(self.vapi.cli("show nat44 ei interfaces"))
+ self.logger.info(self.vapi.cli("show nat44 ei static mappings"))
+ self.logger.info(self.vapi.cli("show nat44 ei interface address"))
+ self.logger.info(self.vapi.cli("show nat44 ei sessions detail"))
+ self.logger.info(self.vapi.cli("show nat44 ei hash tables detail"))
+ self.logger.info(self.vapi.cli("show nat44 ei ha"))
self.logger.info(
- self.vapi.cli("show nat addr-port-assignment-alg"))
- self.logger.info(self.vapi.cli("show nat ha"))
+ self.vapi.cli("show nat44 ei addr-port-assignment-alg"))
+
+ def test_outside_address_distribution(self):
+ """ Outside address distribution based on source address """
+
+ x = 100
+ nat_addresses = []
+
+ for i in range(1, x):
+ a = "10.0.0.%d" % i
+ nat_addresses.append(a)
+
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
+ sw_if_index=self.pg0.sw_if_index,
+ flags=flags, is_add=1)
+ self.vapi.nat44_ei_interface_add_del_feature(
+ sw_if_index=self.pg1.sw_if_index,
+ is_add=1)
+
+ self.vapi.nat44_ei_add_del_address_range(
+ first_ip_address=nat_addresses[0],
+ last_ip_address=nat_addresses[-1],
+ vrf_id=0xFFFFFFFF, is_add=1)
+
+ self.pg0.generate_remote_hosts(x)
+
+ pkts = []
+ for i in range(x):
+ info = self.create_packet_info(self.pg0, self.pg1)
+ payload = self.info_to_payload(info)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=self.pg0.remote_hosts[i].ip4,
+ dst=self.pg1.remote_ip4) /
+ UDP(sport=7000+i, dport=8000+i) /
+ Raw(payload))
+ info.data = p
+ pkts.append(p)
+
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ recvd = self.pg1.get_capture(len(pkts))
+ for p_recvd in recvd:
+ payload_info = self.payload_to_info(p_recvd[Raw])
+ packet_index = payload_info.index
+ info = self._packet_infos[packet_index]
+ self.assertTrue(info is not None)
+ self.assertEqual(packet_index, info.index)
+ p_sent = info.data
+ packed = socket.inet_aton(p_sent[IP].src)
+ numeric = struct.unpack("!L", packed)[0]
+ numeric = socket.htonl(numeric)
+ a = nat_addresses[(numeric-1) % len(nat_addresses)]
+ self.assertEqual(
+ a, p_recvd[IP].src,
+ "Invalid packet (src IP %s translated to %s, but expected %s)"
+ % (p_sent[IP].src, p_recvd[IP].src, a))
class TestNAT44Out2InDPO(MethodHolder):
@classmethod
def setUpClass(cls):
super(TestNAT44Out2InDPO, cls).setUpClass()
- cls.vapi.cli("set log class nat level debug")
+ cls.vapi.cli("set log class nat44-ei level debug")
cls.tcp_port_in = 6303
cls.tcp_port_out = 6303
def setUp(self):
super(TestNAT44Out2InDPO, self).setUp()
- flags = self.nat44_config_flags.NAT44_API_IS_OUT2IN_DPO
- self.vapi.nat44_plugin_enable_disable(enable=1, flags=flags)
+ flags = self.config_flags.NAT44_EI_OUT2IN_DPO
+ self.vapi.nat44_ei_plugin_enable_disable(enable=1, flags=flags)
def tearDown(self):
super(TestNAT44Out2InDPO, self).tearDown()
if not self.vpp_dead:
- self.vapi.nat44_plugin_enable_disable(enable=0)
+ self.vapi.nat44_ei_plugin_enable_disable(enable=0)
self.vapi.cli("clear logging")
def configure_xlat(self):
def test_464xlat_ce(self):
""" Test 464XLAT CE with NAT44EI """
- nat_config = self.vapi.nat_show_config()
- self.assertEqual(1, nat_config.out2in_dpo)
-
self.configure_xlat()
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags, is_add=1)
- self.vapi.nat44_add_del_address_range(first_ip_address=self.nat_addr_n,
- last_ip_address=self.nat_addr_n,
- vrf_id=0xFFFFFFFF, is_add=1)
+ self.vapi.nat44_ei_add_del_address_range(
+ first_ip_address=self.nat_addr_n,
+ last_ip_address=self.nat_addr_n,
+ vrf_id=0xFFFFFFFF, is_add=1)
out_src_ip6 = self.compose_ip6(self.dst_ip4, self.dst_ip6_pfx,
self.dst_ip6_pfx_len)
capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
finally:
- self.vapi.nat44_interface_add_del_feature(
+ self.vapi.nat44_ei_interface_add_del_feature(
sw_if_index=self.pg0.sw_if_index,
flags=flags)
- self.vapi.nat44_add_del_address_range(
+ self.vapi.nat44_ei_add_del_address_range(
first_ip_address=self.nat_addr_n,
last_ip_address=self.nat_addr_n,
vrf_id=0xFFFFFFFF)
self.verify_capture_in(capture, self.pg0)
+class TestNAT44EIMW(MethodHolder):
+ """ NAT44EI Test Cases (multiple workers) """
+ vpp_worker_count = 2
+ max_translations = 10240
+ max_users = 10240
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestNAT44EIMW, cls).setUpClass()
+ cls.vapi.cli("set log class nat level debug")
+
+ cls.tcp_port_in = 6303
+ cls.tcp_port_out = 6303
+ cls.udp_port_in = 6304
+ cls.udp_port_out = 6304
+ cls.icmp_id_in = 6305
+ cls.icmp_id_out = 6305
+ cls.nat_addr = '10.0.0.3'
+ cls.ipfix_src_port = 4739
+ cls.ipfix_domain_id = 1
+ cls.tcp_external_port = 80
+ cls.udp_external_port = 69
+
+ cls.create_pg_interfaces(range(10))
+ cls.interfaces = list(cls.pg_interfaces[0:4])
+
+ for i in cls.interfaces:
+ i.admin_up()
+ i.config_ip4()
+ i.resolve_arp()
+
+ cls.pg0.generate_remote_hosts(3)
+ cls.pg0.configure_ipv4_neighbors()
+
+ cls.pg1.generate_remote_hosts(1)
+ cls.pg1.configure_ipv4_neighbors()
+
+ cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
+ cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
+ cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
+
+ cls.pg4._local_ip4 = "172.16.255.1"
+ cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
+ cls.pg4.set_table_ip4(10)
+ cls.pg5._local_ip4 = "172.17.255.3"
+ cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
+ cls.pg5.set_table_ip4(10)
+ cls.pg6._local_ip4 = "172.16.255.1"
+ cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
+ cls.pg6.set_table_ip4(20)
+ for i in cls.overlapping_interfaces:
+ i.config_ip4()
+ i.admin_up()
+ i.resolve_arp()
+
+ cls.pg7.admin_up()
+ cls.pg8.admin_up()
+
+ cls.pg9.generate_remote_hosts(2)
+ cls.pg9.config_ip4()
+ cls.vapi.sw_interface_add_del_address(
+ sw_if_index=cls.pg9.sw_if_index,
+ prefix="10.0.0.1/24")
+
+ cls.pg9.admin_up()
+ cls.pg9.resolve_arp()
+ cls.pg9._remote_hosts[1]._ip4 = cls.pg9._remote_hosts[0]._ip4
+ cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
+ cls.pg9.resolve_arp()
+
+ def setUp(self):
+ super(TestNAT44EIMW, self).setUp()
+ self.vapi.nat44_ei_plugin_enable_disable(
+ sessions=self.max_translations,
+ users=self.max_users, enable=1)
+
+ def tearDown(self):
+ super(TestNAT44EIMW, self).tearDown()
+ if not self.vpp_dead:
+ self.vapi.nat44_ei_ipfix_enable_disable(
+ domain_id=self.ipfix_domain_id,
+ src_port=self.ipfix_src_port,
+ enable=0)
+ self.ipfix_src_port = 4739
+ self.ipfix_domain_id = 1
+
+ self.vapi.nat44_ei_plugin_enable_disable(enable=0)
+ self.vapi.cli("clear logging")
+
+ def test_hairpinning(self):
+ """ NAT44EI hairpinning - 1:1 NAPT """
+
+ host = self.pg0.remote_hosts[0]
+ server = self.pg0.remote_hosts[1]
+ host_in_port = 1234
+ host_out_port = 0
+ server_in_port = 5678
+ server_out_port = 8765
+ worker_1 = 1
+ worker_2 = 2
+
+ self.nat44_add_address(self.nat_addr)
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
+ sw_if_index=self.pg0.sw_if_index,
+ flags=flags, is_add=1)
+ self.vapi.nat44_ei_interface_add_del_feature(
+ sw_if_index=self.pg1.sw_if_index,
+ is_add=1)
+
+ # add static mapping for server
+ self.nat44_add_static_mapping(server.ip4, self.nat_addr,
+ server_in_port, server_out_port,
+ proto=IP_PROTOS.tcp)
+
+ cnt = self.statistics['/nat44-ei/hairpinning']
+ # send packet from host to server
+ p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
+ 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)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ 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)
+ self.assert_packet_checksums_valid(p)
+ host_out_port = tcp.sport
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ after = self.statistics['/nat44-ei/hairpinning']
+
+ if_idx = self.pg0.sw_if_index
+ self.assertEqual(after[worker_2][if_idx] - cnt[worker_1][if_idx], 1)
+
+ # send reply from server to host
+ p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
+ 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)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ 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)
+ self.assert_packet_checksums_valid(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ after = self.statistics['/nat44-ei/hairpinning']
+ if_idx = self.pg0.sw_if_index
+ self.assertEqual(after[worker_1][if_idx] - cnt[worker_1][if_idx], 1)
+ self.assertEqual(after[worker_2][if_idx] - cnt[worker_2][if_idx], 2)
+
+ def test_hairpinning2(self):
+ """ NAT44EI hairpinning - 1:1 NAT"""
+
+ server1_nat_ip = "10.0.0.10"
+ server2_nat_ip = "10.0.0.11"
+ host = self.pg0.remote_hosts[0]
+ server1 = self.pg0.remote_hosts[1]
+ server2 = self.pg0.remote_hosts[2]
+ server_tcp_port = 22
+ server_udp_port = 20
+
+ self.nat44_add_address(self.nat_addr)
+ flags = self.config_flags.NAT44_EI_IF_INSIDE
+ self.vapi.nat44_ei_interface_add_del_feature(
+ sw_if_index=self.pg0.sw_if_index,
+ flags=flags, is_add=1)
+ self.vapi.nat44_ei_interface_add_del_feature(
+ sw_if_index=self.pg1.sw_if_index,
+ is_add=1)
+
+ # add static mapping for servers
+ self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
+ self.nat44_add_static_mapping(server2.ip4, server2_nat_ip)
+
+ # host to server1
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=host.ip4, dst=server1_nat_ip) /
+ TCP(sport=self.tcp_port_in, dport=server_tcp_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=host.ip4, dst=server1_nat_ip) /
+ UDP(sport=self.udp_port_in, dport=server_udp_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=host.ip4, dst=server1_nat_ip) /
+ ICMP(id=self.icmp_id_in, type='echo-request'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ 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)
+ self.assertEqual(packet[TCP].dport, server_tcp_port)
+ self.tcp_port_out = packet[TCP].sport
+ self.assert_packet_checksums_valid(packet)
+ elif packet.haslayer(UDP):
+ self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
+ self.assertEqual(packet[UDP].dport, server_udp_port)
+ self.udp_port_out = packet[UDP].sport
+ else:
+ self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
+ self.icmp_id_out = packet[ICMP].id
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # server1 to host
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ 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.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.nat_addr) /
+ ICMP(id=self.icmp_id_out, type='echo-reply'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IP].src, server1_nat_ip)
+ self.assertEqual(packet[IP].dst, host.ip4)
+ if packet.haslayer(TCP):
+ self.assertEqual(packet[TCP].dport, self.tcp_port_in)
+ self.assertEqual(packet[TCP].sport, server_tcp_port)
+ self.assert_packet_checksums_valid(packet)
+ elif packet.haslayer(UDP):
+ self.assertEqual(packet[UDP].dport, self.udp_port_in)
+ self.assertEqual(packet[UDP].sport, server_udp_port)
+ else:
+ self.assertEqual(packet[ICMP].id, self.icmp_id_in)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # server2 to server1
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server2.ip4, dst=server1_nat_ip) /
+ TCP(sport=self.tcp_port_in, dport=server_tcp_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server2.ip4, dst=server1_nat_ip) /
+ UDP(sport=self.udp_port_in, dport=server_udp_port))
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server2.ip4, dst=server1_nat_ip) /
+ ICMP(id=self.icmp_id_in, type='echo-request'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IP].src, server2_nat_ip)
+ self.assertEqual(packet[IP].dst, server1.ip4)
+ if packet.haslayer(TCP):
+ self.assertEqual(packet[TCP].sport, self.tcp_port_in)
+ self.assertEqual(packet[TCP].dport, server_tcp_port)
+ self.tcp_port_out = packet[TCP].sport
+ self.assert_packet_checksums_valid(packet)
+ elif packet.haslayer(UDP):
+ self.assertEqual(packet[UDP].sport, self.udp_port_in)
+ self.assertEqual(packet[UDP].dport, server_udp_port)
+ self.udp_port_out = packet[UDP].sport
+ else:
+ self.assertEqual(packet[ICMP].id, self.icmp_id_in)
+ self.icmp_id_out = packet[ICMP].id
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # server1 to server2
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=server1.ip4, dst=server2_nat_ip) /
+ 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=server2_nat_ip) /
+ 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=server2_nat_ip) /
+ ICMP(id=self.icmp_id_out, type='echo-reply'))
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IP].src, server1_nat_ip)
+ self.assertEqual(packet[IP].dst, server2.ip4)
+ if packet.haslayer(TCP):
+ self.assertEqual(packet[TCP].dport, self.tcp_port_in)
+ self.assertEqual(packet[TCP].sport, server_tcp_port)
+ self.assert_packet_checksums_valid(packet)
+ elif packet.haslayer(UDP):
+ self.assertEqual(packet[UDP].dport, self.udp_port_in)
+ self.assertEqual(packet[UDP].sport, server_udp_port)
+ else:
+ self.assertEqual(packet[ICMP].id, self.icmp_id_in)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff