6 from framework import VppTestCase, VppTestRunner
7 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
8 from vpp_pg_interface import is_ipv6_misc
10 from scapy.packet import Raw
11 from scapy.layers.l2 import Ether, Dot1Q
12 from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, \
13 ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation, \
14 ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo
16 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
17 in6_mactoifaceid, in6_ismaddr
18 from scapy.utils import inet_pton, inet_ntop
22 euid = in6_mactoifaceid(mac)
23 addr = "fe80::" + euid
27 class TestIPv6(VppTestCase):
28 """ IPv6 Test Case """
32 super(TestIPv6, cls).setUpClass()
36 Perform test setup before test case.
39 - create 3 pg interfaces
40 - untagged pg0 interface
41 - Dot1Q subinterface on pg1
42 - Dot1AD subinterface on pg2
44 - put it into UP state
46 - resolve neighbor address using NDP
47 - configure 200 fib entries
49 :ivar list interfaces: pg interfaces and subinterfaces.
50 :ivar dict flows: IPv4 packet flows in test.
51 :ivar list pg_if_packet_sizes: packet sizes in test.
53 *TODO:* Create AD sub interface
55 super(TestIPv6, self).setUp()
57 # create 3 pg interfaces
58 self.create_pg_interfaces(range(3))
60 # create 2 subinterfaces for p1 and pg2
61 self.sub_interfaces = [
62 VppDot1QSubint(self, self.pg1, 100),
63 VppDot1QSubint(self, self.pg2, 200)
64 # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
67 # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
69 self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
70 self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
71 self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
74 self.pg_if_packet_sizes = [64, 512, 1518, 9018]
75 self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
77 self.interfaces = list(self.pg_interfaces)
78 self.interfaces.extend(self.sub_interfaces)
80 # setup all interfaces
81 for i in self.interfaces:
86 # config 2M FIB entries
87 self.config_fib_entries(200)
90 """Run standard test teardown and log ``show ip6 neighbors``."""
91 for i in self.sub_interfaces:
97 super(TestIPv6, self).tearDown()
99 self.logger.info(self.vapi.cli("show ip6 neighbors"))
100 # info(self.vapi.cli("show ip6 fib")) # many entries
102 def config_fib_entries(self, count):
103 """For each interface add to the FIB table *count* routes to
104 "fd02::1/128" destination with interface's local address as next-hop
107 :param int count: Number of FIB entries.
109 - *TODO:* check if the next-hop address shouldn't be remote address
110 instead of local address.
112 n_int = len(self.interfaces)
115 dest_addr = socket.inet_pton(socket.AF_INET6, "fd02::1")
117 for i in self.interfaces:
118 next_hop_address = i.local_ip6n
119 for j in range(count / n_int):
120 self.vapi.ip_add_del_route(
121 dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
123 if counter / count * 100 > percent:
124 self.logger.info("Configure %d FIB entries .. %d%% done" %
128 def create_stream(self, src_if, packet_sizes):
129 """Create input packet stream for defined interface.
131 :param VppInterface src_if: Interface to create packet stream for.
132 :param list packet_sizes: Required packet sizes.
135 for i in range(0, 257):
136 dst_if = self.flows[src_if][i % 2]
137 info = self.create_packet_info(src_if, dst_if)
138 payload = self.info_to_payload(info)
139 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
140 IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
141 UDP(sport=1234, dport=1234) /
144 if isinstance(src_if, VppSubInterface):
145 p = src_if.add_dot1_layer(p)
146 size = packet_sizes[(i // 2) % len(packet_sizes)]
147 self.extend_packet(p, size)
151 def verify_capture(self, dst_if, capture):
152 """Verify captured input packet stream for defined interface.
154 :param VppInterface dst_if: Interface to verify captured packet stream
156 :param list capture: Captured packet stream.
158 self.logger.info("Verifying capture on interface %s" % dst_if.name)
160 for i in self.interfaces:
161 last_info[i.sw_if_index] = None
163 dst_sw_if_index = dst_if.sw_if_index
164 if hasattr(dst_if, 'parent'):
166 for packet in capture:
168 # Check VLAN tags and Ethernet header
169 packet = dst_if.remove_dot1_layer(packet)
170 self.assertTrue(Dot1Q not in packet)
174 payload_info = self.payload_to_info(str(packet[Raw]))
175 packet_index = payload_info.index
176 self.assertEqual(payload_info.dst, dst_sw_if_index)
178 "Got packet on port %s: src=%u (id=%u)" %
179 (dst_if.name, payload_info.src, packet_index))
180 next_info = self.get_next_packet_info_for_interface2(
181 payload_info.src, dst_sw_if_index,
182 last_info[payload_info.src])
183 last_info[payload_info.src] = next_info
184 self.assertTrue(next_info is not None)
185 self.assertEqual(packet_index, next_info.index)
186 saved_packet = next_info.data
187 # Check standard fields
188 self.assertEqual(ip.src, saved_packet[IPv6].src)
189 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
190 self.assertEqual(udp.sport, saved_packet[UDP].sport)
191 self.assertEqual(udp.dport, saved_packet[UDP].dport)
193 self.logger.error(ppp("Unexpected or invalid packet:", packet))
195 for i in self.interfaces:
196 remaining_packet = self.get_next_packet_info_for_interface2(
197 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
198 self.assertTrue(remaining_packet is None,
199 "Interface %s: Packet expected from interface %s "
200 "didn't arrive" % (dst_if.name, i.name))
206 - Create IPv6 stream for pg0 interface
207 - Create IPv6 tagged streams for pg1's and pg2's subinterface.
208 - Send and verify received packets on each interface.
211 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
212 self.pg0.add_stream(pkts)
214 for i in self.sub_interfaces:
215 pkts = self.create_stream(i, self.sub_if_packet_sizes)
216 i.parent.add_stream(pkts)
218 self.pg_enable_capture(self.pg_interfaces)
221 pkts = self.pg0.get_capture()
222 self.verify_capture(self.pg0, pkts)
224 for i in self.sub_interfaces:
225 pkts = i.parent.get_capture()
226 self.verify_capture(i, pkts)
228 def send_and_assert_no_replies(self, intf, pkts, remark):
229 intf.add_stream(pkts)
230 self.pg_enable_capture(self.pg_interfaces)
232 intf.assert_nothing_captured(remark=remark)
235 """ IPv6 Neighbour Solicitation Exceptions
238 - Send an NS Sourced from an address not covered by the link sub-net
239 - Send an NS to an mcast address the router has not joined
240 - Send NS for a target address the router does not onn.
244 # An NS from a non link source address
246 nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
247 d = inet_ntop(socket.AF_INET6, nsma)
249 p = (Ether(dst=in6_getnsmac(nsma)) /
250 IPv6(dst=d, src="2002::2") /
251 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
252 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
255 self.send_and_assert_no_replies(
257 "No response to NS source by address not on sub-net")
260 # An NS for sent to a solicited mcast group the router is
261 # not a member of FAILS
264 nsma = in6_getnsma(inet_pton(socket.AF_INET6, "fd::ffff"))
265 d = inet_ntop(socket.AF_INET6, nsma)
267 p = (Ether(dst=in6_getnsmac(nsma)) /
268 IPv6(dst=d, src=self.pg0.remote_ip6) /
269 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
270 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
273 self.send_and_assert_no_replies(
275 "No response to NS sent to unjoined mcast address")
278 # An NS whose target address is one the router does not own
280 nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
281 d = inet_ntop(socket.AF_INET6, nsma)
283 p = (Ether(dst=in6_getnsmac(nsma)) /
284 IPv6(dst=d, src=self.pg0.remote_ip6) /
285 ICMPv6ND_NS(tgt="fd::ffff") /
286 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
289 self.send_and_assert_no_replies(self.pg0, pkts,
290 "No response to NS for unknown target")
292 def validate_ra(self, intf, rx, dst_ip=None, mtu=9000, pi_opt=None):
294 dst_ip = intf.remote_ip6
296 # unicasted packets must come to the unicast mac
297 self.assertEqual(rx[Ether].dst, intf.remote_mac)
299 # and from the router's MAC
300 self.assertEqual(rx[Ether].src, intf.local_mac)
302 # the rx'd RA should be addressed to the sender's source
303 self.assertTrue(rx.haslayer(ICMPv6ND_RA))
304 self.assertEqual(in6_ptop(rx[IPv6].dst),
307 # and come from the router's link local
308 self.assertTrue(in6_islladdr(rx[IPv6].src))
309 self.assertEqual(in6_ptop(rx[IPv6].src),
310 in6_ptop(mk_ll_addr(intf.local_mac)))
312 # it should contain the links MTU
314 self.assertEqual(ra[ICMPv6NDOptMTU].mtu, mtu)
316 # it should contain the source's link layer address option
317 sll = ra[ICMPv6NDOptSrcLLAddr]
318 self.assertEqual(sll.lladdr, intf.local_mac)
321 # the RA should not contain prefix information
322 self.assertFalse(ra.haslayer(ICMPv6NDOptPrefixInfo))
324 raos = rx.getlayer(ICMPv6NDOptPrefixInfo, 1)
326 # the options are nested in the scapy packet in way that i cannot
327 # decipher how to decode. this 1st layer of option always returns
328 # nested classes, so a direct obj1=obj2 comparison always fails.
329 # however, the getlayer(.., 2) does give one instnace.
330 # so we cheat here and construct a new opt instnace for comparison
331 rd = ICMPv6NDOptPrefixInfo(prefixlen=raos.prefixlen,
335 if type(pi_opt) is list:
336 for ii in range(len(pi_opt)):
337 self.assertEqual(pi_opt[ii], rd)
338 rd = rx.getlayer(ICMPv6NDOptPrefixInfo, ii+2)
340 self.assertEqual(pi_opt, raos)
342 def send_and_expect_ra(self, intf, pkts, remark, dst_ip=None,
343 filter_out_fn=is_ipv6_misc,
345 intf.add_stream(pkts)
346 self.pg_enable_capture(self.pg_interfaces)
348 rx = intf.get_capture(1, filter_out_fn=filter_out_fn)
350 self.assertEqual(len(rx), 1)
352 self.validate_ra(intf, rx, dst_ip, pi_opt=opt)
355 """ IPv6 Router Solicitation Exceptions
361 # Before we begin change the IPv6 RA responses to use the unicast
362 # address - that way we will not confuse them with the periodic
363 # RAs which go to the mcast address
364 # Sit and wait for the first periodic RA.
368 self.pg0.ip6_ra_config(send_unicast=1)
371 # An RS from a link source address
372 # - expect an RA in return
374 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
375 IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
378 self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
381 # For the next RS sent the RA should be rate limited
383 self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
386 # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
387 # so we need to do this before each test below so as not to drop
388 # packets for rate limiting reasons. Test this works here.
390 self.pg0.ip6_ra_config(send_unicast=1)
391 self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
394 # An RS sent from a non-link local source
396 self.pg0.ip6_ra_config(send_unicast=1)
397 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
398 IPv6(dst=self.pg0.local_ip6, src="2002::ffff") /
401 self.send_and_assert_no_replies(self.pg0, pkts,
402 "RS from non-link source")
405 # Source an RS from a link local address
407 self.pg0.ip6_ra_config(send_unicast=1)
408 ll = mk_ll_addr(self.pg0.remote_mac)
409 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
410 IPv6(dst=self.pg0.local_ip6, src=ll) /
413 self.send_and_expect_ra(self.pg0, pkts,
414 "RS sourced from link-local",
418 # Send the RS multicast
420 self.pg0.ip6_ra_config(send_unicast=1)
421 dmac = in6_getnsmac(inet_pton(socket.AF_INET6, "ff02::2"))
422 ll = mk_ll_addr(self.pg0.remote_mac)
423 p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
424 IPv6(dst="ff02::2", src=ll) /
427 self.send_and_expect_ra(self.pg0, pkts,
428 "RS sourced from link-local",
432 # Source from the unspecified address ::. This happens when the RS
433 # is sent before the host has a configured address/sub-net,
434 # i.e. auto-config. Since the sender has no IP address, the reply
435 # comes back mcast - so the capture needs to not filter this.
436 # If we happen to pick up the periodic RA at this point then so be it,
439 self.pg0.ip6_ra_config(send_unicast=1, suppress=1)
440 p = (Ether(dst=dmac, src=self.pg0.remote_mac) /
441 IPv6(dst="ff02::2", src="::") /
444 self.send_and_expect_ra(self.pg0, pkts,
445 "RS sourced from unspecified",
450 # Configure The RA to announce the links prefix
452 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
453 self.pg0.local_ip6_prefix_len)
456 # RAs should now contain the prefix information option
458 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
459 prefix=self.pg0.local_ip6,
463 self.pg0.ip6_ra_config(send_unicast=1)
464 ll = mk_ll_addr(self.pg0.remote_mac)
465 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
466 IPv6(dst=self.pg0.local_ip6, src=ll) /
468 self.send_and_expect_ra(self.pg0, p,
469 "RA with prefix-info",
474 # Change the prefix info to not off-link
477 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
478 self.pg0.local_ip6_prefix_len,
481 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
482 prefix=self.pg0.local_ip6,
486 self.pg0.ip6_ra_config(send_unicast=1)
487 self.send_and_expect_ra(self.pg0, p,
488 "RA with Prefix info with L-flag=0",
493 # Change the prefix info to not off-link, no-autoconfig
494 # L and A flag are clear in the advert
496 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
497 self.pg0.local_ip6_prefix_len,
501 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
502 prefix=self.pg0.local_ip6,
506 self.pg0.ip6_ra_config(send_unicast=1)
507 self.send_and_expect_ra(self.pg0, p,
508 "RA with Prefix info with A & L-flag=0",
513 # Change the flag settings back to the defaults
514 # L and A flag are set in the advert
516 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
517 self.pg0.local_ip6_prefix_len)
519 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
520 prefix=self.pg0.local_ip6,
524 self.pg0.ip6_ra_config(send_unicast=1)
525 self.send_and_expect_ra(self.pg0, p,
526 "RA with Prefix info",
531 # Change the prefix info to not off-link, no-autoconfig
532 # L and A flag are clear in the advert
534 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
535 self.pg0.local_ip6_prefix_len,
539 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
540 prefix=self.pg0.local_ip6,
544 self.pg0.ip6_ra_config(send_unicast=1)
545 self.send_and_expect_ra(self.pg0, p,
546 "RA with Prefix info with A & L-flag=0",
551 # Use the reset to defults option to revert to defaults
552 # L and A flag are clear in the advert
554 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
555 self.pg0.local_ip6_prefix_len,
558 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
559 prefix=self.pg0.local_ip6,
563 self.pg0.ip6_ra_config(send_unicast=1)
564 self.send_and_expect_ra(self.pg0, p,
565 "RA with Prefix reverted to defaults",
570 # Advertise Another prefix. With no L-flag/A-flag
572 self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
573 self.pg1.local_ip6_prefix_len,
577 opt = [ICMPv6NDOptPrefixInfo(prefixlen=self.pg0.local_ip6_prefix_len,
578 prefix=self.pg0.local_ip6,
581 ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
582 prefix=self.pg1.local_ip6,
586 self.pg0.ip6_ra_config(send_unicast=1)
587 ll = mk_ll_addr(self.pg0.remote_mac)
588 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
589 IPv6(dst=self.pg0.local_ip6, src=ll) /
591 self.send_and_expect_ra(self.pg0, p,
592 "RA with multiple Prefix infos",
597 # Remove the first refix-info - expect the second is still in the
600 self.pg0.ip6_ra_prefix(self.pg0.local_ip6n,
601 self.pg0.local_ip6_prefix_len,
604 opt = ICMPv6NDOptPrefixInfo(prefixlen=self.pg1.local_ip6_prefix_len,
605 prefix=self.pg1.local_ip6,
609 self.pg0.ip6_ra_config(send_unicast=1)
610 self.send_and_expect_ra(self.pg0, p,
611 "RA with Prefix reverted to defaults",
616 # Remove the second prefix-info - expect no prefix-info i nthe adverts
618 self.pg0.ip6_ra_prefix(self.pg1.local_ip6n,
619 self.pg1.local_ip6_prefix_len,
622 self.pg0.ip6_ra_config(send_unicast=1)
623 self.send_and_expect_ra(self.pg0, p,
624 "RA with Prefix reverted to defaults",
628 # Reset the periodic advertisements back to default values
630 self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
632 if __name__ == '__main__':
633 unittest.main(testRunner=VppTestRunner)