test/test_sixrd.py

   1 #!/usr/bin/env python3
   2 """ 6RD RFC5969 functional tests """
   3
   4 import unittest
   5 from scapy.layers.inet import IP, UDP, Ether
   6 from scapy.layers.inet6 import IPv6
   7 from scapy.packet import Raw
   8 from framework import VppTestCase, VppTestRunner
   9 from vpp_ip import DpoProto
  10 from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable, FibPathProto
  11 from socket import AF_INET, AF_INET6, inet_pton
  12
  13 """ Test6rd is a subclass of  VPPTestCase classes.
  14
  15 6RD tests.
  16
  17 """
  18
  19
  20 class Test6RD(VppTestCase):
  21     """6RD Test Case"""
  22
  23     @classmethod
  24     def setUpClass(cls):
  25         super(Test6RD, cls).setUpClass()
  26         cls.create_pg_interfaces(range(4))
  27         cls.interfaces = list(cls.pg_interfaces)
  28
  29     @classmethod
  30     def tearDownClass(cls):
  31         super(Test6RD, cls).tearDownClass()
  32
  33     def setUp(self):
  34         super(Test6RD, self).setUp()
  35         t4 = VppIpTable(self, 10)
  36         t6 = VppIpTable(self, 20, True)
  37
  38         t4.add_vpp_config()
  39         t6.add_vpp_config()
  40
  41         for n in range(4):
  42             i = self.pg_interfaces[n]
  43             i.admin_up()
  44             if n > 1:
  45                 i.set_table_ip4(10)
  46                 i.set_table_ip6(20)
  47             i.config_ip4()
  48             i.config_ip6()
  49             i.disable_ipv6_ra()
  50             i.resolve_arp()
  51             i.resolve_ndp()
  52
  53     def tearDown(self):
  54         for i in self.pg_interfaces:
  55             i.unconfig_ip4()
  56             i.unconfig_ip6()
  57             i.set_table_ip4(0)
  58             i.set_table_ip6(0)
  59             i.admin_down()
  60         super(Test6RD, self).tearDown()
  61
  62     def validate_6in4(self, rx, expected):
  63         if IP not in rx:
  64             self.fail()
  65         if IPv6 not in rx:
  66             self.fail()
  67
  68         self.assertEqual(rx[IP].src, expected[IP].src)
  69         self.assertEqual(rx[IP].dst, expected[IP].dst)
  70         self.assertEqual(rx[IP].proto, expected[IP].proto)
  71         self.assertEqual(rx[IPv6].src, expected[IPv6].src)
  72         self.assertEqual(rx[IPv6].dst, expected[IPv6].dst)
  73
  74     def validate_4in6(self, rx, expected):
  75         if IPv6 not in rx:
  76             self.fail()
  77         if IP in rx:
  78             self.fail()
  79
  80         self.assertEqual(rx[IPv6].src, expected[IPv6].src)
  81         self.assertEqual(rx[IPv6].dst, expected[IPv6].dst)
  82         self.assertEqual(rx[IPv6].nh, expected[IPv6].nh)
  83
  84     def payload(self, len):
  85         return "x" * len
  86
  87     def test_6rd_ip6_to_ip4(self):
  88         """ip6 -> ip4 (encap) 6rd test"""
  89         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
  90         p_ip6 = IPv6(src="1::1", dst="2002:AC10:0202::1", nh="UDP")
  91
  92         rv = self.vapi.ipip_6rd_add_tunnel(
  93             ip6_table_id=0,
  94             ip4_table_id=0,
  95             ip6_prefix="2002::/16",
  96             ip4_prefix="0.0.0.0/0",
  97             ip4_src=self.pg0.local_ip4,
  98             security_check=True,
  99         )
 100
 101         self.tunnel_index = rv.sw_if_index
 102
 103         self.vapi.cli("show ip6 fib")
 104         p_payload = UDP(sport=1234, dport=1234)
 105         p = p_ether / p_ip6 / p_payload
 106
 107         p_reply = (
 108             IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4, proto="ipv6") / p_ip6
 109         )
 110
 111         rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
 112         for p in rx:
 113             self.validate_6in4(p, p_reply)
 114
 115         # MTU tests (default is 1480)
 116         plen = 1481 - 40 - 8
 117         p_ip6 = IPv6(src="1::1", dst="2002:AC10:0202::1")
 118         p_payload = UDP(sport=1234, dport=1234) / Raw(self.payload(plen))
 119         p = p_ether / p_ip6 / p_payload
 120
 121         p_reply = (
 122             IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4, proto="ipv6") / p_ip6
 123         )
 124
 125         rx = self.send_and_assert_no_replies(self.pg0, p * 10)
 126         self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)
 127
 128     def test_6rd_ip6_to_ip4_vrf(self):
 129         """ip6 -> ip4 (encap) 6rd VRF test"""
 130         p_ether = Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 131         p_ip6 = IPv6(src="1::1", dst="2002:AC10:0402::1", nh="UDP")
 132
 133         rv = self.vapi.ipip_6rd_add_tunnel(
 134             ip6_table_id=20,
 135             ip4_table_id=10,
 136             ip6_prefix="2002::/16",
 137             ip4_prefix="0.0.0.0/0",
 138             ip4_src=self.pg2.local_ip4,
 139             security_check=True,
 140         )
 141         self.tunnel_index = rv.sw_if_index
 142
 143         self.vapi.cli("show ip6 fib")
 144         p_payload = UDP(sport=1234, dport=1234)
 145         p = p_ether / p_ip6 / p_payload
 146
 147         p_reply = (
 148             IP(src=self.pg2.local_ip4, dst=self.pg3.remote_ip4, proto="ipv6") / p_ip6
 149         )
 150
 151         rx = self.send_and_expect(self.pg2, p * 10, self.pg3)
 152         for p in rx:
 153             self.validate_6in4(p, p_reply)
 154
 155         # MTU tests (default is 1480)
 156         plen = 1481 - 40 - 8
 157         p_ip6 = IPv6(src="1::1", dst="2002:AC10:0402::1")
 158         p_payload = UDP(sport=1234, dport=1234) / Raw(self.payload(plen))
 159         p = p_ether / p_ip6 / p_payload
 160
 161         p_reply = (
 162             IP(src=self.pg2.local_ip4, dst=self.pg3.remote_ip4, proto="ipv6") / p_ip6
 163         )
 164
 165         rx = self.send_and_assert_no_replies(self.pg0, p * 10)
 166         self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)
 167
 168     def test_6rd_ip4_to_ip6(self):
 169         """ip4 -> ip6 (decap) 6rd test"""
 170
 171         rv = self.vapi.ipip_6rd_add_tunnel(
 172             ip6_table_id=0,
 173             ip4_table_id=0,
 174             ip6_prefix="2002::/16",
 175             ip4_prefix="0.0.0.0/0",
 176             ip4_src=self.pg0.local_ip4,
 177             security_check=True,
 178         )
 179         self.tunnel_index = rv.sw_if_index
 180         rv = self.vapi.ipip_6rd_del_tunnel(rv.sw_if_index)
 181         rv = self.vapi.ipip_6rd_add_tunnel(
 182             ip6_table_id=0,
 183             ip4_table_id=0,
 184             ip6_prefix="2002::/16",
 185             ip4_prefix="0.0.0.0/0",
 186             ip4_src=self.pg0.local_ip4,
 187             security_check=True,
 188         )
 189         self.tunnel_index = rv.sw_if_index
 190
 191         p_ip6 = IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(
 192             sport=1234, dport=1234
 193         )
 194
 195         p = (
 196             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 197             / IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
 198             / p_ip6
 199         )
 200
 201         p_reply = p_ip6
 202
 203         rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
 204         for p in rx:
 205             self.validate_4in6(p, p_reply)
 206         self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)
 207
 208     def test_6rd_ip4_to_ip6_vrf(self):
 209         """ip4 -> ip6 (decap) 6rd VRF test"""
 210
 211         rv = self.vapi.ipip_6rd_add_tunnel(
 212             ip6_table_id=20,
 213             ip4_table_id=10,
 214             ip6_prefix="2002::/16",
 215             ip4_prefix="0.0.0.0/0",
 216             ip4_src=self.pg2.local_ip4,
 217             security_check=True,
 218         )
 219         self.tunnel_index = rv.sw_if_index
 220         rv = self.vapi.ipip_6rd_del_tunnel(rv.sw_if_index)
 221         rv = self.vapi.ipip_6rd_add_tunnel(
 222             ip6_table_id=20,
 223             ip4_table_id=10,
 224             ip6_prefix="2002::/16",
 225             ip4_prefix="0.0.0.0/0",
 226             ip4_src=self.pg2.local_ip4,
 227             security_check=True,
 228         )
 229         self.tunnel_index = rv.sw_if_index
 230         self.vapi.sw_interface_set_table(self.tunnel_index, 1, 20)
 231
 232         p_ip6 = IPv6(src="2002:AC10:0402::1", dst=self.pg3.remote_ip6) / UDP(
 233             sport=1234, dport=1234
 234         )
 235
 236         p = (
 237             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 238             / IP(src=self.pg3.remote_ip4, dst=self.pg2.local_ip4)
 239             / p_ip6
 240         )
 241
 242         p_reply = p_ip6
 243
 244         rx = self.send_and_expect(self.pg2, p * 10, self.pg3)
 245         for p in rx:
 246             self.validate_4in6(p, p_reply)
 247         self.vapi.sw_interface_set_table(self.tunnel_index, 1, 0)
 248         self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)
 249
 250     def test_6rd_ip4_to_ip6_multiple(self):
 251         """ip4 -> ip6 (decap) 6rd test"""
 252
 253         self.tunnel_index = []
 254         rv = self.vapi.ipip_6rd_add_tunnel(
 255             ip6_table_id=0,
 256             ip4_table_id=0,
 257             ip6_prefix="2002::/16",
 258             ip4_prefix="0.0.0.0/0",
 259             ip4_src=self.pg0.local_ip4,
 260             security_check=True,
 261         )
 262         self.tunnel_index.append(rv.sw_if_index)
 263         rv = self.vapi.ipip_6rd_add_tunnel(
 264             ip6_table_id=0,
 265             ip4_table_id=0,
 266             ip6_prefix="2003::/16",
 267             ip4_prefix="0.0.0.0/0",
 268             ip4_src=self.pg1.local_ip4,
 269             security_check=True,
 270         )
 271
 272         self.tunnel_index.append(rv.sw_if_index)
 273
 274         self.vapi.cli("show ip6 fib")
 275         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 276         p_ip4 = IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
 277         p_ip6_1 = IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(
 278             sport=1234, dport=1234
 279         )
 280         p_ip6_2 = IPv6(src="2003:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(
 281             sport=1234, dport=1234
 282         )
 283
 284         p = p_ether / p_ip4 / p_ip6_1
 285         rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
 286         for p in rx:
 287             self.validate_4in6(p, p_ip6_1)
 288
 289         p = p_ether / p_ip4 / p_ip6_2
 290         rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
 291         for p in rx:
 292             self.validate_4in6(p, p_ip6_2)
 293         for i in self.tunnel_index:
 294             self.vapi.ipip_6rd_del_tunnel(i)
 295
 296     def test_6rd_ip4_to_ip6_suffix(self):
 297         """ip4 -> ip6 (decap) 6rd test"""
 298
 299         rv = self.vapi.ipip_6rd_add_tunnel(
 300             ip6_table_id=0,
 301             ip4_table_id=0,
 302             ip6_prefix="2002::/16",
 303             ip4_prefix="172.0.0.0/8",
 304             ip4_src=self.pg0.local_ip4,
 305             security_check=True,
 306         )
 307         self.tunnel_index = rv.sw_if_index
 308
 309         self.vapi.cli("show ip6 fib")
 310         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 311         p_ip4 = IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
 312         p_ip6 = IPv6(src="2002:1002:0200::1", dst=self.pg1.remote_ip6) / UDP(
 313             sport=1234, dport=1234
 314         )
 315
 316         p = p_ether / p_ip4 / p_ip6
 317         rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
 318         for p in rx:
 319             self.validate_4in6(p, p_ip6)
 320         self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)
 321
 322     def test_6rd_ip4_to_ip6_sec_check(self):
 323         """ip4 -> ip6 (decap) security check 6rd test"""
 324
 325         rv = self.vapi.ipip_6rd_add_tunnel(
 326             ip6_table_id=0,
 327             ip4_table_id=0,
 328             ip6_prefix="2002::/16",
 329             ip4_prefix="0.0.0.0/0",
 330             ip4_src=self.pg0.local_ip4,
 331             security_check=True,
 332         )
 333         self.tunnel_index = rv.sw_if_index
 334
 335         self.vapi.cli("show ip6 fib")
 336         p_ip6 = IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(
 337             sport=1234, dport=1234
 338         )
 339         p_ip6_fail = IPv6(src="2002:DEAD:0202::1", dst=self.pg1.remote_ip6) / UDP(
 340             sport=1234, dport=1234
 341         )
 342
 343         p = (
 344             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 345             / IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
 346             / p_ip6
 347         )
 348
 349         p_reply = p_ip6
 350
 351         rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
 352         for p in rx:
 353             self.validate_4in6(p, p_reply)
 354
 355         p = (
 356             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 357             / IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
 358             / p_ip6_fail
 359         )
 360         rx = self.send_and_assert_no_replies(self.pg0, p * 10)
 361         self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)
 362
 363     def test_6rd_bgp_tunnel(self):
 364         """6rd BGP tunnel"""
 365
 366         rv = self.vapi.ipip_6rd_add_tunnel(
 367             ip6_table_id=0,
 368             ip4_table_id=0,
 369             ip6_prefix="2002::/16",
 370             ip4_prefix="0.0.0.0/0",
 371             ip4_src=self.pg0.local_ip4,
 372             security_check=False,
 373         )
 374
 375         self.tunnel_index = rv.sw_if_index
 376
 377         default_route = VppIpRoute(
 378             self, "DEAD::", 16, [VppRoutePath("2002:0808:0808::", self.tunnel_index)]
 379         )
 380         default_route.add_vpp_config()
 381
 382         ip4_route = VppIpRoute(
 383             self, "8.0.0.0", 8, [VppRoutePath(self.pg1.remote_ip4, 0xFFFFFFFF)]
 384         )
 385         ip4_route.add_vpp_config()
 386
 387         # Via recursive route 6 -> 4
 388         p = (
 389             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 390             / IPv6(src="1::1", dst="DEAD:BEEF::1")
 391             / UDP(sport=1234, dport=1234)
 392         )
 393
 394         p_reply = IP(src=self.pg0.local_ip4, dst="8.8.8.8", proto="ipv6") / IPv6(
 395             src="1::1", dst="DEAD:BEEF::1", nh="UDP"
 396         )
 397
 398         rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
 399         for p in rx:
 400             self.validate_6in4(p, p_reply)
 401
 402         # Via recursive route 4 -> 6 (Security check must be disabled)
 403         p_ip6 = IPv6(src="DEAD:BEEF::1", dst=self.pg1.remote_ip6) / UDP(
 404             sport=1234, dport=1234
 405         )
 406         p = (
 407             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 408             / IP(src="8.8.8.8", dst=self.pg0.local_ip4)
 409             / p_ip6
 410         )
 411
 412         p_reply = p_ip6
 413
 414         rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
 415         for p in rx:
 416             self.validate_4in6(p, p_reply)
 417         ip4_route.remove_vpp_config()
 418         default_route.remove_vpp_config()
 419         self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)
 420
 421
 422 if __name__ == "__main__":
 423     unittest.main(testRunner=VppTestRunner)