5 from socket import AF_INET6
7 from framework import VppTestCase, VppTestRunner
8 from vpp_ip import DpoProto
9 from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
10 from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \
11 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 Dynamic Proxy plugin Test Case """
28 super(TestSRv6, self).setUpClass()
31 """ Perform test setup before each test case.
33 super(TestSRv6, self).setUp()
35 # packet sizes, inclusive L2 overhead
36 self.pg_packet_sizes = [64, 512, 1518, 9018]
39 self.reset_packet_infos()
42 """ Clean up test setup after each test case.
44 self.teardown_interfaces()
46 super(TestSRv6, self).tearDown()
48 def configure_interface(self,
50 ipv6=False, ipv4=False,
51 ipv6_table_id=0, ipv4_table_id=0):
52 """ Configure interface.
53 :param ipv6: configure IPv6 on interface
54 :param ipv4: configure IPv4 on interface
55 :param ipv6_table_id: FIB table_id for IPv6
56 :param ipv4_table_id: FIB table_id for IPv4
58 self.logger.debug("Configuring interface %s" % (interface.name))
60 self.logger.debug("Configuring IPv6")
61 interface.set_table_ip6(ipv6_table_id)
62 interface.config_ip6()
63 interface.resolve_ndp(timeout=5)
65 self.logger.debug("Configuring IPv4")
66 interface.set_table_ip4(ipv4_table_id)
67 interface.config_ip4()
68 interface.resolve_arp()
71 def setup_interfaces(self, ipv6=[], ipv4=[],
72 ipv6_table_id=[], ipv4_table_id=[]):
73 """ Create and configure interfaces.
75 :param ipv6: list of interface IPv6 capabilities
76 :param ipv4: list of interface IPv4 capabilities
77 :param ipv6_table_id: list of intf IPv6 FIB table_ids
78 :param ipv4_table_id: list of intf IPv4 FIB table_ids
79 :returns: List of created interfaces.
81 # how many interfaces?
86 self.logger.debug("Creating and configuring %d interfaces" % (count))
88 # fill up ipv6 and ipv4 lists if needed
89 # not enabled (False) is the default
91 ipv6 += (count - len(ipv6)) * [False]
93 ipv4 += (count - len(ipv4)) * [False]
95 # fill up table_id lists if needed
96 # table_id 0 (global) is the default
97 if len(ipv6_table_id) < count:
98 ipv6_table_id += (count - len(ipv6_table_id)) * [0]
99 if len(ipv4_table_id) < count:
100 ipv4_table_id += (count - len(ipv4_table_id)) * [0]
102 # create 'count' pg interfaces
103 self.create_pg_interfaces(range(count))
105 # setup all interfaces
106 for i in range(count):
107 intf = self.pg_interfaces[i]
108 self.configure_interface(intf,
110 ipv6_table_id[i], ipv4_table_id[i])
113 self.logger.debug(self.vapi.cli("show ip6 neighbors"))
115 self.logger.debug(self.vapi.cli("show ip arp"))
116 self.logger.debug(self.vapi.cli("show interface"))
117 self.logger.debug(self.vapi.cli("show hardware"))
119 return self.pg_interfaces
121 def teardown_interfaces(self):
122 """ Unconfigure and bring down interface.
124 self.logger.debug("Tearing down interfaces")
125 # tear down all interfaces
126 # AFAIK they cannot be deleted
127 for i in self.pg_interfaces:
128 self.logger.debug("Tear down interface %s" % (i.name))
134 def test_SRv6_End_AD_IPv6(self):
135 """ Test SRv6 End.AD behavior with IPv6 traffic.
137 self.src_addr = 'a0::'
138 self.sid_list = ['a1::', 'a2::a6', 'a3::']
139 self.test_sid_index = 1
141 # send traffic to one destination interface
142 # source and destination interfaces are IPv6 only
143 self.setup_interfaces(ipv6=[True, True])
145 # configure route to next segment
146 route = VppIpRoute(self, self.sid_list[self.test_sid_index + 1], 128,
147 [VppRoutePath(self.pg0.remote_ip6,
148 self.pg0.sw_if_index,
149 proto=DpoProto.DPO_PROTO_IP6)],
151 route.add_vpp_config()
153 # configure SRv6 localSID behavior
154 cli_str = "sr localsid address " + \
155 self.sid_list[self.test_sid_index] + \
156 " behavior end.ad" + \
157 " nh " + self.pg1.remote_ip6 + \
158 " oif " + self.pg1.name + \
159 " iif " + self.pg1.name
160 self.vapi.cli(cli_str)
163 self.logger.debug(self.vapi.cli("show sr localsid"))
165 # send one packet per packet size
166 count = len(self.pg_packet_sizes)
168 # prepare IPv6 in SRv6 headers
169 packet_header1 = self.create_packet_header_IPv6_SRH_IPv6(
170 srcaddr=self.src_addr,
171 sidlist=self.sid_list[::-1],
172 segleft=len(self.sid_list) - self.test_sid_index - 1)
174 # generate packets (pg0->pg1)
175 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
176 self.pg_packet_sizes, count)
178 # send packets and verify received packets
179 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
180 self.compare_rx_tx_packet_End_AD_IPv6_out)
182 # log the localsid counters
183 self.logger.info(self.vapi.cli("show sr localsid"))
185 # prepare IPv6 header for returning packets
186 packet_header2 = self.create_packet_header_IPv6()
188 # generate returning packets (pg1->pg0)
189 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
190 self.pg_packet_sizes, count)
192 # send packets and verify received packets
193 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
194 self.compare_rx_tx_packet_End_AD_IPv6_in)
196 # log the localsid counters
197 self.logger.info(self.vapi.cli("show sr localsid"))
199 # remove SRv6 localSIDs
200 cli_str = "sr localsid del address " + \
201 self.sid_list[self.test_sid_index]
202 self.vapi.cli(cli_str)
205 self.teardown_interfaces()
207 def compare_rx_tx_packet_End_AD_IPv6_out(self, tx_pkt, rx_pkt):
208 """ Compare input and output packet after passing End.AD with IPv6
210 :param tx_pkt: transmitted packet
211 :param rx_pkt: received packet
214 # get first (outer) IPv6 header of rx'ed packet
215 rx_ip = rx_pkt.getlayer(IPv6)
217 tx_ip = tx_pkt.getlayer(IPv6)
218 tx_ip2 = tx_pkt.getlayer(IPv6, 2)
220 # verify if rx'ed packet has no SRH
221 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
223 # the whole rx_ip pkt should be equal to tx_ip2
224 # except for the hlim field
225 # -> adjust tx'ed hlim to expected hlim
226 tx_ip2.hlim = tx_ip2.hlim - 1
228 self.assertEqual(rx_ip, tx_ip2)
230 self.logger.debug("packet verification: SUCCESS")
232 def compare_rx_tx_packet_End_AD_IPv6_in(self, tx_pkt, rx_pkt):
233 """ Compare input and output packet after passing End.AD
235 :param tx_pkt: transmitted packet
236 :param rx_pkt: received packet
239 # get first (outer) IPv6 header of rx'ed packet
240 rx_ip = rx_pkt.getlayer(IPv6)
241 # received ip.src should be equal to SR Policy source
242 self.assertEqual(rx_ip.src, self.src_addr)
243 # received ip.dst should be equal to expected sidlist next segment
244 self.assertEqual(rx_ip.dst, self.sid_list[self.test_sid_index + 1])
246 # rx'ed packet should have SRH
247 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
250 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
251 # rx'ed seglist should be equal to SID-list in reversed order
252 self.assertEqual(rx_srh.addresses, self.sid_list[::-1])
253 # segleft should be equal to previous segleft value minus 1
254 self.assertEqual(rx_srh.segleft,
255 len(self.sid_list) - self.test_sid_index - 2)
256 # lastentry should be equal to the SID-list length minus 1
257 self.assertEqual(rx_srh.lastentry, len(self.sid_list) - 1)
259 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
260 # except for the hop-limit field
261 tx_ip = tx_pkt.getlayer(IPv6)
262 # -> update tx'ed hlim to the expected hlim
265 self.assertEqual(rx_srh.payload, tx_ip)
267 self.logger.debug("packet verification: SUCCESS")
269 def test_SRv6_End_AD_IPv4(self):
270 """ Test SRv6 End.AD behavior with IPv4 traffic.
272 self.src_addr = 'a0::'
273 self.sid_list = ['a1::', 'a2::a4', 'a3::']
274 self.test_sid_index = 1
276 # send traffic to one destination interface
277 # source and destination interfaces are IPv6 only
278 self.setup_interfaces(ipv6=[True, False], ipv4=[False, True])
280 # configure route to next segment
281 route = VppIpRoute(self, self.sid_list[self.test_sid_index + 1], 128,
282 [VppRoutePath(self.pg0.remote_ip6,
283 self.pg0.sw_if_index,
284 proto=DpoProto.DPO_PROTO_IP6)],
286 route.add_vpp_config()
288 # configure SRv6 localSID behavior
289 cli_str = "sr localsid address " + \
290 self.sid_list[self.test_sid_index] + \
291 " behavior end.ad" + \
292 " nh " + self.pg1.remote_ip4 + \
293 " oif " + self.pg1.name + \
294 " iif " + self.pg1.name
295 self.vapi.cli(cli_str)
298 self.logger.debug(self.vapi.cli("show sr localsid"))
300 # send one packet per packet size
301 count = len(self.pg_packet_sizes)
303 # prepare IPv4 in SRv6 headers
304 packet_header1 = self.create_packet_header_IPv6_SRH_IPv4(
305 srcaddr=self.src_addr,
306 sidlist=self.sid_list[::-1],
307 segleft=len(self.sid_list) - self.test_sid_index - 1)
309 # generate packets (pg0->pg1)
310 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
311 self.pg_packet_sizes, count)
313 # send packets and verify received packets
314 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
315 self.compare_rx_tx_packet_End_AD_IPv4_out)
317 # log the localsid counters
318 self.logger.info(self.vapi.cli("show sr localsid"))
320 # prepare IPv6 header for returning packets
321 packet_header2 = self.create_packet_header_IPv4()
323 # generate returning packets (pg1->pg0)
324 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
325 self.pg_packet_sizes, count)
327 # send packets and verify received packets
328 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
329 self.compare_rx_tx_packet_End_AD_IPv4_in)
331 # log the localsid counters
332 self.logger.info(self.vapi.cli("show sr localsid"))
334 # remove SRv6 localSIDs
335 cli_str = "sr localsid del address " + \
336 self.sid_list[self.test_sid_index]
337 self.vapi.cli(cli_str)
340 self.teardown_interfaces()
342 def compare_rx_tx_packet_End_AD_IPv4_out(self, tx_pkt, rx_pkt):
343 """ Compare input and output packet after passing End.AD with IPv4
345 :param tx_pkt: transmitted packet
346 :param rx_pkt: received packet
349 # get IPv4 header of rx'ed packet
350 rx_ip = rx_pkt.getlayer(IP)
352 tx_ip = tx_pkt.getlayer(IPv6)
353 tx_ip2 = tx_pkt.getlayer(IP)
355 # verify if rx'ed packet has no SRH
356 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
358 # the whole rx_ip pkt should be equal to tx_ip2
359 # except for the ttl field and ip checksum
360 # -> adjust tx'ed ttl to expected ttl
361 tx_ip2.ttl = tx_ip2.ttl - 1
362 # -> set tx'ed ip checksum to None and let scapy recompute
364 # read back the pkt (with str()) to force computing these fields
365 # probably other ways to accomplish this are possible
366 tx_ip2 = IP(str(tx_ip2))
368 self.assertEqual(rx_ip, tx_ip2)
370 self.logger.debug("packet verification: SUCCESS")
372 def compare_rx_tx_packet_End_AD_IPv4_in(self, tx_pkt, rx_pkt):
373 """ Compare input and output packet after passing End.AD
375 :param tx_pkt: transmitted packet
376 :param rx_pkt: received packet
379 # get first (outer) IPv6 header of rx'ed packet
380 rx_ip = rx_pkt.getlayer(IPv6)
381 # received ip.src should be equal to SR Policy source
382 self.assertEqual(rx_ip.src, self.src_addr)
383 # received ip.dst should be equal to expected sidlist next segment
384 self.assertEqual(rx_ip.dst, self.sid_list[self.test_sid_index + 1])
386 # rx'ed packet should have SRH
387 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
390 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
391 # rx'ed seglist should be equal to SID-list in reversed order
392 self.assertEqual(rx_srh.addresses, self.sid_list[::-1])
393 # segleft should be equal to previous segleft value minus 1
394 self.assertEqual(rx_srh.segleft,
395 len(self.sid_list) - self.test_sid_index - 2)
396 # lastentry should be equal to the SID-list length minus 1
397 self.assertEqual(rx_srh.lastentry, len(self.sid_list) - 1)
399 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
400 # except for the ttl field and ip checksum
401 tx_ip = tx_pkt.getlayer(IP)
402 # -> adjust tx'ed ttl to expected ttl
403 tx_ip.ttl = tx_ip.ttl - 1
404 # -> set tx'ed ip checksum to None and let scapy recompute
406 # -> read back the pkt (with str()) to force computing these fields
407 # probably other ways to accomplish this are possible
408 self.assertEqual(rx_srh.payload, IP(str(tx_ip)))
410 self.logger.debug("packet verification: SUCCESS")
412 def test_SRv6_End_AD_L2(self):
413 """ Test SRv6 End.AD behavior with L2 traffic.
415 self.src_addr = 'a0::'
416 self.sid_list = ['a1::', 'a2::a4', 'a3::']
417 self.test_sid_index = 1
419 # send traffic to one destination interface
420 # source and destination interfaces are IPv6 only
421 self.setup_interfaces(ipv6=[True, False])
423 # configure route to next segment
424 route = VppIpRoute(self, self.sid_list[self.test_sid_index + 1], 128,
425 [VppRoutePath(self.pg0.remote_ip6,
426 self.pg0.sw_if_index,
427 proto=DpoProto.DPO_PROTO_IP6)],
429 route.add_vpp_config()
431 # configure SRv6 localSID behavior
432 cli_str = "sr localsid address " + \
433 self.sid_list[self.test_sid_index] + \
434 " behavior end.ad" + \
435 " oif " + self.pg1.name + \
436 " iif " + self.pg1.name
437 self.vapi.cli(cli_str)
440 self.logger.debug(self.vapi.cli("show sr localsid"))
442 # send one packet per packet size
443 count = len(self.pg_packet_sizes)
445 # prepare L2 in SRv6 headers
446 packet_header1 = self.create_packet_header_IPv6_SRH_L2(
447 srcaddr=self.src_addr,
448 sidlist=self.sid_list[::-1],
449 segleft=len(self.sid_list) - self.test_sid_index - 1,
452 # generate packets (pg0->pg1)
453 pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
454 self.pg_packet_sizes, count)
456 # send packets and verify received packets
457 self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
458 self.compare_rx_tx_packet_End_AD_L2_out)
460 # log the localsid counters
461 self.logger.info(self.vapi.cli("show sr localsid"))
463 # prepare L2 header for returning packets
464 packet_header2 = self.create_packet_header_L2()
466 # generate returning packets (pg1->pg0)
467 pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
468 self.pg_packet_sizes, count)
470 # send packets and verify received packets
471 self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
472 self.compare_rx_tx_packet_End_AD_L2_in)
474 # log the localsid counters
475 self.logger.info(self.vapi.cli("show sr localsid"))
477 # remove SRv6 localSIDs
478 cli_str = "sr localsid del address " + \
479 self.sid_list[self.test_sid_index]
480 self.vapi.cli(cli_str)
483 self.teardown_interfaces()
485 def compare_rx_tx_packet_End_AD_L2_out(self, tx_pkt, rx_pkt):
486 """ Compare input and output packet after passing End.AD with L2
488 :param tx_pkt: transmitted packet
489 :param rx_pkt: received packet
492 # get IPv4 header of rx'ed packet
493 rx_eth = rx_pkt.getlayer(Ether)
495 tx_ip = tx_pkt.getlayer(IPv6)
496 # we can't just get the 2nd Ether layer
497 # get the Raw content and dissect it as Ether
498 tx_eth1 = Ether(str(tx_pkt[Raw]))
500 # verify if rx'ed packet has no SRH
501 self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
503 # the whole rx_eth pkt should be equal to tx_eth1
504 self.assertEqual(rx_eth, tx_eth1)
506 self.logger.debug("packet verification: SUCCESS")
508 def compare_rx_tx_packet_End_AD_L2_in(self, tx_pkt, rx_pkt):
509 """ Compare input and output packet after passing End.AD
511 :param tx_pkt: transmitted packet
512 :param rx_pkt: received packet
516 # get first (outer) IPv6 header of rx'ed packet
517 rx_ip = rx_pkt.getlayer(IPv6)
518 # received ip.src should be equal to SR Policy source
519 self.assertEqual(rx_ip.src, self.src_addr)
520 # received ip.dst should be equal to expected sidlist next segment
521 self.assertEqual(rx_ip.dst, self.sid_list[self.test_sid_index + 1])
523 # rx'ed packet should have SRH
524 self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
527 rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
528 # rx'ed seglist should be equal to SID-list in reversed order
529 self.assertEqual(rx_srh.addresses, self.sid_list[::-1])
530 # segleft should be equal to previous segleft value minus 1
531 self.assertEqual(rx_srh.segleft,
532 len(self.sid_list) - self.test_sid_index - 2)
533 # lastentry should be equal to the SID-list length minus 1
534 self.assertEqual(rx_srh.lastentry, len(self.sid_list) - 1)
536 # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
537 tx_ether = tx_pkt.getlayer(Ether)
538 self.assertEqual(Ether(str(rx_srh.payload)), tx_ether)
540 self.logger.debug("packet verification: SUCCESS")
542 def create_stream(self, src_if, dst_if, packet_header, packet_sizes,
544 """Create SRv6 input packet stream for defined interface.
546 :param VppInterface src_if: Interface to create packet stream for
547 :param VppInterface dst_if: destination interface of packet stream
548 :param packet_header: Layer3 scapy packet headers,
549 L2 is added when not provided,
550 Raw(payload) with packet_info is added
551 :param list packet_sizes: packet stream pckt sizes,sequentially applied
552 to packets in stream have
553 :param int count: number of packets in packet stream
554 :return: list of packets
556 self.logger.info("Creating packets")
558 for i in range(0, count - 1):
559 payload_info = self.create_packet_info(src_if, dst_if)
561 "Creating packet with index %d" % (payload_info.index))
562 payload = self.info_to_payload(payload_info)
563 # add L2 header if not yet provided in packet_header
564 if packet_header.getlayer(0).name == 'Ethernet':
565 p = packet_header / Raw(payload)
567 p = Ether(dst=src_if.local_mac, src=src_if.remote_mac) / \
568 packet_header / Raw(payload)
569 size = packet_sizes[i % len(packet_sizes)]
570 self.logger.debug("Packet size %d" % (size))
571 self.extend_packet(p, size)
572 # we need to store the packet with the automatic fields computed
573 # read back the dumped packet (with str())
574 # to force computing these fields
575 # probably other ways are possible
577 payload_info.data = p.copy()
578 self.logger.debug(ppp("Created packet:", p))
580 self.logger.info("Done creating packets")
583 def send_and_verify_pkts(self, input, pkts, output, compare_func):
584 """Send packets and verify received packets using compare_func
586 :param input: ingress interface of DUT
587 :param pkts: list of packets to transmit
588 :param output: egress interface of DUT
589 :param compare_func: function to compare in and out packets
591 # add traffic stream to input interface
592 input.add_stream(pkts)
594 # enable capture on all interfaces
595 self.pg_enable_capture(self.pg_interfaces)
598 self.logger.info("Starting traffic")
602 self.logger.info("Getting packet capture")
603 capture = output.get_capture()
605 # assert nothing was captured on input interface
606 # input.assert_nothing_captured()
608 # verify captured packets
609 self.verify_captured_pkts(output, capture, compare_func)
611 def create_packet_header_IPv6(self):
612 """Create packet header: IPv6 header, UDP header
614 :param dst: IPv6 destination address
616 IPv6 source address is 1234::1
617 IPv6 destination address is 4321::1
618 UDP source port and destination port are 1234
621 p = IPv6(src='1234::1', dst='4321::1') / UDP(sport=1234, dport=1234)
624 def create_packet_header_IPv6_SRH_IPv6(self, srcaddr, sidlist, segleft):
625 """Create packet header: IPv6 encapsulated in SRv6:
626 IPv6 header with SRH, IPv6 header, UDP header
628 :param int srcaddr: outer source address
629 :param list sidlist: segment list of outer IPv6 SRH
630 :param int segleft: segments-left field of outer IPv6 SRH
632 Outer IPv6 source address is set to srcaddr
633 Outer IPv6 destination address is set to sidlist[segleft]
634 Inner IPv6 source addresses is 1234::1
635 Inner IPv6 destination address is 4321::1
636 UDP source port and destination port are 1234
639 p = IPv6(src=srcaddr, dst=sidlist[segleft]) / \
640 IPv6ExtHdrSegmentRouting(addresses=sidlist,
641 segleft=segleft, nh=41) / \
642 IPv6(src='1234::1', dst='4321::1') / \
643 UDP(sport=1234, dport=1234)
646 def create_packet_header_IPv4(self):
647 """Create packet header: IPv4 header, UDP header
649 :param dst: IPv4 destination address
651 IPv4 source address is 123.1.1.1
652 IPv4 destination address is 124.1.1.1
653 UDP source port and destination port are 1234
656 p = IP(src='123.1.1.1', dst='124.1.1.1') / UDP(sport=1234, dport=1234)
659 def create_packet_header_IPv6_SRH_IPv4(self, srcaddr, sidlist, segleft):
660 """Create packet header: IPv4 encapsulated in SRv6:
661 IPv6 header with SRH, IPv4 header, UDP header
663 :param int srcaddr: outer source address
664 :param list sidlist: segment list of outer IPv6 SRH
665 :param int segleft: segments-left field of outer IPv6 SRH
667 Outer IPv6 source address is set to srcaddr
668 Outer IPv6 destination address is set to sidlist[segleft]
669 Inner IPv4 source address is 123.1.1.1
670 Inner IPv4 destination address is 124.1.1.1
671 UDP source port and destination port are 1234
674 p = IPv6(src=srcaddr, dst=sidlist[segleft]) / \
675 IPv6ExtHdrSegmentRouting(addresses=sidlist,
676 segleft=segleft, nh=4) / \
677 IP(src='123.1.1.1', dst='124.1.1.1') / \
678 UDP(sport=1234, dport=1234)
681 def create_packet_header_L2(self, vlan=0):
682 """Create packet header: L2 header
684 :param vlan: if vlan!=0 then add 802.1q header
686 # Note: the dst addr ('00:55:44:33:22:11') is used in
687 # the compare function compare_rx_tx_packet_T_Encaps_L2
688 # to detect presence of L2 in SRH payload
689 p = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
693 p /= Dot1Q(vlan=vlan, type=etype)
698 def create_packet_header_IPv6_SRH_L2(self, srcaddr, sidlist, segleft,
700 """Create packet header: L2 encapsulated in SRv6:
701 IPv6 header with SRH, L2
703 :param int srcaddr: IPv6 source address
704 :param list sidlist: segment list of outer IPv6 SRH
705 :param int segleft: segments-left field of outer IPv6 SRH
706 :param vlan: L2 vlan; if vlan!=0 then add 802.1q header
708 IPv6 source address is set to srcaddr
709 IPv6 destination address is set to sidlist[segleft]
711 eth = Ether(src='00:11:22:33:44:55', dst='00:55:44:33:22:11')
715 eth /= Dot1Q(vlan=vlan, type=etype)
719 p = IPv6(src=srcaddr, dst=sidlist[segleft]) / \
720 IPv6ExtHdrSegmentRouting(addresses=sidlist,
721 segleft=segleft, nh=59) / \
725 def get_payload_info(self, packet):
726 """ Extract the payload_info from the packet
728 # in most cases, payload_info is in packet[Raw]
729 # but packet[Raw] gives the complete payload
730 # (incl L2 header) for the T.Encaps L2 case
732 payload_info = self.payload_to_info(str(packet[Raw]))
735 # remote L2 header from packet[Raw]:
736 # take packet[Raw], convert it to an Ether layer
737 # and then extract Raw from it
738 payload_info = self.payload_to_info(
739 str(Ether(str(packet[Raw]))[Raw]))
743 def verify_captured_pkts(self, dst_if, capture, compare_func):
745 Verify captured packet stream for specified interface.
746 Compare ingress with egress packets using the specified compare fn
748 :param dst_if: egress interface of DUT
749 :param capture: captured packets
750 :param compare_func: function to compare in and out packet
752 self.logger.info("Verifying capture on interface %s using function %s"
753 % (dst_if.name, compare_func.func_name))
756 for i in self.pg_interfaces:
757 last_info[i.sw_if_index] = None
758 dst_sw_if_index = dst_if.sw_if_index
760 for packet in capture:
762 # extract payload_info from packet's payload
763 payload_info = self.get_payload_info(packet)
764 packet_index = payload_info.index
766 self.logger.debug("Verifying packet with index %d"
768 # packet should have arrived on the expected interface
769 self.assertEqual(payload_info.dst, dst_sw_if_index)
771 "Got packet on interface %s: src=%u (idx=%u)" %
772 (dst_if.name, payload_info.src, packet_index))
774 # search for payload_info with same src and dst if_index
775 # this will give us the transmitted packet
776 next_info = self.get_next_packet_info_for_interface2(
777 payload_info.src, dst_sw_if_index,
778 last_info[payload_info.src])
779 last_info[payload_info.src] = next_info
780 # next_info should not be None
781 self.assertTrue(next_info is not None)
782 # index of tx and rx packets should be equal
783 self.assertEqual(packet_index, next_info.index)
784 # data field of next_info contains the tx packet
785 txed_packet = next_info.data
787 self.logger.debug(ppp("Transmitted packet:",
788 txed_packet)) # ppp=Pretty Print Packet
790 self.logger.debug(ppp("Received packet:", packet))
792 # compare rcvd packet with expected packet using compare_func
793 compare_func(txed_packet, packet)
796 self.logger.error(ppp("Unexpected or invalid packet:", packet))
799 # have all expected packets arrived?
800 for i in self.pg_interfaces:
801 remaining_packet = self.get_next_packet_info_for_interface2(
802 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
803 self.assertTrue(remaining_packet is None,
804 "Interface %s: Packet expected from interface %s "
805 "didn't arrive" % (dst_if.name, i.name))
808 if __name__ == '__main__':
809 unittest.main(testRunner=VppTestRunner)