5 from socket import AF_INET6
7 from framework import VppTestCase, VppTestRunner
8 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto, VppIpTable
9 from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \
10 SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes
13 from scapy.packet import Raw
14 from scapy.layers.l2 import Ether, Dot1Q
15 from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting
16 from scapy.layers.inet import IP, UDP
18 from scapy.utils import inet_pton, inet_ntop
23 class TestSRv6(VppTestCase):
24 """ SRv6 Static Proxy plugin Test Case """
28 super(TestSRv6, self).setUpClass()
31 def tearDownClass(cls):
32 super(TestSRv6, cls).tearDownClass()
35 """ Perform test setup before each test case.
37 super(TestSRv6, self).setUp()
39 # packet sizes, inclusive L2 overhead
40 self.pg_packet_sizes = [64, 512, 1518, 9018]
43 self.reset_packet_infos()
46 """ Clean up test setup after each test case.
48 self.teardown_interfaces()
50 super(TestSRv6, self).tearDown()
52 def configure_interface(self,
54 ipv6=False, ipv4=False,
55 ipv6_table_id=0, ipv4_table_id=0):
56 """ Configure interface.
57 :param ipv6: configure IPv6 on interface
58 :param ipv4: configure IPv4 on interface
59 :param ipv6_table_id: FIB table_id for IPv6
60 :param ipv4_table_id: FIB table_id for IPv4
62 self.logger.debug("Configuring interface %s" % (interface.name))
64 self.logger.debug("Configuring IPv6")
65 interface.set_table_ip6(ipv6_table_id)
66 interface.config_ip6()
67 interface.resolve_ndp(timeout=5)
69 self.logger.debug("Configuring IPv4")
70 interface.set_table_ip4(ipv4_table_id)
71 interface.config_ip4()
72 interface.resolve_arp()
75 def setup_interfaces(self, ipv6=[], ipv4=[],
76 ipv6_table_id=[], ipv4_table_id=[]):
77 """ Create and configure interfaces.
79 :param ipv6: list of interface IPv6 capabilities
80 :param ipv4: list of interface IPv4 capabilities
81 :param ipv6_table_id: list of intf IPv6 FIB table_ids
82 :param ipv4_table_id: list of intf IPv4 FIB table_ids
83 :returns: List of created interfaces.
85 # how many interfaces?
90 self.logger.debug("Creating and configuring %d interfaces" % (count))
92 # fill up ipv6 and ipv4 lists if needed
93 # not enabled (False) is the default
95 ipv6 += (count - len(ipv6)) * [False]
97 ipv4 += (count - len(ipv4)) * [False]
99 # fill up table_id lists if needed
100 # table_id 0 (global) is the default
101 if len(ipv6_table_id) < count:
102 ipv6_table_id += (count - len(ipv6_table_id)) * [0]
103 if len(ipv4_table_id) < count:
104 ipv4_table_id += (count - len(ipv4_table_id)) * [0]
106 # create 'count' pg interfaces
107 self.create_pg_interfaces(range(count))
109 # setup all interfaces
110 for i in range(count):
111 intf = self.pg_interfaces[i]
112 self.configure_interface(intf,
114 ipv6_table_id[i], ipv4_table_id[i])
117 self.logger.debug(self.vapi.cli("show ip6 neighbors"))
119 self.logger.debug(self.vapi.cli("show ip arp"))
120 self.logger.debug(self.vapi.cli("show interface"))
121 self.logger.debug(self.vapi.cli("show hardware"))
123 return self.pg_interfaces
125 def teardown_interfaces(self):
126 """ Unconfigure and bring down interface.
128 self.logger.debug("Tearing down interfaces")
129 # tear down all interfaces
130 # AFAIK they cannot be deleted
131 for i in self.pg_interfaces:
132 self.logger.debug("Tear down interface %s" % (i.name))
138 def test_SRv6_End_AS_IPv6_noSRH(self):
139 """ Test SRv6 End.AS behavior with IPv6 traffic and no SRH rewrite.
141 self.run_SRv6_End_AS_IPv6(
142 sid_list=['a1::', 'a2::a6', 'a3::'],
144 rewrite_src_addr='a2::')
146 def test_SRv6_End_AS_IPv6_SRH(self):
147 """ Test SRv6 End.AS behavior with IPv6 traffic and SRH rewrite.
149 self.run_SRv6_End_AS_IPv6(
150 sid_list=['a1::a6', 'a2::', 'a3::'],
152 rewrite_src_addr='a1::')
154 def test_SRv6_End_AS_IPv4_noSRH(self):
155 """ Test SRv6 End.AS behavior with IPv4 traffic and no SRH rewrite.
157 self.run_SRv6_End_AS_IPv4(
158 sid_list=['a1::', 'a2::a6', 'a3::'],
160 rewrite_src_addr='a2::')
162 def test_SRv6_End_AS_IPv4_SRH(self):
163 """ Test SRv6 End.AS behavior with IPv4 traffic and SRH rewrite.
165 self.run_SRv6_End_AS_IPv4(
166 sid_list=['a1::a6', 'a2::', 'a3::'],
168 rewrite_src_addr='a1::')
170 def test_SRv6_End_AS_L2_noSRH(self):
171 """ Test SRv6 End.AS behavior with L2 traffic and no SRH rewrite.
173 self.run_SRv6_End_AS_L2(
174 sid_list=['a1::', 'a2::a6', 'a3::'],
176 rewrite_src_addr='a2::')
178 def test_SRv6_End_AS_L2_SRH(self):
179 """ Test SRv6 End.AS behavior with L2 traffic and SRH rewrite.
181 self.run_SRv6_End_AS_L2(
182 sid_list=['a1::a6', 'a2::', 'a3::'],
184 rewrite_src_addr='a1::')
186 def run_SRv6_End_AS_L2(self, sid_list, test_sid_index, rewrite_src_addr):
187 """ Run SRv6 End.AS test with L2 traffic.
189 self.rewrite_src_addr = rewrite_src_addr
190 self.rewrite_sid_list = sid_list[test_sid_index + 1::]
192 # send traffic to one destination interface
193 # source and destination interfaces are IPv6 only
194 self.setup_interfaces(ipv6=[True, False])
196 # configure route to next segment
197 route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
198 [VppRoutePath(self.pg0.remote_ip6,
199 self.pg0.sw_if_index,
200 proto=DpoProto.DPO_PROTO_IP6)],
202 route.add_vpp_config()
204 # configure SRv6 localSID behavior
205 cli_str = "sr localsid address " + sid_list[test_sid_index] \
206 + " behavior end.as" \
207 + " oif " + self.pg1.name \
208 + " iif " + self.pg1.name \
209 + " src " + self.rewrite_src_addr
210 for s in self.rewrite_sid_list:
211 cli_str += " next " + s
212 self.vapi.cli(cli_str)
215 self.logger.debug(self.vapi.cli("show sr localsid"))
217 # send one packet per packet size
218 count = len(self.pg_packet_sizes)
220 # prepare L2 in SRv6 headers
221 packet_header1 = self.create_packet_header_IPv6_SRH_L2(
222 sidlist=sid_list[::-1],
223 segleft=len(sid_list) - test_sid_index - 1,
226 # generate packets (pg0->pg1)
227 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
228 self.pg_packet_sizes, count)
230 # send packets and verify received packets
231 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
232 self.compare_rx_tx_packet_End_AS_L2_out)
234 # log the localsid counters
235 self.logger.info(self.vapi.cli("show sr localsid"))
237 # prepare L2 header for returning packets
238 packet_header2 = self.create_packet_header_L2()
240 # generate returning packets (pg1->pg0)
241 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
242 self.pg_packet_sizes, count)
244 # send packets and verify received packets
245 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
246 self.compare_rx_tx_packet_End_AS_L2_in)
248 # log the localsid counters
249 self.logger.info(self.vapi.cli("show sr localsid"))
251 # remove SRv6 localSIDs
252 self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
255 self.teardown_interfaces()
257 def run_SRv6_End_AS_IPv6(self, sid_list, test_sid_index, rewrite_src_addr):
258 """ Run SRv6 End.AS test with IPv6 traffic.
260 self.rewrite_src_addr = rewrite_src_addr
261 self.rewrite_sid_list = sid_list[test_sid_index + 1::]
263 # send traffic to one destination interface
264 # source and destination interfaces are IPv6 only
265 self.setup_interfaces(ipv6=[True, True])
267 # configure route to next segment
268 route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
269 [VppRoutePath(self.pg0.remote_ip6,
270 self.pg0.sw_if_index,
271 proto=DpoProto.DPO_PROTO_IP6)],
273 route.add_vpp_config()
275 # configure SRv6 localSID behavior
276 cli_str = "sr localsid address " + sid_list[test_sid_index] \
277 + " behavior end.as" \
278 + " nh " + self.pg1.remote_ip6 \
279 + " oif " + self.pg1.name \
280 + " iif " + self.pg1.name \
281 + " src " + self.rewrite_src_addr
282 for s in self.rewrite_sid_list:
283 cli_str += " next " + s
284 self.vapi.cli(cli_str)
287 self.logger.debug(self.vapi.cli("show sr localsid"))
289 # send one packet per packet size
290 count = len(self.pg_packet_sizes)
292 # prepare IPv6 in SRv6 headers
293 packet_header1 = self.create_packet_header_IPv6_SRH_IPv6(
294 sidlist=sid_list[::-1],
295 segleft=len(sid_list) - test_sid_index - 1)
297 # generate packets (pg0->pg1)
298 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
299 self.pg_packet_sizes, count)
301 # send packets and verify received packets
302 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
303 self.compare_rx_tx_packet_End_AS_IPv6_out)
305 # log the localsid counters
306 self.logger.info(self.vapi.cli("show sr localsid"))
308 # prepare IPv6 header for returning packets
309 packet_header2 = self.create_packet_header_IPv6()
311 # generate returning packets (pg1->pg0)
312 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
313 self.pg_packet_sizes, count)
315 # send packets and verify received packets
316 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
317 self.compare_rx_tx_packet_End_AS_IPv6_in)
319 # log the localsid counters
320 self.logger.info(self.vapi.cli("show sr localsid"))
322 # remove SRv6 localSIDs
323 self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
326 self.teardown_interfaces()
328 def run_SRv6_End_AS_IPv4(self, sid_list, test_sid_index, rewrite_src_addr):
329 """ Run SRv6 End.AS test with IPv4 traffic.
331 self.rewrite_src_addr = rewrite_src_addr
332 self.rewrite_sid_list = sid_list[test_sid_index + 1::]
334 # send traffic to one destination interface
335 # source and destination interfaces are IPv6 only
336 self.setup_interfaces(ipv6=[True, False], ipv4=[True, True])
338 # configure route to next segment
339 route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
340 [VppRoutePath(self.pg0.remote_ip6,
341 self.pg0.sw_if_index,
342 proto=DpoProto.DPO_PROTO_IP6)],
344 route.add_vpp_config()
346 # configure SRv6 localSID behavior
347 cli_str = "sr localsid address " + sid_list[test_sid_index] \
348 + " behavior end.as" \
349 + " nh " + self.pg1.remote_ip4 \
350 + " oif " + self.pg1.name \
351 + " iif " + self.pg1.name \
352 + " src " + self.rewrite_src_addr
353 for s in self.rewrite_sid_list:
354 cli_str += " next " + s
355 self.vapi.cli(cli_str)
358 self.logger.debug(self.vapi.cli("show sr localsid"))
360 # send one packet per packet size
361 count = len(self.pg_packet_sizes)
363 # prepare IPv4 in SRv6 headers
364 packet_header1 = self.create_packet_header_IPv6_SRH_IPv4(
365 sidlist=sid_list[::-1],
366 segleft=len(sid_list) - test_sid_index - 1)
368 # generate packets (pg0->pg1)
369 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
370 self.pg_packet_sizes, count)
372 # send packets and verify received packets
373 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
374 self.compare_rx_tx_packet_End_AS_IPv4_out)
376 # log the localsid counters
377 self.logger.info(self.vapi.cli("show sr localsid"))
379 # prepare IPv6 header for returning packets
380 packet_header2 = self.create_packet_header_IPv4()
382 # generate returning packets (pg1->pg0)
383 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
384 self.pg_packet_sizes, count)
386 # send packets and verify received packets
387 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
388 self.compare_rx_tx_packet_End_AS_IPv4_in)
390 # log the localsid counters
391 self.logger.info(self.vapi.cli("show sr localsid"))
393 # remove SRv6 localSIDs
394 self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
397 self.teardown_interfaces()
399 def compare_rx_tx_packet_End_AS_IPv6_in(self, tx_pkt, rx_pkt):
400 """ Compare input and output packet after passing End.AS
402 :param tx_pkt: transmitted packet
403 :param rx_pkt: received packet
406 # get first (outer) IPv6 header of rx'ed packet
407 rx_ip = rx_pkt.getlayer(IPv6)
410 tx_ip = tx_pkt.getlayer(IPv6)
412 # expected segment-list (SRH order)
413 tx_seglist = self.rewrite_sid_list[::-1]
415 # received ip.src should be equal to SR Policy source
416 self.assertEqual(rx_ip.src, self.rewrite_src_addr)
417 # received ip.dst should be equal to expected sidlist[lastentry]
418 self.assertEqual(rx_ip.dst, tx_seglist[-1])
420 if len(tx_seglist) > 1:
421 # rx'ed packet should have SRH
422 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
424 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
425 # rx'ed seglist should be equal to expected seglist
426 self.assertEqual(rx_srh.addresses, tx_seglist)
427 # segleft should be equal to size expected seglist-1
428 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
429 # segleft should be equal to lastentry
430 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
432 payload = rx_srh.payload
434 # rx'ed packet should NOT have SRH
435 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
437 payload = rx_ip.payload
439 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
440 # except for the hop-limit field
441 # -> update tx'ed hlim to the expected hlim
442 tx_ip.hlim = tx_ip.hlim - 1
444 self.assertEqual(payload, tx_ip)
446 self.logger.debug("packet verification: SUCCESS")
448 def compare_rx_tx_packet_End_AS_IPv4_in(self, tx_pkt, rx_pkt):
449 """ Compare input and output packet after passing End.AS
451 :param tx_pkt: transmitted packet
452 :param rx_pkt: received packet
455 # get first (outer) IPv6 header of rx'ed packet
456 rx_ip = rx_pkt.getlayer(IPv6)
459 tx_ip = tx_pkt.getlayer(IP)
461 # expected segment-list (SRH order)
462 tx_seglist = self.rewrite_sid_list[::-1]
464 # received ip.src should be equal to SR Policy source
465 self.assertEqual(rx_ip.src, self.rewrite_src_addr)
466 # received ip.dst should be equal to expected sidlist[lastentry]
467 self.assertEqual(rx_ip.dst, tx_seglist[-1])
469 if len(tx_seglist) > 1:
470 # rx'ed packet should have SRH and IPv4 header
471 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
472 self.assertTrue(rx_ip.payload.haslayer(IP))
474 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
475 # rx'ed seglist should be equal to seglist
476 self.assertEqual(rx_srh.addresses, tx_seglist)
477 # segleft should be equal to size seglist-1
478 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
479 # segleft should be equal to lastentry
480 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
481 payload = rx_srh.payload
483 # rx'ed packet should NOT have SRH
484 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
486 payload = rx_ip.payload
488 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
489 # except for the ttl field and ip checksum
490 # -> adjust tx'ed ttl to expected ttl
491 tx_ip.ttl = tx_ip.ttl - 1
492 # -> set tx'ed ip checksum to None and let scapy recompute
494 # read back the pkt (with str()) to force computing these fields
495 # probably other ways to accomplish this are possible
496 tx_ip = IP(scapy.compat.raw(tx_ip))
498 self.assertEqual(payload, tx_ip)
500 self.logger.debug("packet verification: SUCCESS")
502 def compare_rx_tx_packet_End_AS_L2_in(self, tx_pkt, rx_pkt):
503 """ Compare input and output packet after passing End.AS
505 :param tx_pkt: transmitted packet
506 :param rx_pkt: received packet
509 # get first (outer) IPv6 header of rx'ed packet
510 rx_ip = rx_pkt.getlayer(IPv6)
513 tx_ether = tx_pkt.getlayer(Ether)
515 # expected segment-list (SRH order)
516 tx_seglist = self.rewrite_sid_list[::-1]
518 # received ip.src should be equal to SR Policy source
519 self.assertEqual(rx_ip.src, self.rewrite_src_addr)
520 # received ip.dst should be equal to expected sidlist[lastentry]
521 self.assertEqual(rx_ip.dst, tx_seglist[-1])
523 if len(tx_seglist) > 1:
524 # rx'ed packet should have SRH
525 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
527 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
528 # rx'ed seglist should be equal to seglist
529 self.assertEqual(rx_srh.addresses, tx_seglist)
530 # segleft should be equal to size seglist-1
531 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
532 # segleft should be equal to lastentry
533 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
534 # nh should be "No Next Header" (59)
535 self.assertEqual(rx_srh.nh, 59)
537 payload = rx_srh.payload
539 # rx'ed packet should NOT have SRH
540 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
542 payload = rx_ip.payload
544 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
545 self.assertEqual(Ether(scapy.compat.raw(payload)), tx_ether)
547 self.logger.debug("packet verification: SUCCESS")
549 def compare_rx_tx_packet_End_AS_IPv6_out(self, tx_pkt, rx_pkt):
550 """ Compare input and output packet after passing End.AS with IPv6
552 :param tx_pkt: transmitted packet
553 :param rx_pkt: received packet
556 # get first (outer) IPv6 header of rx'ed packet
557 rx_ip = rx_pkt.getlayer(IPv6)
559 tx_ip = tx_pkt.getlayer(IPv6)
560 tx_ip2 = tx_pkt.getlayer(IPv6, 2)
562 # verify if rx'ed packet has no SRH
563 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
565 # the whole rx_ip pkt should be equal to tx_ip2
566 # except for the hlim field
567 # -> adjust tx'ed hlim to expected hlim
568 tx_ip2.hlim = tx_ip2.hlim - 1
570 self.assertEqual(rx_ip, tx_ip2)
572 self.logger.debug("packet verification: SUCCESS")
574 def compare_rx_tx_packet_End_AS_IPv4_out(self, tx_pkt, rx_pkt):
575 """ Compare input and output packet after passing End.AS with IPv4
577 :param tx_pkt: transmitted packet
578 :param rx_pkt: received packet
581 # get IPv4 header of rx'ed packet
582 rx_ip = rx_pkt.getlayer(IP)
584 tx_ip = tx_pkt.getlayer(IPv6)
585 tx_ip2 = tx_pkt.getlayer(IP)
587 # verify if rx'ed packet has no SRH
588 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
590 # the whole rx_ip pkt should be equal to tx_ip2
591 # except for the ttl field and ip checksum
592 # -> adjust tx'ed ttl to expected ttl
593 tx_ip2.ttl = tx_ip2.ttl - 1
594 # -> set tx'ed ip checksum to None and let scapy recompute
596 # read back the pkt (with str()) to force computing these fields
597 # probably other ways to accomplish this are possible
598 tx_ip2 = IP(scapy.compat.raw(tx_ip2))
600 self.assertEqual(rx_ip, tx_ip2)
602 self.logger.debug("packet verification: SUCCESS")
604 def compare_rx_tx_packet_End_AS_L2_out(self, tx_pkt, rx_pkt):
605 """ Compare input and output packet after passing End.AS with L2
607 :param tx_pkt: transmitted packet
608 :param rx_pkt: received packet
611 # get IPv4 header of rx'ed packet
612 rx_eth = rx_pkt.getlayer(Ether)
614 tx_ip = tx_pkt.getlayer(IPv6)
615 # we can't just get the 2nd Ether layer
616 # get the Raw content and dissect it as Ether
617 tx_eth1 = Ether(scapy.compat.raw(tx_pkt[Raw]))
619 # verify if rx'ed packet has no SRH
620 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
622 # the whole rx_eth pkt should be equal to tx_eth1
623 self.assertEqual(rx_eth, tx_eth1)
625 self.logger.debug("packet verification: SUCCESS")
627 def create_stream(self, src_if, dst_if, packet_header, packet_sizes,
629 """Create SRv6 input packet stream for defined interface.
631 :param VppInterface src_if: Interface to create packet stream for
632 :param VppInterface dst_if: destination interface of packet stream
633 :param packet_header: Layer3 scapy packet headers,
634 L2 is added when not provided,
635 Raw(payload) with packet_info is added
636 :param list packet_sizes: packet stream pckt sizes,sequentially applied
637 to packets in stream have
638 :param int count: number of packets in packet stream
639 :return: list of packets
641 self.logger.info("Creating packets")
643 for i in range(0, count-1):
644 payload_info = self.create_packet_info(src_if, dst_if)
646 "Creating packet with index %d" % (payload_info.index))
647 payload = self.info_to_payload(payload_info)
648 # add L2 header if not yet provided in packet_header
649 if packet_header.getlayer(0).name == 'Ethernet':
653 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
656 size = packet_sizes[i % len(packet_sizes)]
657 self.logger.debug("Packet size %d" % (size))
658 self.extend_packet(p, size)
659 # we need to store the packet with the automatic fields computed
660 # read back the dumped packet (with str())
661 # to force computing these fields
662 # probably other ways are possible
663 p = Ether(scapy.compat.raw(p))
664 payload_info.data = p.copy()
665 self.logger.debug(ppp("Created packet:", p))
667 self.logger.info("Done creating packets")
670 def send_and_verify_pkts(self, input, pkts, output, compare_func):
671 """Send packets and verify received packets using compare_func
673 :param input: ingress interface of DUT
674 :param pkts: list of packets to transmit
675 :param output: egress interface of DUT
676 :param compare_func: function to compare in and out packets
678 # add traffic stream to input interface
679 input.add_stream(pkts)
681 # enable capture on all interfaces
682 self.pg_enable_capture(self.pg_interfaces)
685 self.logger.info("Starting traffic")
689 self.logger.info("Getting packet capture")
690 capture = output.get_capture()
692 # assert nothing was captured on input interface
693 # input.assert_nothing_captured()
695 # verify captured packets
696 self.verify_captured_pkts(output, capture, compare_func)
698 def create_packet_header_IPv6(self):
699 """Create packet header: IPv6 header, UDP header
701 :param dst: IPv6 destination address
703 IPv6 source address is 1234::1
704 IPv6 destination address is 4321::1
705 UDP source port and destination port are 1234
708 p = (IPv6(src='1234::1', dst='4321::1') /
709 UDP(sport=1234, dport=1234))
712 def create_packet_header_IPv6_SRH_IPv6(self, sidlist, segleft):
713 """Create packet header: IPv6 encapsulated in SRv6:
714 IPv6 header with SRH, IPv6 header, UDP header
716 :param list sidlist: segment list of outer IPv6 SRH
717 :param int segleft: segments-left field of outer IPv6 SRH
719 Outer IPv6 source address is set to 5678::1
720 Outer IPv6 destination address is set to sidlist[segleft]
721 IPv6 source addresses is 1234::1
722 IPv6 destination address is 4321::1
723 UDP source port and destination port are 1234
726 p = (IPv6(src='5678::1', dst=sidlist[segleft]) /
727 IPv6ExtHdrSegmentRouting(addresses=sidlist,
728 segleft=segleft, nh=41) /
729 IPv6(src='1234::1', dst='4321::1') /
730 UDP(sport=1234, dport=1234))
733 def create_packet_header_IPv4(self):
734 """Create packet header: IPv4 header, UDP header
736 :param dst: IPv4 destination address
738 IPv4 source address is 123.1.1.1
739 IPv4 destination address is 124.1.1.1
740 UDP source port and destination port are 1234
743 p = (IP(src='123.1.1.1', dst='124.1.1.1') /
744 UDP(sport=1234, dport=1234))
747 def create_packet_header_IPv6_SRH_IPv4(self, sidlist, segleft):
748 """Create packet header: IPv4 encapsulated in SRv6:
749 IPv6 header with SRH, IPv4 header, UDP header
751 :param ipv4address dst: inner IPv4 destination address
752 :param list sidlist: segment list of outer IPv6 SRH
753 :param int segleft: segments-left field of outer IPv6 SRH
755 Outer IPv6 destination address is set to sidlist[segleft]
756 IPv6 source address is 1234::1
757 IPv4 source address is 123.1.1.1
758 IPv4 destination address is 124.1.1.1
759 UDP source port and destination port are 1234
762 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
763 IPv6ExtHdrSegmentRouting(addresses=sidlist,
764 segleft=segleft, nh=4) /
765 IP(src='123.1.1.1', dst='124.1.1.1') /
766 UDP(sport=1234, dport=1234))
769 def create_packet_header_L2(self, vlan=0):
770 """Create packet header: L2 header
772 :param vlan: if vlan!=0 then add 802.1q header
774 # Note: the dst addr ('00:55:44:33:22:11') is used in
775 # the compare function compare_rx_tx_packet_T_Encaps_L2
776 # to detect presence of L2 in SRH payload
777 p = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
781 p /= Dot1Q(vlan=vlan, type=etype)
786 def create_packet_header_IPv6_SRH_L2(self, sidlist, segleft, vlan=0):
787 """Create packet header: L2 encapsulated in SRv6:
788 IPv6 header with SRH, L2
790 :param list sidlist: segment list of outer IPv6 SRH
791 :param int segleft: segments-left field of outer IPv6 SRH
792 :param vlan: L2 vlan; if vlan!=0 then add 802.1q header
794 Outer IPv6 destination address is set to sidlist[segleft]
795 IPv6 source address is 1234::1
797 eth = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
801 eth /= Dot1Q(vlan=vlan, type=etype)
805 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
806 IPv6ExtHdrSegmentRouting(addresses=sidlist,
807 segleft=segleft, nh=59) /
811 def get_payload_info(self, packet):
812 """ Extract the payload_info from the packet
814 # in most cases, payload_info is in packet[Raw]
815 # but packet[Raw] gives the complete payload
816 # (incl L2 header) for the T.Encaps L2 case
818 payload_info = self.payload_to_info(packet[Raw])
821 # remote L2 header from packet[Raw]:
822 # take packet[Raw], convert it to an Ether layer
823 # and then extract Raw from it
824 payload_info = self.payload_to_info(
825 Ether(scapy.compat.raw(packet[Raw]))[Raw])
829 def verify_captured_pkts(self, dst_if, capture, compare_func):
831 Verify captured packet stream for specified interface.
832 Compare ingress with egress packets using the specified compare fn
834 :param dst_if: egress interface of DUT
835 :param capture: captured packets
836 :param compare_func: function to compare in and out packet
838 self.logger.info("Verifying capture on interface %s using function %s"
839 % (dst_if.name, compare_func.__name__))
842 for i in self.pg_interfaces:
843 last_info[i.sw_if_index] = None
844 dst_sw_if_index = dst_if.sw_if_index
846 for packet in capture:
848 # extract payload_info from packet's payload
849 payload_info = self.get_payload_info(packet)
850 packet_index = payload_info.index
852 self.logger.debug("Verifying packet with index %d"
854 # packet should have arrived on the expected interface
855 self.assertEqual(payload_info.dst, dst_sw_if_index)
857 "Got packet on interface %s: src=%u (idx=%u)" %
858 (dst_if.name, payload_info.src, packet_index))
860 # search for payload_info with same src and dst if_index
861 # this will give us the transmitted packet
862 next_info = self.get_next_packet_info_for_interface2(
863 payload_info.src, dst_sw_if_index,
864 last_info[payload_info.src])
865 last_info[payload_info.src] = next_info
866 # next_info should not be None
867 self.assertTrue(next_info is not None)
868 # index of tx and rx packets should be equal
869 self.assertEqual(packet_index, next_info.index)
870 # data field of next_info contains the tx packet
871 txed_packet = next_info.data
873 self.logger.debug(ppp("Transmitted packet:",
874 txed_packet)) # ppp=Pretty Print Packet
876 self.logger.debug(ppp("Received packet:", packet))
878 # compare rcvd packet with expected packet using compare_func
879 compare_func(txed_packet, packet)
882 self.logger.error(ppp("Unexpected or invalid packet:", packet))
885 # have all expected packets arrived?
886 for i in self.pg_interfaces:
887 remaining_packet = self.get_next_packet_info_for_interface2(
888 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
889 self.assertTrue(remaining_packet is None,
890 "Interface %s: Packet expected from interface %s "
891 "didn't arrive" % (dst_if.name, i.name))
894 if __name__ == '__main__':
895 unittest.main(testRunner=VppTestRunner)