Code Review / vpp.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | review | tree
raw | inline | side by side
fib: fib api updates
[vpp.git] / test / test_map.py
diff --git a/test/test_map.py b/test/test_map.py
index 3d02853..f1388b3 100644 (file)
--- a/test/test_map.py
+++ b/test/test_map.py
@@ -1,12 +1,14 @@
 #!/usr/bin/env python
 
 #!/usr/bin/env python
 
+import ipaddress
 import unittest
 import unittest
-import socket
+from ipaddress import IPv6Network, IPv4Network
 
 from framework import VppTestCase, VppTestRunner
 
 from framework import VppTestCase, VppTestRunner
-from vpp_ip import *
+from vpp_ip import DpoProto
 from vpp_ip_route import VppIpRoute, VppRoutePath
 
 from vpp_ip_route import VppIpRoute, VppRoutePath
 
+import scapy.compat
 from scapy.layers.l2 import Ether, Raw
 from scapy.layers.inet import IP, UDP, ICMP, TCP, fragment
 from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded
 from scapy.layers.l2 import Ether, Raw
 from scapy.layers.inet import IP, UDP, ICMP, TCP, fragment
 from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded
@@ -15,6 +17,14 @@ from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded
 class TestMAP(VppTestCase):
     """ MAP Test Case """
 
 class TestMAP(VppTestCase):
     """ MAP Test Case """
 
+    @classmethod
+    def setUpClass(cls):
+        super(TestMAP, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestMAP, cls).tearDownClass()
+
     def setUp(self):
         super(TestMAP, self).setUp()
 
     def setUp(self):
         super(TestMAP, self).setUp()
 
@@ -56,6 +66,34 @@ class TestMAP(VppTestCase):
         self.assertEqual(rx[IPv6].src, ip6_src)
         self.assertEqual(rx[IPv6].dst, ip6_dst)
 
         self.assertEqual(rx[IPv6].src, ip6_src)
         self.assertEqual(rx[IPv6].dst, ip6_dst)
 
+    def test_api_map_domain_dump(self):
+        map_dst = '2001::/64'
+        map_src = '3000::1/128'
+        client_pfx = '192.168.0.0/16'
+        tag = 'MAP-E tag.'
+        index = self.vapi.map_add_domain(ip4_prefix=client_pfx,
+                                         ip6_prefix=map_dst,
+                                         ip6_src=map_src,
+                                         tag=tag).index
+
+        rv = self.vapi.map_domain_dump()
+
+        # restore the state early so as to not impact subsequent tests.
+        # If an assert fails, we will not get the chance to do it at the end.
+        self.vapi.map_del_domain(index=index)
+
+        self.assertGreater(len(rv), 0,
+                           "Expected output from 'map_domain_dump'")
+
+        # typedefs are returned as ipaddress objects.
+        # wrap results in str() ugh! to avoid the need to call unicode.
+        self.assertEqual(str(rv[0].ip4_prefix), client_pfx)
+        self.assertEqual(str(rv[0].ip6_prefix), map_dst)
+        self.assertEqual(str(rv[0].ip6_src), map_src)
+
+        self.assertEqual(rv[0].tag, tag,
+                         "output produced incorrect tag value.")
+
     def test_map_e(self):
         """ MAP-E """
 
     def test_map_e(self):
         """ MAP-E """
 
@@ -68,18 +106,36 @@ class TestMAP(VppTestCase):
                                map_br_pfx,
                                map_br_pfx_len,
                                [VppRoutePath(self.pg1.remote_ip6,
                                map_br_pfx,
                                map_br_pfx_len,
                                [VppRoutePath(self.pg1.remote_ip6,
-                                             self.pg1.sw_if_index,
-                                             proto=DpoProto.DPO_PROTO_IP6)],
-                               is_ip6=1)
+                                             self.pg1.sw_if_index)])
         map_route.add_vpp_config()
 
         #
         # Add a domain that maps from pg0 to pg1
         #
         map_route.add_vpp_config()
 
         #
         # Add a domain that maps from pg0 to pg1
         #
