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, IPv6ExtHdrFragment
   6 from scapy.all import fragment, fragment6, RandShort, defragment6
   7 from framework import VppTestCase, VppTestRunner
   8 from vpp_ip import DpoProto
   9 from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
  10 from socket import AF_INET, AF_INET6, inet_pton
  11 from util import reassemble4
  12
  13 """ Testipip is a subclass of  VPPTestCase classes.
  14
  15 IPIP tests.
  16
  17 """
  18
  19
  20 class TestIPIP(VppTestCase):
  21     """ IPIP Test Case """
  22
  23     @classmethod
  24     def setUpClass(cls):
  25         super(TestIPIP, cls).setUpClass()
  26         cls.create_pg_interfaces(range(2))
  27         cls.interfaces = list(cls.pg_interfaces)
  28
  29     @classmethod
  30     def tearDownClass(cls):
  31         super(TestIPIP, cls).tearDownClass()
  32
  33     def setUp(self):
  34         super(TestIPIP, self).setUp()
  35         for i in self.interfaces:
  36             i.admin_up()
  37             i.config_ip4()
  38             i.config_ip6()
  39             i.disable_ipv6_ra()
  40             i.resolve_arp()
  41             i.resolve_ndp()
  42
  43     def tearDown(self):
  44         super(TestIPIP, self).tearDown()
  45         if not self.vpp_dead:
  46             for i in self.pg_interfaces:
  47                 i.unconfig_ip4()
  48                 i.unconfig_ip6()
  49                 i.admin_down()
  50
  51     def validate(self, rx, expected):
  52         self.assertEqual(rx, expected.__class__(expected))
  53
  54     def generate_ip4_frags(self, payload_length, fragment_size):
  55         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
  56         p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
  57         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
  58         outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
  59                                   id=RandShort(),
  60                                   dst=self.pg0.local_ip4) / p_ip4 / p_payload)
  61         frags = fragment(outer_ip4, fragment_size)
  62         p4_reply = (p_ip4 / p_payload)
  63         p4_reply.ttl -= 1
  64         return frags, p4_reply
  65
  66     def test_ipip4(self):
  67         """ ip{v4,v6} over ip4 test """
  68         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
  69         p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
  70         p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
  71         p_payload = UDP(sport=1234, dport=1234)
  72
  73         # IPv4 transport
  74         rv = self.vapi.ipip_add_tunnel(
  75             src_address=self.pg0.local_ip4n,
  76             dst_address=self.pg1.remote_ip4n,
  77             is_ipv6=0, tc_tos=0xFF)
  78         sw_if_index = rv.sw_if_index
  79
  80         # Set interface up and enable IP on it
  81         self.vapi.sw_interface_set_flags(sw_if_index, 1)
  82         self.vapi.sw_interface_set_unnumbered(
  83             sw_if_index=self.pg0.sw_if_index,
  84             unnumbered_sw_if_index=sw_if_index)
  85
  86         # Add IPv4 and IPv6 routes via tunnel interface
  87         ip4_via_tunnel = VppIpRoute(
  88             self, "130.67.0.0", 16,
  89             [VppRoutePath("0.0.0.0",
  90                           sw_if_index,
  91                           proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
  92         ip4_via_tunnel.add_vpp_config()
  93
  94         ip6_via_tunnel = VppIpRoute(
  95             self, "dead::", 16,
  96             [VppRoutePath("::",
  97                           sw_if_index,
  98                           proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
  99         ip6_via_tunnel.add_vpp_config()
 100
 101         # IPv6 in to IPv4 tunnel
 102         p6 = (p_ether / p_ip6 / p_payload)
 103         p_inner_ip6 = p_ip6
 104         p_inner_ip6.hlim -= 1
 105         p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
 106                        proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload)
 107         p6_reply.ttl -= 1
 108         rx = self.send_and_expect(self.pg0, p6 * 10, self.pg1)
 109         for p in rx:
 110             self.validate(p[1], p6_reply)
 111
 112         # IPv4 in to IPv4 tunnel
 113         p4 = (p_ether / p_ip4 / p_payload)
 114         p_ip4_inner = p_ip4
 115         p_ip4_inner.ttl -= 1
 116         p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
 117                        tos=42) /
 118                     p_ip4_inner / p_payload)
 119         p4_reply.ttl -= 1
 120         p4_reply.id = 0
 121         rx = self.send_and_expect(self.pg0, p4 * 10, self.pg1)
 122         for p in rx:
 123             self.validate(p[1], p4_reply)
 124
 125         # Decapsulation
 126         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 127
 128         # IPv4 tunnel to IPv4
 129         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
 130         p4 = (p_ether / IP(src=self.pg1.remote_ip4,
 131                            dst=self.pg0.local_ip4) / p_ip4 / p_payload)
 132         p4_reply = (p_ip4 / p_payload)
 133         p4_reply.ttl -= 1
 134         rx = self.send_and_expect(self.pg1, p4 * 10, self.pg0)
 135         for p in rx:
 136             self.validate(p[1], p4_reply)
 137
 138         err = self.statistics.get_counter(
 139             '/err/ipip4-input/packets decapsulated')
 140         self.assertEqual(err, 10)
 141
 142         # IPv4 tunnel to IPv6
 143         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
 144         p6 = (p_ether / IP(src=self.pg1.remote_ip4,
 145                            dst=self.pg0.local_ip4) / p_ip6 / p_payload)
 146         p6_reply = (p_ip6 / p_payload)
 147         p6_reply.hlim = 63
 148         rx = self.send_and_expect(self.pg1, p6 * 10, self.pg0)
 149         for p in rx:
 150             self.validate(p[1], p6_reply)
 151
 152         err = self.statistics.get_counter(
 153             '/err/ipip4-input/packets decapsulated')
 154         self.assertEqual(err, 20)
 155
 156         #
 157         # Fragmentation / Reassembly and Re-fragmentation
 158         #
 159         rv = self.vapi.ip_reassembly_enable_disable(
 160             sw_if_index=self.pg1.sw_if_index,
 161             enable_ip4=1)
 162
 163         # Send lots of fragments, verify reassembled packet
 164         frags, p4_reply = self.generate_ip4_frags(3131, 1400)
 165         f = []
 166         for i in range(0, 1000):
 167             f.extend(frags)
 168         self.pg1.add_stream(f)
 169         self.pg_enable_capture()
 170         self.pg_start()
 171         rx = self.pg0.get_capture(1000)
 172
 173         for p in rx:
 174             self.validate(p[1], p4_reply)
 175
 176         err = self.statistics.get_counter(
 177             '/err/ipip4-input/packets decapsulated')
 178         self.assertEqual(err, 1020)
 179
 180         f = []
 181         r = []
 182         for i in range(1, 90):
 183             frags, p4_reply = self.generate_ip4_frags(i * 100, 1000)
 184             f.extend(frags)
 185             r.extend(p4_reply)
 186         self.pg_enable_capture()
 187         self.pg1.add_stream(f)
 188         self.pg_start()
 189         rx = self.pg0.get_capture(89)
 190         i = 0
 191         for p in rx:
 192             self.validate(p[1], r[i])
 193             i += 1
 194
 195         # Now try with re-fragmentation
 196         #
 197         # Send fragments to tunnel head-end, for the tunnel head end
 198         # to reassemble and then refragment
 199         #
 200         self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
 201         frags, p4_reply = self.generate_ip4_frags(3123, 1200)
 202         self.pg_enable_capture()
 203         self.pg1.add_stream(frags)
 204         self.pg_start()
 205         rx = self.pg0.get_capture(6)
 206         reass_pkt = reassemble4(rx)
 207         p4_reply.ttl -= 1
 208         p4_reply.id = 256
 209         self.validate(reass_pkt, p4_reply)
 210
 211         self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
 212         frags, p4_reply = self.generate_ip4_frags(3123, 1200)
 213         self.pg_enable_capture()
 214         self.pg1.add_stream(frags)
 215         self.pg_start()
 216         rx = self.pg0.get_capture(2)
 217         reass_pkt = reassemble4(rx)
 218         p4_reply.ttl -= 1
 219         p4_reply.id = 512
 220         self.validate(reass_pkt, p4_reply)
 221
 222     def test_ipip_create(self):
 223         """ ipip create / delete interface test """
 224         rv = self.vapi.ipip_add_tunnel(
 225             src_address=inet_pton(AF_INET, '1.2.3.4'),
 226             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
 227         sw_if_index = rv.sw_if_index
 228         self.vapi.ipip_del_tunnel(sw_if_index)
 229
 230     def test_ipip_vrf_create(self):
 231         """ ipip create / delete interface VRF test """
 232
 233         t = VppIpTable(self, 20)
 234         t.add_vpp_config()
 235         rv = self.vapi.ipip_add_tunnel(
 236             src_address=inet_pton(AF_INET, '1.2.3.4'),
 237             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
 238             table_id=20)
 239         sw_if_index = rv.sw_if_index
 240         self.vapi.ipip_del_tunnel(sw_if_index)
 241
 242     def payload(self, len):
 243         return 'x' * len
 244
 245
 246 class TestIPIP6(VppTestCase):
 247     """ IPIP6 Test Case """
 248
 249     @classmethod
 250     def setUpClass(cls):
 251         super(TestIPIP6, cls).setUpClass()
 252         cls.create_pg_interfaces(range(2))
 253         cls.interfaces = list(cls.pg_interfaces)
 254
 255     @classmethod
 256     def tearDownClass(cls):
 257         super(TestIPIP6, cls).tearDownClass()
 258
 259     def setUp(self):
 260         super(TestIPIP6, self).setUp()
 261         for i in self.interfaces:
 262             i.admin_up()
 263             i.config_ip4()
 264             i.config_ip6()
 265             i.disable_ipv6_ra()
 266             i.resolve_arp()
 267             i.resolve_ndp()
 268         self.setup_tunnel()
 269
 270     def tearDown(self):
 271         if not self.vpp_dead:
 272             self.destroy_tunnel()
 273             for i in self.pg_interfaces:
 274                 i.unconfig_ip4()
 275                 i.unconfig_ip6()
 276                 i.admin_down()
 277             super(TestIPIP6, self).tearDown()
 278
 279     def setup_tunnel(self):
 280         # IPv6 transport
 281         rv = self.vapi.ipip_add_tunnel(
 282             src_address=self.pg0.local_ip6n,
 283             dst_address=self.pg1.remote_ip6n, tc_tos=255)
 284
 285         sw_if_index = rv.sw_if_index
 286         self.tunnel_if_index = sw_if_index
 287         self.vapi.sw_interface_set_flags(sw_if_index, 1)
 288         self.vapi.sw_interface_set_unnumbered(
 289             sw_if_index=self.pg0.sw_if_index,
 290             unnumbered_sw_if_index=sw_if_index)
 291
 292         # Add IPv4 and IPv6 routes via tunnel interface
 293         ip4_via_tunnel = VppIpRoute(
 294             self, "130.67.0.0", 16,
 295             [VppRoutePath("0.0.0.0",
 296                           sw_if_index,
 297                           proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
 298         ip4_via_tunnel.add_vpp_config()
 299
 300         ip6_via_tunnel = VppIpRoute(
 301             self, "dead::", 16,
 302             [VppRoutePath("::",
 303                           sw_if_index,
 304                           proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
 305         ip6_via_tunnel.add_vpp_config()
 306
 307         self.tunnel_ip6_via_tunnel = ip6_via_tunnel
 308         self.tunnel_ip4_via_tunnel = ip4_via_tunnel
 309
 310     def destroy_tunnel(self):
 311         # IPv6 transport
 312         self.tunnel_ip4_via_tunnel.remove_vpp_config()
 313         self.tunnel_ip6_via_tunnel.remove_vpp_config()
 314
 315         rv = self.vapi.ipip_del_tunnel(sw_if_index=self.tunnel_if_index)
 316
 317     def validate(self, rx, expected):
 318         self.assertEqual(rx, expected.__class__(expected))
 319
 320     def generate_ip6_frags(self, payload_length, fragment_size):
 321         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 322         p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
 323         p_ip6 = IPv6(src="1::1", dst=self.pg0.remote_ip6)
 324         outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 325                                     dst=self.pg0.local_ip6) /
 326                      IPv6ExtHdrFragment() / p_ip6 / p_payload)
 327         frags = fragment6(outer_ip6, fragment_size)
 328         p6_reply = (p_ip6 / p_payload)
 329         p6_reply.hlim -= 1
 330         return frags, p6_reply
 331
 332     def generate_ip6_hairpin_frags(self, payload_length, fragment_size):
 333         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 334         p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
 335         p_ip6 = IPv6(src="1::1", dst="dead::1")
 336         outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 337                                     dst=self.pg0.local_ip6) /
 338                      IPv6ExtHdrFragment() / p_ip6 / p_payload)
 339         frags = fragment6(outer_ip6, fragment_size)
 340         p_ip6.hlim -= 1
 341         p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
 342                          hlim=63) / p_ip6 / p_payload)
 343
 344         return frags, p6_reply
 345
 346     def test_encap(self):
 347         """ ip{v4,v6} over ip6 test encap """
 348         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 349         p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
 350         p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
 351         p_payload = UDP(sport=1234, dport=1234)
 352
 353         # Encapsulation
 354         # IPv6 in to IPv6 tunnel
 355         p6 = (p_ether / p_ip6 / p_payload)
 356         p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
 357                          hlim=64, tc=42) /
 358                     p_ip6 / p_payload)
 359         p6_reply[1].hlim -= 1
 360         rx = self.send_and_expect(self.pg0, p6 * 11, self.pg1)
 361         for p in rx:
 362             self.validate(p[1], p6_reply)
 363
 364         # IPv4 in to IPv6 tunnel
 365         p4 = (p_ether / p_ip4 / p_payload)
 366         p4_reply = (IPv6(src=self.pg0.local_ip6,
 367                          dst=self.pg1.remote_ip6, hlim=64, tc=42) /
 368                     p_ip4 / p_payload)
 369         p4_reply[1].ttl -= 1
 370         rx = self.send_and_expect(self.pg0, p4 * 11, self.pg1)
 371         for p in rx:
 372             self.validate(p[1], p4_reply)
 373
 374     def test_decap(self):
 375         """ ip{v4,v6} over ip6 test decap """
 376
 377         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 378         p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
 379         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
 380         p_payload = UDP(sport=1234, dport=1234)
 381
 382         # Decapsulation
 383         # IPv6 tunnel to IPv4
 384
 385         p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 386                              dst=self.pg0.local_ip6) / p_ip4 / p_payload)
 387         p4_reply = (p_ip4 / p_payload)
 388         p4_reply.ttl -= 1
 389         rx = self.send_and_expect(self.pg1, p4 * 11, self.pg0)
 390         for p in rx:
 391             self.validate(p[1], p4_reply)
 392
 393         # IPv6 tunnel to IPv6
 394         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
 395         p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 396                              dst=self.pg0.local_ip6) / p_ip6 / p_payload)
 397         p6_reply = (p_ip6 / p_payload)
 398         p6_reply.hlim = 63
 399         rx = self.send_and_expect(self.pg1, p6 * 11, self.pg0)
 400         for p in rx:
 401             self.validate(p[1], p6_reply)
 402
 403     def test_frag(self):
 404         """ ip{v4,v6} over ip6 test frag """
 405
 406         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 407         p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
 408         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
 409         p_payload = UDP(sport=1234, dport=1234)
 410
 411         #
 412         # Fragmentation / Reassembly and Re-fragmentation
 413         #
 414         rv = self.vapi.ip_reassembly_enable_disable(
 415             sw_if_index=self.pg1.sw_if_index,
 416             enable_ip6=1)
 417
 418         # Send lots of fragments, verify reassembled packet
 419         before_cnt = self.statistics.get_counter(
 420             '/err/ipip6-input/packets decapsulated')
 421         frags, p6_reply = self.generate_ip6_frags(3131, 1400)
 422         f = []
 423         for i in range(0, 1000):
 424             f.extend(frags)
 425         self.pg1.add_stream(f)
 426         self.pg_enable_capture()
 427         self.pg_start()
 428         rx = self.pg0.get_capture(1000)
 429
 430         for p in rx:
 431             self.validate(p[1], p6_reply)
 432
 433         cnt = self.statistics.get_counter(
 434             '/err/ipip6-input/packets decapsulated')
 435         self.assertEqual(cnt, before_cnt + 1000)
 436
 437         f = []
 438         r = []
 439         # TODO: Check out why reassembly of atomic fragments don't work
 440         for i in range(10, 90):
 441             frags, p6_reply = self.generate_ip6_frags(i * 100, 1000)
 442             f.extend(frags)
 443             r.extend(p6_reply)
 444         self.pg_enable_capture()
 445         self.pg1.add_stream(f)
 446         self.pg_start()
 447         rx = self.pg0.get_capture(80)
 448         i = 0
 449         for p in rx:
 450             self.validate(p[1], r[i])
 451             i += 1
 452
 453         # Simple fragmentation
 454         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 455         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
 456
 457         # IPv6 in to IPv6 tunnel
 458         p_payload = UDP(sport=1234, dport=1234) / self.payload(1300)
 459
 460         p6 = (p_ether / p_ip6 / p_payload)
 461         p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
 462                          hlim=63, tc=42) /
 463                     p_ip6 / p_payload)
 464         p6_reply[1].hlim -= 1
 465         self.pg_enable_capture()
 466         self.pg0.add_stream(p6)
 467         self.pg_start()
 468         rx = self.pg1.get_capture(2)
 469
 470         # Scapy defragment doesn't deal well with multiple layers
 471         # of same type / Ethernet header first
 472         f = [p[1] for p in rx]
 473         reass_pkt = defragment6(f)
 474         self.validate(reass_pkt, p6_reply)
 475
 476         # Now try with re-fragmentation
 477         #
 478         # Send large fragments to tunnel head-end, for the tunnel head end
 479         # to reassemble and then refragment out the tunnel again.
 480         # Hair-pinning
 481         #
 482         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
 483         frags, p6_reply = self.generate_ip6_hairpin_frags(8000, 1200)
 484         self.pg_enable_capture()
 485         self.pg1.add_stream(frags)
 486         self.pg_start()
 487         rx = self.pg1.get_capture(7)
 488         f = [p[1] for p in rx]
 489         reass_pkt = defragment6(f)
 490         p6_reply.id = 256
 491         self.validate(reass_pkt, p6_reply)
 492
 493     def test_ipip_create(self):
 494         """ ipip create / delete interface test """
 495         rv = self.vapi.ipip_add_tunnel(
 496             src_address=inet_pton(AF_INET, '1.2.3.4'),
 497             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
 498         sw_if_index = rv.sw_if_index
 499         self.vapi.ipip_del_tunnel(sw_if_index)
 500
 501     def test_ipip_vrf_create(self):
 502         """ ipip create / delete interface VRF test """
 503
 504         t = VppIpTable(self, 20)
 505         t.add_vpp_config()
 506         rv = self.vapi.ipip_add_tunnel(
 507             src_address=inet_pton(AF_INET, '1.2.3.4'),
 508             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
 509             table_id=20)
 510         sw_if_index = rv.sw_if_index
 511         self.vapi.ipip_del_tunnel(sw_if_index)
 512
 513     def payload(self, len):
 514         return 'x' * len
 515
 516
 517 if __name__ == '__main__':
 518     unittest.main(testRunner=VppTestRunner)