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