+ def test_frag_in_order(self):
+ """ NAT64 translate fragments arriving in order """
+ self.tcp_port_in = random.randint(1025, 65535)
+
+ 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)
+
+ reass = self.vapi.nat_reass_dump()
+ reass_n_start = len(reass)
+
+ # in2out
+ data = 'a' * 200
+ pkts = self.create_stream_frag_ip6(self.pg0, self.pg1.remote_ip4,
+ self.tcp_port_in, 20, data)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ frags = self.pg1.get_capture(len(pkts))
+ p = self.reass_frags_and_verify(frags,
+ self.nat_addr,
+ self.pg1.remote_ip4)
+ self.assertEqual(p[TCP].dport, 20)
+ self.assertNotEqual(p[TCP].sport, self.tcp_port_in)
+ self.tcp_port_out = p[TCP].sport
+ self.assertEqual(data, p[Raw].load)
+
+ # out2in
+ data = "A" * 4 + "b" * 16 + "C" * 3
+ pkts = self.create_stream_frag(self.pg1,
+ self.nat_addr,
+ 20,
+ self.tcp_port_out,
+ data)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ frags = self.pg0.get_capture(len(pkts))
+ src = self.compose_ip6(self.pg1.remote_ip4, '64:ff9b::', 96)
+ p = self.reass_frags_and_verify_ip6(frags, src, self.pg0.remote_ip6)
+ self.assertEqual(p[TCP].sport, 20)
+ self.assertEqual(p[TCP].dport, self.tcp_port_in)
+ self.assertEqual(data, p[Raw].load)
+
+ reass = self.vapi.nat_reass_dump()
+ reass_n_end = len(reass)
+
+ self.assertEqual(reass_n_end - reass_n_start, 2)
+
+ def test_reass_hairpinning(self):
+ """ NAT64 fragments hairpinning """
+ data = 'a' * 200
+ client = self.pg0.remote_hosts[0]
+ server = self.pg0.remote_hosts[1]
+ server_in_port = random.randint(1025, 65535)
+ server_out_port = random.randint(1025, 65535)
+ client_in_port = random.randint(1025, 65535)
+ ip = IPv6(src=''.join(['64:ff9b::', self.nat_addr]))
+ nat_addr_ip6 = ip.src
+
+ 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)
+
+ # add static BIB entry for server
+ self.vapi.nat64_add_del_static_bib(server.ip6n,
+ self.nat_addr_n,
+ server_in_port,
+ server_out_port,
+ IP_PROTOS.tcp)
+
+ # send packet from host to server
+ pkts = self.create_stream_frag_ip6(self.pg0,
+ self.nat_addr,
+ client_in_port,
+ server_out_port,
+ data)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ frags = self.pg0.get_capture(len(pkts))
+ p = self.reass_frags_and_verify_ip6(frags, nat_addr_ip6, server.ip6)
+ self.assertNotEqual(p[TCP].sport, client_in_port)
+ self.assertEqual(p[TCP].dport, server_in_port)
+ self.assertEqual(data, p[Raw].load)
+
+ def test_frag_out_of_order(self):
+ """ NAT64 translate fragments arriving out of order """
+ self.tcp_port_in = random.randint(1025, 65535)
+
+ 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)
+
+ # in2out
+ data = 'a' * 200
+ pkts = self.create_stream_frag_ip6(self.pg0, self.pg1.remote_ip4,
+ self.tcp_port_in, 20, data)
+ pkts.reverse()
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ frags = self.pg1.get_capture(len(pkts))
+ p = self.reass_frags_and_verify(frags,
+ self.nat_addr,
+ self.pg1.remote_ip4)
+ self.assertEqual(p[TCP].dport, 20)
+ self.assertNotEqual(p[TCP].sport, self.tcp_port_in)
+ self.tcp_port_out = p[TCP].sport
+ self.assertEqual(data, p[Raw].load)
+
+ # out2in
+ data = "A" * 4 + "B" * 16 + "C" * 3
+ pkts = self.create_stream_frag(self.pg1,
+ self.nat_addr,
+ 20,
+ self.tcp_port_out,
+ data)
+ pkts.reverse()
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ frags = self.pg0.get_capture(len(pkts))
+ src = self.compose_ip6(self.pg1.remote_ip4, '64:ff9b::', 96)
+ p = self.reass_frags_and_verify_ip6(frags, src, self.pg0.remote_ip6)
+ self.assertEqual(p[TCP].sport, 20)
+ self.assertEqual(p[TCP].dport, self.tcp_port_in)
+ self.assertEqual(data, p[Raw].load)
+
+ def test_interface_addr(self):
+ """ Acquire NAT64 pool addresses from interface """
+ self.vapi.nat64_add_interface_addr(self.pg4.sw_if_index)
+
+ # no address in NAT64 pool
+ adresses = self.vapi.nat44_address_dump()
+ self.assertEqual(0, len(adresses))
+
+ # configure interface address and check NAT64 address pool
+ self.pg4.config_ip4()
+ addresses = self.vapi.nat64_pool_addr_dump()
+ self.assertEqual(len(addresses), 1)
+ self.assertEqual(addresses[0].address, self.pg4.local_ip4n)
+
+ # remove interface address and check NAT64 address pool
+ self.pg4.unconfig_ip4()
+ addresses = self.vapi.nat64_pool_addr_dump()
+ self.assertEqual(0, len(adresses))
+