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 framework import VppTestCase, VppTestRunner
   7 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
   8 from socket import AF_INET, AF_INET6, inet_pton
   9
  10 """ Testipip is a subclass of  VPPTestCase classes.
  11
  12 IPIP tests.
  13
  14 """
  15
  16
  17 class TestIPIP(VppTestCase):
  18     """ IPIP Test Case """
  19
  20     @classmethod
  21     def setUpClass(cls):
  22         super(TestIPIP, cls).setUpClass()
  23         cls.create_pg_interfaces(range(2))
  24         cls.interfaces = list(cls.pg_interfaces)
  25
  26     def setUp(cls):
  27         super(TestIPIP, cls).setUp()
  28         for i in cls.interfaces:
  29             i.admin_up()
  30             i.config_ip4()
  31             i.config_ip6()
  32             i.disable_ipv6_ra()
  33             i.resolve_arp()
  34             i.resolve_ndp()
  35
  36     def tearDown(self):
  37         super(TestIPIP, self).tearDown()
  38         if not self.vpp_dead:
  39             for i in self.pg_interfaces:
  40                 i.unconfig_ip4()
  41                 i.unconfig_ip6()
  42                 i.admin_down()
  43
  44     def validate(self, rx, expected):
  45         self.assertEqual(rx, expected.__class__(str(expected)))
  46
  47     def test_ipip4(self):
  48         """ ip{v4,v6} over ip4 test """
  49         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
  50         p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP')
  51         p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1")
  52         p_payload = UDP(sport=1234, dport=1234)
  53
  54         # IPv4 transport
  55         rv = self.vapi.ipip_add_tunnel(
  56             src_address=self.pg0.local_ip4n,
  57             dst_address=self.pg1.remote_ip4n,
  58             is_ipv6=0)
  59         sw_if_index = rv.sw_if_index
  60
  61         # Set interface up and enable IP on it
  62         self.vapi.sw_interface_set_flags(sw_if_index, 1)
  63         self.vapi.sw_interface_set_unnumbered(
  64             ip_sw_if_index=self.pg0.sw_if_index,
  65             sw_if_index=sw_if_index)
  66
  67         # Add IPv4 and IPv6 routes via tunnel interface
  68         ip4_via_tunnel = VppIpRoute(
  69             self, "130.67.0.0", 16,
  70             [VppRoutePath("0.0.0.0",
  71                           sw_if_index,
  72                           proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
  73         ip4_via_tunnel.add_vpp_config()
  74
  75         ip6_via_tunnel = VppIpRoute(
  76             self, "dead::", 16,
  77             [VppRoutePath("::",
  78                           sw_if_index,
  79                           proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
  80         ip6_via_tunnel.add_vpp_config()
  81
  82         # IPv6 in to IPv4 tunnel
  83         p6 = (p_ether / p_ip6 / p_payload)
  84         p_inner_ip6 = p_ip6
  85         p_inner_ip6.hlim -= 1
  86         p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
  87                        proto='ipv6', id=0) / p_inner_ip6 / p_payload)
  88         p6_reply.ttl -= 1
  89         rx = self.send_and_expect(self.pg0, p6*10, self.pg1)
  90         for p in rx:
  91             self.validate(p[1], p6_reply)
  92
  93         # IPv4 in to IPv4 tunnel
  94         p4 = (p_ether / p_ip4 / p_payload)
  95         p_ip4_inner = p_ip4
  96         p_ip4_inner.ttl -= 1
  97         p4_reply = (IP(src=self.pg0.local_ip4,
  98                        dst=self.pg1.remote_ip4) / p_ip4_inner / p_payload)
  99         p4_reply.ttl -= 1
 100         p4_reply.id = 0
 101         rx = self.send_and_expect(self.pg0, p4*10, self.pg1)
 102         for p in rx:
 103             self.validate(p[1], p4_reply)
 104
 105         # Decapsulation
 106         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 107
 108         # IPv4 tunnel to IPv4
 109         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
 110         p4 = (p_ether / IP(src=self.pg1.remote_ip4,
 111                            dst=self.pg0.local_ip4) / p_ip4 / p_payload)
 112         p4_reply = (p_ip4 / p_payload)
 113         p4_reply.ttl -= 1
 114         rx = self.send_and_expect(self.pg1, p4*10, self.pg0)
 115         for p in rx:
 116             self.validate(p[1], p4_reply)
 117
 118         # IPv4 tunnel to IPv6
 119         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
 120         p6 = (p_ether / IP(src=self.pg1.remote_ip4,
 121                            dst=self.pg0.local_ip4) / p_ip6 / p_payload)
 122         p6_reply = (p_ip6 / p_payload)
 123         p6_reply.hlim = 63
 124         rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
 125         for p in rx:
 126             self.validate(p[1], p6_reply)
 127
 128     def test_ipip6(self):
 129         """ ip{v4,v6} over ip6 test """
 130         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 131         p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP')
 132         p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1")
 133         p_payload = UDP(sport=1234, dport=1234)
 134
 135         # IPv6 transport
 136         rv = self.vapi.ipip_add_tunnel(
 137             src_address=self.pg0.local_ip6n,
 138             dst_address=self.pg1.remote_ip6n)
 139
 140         sw_if_index = rv.sw_if_index
 141
 142         self.vapi.sw_interface_set_flags(sw_if_index, 1)
 143         self.vapi.sw_interface_set_unnumbered(
 144             ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
 145
 146         # Add IPv4 and IPv6 routes via tunnel interface
 147         ip4_via_tunnel = VppIpRoute(
 148             self, "130.67.0.0", 16,
 149             [VppRoutePath("0.0.0.0",
 150                           sw_if_index,
 151                           proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
 152         ip4_via_tunnel.add_vpp_config()
 153
 154         ip6_via_tunnel = VppIpRoute(
 155             self, "dead::", 16,
 156             [VppRoutePath("::",
 157                           sw_if_index,
 158                           proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
 159         ip6_via_tunnel.add_vpp_config()
 160
 161         # Encapsulation
 162
 163         # IPv6 in to IPv6 tunnel
 164         p6 = (p_ether / p_ip6 / p_payload)
 165         p6_reply = (IPv6(src=self.pg0.local_ip6,
 166                          dst=self.pg1.remote_ip6, hlim=63) / p_ip6 / p_payload)
 167         p6_reply[1].hlim -= 1
 168         rx = self.send_and_expect(self.pg0, p6*10, self.pg1)
 169         for p in rx:
 170             self.validate(p[1], p6_reply)
 171
 172         # IPv4 in to IPv6 tunnel
 173         p4 = (p_ether / p_ip4 / p_payload)
 174         p4_reply = (IPv6(src=self.pg0.local_ip6,
 175                          dst=self.pg1.remote_ip6, hlim=63) / p_ip4 / p_payload)
 176         p4_reply[1].ttl -= 1
 177         rx = self.send_and_expect(self.pg0, p4*10, self.pg1)
 178         for p in rx:
 179             self.validate(p[1], p4_reply)
 180
 181         # Decapsulation
 182
 183         p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
 184
 185         # IPv6 tunnel to IPv4
 186         p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
 187         p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 188                              dst=self.pg0.local_ip6) / p_ip4 / p_payload)
 189         p4_reply = (p_ip4 / p_payload)
 190         p4_reply.ttl -= 1
 191         rx = self.send_and_expect(self.pg1, p4*10, self.pg0)
 192         for p in rx:
 193             self.validate(p[1], p4_reply)
 194
 195         # IPv6 tunnel to IPv6
 196         p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
 197         p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
 198                              dst=self.pg0.local_ip6) / p_ip6 / p_payload)
 199         p6_reply = (p_ip6 / p_payload)
 200         p6_reply.hlim = 63
 201         rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
 202         for p in rx:
 203             self.validate(p[1], p6_reply)
 204
 205     def test_ipip_create(self):
 206         """ ipip create / delete interface test """
 207         rv = self.vapi.ipip_add_tunnel(
 208             src_address=inet_pton(AF_INET, '1.2.3.4'),
 209             dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
 210         sw_if_index = rv.sw_if_index
 211         self.vapi.ipip_del_tunnel(sw_if_index)
 212
 213
 214 if __name__ == '__main__':
 215     unittest.main(testRunner=VppTestRunner)