NAT64 to use IPv4 address from interface (VPP-1051)

[vpp.git] / test / test_nat.py

diff --git a/test/test_nat.py b/test/test_nat.py
index 2afa1dd..76c5282 100644 (file)
--- a/test/test_nat.py
+++ b/test/test_nat.py
@@ -3,19 +3,24 @@
 import socket
 import unittest
 import struct
 import socket
 import unittest
 import struct

+import StringIO
+import random

 
 from framework import VppTestCase, VppTestRunner, running_extended_tests
 
 from framework import VppTestCase, VppTestRunner, running_extended_tests

+from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto

 from scapy.layers.inet import IP, TCP, UDP, ICMP
 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
 from scapy.layers.inet import IP, TCP, UDP, ICMP
 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply

-from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6
+from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment

 from scapy.layers.l2 import Ether, ARP, GRE
 from scapy.data import IP_PROTOS
 from scapy.layers.l2 import Ether, ARP, GRE
 from scapy.data import IP_PROTOS

-from scapy.packet import bind_layers
+from scapy.packet import bind_layers, Raw
+from scapy.all import fragment6

 from util import ppp
 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
 from time import sleep
 from util import ip4_range
 from util import ppp
 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
 from time import sleep
 from util import ip4_range

+from util import mactobinary

 
 
 class MethodHolder(VppTestCase):
 
 
 class MethodHolder(VppTestCase):


@@ -462,6 +467,121 @@ class MethodHolder(VppTestCase):
                                       "(inside network):", packet))
                 raise
 
                                       "(inside network):", packet))
                 raise
 

+    def create_stream_frag(self, src_if, dst, sport, dport, data):
+        """
+        Create fragmented packet stream
+
+        :param src_if: Source interface
+        :param dst: Destination IPv4 address
+        :param sport: Source TCP port
+        :param dport: Destination TCP port
+        :param data: Payload data
+        :returns: Fragmets
+        """
+        id = random.randint(0, 65535)
+        p = (IP(src=src_if.remote_ip4, dst=dst) /
+             TCP(sport=sport, dport=dport) /
+             Raw(data))
+        p = p.__class__(str(p))
+        chksum = p['TCP'].chksum
+        pkts = []
+        p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
+             IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=0, id=id) /
+             TCP(sport=sport, dport=dport, chksum=chksum) /
+             Raw(data[0:4]))
+        pkts.append(p)
+        p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
+             IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=3, id=id,
+                proto=IP_PROTOS.tcp) /
+             Raw(data[4:20]))
+        pkts.append(p)
+        p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
+             IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=IP_PROTOS.tcp,
+                id=id) /
+             Raw(data[20:]))
+        pkts.append(p)
+        return pkts
+
+    def create_stream_frag_ip6(self, src_if, dst, sport, dport, data,
+                               pref=None, plen=0, frag_size=128):
+        """
+        Create fragmented packet stream
+
+        :param src_if: Source interface
+        :param dst: Destination IPv4 address
+        :param sport: Source TCP port
+        :param dport: Destination TCP port
+        :param data: Payload data
+        :param pref: NAT64 prefix
+        :param plen: NAT64 prefix length
+        :param fragsize: size of fragments
+        :returns: Fragmets
+        """
+        if pref is None:
+            dst_ip6 = ''.join(['64:ff9b::', dst])
+        else:
+            dst_ip6 = self.compose_ip6(dst, pref, plen)
+
+        p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
+             IPv6(src=src_if.remote_ip6, dst=dst_ip6) /
+             IPv6ExtHdrFragment(id=random.randint(0, 65535)) /
+             TCP(sport=sport, dport=dport) /
+             Raw(data))
+
+        return fragment6(p, frag_size)
+
+    def reass_frags_and_verify(self, frags, src, dst):
+        """
+        Reassemble and verify fragmented packet
+
+        :param frags: Captured fragments
+        :param src: Source IPv4 address to verify
+        :param dst: Destination IPv4 address to verify
+
+        :returns: Reassembled IPv4 packet
+        """
+        buffer = StringIO.StringIO()
+        for p in frags:
+            self.assertEqual(p[IP].src, src)
+            self.assertEqual(p[IP].dst, dst)
+            self.check_ip_checksum(p)
+            buffer.seek(p[IP].frag * 8)
+            buffer.write(p[IP].payload)
+        ip = frags[0].getlayer(IP)
+        ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst,
+                proto=frags[0][IP].proto)
+        if ip.proto == IP_PROTOS.tcp:
+            p = (ip / TCP(buffer.getvalue()))
+            self.check_tcp_checksum(p)
+        elif ip.proto == IP_PROTOS.udp:
+            p = (ip / UDP(buffer.getvalue()))
+        return p
+
+    def reass_frags_and_verify_ip6(self, frags, src, dst):
+        """
+        Reassemble and verify fragmented packet
+
+        :param frags: Captured fragments
+        :param src: Source IPv6 address to verify
+        :param dst: Destination IPv6 address to verify
+
+        :returns: Reassembled IPv6 packet
+        """
+        buffer = StringIO.StringIO()
+        for p in frags:
+            self.assertEqual(p[IPv6].src, src)
+            self.assertEqual(p[IPv6].dst, dst)
+            buffer.seek(p[IPv6ExtHdrFragment].offset * 8)
+            buffer.write(p[IPv6ExtHdrFragment].payload)
+        ip = IPv6(src=frags[0][IPv6].src, dst=frags[0][IPv6].dst,
+                  nh=frags[0][IPv6ExtHdrFragment].nh)
+        if ip.nh == IP_PROTOS.tcp:
+            p = (ip / TCP(buffer.getvalue()))
+            self.check_tcp_checksum(p)
+        elif ip.nh == IP_PROTOS.udp:
+            p = (ip / UDP(buffer.getvalue()))
+        return p
+

     def verify_ipfix_nat44_ses(self, data):
         """
         Verify IPFIX NAT44 session create/delete event
     def verify_ipfix_nat44_ses(self, data):
         """
         Verify IPFIX NAT44 session create/delete event


