test/test_ip6.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 from socket import AF_INET6
   5
   6 from framework import VppTestCase, VppTestRunner
   7 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
   8 from vpp_pg_interface import is_ipv6_misc
   9 from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, VppIpMRoute, \
  10     VppMRoutePath, MRouteItfFlags, MRouteEntryFlags, VppMplsIpBind, \
  11     VppMplsRoute, DpoProto, VppMplsTable
  12 from vpp_neighbor import find_nbr, VppNeighbor
  13
  14 from scapy.packet import Raw
  15 from scapy.layers.l2 import Ether, Dot1Q
  16 from scapy.layers.inet6 import IPv6, UDP, TCP, ICMPv6ND_NS, ICMPv6ND_RS, \
  17     ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation, \
  18     ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
  19     ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \
  20     ICMPv6TimeExceeded
  21
  22 from util import ppp
  23 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
  24     in6_mactoifaceid, in6_ismaddr
  25 from scapy.utils import inet_pton, inet_ntop
  26 from scapy.contrib.mpls import MPLS
  27
  28
  29 def mk_ll_addr(mac):
  30     euid = in6_mactoifaceid(mac)
  31     addr = "fe80::" + euid
  32     return addr
  33
  34
  35 class TestIPv6ND(VppTestCase):
  36     def validate_ra(self, intf, rx, dst_ip=None):
  37         if not dst_ip:
  38             dst_ip = intf.remote_ip6
  39
  40         # unicasted packets must come to the unicast mac
  41         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  42
  43         # and from the router's MAC
  44         self.assertEqual(rx[Ether].src, intf.local_mac)
  45
  46         # the rx'd RA should be addressed to the sender's source
  47         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
  48         self.assertEqual(in6_ptop(rx[IPv6].dst),
  49                          in6_ptop(dst_ip))
  50
  51         # and come from the router's link local
  52         self.assertTrue(in6_islladdr(rx[IPv6].src))
  53         self.assertEqual(in6_ptop(rx[IPv6].src),
  54                          in6_ptop(mk_ll_addr(intf.local_mac)))
  55
  56     def validate_na(self, intf, rx, dst_ip=None, tgt_ip=None):
  57         if not dst_ip:
  58             dst_ip = intf.remote_ip6
  59         if not tgt_ip:
  60             dst_ip = intf.local_ip6
  61
  62         # unicasted packets must come to the unicast mac
  63         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  64
  65         # and from the router's MAC
  66         self.assertEqual(rx[Ether].src, intf.local_mac)
  67
  68         # the rx'd NA should be addressed to the sender's source
  69         self.assertTrue(rx.haslayer(ICMPv6ND_NA))
  70         self.assertEqual(in6_ptop(rx[IPv6].dst),
  71                          in6_ptop(dst_ip))
  72
  73         # and come from the target address
  74         self.assertEqual(in6_ptop(rx[IPv6].src), in6_ptop(tgt_ip))
  75
  76         # Dest link-layer options should have the router's MAC
  77         dll = rx[ICMPv6NDOptDstLLAddr]
  78         self.assertEqual(dll.lladdr, intf.local_mac)
  79
  80     def validate_ns(self, intf, rx, tgt_ip):
  81         nsma = in6_getnsma(inet_pton(AF_INET6, tgt_ip))
  82         dst_ip = inet_ntop(AF_INET6, nsma)
  83
  84         # NS is broadcast
  85         self.assertEqual(rx[Ether].dst, "ff:ff:ff:ff:ff:ff")
  86
  87         # and from the router's MAC
  88         self.assertEqual(rx[Ether].src, intf.local_mac)
  89
  90         # the rx'd NS should be addressed to an mcast address
  91         # derived from the target address
  92         self.assertEqual(in6_ptop(rx[IPv6].dst), in6_ptop(dst_ip))
  93
  94         # expect the tgt IP in the NS header
  95         ns = rx[ICMPv6ND_NS]
  96         self.assertEqual(in6_ptop(ns.tgt), in6_ptop(tgt_ip))
  97
  98         # packet is from the router's local address
  99         self.assertEqual(in6_ptop(rx[IPv6].src), intf.local_ip6)
 100
 101         # Src link-layer options should have the router's MAC
 102         sll = rx[ICMPv6NDOptSrcLLAddr]
 103         self.assertEqual(sll.lladdr, intf.local_mac)
 104
 105     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 106                            filter_out_fn=is_ipv6_misc):
 107         intf.add_stream(pkts)
 108         self.pg_enable_capture(self.pg_interfaces)
 109         self.pg_start()
 110         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 111
 112         self.assertEqual(len(rx), 1)
 113         rx = rx[0]
 114         self.validate_ra(intf, rx, dst_ip)
 115
 116     def send_and_expect_na(self, intf, pkts, remark, dst_ip=None,
 117                            tgt_ip=None,
 118                            filter_out_fn=is_ipv6_misc):
 119         intf.add_stream(pkts)
 120         self.pg_enable_capture(self.pg_interfaces)
 121         self.pg_start()
 122         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 123
 124         self.assertEqual(len(rx), 1)
 125         rx = rx[0]
 126         self.validate_na(intf, rx, dst_ip, tgt_ip)
 127
 128     def send_and_expect_ns(self, tx_intf, rx_intf, pkts, tgt_ip,
 129                            filter_out_fn=is_ipv6_misc):
 130         tx_intf.add_stream(pkts)
 131         self.pg_enable_capture(self.pg_interfaces)
 132         self.pg_start()
 133         rx = rx_intf.get_capture(1, filter_out_fn=filter_out_fn)
 134
 135         self.assertEqual(len(rx), 1)
 136         rx = rx[0]
 137         self.validate_ns(rx_intf, rx, tgt_ip)
 138
 139     def send_and_assert_no_replies(self, intf, pkts, remark):
 140         intf.add_stream(pkts)
 141         self.pg_enable_capture(self.pg_interfaces)
 142         self.pg_start()
 143         for i in self.pg_interfaces:
 144             i.get_capture(0)
 145             i.assert_nothing_captured(remark=remark)
 146
 147     def verify_ip(self, rx, smac, dmac, sip, dip):
 148         ether = rx[Ether]
 149         self.assertEqual(ether.dst, dmac)
 150         self.assertEqual(ether.src, smac)
 151
 152         ip = rx[IPv6]
 153         self.assertEqual(ip.src, sip)
 154         self.assertEqual(ip.dst, dip)
 155
 156
 157 class TestIPv6(TestIPv6ND):
 158     """ IPv6 Test Case """
 159
 160     @classmethod
 161     def setUpClass(cls):
 162         super(TestIPv6, cls).setUpClass()
 163
 164     def setUp(self):
 165         """
 166         Perform test setup before test case.
 167
 168         **Config:**
 169             - create 3 pg interfaces
 170                 - untagged pg0 interface
 171                 - Dot1Q subinterface on pg1
 172                 - Dot1AD subinterface on pg2
 173             - setup interfaces:
 174                 - put it into UP state
 175                 - set IPv6 addresses
 176                 - resolve neighbor address using NDP
 177             - configure 200 fib entries
 178
 179         :ivar list interfaces: pg interfaces and subinterfaces.
 180         :ivar dict flows: IPv4 packet flows in test.
 181         :ivar list pg_if_packet_sizes: packet sizes in test.
 182
 183         *TODO:* Create AD sub interface
 184         """
 185         super(TestIPv6, self).setUp()
 186
 187         # create 3 pg interfaces
 188         self.create_pg_interfaces(range(3))
 189
 190         # create 2 subinterfaces for p1 and pg2
 191         self.sub_interfaces = [
 192             VppDot1QSubint(self, self.pg1, 100),
 193             VppDot1QSubint(self, self.pg2, 200)
 194             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
 195         ]
 196
 197         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
 198         self.flows = dict()
 199         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
 200         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
 201         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
 202
 203         # packet sizes
 204         self.pg_if_packet_sizes = [64, 512, 1518, 9018]
 205         self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
 206
 207         self.interfaces = list(self.pg_interfaces)
 208         self.interfaces.extend(self.sub_interfaces)
 209
 210         # setup all interfaces
 211         for i in self.interfaces:
 212             i.admin_up()
 213             i.config_ip6()
 214             i.resolve_ndp()
 215
 216         # config 2M FIB entries
 217         self.config_fib_entries(200)
 218
 219     def tearDown(self):
 220         """Run standard test teardown and log ``show ip6 neighbors``."""
 221         for i in self.sub_interfaces:
 222             i.unconfig_ip6()
 223             i.ip6_disable()
 224             i.admin_down()
 225             i.remove_vpp_config()
 226
 227         super(TestIPv6, self).tearDown()
 228         if not self.vpp_dead:
 229             self.logger.info(self.vapi.cli("show ip6 neighbors"))
 230             # info(self.vapi.cli("show ip6 fib"))  # many entries
 231
 232     def config_fib_entries(self, count):
 233         """For each interface add to the FIB table *count* routes to
 234         "fd02::1/128" destination with interface's local address as next-hop
 235         address.
 236
 237         :param int count: Number of FIB entries.
 238
 239         - *TODO:* check if the next-hop address shouldn't be remote address
 240           instead of local address.
 241         """
 242         n_int = len(self.interfaces)
 243         percent = 0
 244         counter = 0.0
 245         dest_addr = inet_pton(AF_INET6, "fd02::1")
 246         dest_addr_len = 128
 247         for i in self.interfaces:
 248             next_hop_address = i.local_ip6n
 249             for j in range(count / n_int):
 250                 self.vapi.ip_add_del_route(
 251                     dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
 252                 counter += 1
 253                 if counter / count * 100 > percent:
 254                     self.logger.info("Configure %d FIB entries .. %d%% done" %
 255                                      (count, percent))
 256                     percent += 1
 257
 258     def create_stream(self, src_if, packet_sizes):
 259         """Create input packet stream for defined interface.
 260
 261         :param VppInterface src_if: Interface to create packet stream for.
 262         :param list packet_sizes: Required packet sizes.
 263         """
 264         pkts = []
 265         for i in range(0, 257):
 266             dst_if = self.flows[src_if][i % 2]
 267             info = self.create_packet_info(src_if, dst_if)
 268             payload = self.info_to_payload(info)
 269             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 270                  IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
 271                  UDP(sport=1234, dport=1234) /
 272                  Raw(payload))
 273             info.data = p.copy()
 274             if isinstance(src_if, VppSubInterface):
 275                 p = src_if.add_dot1_layer(p)
 276             size = packet_sizes[(i // 2) % len(packet_sizes)]
 277             self.extend_packet(p, size)
 278             pkts.append(p)
 279         return pkts
 280
 281     def verify_capture(self, dst_if, capture):
 282         """Verify captured input packet stream for defined interface.
 283
 284         :param VppInterface dst_if: Interface to verify captured packet stream
 285                                     for.
 286         :param list capture: Captured packet stream.
 287         """
 288         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 289         last_info = dict()
 290         for i in self.interfaces:
 291             last_info[i.sw_if_index] = None
 292         is_sub_if = False
 293         dst_sw_if_index = dst_if.sw_if_index
 294         if hasattr(dst_if, 'parent'):
 295             is_sub_if = True
 296         for packet in capture:
 297             if is_sub_if:
 298                 # Check VLAN tags and Ethernet header
 299                 packet = dst_if.remove_dot1_layer(packet)
 300             self.assertTrue(Dot1Q not in packet)
 301             try:
 302                 ip = packet[IPv6]
 303                 udp = packet[UDP]
 304                 payload_info = self.payload_to_info(str(packet[Raw]))
 305                 packet_index = payload_info.index
 306                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 307                 self.logger.debug(
 308                     "Got packet on port %s: src=%u (id=%u)" %
 309                     (dst_if.name, payload_info.src, packet_index))
 310                 next_info = self.get_next_packet_info_for_interface2(
 311                     payload_info.src, dst_sw_if_index,
 312                     last_info[payload_info.src])
 313                 last_info[payload_info.src] = next_info
 314                 self.assertTrue(next_info is not None)
 315                 self.assertEqual(packet_index, next_info.index)
 316                 saved_packet = next_info.data
 317                 # Check standard fields
 318                 self.assertEqual(ip.src, saved_packet[IPv6].src)
 319                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
 320                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 321                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 322             except:
 323                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 324                 raise
 325         for i in self.interfaces:
 326             remaining_packet = self.get_next_packet_info_for_interface2(
 327                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 328             self.assertTrue(remaining_packet is None,
 329                             "Interface %s: Packet expected from interface %s "
 330                             "didn't arrive" % (dst_if.name, i.name))
 331
 332     def test_fib(self):
 333         """ IPv6 FIB test
 334
 335         Test scenario:
 336             - Create IPv6 stream for pg0 interface
 337             - Create IPv6 tagged streams for pg1's and pg2's subinterface.
 338             - Send and verify received packets on each interface.
 339         """
 340
 341         pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
 342         self.pg0.add_stream(pkts)
 343
 344         for i in self.sub_interfaces:
 345             pkts = self.create_stream(i, self.sub_if_packet_sizes)
 346             i.parent.add_stream(pkts)
 347
 348         self.pg_enable_capture(self.pg_interfaces)
 349         self.pg_start()
 350
 351         pkts = self.pg0.get_capture()
 352         self.verify_capture(self.pg0, pkts)
 353
 354         for i in self.sub_interfaces:
 355             pkts = i.parent.get_capture()
 356             self.verify_capture(i, pkts)
 357
 358     def test_ns(self):
 359         """ IPv6 Neighbour Solicitation Exceptions
 360
 361         Test scenario:
 362            - Send an NS Sourced from an address not covered by the link sub-net
 363            - Send an NS to an mcast address the router has not joined
 364            - Send NS for a target address the router does not onn.
 365         """
 366
 367         #
 368         # An NS from a non link source address
 369         #
 370         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 371         d = inet_ntop(AF_INET6, nsma)
 372
 373         p = (Ether(dst=in6_getnsmac(nsma)) /
 374              IPv6(dst=d, src="2002::2") /
 375              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 376              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 377         pkts = [p]
 378
 379         self.send_and_assert_no_replies(
 380             self.pg0, pkts,
 381             "No response to NS source by address not on sub-net")
 382
 383         #
 384         # An NS for sent to a solicited mcast group the router is
 385         # not a member of FAILS
 386         #
 387         if 0:
 388             nsma = in6_getnsma(inet_pton(AF_INET6, "fd::ffff"))
 389             d = inet_ntop(AF_INET6, nsma)
 390
 391             p = (Ether(dst=in6_getnsmac(nsma)) /
 392                  IPv6(dst=d, src=self.pg0.remote_ip6) /
 393                  ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 394                  ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 395             pkts = [p]
 396
 397             self.send_and_assert_no_replies(
 398                 self.pg0, pkts,
 399                 "No response to NS sent to unjoined mcast address")
 400
 401         #
 402         # An NS whose target address is one the router does not own
 403         #
 404         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 405         d = inet_ntop(AF_INET6, nsma)
 406
 407         p = (Ether(dst=in6_getnsmac(nsma)) /
 408              IPv6(dst=d, src=self.pg0.remote_ip6) /
 409              ICMPv6ND_NS(tgt="fd::ffff") /
 410              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 411         pkts = [p]
 412
 413         self.send_and_assert_no_replies(self.pg0, pkts,
 414                                         "No response to NS for unknown target")
 415
 416         #
 417         # A neighbor entry that has no associated FIB-entry
 418         #
 419         self.pg0.generate_remote_hosts(4)
 420         nd_entry = VppNeighbor(self,
 421                                self.pg0.sw_if_index,
 422                                self.pg0.remote_hosts[2].mac,
 423                                self.pg0.remote_hosts[2].ip6,
 424                                af=AF_INET6,
 425                                is_no_fib_entry=1)
 426         nd_entry.add_vpp_config()
 427
 428         #
 429         # check we have the neighbor, but no route
 430         #
 431         self.assertTrue(find_nbr(self,
 432                                  self.pg0.sw_if_index,
 433                                  self.pg0._remote_hosts[2].ip6,
 434                                  inet=AF_INET6))
 435         self.assertFalse(find_route(self,
 436                                     self.pg0._remote_hosts[2].ip6,
 437                                     128,
 438                                     inet=AF_INET6))
 439
 440         #
 441         # send an NS from a link local address to the interface's global
 442         # address
 443         #
 444         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 445              IPv6(dst=d, src=self.pg0._remote_hosts[2].ip6_ll) /
 446              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 447              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 448
 449         self.send_and_expect_na(self.pg0, p,
 450                                 "NS from link-local",
 451                                 dst_ip=self.pg0._remote_hosts[2].ip6_ll,
 452                                 tgt_ip=self.pg0.local_ip6)
 453
 454         #
 455         # we should have learned an ND entry for the peer's link-local
 456         # but not inserted a route to it in the FIB
 457         #
 458         self.assertTrue(find_nbr(self,
 459                                  self.pg0.sw_if_index,
 460                                  self.pg0._remote_hosts[2].ip6_ll,
 461                                  inet=AF_INET6))
 462         self.assertFalse(find_route(self,
 463                                     self.pg0._remote_hosts[2].ip6_ll,
 464                                     128,
 465                                     inet=AF_INET6))
 466
 467         #
 468         # An NS to the router's own Link-local
 469         #
 470         p = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 471              IPv6(dst=d, src=self.pg0._remote_hosts[3].ip6_ll) /
 472              ICMPv6ND_NS(tgt=self.pg0.local_ip6_ll) /
 473              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 474
 475         self.send_and_expect_na(self.pg0, p,
 476                                 "NS to/from link-local",
 477                                 dst_ip=self.pg0._remote_hosts[3].ip6_ll,
 478                                 tgt_ip=self.pg0.local_ip6_ll)
 479
 480         #
 481         # we should have learned an ND entry for the peer's link-local
 482         # but not inserted a route to it in the FIB
 483         #
 484         self.assertTrue(find_nbr(self,
 485                                  self.pg0.sw_if_index,
 486                                  self.pg0._remote_hosts[3].ip6_ll,
 487                                  inet=AF_INET6))
 488         self.assertFalse(find_route(self,
 489                                     self.pg0._remote_hosts[3].ip6_ll,
 490                                     128,
 491                                     inet=AF_INET6))
 492
 493     def test_ns_duplicates(self):
 494         """ ND Duplicates"""
 495
 496         #
 497         # Generate some hosts on the LAN
 498         #
 499         self.pg1.generate_remote_hosts(3)
 500
 501         #
 502         # Add host 1 on pg1 and pg2
 503         #
 504         ns_pg1 = VppNeighbor(self,
 505                              self.pg1.sw_if_index,
 506                              self.pg1.remote_hosts[1].mac,
 507                              self.pg1.remote_hosts[1].ip6,
 508                              af=AF_INET6)
 509         ns_pg1.add_vpp_config()
 510         ns_pg2 = VppNeighbor(self,
 511                              self.pg2.sw_if_index,
 512                              self.pg2.remote_mac,
 513                              self.pg1.remote_hosts[1].ip6,
 514                              af=AF_INET6)
 515         ns_pg2.add_vpp_config()
 516
 517         #
 518         # IP packet destined for pg1 remote host arrives on pg1 again.
 519         #
 520         p = (Ether(dst=self.pg0.local_mac,
 521                    src=self.pg0.remote_mac) /
 522              IPv6(src=self.pg0.remote_ip6,
 523                   dst=self.pg1.remote_hosts[1].ip6) /
 524              UDP(sport=1234, dport=1234) /
 525              Raw())
 526
 527         self.pg0.add_stream(p)
 528         self.pg_enable_capture(self.pg_interfaces)
 529         self.pg_start()
 530
 531         rx1 = self.pg1.get_capture(1)
 532
 533         self.verify_ip(rx1[0],
 534                        self.pg1.local_mac,
 535                        self.pg1.remote_hosts[1].mac,
 536                        self.pg0.remote_ip6,
 537                        self.pg1.remote_hosts[1].ip6)
 538
 539         #
 540         # remove the duplicate on pg1
 541         # packet stream shoud generate NSs out of pg1
 542         #
 543         ns_pg1.remove_vpp_config()
 544
 545         self.send_and_expect_ns(self.pg0, self.pg1,
 546                                 p, self.pg1.remote_hosts[1].ip6)
 547
 548         #
 549         # Add it back
 550         #
 551         ns_pg1.add_vpp_config()
 552
 553         self.pg0.add_stream(p)
 554         self.pg_enable_capture(self.pg_interfaces)
 555         self.pg_start()
 556
 557         rx1 = self.pg1.get_capture(1)
 558
 559         self.verify_ip(rx1[0],
 560                        self.pg1.local_mac,
 561                        self.pg1.remote_hosts[1].mac,
 562                        self.pg0.remote_ip6,
 563                        self.pg1.remote_hosts[1].ip6)
 564
 565     def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None):
 566         if not dst_ip:
 567             dst_ip = intf.remote_ip6
 568
 569         # unicasted packets must come to the unicast mac
 570         self.assertEqual(rx[Ether].dst, intf.remote_mac)
 571
 572         # and from the router's MAC
 573         self.assertEqual(rx[Ether].src, intf.local_mac)
 574
 575         # the rx'd RA should be addressed to the sender's source
 576         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
 577         self.assertEqual(in6_ptop(rx[IPv6].dst),
 578                          in6_ptop(dst_ip))
 579
 580         # and come from the router's link local
 581         self.assertTrue(in6_islladdr(rx[IPv6].src))
 582         self.assertEqual(in6_ptop(rx[IPv6].src),
 583                          in6_ptop(mk_ll_addr(intf.local_mac)))
 584
 585         # it should contain the links MTU
 586         ra = rx[ICMPv6ND_RA]
 587         self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
 588
 589         # it should contain the source's link layer address option
 590         sll = ra[ICMPv6NDOptSrcLLAddr]
 591         self.assertEqual(sll.lladdr, intf.local_mac)
 592
 593         if not pi_opt:
 594             # the RA should not contain prefix information
 595             self.assertFalse(ra.haslayer(ICMPv6NDOptPrefixInfo))
 596         else:
 597             raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
 598
 599             # the options are nested in the scapy packet in way that i cannot
 600             # decipher how to decode. this 1st layer of option always returns
 601             # nested classes, so a direct obj1=obj2 comparison always fails.
 602             # however, the getlayer(.., 2) does give one instnace.
 603             # so we cheat here and construct a new opt instnace for comparison
 604             rd = ICMPv6NDOptPrefixInfo(prefixlen=raos.prefixlen,
 605                                        prefix=raos.prefix,
 606                                        L=raos.L,
 607                                        A=raos.A)
 608             if type(pi_opt) is list:
 609                 for ii in range(len(pi_opt)):
 610                     self.assertEqual(pi_opt[ii], rd)
 611                     rd = rx.getlayer(ICMPv6NDOptPrefixInfo, ii+2)
 612             else:
 613                 self.assertEqual(pi_opt, raos)
 614
 615     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 616                            filter_out_fn=is_ipv6_misc,
 617                            opt=None):
 618         intf.add_stream(pkts)
 619         self.pg_enable_capture(self.pg_interfaces)
 620         self.pg_start()
 621         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 622
 623         self.assertEqual(len(rx), 1)
 624         rx = rx[0]
 625         self.validate_ra(intf, rx, dst_ip, pi_opt=opt)
 626
 627     def test_rs(self):
 628         """ IPv6 Router Solicitation Exceptions
 629
 630         Test scenario:
 631         """
 632
 633         #
 634         # Before we begin change the IPv6 RA responses to use the unicast
 635         # address - that way we will not confuse them with the periodic
 636         # RAs which go to the mcast address
 637         # Sit and wait for the first periodic RA.
 638         #
 639         # TODO
 640         #
 641         self.pg0.ip6_ra_config(send_unicast=1)
 642
 643         #
 644         # An RS from a link source address
 645         #  - expect an RA in return
 646         #
 647         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 648              IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
 649              ICMPv6ND_RS())
 650         pkts = [p]
 651         self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
 652
 653         #
 654         # For the next RS sent the RA should be rate limited
 655         #
 656         self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
 657
 658         #
 659         # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
 660         # so we need to do this before each test below so as not to drop
 661         # packets for rate limiting reasons. Test this works here.
 662         #
 663         self.pg0.ip6_ra_config(send_unicast=1)
 664         self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
 665
 666         #
 667         # An RS sent from a non-link local source
 668         #
 669         self.pg0.ip6_ra_config(send_unicast=1)
 670         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 671              IPv6(dst=self.pg0.local_ip6, src="2002::ffff") /
 672              ICMPv6ND_RS())
 673         pkts = [p]
 674         self.send_and_assert_no_replies(self.pg0, pkts,
 675                                         "RS from non-link source")
 676
 677         #
 678         # Source an RS from a link local address
 679         #
 680         self.pg0.ip6_ra_config(send_unicast=1)
 681         ll = mk_ll_addr(self.pg0.remote_mac)
 682         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 683              IPv6(dst=self.pg0.local_ip6, src=ll) /
 684              ICMPv6ND_RS())
 685         pkts = [p]
 686         self.send_and_expect_ra(self.pg0, pkts,
 687                                 "RS sourced from link-local",
 688                                 dst_ip=ll)
 689
 690         #
 691         # Send the RS multicast
 692         #
 693         self.pg0.ip6_ra_config(send_unicast=1)
 694         dmac = in6_getnsmac(inet_pton(AF_INET6, "ff02::2"))
 695         ll = mk_ll_addr(self.pg0.remote_mac)
 696         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 697              IPv6(dst="ff02::2", src=ll) /
 698              ICMPv6ND_RS())
 699         pkts = [p]
 700         self.send_and_expect_ra(self.pg0, pkts,
 701                                 "RS sourced from link-local",
 702                                 dst_ip=ll)
 703
 704         #
 705         # Source from the unspecified address ::. This happens when the RS
 706         # is sent before the host has a configured address/sub-net,
 707         # i.e. auto-config. Since the sender has no IP address, the reply
 708         # comes back mcast - so the capture needs to not filter this.
 709         # If we happen to pick up the periodic RA at this point then so be it,
 710         # it's not an error.
 711         #
 712         self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
 713         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 714              IPv6(dst="ff02::2", src="::") /
 715              ICMPv6ND_RS())
 716         pkts = [p]
 717         self.send_and_expect_ra(self.pg0, pkts,
 718                                 "RS sourced from unspecified",
 719                                 dst_ip="ff02::1",
 720                                 filter_out_fn=None)
 721
 722         #
 723         # Configure The RA to announce the links prefix
 724         #
 725         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 726                                self.pg0.local_ip6_prefix_len)
 727
 728         #
 729         # RAs should now contain the prefix information option
 730         #
 731         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 732                                     prefix=self.pg0.local_ip6,
 733                                     L=1,
 734                                     A=1)
 735
 736         self.pg0.ip6_ra_config(send_unicast=1)
 737         ll = mk_ll_addr(self.pg0.remote_mac)
 738         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 739              IPv6(dst=self.pg0.local_ip6, src=ll) /
 740              ICMPv6ND_RS())
 741         self.send_and_expect_ra(self.pg0, p,
 742                                 "RA with prefix-info",
 743                                 dst_ip=ll,
 744                                 opt=opt)
 745
 746         #
 747         # Change the prefix info to not off-link
 748         #  L-flag is clear
 749         #
 750         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 751                                self.pg0.local_ip6_prefix_len,
 752                                off_link=1)
 753
 754         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 755                                     prefix=self.pg0.local_ip6,
 756                                     L=0,
 757                                     A=1)
 758
 759         self.pg0.ip6_ra_config(send_unicast=1)
 760         self.send_and_expect_ra(self.pg0, p,
 761                                 "RA with Prefix info with L-flag=0",
 762                                 dst_ip=ll,
 763                                 opt=opt)
 764
 765         #
 766         # Change the prefix info to not off-link, no-autoconfig
 767         #  L and A flag are clear in the advert
 768         #
 769         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 770                                self.pg0.local_ip6_prefix_len,
 771                                off_link=1,
 772                                no_autoconfig=1)
 773
 774         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 775                                     prefix=self.pg0.local_ip6,
 776                                     L=0,
 777                                     A=0)
 778
 779         self.pg0.ip6_ra_config(send_unicast=1)
 780         self.send_and_expect_ra(self.pg0, p,
 781                                 "RA with Prefix info with A & L-flag=0",
 782                                 dst_ip=ll,
 783                                 opt=opt)
 784
 785         #
 786         # Change the flag settings back to the defaults
 787         #  L and A flag are set in the advert
 788         #
 789         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 790                                self.pg0.local_ip6_prefix_len)
 791
 792         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 793                                     prefix=self.pg0.local_ip6,
 794                                     L=1,
 795                                     A=1)
 796
 797         self.pg0.ip6_ra_config(send_unicast=1)
 798         self.send_and_expect_ra(self.pg0, p,
 799                                 "RA with Prefix info",
 800                                 dst_ip=ll,
 801                                 opt=opt)
 802
 803         #
 804         # Change the prefix info to not off-link, no-autoconfig
 805         #  L and A flag are clear in the advert
 806         #
 807         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 808                                self.pg0.local_ip6_prefix_len,
 809                                off_link=1,
 810                                no_autoconfig=1)
 811
 812         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 813                                     prefix=self.pg0.local_ip6,
 814                                     L=0,
 815                                     A=0)
 816
 817         self.pg0.ip6_ra_config(send_unicast=1)
 818         self.send_and_expect_ra(self.pg0, p,
 819                                 "RA with Prefix info with A & L-flag=0",
 820                                 dst_ip=ll,
 821                                 opt=opt)
 822
 823         #
 824         # Use the reset to defults option to revert to defaults
 825         #  L and A flag are clear in the advert
 826         #
 827         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 828                                self.pg0.local_ip6_prefix_len,
 829                                use_default=1)
 830
 831         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 832                                     prefix=self.pg0.local_ip6,
 833                                     L=1,
 834                                     A=1)
 835
 836         self.pg0.ip6_ra_config(send_unicast=1)
 837         self.send_and_expect_ra(self.pg0, p,
 838                                 "RA with Prefix reverted to defaults",
 839                                 dst_ip=ll,
 840                                 opt=opt)
 841
 842         #
 843         # Advertise Another prefix. With no L-flag/A-flag
 844         #
 845         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 846                                self.pg1.local_ip6_prefix_len,
 847                                off_link=1,
 848                                no_autoconfig=1)
 849
 850         opt = [ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 851                                      prefix=self.pg0.local_ip6,
 852                                      L=1,
 853                                      A=1),
 854                ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
 855                                      prefix=self.pg1.local_ip6,
 856                                      L=0,
 857                                      A=0)]
 858
 859         self.pg0.ip6_ra_config(send_unicast=1)
 860         ll = mk_ll_addr(self.pg0.remote_mac)
 861         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 862              IPv6(dst=self.pg0.local_ip6, src=ll) /
 863              ICMPv6ND_RS())
 864         self.send_and_expect_ra(self.pg0, p,
 865                                 "RA with multiple Prefix infos",
 866                                 dst_ip=ll,
 867                                 opt=opt)
 868
 869         #
 870         # Remove the first refix-info - expect the second is still in the
 871         # advert
 872         #
 873         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 874                                self.pg0.local_ip6_prefix_len,
 875                                is_no=1)
 876
 877         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
 878                                     prefix=self.pg1.local_ip6,
 879                                     L=0,
 880                                     A=0)
 881
 882         self.pg0.ip6_ra_config(send_unicast=1)
 883         self.send_and_expect_ra(self.pg0, p,
 884                                 "RA with Prefix reverted to defaults",
 885                                 dst_ip=ll,
 886                                 opt=opt)
 887
 888         #
 889         # Remove the second prefix-info - expect no prefix-info i nthe adverts
 890         #
 891         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 892                                self.pg1.local_ip6_prefix_len,
 893                                is_no=1)
 894
 895         self.pg0.ip6_ra_config(send_unicast=1)
 896         self.send_and_expect_ra(self.pg0, p,
 897                                 "RA with Prefix reverted to defaults",
 898                                 dst_ip=ll)
 899
 900         #
 901         # Reset the periodic advertisements back to default values
 902         #
 903         self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
 904
 905
 906 class IPv6NDProxyTest(TestIPv6ND):
 907     """ IPv6 ND ProxyTest Case """
 908
 909     def setUp(self):
 910         super(IPv6NDProxyTest, self).setUp()
 911
 912         # create 3 pg interfaces
 913         self.create_pg_interfaces(range(3))
 914
 915         # pg0 is the master interface, with the configured subnet
 916         self.pg0.admin_up()
 917         self.pg0.config_ip6()
 918         self.pg0.resolve_ndp()
 919
 920         self.pg1.ip6_enable()
 921         self.pg2.ip6_enable()
 922
 923     def tearDown(self):
 924         super(IPv6NDProxyTest, self).tearDown()
 925
 926     def test_nd_proxy(self):
 927         """ IPv6 Proxy ND """
 928
 929         #
 930         # Generate some hosts in the subnet that we are proxying
 931         #
 932         self.pg0.generate_remote_hosts(8)
 933
 934         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 935         d = inet_ntop(AF_INET6, nsma)
 936
 937         #
 938         # Send an NS for one of those remote hosts on one of the proxy links
 939         # expect no response since it's from an address that is not
 940         # on the link that has the prefix configured
 941         #
 942         ns_pg1 = (Ether(dst=in6_getnsmac(nsma), src=self.pg1.remote_mac) /
 943                   IPv6(dst=d, src=self.pg0._remote_hosts[2].ip6) /
 944                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 945                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
 946
 947         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Off link NS")
 948
 949         #
 950         # Add proxy support for the host
 951         #
 952         self.vapi.ip6_nd_proxy(
 953             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
 954             self.pg1.sw_if_index)
 955
 956         #
 957         # try that NS again. this time we expect an NA back
 958         #
 959         self.send_and_expect_na(self.pg1, ns_pg1,
 960                                 "NS to proxy entry",
 961                                 dst_ip=self.pg0._remote_hosts[2].ip6,
 962                                 tgt_ip=self.pg0.local_ip6)
 963
 964         #
 965         # ... and that we have an entry in the ND cache
 966         #
 967         self.assertTrue(find_nbr(self,
 968                                  self.pg1.sw_if_index,
 969                                  self.pg0._remote_hosts[2].ip6,
 970                                  inet=AF_INET6))
 971
 972         #
 973         # ... and we can route traffic to it
 974         #
 975         t = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 976              IPv6(dst=self.pg0._remote_hosts[2].ip6,
 977                   src=self.pg0.remote_ip6) /
 978              UDP(sport=10000, dport=20000) /
 979              Raw('\xa5' * 100))
 980
 981         self.pg0.add_stream(t)
 982         self.pg_enable_capture(self.pg_interfaces)
 983         self.pg_start()
 984         rx = self.pg1.get_capture(1)
 985         rx = rx[0]
 986
 987         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
 988         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
 989
 990         self.assertEqual(rx[IPv6].src, t[IPv6].src)
 991         self.assertEqual(rx[IPv6].dst, t[IPv6].dst)
 992
 993         #
 994         # Test we proxy for the host on the main interface
 995         #
 996         ns_pg0 = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 997                   IPv6(dst=d, src=self.pg0.remote_ip6) /
 998                   ICMPv6ND_NS(tgt=self.pg0._remote_hosts[2].ip6) /
 999                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
