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
21 class TestSRv6(VppTestCase):
22 """ SRv6 Static Proxy plugin Test Case """
26 super(TestSRv6, self).setUpClass()
29 def tearDownClass(cls):
30 super(TestSRv6, cls).tearDownClass()
33 """ Perform test setup before each test case.
35 super(TestSRv6, self).setUp()
37 # packet sizes, inclusive L2 overhead
38 self.pg_packet_sizes = [64, 512, 1518, 9018]
41 self.reset_packet_infos()
44 """ Clean up test setup after each test case.
46 self.teardown_interfaces()
48 super(TestSRv6, self).tearDown()
50 def configure_interface(self,
52 ipv6=False, ipv4=False,
53 ipv6_table_id=0, ipv4_table_id=0):
54 """ Configure interface.
55 :param ipv6: configure IPv6 on interface
56 :param ipv4: configure IPv4 on interface
57 :param ipv6_table_id: FIB table_id for IPv6
58 :param ipv4_table_id: FIB table_id for IPv4
60 self.logger.debug("Configuring interface %s" % (interface.name))
62 self.logger.debug("Configuring IPv6")
63 interface.set_table_ip6(ipv6_table_id)
64 interface.config_ip6()
65 interface.resolve_ndp(timeout=5)
67 self.logger.debug("Configuring IPv4")
68 interface.set_table_ip4(ipv4_table_id)
69 interface.config_ip4()
70 interface.resolve_arp()
73 def setup_interfaces(self, ipv6=[], ipv4=[],
74 ipv6_table_id=[], ipv4_table_id=[]):
75 """ Create and configure interfaces.
77 :param ipv6: list of interface IPv6 capabilities
78 :param ipv4: list of interface IPv4 capabilities
79 :param ipv6_table_id: list of intf IPv6 FIB table_ids
80 :param ipv4_table_id: list of intf IPv4 FIB table_ids
81 :returns: List of created interfaces.
83 # how many interfaces?
88 self.logger.debug("Creating and configuring %d interfaces" % (count))
90 # fill up ipv6 and ipv4 lists if needed
91 # not enabled (False) is the default
93 ipv6 += (count - len(ipv6)) * [False]
95 ipv4 += (count - len(ipv4)) * [False]
97 # fill up table_id lists if needed
98 # table_id 0 (global) is the default
99 if len(ipv6_table_id) < count:
100 ipv6_table_id += (count - len(ipv6_table_id)) * [0]
101 if len(ipv4_table_id) < count:
102 ipv4_table_id += (count - len(ipv4_table_id)) * [0]
104 # create 'count' pg interfaces
105 self.create_pg_interfaces(range(count))
107 # setup all interfaces
108 for i in range(count):
109 intf = self.pg_interfaces[i]
110 self.configure_interface(intf,
112 ipv6_table_id[i], ipv4_table_id[i])
115 self.logger.debug(self.vapi.cli("show ip6 neighbors"))
117 self.logger.debug(self.vapi.cli("show ip4 neighbors"))
118 self.logger.debug(self.vapi.cli("show interface"))
119 self.logger.debug(self.vapi.cli("show hardware"))
121 return self.pg_interfaces
123 def teardown_interfaces(self):
124 """ Unconfigure and bring down interface.
126 self.logger.debug("Tearing down interfaces")
127 # tear down all interfaces
128 # AFAIK they cannot be deleted
129 for i in self.pg_interfaces:
130 self.logger.debug("Tear down interface %s" % (i.name))
136 def test_SRv6_End_AS_IPv6_noSRH(self):
137 """ Test SRv6 End.AS behavior with IPv6 traffic and no SRH rewrite.
139 self.run_SRv6_End_AS_IPv6(
140 sid_list=['a1::', 'a2::a6', 'a3::'],
142 rewrite_src_addr='a2::')
144 def test_SRv6_End_AS_IPv6_SRH(self):
145 """ Test SRv6 End.AS behavior with IPv6 traffic and SRH rewrite.
147 self.run_SRv6_End_AS_IPv6(
148 sid_list=['a1::a6', 'a2::', 'a3::'],
150 rewrite_src_addr='a1::')
152 def test_SRv6_End_AS_IPv4_noSRH(self):
153 """ Test SRv6 End.AS behavior with IPv4 traffic and no SRH rewrite.
155 self.run_SRv6_End_AS_IPv4(
156 sid_list=['a1::', 'a2::a6', 'a3::'],
158 rewrite_src_addr='a2::')
160 def test_SRv6_End_AS_IPv4_SRH(self):
161 """ Test SRv6 End.AS behavior with IPv4 traffic and SRH rewrite.
163 self.run_SRv6_End_AS_IPv4(
164 sid_list=['a1::a6', 'a2::', 'a3::'],
166 rewrite_src_addr='a1::')
168 def test_SRv6_End_AS_L2_noSRH(self):
169 """ Test SRv6 End.AS behavior with L2 traffic and no SRH rewrite.
171 self.run_SRv6_End_AS_L2(
172 sid_list=['a1::', 'a2::a6', 'a3::'],
174 rewrite_src_addr='a2::')
176 def test_SRv6_End_AS_L2_SRH(self):
177 """ Test SRv6 End.AS behavior with L2 traffic and SRH rewrite.
179 self.run_SRv6_End_AS_L2(
180 sid_list=['a1::a6', 'a2::', 'a3::'],
182 rewrite_src_addr='a1::')
184 def run_SRv6_End_AS_L2(self, sid_list, test_sid_index, rewrite_src_addr):
185 """ Run SRv6 End.AS test with L2 traffic.
187 self.rewrite_src_addr = rewrite_src_addr
188 self.rewrite_sid_list = sid_list[test_sid_index + 1::]
190 # send traffic to one destination interface
191 # source and destination interfaces are IPv6 only
192 self.setup_interfaces(ipv6=[True, False])
194 # configure route to next segment
195 route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
196 [VppRoutePath(self.pg0.remote_ip6,
197 self.pg0.sw_if_index)])
198 route.add_vpp_config()
200 # configure SRv6 localSID behavior
201 cli_str = "sr localsid address " + sid_list[test_sid_index] \
202 + " behavior end.as" \
203 + " oif " + self.pg1.name \
204 + " iif " + self.pg1.name \
205 + " src " + self.rewrite_src_addr
206 for s in self.rewrite_sid_list:
207 cli_str += " next " + s
208 self.vapi.cli(cli_str)
211 self.logger.debug(self.vapi.cli("show sr localsid"))
213 # send one packet per packet size
214 count = len(self.pg_packet_sizes)
216 # prepare L2 in SRv6 headers
217 packet_header1 = self.create_packet_header_IPv6_SRH_L2(
218 sidlist=sid_list[::-1],
219 segleft=len(sid_list) - test_sid_index - 1,
222 # generate packets (pg0->pg1)
223 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
224 self.pg_packet_sizes, count)
226 # send packets and verify received packets
227 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
228 self.compare_rx_tx_packet_End_AS_L2_out)
230 # log the localsid counters
231 self.logger.info(self.vapi.cli("show sr localsid"))
233 # prepare L2 header for returning packets
234 packet_header2 = self.create_packet_header_L2()
236 # generate returning packets (pg1->pg0)
237 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
238 self.pg_packet_sizes, count)
240 # send packets and verify received packets
241 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
242 self.compare_rx_tx_packet_End_AS_L2_in)
244 # log the localsid counters
245 self.logger.info(self.vapi.cli("show sr localsid"))
247 # remove SRv6 localSIDs
248 self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
251 self.teardown_interfaces()
253 def run_SRv6_End_AS_IPv6(self, sid_list, test_sid_index, rewrite_src_addr):
254 """ Run SRv6 End.AS test with IPv6 traffic.
256 self.rewrite_src_addr = rewrite_src_addr
257 self.rewrite_sid_list = sid_list[test_sid_index + 1::]
259 # send traffic to one destination interface
260 # source and destination interfaces are IPv6 only
261 self.setup_interfaces(ipv6=[True, True])
263 # configure route to next segment
264 route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
265 [VppRoutePath(self.pg0.remote_ip6,
266 self.pg0.sw_if_index)])
267 route.add_vpp_config()
269 # configure SRv6 localSID behavior
270 cli_str = "sr localsid address " + sid_list[test_sid_index] \
271 + " behavior end.as" \
272 + " nh " + self.pg1.remote_ip6 \
273 + " oif " + self.pg1.name \
274 + " iif " + self.pg1.name \
275 + " src " + self.rewrite_src_addr
276 for s in self.rewrite_sid_list:
277 cli_str += " next " + s
278 self.vapi.cli(cli_str)
281 self.logger.debug(self.vapi.cli("show sr localsid"))
283 # send one packet per packet size
284 count = len(self.pg_packet_sizes)
286 # prepare IPv6 in SRv6 headers
287 packet_header1 = self.create_packet_header_IPv6_SRH_IPv6(
288 sidlist=sid_list[::-1],
289 segleft=len(sid_list) - test_sid_index - 1)
291 # generate packets (pg0->pg1)
292 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
293 self.pg_packet_sizes, count)
295 # send packets and verify received packets
296 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
297 self.compare_rx_tx_packet_End_AS_IPv6_out)
299 # log the localsid counters
300 self.logger.info(self.vapi.cli("show sr localsid"))
302 # prepare IPv6 header for returning packets
303 packet_header2 = self.create_packet_header_IPv6()
305 # generate returning packets (pg1->pg0)
306 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
307 self.pg_packet_sizes, count)
309 # send packets and verify received packets
310 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
311 self.compare_rx_tx_packet_End_AS_IPv6_in)
313 # log the localsid counters
314 self.logger.info(self.vapi.cli("show sr localsid"))
316 # remove SRv6 localSIDs
317 self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
320 self.teardown_interfaces()
322 def run_SRv6_End_AS_IPv4(self, sid_list, test_sid_index, rewrite_src_addr):
323 """ Run SRv6 End.AS test with IPv4 traffic.
325 self.rewrite_src_addr = rewrite_src_addr
326 self.rewrite_sid_list = sid_list[test_sid_index + 1::]
328 # send traffic to one destination interface
329 # source and destination interfaces are IPv6 only
330 self.setup_interfaces(ipv6=[True, False], ipv4=[True, True])
332 # configure route to next segment
333 route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
334 [VppRoutePath(self.pg0.remote_ip6,
335 self.pg0.sw_if_index)])
336 route.add_vpp_config()
338 # configure SRv6 localSID behavior
339 cli_str = "sr localsid address " + sid_list[test_sid_index] \
340 + " behavior end.as" \
341 + " nh " + self.pg1.remote_ip4 \
342 + " oif " + self.pg1.name \
343 + " iif " + self.pg1.name \
344 + " src " + self.rewrite_src_addr
345 for s in self.rewrite_sid_list:
346 cli_str += " next " + s
347 self.vapi.cli(cli_str)
350 self.logger.debug(self.vapi.cli("show sr localsid"))
352 # send one packet per packet size
353 count = len(self.pg_packet_sizes)
355 # prepare IPv4 in SRv6 headers
356 packet_header1 = self.create_packet_header_IPv6_SRH_IPv4(
357 sidlist=sid_list[::-1],
358 segleft=len(sid_list) - test_sid_index - 1)
360 # generate packets (pg0->pg1)
361 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
362 self.pg_packet_sizes, count)
364 # send packets and verify received packets
365 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
366 self.compare_rx_tx_packet_End_AS_IPv4_out)
368 # log the localsid counters
369 self.logger.info(self.vapi.cli("show sr localsid"))
371 # prepare IPv6 header for returning packets
372 packet_header2 = self.create_packet_header_IPv4()
374 # generate returning packets (pg1->pg0)
375 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
376 self.pg_packet_sizes, count)
378 # send packets and verify received packets
379 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
380 self.compare_rx_tx_packet_End_AS_IPv4_in)
382 # log the localsid counters
383 self.logger.info(self.vapi.cli("show sr localsid"))
385 # remove SRv6 localSIDs
386 self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
389 self.teardown_interfaces()
391 def compare_rx_tx_packet_End_AS_IPv6_in(self, tx_pkt, rx_pkt):
392 """ Compare input and output packet after passing End.AS
394 :param tx_pkt: transmitted packet
395 :param rx_pkt: received packet
398 # get first (outer) IPv6 header of rx'ed packet
399 rx_ip = rx_pkt.getlayer(IPv6)
402 tx_ip = tx_pkt.getlayer(IPv6)
404 # expected segment-list (SRH order)
405 tx_seglist = self.rewrite_sid_list[::-1]
407 # received ip.src should be equal to SR Policy source
408 self.assertEqual(rx_ip.src, self.rewrite_src_addr)
409 # received ip.dst should be equal to expected sidlist[lastentry]
410 self.assertEqual(rx_ip.dst, tx_seglist[-1])
412 if len(tx_seglist) > 1:
413 # rx'ed packet should have SRH
414 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
416 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
417 # rx'ed seglist should be equal to expected seglist
418 self.assertEqual(rx_srh.addresses, tx_seglist)
419 # segleft should be equal to size expected seglist-1
420 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
421 # segleft should be equal to lastentry
422 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
424 payload = rx_srh.payload
426 # rx'ed packet should NOT have SRH
427 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
429 payload = rx_ip.payload
431 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
432 # except for the hop-limit field
433 # -> update tx'ed hlim to the expected hlim
434 tx_ip.hlim = tx_ip.hlim - 1
436 self.assertEqual(payload, tx_ip)
438 self.logger.debug("packet verification: SUCCESS")
440 def compare_rx_tx_packet_End_AS_IPv4_in(self, tx_pkt, rx_pkt):
441 """ Compare input and output packet after passing End.AS
443 :param tx_pkt: transmitted packet
444 :param rx_pkt: received packet
447 # get first (outer) IPv6 header of rx'ed packet
448 rx_ip = rx_pkt.getlayer(IPv6)
451 tx_ip = tx_pkt.getlayer(IP)
453 # expected segment-list (SRH order)
454 tx_seglist = self.rewrite_sid_list[::-1]
456 # received ip.src should be equal to SR Policy source
457 self.assertEqual(rx_ip.src, self.rewrite_src_addr)
458 # received ip.dst should be equal to expected sidlist[lastentry]
459 self.assertEqual(rx_ip.dst, tx_seglist[-1])
461 if len(tx_seglist) > 1:
462 # rx'ed packet should have SRH and IPv4 header
463 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
464 self.assertTrue(rx_ip.payload.haslayer(IP))
466 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
467 # rx'ed seglist should be equal to seglist
468 self.assertEqual(rx_srh.addresses, tx_seglist)
469 # segleft should be equal to size seglist-1
470 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
471 # segleft should be equal to lastentry
472 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
473 payload = rx_srh.payload
475 # rx'ed packet should NOT have SRH
476 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
478 payload = rx_ip.payload
480 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
481 # except for the ttl field and ip checksum
482 # -> adjust tx'ed ttl to expected ttl
483 tx_ip.ttl = tx_ip.ttl - 1
484 # -> set tx'ed ip checksum to None and let scapy recompute
486 # read back the pkt (with str()) to force computing these fields
487 # probably other ways to accomplish this are possible
488 tx_ip = IP(scapy.compat.raw(tx_ip))
490 self.assertEqual(payload, tx_ip)
492 self.logger.debug("packet verification: SUCCESS")
494 def compare_rx_tx_packet_End_AS_L2_in(self, tx_pkt, rx_pkt):
495 """ Compare input and output packet after passing End.AS
497 :param tx_pkt: transmitted packet
498 :param rx_pkt: received packet
501 # get first (outer) IPv6 header of rx'ed packet
502 rx_ip = rx_pkt.getlayer(IPv6)
505 tx_ether = tx_pkt.getlayer(Ether)
507 # expected segment-list (SRH order)
508 tx_seglist = self.rewrite_sid_list[::-1]
510 # received ip.src should be equal to SR Policy source
511 self.assertEqual(rx_ip.src, self.rewrite_src_addr)
512 # received ip.dst should be equal to expected sidlist[lastentry]
513 self.assertEqual(rx_ip.dst, tx_seglist[-1])
515 if len(tx_seglist) > 1:
516 # rx'ed packet should have SRH
517 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
519 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
520 # rx'ed seglist should be equal to seglist
521 self.assertEqual(rx_srh.addresses, tx_seglist)
522 # segleft should be equal to size seglist-1
523 self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
524 # segleft should be equal to lastentry
525 self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
526 # nh should be "No Next Header" (143)
527 self.assertEqual(rx_srh.nh, 143)
529 payload = rx_srh.payload
531 # rx'ed packet should NOT have SRH
532 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
534 payload = rx_ip.payload
536 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
537 self.assertEqual(Ether(scapy.compat.raw(payload)), tx_ether)
539 self.logger.debug("packet verification: SUCCESS")
541 def compare_rx_tx_packet_End_AS_IPv6_out(self, tx_pkt, rx_pkt):
542 """ Compare input and output packet after passing End.AS with IPv6
544 :param tx_pkt: transmitted packet
545 :param rx_pkt: received packet
548 # get first (outer) IPv6 header of rx'ed packet
549 rx_ip = rx_pkt.getlayer(IPv6)
551 tx_ip = tx_pkt.getlayer(IPv6)
552 tx_ip2 = tx_pkt.getlayer(IPv6, 2)
554 # verify if rx'ed packet has no SRH
555 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
557 # the whole rx_ip pkt should be equal to tx_ip2
558 # except for the hlim field
559 # -> adjust tx'ed hlim to expected hlim
560 tx_ip2.hlim = tx_ip2.hlim - 1
562 self.assertEqual(rx_ip, tx_ip2)
564 self.logger.debug("packet verification: SUCCESS")
566 def compare_rx_tx_packet_End_AS_IPv4_out(self, tx_pkt, rx_pkt):
567 """ Compare input and output packet after passing End.AS with IPv4
569 :param tx_pkt: transmitted packet
570 :param rx_pkt: received packet
573 # get IPv4 header of rx'ed packet
574 rx_ip = rx_pkt.getlayer(IP)
576 tx_ip = tx_pkt.getlayer(IPv6)
577 tx_ip2 = tx_pkt.getlayer(IP)
579 # verify if rx'ed packet has no SRH
580 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
582 # the whole rx_ip pkt should be equal to tx_ip2
583 # except for the ttl field and ip checksum
584 # -> adjust tx'ed ttl to expected ttl
585 tx_ip2.ttl = tx_ip2.ttl - 1
586 # -> set tx'ed ip checksum to None and let scapy recompute
588 # read back the pkt (with str()) to force computing these fields
589 # probably other ways to accomplish this are possible
590 tx_ip2 = IP(scapy.compat.raw(tx_ip2))
592 self.assertEqual(rx_ip, tx_ip2)
594 self.logger.debug("packet verification: SUCCESS")
596 def compare_rx_tx_packet_End_AS_L2_out(self, tx_pkt, rx_pkt):
597 """ Compare input and output packet after passing End.AS with L2
599 :param tx_pkt: transmitted packet
600 :param rx_pkt: received packet
603 # get IPv4 header of rx'ed packet
604 rx_eth = rx_pkt.getlayer(Ether)
606 tx_ip = tx_pkt.getlayer(IPv6)
607 # we can't just get the 2nd Ether layer
608 # get the Raw content and dissect it as Ether
609 tx_eth1 = Ether(scapy.compat.raw(tx_pkt[Raw]))
611 # verify if rx'ed packet has no SRH
612 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
614 # the whole rx_eth pkt should be equal to tx_eth1
615 self.assertEqual(rx_eth, tx_eth1)
617 self.logger.debug("packet verification: SUCCESS")
619 def create_stream(self, src_if, dst_if, packet_header, packet_sizes,
621 """Create SRv6 input packet stream for defined interface.
623 :param VppInterface src_if: Interface to create packet stream for
624 :param VppInterface dst_if: destination interface of packet stream
625 :param packet_header: Layer3 scapy packet headers,
626 L2 is added when not provided,
627 Raw(payload) with packet_info is added
628 :param list packet_sizes: packet stream pckt sizes,sequentially applied
629 to packets in stream have
630 :param int count: number of packets in packet stream
631 :return: list of packets
633 self.logger.info("Creating packets")
635 for i in range(0, count-1):
636 payload_info = self.create_packet_info(src_if, dst_if)
638 "Creating packet with index %d" % (payload_info.index))
639 payload = self.info_to_payload(payload_info)
640 # add L2 header if not yet provided in packet_header
641 if packet_header.getlayer(0).name == 'Ethernet':
645 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
648 size = packet_sizes[i % len(packet_sizes)]
649 self.logger.debug("Packet size %d" % (size))
650 self.extend_packet(p, size)
651 # we need to store the packet with the automatic fields computed
652 # read back the dumped packet (with str())
653 # to force computing these fields
654 # probably other ways are possible
655 p = Ether(scapy.compat.raw(p))
656 payload_info.data = p.copy()
657 self.logger.debug(ppp("Created packet:", p))
659 self.logger.info("Done creating packets")
662 def send_and_verify_pkts(self, input, pkts, output, compare_func):
663 """Send packets and verify received packets using compare_func
665 :param input: ingress interface of DUT
666 :param pkts: list of packets to transmit
667 :param output: egress interface of DUT
668 :param compare_func: function to compare in and out packets
670 # add traffic stream to input interface
671 input.add_stream(pkts)
673 # enable capture on all interfaces
674 self.pg_enable_capture(self.pg_interfaces)
677 self.logger.info("Starting traffic")
681 self.logger.info("Getting packet capture")
682 capture = output.get_capture()
684 # assert nothing was captured on input interface
685 # input.assert_nothing_captured()
687 # verify captured packets
688 self.verify_captured_pkts(output, capture, compare_func)
690 def create_packet_header_IPv6(self):
691 """Create packet header: IPv6 header, UDP header
693 :param dst: IPv6 destination address
695 IPv6 source address is 1234::1
696 IPv6 destination address is 4321::1
697 UDP source port and destination port are 1234
700 p = (IPv6(src='1234::1', dst='4321::1') /
701 UDP(sport=1234, dport=1234))
704 def create_packet_header_IPv6_SRH_IPv6(self, sidlist, segleft):
705 """Create packet header: IPv6 encapsulated in SRv6:
706 IPv6 header with SRH, IPv6 header, UDP header
708 :param list sidlist: segment list of outer IPv6 SRH
709 :param int segleft: segments-left field of outer IPv6 SRH
711 Outer IPv6 source address is set to 5678::1
712 Outer IPv6 destination address is set to sidlist[segleft]
713 IPv6 source addresses is 1234::1
714 IPv6 destination address is 4321::1
715 UDP source port and destination port are 1234
718 p = (IPv6(src='5678::1', dst=sidlist[segleft]) /
719 IPv6ExtHdrSegmentRouting(addresses=sidlist,
720 segleft=segleft, nh=41) /
721 IPv6(src='1234::1', dst='4321::1') /
722 UDP(sport=1234, dport=1234))
725 def create_packet_header_IPv4(self):
726 """Create packet header: IPv4 header, UDP header
728 :param dst: IPv4 destination address
730 IPv4 source address is 123.1.1.1
731 IPv4 destination address is 124.1.1.1
732 UDP source port and destination port are 1234
735 p = (IP(src='123.1.1.1', dst='124.1.1.1') /
736 UDP(sport=1234, dport=1234))
739 def create_packet_header_IPv6_SRH_IPv4(self, sidlist, segleft):
740 """Create packet header: IPv4 encapsulated in SRv6:
741 IPv6 header with SRH, IPv4 header, UDP header
743 :param ipv4address dst: inner IPv4 destination address
744 :param list sidlist: segment list of outer IPv6 SRH
745 :param int segleft: segments-left field of outer IPv6 SRH
747 Outer IPv6 destination address is set to sidlist[segleft]
748 IPv6 source address is 1234::1
749 IPv4 source address is 123.1.1.1
750 IPv4 destination address is 124.1.1.1
751 UDP source port and destination port are 1234
754 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
755 IPv6ExtHdrSegmentRouting(addresses=sidlist,
756 segleft=segleft, nh=4) /
757 IP(src='123.1.1.1', dst='124.1.1.1') /
758 UDP(sport=1234, dport=1234))
761 def create_packet_header_L2(self, vlan=0):
762 """Create packet header: L2 header
764 :param vlan: if vlan!=0 then add 802.1q header
766 # Note: the dst addr ('00:55:44:33:22:11') is used in
767 # the compare function compare_rx_tx_packet_T_Encaps_L2
768 # to detect presence of L2 in SRH payload
769 p = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
773 p /= Dot1Q(vlan=vlan, type=etype)
778 def create_packet_header_IPv6_SRH_L2(self, sidlist, segleft, vlan=0):
779 """Create packet header: L2 encapsulated in SRv6:
780 IPv6 header with SRH, L2
782 :param list sidlist: segment list of outer IPv6 SRH
783 :param int segleft: segments-left field of outer IPv6 SRH
784 :param vlan: L2 vlan; if vlan!=0 then add 802.1q header
786 Outer IPv6 destination address is set to sidlist[segleft]
787 IPv6 source address is 1234::1
789 eth = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
793 eth /= Dot1Q(vlan=vlan, type=etype)
797 p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
798 IPv6ExtHdrSegmentRouting(addresses=sidlist,
799 segleft=segleft, nh=143) /
803 def get_payload_info(self, packet):
804 """ Extract the payload_info from the packet
806 # in most cases, payload_info is in packet[Raw]
807 # but packet[Raw] gives the complete payload
808 # (incl L2 header) for the T.Encaps L2 case
810 payload_info = self.payload_to_info(packet[Raw])
813 # remote L2 header from packet[Raw]:
814 # take packet[Raw], convert it to an Ether layer
815 # and then extract Raw from it
816 payload_info = self.payload_to_info(
817 Ether(scapy.compat.raw(packet[Raw]))[Raw])
821 def verify_captured_pkts(self, dst_if, capture, compare_func):
823 Verify captured packet stream for specified interface.
824 Compare ingress with egress packets using the specified compare fn
826 :param dst_if: egress interface of DUT
827 :param capture: captured packets
828 :param compare_func: function to compare in and out packet
830 self.logger.info("Verifying capture on interface %s using function %s"
831 % (dst_if.name, compare_func.__name__))
834 for i in self.pg_interfaces:
835 last_info[i.sw_if_index] = None
836 dst_sw_if_index = dst_if.sw_if_index
838 for packet in capture:
840 # extract payload_info from packet's payload
841 payload_info = self.get_payload_info(packet)
842 packet_index = payload_info.index
844 self.logger.debug("Verifying packet with index %d"
846 # packet should have arrived on the expected interface
847 self.assertEqual(payload_info.dst, dst_sw_if_index)
849 "Got packet on interface %s: src=%u (idx=%u)" %
850 (dst_if.name, payload_info.src, packet_index))
852 # search for payload_info with same src and dst if_index
853 # this will give us the transmitted packet
854 next_info = self.get_next_packet_info_for_interface2(
855 payload_info.src, dst_sw_if_index,
856 last_info[payload_info.src])
857 last_info[payload_info.src] = next_info
858 # next_info should not be None
859 self.assertTrue(next_info is not None)
860 # index of tx and rx packets should be equal
861 self.assertEqual(packet_index, next_info.index)
862 # data field of next_info contains the tx packet
863 txed_packet = next_info.data
865 self.logger.debug(ppp("Transmitted packet:",
866 txed_packet)) # ppp=Pretty Print Packet
868 self.logger.debug(ppp("Received packet:", packet))
870 # compare rcvd packet with expected packet using compare_func
871 compare_func(txed_packet, packet)
874 self.logger.error(ppp("Unexpected or invalid packet:", packet))
877 # have all expected packets arrived?
878 for i in self.pg_interfaces:
879 remaining_packet = self.get_next_packet_info_for_interface2(
880 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
881 self.assertTrue(remaining_packet is None,
882 "Interface %s: Packet expected from interface %s "
883 "didn't arrive" % (dst_if.name, i.name))
886 if __name__ == '__main__':
887 unittest.main(testRunner=VppTestRunner)