@@ -584,6 +704,8 @@ class TestNAT44(MethodHolder):
             cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
             cls.pg9.resolve_arp()
 
             cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
             cls.pg9.resolve_arp()
 

+            random.seed()
+

         except Exception:
             super(TestNAT44, cls).tearDownClass()
             raise
         except Exception:
             super(TestNAT44, cls).tearDownClass()
             raise


@@ -659,7 +781,9 @@ class TestNAT44(MethodHolder):
                 lb_sm.external_port,
                 lb_sm.protocol,
                 lb_sm.vrf_id,
                 lb_sm.external_port,
                 lb_sm.protocol,
                 lb_sm.vrf_id,

-                is_add=0)
+                is_add=0,
+                local_num=0,
+                locals=[])

 
         adresses = self.vapi.nat44_address_dump()
         for addr in adresses:
 
         adresses = self.vapi.nat44_address_dump()
         for addr in adresses:


@@ -667,6 +791,9 @@ class TestNAT44(MethodHolder):
                                                   addr.ip_address,
                                                   is_add=0)
 
                                                   addr.ip_address,
                                                   is_add=0)
 

+        self.vapi.nat_set_reass()
+        self.vapi.nat_set_reass(is_ip6=1)
+

     def nat44_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
                                  local_port=0, external_port=0, vrf_id=0,
                                  is_add=1, external_sw_if_index=0xFFFFFFFF,
     def nat44_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
                                  local_port=0, external_port=0, vrf_id=0,
                                  is_add=1, external_sw_if_index=0xFFFFFFFF,


@@ -1885,11 +2012,11 @@ class TestNAT44(MethodHolder):
         """ NAT44 interfaces without configured IP address """
 
         self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
         """ NAT44 interfaces without configured IP address """
 
         self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,

-                                      self.pg7.remote_mac,
+                                      mactobinary(self.pg7.remote_mac),

                                       self.pg7.remote_ip4n,
                                       is_static=1)
         self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
                                       self.pg7.remote_ip4n,
                                       is_static=1)
         self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,

-                                      self.pg8.remote_mac,
+                                      mactobinary(self.pg8.remote_mac),

                                       self.pg8.remote_ip4n,
                                       is_static=1)
 
                                       self.pg8.remote_ip4n,
                                       is_static=1)
 


@@ -1927,11 +2054,11 @@ class TestNAT44(MethodHolder):
         """ NAT44 interfaces without configured IP address - 1:1 NAT """
 
         self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
         """ NAT44 interfaces without configured IP address - 1:1 NAT """
 
         self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,

