5 from socket import AF_INET6
7 from framework import VppTestCase, VppTestRunner
8 from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto, 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 route.add_vpp_config()
202 # configure SRv6 localSID behavior
203 cli_str = "sr localsid address " + sid_list[test_sid_index] \
204 + " behavior end.as" \
205 + " oif " + self.pg1.name \
206 + " iif " + self.pg1.name \
207 + " src " + self.rewrite_src_addr
208 for s in self.rewrite_sid_list:
209 cli_str += " next " + s
210 self.vapi.cli(cli_str)
213 self.logger.debug(self.vapi.cli("show sr localsid"))
215 # send one packet per packet size
216 count = len(self.pg_packet_sizes)
218 # prepare L2 in SRv6 headers
219 packet_header1 = self.create_packet_header_IPv6_SRH_L2(
220 sidlist=sid_list[::-1],
221 segleft=len(sid_list) - test_sid_index - 1,
224 # generate packets (pg0->pg1)
225 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
226 self.pg_packet_sizes, count)
228 # send packets and verify received packets
229 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
230 self.compare_rx_tx_packet_End_AS_L2_out)
232 # log the localsid counters
233 self.logger.info(self.vapi.cli("show sr localsid"))
235 # prepare L2 header for returning packets
236 packet_header2 = self.create_packet_header_L2()
238 # generate returning packets (pg1->pg0)
239 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
240 self.pg_packet_sizes, count)
242 # send packets and verify received packets
243 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
244 self.compare_rx_tx_packet_End_AS_L2_in)
246 # log the localsid counters
247 self.logger.info(self.vapi.cli("show sr localsid"))
249 # remove SRv6 localSIDs
250 self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
253 self.teardown_interfaces()
255 def run_SRv6_End_AS_IPv6(self, sid_list, test_sid_index, rewrite_src_addr):
256 """ Run SRv6 End.AS test with IPv6 traffic.
258 self.rewrite_src_addr = rewrite_src_addr
259 self.rewrite_sid_list = sid_list[test_sid_index + 1::]
261 # send traffic to one destination interface
262 # source and destination interfaces are IPv6 only
263 self.setup_interfaces(ipv6=[True, True])
265 # configure route to next segment
266 route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
267 [VppRoutePath(self.pg0.remote_ip6,
268 self.pg0.sw_if_index)])
269 route.add_vpp_config()
271 # configure SRv6 localSID behavior
272 cli_str = "sr localsid address " + sid_list[test_sid_index] \
273 + " behavior end.as" \
274 + " nh " + self.pg1.remote_ip6 \
275 + " oif " + self.pg1.name \
276 + " iif " + self.pg1.name \
277 + " src " + self.rewrite_src_addr
278 for s in self.rewrite_sid_list:
279 cli_str += " next " + s
280 self.vapi.cli(cli_str)
283 self.logger.debug(self.vapi.cli("show sr localsid"))
285 # send one packet per packet size
286 count = len(self.pg_packet_sizes)
288 # prepare IPv6 in SRv6 headers
289 packet_header1 = self.create_packet_header_IPv6_SRH_IPv6(
290 sidlist=sid_list[::-1],
291 segleft=len(sid_list) - test_sid_index - 1)
293 # generate packets (pg0->pg1)
294 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
295 self.pg_packet_sizes, count)
297 # send packets and verify received packets
298 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
299 self.compare_rx_tx_packet_End_AS_IPv6_out)
301 # log the localsid counters
302 self.logger.info(self.vapi.cli("show sr localsid"))
304 # prepare IPv6 header for returning packets
305 packet_header2 = self.create_packet_header_IPv6()
307 # generate returning packets (pg1->pg0)
308 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
309 self.pg_packet_sizes, count)
311 # send packets and verify received packets
312 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
313 self.compare_rx_tx_packet_End_AS_IPv6_in)
315 # log the localsid counters
316 self.logger.info(self.vapi.cli("show sr localsid"))
318 # remove SRv6 localSIDs
319 self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
322 self.teardown_interfaces()
324 def run_SRv6_End_AS_IPv4(self, sid_list, test_sid_index, rewrite_src_addr):
325 """ Run SRv6 End.AS test with IPv4 traffic.
327 self.rewrite_src_addr = rewrite_src_addr
328 self.rewrite_sid_list = sid_list[test_sid_index + 1::]
330 # send traffic to one destination interface
331 # source and destination interfaces are IPv6 only
332 self.setup_interfaces(ipv6=[True, False], ipv4=[True, True])
334 # configure route to next segment
335 route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
336 [VppRoutePath(self.pg0.remote_ip6,
337 self.pg0.sw_if_index)])
338 route.add_vpp_config()
340 # configure SRv6 localSID behavior
341 cli_str = "sr localsid address " + sid_list[test_sid_index] \
342 + " behavior end.as" \
343 + " nh " + self.pg1.remote_ip4 \
344 + " oif " + self.pg1.name \
345 + " iif " + self.pg1.name \
346 + " src " + self.rewrite_src_addr
347 for s in self.rewrite_sid_list:
348 cli_str += " next " + s
349 self.vapi.cli(cli_str)
352 self.logger.debug(self.vapi.cli("show sr localsid"))
354 # send one packet per packet size
355 count = len(self.pg_packet_sizes)
357 # prepare IPv4 in SRv6 headers
358 packet_header1 = self.create_packet_header_IPv6_SRH_IPv4(
359 sidlist=sid_list[::-1],
360 segleft=len(sid_list) - test_sid_index - 1)
362 # generate packets (pg0->pg1)
363 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
364 self.pg_packet_sizes, count)
366 # send packets and verify received packets
367 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
368 self.compare_rx_tx_packet_End_AS_IPv4_out)
370 # log the localsid counters
371 self.logger.info(self.vapi.cli("show sr localsid"))
373 # prepare IPv6 header for returning packets
374 packet_header2 = self.create_packet_header_IPv4()
376 # generate returning packets (pg1->pg0)
377 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
378 self.pg_packet_sizes, count)
380 # send packets and verify received packets
381 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
382 self.compare_rx_tx_packet_End_AS_IPv4_in)
384 # log the localsid counters
385 self.logger.info(self.vapi.cli("show sr localsid"))
387 # remove SRv6 localSIDs
388 self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
391 self.teardown_interfaces()
393 def compare_rx_tx_packet_End_AS_IPv6_in(self, tx_pkt, rx_pkt):
394 """ Compare input and output packet after passing End.AS
396 :param tx_pkt: transmitted packet
397 :param rx_pkt: received packet
400 # get first (outer) IPv6 header of rx'ed packet
401 rx_ip = rx_pkt.getlayer(IPv6)
404 tx_ip = tx_pkt.getlayer(IPv6)
406 # expected segment-list (SRH order)
407 tx_seglist = self.rewrite_sid_list[::-1]
409 # received ip.src should be equal to SR Policy source
410 self.assertEqual(rx_ip.src, self.rewrite_src_addr)
411 # received ip.dst should be equal to expected sidlist[lastentry]
412 self.assertEqual(rx_ip.dst, tx_seglist[-1])
414 if len(tx_seglist) > 1:
415 # rx'ed packet should have SRH
416 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
418 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
419 # rx'ed seglist should be equal to expected seglist
420 self.assertEqual(rx_srh.addresses, tx_seglist)
421 # segleft should be equal to size expected seglist-1
422 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
423 # segleft should be equal to lastentry
424 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
426 payload = rx_srh.payload
428 # rx'ed packet should NOT have SRH
429 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
431 payload = rx_ip.payload
433 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
434 # except for the hop-limit field
435 # -> update tx'ed hlim to the expected hlim
436 tx_ip.hlim = tx_ip.hlim - 1
438 self.assertEqual(payload, tx_ip)
440 self.logger.debug("packet verification: SUCCESS")
442 def compare_rx_tx_packet_End_AS_IPv4_in(self, tx_pkt, rx_pkt):
443 """ Compare input and output packet after passing End.AS
445 :param tx_pkt: transmitted packet
446 :param rx_pkt: received packet
449 # get first (outer) IPv6 header of rx'ed packet
450 rx_ip = rx_pkt.getlayer(IPv6)
453 tx_ip = tx_pkt.getlayer(IP)
455 # expected segment-list (SRH order)
456 tx_seglist = self.rewrite_sid_list[::-1]
458 # received ip.src should be equal to SR Policy source
459 self.assertEqual(rx_ip.src, self.rewrite_src_addr)
460 # received ip.dst should be equal to expected sidlist[lastentry]
461 self.assertEqual(rx_ip.dst, tx_seglist[-1])
463 if len(tx_seglist) > 1:
464 # rx'ed packet should have SRH and IPv4 header
465 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
466 self.assertTrue(rx_ip.payload.haslayer(IP))
468 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
469 # rx'ed seglist should be equal to seglist
470 self.assertEqual(rx_srh.addresses, tx_seglist)
471 # segleft should be equal to size seglist-1
472 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
473 # segleft should be equal to lastentry
474 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
475 payload = rx_srh.payload
477 # rx'ed packet should NOT have SRH
478 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
480 payload = rx_ip.payload
482 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
483 # except for the ttl field and ip checksum
484 # -> adjust tx'ed ttl to expected ttl
485 tx_ip.ttl = tx_ip.ttl - 1
486 # -> set tx'ed ip checksum to None and let scapy recompute
488 # read back the pkt (with str()) to force computing these fields
489 # probably other ways to accomplish this are possible
490 tx_ip = IP(scapy.compat.raw(tx_ip))
492 self.assertEqual(payload, tx_ip)
494 self.logger.debug("packet verification: SUCCESS")
496 def compare_rx_tx_packet_End_AS_L2_in(self, tx_pkt, rx_pkt):
497 """ Compare input and output packet after passing End.AS
499 :param tx_pkt: transmitted packet
500 :param rx_pkt: received packet
503 # get first (outer) IPv6 header of rx'ed packet
504 rx_ip = rx_pkt.getlayer(IPv6)
507 tx_ether = tx_pkt.getlayer(Ether)
509 # expected segment-list (SRH order)
510 tx_seglist = self.rewrite_sid_list[::-1]
512 # received ip.src should be equal to SR Policy source
513 self.assertEqual(rx_ip.src, self.rewrite_src_addr)
514 # received ip.dst should be equal to expected sidlist[lastentry]
515 self.assertEqual(rx_ip.dst, tx_seglist[-1])
517 if len(tx_seglist) > 1:
518 # rx'ed packet should have SRH
519 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
521 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
522 # rx'ed seglist should be equal to seglist
523 self.assertEqual(rx_srh.addresses, tx_seglist)
524 # segleft should be equal to size seglist-1
525 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
526 # segleft should be equal to lastentry
527 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
528 # nh should be "No Next Header" (59)
529 self.assertEqual(rx_srh.nh, 59)
531 payload = rx_srh.payload
533 # rx'ed packet should NOT have SRH
534 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
536 payload = rx_ip.payload
538 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
539 self.assertEqual(Ether(scapy.compat.raw(payload)), tx_ether)
541 self.logger.debug("packet verification: SUCCESS")
543 def compare_rx_tx_packet_End_AS_IPv6_out(self, tx_pkt, rx_pkt):
544 """ Compare input and output packet after passing End.AS with IPv6
546 :param tx_pkt: transmitted packet
547 :param rx_pkt: received packet
550 # get first (outer) IPv6 header of rx'ed packet
551 rx_ip = rx_pkt.getlayer(IPv6)
553 tx_ip = tx_pkt.getlayer(IPv6)
554 tx_ip2 = tx_pkt.getlayer(IPv6, 2)
556 # verify if rx'ed packet has no SRH
557 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
559 # the whole rx_ip pkt should be equal to tx_ip2
560 # except for the hlim field
561 # -> adjust tx'ed hlim to expected hlim
562 tx_ip2.hlim = tx_ip2.hlim - 1
564 self.assertEqual(rx_ip, tx_ip2)
566 self.logger.debug("packet verification: SUCCESS")
568 def compare_rx_tx_packet_End_AS_IPv4_out(self, tx_pkt, rx_pkt):
569 """ Compare input and output packet after passing End.AS with IPv4
571 :param tx_pkt: transmitted packet
572 :param rx_pkt: received packet
575 # get IPv4 header of rx'ed packet
576 rx_ip = rx_pkt.getlayer(IP)
578 tx_ip = tx_pkt.getlayer(IPv6)
579 tx_ip2 = tx_pkt.getlayer(IP)
581 # verify if rx'ed packet has no SRH
582 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
584 # the whole rx_ip pkt should be equal to tx_ip2
585 # except for the ttl field and ip checksum
586 # -> adjust tx'ed ttl to expected ttl
587 tx_ip2.ttl = tx_ip2.ttl - 1
588 # -> set tx'ed ip checksum to None and let scapy recompute
590 # read back the pkt (with str()) to force computing these fields
591 # probably other ways to accomplish this are possible
592 tx_ip2 = IP(scapy.compat.raw(tx_ip2))
594 self.assertEqual(rx_ip, tx_ip2)
596 self.logger.debug("packet verification: SUCCESS")
598 def compare_rx_tx_packet_End_AS_L2_out(self, tx_pkt, rx_pkt):
599 """ Compare input and output packet after passing End.AS with L2
601 :param tx_pkt: transmitted packet
602 :param rx_pkt: received packet
605 # get IPv4 header of rx'ed packet
606 rx_eth = rx_pkt.getlayer(Ether)
608 tx_ip = tx_pkt.getlayer(IPv6)
609 # we can't just get the 2nd Ether layer
610 # get the Raw content and dissect it as Ether
611 tx_eth1 = Ether(scapy.compat.raw(tx_pkt[Raw]))
613 # verify if rx'ed packet has no SRH
614 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
616 # the whole rx_eth pkt should be equal to tx_eth1
617 self.assertEqual(rx_eth, tx_eth1)
619 self.logger.debug("packet verification: SUCCESS")
621 def create_stream(self, src_if, dst_if, packet_header, packet_sizes,
623 """Create SRv6 input packet stream for defined interface.
625 :param VppInterface src_if: Interface to create packet stream for
626 :param VppInterface dst_if: destination interface of packet stream
627 :param packet_header: Layer3 scapy packet headers,
628 L2 is added when not provided,
629 Raw(payload) with packet_info is added
630 :param list packet_sizes: packet stream pckt sizes,sequentially applied
631 to packets in stream have
632 :param int count: number of packets in packet stream
633 :return: list of packets
635 self.logger.info("Creating packets")
637 for i in range(0, count-1):
638 payload_info = self.create_packet_info(src_if, dst_if)
640 "Creating packet with index %d" % (payload_info.index))
641 payload = self.info_to_payload(payload_info)
642 # add L2 header if not yet provided in packet_header
643 if packet_header.getlayer(0).name == 'Ethernet':
647 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
650 size = packet_sizes[i % len(packet_sizes)]
651 self.logger.debug("Packet size %d" % (size))
652 self.extend_packet(p, size)
653 # we need to store the packet with the automatic fields computed
654 # read back the dumped packet (with str())
655 # to force computing these fields
656 # probably other ways are possible
657 p = Ether(scapy.compat.raw(p))
658 payload_info.data = p.copy()
659 self.logger.debug(ppp("Created packet:", p))
661 self.logger.info("Done creating packets")
664 def send_and_verify_pkts(self, input, pkts, output, compare_func):
665 """Send packets and verify received packets using compare_func
667 :param input: ingress interface of DUT
668 :param pkts: list of packets to transmit
669 :param output: egress interface of DUT
670 :param compare_func: function to compare in and out packets
672 # add traffic stream to input interface
673 input.add_stream(pkts)
675 # enable capture on all interfaces
676 self.pg_enable_capture(self.pg_interfaces)
679 self.logger.info("Starting traffic")
683 self.logger.info("Getting packet capture")
684 capture = output.get_capture()
686 # assert nothing was captured on input interface
687 # input.assert_nothing_captured()
689 # verify captured packets
690 self.verify_captured_pkts(output, capture, compare_func)
692 def create_packet_header_IPv6(self):
693 """Create packet header: IPv6 header, UDP header
695 :param dst: IPv6 destination address
697 IPv6 source address is 1234::1
698 IPv6 destination address is 4321::1
699 UDP source port and destination port are 1234
702 p = (IPv6(src='1234::1', dst='4321::1') /
703 UDP(sport=1234, dport=1234))
706 def create_packet_header_IPv6_SRH_IPv6(self, sidlist, segleft):
707 """Create packet header: IPv6 encapsulated in SRv6:
708 IPv6 header with SRH, IPv6 header, UDP header
710 :param list sidlist: segment list of outer IPv6 SRH
711 :param int segleft: segments-left field of outer IPv6 SRH
713 Outer IPv6 source address is set to 5678::1
714 Outer IPv6 destination address is set to sidlist[segleft]
715 IPv6 source addresses is 1234::1
716 IPv6 destination address is 4321::1
717 UDP source port and destination port are 1234
720 p = (IPv6(src='5678::1', dst=sidlist[segleft]) /
721 IPv6ExtHdrSegmentRouting(addresses=sidlist,
722 segleft=segleft, nh=41) /
723 IPv6(src='1234::1', dst='4321::1') /
724 UDP(sport=1234, dport=1234))
727 def create_packet_header_IPv4(self):
728 """Create packet header: IPv4 header, UDP header
730 :param dst: IPv4 destination address
732 IPv4 source address is 123.1.1.1
733 IPv4 destination address is 124.1.1.1
734 UDP source port and destination port are 1234
737 p = (IP(src='123.1.1.1', dst='124.1.1.1') /
738 UDP(sport=1234, dport=1234))
741 def create_packet_header_IPv6_SRH_IPv4(self, sidlist, segleft):
742 """Create packet header: IPv4 encapsulated in SRv6:
743 IPv6 header with SRH, IPv4 header, UDP header
745 :param ipv4address dst: inner IPv4 destination address
746 :param list sidlist: segment list of outer IPv6 SRH
747 :param int segleft: segments-left field of outer IPv6 SRH
749 Outer IPv6 destination address is set to sidlist[segleft]
750 IPv6 source address is 1234::1
751 IPv4 source address is 123.1.1.1
752 IPv4 destination address is 124.1.1.1
753 UDP source port and destination port are 1234
756 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
757 IPv6ExtHdrSegmentRouting(addresses=sidlist,
758 segleft=segleft, nh=4) /
759 IP(src='123.1.1.1', dst='124.1.1.1') /
760 UDP(sport=1234, dport=1234))
763 def create_packet_header_L2(self, vlan=0):
764 """Create packet header: L2 header
766 :param vlan: if vlan!=0 then add 802.1q header
768 # Note: the dst addr ('00:55:44:33:22:11') is used in
769 # the compare function compare_rx_tx_packet_T_Encaps_L2
770 # to detect presence of L2 in SRH payload
771 p = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
775 p /= Dot1Q(vlan=vlan, type=etype)
780 def create_packet_header_IPv6_SRH_L2(self, sidlist, segleft, vlan=0):
781 """Create packet header: L2 encapsulated in SRv6:
782 IPv6 header with SRH, L2
784 :param list sidlist: segment list of outer IPv6 SRH
785 :param int segleft: segments-left field of outer IPv6 SRH
786 :param vlan: L2 vlan; if vlan!=0 then add 802.1q header
788 Outer IPv6 destination address is set to sidlist[segleft]
789 IPv6 source address is 1234::1
791 eth = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
795 eth /= Dot1Q(vlan=vlan, type=etype)
799 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
800 IPv6ExtHdrSegmentRouting(addresses=sidlist,
801 segleft=segleft, nh=59) /
805 def get_payload_info(self, packet):
806 """ Extract the payload_info from the packet
808 # in most cases, payload_info is in packet[Raw]
809 # but packet[Raw] gives the complete payload
810 # (incl L2 header) for the T.Encaps L2 case
812 payload_info = self.payload_to_info(packet[Raw])
815 # remote L2 header from packet[Raw]:
816 # take packet[Raw], convert it to an Ether layer
817 # and then extract Raw from it
818 payload_info = self.payload_to_info(
819 Ether(scapy.compat.raw(packet[Raw]))[Raw])
823 def verify_captured_pkts(self, dst_if, capture, compare_func):
825 Verify captured packet stream for specified interface.
826 Compare ingress with egress packets using the specified compare fn
828 :param dst_if: egress interface of DUT
829 :param capture: captured packets
830 :param compare_func: function to compare in and out packet
832 self.logger.info("Verifying capture on interface %s using function %s"
833 % (dst_if.name, compare_func.__name__))
836 for i in self.pg_interfaces:
837 last_info[i.sw_if_index] = None
838 dst_sw_if_index = dst_if.sw_if_index
840 for packet in capture:
842 # extract payload_info from packet's payload
843 payload_info = self.get_payload_info(packet)
844 packet_index = payload_info.index
846 self.logger.debug("Verifying packet with index %d"
848 # packet should have arrived on the expected interface
849 self.assertEqual(payload_info.dst, dst_sw_if_index)
851 "Got packet on interface %s: src=%u (idx=%u)" %
852 (dst_if.name, payload_info.src, packet_index))
854 # search for payload_info with same src and dst if_index
855 # this will give us the transmitted packet
856 next_info = self.get_next_packet_info_for_interface2(
857 payload_info.src, dst_sw_if_index,
858 last_info[payload_info.src])
859 last_info[payload_info.src] = next_info
860 # next_info should not be None
861 self.assertTrue(next_info is not None)
862 # index of tx and rx packets should be equal
863 self.assertEqual(packet_index, next_info.index)
864 # data field of next_info contains the tx packet
865 txed_packet = next_info.data
867 self.logger.debug(ppp("Transmitted packet:",
868 txed_packet)) # ppp=Pretty Print Packet
870 self.logger.debug(ppp("Received packet:", packet))
872 # compare rcvd packet with expected packet using compare_func
873 compare_func(txed_packet, packet)
876 self.logger.error(ppp("Unexpected or invalid packet:", packet))
879 # have all expected packets arrived?
880 for i in self.pg_interfaces:
881 remaining_packet = self.get_next_packet_info_for_interface2(
882 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
883 self.assertTrue(remaining_packet is None,
884 "Interface %s: Packet expected from interface %s "
885 "didn't arrive" % (dst_if.name, i.name))
888 if __name__ == '__main__':
889 unittest.main(testRunner=VppTestRunner)