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         err = '/err/ipip4-input/packets decapsulated'
 148         cnt = self.statistics.dump_str(err)
 149         self.assertEqual(cnt[err], 10)
 150
 151         # IPv4 tunnel to IPv6
 152         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
 153         p6 = (p_ether / IP(src=self.pg1.remote_ip4,
 154                            dst=self.pg0.local_ip4) / p_ip6 / p_payload)
 155         p6_reply = (p_ip6 / p_payload)
 156         p6_reply.hlim = 63
 157         rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
 158         for p in rx:
 159             self.validate(p[1], p6_reply)
 160
 161         cnt = self.statistics.dump_str(err)
 162         self.assertEqual(cnt[err], 20)
 163
 164         #
 165         # Fragmentation / Reassembly and Re-fragmentation
 166         #
 167         rv = self.vapi.ip_reassembly_enable_disable(
 168             sw_if_index=self.pg1.sw_if_index,
 169             enable_ip4=1)
 170
 171         # Send lots of fragments, verify reassembled packet
 172         frags, p4_reply = self.generate_frags(3131, 1400)
 173         f = []
 174         for i in range(0, 1000):
 175             f.extend(frags)
 176         self.pg1.add_stream(f)
 177         self.pg_enable_capture()
 178         self.pg_start()
 179         rx = self.pg0.get_capture(1000)
 180
 181         for p in rx:
 182             self.validate(p[1], p4_reply)
 183
 184         cnt = self.statistics.dump_str(err)
 185         self.assertEqual(cnt[err], 1020)
 186
 187         f = []
 188         r = []
 189         for i in range(1, 90):
 190             frags, p4_reply = self.generate_frags(i * 100, 1000)
 191             f.extend(frags)
 192             r.extend(p4_reply)
 193         self.pg_enable_capture()
 194         self.pg1.add_stream(f)
 195         self.pg_start()
 196         rx = self.pg0.get_capture(89)
 197         i = 0
 198         for p in rx:
 199             self.validate(p[1], r[i])
 200             i += 1
 201
 202         # Now try with re-fragmentation
 203         #
 204         # Send fragments to tunnel head-end, for the tunnel head end
 205         # to reassemble and then refragment
 206         #
 207         self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
 208         frags, p4_reply = self.generate_frags(3123, 1200)
 209         self.pg_enable_capture()
 210         self.pg1.add_stream(frags)
 211         self.pg_start()
 212         rx = self.pg0.get_capture(6)
 213         reass_pkt = reassemble(rx)
 214         p4_reply.ttl -= 1
 215         p4_reply.id = 256
 216         self.validate(reass_pkt, p4_reply)
 217
 218         self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
 219         frags, p4_reply = self.generate_frags(3123, 1200)
 220         self.pg_enable_capture()
 221         self.pg1.add_stream(frags)
 222         self.pg_start()
 223         rx = self.pg0.get_capture(2)
 224         reass_pkt = reassemble(rx)
 225         p4_reply.ttl -= 1
 226         p4_reply.id = 512
 227         self.validate(reass_pkt, p4_reply)
 228
 229     def test_ipip6(self):
 230         """ ip{v4,v6} over ip6 test """
 231         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 232         p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
 233         p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
 234         p_payload = UDP(sport=1234, dport=1234)
 235
 236         # IPv6 transport
 237         rv = self.vapi.ipip_add_tunnel(
 238             src_address=self.pg0.local_ip6n,
 239             dst_address=self.pg1.remote_ip6n, tc_tos=255)
 240
 241         sw_if_index = rv.sw_if_index
 242
 243         self.vapi.sw_interface_set_flags(sw_if_index, 1)
 244         self.vapi.sw_interface_set_unnumbered(
 245             ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
 246
 247         # Add IPv4 and IPv6 routes via tunnel interface
 248         ip4_via_tunnel = VppIpRoute(
 249             self, "130.67.0.0", 16,
 250             [VppRoutePath("0.0.0.0",
 251                           sw_if_index,
 252                           proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
 253         ip4_via_tunnel.add_vpp_config()
 254
 255         ip6_via_tunnel = VppIpRoute(
 256             self, "dead::", 16,
 257             [VppRoutePath("::",
 258                           sw_if_index,
 259                           proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
 260         ip6_via_tunnel.add_vpp_config()
 261
 262         # Encapsulation
 263
 264         # IPv6 in to IPv6 tunnel
 265         p6 = (p_ether / p_ip6 / p_payload)
 266         p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
 267                          hlim=63, tc=42) /
 268                     p_ip6 / p_payload)
 269         p6_reply[1].hlim -= 1
 270         rx = self.send_and_expect(self.pg0, p6*11, self.pg1)
 271         for p in rx:
 272             self.validate(p[1], p6_reply)
 273
 274         # IPv4 in to IPv6 tunnel
 275         p4 = (p_ether / p_ip4 / p_payload)
 276         p4_reply = (IPv6(src=self.pg0.local_ip6,
 277                          dst=self.pg1.remote_ip6, hlim=63, tc=42) /
 278                     p_ip4 / p_payload)
 279         p4_reply[1].ttl -= 1
 280         rx = self.send_and_expect(self.pg0, p4*11, self.pg1)
 281         for p in rx:
 282             self.validate(p[1], p4_reply)
 283
 284         # Decapsulation
 285
 286         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 287
 288         # IPv6 tunnel to IPv4
 289         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
 290         p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 291                              dst=self.pg0.local_ip6) / p_ip4 / p_payload)
 292         p4_reply = (p_ip4 / p_payload)
 293         p4_reply.ttl -= 1
 294         rx = self.send_and_expect(self.pg1, p4*11, self.pg0)
 295         for p in rx:
 296             self.validate(p[1], p4_reply)
 297
 298         # IPv6 tunnel to IPv6
 299         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
 300         p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 301                              dst=self.pg0.local_ip6) / p_ip6 / p_payload)
 302         p6_reply = (p_ip6 / p_payload)
 303         p6_reply.hlim = 63
 304         rx = self.send_and_expect(self.pg1, p6*11, self.pg0)
 305         for p in rx:
 306             self.validate(p[1], p6_reply)
 307
 308     def test_ipip_create(self):
 309         """ ipip create / delete interface test """
 310         rv = self.vapi.ipip_add_tunnel(
 311             src_address=inet_pton(AF_INET, '1.2.3.4'),
 312             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
 313         sw_if_index = rv.sw_if_index
 314         self.vapi.ipip_del_tunnel(sw_if_index)
 315
 316     def test_ipip_vrf_create(self):
 317         """ ipip create / delete interface VRF test """
 318
 319         t = VppIpTable(self, 20)
 320         t.add_vpp_config()
 321         rv = self.vapi.ipip_add_tunnel(
 322             src_address=inet_pton(AF_INET, '1.2.3.4'),
 323             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
 324             table_id=20)
 325         sw_if_index = rv.sw_if_index
 326         self.vapi.ipip_del_tunnel(sw_if_index)
 327
 328     def payload(self, len):
 329         return 'x' * len
 330
 331
 332 if __name__ == '__main__':
 333     unittest.main(testRunner=VppTestRunner)