-                                      self.pg7.remote_mac,
+                                      mactobinary(self.pg7.remote_mac),

                                       self.pg7.remote_ip4n,
                                       is_static=1)
         self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
                                       self.pg7.remote_ip4n,
                                       is_static=1)
         self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,

-                                      self.pg8.remote_mac,
+                                      mactobinary(self.pg8.remote_mac),

                                       self.pg8.remote_ip4n,
                                       is_static=1)
 
                                       self.pg8.remote_ip4n,
                                       is_static=1)
 


@@ -1973,11 +2100,11 @@ class TestNAT44(MethodHolder):
         self.icmp_id_out = 30608
 
         self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
         self.icmp_id_out = 30608
 
         self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,

-                                      self.pg7.remote_mac,
+                                      mactobinary(self.pg7.remote_mac),

                                       self.pg7.remote_ip4n,
                                       is_static=1)
         self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
                                       self.pg7.remote_ip4n,
                                       is_static=1)
         self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,

-                                      self.pg8.remote_mac,
+                                      mactobinary(self.pg8.remote_mac),

                                       self.pg8.remote_ip4n,
                                       is_static=1)
 
                                       self.pg8.remote_ip4n,
                                       is_static=1)
 


@@ -2449,10 +2576,191 @@ class TestNAT44(MethodHolder):
             self.logger.error(ppp("Unexpected or invalid packet:", p))
             raise
 
             self.logger.error(ppp("Unexpected or invalid packet:", p))
             raise
 

+    def test_del_session(self):
+        """ Delete NAT44 session """
+        self.nat44_add_address(self.nat_addr)
+        self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+        self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+                                                  is_inside=0)
+
+        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))
+
+        sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
+        nsessions = len(sessions)
+
+        self.vapi.nat44_del_session(sessions[0].inside_ip_address,
+                                    sessions[0].inside_port,
+                                    sessions[0].protocol)
+        self.vapi.nat44_del_session(sessions[1].outside_ip_address,
+                                    sessions[1].outside_port,
+                                    sessions[1].protocol,
+                                    is_in=0)
+
+        sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
+        self.assertEqual(nsessions - len(sessions), 2)
+
+    def test_set_get_reass(self):
+        """ NAT44 set/get virtual fragmentation reassembly """
+        reas_cfg1 = self.vapi.nat_get_reass()
+
+        self.vapi.nat_set_reass(timeout=reas_cfg1.ip4_timeout + 5,
+                                max_reass=reas_cfg1.ip4_max_reass * 2,
+                                max_frag=reas_cfg1.ip4_max_frag * 2)
+
+        reas_cfg2 = self.vapi.nat_get_reass()
+
+        self.assertEqual(reas_cfg1.ip4_timeout + 5, reas_cfg2.ip4_timeout)
+        self.assertEqual(reas_cfg1.ip4_max_reass * 2, reas_cfg2.ip4_max_reass)
+        self.assertEqual(reas_cfg1.ip4_max_frag * 2, reas_cfg2.ip4_max_frag)
+
+        self.vapi.nat_set_reass(drop_frag=1)
+        self.assertTrue(self.vapi.nat_get_reass().ip4_drop_frag)
+
+    def test_frag_in_order(self):
+        """ NAT44 translate fragments arriving in order """
+        self.nat44_add_address(self.nat_addr)
+        self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+        self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+                                                  is_inside=0)
+
+        data = "A" * 4 + "B" * 16 + "C" * 3
+        self.tcp_port_in = random.randint(1025, 65535)
+
+        reass = self.vapi.nat_reass_dump()
+        reass_n_start = len(reass)
+
+        # in2out
+        pkts = self.create_stream_frag(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
+        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))
+        p = self.reass_frags_and_verify(frags,
+                                        self.pg1.remote_ip4,
+                                        self.pg0.remote_ip4)
+        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):
+        """ NAT44 fragments hairpinning """
+        host = self.pg0.remote_hosts[0]
+        server = self.pg0.remote_hosts[1]
+        host_in_port = random.randint(1025, 65535)
+        host_out_port = 0
+        server_in_port = random.randint(1025, 65535)
+        server_out_port = random.randint(1025, 65535)
+        data = "A" * 4 + "B" * 16 + "C" * 3
+
+        self.nat44_add_address(self.nat_addr)
+        self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+        self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+                                                  is_inside=0)
+        # add static mapping for server
+        self.nat44_add_static_mapping(server.ip4, self.nat_addr,
+                                      server_in_port, server_out_port,
+                                      proto=IP_PROTOS.tcp)
+
+        # send packet from host to server
+        pkts = self.create_stream_frag(self.pg0,
+                                       self.nat_addr,
+                                       host_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(frags,
+                                        self.nat_addr,
+                                        server.ip4)
+        self.assertNotEqual(p[TCP].sport, host_in_port)
+        self.assertEqual(p[TCP].dport, server_in_port)
+        self.assertEqual(data, p[Raw].load)
+
+    def test_frag_out_of_order(self):
+        """ NAT44 translate fragments arriving out of order """
+        self.nat44_add_address(self.nat_addr)
+        self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+        self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+                                                  is_inside=0)
+
+        data = "A" * 4 + "B" * 16 + "C" * 3
+        random.randint(1025, 65535)
+
+        # in2out
+        pkts = self.create_stream_frag(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
+        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))
+        p = self.reass_frags_and_verify(frags,
+                                        self.pg1.remote_ip4,
+                                        self.pg0.remote_ip4)
+        self.assertEqual(p[TCP].sport, 20)
+        self.assertEqual(p[TCP].dport, self.tcp_port_in)
+        self.assertEqual(data, p[Raw].load)
+

     def tearDown(self):
         super(TestNAT44, self).tearDown()
         if not self.vpp_dead:
             self.logger.info(self.vapi.cli("show nat44 verbose"))
     def tearDown(self):
         super(TestNAT44, self).tearDown()
         if not self.vpp_dead:
             self.logger.info(self.vapi.cli("show nat44 verbose"))

