test/test_ipip.py

   1 #!/usr/bin/env python
   2 """IP{4,6} over IP{v,6} tunnel functional tests"""
   3
   4 import unittest
   5 from scapy.layers.inet6 import IPv6, Ether, IP, UDP
   6 from scapy.all import fragment, RandShort
   7 from framework import VppTestCase, VppTestRunner
   8 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto, VppIpTable
   9 from socket import AF_INET, AF_INET6, inet_pton
  10 import StringIO
  11
  12 """ Testipip is a subclass of  VPPTestCase classes.
  13
  14 IPIP tests.
  15
  16 """
  17
  18
  19 def reassemble(listoffragments):
  20     buffer = StringIO.StringIO()
  21     first = listoffragments[0]
  22     buffer.seek(20)
  23     for pkt in listoffragments:
  24         buffer.seek(pkt[IP].frag*8)
  25         buffer.write(pkt[IP].payload)
  26     first.len = len(buffer.getvalue()) + 20
  27     first.flags = 0
  28     del(first.chksum)
  29     header = str(first[IP])[:20]
  30     return first[IP].__class__(header + buffer.getvalue())
  31
  32
  33 class TestIPIP(VppTestCase):
  34     """ IPIP Test Case """
  35
  36     @classmethod
  37     def setUpClass(cls):
  38         super(TestIPIP, cls).setUpClass()
  39         cls.create_pg_interfaces(range(2))
  40         cls.interfaces = list(cls.pg_interfaces)
  41
  42     def setUp(cls):
  43         super(TestIPIP, cls).setUp()
  44         for i in cls.interfaces:
  45             i.admin_up()
  46             i.config_ip4()
  47             i.config_ip6()
  48             i.disable_ipv6_ra()
  49             i.resolve_arp()
  50             i.resolve_ndp()
  51
  52     def tearDown(self):
  53         super(TestIPIP, self).tearDown()
  54         if not self.vpp_dead:
  55             for i in self.pg_interfaces:
  56                 i.unconfig_ip4()
  57                 i.unconfig_ip6()
  58                 i.admin_down()
  59
  60     def validate(self, rx, expected):
  61         self.assertEqual(rx, expected.__class__(str(expected)))
  62
  63     def generate_frags(self, payload_length, fragment_size):
  64         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
  65         p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
  66         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
  67         outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
  68                                   id=RandShort(),
  69                                   dst=self.pg0.local_ip4) / p_ip4 / p_payload)
  70         frags = fragment(outer_ip4, fragment_size)
  71         p4_reply = (p_ip4 / p_payload)
  72         p4_reply.ttl -= 1
  73         return frags, p4_reply
  74
  75     def test_ipip4(self):
  76         """ ip{v4,v6} over ip4 test """
  77         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
  78         p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
  79         p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
  80         p_payload = UDP(sport=1234, dport=1234)
  81
  82         # IPv4 transport
  83         rv = self.vapi.ipip_add_tunnel(
  84             src_address=self.pg0.local_ip4n,
  85             dst_address=self.pg1.remote_ip4n,
  86             is_ipv6=0, tc_tos=0xFF)
  87         sw_if_index = rv.sw_if_index
  88
  89         # Set interface up and enable IP on it
  90         self.vapi.sw_interface_set_flags(sw_if_index, 1)
  91         self.vapi.sw_interface_set_unnumbered(
  92             ip_sw_if_index=self.pg0.sw_if_index,
  93             sw_if_index=sw_if_index)
  94
  95         # Add IPv4 and IPv6 routes via tunnel interface
  96         ip4_via_tunnel = VppIpRoute(
  97             self, "130.67.0.0", 16,
  98             [VppRoutePath("0.0.0.0",
  99                           sw_if_index,
 100                           proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
 101         ip4_via_tunnel.add_vpp_config()
 102
 103         ip6_via_tunnel = VppIpRoute(
 104             self, "dead::", 16,
 105             [VppRoutePath("::",
 106                           sw_if_index,
 107                           proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
 108         ip6_via_tunnel.add_vpp_config()
 109
 110         # IPv6 in to IPv4 tunnel
 111         p6 = (p_ether / p_ip6 / p_payload)
 112         p_inner_ip6 = p_ip6
 113         p_inner_ip6.hlim -= 1
 114         p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
 115                        proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload)
 116         p6_reply.ttl -= 1
 117         rx = self.send_and_expect(self.pg0, p6*10, self.pg1)
 118         for p in rx:
 119             self.validate(p[1], p6_reply)
 120
 121         # IPv4 in to IPv4 tunnel
 122         p4 = (p_ether / p_ip4 / p_payload)
 123         p_ip4_inner = p_ip4
 124         p_ip4_inner.ttl -= 1
 125         p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
 126                        tos=42) /
 127                     p_ip4_inner / p_payload)
 128         p4_reply.ttl -= 1
 129         p4_reply.id = 0
 130         rx = self.send_and_expect(self.pg0, p4*10, self.pg1)
 131         for p in rx:
 132             self.validate(p[1], p4_reply)
 133
 134         # Decapsulation
 135         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 136
 137         # IPv4 tunnel to IPv4
 138         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
 139         p4 = (p_ether / IP(src=self.pg1.remote_ip4,
 140                            dst=self.pg0.local_ip4) / p_ip4 / p_payload)
 141         p4_reply = (p_ip4 / p_payload)
 142         p4_reply.ttl -= 1
 143         rx = self.send_and_expect(self.pg1, p4*10, self.pg0)
 144         for p in rx:
 145             self.validate(p[1], p4_reply)
 146
 147         # IPv4 tunnel to IPv6
 148         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
 149         p6 = (p_ether / IP(src=self.pg1.remote_ip4,
 150                            dst=self.pg0.local_ip4) / p_ip6 / p_payload)
 151         p6_reply = (p_ip6 / p_payload)
 152         p6_reply.hlim = 63
 153         rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
 154         for p in rx:
 155             self.validate(p[1], p6_reply)
 156
 157         #
 158         # Fragmentation / Reassembly and Re-fragmentation
 159         #
 160         rv = self.vapi.ip_reassembly_enable_disable(
 161             sw_if_index=self.pg1.sw_if_index,
 162             enable_ip4=1)
 163
 164         # Send lots of fragments, verify reassembled packet
 165         frags, p4_reply = self.generate_frags(3131, 1400)
 166         f = []
 167         for i in range(0, 1000):
 168             f.extend(frags)
 169         self.pg1.add_stream(f)
 170         self.pg_enable_capture()
 171         self.pg_start()
 172         rx = self.pg0.get_capture(1000)
 173
 174         for p in rx:
 175             self.validate(p[1], p4_reply)
 176
 177         f = []
 178         r = []
 179         for i in range(1, 90):
 180             frags, p4_reply = self.generate_frags(i * 100, 1000)
 181             f.extend(frags)
 182             r.extend(p4_reply)
 183         self.pg_enable_capture()
 184         self.pg1.add_stream(f)
 185         self.pg_start()
 186         rx = self.pg0.get_capture(89)
 187         i = 0
 188         for p in rx:
 189             self.validate(p[1], r[i])
 190             i += 1
 191
 192         # Now try with re-fragmentation
 193         #
 194         # Send fragments to tunnel head-end, for the tunnel head end
 195         # to reassemble and then refragment
 196         #
 197         self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
 198         frags, p4_reply = self.generate_frags(3123, 1200)
 199         self.pg_enable_capture()
 200         self.pg1.add_stream(frags)
 201         self.pg_start()
 202         rx = self.pg0.get_capture(6)
 203         reass_pkt = reassemble(rx)
 204         p4_reply.ttl -= 1
 205         p4_reply.id = 256
 206         self.validate(reass_pkt, p4_reply)
 207
 208         self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
 209         frags, p4_reply = self.generate_frags(3123, 1200)
 210         self.pg_enable_capture()
 211         self.pg1.add_stream(frags)
 212         self.pg_start()
 213         rx = self.pg0.get_capture(2)
 214         reass_pkt = reassemble(rx)
 215         p4_reply.ttl -= 1
 216         p4_reply.id = 512
 217         self.validate(reass_pkt, p4_reply)
 218
 219     def test_ipip6(self):
 220         """ ip{v4,v6} over ip6 test """
 221         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 222         p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
 223         p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
 224         p_payload = UDP(sport=1234, dport=1234)
 225
 226         # IPv6 transport
 227         rv = self.vapi.ipip_add_tunnel(
 228             src_address=self.pg0.local_ip6n,
 229             dst_address=self.pg1.remote_ip6n, tc_tos=255)
 230
 231         sw_if_index = rv.sw_if_index
 232
 233         self.vapi.sw_interface_set_flags(sw_if_index, 1)
 234         self.vapi.sw_interface_set_unnumbered(
 235             ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
 236
 237         # Add IPv4 and IPv6 routes via tunnel interface
 238         ip4_via_tunnel = VppIpRoute(
 239             self, "130.67.0.0", 16,
 240             [VppRoutePath("0.0.0.0",
 241                           sw_if_index,
 242                           proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
 243         ip4_via_tunnel.add_vpp_config()
 244
 245         ip6_via_tunnel = VppIpRoute(
 246             self, "dead::", 16,
 247             [VppRoutePath("::",
 248                           sw_if_index,
 249                           proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
 250         ip6_via_tunnel.add_vpp_config()
 251
 252         # Encapsulation
 253
 254         # IPv6 in to IPv6 tunnel
 255         p6 = (p_ether / p_ip6 / p_payload)
 256         p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
 257                          hlim=63, tc=42) /
 258                     p_ip6 / p_payload)
 259         p6_reply[1].hlim -= 1
 260         rx = self.send_and_expect(self.pg0, p6*11, self.pg1)
 261         for p in rx:
 262             self.validate(p[1], p6_reply)
 263
 264         # IPv4 in to IPv6 tunnel
 265         p4 = (p_ether / p_ip4 / p_payload)
 266         p4_reply = (IPv6(src=self.pg0.local_ip6,
 267                          dst=self.pg1.remote_ip6, hlim=63, tc=42) /
 268                     p_ip4 / p_payload)
 269         p4_reply[1].ttl -= 1
 270         rx = self.send_and_expect(self.pg0, p4*11, self.pg1)
 271         for p in rx:
 272             self.validate(p[1], p4_reply)
 273
 274         # Decapsulation
 275
 276         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 277
 278         # IPv6 tunnel to IPv4
 279         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
 280         p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 281                              dst=self.pg0.local_ip6) / p_ip4 / p_payload)
 282         p4_reply = (p_ip4 / p_payload)
 283         p4_reply.ttl -= 1
 284         rx = self.send_and_expect(self.pg1, p4*11, self.pg0)
 285         for p in rx:
 286             self.validate(p[1], p4_reply)
 287
 288         # IPv6 tunnel to IPv6
 289         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
 290         p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 291                              dst=self.pg0.local_ip6) / p_ip6 / p_payload)
 292         p6_reply = (p_ip6 / p_payload)
 293         p6_reply.hlim = 63
 294         rx = self.send_and_expect(self.pg1, p6*11, self.pg0)
 295         for p in rx:
 296             self.validate(p[1], p6_reply)
 297
 298     def test_ipip_create(self):
 299         """ ipip create / delete interface test """
 300         rv = self.vapi.ipip_add_tunnel(
 301             src_address=inet_pton(AF_INET, '1.2.3.4'),
 302             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
 303         sw_if_index = rv.sw_if_index
 304         self.vapi.ipip_del_tunnel(sw_if_index)
 305
 306     def test_ipip_vrf_create(self):
 307         """ ipip create / delete interface VRF test """
 308
 309         t = VppIpTable(self, 20)
 310         t.add_vpp_config()
 311         rv = self.vapi.ipip_add_tunnel(
 312             src_address=inet_pton(AF_INET, '1.2.3.4'),
 313             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
 314             table_id=20)
 315         sw_if_index = rv.sw_if_index
 316         self.vapi.ipip_del_tunnel(sw_if_index)
 317
 318     def payload(self, len):
 319         return 'x' * len
 320
 321
 322 if __name__ == '__main__':
 323     unittest.main(testRunner=VppTestRunner)