1000
1001         self.send_and_expect_na(self.pg0, ns_pg0,
1002                                 "NS to proxy entry on main",
1003                                 tgt_ip=self.pg0._remote_hosts[2].ip6,
1004                                 dst_ip=self.pg0.remote_ip6)
1005
1006         #
1007         # Setup and resolve proxy for another host on another interface
1008         #
1009         ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
1010                   IPv6(dst=d, src=self.pg0._remote_hosts[3].ip6) /
1011                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
1012                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
1013
1014         self.vapi.ip6_nd_proxy(
1015             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1016             self.pg2.sw_if_index)
1017
1018         self.send_and_expect_na(self.pg2, ns_pg2,
1019                                 "NS to proxy entry other interface",
1020                                 dst_ip=self.pg0._remote_hosts[3].ip6,
1021                                 tgt_ip=self.pg0.local_ip6)
1022
1023         self.assertTrue(find_nbr(self,
1024                                  self.pg2.sw_if_index,
1025                                  self.pg0._remote_hosts[3].ip6,
1026                                  inet=AF_INET6))
1027
1028         #
1029         # hosts can communicate. pg2->pg1
1030         #
1031         t2 = (Ether(dst=self.pg2.local_mac,
1032                     src=self.pg0.remote_hosts[3].mac) /
1033               IPv6(dst=self.pg0._remote_hosts[2].ip6,
1034                    src=self.pg0._remote_hosts[3].ip6) /
1035               UDP(sport=10000, dport=20000) /
1036               Raw('\xa5' * 100))
1037
1038         self.pg2.add_stream(t2)
1039         self.pg_enable_capture(self.pg_interfaces)
1040         self.pg_start()
1041         rx = self.pg1.get_capture(1)
1042         rx = rx[0]
1043
1044         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
1045         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
1046
1047         self.assertEqual(rx[IPv6].src, t2[IPv6].src)
1048         self.assertEqual(rx[IPv6].dst, t2[IPv6].dst)
1049
1050         #
1051         # remove the proxy configs
1052         #
1053         self.vapi.ip6_nd_proxy(
1054             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
1055             self.pg1.sw_if_index,
1056             is_del=1)
1057         self.vapi.ip6_nd_proxy(
1058             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
1059             self.pg2.sw_if_index,
1060             is_del=1)
1061
1062         self.assertFalse(find_nbr(self,
1063                                   self.pg2.sw_if_index,
1064                                   self.pg0._remote_hosts[3].ip6,
1065                                   inet=AF_INET6))
1066         self.assertFalse(find_nbr(self,
1067                                   self.pg1.sw_if_index,
1068                                   self.pg0._remote_hosts[2].ip6,
1069                                   inet=AF_INET6))
1070
1071         #
1072         # no longer proxy-ing...
1073         #
1074         self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
1075         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
1076         self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
1077
1078         #
1079         # no longer forwarding. traffic generates NS out of the glean/main
1080         # interface
1081         #
1082         self.pg2.add_stream(t2)
1083         self.pg_enable_capture(self.pg_interfaces)
1084         self.pg_start()
1085
1086         rx = self.pg0.get_capture(1)
1087
1088         self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
1089
1090
1091 class TestIPNull(VppTestCase):
1092     """ IPv6 routes via NULL """
1093
1094     def setUp(self):
1095         super(TestIPNull, self).setUp()
1096
1097         # create 2 pg interfaces
1098         self.create_pg_interfaces(range(1))
1099
1100         for i in self.pg_interfaces:
1101             i.admin_up()
1102             i.config_ip6()
1103             i.resolve_ndp()
1104
1105     def tearDown(self):
1106         super(TestIPNull, self).tearDown()
1107         for i in self.pg_interfaces:
1108             i.unconfig_ip6()
1109             i.admin_down()
1110
1111     def test_ip_null(self):
1112         """ IP NULL route """
1113
1114         p = (Ether(src=self.pg0.remote_mac,
1115                    dst=self.pg0.local_mac) /
1116              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
1117              UDP(sport=1234, dport=1234) /
1118              Raw('\xa5' * 100))
1119
1120         #
1121         # A route via IP NULL that will reply with ICMP unreachables
1122         #
1123         ip_unreach = VppIpRoute(self, "2001::", 64, [], is_unreach=1, is_ip6=1)
1124         ip_unreach.add_vpp_config()
1125
1126         self.pg0.add_stream(p)
1127         self.pg_enable_capture(self.pg_interfaces)
1128         self.pg_start()
1129
1130         rx = self.pg0.get_capture(1)
1131         rx = rx[0]
1132         icmp = rx[ICMPv6DestUnreach]
1133
1134         # 0 = "No route to destination"
1135         self.assertEqual(icmp.code, 0)
1136
1137         # ICMP is rate limited. pause a bit
1138         self.sleep(1)
1139
1140         #
1141         # A route via IP NULL that will reply with ICMP prohibited
1142         #
1143         ip_prohibit = VppIpRoute(self, "2001::1", 128, [],
1144                                  is_prohibit=1, is_ip6=1)
1145         ip_prohibit.add_vpp_config()
1146
1147         self.pg0.add_stream(p)
1148         self.pg_enable_capture(self.pg_interfaces)
1149         self.pg_start()
1150
1151         rx = self.pg0.get_capture(1)
1152         rx = rx[0]
1153         icmp = rx[ICMPv6DestUnreach]
1154
1155         # 1 = "Communication with destination administratively prohibited"
1156         self.assertEqual(icmp.code, 1)
1157
1158
1159 class TestIPDisabled(VppTestCase):
1160     """ IPv6 disabled """
1161
1162     def setUp(self):
1163         super(TestIPDisabled, self).setUp()
1164
1165         # create 2 pg interfaces
1166         self.create_pg_interfaces(range(2))
1167
1168         # PG0 is IP enalbed
1169         self.pg0.admin_up()
1170         self.pg0.config_ip6()
1171         self.pg0.resolve_ndp()
1172
1173         # PG 1 is not IP enabled
1174         self.pg1.admin_up()
1175
1176     def tearDown(self):
1177         super(TestIPDisabled, self).tearDown()
1178         for i in self.pg_interfaces:
1179             i.unconfig_ip4()
1180             i.admin_down()
1181
1182     def send_and_assert_no_replies(self, intf, pkts, remark):
1183         intf.add_stream(pkts)
1184         self.pg_enable_capture(self.pg_interfaces)
1185         self.pg_start()
1186         for i in self.pg_interfaces:
1187             i.get_capture(0)
1188             i.assert_nothing_captured(remark=remark)
1189
1190     def test_ip_disabled(self):
1191         """ IP Disabled """
1192
1193         #
1194         # An (S,G).
1195         # one accepting interface, pg0, 2 forwarding interfaces
1196         #
1197         route_ff_01 = VppIpMRoute(
1198             self,
1199             "::",
1200             "ffef::1", 128,
1201             MRouteEntryFlags.MFIB_ENTRY_FLAG_NONE,
1202             [VppMRoutePath(self.pg1.sw_if_index,
1203                            MRouteItfFlags.MFIB_ITF_FLAG_ACCEPT),
1204              VppMRoutePath(self.pg0.sw_if_index,
1205                            MRouteItfFlags.MFIB_ITF_FLAG_FORWARD)],
1206             is_ip6=1)
1207         route_ff_01.add_vpp_config()
1208
1209         pu = (Ether(src=self.pg1.remote_mac,
1210                     dst=self.pg1.local_mac) /
1211               IPv6(src="2001::1", dst=self.pg0.remote_ip6) /
1212               UDP(sport=1234, dport=1234) /
1213               Raw('\xa5' * 100))
1214         pm = (Ether(src=self.pg1.remote_mac,
1215                     dst=self.pg1.local_mac) /
1216               IPv6(src="2001::1", dst="ffef::1") /
1217               UDP(sport=1234, dport=1234) /
1218               Raw('\xa5' * 100))
1219
1220         #
1221         # PG1 does not forward IP traffic
1222         #
1223         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1224         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1225
1226         #
1227         # IP enable PG1
1228         #
1229         self.pg1.config_ip6()
1230
1231         #
1232         # Now we get packets through
1233         #
1234         self.pg1.add_stream(pu)
1235         self.pg_enable_capture(self.pg_interfaces)
1236         self.pg_start()
1237         rx = self.pg0.get_capture(1)
1238
1239         self.pg1.add_stream(pm)
1240         self.pg_enable_capture(self.pg_interfaces)
1241         self.pg_start()
1242         rx = self.pg0.get_capture(1)
1243
1244         #
1245         # Disable PG1
1246         #
1247         self.pg1.unconfig_ip6()
1248
1249         #
1250         # PG1 does not forward IP traffic
1251         #
1252         self.send_and_assert_no_replies(self.pg1, pu, "IPv6 disabled")
1253         self.send_and_assert_no_replies(self.pg1, pm, "IPv6 disabled")
1254
1255
1256 class TestIP6LoadBalance(VppTestCase):
1257     """ IPv6 Load-Balancing """
1258
1259     def setUp(self):
1260         super(TestIP6LoadBalance, self).setUp()
1261
1262         self.create_pg_interfaces(range(5))
1263
1264         mpls_tbl = VppMplsTable(self, 0)
1265         mpls_tbl.add_vpp_config()
1266
1267         for i in self.pg_interfaces:
1268             i.admin_up()
1269             i.config_ip6()
1270             i.resolve_ndp()
1271             i.enable_mpls()
1272
1273     def tearDown(self):
1274         for i in self.pg_interfaces:
1275             i.unconfig_ip6()
1276             i.admin_down()
1277             i.disable_mpls()
1278         super(TestIP6LoadBalance, self).tearDown()
1279
1280     def send_and_expect_load_balancing(self, input, pkts, outputs):
1281         self.vapi.cli("clear trace")
1282         input.add_stream(pkts)
1283         self.pg_enable_capture(self.pg_interfaces)
1284         self.pg_start()
1285         for oo in outputs:
1286             rx = oo._get_capture(1)
1287             self.assertNotEqual(0, len(rx))
1288
1289     def send_and_expect_one_itf(self, input, pkts, itf):
1290         self.vapi.cli("clear trace")
1291         input.add_stream(pkts)
1292         self.pg_enable_capture(self.pg_interfaces)
1293         self.pg_start()
1294         rx = itf.get_capture(len(pkts))
1295
1296     def test_ip6_load_balance(self):
1297         """ IPv6 Load-Balancing """
1298
1299         #
1300         # An array of packets that differ only in the destination port
1301         #  - IP only
1302         #  - MPLS EOS
1303         #  - MPLS non-EOS
1304         #  - MPLS non-EOS with an entropy label
1305         #
1306         port_ip_pkts = []
1307         port_mpls_pkts = []
1308         port_mpls_neos_pkts = []
1309         port_ent_pkts = []
1310
1311         #
1312         # An array of packets that differ only in the source address
1313         #
1314         src_ip_pkts = []
1315         src_mpls_pkts = []
1316
1317         for ii in range(65):
1318             port_ip_hdr = (IPv6(dst="3000::1", src="3000:1::1") /
1319                            UDP(sport=1234, dport=1234 + ii) /
1320                            Raw('\xa5' * 100))
1321             port_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1322                                        dst=self.pg0.local_mac) /
1323                                  port_ip_hdr))
1324             port_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1325                                          dst=self.pg0.local_mac) /
1326                                    MPLS(label=66, ttl=2) /
1327                                    port_ip_hdr))
1328             port_mpls_neos_pkts.append((Ether(src=self.pg0.remote_mac,
1329                                               dst=self.pg0.local_mac) /
1330                                         MPLS(label=67, ttl=2) /
1331                                         MPLS(label=77, ttl=2) /
1332                                         port_ip_hdr))
1333             port_ent_pkts.append((Ether(src=self.pg0.remote_mac,
1334                                         dst=self.pg0.local_mac) /
1335                                   MPLS(label=67, ttl=2) /
1336                                   MPLS(label=14, ttl=2) /
1337                                   MPLS(label=999, ttl=2) /
1338                                   port_ip_hdr))
1339             src_ip_hdr = (IPv6(dst="3000::1", src="3000:1::%d" % ii) /
1340                           UDP(sport=1234, dport=1234) /
1341                           Raw('\xa5' * 100))
1342             src_ip_pkts.append((Ether(src=self.pg0.remote_mac,
1343                                       dst=self.pg0.local_mac) /
1344                                 src_ip_hdr))
1345             src_mpls_pkts.append((Ether(src=self.pg0.remote_mac,
1346                                         dst=self.pg0.local_mac) /
1347                                   MPLS(label=66, ttl=2) /
1348                                   src_ip_hdr))
1349
1350         #
1351         # A route for the IP pacekts
1352         #
1353         route_3000_1 = VppIpRoute(self, "3000::1", 128,
1354                                   [VppRoutePath(self.pg1.remote_ip6,
1355                                                 self.pg1.sw_if_index,
1356                                                 proto=DpoProto.DPO_PROTO_IP6),
1357                                    VppRoutePath(self.pg2.remote_ip6,
1358                                                 self.pg2.sw_if_index,
1359                                                 proto=DpoProto.DPO_PROTO_IP6)],
1360                                   is_ip6=1)
1361         route_3000_1.add_vpp_config()
1362
1363         #
1364         # a local-label for the EOS packets
1365         #
1366         binding = VppMplsIpBind(self, 66, "3000::1", 128, is_ip6=1)
1367         binding.add_vpp_config()
1368
1369         #
1370         # An MPLS route for the non-EOS packets
1371         #
1372         route_67 = VppMplsRoute(self, 67, 0,
1373                                 [VppRoutePath(self.pg1.remote_ip6,
1374                                               self.pg1.sw_if_index,
1375                                               labels=[67],
1376                                               proto=DpoProto.DPO_PROTO_IP6),
1377                                  VppRoutePath(self.pg2.remote_ip6,
1378                                               self.pg2.sw_if_index,
1379                                               labels=[67],
1380                                               proto=DpoProto.DPO_PROTO_IP6)])
1381         route_67.add_vpp_config()
1382
1383         #
1384         # inject the packet on pg0 - expect load-balancing across the 2 paths
1385         #  - since the default hash config is to use IP src,dst and port
1386         #    src,dst
1387         # We are not going to ensure equal amounts of packets across each link,
1388         # since the hash algorithm is statistical and therefore this can never
1389         # be guaranteed. But wuth 64 different packets we do expect some
1390         # balancing. So instead just ensure there is traffic on each link.
1391         #
1392         self.send_and_expect_load_balancing(self.pg0, port_ip_pkts,
1393                                             [self.pg1, self.pg2])
1394         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1395                                             [self.pg1, self.pg2])
1396         self.send_and_expect_load_balancing(self.pg0, port_mpls_pkts,
1397                                             [self.pg1, self.pg2])
1398         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1399                                             [self.pg1, self.pg2])
1400         self.send_and_expect_load_balancing(self.pg0, port_mpls_neos_pkts,
1401                                             [self.pg1, self.pg2])
1402
1403         #
1404         # The packets with Entropy label in should not load-balance,
1405         # since the Entorpy value is fixed.
1406         #
1407         self.send_and_expect_one_itf(self.pg0, port_ent_pkts, self.pg1)
1408
1409         #
1410         # change the flow hash config so it's only IP src,dst
1411         #  - now only the stream with differing source address will
1412         #    load-balance
1413         #
1414         self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=0, dport=0)
1415
1416         self.send_and_expect_load_balancing(self.pg0, src_ip_pkts,
1417                                             [self.pg1, self.pg2])
1418         self.send_and_expect_load_balancing(self.pg0, src_mpls_pkts,
1419                                             [self.pg1, self.pg2])
1420         self.send_and_expect_one_itf(self.pg0, port_ip_pkts, self.pg2)
1421
1422         #
1423         # change the flow hash config back to defaults
1424         #
1425         self.vapi.set_ip_flow_hash(0, is_ip6=1, src=1, dst=1, sport=1, dport=1)
1426
1427         #
1428         # Recursive prefixes
1429         #  - testing that 2 stages of load-balancing occurs and there is no
1430         #    polarisation (i.e. only 2 of 4 paths are used)
1431         #
1432         port_pkts = []
1433         src_pkts = []
1434
1435         for ii in range(257):
1436             port_pkts.append((Ether(src=self.pg0.remote_mac,
1437                                     dst=self.pg0.local_mac) /
1438                               IPv6(dst="4000::1", src="4000:1::1") /
1439                               UDP(sport=1234, dport=1234 + ii) /
1440                               Raw('\xa5' * 100)))
1441             src_pkts.append((Ether(src=self.pg0.remote_mac,
1442                                    dst=self.pg0.local_mac) /
1443                              IPv6(dst="4000::1", src="4000:1::%d" % ii) /
1444                              UDP(sport=1234, dport=1234) /
1445                              Raw('\xa5' * 100)))
1446
1447         route_3000_2 = VppIpRoute(self, "3000::2", 128,
1448                                   [VppRoutePath(self.pg3.remote_ip6,
1449                                                 self.pg3.sw_if_index,
1450                                                 proto=DpoProto.DPO_PROTO_IP6),
1451                                    VppRoutePath(self.pg4.remote_ip6,
1452                                                 self.pg4.sw_if_index,
1453                                                 proto=DpoProto.DPO_PROTO_IP6)],
1454                                   is_ip6=1)
1455         route_3000_2.add_vpp_config()
1456
1457         route_4000_1 = VppIpRoute(self, "4000::1", 128,
1458                                   [VppRoutePath("3000::1",
1459                                                 0xffffffff,
1460                                                 proto=DpoProto.DPO_PROTO_IP6),
1461                                    VppRoutePath("3000::2",
1462                                                 0xffffffff,
1463                                                 proto=DpoProto.DPO_PROTO_IP6)],
1464                                   is_ip6=1)
1465         route_4000_1.add_vpp_config()
1466
1467         #
1468         # inject the packet on pg0 - expect load-balancing across all 4 paths
1469         #
1470         self.vapi.cli("clear trace")
1471         self.send_and_expect_load_balancing(self.pg0, port_pkts,
1472                                             [self.pg1, self.pg2,
1473                                              self.pg3, self.pg4])
1474         self.send_and_expect_load_balancing(self.pg0, src_pkts,
1475                                             [self.pg1, self.pg2,
1476                                              self.pg3, self.pg4])
1477
1478         #
1479         # Recursive prefixes
1480         #  - testing that 2 stages of load-balancing no choices
1481         #
1482         port_pkts = []
1483
1484         for ii in range(257):
1485             port_pkts.append((Ether(src=self.pg0.remote_mac,
1486                                     dst=self.pg0.local_mac) /
1487                               IPv6(dst="6000::1", src="6000:1::1") /
1488                               UDP(sport=1234, dport=1234 + ii) /
1489                               Raw('\xa5' * 100)))
1490
1491         route_5000_2 = VppIpRoute(self, "5000::2", 128,
1492                                   [VppRoutePath(self.pg3.remote_ip6,
1493                                                 self.pg3.sw_if_index,
1494                                                 proto=DpoProto.DPO_PROTO_IP6)],
1495                                   is_ip6=1)
1496         route_5000_2.add_vpp_config()
1497
1498         route_6000_1 = VppIpRoute(self, "6000::1", 128,
1499                                   [VppRoutePath("5000::2",
1500                                                 0xffffffff,
1501                                                 proto=DpoProto.DPO_PROTO_IP6)],
1502                                   is_ip6=1)
1503         route_6000_1.add_vpp_config()
1504
1505         #
1506         # inject the packet on pg0 - expect load-balancing across all 4 paths
1507         #
1508         self.vapi.cli("clear trace")
1509         self.send_and_expect_one_itf(self.pg0, port_pkts, self.pg3)
1510
1511
1512 class TestIP6Punt(VppTestCase):
1513     """ IPv6 Punt Police/Redirect """
1514
1515     def setUp(self):
1516         super(TestIP6Punt, self).setUp()
1517
1518         self.create_pg_interfaces(range(2))
1519
1520         for i in self.pg_interfaces:
1521             i.admin_up()
1522             i.config_ip6()
1523             i.resolve_ndp()
1524
1525     def tearDown(self):
1526         super(TestIP6Punt, self).tearDown()
1527         for i in self.pg_interfaces:
1528             i.unconfig_ip6()
1529             i.admin_down()
1530
1531     def send_and_expect(self, input, pkts, output):
1532         input.add_stream(pkts)
1533         self.pg_enable_capture(self.pg_interfaces)
1534         self.pg_start()
1535         rx = output.get_capture(len(pkts))
1536         return rx
1537
1538     def send_and_assert_no_replies(self, intf, pkts, remark):
1539         self.vapi.cli("clear trace")
1540         intf.add_stream(pkts)
1541         self.pg_enable_capture(self.pg_interfaces)
1542         self.pg_start()
1543         for i in self.pg_interfaces:
1544             i.get_capture(0)
1545             i.assert_nothing_captured(remark=remark)
1546
1547     def test_ip_punt(self):
1548         """ IP6 punt police and redirect """
1549
1550         p = (Ether(src=self.pg0.remote_mac,
1551                    dst=self.pg0.local_mac) /
1552              IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
1553              TCP(sport=1234, dport=1234) /
1554              Raw('\xa5' * 100))
1555
1556         pkts = p * 1025
1557
1558         #
1559         # Configure a punt redirect via pg1.
1560         #
1561         nh_addr = inet_pton(AF_INET6,
1562                             self.pg1.remote_ip6)
1563         self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
1564                                    self.pg1.sw_if_index,
1565                                    nh_addr,
1566                                    is_ip6=1)
1567
1568         self.send_and_expect(self.pg0, pkts, self.pg1)
1569
1570         #
1571         # add a policer
1572         #
1573         policer = self.vapi.policer_add_del("ip6-punt", 400, 0, 10, 0,
1574                                             rate_type=1)
1575         self.vapi.ip_punt_police(policer.policer_index, is_ip6=1)
1576
1577         self.vapi.cli("clear trace")
1578         self.pg0.add_stream(pkts)
1579         self.pg_enable_capture(self.pg_interfaces)
1580         self.pg_start()
1581
1582         #
1583         # the number of packet recieved should be greater than 0,
1584         # but not equal to the number sent, since some were policed
1585         #
1586         rx = self.pg1._get_capture(1)
1587         self.assertTrue(len(rx) > 0)
1588         self.assertTrue(len(rx) < len(pkts))
1589
1590         #
1591         # remove the poilcer. back to full rx
1592         #
1593         self.vapi.ip_punt_police(policer.policer_index, is_add=0, is_ip6=1)
1594         self.vapi.policer_add_del("ip6-punt", 400, 0, 10, 0,
1595                                   rate_type=1, is_add=0)
1596         self.send_and_expect(self.pg0, pkts, self.pg1)
1597
1598         #
1599         # remove the redirect. expect full drop.
1600         #
1601         self.vapi.ip_punt_redirect(self.pg0.sw_if_index,
1602                                    self.pg1.sw_if_index,
1603                                    nh_addr,
1604                                    is_add=0,
1605                                    is_ip6=1)
1606         self.send_and_assert_no_replies(self.pg0, pkts,
1607                                         "IP no punt config")
1608
1609         #
1610         # Add a redirect that is not input port selective
1611         #
1612         self.vapi.ip_punt_redirect(0xffffffff,
1613                                    self.pg1.sw_if_index,
1614                                    nh_addr,
1615                                    is_ip6=1)
1616         self.send_and_expect(self.pg0, pkts, self.pg1)
1617
1618         self.vapi.ip_punt_redirect(0xffffffff,
1619                                    self.pg1.sw_if_index,
1620                                    nh_addr,
1621                                    is_add=0,
1622                                    is_ip6=1)
1623
1624
1625 class TestIP6Input(VppTestCase):
1626     """ IPv6 Input Exceptions """
1627
1628     def setUp(self):
1629         super(TestIP6Input, self).setUp()
1630
1631         self.create_pg_interfaces(range(2))
1632
1633         for i in self.pg_interfaces:
1634             i.admin_up()
1635             i.config_ip6()
1636             i.resolve_ndp()
1637
1638     def tearDown(self):
1639         super(TestIP6Input, self).tearDown()
1640         for i in self.pg_interfaces:
1641             i.unconfig_ip6()
1642             i.admin_down()
1643
1644     def send_and_expect(self, input, pkts, output):
1645         self.vapi.cli("clear trace")
1646         input.add_stream(pkts)
1647         self.pg_enable_capture(self.pg_interfaces)
1648         self.pg_start()
1649         rx = output.get_capture(len(pkts))
1650         return rx
1651
1652     def send_and_assert_no_replies(self, intf, pkts, remark):
1653         self.vapi.cli("clear trace")
1654         intf.add_stream(pkts)
1655         self.pg_enable_capture(self.pg_interfaces)
1656         self.pg_start()
1657         for i in self.pg_interfaces:
1658             i.get_capture(0)
1659             i.assert_nothing_captured(remark=remark)
1660
1661     def test_ip_input(self):
1662         """ IP6 Input Exceptions """
1663
1664         #
1665         # bad version - this is dropped
1666         #
1667         p_version = (Ether(src=self.pg0.remote_mac,
1668                            dst=self.pg0.local_mac) /
1669                      IPv6(src=self.pg0.remote_ip6,
1670                           dst=self.pg1.remote_ip6,
1671                           version=3) /
1672                      UDP(sport=1234, dport=1234) /
1673                      Raw('\xa5' * 100))
1674
1675         self.send_and_assert_no_replies(self.pg0, p_version * 65,
1676                                         "funky version")
1677
1678         #
1679         # hop limit - IMCP replies
1680         #
1681         p_version = (Ether(src=self.pg0.remote_mac,
1682                            dst=self.pg0.local_mac) /
1683                      IPv6(src=self.pg0.remote_ip6,
1684                           dst=self.pg1.remote_ip6,
1685                           hlim=1) /
1686                      UDP(sport=1234, dport=1234) /
1687                      Raw('\xa5' * 100))
1688
1689         rx = self.send_and_expect(self.pg0, p_version * 65, self.pg0)
1690         rx = rx[0]
1691         icmp = rx[ICMPv6TimeExceeded]
1692         self.assertEqual(icmp.type, 3)
1693         # 0: "hop limit exceeded in transit",
1694         self.assertEqual(icmp.code, 0)
1695
1696
1697 if __name__ == '__main__':
1698     unittest.main(testRunner=VppTestRunner)