+            self.logger.info(self.vapi.cli("show nat virtual-reassembly"))

             self.clear_nat44()
 
 
             self.clear_nat44()
 
 


@@ -3120,7 +3428,7 @@ class TestNAT64(MethodHolder):
             cls.vrf1_nat_addr_n = socket.inet_pton(socket.AF_INET,
                                                    cls.vrf1_nat_addr)
 
             cls.vrf1_nat_addr_n = socket.inet_pton(socket.AF_INET,
                                                    cls.vrf1_nat_addr)
 

-            cls.create_pg_interfaces(range(4))
+            cls.create_pg_interfaces(range(5))

             cls.ip6_interfaces = list(cls.pg_interfaces[0:1])
             cls.ip6_interfaces.append(cls.pg_interfaces[2])
             cls.ip4_interfaces = list(cls.pg_interfaces[1:2])
             cls.ip6_interfaces = list(cls.pg_interfaces[0:1])
             cls.ip6_interfaces.append(cls.pg_interfaces[2])
             cls.ip4_interfaces = list(cls.pg_interfaces[1:2])


@@ -3897,6 +4205,157 @@ class TestNAT64(MethodHolder):
             self.logger.error(ppp("Unexpected or invalid packet:", p))
             raise
 
             self.logger.error(ppp("Unexpected or invalid packet:", p))
             raise
 

+    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))
+

     def nat64_get_ses_num(self):
         """
         Return number of active NAT64 sessions.
     def nat64_get_ses_num(self):
         """
         Return number of active NAT64 sessions.


@@ -3975,7 +4434,155 @@ class TestNAT64(MethodHolder):
             self.logger.info(self.vapi.cli("show nat64 prefix"))
             self.logger.info(self.vapi.cli("show nat64 bib all"))
             self.logger.info(self.vapi.cli("show nat64 session table all"))
             self.logger.info(self.vapi.cli("show nat64 prefix"))
             self.logger.info(self.vapi.cli("show nat64 bib all"))
             self.logger.info(self.vapi.cli("show nat64 session table all"))

+            self.logger.info(self.vapi.cli("show nat virtual-reassembly"))

             self.clear_nat64()
 
             self.clear_nat64()
 

