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
  10 from vpp_neighbor import find_nbr, VppNeighbor
  11
  12 from scapy.packet import Raw
  13 from scapy.layers.l2 import Ether, Dot1Q
  14 from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, \
  15     ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation, \
  16     ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
  17     ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types
  18
  19 from util import ppp
  20 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
  21     in6_mactoifaceid, in6_ismaddr
  22 from scapy.utils import inet_pton, inet_ntop
  23
  24
  25 def mk_ll_addr(mac):
  26     euid = in6_mactoifaceid(mac)
  27     addr = "fe80::" + euid
  28     return addr
  29
  30
  31 class TestIPv6ND(VppTestCase):
  32     def validate_ra(self, intf, rx, dst_ip=None):
  33         if not dst_ip:
  34             dst_ip = intf.remote_ip6
  35
  36         # unicasted packets must come to the unicast mac
  37         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  38
  39         # and from the router's MAC
  40         self.assertEqual(rx[Ether].src, intf.local_mac)
  41
  42         # the rx'd RA should be addressed to the sender's source
  43         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
  44         self.assertEqual(in6_ptop(rx[IPv6].dst),
  45                          in6_ptop(dst_ip))
  46
  47         # and come from the router's link local
  48         self.assertTrue(in6_islladdr(rx[IPv6].src))
  49         self.assertEqual(in6_ptop(rx[IPv6].src),
  50                          in6_ptop(mk_ll_addr(intf.local_mac)))
  51
  52     def validate_na(self, intf, rx, dst_ip=None, tgt_ip=None):
  53         if not dst_ip:
  54             dst_ip = intf.remote_ip6
  55         if not tgt_ip:
  56             dst_ip = intf.local_ip6
  57
  58         # unicasted packets must come to the unicast mac
  59         self.assertEqual(rx[Ether].dst, intf.remote_mac)
  60
  61         # and from the router's MAC
  62         self.assertEqual(rx[Ether].src, intf.local_mac)
  63
  64         # the rx'd NA should be addressed to the sender's source
  65         self.assertTrue(rx.haslayer(ICMPv6ND_NA))
  66         self.assertEqual(in6_ptop(rx[IPv6].dst),
  67                          in6_ptop(dst_ip))
  68
  69         # and come from the target address
  70         self.assertEqual(in6_ptop(rx[IPv6].src), in6_ptop(tgt_ip))
  71
  72         # Dest link-layer options should have the router's MAC
  73         dll = rx[ICMPv6NDOptDstLLAddr]
  74         self.assertEqual(dll.lladdr, intf.local_mac)
  75
  76     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
  77                            filter_out_fn=is_ipv6_misc):
  78         intf.add_stream(pkts)
  79         self.pg0.add_stream(pkts)
  80         self.pg_enable_capture(self.pg_interfaces)
  81         self.pg_start()
  82         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
  83
  84         self.assertEqual(len(rx), 1)
  85         rx = rx[0]
  86         self.validate_ra(intf, rx, dst_ip)
  87
  88     def send_and_assert_no_replies(self, intf, pkts, remark):
  89         intf.add_stream(pkts)
  90         self.pg_enable_capture(self.pg_interfaces)
  91         self.pg_start()
  92         intf.assert_nothing_captured(remark=remark)
  93
  94
  95 class TestIPv6(TestIPv6ND):
  96     """ IPv6 Test Case """
  97
  98     @classmethod
  99     def setUpClass(cls):
 100         super(TestIPv6, cls).setUpClass()
 101
 102     def setUp(self):
 103         """
 104         Perform test setup before test case.
 105
 106         **Config:**
 107             - create 3 pg interfaces
 108                 - untagged pg0 interface
 109                 - Dot1Q subinterface on pg1
 110                 - Dot1AD subinterface on pg2
 111             - setup interfaces:
 112                 - put it into UP state
 113                 - set IPv6 addresses
 114                 - resolve neighbor address using NDP
 115             - configure 200 fib entries
 116
 117         :ivar list interfaces: pg interfaces and subinterfaces.
 118         :ivar dict flows: IPv4 packet flows in test.
 119         :ivar list pg_if_packet_sizes: packet sizes in test.
 120
 121         *TODO:* Create AD sub interface
 122         """
 123         super(TestIPv6, self).setUp()
 124
 125         # create 3 pg interfaces
 126         self.create_pg_interfaces(range(3))
 127
 128         # create 2 subinterfaces for p1 and pg2
 129         self.sub_interfaces = [
 130             VppDot1QSubint(self, self.pg1, 100),
 131             VppDot1QSubint(self, self.pg2, 200)
 132             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
 133         ]
 134
 135         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
 136         self.flows = dict()
 137         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
 138         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
 139         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
 140
 141         # packet sizes
 142         self.pg_if_packet_sizes = [64, 512, 1518, 9018]
 143         self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
 144
 145         self.interfaces = list(self.pg_interfaces)
 146         self.interfaces.extend(self.sub_interfaces)
 147
 148         # setup all interfaces
 149         for i in self.interfaces:
 150             i.admin_up()
 151             i.config_ip6()
 152             i.resolve_ndp()
 153
 154         # config 2M FIB entries
 155         self.config_fib_entries(200)
 156
 157     def tearDown(self):
 158         """Run standard test teardown and log ``show ip6 neighbors``."""
 159         for i in self.sub_interfaces:
 160             i.unconfig_ip6()
 161             i.ip6_disable()
 162             i.admin_down()
 163             i.remove_vpp_config()
 164
 165         super(TestIPv6, self).tearDown()
 166         if not self.vpp_dead:
 167             self.logger.info(self.vapi.cli("show ip6 neighbors"))
 168             # info(self.vapi.cli("show ip6 fib"))  # many entries
 169
 170     def config_fib_entries(self, count):
 171         """For each interface add to the FIB table *count* routes to
 172         "fd02::1/128" destination with interface's local address as next-hop
 173         address.
 174
 175         :param int count: Number of FIB entries.
 176
 177         - *TODO:* check if the next-hop address shouldn't be remote address
 178           instead of local address.
 179         """
 180         n_int = len(self.interfaces)
 181         percent = 0
 182         counter = 0.0
 183         dest_addr = inet_pton(AF_INET6, "fd02::1")
 184         dest_addr_len = 128
 185         for i in self.interfaces:
 186             next_hop_address = i.local_ip6n
 187             for j in range(count / n_int):
 188                 self.vapi.ip_add_del_route(
 189                     dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
 190                 counter += 1
 191                 if counter / count * 100 > percent:
 192                     self.logger.info("Configure %d FIB entries .. %d%% done" %
 193                                      (count, percent))
 194                     percent += 1
 195
 196     def create_stream(self, src_if, packet_sizes):
 197         """Create input packet stream for defined interface.
 198
 199         :param VppInterface src_if: Interface to create packet stream for.
 200         :param list packet_sizes: Required packet sizes.
 201         """
 202         pkts = []
 203         for i in range(0, 257):
 204             dst_if = self.flows[src_if][i % 2]
 205             info = self.create_packet_info(src_if, dst_if)
 206             payload = self.info_to_payload(info)
 207             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 208                  IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
 209                  UDP(sport=1234, dport=1234) /
 210                  Raw(payload))
 211             info.data = p.copy()
 212             if isinstance(src_if, VppSubInterface):
 213                 p = src_if.add_dot1_layer(p)
 214             size = packet_sizes[(i // 2) % len(packet_sizes)]
 215             self.extend_packet(p, size)
 216             pkts.append(p)
 217         return pkts
 218
 219     def verify_capture(self, dst_if, capture):
 220         """Verify captured input packet stream for defined interface.
 221
 222         :param VppInterface dst_if: Interface to verify captured packet stream
 223                                     for.
 224         :param list capture: Captured packet stream.
 225         """
 226         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 227         last_info = dict()
 228         for i in self.interfaces:
 229             last_info[i.sw_if_index] = None
 230         is_sub_if = False
 231         dst_sw_if_index = dst_if.sw_if_index
 232         if hasattr(dst_if, 'parent'):
 233             is_sub_if = True
 234         for packet in capture:
 235             if is_sub_if:
 236                 # Check VLAN tags and Ethernet header
 237                 packet = dst_if.remove_dot1_layer(packet)
 238             self.assertTrue(Dot1Q not in packet)
 239             try:
 240                 ip = packet[IPv6]
 241                 udp = packet[UDP]
 242                 payload_info = self.payload_to_info(str(packet[Raw]))
 243                 packet_index = payload_info.index
 244                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 245                 self.logger.debug(
 246                     "Got packet on port %s: src=%u (id=%u)" %
 247                     (dst_if.name, payload_info.src, packet_index))
 248                 next_info = self.get_next_packet_info_for_interface2(
 249                     payload_info.src, dst_sw_if_index,
 250                     last_info[payload_info.src])
 251                 last_info[payload_info.src] = next_info
 252                 self.assertTrue(next_info is not None)
 253                 self.assertEqual(packet_index, next_info.index)
 254                 saved_packet = next_info.data
 255                 # Check standard fields
 256                 self.assertEqual(ip.src, saved_packet[IPv6].src)
 257                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
 258                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 259                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 260             except:
 261                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 262                 raise
 263         for i in self.interfaces:
 264             remaining_packet = self.get_next_packet_info_for_interface2(
 265                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 266             self.assertTrue(remaining_packet is None,
 267                             "Interface %s: Packet expected from interface %s "
 268                             "didn't arrive" % (dst_if.name, i.name))
 269
 270     def test_fib(self):
 271         """ IPv6 FIB test
 272
 273         Test scenario:
 274             - Create IPv6 stream for pg0 interface
 275             - Create IPv6 tagged streams for pg1's and pg2's subinterface.
 276             - Send and verify received packets on each interface.
 277         """
 278
 279         pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
 280         self.pg0.add_stream(pkts)
 281
 282         for i in self.sub_interfaces:
 283             pkts = self.create_stream(i, self.sub_if_packet_sizes)
 284             i.parent.add_stream(pkts)
 285
 286         self.pg_enable_capture(self.pg_interfaces)
 287         self.pg_start()
 288
 289         pkts = self.pg0.get_capture()
 290         self.verify_capture(self.pg0, pkts)
 291
 292         for i in self.sub_interfaces:
 293             pkts = i.parent.get_capture()
 294             self.verify_capture(i, pkts)
 295
 296     def test_ns(self):
 297         """ IPv6 Neighbour Solicitation Exceptions
 298
 299         Test scenario:
 300            - Send an NS Sourced from an address not covered by the link sub-net
 301            - Send an NS to an mcast address the router has not joined
 302            - Send NS for a target address the router does not onn.
 303         """
 304
 305         #
 306         # An NS from a non link source address
 307         #
 308         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 309         d = inet_ntop(AF_INET6, nsma)
 310
 311         p = (Ether(dst=in6_getnsmac(nsma)) /
 312              IPv6(dst=d, src="2002::2") /
 313              ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 314              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 315         pkts = [p]
 316
 317         self.send_and_assert_no_replies(
 318             self.pg0, pkts,
 319             "No response to NS source by address not on sub-net")
 320
 321         #
 322         # An NS for sent to a solicited mcast group the router is
 323         # not a member of FAILS
 324         #
 325         if 0:
 326             nsma = in6_getnsma(inet_pton(AF_INET6, "fd::ffff"))
 327             d = inet_ntop(AF_INET6, nsma)
 328
 329             p = (Ether(dst=in6_getnsmac(nsma)) /
 330                  IPv6(dst=d, src=self.pg0.remote_ip6) /
 331                  ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 332                  ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 333             pkts = [p]
 334
 335             self.send_and_assert_no_replies(
 336                 self.pg0, pkts,
 337                 "No response to NS sent to unjoined mcast address")
 338
 339         #
 340         # An NS whose target address is one the router does not own
 341         #
 342         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 343         d = inet_ntop(AF_INET6, nsma)
 344
 345         p = (Ether(dst=in6_getnsmac(nsma)) /
 346              IPv6(dst=d, src=self.pg0.remote_ip6) /
 347              ICMPv6ND_NS(tgt="fd::ffff") /
 348              ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 349         pkts = [p]
 350
 351         self.send_and_assert_no_replies(self.pg0, pkts,
 352                                         "No response to NS for unknown target")
 353
 354         #
 355         # A neighbor entry that has no associated FIB-entry
 356         #
 357         self.pg0.generate_remote_hosts(4)
 358         nd_entry = VppNeighbor(self,
 359                                self.pg0.sw_if_index,
 360                                self.pg0.remote_hosts[2].mac,
 361                                self.pg0.remote_hosts[2].ip6,
 362                                af=AF_INET6,
 363                                is_no_fib_entry=1)
 364         nd_entry.add_vpp_config()
 365
 366         #
 367         # check we have the neighbor, but no route
 368         #
 369         self.assertTrue(find_nbr(self,
 370                                  self.pg0.sw_if_index,
 371                                  self.pg0._remote_hosts[2].ip6,
 372                                  inet=AF_INET6))
 373         self.assertFalse(find_route(self,
 374                                     self.pg0._remote_hosts[2].ip6,
 375                                     128,
 376                                     inet=AF_INET6))
 377
 378     def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None):
 379         if not dst_ip:
 380             dst_ip = intf.remote_ip6
 381
 382         # unicasted packets must come to the unicast mac
 383         self.assertEqual(rx[Ether].dst, intf.remote_mac)
 384
 385         # and from the router's MAC
 386         self.assertEqual(rx[Ether].src, intf.local_mac)
 387
 388         # the rx'd RA should be addressed to the sender's source
 389         self.assertTrue(rx.haslayer(ICMPv6ND_RA))
 390         self.assertEqual(in6_ptop(rx[IPv6].dst),
 391                          in6_ptop(dst_ip))
 392
 393         # and come from the router's link local
 394         self.assertTrue(in6_islladdr(rx[IPv6].src))
 395         self.assertEqual(in6_ptop(rx[IPv6].src),
 396                          in6_ptop(mk_ll_addr(intf.local_mac)))
 397
 398         # it should contain the links MTU
 399         ra = rx[ICMPv6ND_RA]
 400         self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
 401
 402         # it should contain the source's link layer address option
 403         sll = ra[ICMPv6NDOptSrcLLAddr]
 404         self.assertEqual(sll.lladdr, intf.local_mac)
 405
 406         if not pi_opt:
 407             # the RA should not contain prefix information
 408             self.assertFalse(ra.haslayer(ICMPv6NDOptPrefixInfo))
 409         else:
 410             raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
 411
 412             # the options are nested in the scapy packet in way that i cannot
 413             # decipher how to decode. this 1st layer of option always returns
 414             # nested classes, so a direct obj1=obj2 comparison always fails.
 415             # however, the getlayer(.., 2) does give one instnace.
 416             # so we cheat here and construct a new opt instnace for comparison
 417             rd = ICMPv6NDOptPrefixInfo(prefixlen=raos.prefixlen,
 418                                        prefix=raos.prefix,
 419                                        L=raos.L,
 420                                        A=raos.A)
 421             if type(pi_opt) is list:
 422                 for ii in range(len(pi_opt)):
 423                     self.assertEqual(pi_opt[ii], rd)
 424                     rd = rx.getlayer(ICMPv6NDOptPrefixInfo, ii+2)
 425             else:
 426                 self.assertEqual(pi_opt, raos)
 427
 428     def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
 429                            filter_out_fn=is_ipv6_misc,
 430                            opt=None):
 431         intf.add_stream(pkts)
 432         self.pg_enable_capture(self.pg_interfaces)
 433         self.pg_start()
 434         rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
 435
 436         self.assertEqual(len(rx), 1)
 437         rx = rx[0]
 438         self.validate_ra(intf, rx, dst_ip, pi_opt=opt)
 439
 440     def test_rs(self):
 441         """ IPv6 Router Solicitation Exceptions
 442
 443         Test scenario:
 444         """
 445
 446         #
 447         # Before we begin change the IPv6 RA responses to use the unicast
 448         # address - that way we will not confuse them with the periodic
 449         # RAs which go to the mcast address
 450         # Sit and wait for the first periodic RA.
 451         #
 452         # TODO
 453         #
 454         self.pg0.ip6_ra_config(send_unicast=1)
 455
 456         #
 457         # An RS from a link source address
 458         #  - expect an RA in return
 459         #
 460         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 461              IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
 462              ICMPv6ND_RS())
 463         pkts = [p]
 464         self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
 465
 466         #
 467         # For the next RS sent the RA should be rate limited
 468         #
 469         self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
 470
 471         #
 472         # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
 473         # so we need to do this before each test below so as not to drop
 474         # packets for rate limiting reasons. Test this works here.
 475         #
 476         self.pg0.ip6_ra_config(send_unicast=1)
 477         self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
 478
 479         #
 480         # An RS sent from a non-link local source
 481         #
 482         self.pg0.ip6_ra_config(send_unicast=1)
 483         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 484              IPv6(dst=self.pg0.local_ip6, src="2002::ffff") /
 485              ICMPv6ND_RS())
 486         pkts = [p]
 487         self.send_and_assert_no_replies(self.pg0, pkts,
 488                                         "RS from non-link source")
 489
 490         #
 491         # Source an RS from a link local address
 492         #
 493         self.pg0.ip6_ra_config(send_unicast=1)
 494         ll = mk_ll_addr(self.pg0.remote_mac)
 495         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 496              IPv6(dst=self.pg0.local_ip6, src=ll) /
 497              ICMPv6ND_RS())
 498         pkts = [p]
 499         self.send_and_expect_ra(self.pg0, pkts,
 500                                 "RS sourced from link-local",
 501                                 dst_ip=ll)
 502
 503         #
 504         # Send the RS multicast
 505         #
 506         self.pg0.ip6_ra_config(send_unicast=1)
 507         dmac = in6_getnsmac(inet_pton(AF_INET6, "ff02::2"))
 508         ll = mk_ll_addr(self.pg0.remote_mac)
 509         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 510              IPv6(dst="ff02::2", src=ll) /
 511              ICMPv6ND_RS())
 512         pkts = [p]
 513         self.send_and_expect_ra(self.pg0, pkts,
 514                                 "RS sourced from link-local",
 515                                 dst_ip=ll)
 516
 517         #
 518         # Source from the unspecified address ::. This happens when the RS
 519         # is sent before the host has a configured address/sub-net,
 520         # i.e. auto-config. Since the sender has no IP address, the reply
 521         # comes back mcast - so the capture needs to not filter this.
 522         # If we happen to pick up the periodic RA at this point then so be it,
 523         # it's not an error.
 524         #
 525         self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
 526         p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
 527              IPv6(dst="ff02::2", src="::") /
 528              ICMPv6ND_RS())
 529         pkts = [p]
 530         self.send_and_expect_ra(self.pg0, pkts,
 531                                 "RS sourced from unspecified",
 532                                 dst_ip="ff02::1",
 533                                 filter_out_fn=None)
 534
 535         #
 536         # Configure The RA to announce the links prefix
 537         #
 538         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 539                                self.pg0.local_ip6_prefix_len)
 540
 541         #
 542         # RAs should now contain the prefix information option
 543         #
 544         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 545                                     prefix=self.pg0.local_ip6,
 546                                     L=1,
 547                                     A=1)
 548
 549         self.pg0.ip6_ra_config(send_unicast=1)
 550         ll = mk_ll_addr(self.pg0.remote_mac)
 551         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 552              IPv6(dst=self.pg0.local_ip6, src=ll) /
 553              ICMPv6ND_RS())
 554         self.send_and_expect_ra(self.pg0, p,
 555                                 "RA with prefix-info",
 556                                 dst_ip=ll,
 557                                 opt=opt)
 558
 559         #
 560         # Change the prefix info to not off-link
 561         #  L-flag is clear
 562         #
 563         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 564                                self.pg0.local_ip6_prefix_len,
 565                                off_link=1)
 566
 567         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 568                                     prefix=self.pg0.local_ip6,
 569                                     L=0,
 570                                     A=1)
 571
 572         self.pg0.ip6_ra_config(send_unicast=1)
 573         self.send_and_expect_ra(self.pg0, p,
 574                                 "RA with Prefix info with L-flag=0",
 575                                 dst_ip=ll,
 576                                 opt=opt)
 577
 578         #
 579         # Change the prefix info to not off-link, no-autoconfig
 580         #  L and A flag are clear in the advert
 581         #
 582         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 583                                self.pg0.local_ip6_prefix_len,
 584                                off_link=1,
 585                                no_autoconfig=1)
 586
 587         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 588                                     prefix=self.pg0.local_ip6,
 589                                     L=0,
 590                                     A=0)
 591
 592         self.pg0.ip6_ra_config(send_unicast=1)
 593         self.send_and_expect_ra(self.pg0, p,
 594                                 "RA with Prefix info with A & L-flag=0",
 595                                 dst_ip=ll,
 596                                 opt=opt)
 597
 598         #
 599         # Change the flag settings back to the defaults
 600         #  L and A flag are set in the advert
 601         #
 602         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 603                                self.pg0.local_ip6_prefix_len)
 604
 605         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 606                                     prefix=self.pg0.local_ip6,
 607                                     L=1,
 608                                     A=1)
 609
 610         self.pg0.ip6_ra_config(send_unicast=1)
 611         self.send_and_expect_ra(self.pg0, p,
 612                                 "RA with Prefix info",
 613                                 dst_ip=ll,
 614                                 opt=opt)
 615
 616         #
 617         # Change the prefix info to not off-link, no-autoconfig
 618         #  L and A flag are clear in the advert
 619         #
 620         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 621                                self.pg0.local_ip6_prefix_len,
 622                                off_link=1,
 623                                no_autoconfig=1)
 624
 625         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 626                                     prefix=self.pg0.local_ip6,
 627                                     L=0,
 628                                     A=0)
 629
 630         self.pg0.ip6_ra_config(send_unicast=1)
 631         self.send_and_expect_ra(self.pg0, p,
 632                                 "RA with Prefix info with A & L-flag=0",
 633                                 dst_ip=ll,
 634                                 opt=opt)
 635
 636         #
 637         # Use the reset to defults option to revert to defaults
 638         #  L and A flag are clear in the advert
 639         #
 640         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 641                                self.pg0.local_ip6_prefix_len,
 642                                use_default=1)
 643
 644         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 645                                     prefix=self.pg0.local_ip6,
 646                                     L=1,
 647                                     A=1)
 648
 649         self.pg0.ip6_ra_config(send_unicast=1)
 650         self.send_and_expect_ra(self.pg0, p,
 651                                 "RA with Prefix reverted to defaults",
 652                                 dst_ip=ll,
 653                                 opt=opt)
 654
 655         #
 656         # Advertise Another prefix. With no L-flag/A-flag
 657         #
 658         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 659                                self.pg1.local_ip6_prefix_len,
 660                                off_link=1,
 661                                no_autoconfig=1)
 662
 663         opt = [ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
 664                                      prefix=self.pg0.local_ip6,
 665                                      L=1,
 666                                      A=1),
 667                ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
 668                                      prefix=self.pg1.local_ip6,
 669                                      L=0,
 670                                      A=0)]
 671
 672         self.pg0.ip6_ra_config(send_unicast=1)
 673         ll = mk_ll_addr(self.pg0.remote_mac)
 674         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 675              IPv6(dst=self.pg0.local_ip6, src=ll) /
 676              ICMPv6ND_RS())
 677         self.send_and_expect_ra(self.pg0, p,
 678                                 "RA with multiple Prefix infos",
 679                                 dst_ip=ll,
 680                                 opt=opt)
 681
 682         #
 683         # Remove the first refix-info - expect the second is still in the
 684         # advert
 685         #
 686         self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
 687                                self.pg0.local_ip6_prefix_len,
 688                                is_no=1)
 689
 690         opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
 691                                     prefix=self.pg1.local_ip6,
 692                                     L=0,
 693                                     A=0)
 694
 695         self.pg0.ip6_ra_config(send_unicast=1)
 696         self.send_and_expect_ra(self.pg0, p,
 697                                 "RA with Prefix reverted to defaults",
 698                                 dst_ip=ll,
 699                                 opt=opt)
 700
 701         #
 702         # Remove the second prefix-info - expect no prefix-info i nthe adverts
 703         #
 704         self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
 705                                self.pg1.local_ip6_prefix_len,
 706                                is_no=1)
 707
 708         self.pg0.ip6_ra_config(send_unicast=1)
 709         self.send_and_expect_ra(self.pg0, p,
 710                                 "RA with Prefix reverted to defaults",
 711                                 dst_ip=ll)
 712
 713         #
 714         # Reset the periodic advertisements back to default values
 715         #
 716         self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
 717
 718
 719 class IPv6NDProxyTest(TestIPv6ND):
 720     """ IPv6 ND ProxyTest Case """
 721
 722     def setUp(self):
 723         super(IPv6NDProxyTest, self).setUp()
 724
 725         # create 3 pg interfaces
 726         self.create_pg_interfaces(range(3))
 727
 728         # pg0 is the master interface, with the configured subnet
 729         self.pg0.admin_up()
 730         self.pg0.config_ip6()
 731         self.pg0.resolve_ndp()
 732
 733         self.pg1.ip6_enable()
 734         self.pg2.ip6_enable()
 735
 736     def tearDown(self):
 737         super(IPv6NDProxyTest, self).tearDown()
 738
 739     def test_nd_proxy(self):
 740         """ IPv6 Proxy ND """
 741
 742         #
 743         # Generate some hosts in the subnet that we are proxying
 744         #
 745         self.pg0.generate_remote_hosts(8)
 746
 747         nsma = in6_getnsma(inet_pton(AF_INET6, self.pg0.local_ip6))
 748         d = inet_ntop(AF_INET6, nsma)
 749
 750         #
 751         # Send an NS for one of those remote hosts on one of the proxy links
 752         # expect no response since it's from an address that is not
 753         # on the link that has the prefix configured
 754         #
 755         ns_pg1 = (Ether(dst=in6_getnsmac(nsma), src=self.pg1.remote_mac) /
 756                   IPv6(dst=d, src=self.pg0._remote_hosts[2].ip6) /
 757                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 758                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
 759
 760         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Off link NS")
 761
 762         #
 763         # Add proxy support for the host
 764         #
 765         self.vapi.ip6_nd_proxy(
 766             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
 767             self.pg1.sw_if_index)
 768
 769         #
 770         # try that NS again. this time we expect an NA back
 771         #
 772         self.pg1.add_stream(ns_pg1)
 773         self.pg_enable_capture(self.pg_interfaces)
 774         self.pg_start()
 775         rx = self.pg1.get_capture(1)
 776
 777         self.validate_na(self.pg1, rx[0],
 778                          dst_ip=self.pg0._remote_hosts[2].ip6,
 779                          tgt_ip=self.pg0.local_ip6)
 780
 781         #
 782         # ... and that we have an entry in the ND cache
 783         #
 784         self.assertTrue(find_nbr(self,
 785                                  self.pg1.sw_if_index,
 786                                  self.pg0._remote_hosts[2].ip6,
 787                                  inet=AF_INET6))
 788
 789         #
 790         # ... and we can route traffic to it
 791         #
 792         t = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 793              IPv6(dst=self.pg0._remote_hosts[2].ip6,
 794                   src=self.pg0.remote_ip6) /
 795              UDP(sport=10000, dport=20000) /
 796              Raw('\xa5' * 100))
 797
 798         self.pg0.add_stream(t)
 799         self.pg_enable_capture(self.pg_interfaces)
 800         self.pg_start()
 801         rx = self.pg1.get_capture(1)
 802         rx = rx[0]
 803
 804         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
 805         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
 806
 807         self.assertEqual(rx[IPv6].src, t[IPv6].src)
 808         self.assertEqual(rx[IPv6].dst, t[IPv6].dst)
 809
 810         #
 811         # Test we proxy for the host on the main interface
 812         #
 813         ns_pg0 = (Ether(dst=in6_getnsmac(nsma), src=self.pg0.remote_mac) /
 814                   IPv6(dst=d, src=self.pg0.remote_ip6) /
 815                   ICMPv6ND_NS(tgt=self.pg0._remote_hosts[2].ip6) /
 816                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
 817
 818         self.pg0.add_stream(ns_pg0)
 819         self.pg_enable_capture(self.pg_interfaces)
 820         self.pg_start()
 821         rx = self.pg0.get_capture(1)
 822
 823         self.validate_na(self.pg0, rx[0],
 824                          tgt_ip=self.pg0._remote_hosts[2].ip6,
 825                          dst_ip=self.pg0.remote_ip6)
 826
 827         #
 828         # Setup and resolve proxy for another host on another interface
 829         #
 830         ns_pg2 = (Ether(dst=in6_getnsmac(nsma), src=self.pg2.remote_mac) /
 831                   IPv6(dst=d, src=self.pg0._remote_hosts[3].ip6) /
 832                   ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
 833                   ICMPv6NDOptSrcLLAddr(lladdr=self.pg0._remote_hosts[2].mac))
 834
 835         self.vapi.ip6_nd_proxy(
 836             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
 837             self.pg2.sw_if_index)
 838
 839         self.pg2.add_stream(ns_pg2)
 840         self.pg_enable_capture(self.pg_interfaces)
 841         self.pg_start()
 842         rx = self.pg2.get_capture(1)
 843
 844         self.validate_na(self.pg2, rx[0],
 845                          dst_ip=self.pg0._remote_hosts[3].ip6,
 846                          tgt_ip=self.pg0.local_ip6)
 847
 848         self.assertTrue(find_nbr(self,
 849                                  self.pg2.sw_if_index,
 850                                  self.pg0._remote_hosts[3].ip6,
 851                                  inet=AF_INET6))
 852
 853         #
 854         # hosts can communicate. pg2->pg1
 855         #
 856         t2 = (Ether(dst=self.pg2.local_mac,
 857                     src=self.pg0.remote_hosts[3].mac) /
 858               IPv6(dst=self.pg0._remote_hosts[2].ip6,
 859                    src=self.pg0._remote_hosts[3].ip6) /
 860               UDP(sport=10000, dport=20000) /
 861               Raw('\xa5' * 100))
 862
 863         self.pg2.add_stream(t2)
 864         self.pg_enable_capture(self.pg_interfaces)
 865         self.pg_start()
 866         rx = self.pg1.get_capture(1)
 867         rx = rx[0]
 868
 869         self.assertEqual(rx[Ether].dst, self.pg0._remote_hosts[2].mac)
 870         self.assertEqual(rx[Ether].src, self.pg1.local_mac)
 871
 872         self.assertEqual(rx[IPv6].src, t2[IPv6].src)
 873         self.assertEqual(rx[IPv6].dst, t2[IPv6].dst)
 874
 875         #
 876         # remove the proxy configs
 877         #
 878         self.vapi.ip6_nd_proxy(
 879             inet_pton(AF_INET6, self.pg0._remote_hosts[2].ip6),
 880             self.pg1.sw_if_index,
 881             is_del=1)
 882         self.vapi.ip6_nd_proxy(
 883             inet_pton(AF_INET6, self.pg0._remote_hosts[3].ip6),
 884             self.pg2.sw_if_index,
 885             is_del=1)
 886
 887         self.assertFalse(find_nbr(self,
 888                                   self.pg2.sw_if_index,
 889                                   self.pg0._remote_hosts[3].ip6,
 890                                   inet=AF_INET6))
 891         self.assertFalse(find_nbr(self,
 892                                   self.pg1.sw_if_index,
 893                                   self.pg0._remote_hosts[2].ip6,
 894                                   inet=AF_INET6))
 895
 896         #
 897         # no longer proxy-ing...
 898         #
 899         self.send_and_assert_no_replies(self.pg0, ns_pg0, "Proxy unconfigured")
 900         self.send_and_assert_no_replies(self.pg1, ns_pg1, "Proxy unconfigured")
 901         self.send_and_assert_no_replies(self.pg2, ns_pg2, "Proxy unconfigured")
 902
 903         #
 904         # no longer forwarding. traffic generates NS out of the glean/main
 905         # interface
 906         #
 907         self.pg2.add_stream(t2)
 908         self.pg_enable_capture(self.pg_interfaces)
 909         self.pg_start()
 910
 911         rx = self.pg0.get_capture(1)
 912
 913         self.assertTrue(rx[0].haslayer(ICMPv6ND_NS))
 914
 915
 916 class TestIPNull(VppTestCase):
 917     """ IPv6 routes via NULL """
 918
 919     def setUp(self):
 920         super(TestIPNull, self).setUp()
 921
 922         # create 2 pg interfaces
 923         self.create_pg_interfaces(range(1))
 924
 925         for i in self.pg_interfaces:
 926             i.admin_up()
 927             i.config_ip6()
 928             i.resolve_ndp()
 929
 930     def tearDown(self):
 931         super(TestIPNull, self).tearDown()
 932         for i in self.pg_interfaces:
 933             i.unconfig_ip6()
 934             i.admin_down()
 935
 936     def test_ip_null(self):
 937         """ IP NULL route """
 938
 939         p = (Ether(src=self.pg0.remote_mac,
 940                    dst=self.pg0.local_mac) /
 941              IPv6(src=self.pg0.remote_ip6, dst="2001::1") /
 942              UDP(sport=1234, dport=1234) /
 943              Raw('\xa5' * 100))
 944
 945         #
 946         # A route via IP NULL that will reply with ICMP unreachables
 947         #
 948         ip_unreach = VppIpRoute(self, "2001::", 64, [], is_unreach=1, is_ip6=1)
 949         ip_unreach.add_vpp_config()
 950
 951         self.pg0.add_stream(p)
 952         self.pg_enable_capture(self.pg_interfaces)
 953         self.pg_start()
 954
 955         rx = self.pg0.get_capture(1)
 956         rx = rx[0]
 957         icmp = rx[ICMPv6DestUnreach]
 958
 959         # 0 = "No route to destination"
 960         self.assertEqual(icmp.code, 0)
 961
 962         # ICMP is rate limited. pause a bit
 963         self.sleep(1)
 964
 965         #
 966         # A route via IP NULL that will reply with ICMP prohibited
 967         #
 968         ip_prohibit = VppIpRoute(self, "2001::1", 128, [],
 969                                  is_prohibit=1, is_ip6=1)
 970         ip_prohibit.add_vpp_config()
 971
 972         self.pg0.add_stream(p)
 973         self.pg_enable_capture(self.pg_interfaces)
 974         self.pg_start()
 975
 976         rx = self.pg0.get_capture(1)
 977         rx = rx[0]
 978         icmp = rx[ICMPv6DestUnreach]
 979
 980         # 1 = "Communication with destination administratively prohibited"
 981         self.assertEqual(icmp.code, 1)
 982
 983
 984 if __name__ == '__main__':
 985     unittest.main(testRunner=VppTestRunner)