-        map_dst = VppIp6Prefix(map_br_pfx, map_br_pfx_len).encode()
-        map_src = VppIp6Prefix("3000::1", 128).encode()
-        client_pfx = VppIp4Prefix("192.168.0.0", 16).encode()
-        self.vapi.map_add_domain(map_dst, map_src, client_pfx)
+        map_dst = '2001::/64'
+        map_src = '3000::1/128'
+        client_pfx = '192.168.0.0/16'
+        tag = 'MAP-E tag.'
+        self.vapi.map_add_domain(ip4_prefix=client_pfx,
+                                 ip6_prefix=map_dst,
+                                 ip6_src=map_src,
+                                 tag=tag)
+
+        # Enable MAP on interface.
+        self.vapi.map_if_enable_disable(is_enable=1,
+                                        sw_if_index=self.pg0.sw_if_index,
+                                        is_translation=0)
+
+        # Ensure MAP doesn't steal all packets!
+        v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+              IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
+              UDP(sport=20000, dport=10000) /
+              Raw('\xa5' * 100))
+        rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
+        v4_reply = v4[1]
+        v4_reply.ttl -= 1
+        for p in rx:
+            self.validate(p[1], v4_reply)
 
         #
         # Fire in a v4 packet that will be encapped to the BR
 
         #
         # Fire in a v4 packet that will be encapped to the BR
@@ -91,6 +147,22 @@ class TestMAP(VppTestCase):
 
         self.send_and_assert_encapped(v4, "3000::1", "2001::c0a8:0:0")
 
 
         self.send_and_assert_encapped(v4, "3000::1", "2001::c0a8:0:0")
 
+        # Enable MAP on interface.
+        self.vapi.map_if_enable_disable(is_enable=1,
+                                        sw_if_index=self.pg1.sw_if_index,
+                                        is_translation=0)
+
+        # Ensure MAP doesn't steal all packets
+        v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+              IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
+              UDP(sport=20000, dport=10000) /
+              Raw('\xa5' * 100))
+        rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
+        v6_reply = v6[1]
+        v6_reply.hlim -= 1
+        for p in rx:
+            self.validate(p[1], v6_reply)
+
         #
         # Fire in a V6 encapped packet.
         #  expect a decapped packet on the inside ip4 link
         #
         # Fire in a V6 encapped packet.
         #  expect a decapped packet on the inside ip4 link
@@ -119,18 +191,15 @@ class TestMAP(VppTestCase):
         self.vapi.ppcli("map params pre-resolve ip6-nh 4001::1")
 
         self.send_and_assert_no_replies(self.pg0, v4,
         self.vapi.ppcli("map params pre-resolve ip6-nh 4001::1")
 
         self.send_and_assert_no_replies(self.pg0, v4,
-                                        "resovled via default route")
+                                        "resolved via default route")
 
         #
         # Add a route to 4001::1. Expect the encapped traffic to be
         # sent via that routes next-hop
         #
 
         #
         # Add a route to 4001::1. Expect the encapped traffic to be
         # sent via that routes next-hop
         #
