return pkts
- def create_stream_in_ip6(self, in_if, out_if, hlim=64):
+ def compose_ip6(self, ip4, pref, plen):
+ """
+ Compose IPv4-embedded IPv6 addresses
+
+ :param ip4: IPv4 address
+ :param pref: IPv6 prefix
+ :param plen: IPv6 prefix length
+ :returns: IPv4-embedded IPv6 addresses
+ """
+ pref_n = list(socket.inet_pton(socket.AF_INET6, pref))
+ ip4_n = list(socket.inet_pton(socket.AF_INET, ip4))
+ if plen == 32:
+ pref_n[4] = ip4_n[0]
+ pref_n[5] = ip4_n[1]
+ pref_n[6] = ip4_n[2]
+ pref_n[7] = ip4_n[3]
+ elif plen == 40:
+ pref_n[5] = ip4_n[0]
+ pref_n[6] = ip4_n[1]
+ pref_n[7] = ip4_n[2]
+ pref_n[9] = ip4_n[3]
+ elif plen == 48:
+ pref_n[6] = ip4_n[0]
+ pref_n[7] = ip4_n[1]
+ pref_n[9] = ip4_n[2]
+ pref_n[10] = ip4_n[3]
+ elif plen == 56:
+ pref_n[7] = ip4_n[0]
+ pref_n[9] = ip4_n[1]
+ pref_n[10] = ip4_n[2]
+ pref_n[11] = ip4_n[3]
+ elif plen == 64:
+ pref_n[9] = ip4_n[0]
+ pref_n[10] = ip4_n[1]
+ pref_n[11] = ip4_n[2]
+ pref_n[12] = ip4_n[3]
+ elif plen == 96:
+ pref_n[12] = ip4_n[0]
+ pref_n[13] = ip4_n[1]
+ pref_n[14] = ip4_n[2]
+ pref_n[15] = ip4_n[3]
+ return socket.inet_ntop(socket.AF_INET6, ''.join(pref_n))
+
+ def create_stream_in_ip6(self, in_if, out_if, hlim=64, pref=None, plen=0):
"""
Create IPv6 packet stream for inside network
:param in_if: Inside interface
:param out_if: Outside interface
:param ttl: Hop Limit of generated packets
+ :param pref: NAT64 prefix
+ :param plen: NAT64 prefix length
"""
pkts = []
- dst = ''.join(['64:ff9b::', out_if.remote_ip4])
+ if pref is None:
+ dst = ''.join(['64:ff9b::', out_if.remote_ip4])
+ else:
+ dst = self.compose_ip6(out_if.remote_ip4, pref, plen)
+
# TCP
p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
intf.is_inside,
is_add=0)
+ interfaces = self.vapi.snat_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)
+
static_mappings = self.vapi.snat_static_mapping_dump()
for sm in static_mappings:
self.vapi.snat_add_static_mapping(sm.local_ip_address,
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
- def test_hairpinning_unknown_proto(self):
+ def test_hairpinning_static_unknown_proto(self):
""" 1:1 NAT translate packet with unknown protocol - hairpinning """
host = self.pg0.remote_hosts[0]
self.logger.error(ppp("Unexpected or invalid packet:", packet))
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)
+
+ # in2out
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ TCP(sport=self.tcp_port_in, dport=20))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg1.get_capture(1)
+
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
+ GRE() /
+ IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg1.get_capture(1)
+ packet = p[0]
+ try:
+ self.assertEqual(packet[IP].src, self.snat_addr)
+ self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
+ self.assertTrue(packet.haslayer(GRE))
+ self.check_ip_checksum(packet)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # out2in
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
+ GRE() /
+ IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg0.get_capture(1)
+ packet = p[0]
+ try:
+ self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
+ self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
+ self.assertTrue(packet.haslayer(GRE))
+ self.check_ip_checksum(packet)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ def test_hairpinning_unknown_proto(self):
+ """ SNAT translate packet with unknown protocol - hairpinning """
+ 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
+ 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)
+
+ # add static mapping for server
+ self.snat_add_static_mapping(server.ip4, server_nat_ip)
+
+ # host to server
+ p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
+ IP(src=host.ip4, dst=server_nat_ip) /
+ 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 = (Ether(dst=self.pg0.local_mac, src=host.mac) /
+ IP(src=host.ip4, dst=server_nat_ip) /
+ GRE() /
+ IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg0.get_capture(1)
+ packet = p[0]
+ try:
+ self.assertEqual(packet[IP].src, self.snat_addr)
+ self.assertEqual(packet[IP].dst, server.ip4)
+ self.assertTrue(packet.haslayer(GRE))
+ self.check_ip_checksum(packet)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # server to host
+ p = (Ether(dst=self.pg0.local_mac, src=server.mac) /
+ IP(src=server.ip4, dst=self.snat_addr) /
+ GRE() /
+ IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg0.get_capture(1)
+ packet = p[0]
+ try:
+ self.assertEqual(packet[IP].src, server_nat_ip)
+ self.assertEqual(packet[IP].dst, host.ip4)
+ self.assertTrue(packet.haslayer(GRE))
+ self.check_ip_checksum(packet)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ 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)
+
+ # in2out
+ pkts = self.create_stream_in(self.pg0, self.pg1)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture)
+
+ # out2in
+ pkts = self.create_stream_out(self.pg1)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg0)
+
+ def test_output_feature_vrf_aware(self):
+ """ S-NAT interface output feature VRF aware (in2out postrouting) """
+ nat_ip_vrf10 = "10.0.0.10"
+ nat_ip_vrf20 = "10.0.0.20"
+
+ self.vapi.ip_add_del_route(dst_address=self.pg3.remote_ip4n,
+ dst_address_length=32,
+ next_hop_address=self.pg3.remote_ip4n,
+ next_hop_sw_if_index=self.pg3.sw_if_index,
+ table_id=10)
+ self.vapi.ip_add_del_route(dst_address=self.pg3.remote_ip4n,
+ dst_address_length=32,
+ next_hop_address=self.pg3.remote_ip4n,
+ 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)
+
+ # in2out VRF 10
+ pkts = self.create_stream_in(self.pg4, self.pg3)
+ self.pg4.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg3.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip=nat_ip_vrf10)
+
+ # out2in VRF 10
+ pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf10)
+ self.pg3.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg4.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg4)
+
+ # in2out VRF 20
+ pkts = self.create_stream_in(self.pg6, self.pg3)
+ self.pg6.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg3.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip=nat_ip_vrf20)
+
+ # out2in VRF 20
+ pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf20)
+ self.pg3.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg6.get_capture(len(pkts))
+ self.verify_capture_in(capture, self.pg6)
+
+ def test_output_feature_hairpinning(self):
+ """ S-NAT interface output feature hairpinning (in2out postrouting) """
+ 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
+
+ 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)
+
+ # 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)
+
+ # send packet from host to server
+ p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
+ IP(src=host.ip4, dst=self.snat_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.snat_addr)
+ self.assertEqual(ip.dst, server.ip4)
+ self.assertNotEqual(tcp.sport, host_in_port)
+ self.assertEqual(tcp.dport, server_in_port)
+ self.check_tcp_checksum(p)
+ host_out_port = tcp.sport
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ # send reply from server to host
+ p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
+ IP(src=server.ip4, dst=self.snat_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.snat_addr)
+ self.assertEqual(ip.dst, host.ip4)
+ self.assertEqual(tcp.sport, server_out_port)
+ self.assertEqual(tcp.dport, host_in_port)
+ self.check_tcp_checksum(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:"), p)
+ raise
+
def tearDown(self):
super(TestSNAT, self).tearDown()
if not self.vpp_dead:
dms = self.vapi.snat_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,
# verify IPFIX logging
self.vapi.cli("ipfix flush") # FIXME this should be an API call
+ sleep(1)
capture = self.pg2.get_capture(2)
ipfix = IPFIXDecoder()
# first load template
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
+ def test_prefix(self):
+ """ NAT64 Network-Specific Prefix """
+
+ self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
+ self.nat_addr_n)
+ self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
+ self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
+ self.vapi.nat64_add_del_pool_addr_range(self.vrf1_nat_addr_n,
+ self.vrf1_nat_addr_n,
+ vrf_id=self.vrf1_id)
+ self.vapi.nat64_add_del_interface(self.pg2.sw_if_index)
+
+ # Add global prefix
+ global_pref64 = "2001:db8::"
+ global_pref64_n = socket.inet_pton(socket.AF_INET6, global_pref64)
+ global_pref64_len = 32
+ self.vapi.nat64_add_del_prefix(global_pref64_n, global_pref64_len)
+
+ prefix = self.vapi.nat64_prefix_dump()
+ self.assertEqual(len(prefix), 1)
+ self.assertEqual(prefix[0].prefix, global_pref64_n)
+ self.assertEqual(prefix[0].prefix_len, global_pref64_len)
+ self.assertEqual(prefix[0].vrf_id, 0)
+
+ # Add tenant specific prefix
+ vrf1_pref64 = "2001:db8:122:300::"
+ vrf1_pref64_n = socket.inet_pton(socket.AF_INET6, vrf1_pref64)
+ vrf1_pref64_len = 56
+ self.vapi.nat64_add_del_prefix(vrf1_pref64_n,
+ vrf1_pref64_len,
+ vrf_id=self.vrf1_id)
+ prefix = self.vapi.nat64_prefix_dump()
+ self.assertEqual(len(prefix), 2)
+
+ # Global prefix
+ pkts = self.create_stream_in_ip6(self.pg0,
+ self.pg1,
+ pref=global_pref64,
+ plen=global_pref64_len)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip=self.nat_addr,
+ dst_ip=self.pg1.remote_ip4)
+
+ pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ dst_ip = self.compose_ip6(self.pg1.remote_ip4,
+ global_pref64,
+ global_pref64_len)
+ self.verify_capture_in_ip6(capture, dst_ip, self.pg0.remote_ip6)
+
+ # Tenant specific prefix
+ pkts = self.create_stream_in_ip6(self.pg2,
+ self.pg1,
+ pref=vrf1_pref64,
+ plen=vrf1_pref64_len)
+ self.pg2.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ self.verify_capture_out(capture, nat_ip=self.vrf1_nat_addr,
+ dst_ip=self.pg1.remote_ip4)
+
+ pkts = self.create_stream_out(self.pg1, dst_ip=self.vrf1_nat_addr)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg2.get_capture(len(pkts))
+ dst_ip = self.compose_ip6(self.pg1.remote_ip4,
+ vrf1_pref64,
+ vrf1_pref64_len)
+ self.verify_capture_in_ip6(capture, dst_ip, self.pg2.remote_ip6)
+
+ def _test_unknown_proto(self):
+ """ NAT64 translate packet with unknown protocol """
+
+ self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
+ self.nat_addr_n)
+ self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
+ self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
+ remote_ip6 = self.compose_ip6(self.pg1.remote_ip4, '64:ff9b::', 96)
+
+ # in2out
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=remote_ip6) /
+ TCP(sport=self.tcp_port_in, dport=20))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg1.get_capture(1)
+
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=remote_ip6) /
+ GRE() /
+ IP(src=self.pg2.local_ip4, dst=self.pg2.remote_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg1.get_capture(1)
+ packet = p[0]
+ try:
+ 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)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # out2in
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
+ GRE() /
+ IP(src=self.pg2.remote_ip4, dst=self.pg2.local_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg0.get_capture(1)
+ packet = p[0]
+ try:
+ self.assertEqual(packet[IPv6].src, remote_ip6)
+ self.assertEqual(packet[IPv6].dst, self.pgi0.remote_ip6)
+ self.assertTrue(packet.haslayer(GRE))
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ def _test_hairpinning_unknown_proto(self):
+ """ NAT64 translate packet with unknown protocol - hairpinning """
+
+ client = self.pg0.remote_hosts[0]
+ server = self.pg0.remote_hosts[1]
+ server_tcp_in_port = 22
+ server_tcp_out_port = 4022
+ client_tcp_in_port = 1234
+ client_udp_in_port = 1235
+ nat_addr_ip6 = self.compose_ip6(self.nat_addr, '64:ff9b::', 96)
+
+ self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
+ self.nat_addr_n)
+ self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
+ self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
+
+ self.vapi.nat64_add_del_static_bib(server.ip6n,
+ self.nat_addr_n,
+ server_tcp_in_port,
+ server_tcp_out_port,
+ IP_PROTOS.tcp)
+
+ # client to server
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=client.ip6, dst=nat_addr_ip6) /
+ TCP(sport=client_tcp_in_port, dport=server_tcp_out_port))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg0.get_capture(1)
+
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=client.ip6, dst=nat_addr_ip6) /
+ GRE() /
+ IP(src=self.pg2.local_ip4, dst=self.pg2.remote_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg0.get_capture(1)
+ packet = p[0]
+ try:
+ self.assertEqual(packet[IPv6].src, nat_addr_ip6)
+ self.assertEqual(packet[IPv6].dst, server.ip6)
+ self.assertTrue(packet.haslayer(GRE))
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # server to client
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=server.ip6, dst=nat_addr_ip6) /
+ GRE() /
+ IP(src=self.pg2.remote_ip4, dst=self.pg2.local_ip4) /
+ TCP(sport=1234, dport=1234))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ p = self.pg0.get_capture(1)
+ packet = p[0]
+ try:
+ self.assertEqual(packet[IPv6].src, nat_addr_ip6)
+ self.assertEqual(packet[IPv6].dst, client.ip6)
+ self.assertTrue(packet.haslayer(GRE))
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
def nat64_get_ses_num(self):
"""
Return number of active NAT64 sessions.
vrf_id=addr.vrf_id,
is_add=0)
+ prefixes = self.vapi.nat64_prefix_dump()
+ for prefix in prefixes:
+ self.vapi.nat64_add_del_prefix(prefix.prefix,
+ prefix.prefix_len,
+ vrf_id=prefix.vrf_id,
+ is_add=0)
+
def tearDown(self):
super(TestNAT64, self).tearDown()
if not self.vpp_dead:
self.logger.info(self.vapi.cli("show nat64 pool"))
self.logger.info(self.vapi.cli("show nat64 interfaces"))
+ self.logger.info(self.vapi.cli("show nat64 prefix"))
self.logger.info(self.vapi.cli("show nat64 bib tcp"))
self.logger.info(self.vapi.cli("show nat64 bib udp"))
self.logger.info(self.vapi.cli("show nat64 bib icmp"))
Code Review / vpp.git / blobdiff