6 from framework import VppTestCase, VppTestRunner
7 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
9 from scapy.packet import Raw
10 from scapy.layers.l2 import Ether, Dot1Q
11 from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, ICMPv6ND_RA, \
12 ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation
14 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
16 from scapy.utils import inet_pton, inet_ntop
20 euid = in6_mactoifaceid(mac)
21 addr = "fe80::" + euid
25 class TestIPv6(VppTestCase):
26 """ IPv6 Test Case """
30 super(TestIPv6, cls).setUpClass()
34 Perform test setup before test case.
37 - create 3 pg interfaces
38 - untagged pg0 interface
39 - Dot1Q subinterface on pg1
40 - Dot1AD subinterface on pg2
42 - put it into UP state
44 - resolve neighbor address using NDP
45 - configure 200 fib entries
47 :ivar list interfaces: pg interfaces and subinterfaces.
48 :ivar dict flows: IPv4 packet flows in test.
49 :ivar list pg_if_packet_sizes: packet sizes in test.
51 *TODO:* Create AD sub interface
53 super(TestIPv6, self).setUp()
55 # create 3 pg interfaces
56 self.create_pg_interfaces(range(3))
58 # create 2 subinterfaces for p1 and pg2
59 self.sub_interfaces = [
60 VppDot1QSubint(self, self.pg1, 100),
61 VppDot1QSubint(self, self.pg2, 200)
62 # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
65 # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
67 self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
68 self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
69 self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
72 self.pg_if_packet_sizes = [64, 512, 1518, 9018]
73 self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
75 self.interfaces = list(self.pg_interfaces)
76 self.interfaces.extend(self.sub_interfaces)
78 # setup all interfaces
79 for i in self.interfaces:
84 # config 2M FIB entries
85 self.config_fib_entries(200)
88 """Run standard test teardown and log ``show ip6 neighbors``."""
89 for i in self.sub_interfaces:
95 super(TestIPv6, self).tearDown()
97 self.logger.info(self.vapi.cli("show ip6 neighbors"))
98 # info(self.vapi.cli("show ip6 fib")) # many entries
100 def config_fib_entries(self, count):
101 """For each interface add to the FIB table *count* routes to
102 "fd02::1/128" destination with interface's local address as next-hop
105 :param int count: Number of FIB entries.
107 - *TODO:* check if the next-hop address shouldn't be remote address
108 instead of local address.
110 n_int = len(self.interfaces)
113 dest_addr = socket.inet_pton(socket.AF_INET6, "fd02::1")
115 for i in self.interfaces:
116 next_hop_address = i.local_ip6n
117 for j in range(count / n_int):
118 self.vapi.ip_add_del_route(
119 dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
121 if counter / count * 100 > percent:
122 self.logger.info("Configure %d FIB entries .. %d%% done" %
126 def create_stream(self, src_if, packet_sizes):
127 """Create input packet stream for defined interface.
129 :param VppInterface src_if: Interface to create packet stream for.
130 :param list packet_sizes: Required packet sizes.
133 for i in range(0, 257):
134 dst_if = self.flows[src_if][i % 2]
135 info = self.create_packet_info(src_if, dst_if)
136 payload = self.info_to_payload(info)
137 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
138 IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
139 UDP(sport=1234, dport=1234) /
142 if isinstance(src_if, VppSubInterface):
143 p = src_if.add_dot1_layer(p)
144 size = packet_sizes[(i // 2) % len(packet_sizes)]
145 self.extend_packet(p, size)
149 def verify_capture(self, dst_if, capture):
150 """Verify captured input packet stream for defined interface.
152 :param VppInterface dst_if: Interface to verify captured packet stream
154 :param list capture: Captured packet stream.
156 self.logger.info("Verifying capture on interface %s" % dst_if.name)
158 for i in self.interfaces:
159 last_info[i.sw_if_index] = None
161 dst_sw_if_index = dst_if.sw_if_index
162 if hasattr(dst_if, 'parent'):
164 for packet in capture:
166 # Check VLAN tags and Ethernet header
167 packet = dst_if.remove_dot1_layer(packet)
168 self.assertTrue(Dot1Q not in packet)
172 payload_info = self.payload_to_info(str(packet[Raw]))
173 packet_index = payload_info.index
174 self.assertEqual(payload_info.dst, dst_sw_if_index)
175 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
176 (dst_if.name, payload_info.src, packet_index))
177 next_info = self.get_next_packet_info_for_interface2(
178 payload_info.src, dst_sw_if_index,
179 last_info[payload_info.src])
180 last_info[payload_info.src] = next_info
181 self.assertTrue(next_info is not None)
182 self.assertEqual(packet_index, next_info.index)
183 saved_packet = next_info.data
184 # Check standard fields
185 self.assertEqual(ip.src, saved_packet[IPv6].src)
186 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
187 self.assertEqual(udp.sport, saved_packet[UDP].sport)
188 self.assertEqual(udp.dport, saved_packet[UDP].dport)
190 self.logger.error(ppp("Unexpected or invalid packet:", packet))
192 for i in self.interfaces:
193 remaining_packet = self.get_next_packet_info_for_interface2(
194 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
195 self.assertTrue(remaining_packet is None,
196 "Interface %s: Packet expected from interface %s "
197 "didn't arrive" % (dst_if.name, i.name))
203 - Create IPv6 stream for pg0 interface
204 - Create IPv6 tagged streams for pg1's and pg2's subinterface.
205 - Send and verify received packets on each interface.
208 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
209 self.pg0.add_stream(pkts)
211 for i in self.sub_interfaces:
212 pkts = self.create_stream(i, self.sub_if_packet_sizes)
213 i.parent.add_stream(pkts)
215 self.pg_enable_capture(self.pg_interfaces)
218 pkts = self.pg0.get_capture()
219 self.verify_capture(self.pg0, pkts)
221 for i in self.sub_interfaces:
222 pkts = i.parent.get_capture()
223 self.verify_capture(i, pkts)
225 def send_and_assert_no_replies(self, intf, pkts, remark):
226 intf.add_stream(pkts)
227 self.pg_enable_capture(self.pg_interfaces)
229 intf.assert_nothing_captured(remark=remark)
232 """ IPv6 Neighbour Soliciatation Exceptions
235 - Send an NS Sourced from an address not covered by the link sub-net
236 - Send an NS to an mcast address the router has not joined
237 - Send NS for a target address the router does not onn.
241 # An NS from a non link source address
243 nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
244 d = inet_ntop(socket.AF_INET6, nsma)
246 p = (Ether(dst=in6_getnsmac(nsma)) /
247 IPv6(dst=d, src="2002::2") /
248 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
249 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
252 self.send_and_assert_no_replies(self.pg0, pkts,
253 "No response to NS source by address not on sub-net")
256 # An NS for sent to a solicited mcast group the router is not a member of
260 nsma = in6_getnsma(inet_pton(socket.AF_INET6, "fd::ffff"))
261 d = inet_ntop(socket.AF_INET6, nsma)
263 p = (Ether(dst=in6_getnsmac(nsma)) /
264 IPv6(dst=d, src=self.pg0.remote_ip6) /
265 ICMPv6ND_NS(tgt=self.pg0.local_ip6) /
266 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
269 self.send_and_assert_no_replies(self.pg0, pkts,
270 "No response to NS sent to unjoined mcast address")
273 # An NS whose target address is one the router does not own
275 nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
276 d = inet_ntop(socket.AF_INET6, nsma)
278 p = (Ether(dst=in6_getnsmac(nsma)) /
279 IPv6(dst=d, src=self.pg0.remote_ip6) /
280 ICMPv6ND_NS(tgt="fd::ffff") /
281 ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
284 self.send_and_assert_no_replies(self.pg0, pkts,
285 "No response to NS for unknown target")
287 def send_and_expect_ra(self, intf, pkts, remark, src_ip=None):
289 src_ip = intf.remote_ip6
290 intf.add_stream(pkts)
291 self.pg0.add_stream(pkts)
292 self.pg_enable_capture(self.pg_interfaces)
294 rx = intf.get_capture(1)
296 self.assertEqual(len(rx), 1)
299 # the rx'd RA should be addressed to the sender's source
300 self.assertTrue(rx.haslayer(ICMPv6ND_RA))
301 self.assertEqual(in6_ptop(rx[IPv6].dst),
304 # and come from the router's link local
305 self.assertTrue(in6_islladdr(rx[IPv6].src))
306 self.assertEqual(in6_ptop(rx[IPv6].src),
307 in6_ptop(mk_ll_addr(intf.local_mac)))
310 """ IPv6 Router Soliciatation Exceptions
316 # Before we begin change the IPv6 RA responses to use the unicast address
317 # that way we will not confuse them with the periodic Ras which go to the Mcast
320 self.pg0.ip6_ra_config(send_unicast=1)
323 # An RS from a link source address
324 # - expect an RA in return
326 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
327 IPv6(dst=self.pg0.local_ip6, src=self.pg0.remote_ip6) /
330 self.send_and_expect_ra(self.pg0, pkts, "Genuine RS")
333 # For the next RS sent the RA should be rate limited
335 self.send_and_assert_no_replies(self.pg0, pkts, "RA rate limited")
338 # When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
339 # so we need to do this before each test below so as not to drop packets for
340 # rate limiting reasons. Test this works here.
342 self.pg0.ip6_ra_config(send_unicast=1)
343 self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
346 # An RS sent from a non-link local source
348 self.pg0.ip6_ra_config(send_unicast=1)
349 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
350 IPv6(dst=self.pg0.local_ip6, src="2002::ffff") /
353 self.send_and_assert_no_replies(self.pg0, pkts,
354 "RS from non-link source")
357 # Source an RS from a link local address
359 self.pg0.ip6_ra_config(send_unicast=1)
360 ll = mk_ll_addr(self.pg0.remote_mac)
361 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
362 IPv6(dst=self.pg0.local_ip6, src=ll) /
365 self.send_and_expect_ra(
366 self.pg0, pkts, "RS sourced from link-local", src_ip=ll)
369 # Source from the unspecified address ::. This happens when the RS is sent before
370 # the host has a configured address/sub-net, i.e. auto-config.
371 # Since the sender has no IP address, the reply comes back mcast - so the
372 # capture needs to not filter this.
373 # If we happen to pick up the periodic RA at this point then so be it, it's not
376 self.pg0.ip6_ra_config(send_unicast=1)
377 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
378 IPv6(dst=self.pg0.local_ip6, src="::") /
382 self.pg0.add_stream(pkts)
383 self.pg0.add_stream(pkts)
384 self.pg_enable_capture(self.pg_interfaces)
386 capture = self.pg0.get_capture(1, filter_out_fn=None)
389 if (rx.haslayer(ICMPv6ND_RA)):
390 # and come from the router's link local
391 self.assertTrue(in6_islladdr(rx[IPv6].src))
392 self.assertEqual(in6_ptop(rx[IPv6].src),
393 in6_ptop(mk_ll_addr(self.pg0.local_mac)))
394 # sent to the all hosts mcast
395 self.assertEqual(in6_ptop(rx[IPv6].dst), "ff02::1")
398 self.assertTrue(found)
400 @unittest.skip("Unsupported")
402 """ IPv6 Multicast Router Soliciatation Exceptions
408 # An RS from a link source address
409 # - expect an RA in return
411 nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
412 d = inet_ntop(socket.AF_INET6, nsma)
414 p = (Ether(dst=getmacbyip6("ff02::2")) /
415 IPv6(dst=d, src=self.pg0.remote_ip6) /
416 ICMPv6MRD_Solicitation())
419 self.pg0.add_stream(pkts)
420 self.pg_enable_capture(self.pg_interfaces)
422 self.pg0.assert_nothing_captured(
423 remark="No response to NS source by address not on sub-net")
426 # An RS from a non link source address
428 nsma = in6_getnsma(inet_pton(socket.AF_INET6, self.pg0.local_ip6))
429 d = inet_ntop(socket.AF_INET6, nsma)
431 p = (Ether(dst=getmacbyip6("ff02::2")) /
432 IPv6(dst=d, src="2002::2") /
433 ICMPv6MRD_Solicitation())
436 self.send_and_assert_no_replies(self.pg0, pkts,
438 self.pg0.add_stream(pkts)
439 self.pg_enable_capture(self.pg_interfaces)
441 self.pg0.assert_nothing_captured(
442 remark="No response to NS source by address not on sub-net")
445 if __name__ == '__main__':
446 unittest.main(testRunner=VppTestRunner)