-        pre_res_route = VppIpRoute(
-            self, "4001::1", 128,
-            [VppRoutePath(self.pg1.remote_hosts[2].ip6,
-                          self.pg1.sw_if_index,
-                          proto=DpoProto.DPO_PROTO_IP6)],
-            is_ip6=1)
+        pre_res_route = VppIpRoute(self, "4001::1", 128,
+                                   [VppRoutePath(self.pg1.remote_hosts[2].ip6,
+                                                 self.pg1.sw_if_index)])
         pre_res_route.add_vpp_config()
 
         self.send_and_assert_encapped(v4, "3000::1",
         pre_res_route.add_vpp_config()
 
         self.send_and_assert_encapped(v4, "3000::1",
@@ -141,8 +210,7 @@ class TestMAP(VppTestCase):
         # change the route to the pre-solved next-hop
         #
         pre_res_route.modify([VppRoutePath(self.pg1.remote_hosts[3].ip6,
         # change the route to the pre-solved next-hop
         #
         pre_res_route.modify([VppRoutePath(self.pg1.remote_hosts[3].ip6,
-                                           self.pg1.sw_if_index,
-                                           proto=DpoProto.DPO_PROTO_IP6)])
+                                           self.pg1.sw_if_index)])
         pre_res_route.add_vpp_config()
 
         self.send_and_assert_encapped(v4, "3000::1",
         pre_res_route.add_vpp_config()
 
         self.send_and_assert_encapped(v4, "3000::1",
@@ -157,7 +225,7 @@ class TestMAP(VppTestCase):
         self.vapi.ppcli("map params pre-resolve del ip6-nh 4001::1")
 
     def validate(self, rx, expected):
         self.vapi.ppcli("map params pre-resolve del ip6-nh 4001::1")
 
     def validate(self, rx, expected):
-        self.assertEqual(rx, expected.__class__(str(expected)))
+        self.assertEqual(rx, expected.__class__(scapy.compat.raw(expected)))
 
     def payload(self, len):
         return 'x' * len
 
     def payload(self, len):
         return 'x' * len
@@ -168,23 +236,55 @@ class TestMAP(VppTestCase):
         #
         # Add a domain that maps from pg0 to pg1
         #
         #
         # Add a domain that maps from pg0 to pg1
         #
-        map_dst = VppIp6Prefix("2001:db8::", 32).encode()
-        map_src = VppIp6Prefix("1234:5678:90ab:cdef::", 64).encode()
-        ip4_pfx = VppIp4Prefix("192.168.0.0", 24).encode()
-        self.vapi.map_add_domain(map_dst, map_src, ip4_pfx, 16, 6, 4, 1)
+        map_dst = '2001:db8::/32'
+        map_src = '1234:5678:90ab:cdef::/64'
+        ip4_pfx = '192.168.0.0/24'
+        tag = 'MAP-T Tag.'
+
+        self.vapi.map_add_domain(ip6_prefix=map_dst,
+                                 ip4_prefix=ip4_pfx,
+                                 ip6_src=map_src,
+                                 ea_bits_len=16,
+                                 psid_offset=6,
+                                 psid_length=4,
+                                 mtu=1500,
+                                 tag=tag)
 
         # Enable MAP-T on interfaces.
 
         # Enable MAP-T on interfaces.
-
-        # self.vapi.map_if_enable_disable(1, self.pg0.sw_if_index, 1)
-        # self.vapi.map_if_enable_disable(1, self.pg1.sw_if_index, 1)
+        self.vapi.map_if_enable_disable(is_enable=1,
+                                        sw_if_index=self.pg0.sw_if_index,
+                                        is_translation=1)
+        self.vapi.map_if_enable_disable(is_enable=1,
+                                        sw_if_index=self.pg1.sw_if_index,
+                                        is_translation=1)
+
+        # Ensure MAP doesn't steal all packets!
+        v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+              IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
+              UDP(sport=20000, dport=10000) /
+              Raw('\xa5' * 100))
+        rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
+        v4_reply = v4[1]
+        v4_reply.ttl -= 1
+        for p in rx:
+            self.validate(p[1], v4_reply)
+        # Ensure MAP doesn't steal all packets
+        v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+              IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
+              UDP(sport=20000, dport=10000) /
+              Raw('\xa5' * 100))
+        rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
+        v6_reply = v6[1]
+        v6_reply.hlim -= 1
+        for p in rx:
+            self.validate(p[1], v6_reply)
 
         map_route = VppIpRoute(self,
                                "2001:db8::",
                                32,
                                [VppRoutePath(self.pg1.remote_ip6,
                                              self.pg1.sw_if_index,
 
         map_route = VppIpRoute(self,
                                "2001:db8::",
                                32,
                                [VppRoutePath(self.pg1.remote_ip6,
                                              self.pg1.sw_if_index,
-                                             proto=DpoProto.DPO_PROTO_IP6)],
-                               is_ip6=1)
+                                             proto=DpoProto.DPO_PROTO_IP6)])
         map_route.add_vpp_config()
 
         #
         map_route.add_vpp_config()
 
         #
@@ -298,6 +398,40 @@ class TestMAP(VppTestCase):
         # p4_reply.id = 256
         # self.validate(reass_pkt, p4_reply)
 
         # p4_reply.id = 256
         # self.validate(reass_pkt, p4_reply)
 
+        # TCP MSS clamping
+        self.vapi.map_param_set_tcp(1300)
+
+        #
+        # Send a v4 TCP SYN packet that will be translated and MSS clamped
+        #
+        p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+        p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
+        payload = TCP(sport=0xabcd, dport=0xabcd, flags="S",
+                      options=[('MSS', 1460)])
+
+        p4 = (p_ether / p_ip4 / payload)
+        p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
+                              dst="2001:db8:1f0::c0a8:1:f") / payload)
+        p6_translated.hlim -= 1
+        p6_translated[TCP].options = [('MSS', 1300)]
+        rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
+        for p in rx:
+            self.validate(p[1], p6_translated)
+
+        # Send back an IPv6 packet that will be "untranslated"
+        p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
+        p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
+                     dst='1234:5678:90ab:cdef:ac:1001:200:0')
+        p6 = (p_ether6 / p_ip6 / payload)
+        p4_translated = (IP(src='192.168.0.1',
+                            dst=self.pg0.remote_ip4) / payload)
+        p4_translated.id = 0
+        p4_translated.ttl -= 1
+        p4_translated[TCP].options = [('MSS', 1300)]
+        rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
+        for p in rx:
+            self.validate(p[1], p4_translated)
+
 
 if __name__ == '__main__':
     unittest.main(testRunner=VppTestRunner)
 
 if __name__ == '__main__':
     unittest.main(testRunner=VppTestRunner)
Vector Packet Processing