test/test_sixrd.py

   1 #
   2 # 6RD RFC5969 functional tests
   3 #
   4
   5 import unittest
   6 from scapy.layers.inet import IP, UDP, ICMP
   7 from scapy.layers.inet6 import IPv6
   8 from scapy.layers.l2 import Ether, GRE
   9 from scapy.packet import Raw
  10
  11 from framework import VppTestCase
  12 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
  13 from util import ppp
  14 from ipaddress import *
  15
  16 """ Test6rd is a subclass of  VPPTestCase classes.
  17
  18 6RD tests.
  19
  20 """
  21
  22
  23 class Test6RD(VppTestCase):
  24     """ 6RD Test Case """
  25
  26     @classmethod
  27     def setUpClass(cls):
  28         super(Test6RD, cls).setUpClass()
  29         try:
  30             cls.create_pg_interfaces(range(2))
  31             cls.interfaces = list(cls.pg_interfaces)
  32         except Exception:
  33             super(Test6RD, cls).tearDownClass()
  34             raise
  35
  36     def setUp(cls):
  37         super(Test6RD, cls).setUp()
  38         try:
  39             for i in cls.interfaces:
  40                 i.admin_up()
  41                 i.config_ip4()
  42                 i.config_ip6()
  43                 i.disable_ipv6_ra()
  44                 i.resolve_arp()
  45                 i.resolve_ndp()
  46         except Exception:
  47             super(Test6RD, cls).tearDown()
  48             raise
  49
  50     def tearDown(self):
  51         super(Test6RD, self).tearDown()
  52         if not self.vpp_dead:
  53             self.vapi.cli("show hardware")
  54         for i in self.pg_interfaces:
  55             i.unconfig_ip4()
  56             i.unconfig_ip6()
  57             i.admin_down()
  58         if type(self.tunnel_index) is list:
  59             for sw_if_index in self.tunnel_index:
  60                 rv = self.vapi.ipip_6rd_del_tunnel(sw_if_index)
  61                 self.assertEqual(rv.retval, 0)
  62         else:
  63             rv = self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)
  64             self.assertEqual(rv.retval, 0)
  65         self.vapi.cli("show error")
  66
  67     def validate_6in4(self, rx, expected):
  68         if IP not in rx:
  69             self.fail()
  70         if IPv6 not in rx:
  71             self.fail()
  72
  73         self.assertEqual(rx[IP].src, expected[IP].src)
  74         self.assertEqual(rx[IP].dst, expected[IP].dst)
  75         self.assertEqual(rx[IP].proto, expected[IP].proto)
  76         self.assertEqual(rx[IPv6].src, expected[IPv6].src)
  77         self.assertEqual(rx[IPv6].dst, expected[IPv6].dst)
  78
  79     def validate_4in6(self, rx, expected):
  80         if IPv6 not in rx:
  81             self.fail()
  82         if IP in rx:
  83             self.fail()
  84
  85         self.assertTrue(rx[IPv6].src == expected[IPv6].src)
  86         self.assertTrue(rx[IPv6].dst == expected[IPv6].dst)
  87         self.assertTrue(rx[IPv6].nh == expected[IPv6].nh)
  88
  89     def payload(self, len):
  90         return 'x' * len
  91
  92     def test_6rd_ip6_to_ip4(self):
  93         """ ip6 -> ip4 (encap) 6rd test """
  94         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
  95         p_ip6 = IPv6(src="1::1", dst="2002:AC10:0202::1", nh='UDP')
  96
  97         rv = self.vapi.ipip_6rd_add_tunnel(
  98             0, str(ip_address('2002::').packed), 16,
  99             str(ip_address('0.0.0.0').packed), 0,
 100             str(ip_address(self.pg0.local_ip4).packed), True)
 101
 102         self.assertEqual(rv.retval, 0)
 103         self.tunnel_index = rv.sw_if_index
 104
 105         self.vapi.cli("show ip6 fib")
 106         p_payload = UDP(sport=1234, dport=1234)
 107         p = (p_ether / p_ip6 / p_payload)
 108
 109         p_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
 110                       proto='ipv6') / p_ip6)
 111
 112         rx = self.send_and_expect(self.pg0, p*10, self.pg1)
 113         for p in rx:
 114             self.validate_6in4(p, p_reply)
 115
 116         # MTU tests (default is 1480)
 117         plen = 1481 - 40 - 8
 118         p_ip6 = IPv6(src="1::1", dst="2002:AC10:0202::1")
 119         p_payload = UDP(sport=1234, dport=1234) / Raw(self.payload(plen))
 120         p = (p_ether / p_ip6 / p_payload)
 121
 122         p_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
 123                       proto='ipv6') / p_ip6)
 124
 125         rx = self.send_and_assert_no_replies(self.pg0, p*10)
 126
 127     def test_6rd_ip4_to_ip6(self):
 128         """ ip4 -> ip6 (decap) 6rd test """
 129
 130         rv = self.vapi.ipip_6rd_add_tunnel(
 131             0, str(ip_address('2002::').packed),
 132             16, str(ip_address('0.0.0.0').packed),
 133             0, str(ip_address(self.pg0.local_ip4).packed), True)
 134         self.assertEqual(rv.retval, 0)
 135         self.tunnel_index = rv.sw_if_index
 136         rv = self.vapi.ipip_6rd_del_tunnel(rv.sw_if_index)
 137         self.assertEqual(rv.retval, 0)
 138         rv = self.vapi.ipip_6rd_add_tunnel(
 139             0, str(ip_address('2002::').packed),
 140             16, str(ip_address('0.0.0.0').packed),
 141             0, str(ip_address(self.pg0.local_ip4).packed), True)
 142         self.tunnel_index = rv.sw_if_index
 143         self.assertEqual(rv.retval, 0)
 144
 145         p_ip6 = (IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) /
 146                  UDP(sport=1234, dport=1234))
 147
 148         p = (Ether(src=self.pg0.remote_mac,
 149                    dst=self.pg0.local_mac) /
 150              IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4) /
 151              p_ip6)
 152
 153         p_reply = p_ip6
 154
 155         rx = self.send_and_expect(self.pg0, p*10, self.pg1)
 156         for p in rx:
 157             self.validate_4in6(p, p_reply)
 158
 159     def test_6rd_ip4_to_ip6_multiple(self):
 160         """ ip4 -> ip6 (decap) 6rd test """
 161
 162         self.tunnel_index = []
 163         rv = self.vapi.ipip_6rd_add_tunnel(
 164             0, str(ip_address('2002::').packed),
 165             16, str(ip_address('0.0.0.0').packed),
 166             0, str(ip_address(self.pg0.local_ip4).packed), True)
 167         self.assertEqual(rv.retval, 0)
 168         self.tunnel_index.append(rv.sw_if_index)
 169
 170         rv = self.vapi.ipip_6rd_add_tunnel(
 171             0, str(ip_address('2003::').packed),
 172             16, str(ip_address('0.0.0.0').packed),
 173             0, str(ip_address(self.pg1.local_ip4).packed), True)
 174         self.assertEqual(rv.retval, 0)
 175         self.tunnel_index.append(rv.sw_if_index)
 176
 177         self.vapi.cli("show ip6 fib")
 178         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 179         p_ip4 = IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
 180         p_ip6_1 = (IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) /
 181                    UDP(sport=1234, dport=1234))
 182         p_ip6_2 = (IPv6(src="2003:AC10:0202::1", dst=self.pg1.remote_ip6) /
 183                    UDP(sport=1234, dport=1234))
 184
 185         p = (p_ether / p_ip4 / p_ip6_1)
 186         rx = self.send_and_expect(self.pg0, p*10, self.pg1)
 187         for p in rx:
 188             self.validate_4in6(p, p_ip6_1)
 189
 190         p = (p_ether / p_ip4 / p_ip6_2)
 191         rx = self.send_and_expect(self.pg0, p*10, self.pg1)
 192         for p in rx:
 193             self.validate_4in6(p, p_ip6_2)
 194
 195     def test_6rd_ip4_to_ip6_suffix(self):
 196         """ ip4 -> ip6 (decap) 6rd test """
 197
 198         rv = self.vapi.ipip_6rd_add_tunnel(
 199             0, str(ip_address('2002::').packed), 16,
 200             str(ip_address('172.0.0.0').packed), 8,
 201             str(ip_address(self.pg0.local_ip4).packed), True)
 202         self.assertEqual(rv.retval, 0)
 203
 204         self.tunnel_index = rv.sw_if_index
 205
 206         self.vapi.cli("show ip6 fib")
 207         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 208         p_ip4 = IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
 209         p_ip6 = (IPv6(src="2002:1002:0200::1", dst=self.pg1.remote_ip6) /
 210                  UDP(sport=1234, dport=1234))
 211
 212         p = (p_ether / p_ip4 / p_ip6)
 213         rx = self.send_and_expect(self.pg0, p*10, self.pg1)
 214         for p in rx:
 215             self.validate_4in6(p, p_ip6)
 216
 217     def test_6rd_ip4_to_ip6_sec_check(self):
 218         """ ip4 -> ip6 (decap) security check 6rd test """
 219
 220         rv = self.vapi.ipip_6rd_add_tunnel(
 221             0, str(ip_address('2002::').packed),
 222             16, str(ip_address('0.0.0.0').packed),
 223             0, str(ip_address(self.pg0.local_ip4).packed), True)
 224         self.assertEqual(rv.retval, 0)
 225         self.tunnel_index = rv.sw_if_index
 226
 227         self.vapi.cli("show ip6 fib")
 228         p_ip6 = (IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) /
 229                  UDP(sport=1234, dport=1234))
 230         p_ip6_fail = (IPv6(src="2002:DEAD:0202::1", dst=self.pg1.remote_ip6) /
 231                       UDP(sport=1234, dport=1234))
 232
 233         p = (Ether(src=self.pg0.remote_mac,
 234                    dst=self.pg0.local_mac) /
 235              IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4) /
 236              p_ip6)
 237
 238         p_reply = p_ip6
 239
 240         rx = self.send_and_expect(self.pg0, p*10, self.pg1)
 241         for p in rx:
 242             self.validate_4in6(p, p_reply)
 243
 244         p = (Ether(src=self.pg0.remote_mac,
 245                    dst=self.pg0.local_mac) /
 246              IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4) /
 247              p_ip6_fail)
 248         rx = self.send_and_assert_no_replies(self.pg0, p*10)
 249
 250     def test_6rd_bgp_tunnel(self):
 251         """ 6rd BGP tunnel """
 252
 253         rv = self.vapi.ipip_6rd_add_tunnel(
 254             0, str(ip_address('2002::').packed),
 255             16, str(ip_address('0.0.0.0').packed),
 256             0, str(ip_address(self.pg0.local_ip4).packed), False)
 257         self.assertEqual(rv.retval, 0)
 258         self.tunnel_index = rv.sw_if_index
 259
 260         default_route = VppIpRoute(
 261             self, "DEAD::", 16, [VppRoutePath("2002:0808:0808::",
 262                                               self.tunnel_index,
 263                                               proto=DpoProto.DPO_PROTO_IP6)],
 264             is_ip6=1)
 265         default_route.add_vpp_config()
 266
 267         ip4_route = VppIpRoute(self, "8.0.0.0", 8,
 268                                [VppRoutePath(self.pg1.remote_ip4, 0xFFFFFFFF)])
 269         ip4_route.add_vpp_config()
 270
 271         # Via recursive route 6 -> 4
 272         p = (Ether(src=self.pg0.remote_mac,
 273                    dst=self.pg0.local_mac) /
 274              IPv6(src="1::1", dst="DEAD:BEEF::1") /
 275              UDP(sport=1234, dport=1234))
 276
 277         p_reply = (IP(src=self.pg0.local_ip4, dst="8.8.8.8",
 278                       proto='ipv6') /
 279                    IPv6(src='1::1', dst='DEAD:BEEF::1', nh='UDP'))
 280
 281         rx = self.send_and_expect(self.pg0, p*10, self.pg1)
 282         for p in rx:
 283             self.validate_6in4(p, p_reply)
 284
 285         # Via recursive route 4 -> 6 (Security check must be disabled)
 286         p_ip6 = (IPv6(src="DEAD:BEEF::1", dst=self.pg1.remote_ip6) /
 287                  UDP(sport=1234, dport=1234))
 288         p = (Ether(src=self.pg0.remote_mac,
 289                    dst=self.pg0.local_mac) /
 290              IP(src="8.8.8.8", dst=self.pg0.local_ip4) /
 291              p_ip6)
 292
 293         p_reply = p_ip6
 294
 295         rx = self.send_and_expect(self.pg0, p*10, self.pg1)
 296         for p in rx:
 297             self.validate_4in6(p, p_reply)
 298
 299
 300 if __name__ == '__main__':
 301     unittest.main(testRunner=VppTestRunner)