+
+class TestDSlite(MethodHolder):
+    """ DS-Lite Test Cases """
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestDSlite, cls).setUpClass()
+
+        try:
+            cls.nat_addr = '10.0.0.3'
+            cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
+
+            cls.create_pg_interfaces(range(2))
+            cls.pg0.admin_up()
+            cls.pg0.config_ip4()
+            cls.pg0.resolve_arp()
+            cls.pg1.admin_up()
+            cls.pg1.config_ip6()
+            cls.pg1.generate_remote_hosts(2)
+            cls.pg1.configure_ipv6_neighbors()
+
+        except Exception:
+            super(TestDSlite, cls).tearDownClass()
+            raise
+
+    def test_dslite(self):
+        """ Test DS-Lite """
+        self.vapi.dslite_add_del_pool_addr_range(self.nat_addr_n,
+                                                 self.nat_addr_n)
+        aftr_ip4 = '192.0.0.1'
+        aftr_ip4_n = socket.inet_pton(socket.AF_INET, aftr_ip4)
+        aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
+        aftr_ip6_n = socket.inet_pton(socket.AF_INET6, aftr_ip6)
+        self.vapi.dslite_set_aftr_addr(aftr_ip6_n, aftr_ip4_n)
+
+        # UDP
+        p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+             IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[0].ip6) /
+             IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
+             UDP(sport=20000, dport=10000))
+        self.pg1.add_stream(p)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg0.get_capture(1)
+        capture = capture[0]
+        self.assertFalse(capture.haslayer(IPv6))
+        self.assertEqual(capture[IP].src, self.nat_addr)
+        self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
+        self.assertNotEqual(capture[UDP].sport, 20000)
+        self.assertEqual(capture[UDP].dport, 10000)
+        self.check_ip_checksum(capture)
+        out_port = capture[UDP].sport
+
+        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+             IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
+             UDP(sport=10000, dport=out_port))
+        self.pg0.add_stream(p)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg1.get_capture(1)
+        capture = capture[0]
+        self.assertEqual(capture[IPv6].src, aftr_ip6)
+        self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
+        self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
+        self.assertEqual(capture[IP].dst, '192.168.1.1')
+        self.assertEqual(capture[UDP].sport, 10000)
+        self.assertEqual(capture[UDP].dport, 20000)
+        self.check_ip_checksum(capture)
+
+        # TCP
+        p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+             IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
+             IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
+             TCP(sport=20001, dport=10001))
+        self.pg1.add_stream(p)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg0.get_capture(1)
+        capture = capture[0]
+        self.assertFalse(capture.haslayer(IPv6))
+        self.assertEqual(capture[IP].src, self.nat_addr)
+        self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
+        self.assertNotEqual(capture[TCP].sport, 20001)
+        self.assertEqual(capture[TCP].dport, 10001)
+        self.check_ip_checksum(capture)
+        self.check_tcp_checksum(capture)
+        out_port = capture[TCP].sport
+
+        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+             IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
+             TCP(sport=10001, dport=out_port))
+        self.pg0.add_stream(p)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg1.get_capture(1)
+        capture = capture[0]
+        self.assertEqual(capture[IPv6].src, aftr_ip6)
+        self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
+        self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
+        self.assertEqual(capture[IP].dst, '192.168.1.1')
+        self.assertEqual(capture[TCP].sport, 10001)
+        self.assertEqual(capture[TCP].dport, 20001)
+        self.check_ip_checksum(capture)
+        self.check_tcp_checksum(capture)
+
+        # ICMP
+        p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+             IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
+             IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
+             ICMP(id=4000, type='echo-request'))
+        self.pg1.add_stream(p)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg0.get_capture(1)
+        capture = capture[0]
+        self.assertFalse(capture.haslayer(IPv6))
+        self.assertEqual(capture[IP].src, self.nat_addr)
+        self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
+        self.assertNotEqual(capture[ICMP].id, 4000)
+        self.check_ip_checksum(capture)
+        self.check_icmp_checksum(capture)
+        out_id = capture[ICMP].id
+
+        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+             IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
+             ICMP(id=out_id, type='echo-reply'))
+        self.pg0.add_stream(p)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+        capture = self.pg1.get_capture(1)
+        capture = capture[0]
+        self.assertEqual(capture[IPv6].src, aftr_ip6)
+        self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
+        self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
+        self.assertEqual(capture[IP].dst, '192.168.1.1')
+        self.assertEqual(capture[ICMP].id, 4000)
+        self.check_ip_checksum(capture)
+        self.check_icmp_checksum(capture)
+
+    def tearDown(self):
+        super(TestDSlite, self).tearDown()
+        if not self.vpp_dead:
+            self.logger.info(self.vapi.cli("show dslite pool"))
+            self.logger.info(
+                self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
+            self.logger.info(self.vapi.cli("show dslite sessions"))
+

 if __name__ == '__main__':
     unittest.main(testRunner=VppTestRunner)
 if __name__ == '__main__':
     unittest.main(testRunner=VppTestRunner)