test/test_srv6_as.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import binascii
   5 from socket import AF_INET6
   6
   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
  11
  12 from scapy.packet import Raw
  13 from scapy.layers.l2 import Ether, Dot1Q
  14 from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting
  15 from scapy.layers.inet import IP, UDP
  16
  17 from scapy.utils import inet_pton, inet_ntop
  18
  19 from util import ppp
  20
  21
  22 class TestSRv6(VppTestCase):
  23     """ SRv6 Static Proxy plugin Test Case """
  24
  25     @classmethod
  26     def setUpClass(self):
  27         super(TestSRv6, self).setUpClass()
  28
  29     def setUp(self):
  30         """ Perform test setup before each test case.
  31         """
  32         super(TestSRv6, self).setUp()
  33
  34         # packet sizes, inclusive L2 overhead
  35         self.pg_packet_sizes = [64, 512, 1518, 9018]
  36
  37         # reset packet_infos
  38         self.reset_packet_infos()
  39
  40     def tearDown(self):
  41         """ Clean up test setup after each test case.
  42         """
  43         self.teardown_interfaces()
  44
  45         super(TestSRv6, self).tearDown()
  46
  47     def configure_interface(self,
  48                             interface,
  49                             ipv6=False, ipv4=False,
  50                             ipv6_table_id=0, ipv4_table_id=0):
  51         """ Configure interface.
  52         :param ipv6: configure IPv6 on interface
  53         :param ipv4: configure IPv4 on interface
  54         :param ipv6_table_id: FIB table_id for IPv6
  55         :param ipv4_table_id: FIB table_id for IPv4
  56         """
  57         self.logger.debug("Configuring interface %s" % (interface.name))
  58         if ipv6:
  59             self.logger.debug("Configuring IPv6")
  60             interface.set_table_ip6(ipv6_table_id)
  61             interface.config_ip6()
  62             interface.resolve_ndp(timeout=5)
  63         if ipv4:
  64             self.logger.debug("Configuring IPv4")
  65             interface.set_table_ip4(ipv4_table_id)
  66             interface.config_ip4()
  67             interface.resolve_arp()
  68         interface.admin_up()
  69
  70     def setup_interfaces(self, ipv6=[], ipv4=[],
  71                          ipv6_table_id=[], ipv4_table_id=[]):
  72         """ Create and configure interfaces.
  73
  74         :param ipv6: list of interface IPv6 capabilities
  75         :param ipv4: list of interface IPv4 capabilities
  76         :param ipv6_table_id: list of intf IPv6 FIB table_ids
  77         :param ipv4_table_id: list of intf IPv4 FIB table_ids
  78         :returns: List of created interfaces.
  79         """
  80         # how many interfaces?
  81         if len(ipv6):
  82             count = len(ipv6)
  83         else:
  84             count = len(ipv4)
  85         self.logger.debug("Creating and configuring %d interfaces" % (count))
  86
  87         # fill up ipv6 and ipv4 lists if needed
  88         # not enabled (False) is the default
  89         if len(ipv6) < count:
  90             ipv6 += (count - len(ipv6)) * [False]
  91         if len(ipv4) < count:
  92             ipv4 += (count - len(ipv4)) * [False]
  93
  94         # fill up table_id lists if needed
  95         # table_id 0 (global) is the default
  96         if len(ipv6_table_id) < count:
  97             ipv6_table_id += (count - len(ipv6_table_id)) * [0]
  98         if len(ipv4_table_id) < count:
  99             ipv4_table_id += (count - len(ipv4_table_id)) * [0]
 100
 101         # create 'count' pg interfaces
 102         self.create_pg_interfaces(range(count))
 103
 104         # setup all interfaces
 105         for i in range(count):
 106             intf = self.pg_interfaces[i]
 107             self.configure_interface(intf,
 108                                      ipv6[i], ipv4[i],
 109                                      ipv6_table_id[i], ipv4_table_id[i])
 110
 111         if any(ipv6):
 112             self.logger.debug(self.vapi.cli("show ip6 neighbors"))
 113         if any(ipv4):
 114             self.logger.debug(self.vapi.cli("show ip arp"))
 115         self.logger.debug(self.vapi.cli("show interface"))
 116         self.logger.debug(self.vapi.cli("show hardware"))
 117
 118         return self.pg_interfaces
 119
 120     def teardown_interfaces(self):
 121         """ Unconfigure and bring down interface.
 122         """
 123         self.logger.debug("Tearing down interfaces")
 124         # tear down all interfaces
 125         # AFAIK they cannot be deleted
 126         for i in self.pg_interfaces:
 127             self.logger.debug("Tear down interface %s" % (i.name))
 128             i.admin_down()
 129             i.unconfig()
 130             i.set_table_ip4(0)
 131             i.set_table_ip6(0)
 132
 133     def test_SRv6_End_AS_IPv6_noSRH(self):
 134         """ Test SRv6 End.AS behavior with IPv6 traffic and no SRH rewrite.
 135         """
 136         self.run_SRv6_End_AS_IPv6(
 137             sid_list=['a1::', 'a2::a6', 'a3::'],
 138             test_sid_index=1,
 139             rewrite_src_addr='a2::')
 140
 141     def test_SRv6_End_AS_IPv6_SRH(self):
 142         """ Test SRv6 End.AS behavior with IPv6 traffic and SRH rewrite.
 143         """
 144         self.run_SRv6_End_AS_IPv6(
 145             sid_list=['a1::a6', 'a2::', 'a3::'],
 146             test_sid_index=0,
 147             rewrite_src_addr='a1::')
 148
 149     def test_SRv6_End_AS_IPv4_noSRH(self):
 150         """ Test SRv6 End.AS behavior with IPv4 traffic and no SRH rewrite.
 151         """
 152         self.run_SRv6_End_AS_IPv4(
 153             sid_list=['a1::', 'a2::a6', 'a3::'],
 154             test_sid_index=1,
 155             rewrite_src_addr='a2::')
 156
 157     def test_SRv6_End_AS_IPv4_SRH(self):
 158         """ Test SRv6 End.AS behavior with IPv4 traffic and SRH rewrite.
 159         """
 160         self.run_SRv6_End_AS_IPv4(
 161             sid_list=['a1::a6', 'a2::', 'a3::'],
 162             test_sid_index=0,
 163             rewrite_src_addr='a1::')
 164
 165     def run_SRv6_End_AS_IPv6(self, sid_list, test_sid_index, rewrite_src_addr):
 166         """ Run SRv6 End.AS test with IPv6 traffic.
 167         """
 168         self.rewrite_src_addr = rewrite_src_addr
 169         self.rewrite_sid_list = sid_list[test_sid_index + 1::]
 170
 171         # send traffic to one destination interface
 172         # source and destination interfaces are IPv6 only
 173         self.setup_interfaces(ipv6=[True, True])
 174
 175         # configure route to next segment
 176         route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
 177                            [VppRoutePath(self.pg0.remote_ip6,
 178                                          self.pg0.sw_if_index,
 179                                          proto=DpoProto.DPO_PROTO_IP6)],
 180                            is_ip6=1)
 181         route.add_vpp_config()
 182
 183         # configure SRv6 localSID behavior
 184         cli_str = "sr localsid address " + sid_list[test_sid_index] \
 185             + " behavior end.as" \
 186             + " nh " + self.pg1.remote_ip6 \
 187             + " oif " + self.pg1.name \
 188             + " iif " + self.pg1.name \
 189             + " src " + self.rewrite_src_addr
 190         for s in self.rewrite_sid_list:
 191             cli_str += " next " + s
 192         self.vapi.cli(cli_str)
 193
 194         # log the localsids
 195         self.logger.debug(self.vapi.cli("show sr localsid"))
 196
 197         # send one packet per packet size
 198         count = len(self.pg_packet_sizes)
 199
 200         # prepare IPv6 in SRv6 headers
 201         packet_header1 = self.create_packet_header_IPv6_SRH_IPv6(
 202                         sidlist=sid_list[::-1],
 203                         segleft=len(sid_list) - test_sid_index - 1)
 204
 205         # generate packets (pg0->pg1)
 206         pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
 207                                    self.pg_packet_sizes, count)
 208
 209         # send packets and verify received packets
 210         self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
 211                                   self.compare_rx_tx_packet_End_AS_IPv6_out)
 212
 213         # log the localsid counters
 214         self.logger.info(self.vapi.cli("show sr localsid"))
 215
 216         # prepare IPv6 header for returning packets
 217         packet_header2 = self.create_packet_header_IPv6()
 218
 219         # generate returning packets (pg1->pg0)
 220         pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
 221                                    self.pg_packet_sizes, count)
 222
 223         # send packets and verify received packets
 224         self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
 225                                   self.compare_rx_tx_packet_End_AS_IPv6_in)
 226
 227         # log the localsid counters
 228         self.logger.info(self.vapi.cli("show sr localsid"))
 229
 230         # remove SRv6 localSIDs
 231         self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
 232
 233         # cleanup interfaces
 234         self.teardown_interfaces()
 235
 236     def run_SRv6_End_AS_IPv4(self, sid_list, test_sid_index, rewrite_src_addr):
 237         """ Run SRv6 End.AS test with IPv4 traffic.
 238         """
 239         self.rewrite_src_addr = rewrite_src_addr
 240         self.rewrite_sid_list = sid_list[test_sid_index + 1::]
 241
 242         # send traffic to one destination interface
 243         # source and destination interfaces are IPv6 only
 244         self.setup_interfaces(ipv6=[True, False], ipv4=[True, True])
 245
 246         # configure route to next segment
 247         route = VppIpRoute(self, sid_list[test_sid_index + 1], 128,
 248                            [VppRoutePath(self.pg0.remote_ip6,
 249                                          self.pg0.sw_if_index,
 250                                          proto=DpoProto.DPO_PROTO_IP6)],
 251                            is_ip6=1)
 252         route.add_vpp_config()
 253
 254         # configure SRv6 localSID behavior
 255         cli_str = "sr localsid address " + sid_list[test_sid_index] \
 256             + " behavior end.as" \
 257             + " nh " + self.pg1.remote_ip4 \
 258             + " oif " + self.pg1.name \
 259             + " iif " + self.pg1.name \
 260             + " src " + self.rewrite_src_addr
 261         for s in self.rewrite_sid_list:
 262             cli_str += " next " + s
 263         self.vapi.cli(cli_str)
 264
 265         # log the localsids
 266         self.logger.debug(self.vapi.cli("show sr localsid"))
 267
 268         # send one packet per packet size
 269         count = len(self.pg_packet_sizes)
 270
 271         # prepare IPv4 in SRv6 headers
 272         packet_header1 = self.create_packet_header_IPv6_SRH_IPv4(
 273                         sidlist=sid_list[::-1],
 274                         segleft=len(sid_list) - test_sid_index - 1)
 275
 276         # generate packets (pg0->pg1)
 277         pkts1 = self.create_stream(self.pg0, self.pg1, packet_header1,
 278                                    self.pg_packet_sizes, count)
 279
 280         # send packets and verify received packets
 281         self.send_and_verify_pkts(self.pg0, pkts1, self.pg1,
 282                                   self.compare_rx_tx_packet_End_AS_IPv4_out)
 283
 284         # log the localsid counters
 285         self.logger.info(self.vapi.cli("show sr localsid"))
 286
 287         # prepare IPv6 header for returning packets
 288         packet_header2 = self.create_packet_header_IPv4()
 289
 290         # generate returning packets (pg1->pg0)
 291         pkts2 = self.create_stream(self.pg1, self.pg0, packet_header2,
 292                                    self.pg_packet_sizes, count)
 293
 294         # send packets and verify received packets
 295         self.send_and_verify_pkts(self.pg1, pkts2, self.pg0,
 296                                   self.compare_rx_tx_packet_End_AS_IPv4_in)
 297
 298         # log the localsid counters
 299         self.logger.info(self.vapi.cli("show sr localsid"))
 300
 301         # remove SRv6 localSIDs
 302         self.vapi.cli("sr localsid del address " + sid_list[test_sid_index])
 303
 304         # cleanup interfaces
 305         self.teardown_interfaces()
 306
 307     def compare_rx_tx_packet_End_AS_IPv6_in(self, tx_pkt, rx_pkt):
 308         """ Compare input and output packet after passing End.AS
 309
 310         :param tx_pkt: transmitted packet
 311         :param rx_pkt: received packet
 312         """
 313
 314         # get first (outer) IPv6 header of rx'ed packet
 315         rx_ip = rx_pkt.getlayer(IPv6)
 316         rx_srh = None
 317
 318         tx_ip = tx_pkt.getlayer(IPv6)
 319
 320         # expected segment-list (SRH order)
 321         tx_seglist = self.rewrite_sid_list[::-1]
 322
 323         # received ip.src should be equal to SR Policy source
 324         self.assertEqual(rx_ip.src, self.rewrite_src_addr)
 325         # received ip.dst should be equal to expected sidlist[lastentry]
 326         self.assertEqual(rx_ip.dst, tx_seglist[-1])
 327
 328         if len(tx_seglist) > 1:
 329             # rx'ed packet should have SRH
 330             self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
 331             # get SRH
 332             rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
 333             # rx'ed seglist should be equal to expected seglist
 334             self.assertEqual(rx_srh.addresses, tx_seglist)
 335             # segleft should be equal to size expected seglist-1
 336             self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
 337             # segleft should be equal to lastentry
 338             self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
 339             # get payload
 340             payload = rx_srh.payload
 341         else:
 342             # rx'ed packet should NOT have SRH
 343             self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
 344             # get payload
 345             payload = rx_ip.payload
 346
 347         # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
 348         # except for the hop-limit field
 349         #   -> update tx'ed hlim to the expected hlim
 350         tx_ip.hlim = tx_ip.hlim - 1
 351
 352         self.assertEqual(payload, tx_ip)
 353
 354         self.logger.debug("packet verification: SUCCESS")
 355
 356     def compare_rx_tx_packet_End_AS_IPv4_in(self, tx_pkt, rx_pkt):
 357         """ Compare input and output packet after passing End.AS
 358
 359         :param tx_pkt: transmitted packet
 360         :param rx_pkt: received packet
 361         """
 362
 363         # get first (outer) IPv6 header of rx'ed packet
 364         rx_ip = rx_pkt.getlayer(IPv6)
 365         rx_srh = None
 366
 367         tx_ip = tx_pkt.getlayer(IP)
 368
 369         # expected segment-list (SRH order)
 370         tx_seglist = self.rewrite_sid_list[::-1]
 371
 372         # received ip.src should be equal to SR Policy source
 373         self.assertEqual(rx_ip.src, self.rewrite_src_addr)
 374         # received ip.dst should be equal to expected sidlist[lastentry]
 375         self.assertEqual(rx_ip.dst, tx_seglist[-1])
 376
 377         if len(tx_seglist) > 1:
 378             # rx'ed packet should have SRH and IPv4 header
 379             self.assertTrue(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
 380             self.assertTrue(rx_ip.payload.haslayer(IP))
 381             # get SRH
 382             rx_srh = rx_pkt.getlayer(IPv6ExtHdrSegmentRouting)
 383             # rx'ed seglist should be equal to seglist
 384             self.assertEqual(rx_srh.addresses, tx_seglist)
 385             # segleft should be equal to size seglist-1
 386             self.assertEqual(rx_srh.segleft, len(tx_seglist)-1)
 387             # segleft should be equal to lastentry
 388             self.assertEqual(rx_srh.segleft, rx_srh.lastentry)
 389             payload = rx_srh.payload
 390         else:
 391             # rx'ed packet should NOT have SRH
 392             self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
 393             # get payload
 394             payload = rx_ip.payload
 395
 396         # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt
 397         # except for the ttl field and ip checksum
 398         #   -> adjust tx'ed ttl to expected ttl
 399         tx_ip.ttl = tx_ip.ttl - 1
 400         #   -> set tx'ed ip checksum to None and let scapy recompute
 401         tx_ip.chksum = None
 402         # read back the pkt (with str()) to force computing these fields
 403         # probably other ways to accomplish this are possible
 404         tx_ip = IP(str(tx_ip))
 405
 406         self.assertEqual(payload, tx_ip)
 407
 408         self.logger.debug("packet verification: SUCCESS")
 409
 410     def compare_rx_tx_packet_End_AS_IPv6_out(self, tx_pkt, rx_pkt):
 411         """ Compare input and output packet after passing End.AS with IPv6
 412
 413         :param tx_pkt: transmitted packet
 414         :param rx_pkt: received packet
 415         """
 416
 417         # get first (outer) IPv6 header of rx'ed packet
 418         rx_ip = rx_pkt.getlayer(IPv6)
 419
 420         tx_ip = tx_pkt.getlayer(IPv6)
 421         tx_ip2 = tx_pkt.getlayer(IPv6, 2)
 422
 423         # verify if rx'ed packet has no SRH
 424         self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
 425
 426         # the whole rx_ip pkt should be equal to tx_ip2
 427         # except for the hlim field
 428         #   -> adjust tx'ed hlim to expected hlim
 429         tx_ip2.hlim = tx_ip2.hlim - 1
 430
 431         self.assertEqual(rx_ip, tx_ip2)
 432
 433         self.logger.debug("packet verification: SUCCESS")
 434
 435     def compare_rx_tx_packet_End_AS_IPv4_out(self, tx_pkt, rx_pkt):
 436         """ Compare input and output packet after passing End.AS with IPv4
 437
 438         :param tx_pkt: transmitted packet
 439         :param rx_pkt: received packet
 440         """
 441
 442         # get IPv4 header of rx'ed packet
 443         rx_ip = rx_pkt.getlayer(IP)
 444
 445         tx_ip = tx_pkt.getlayer(IPv6)
 446         tx_ip2 = tx_pkt.getlayer(IP)
 447
 448         # verify if rx'ed packet has no SRH
 449         self.assertFalse(rx_pkt.haslayer(IPv6ExtHdrSegmentRouting))
 450
 451         # the whole rx_ip pkt should be equal to tx_ip2
 452         # except for the ttl field and ip checksum
 453         #   -> adjust tx'ed ttl to expected ttl
 454         tx_ip2.ttl = tx_ip2.ttl - 1
 455         #   -> set tx'ed ip checksum to None and let scapy recompute
 456         tx_ip2.chksum = None
 457         # read back the pkt (with str()) to force computing these fields
 458         # probably other ways to accomplish this are possible
 459         tx_ip2 = IP(str(tx_ip2))
 460
 461         self.assertEqual(rx_ip, tx_ip2)
 462
 463         self.logger.debug("packet verification: SUCCESS")
 464
 465     def create_stream(self, src_if, dst_if, packet_header, packet_sizes,
 466                       count):
 467         """Create SRv6 input packet stream for defined interface.
 468
 469         :param VppInterface src_if: Interface to create packet stream for
 470         :param VppInterface dst_if: destination interface of packet stream
 471         :param packet_header: Layer3 scapy packet headers,
 472                 L2 is added when not provided,
 473                 Raw(payload) with packet_info is added
 474         :param list packet_sizes: packet stream pckt sizes,sequentially applied
 475                to packets in stream have
 476         :param int count: number of packets in packet stream
 477         :return: list of packets
 478         """
 479         self.logger.info("Creating packets")
 480         pkts = []
 481         for i in range(0, count-1):
 482             payload_info = self.create_packet_info(src_if, dst_if)
 483             self.logger.debug(
 484                 "Creating packet with index %d" % (payload_info.index))
 485             payload = self.info_to_payload(payload_info)
 486             # add L2 header if not yet provided in packet_header
 487             if packet_header.getlayer(0).name == 'Ethernet':
 488                 p = (packet_header /
 489                      Raw(payload))
 490             else:
 491                 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 492                      packet_header /
 493                      Raw(payload))
 494             size = packet_sizes[i % len(packet_sizes)]
 495             self.logger.debug("Packet size %d" % (size))
 496             self.extend_packet(p, size)
 497             # we need to store the packet with the automatic fields computed
 498             # read back the dumped packet (with str())
 499             # to force computing these fields
 500             # probably other ways are possible
 501             p = Ether(str(p))
 502             payload_info.data = p.copy()
 503             self.logger.debug(ppp("Created packet:", p))
 504             pkts.append(p)
 505         self.logger.info("Done creating packets")
 506         return pkts
 507
 508     def send_and_verify_pkts(self, input, pkts, output, compare_func):
 509         """Send packets and verify received packets using compare_func
 510
 511         :param input: ingress interface of DUT
 512         :param pkts: list of packets to transmit
 513         :param output: egress interface of DUT
 514         :param compare_func: function to compare in and out packets
 515         """
 516         # add traffic stream to input interface
 517         input.add_stream(pkts)
 518
 519         # enable capture on all interfaces
 520         self.pg_enable_capture(self.pg_interfaces)
 521
 522         # start traffic
 523         self.logger.info("Starting traffic")
 524         self.pg_start()
 525
 526         # get output capture
 527         self.logger.info("Getting packet capture")
 528         capture = output.get_capture()
 529
 530         # assert nothing was captured on input interface
 531         # input.assert_nothing_captured()
 532
 533         # verify captured packets
 534         self.verify_captured_pkts(output, capture, compare_func)
 535
 536     def create_packet_header_IPv6(self):
 537         """Create packet header: IPv6 header, UDP header
 538
 539         :param dst: IPv6 destination address
 540
 541         IPv6 source address is 1234::1
 542         IPv6 destination address is 4321::1
 543         UDP source port and destination port are 1234
 544         """
 545
 546         p = (IPv6(src='1234::1', dst='4321::1') /
 547              UDP(sport=1234, dport=1234))
 548         return p
 549
 550     def create_packet_header_IPv6_SRH_IPv6(self, sidlist, segleft):
 551         """Create packet header: IPv6 encapsulated in SRv6:
 552         IPv6 header with SRH, IPv6 header, UDP header
 553
 554         :param list sidlist: segment list of outer IPv6 SRH
 555         :param int segleft: segments-left field of outer IPv6 SRH
 556
 557         Outer IPv6 source address is set to 5678::1
 558         Outer IPv6 destination address is set to sidlist[segleft]
 559         IPv6 source addresses is 1234::1
 560         IPv6 destination address is 4321::1
 561         UDP source port and destination port are 1234
 562         """
 563
 564         p = (IPv6(src='5678::1', dst=sidlist[segleft]) /
 565              IPv6ExtHdrSegmentRouting(addresses=sidlist,
 566                                       segleft=segleft, nh=41) /
 567              IPv6(src='1234::1', dst='4321::1') /
 568              UDP(sport=1234, dport=1234))
 569         return p
 570
 571     def create_packet_header_IPv4(self):
 572         """Create packet header: IPv4 header, UDP header
 573
 574         :param dst: IPv4 destination address
 575
 576         IPv4 source address is 123.1.1.1
 577         IPv4 destination address is 124.1.1.1
 578         UDP source port and destination port are 1234
 579         """
 580
 581         p = (IP(src='123.1.1.1', dst='124.1.1.1') /
 582              UDP(sport=1234, dport=1234))
 583         return p
 584
 585     def create_packet_header_IPv6_SRH_IPv4(self, sidlist, segleft):
 586         """Create packet header: IPv4 encapsulated in SRv6:
 587         IPv6 header with SRH, IPv4 header, UDP header
 588
 589         :param ipv4address dst: inner IPv4 destination address
 590         :param list sidlist: segment list of outer IPv6 SRH
 591         :param int segleft: segments-left field of outer IPv6 SRH
 592
 593         Outer IPv6 destination address is set to sidlist[segleft]
 594         IPv6 source address is 1234::1
 595         IPv4 source address is 123.1.1.1
 596         IPv4 destination address is 124.1.1.1
 597         UDP source port and destination port are 1234
 598         """
 599
 600         p = (IPv6(src='1234::1', dst=sidlist[segleft]) /
 601              IPv6ExtHdrSegmentRouting(addresses=sidlist,
 602                                       segleft=segleft, nh=4) /
 603              IP(src='123.1.1.1', dst='124.1.1.1') /
 604              UDP(sport=1234, dport=1234))
 605         return p
 606
 607     def get_payload_info(self, packet):
 608         """ Extract the payload_info from the packet
 609         """
 610         # in most cases, payload_info is in packet[Raw]
 611         # but packet[Raw] gives the complete payload
 612         # (incl L2 header) for the T.Encaps L2 case
 613         try:
 614             payload_info = self.payload_to_info(str(packet[Raw]))
 615
 616         except:
 617             # remote L2 header from packet[Raw]:
 618             # take packet[Raw], convert it to an Ether layer
 619             # and then extract Raw from it
 620             payload_info = self.payload_to_info(
 621                 str(Ether(str(packet[Raw]))[Raw]))
 622
 623         return payload_info
 624
 625     def verify_captured_pkts(self, dst_if, capture, compare_func):
 626         """
 627         Verify captured packet stream for specified interface.
 628         Compare ingress with egress packets using the specified compare fn
 629
 630         :param dst_if: egress interface of DUT
 631         :param capture: captured packets
 632         :param compare_func: function to compare in and out packet
 633         """
 634         self.logger.info("Verifying capture on interface %s using function %s"
 635                          % (dst_if.name, compare_func.func_name))
 636
 637         last_info = dict()
 638         for i in self.pg_interfaces:
 639             last_info[i.sw_if_index] = None
 640         dst_sw_if_index = dst_if.sw_if_index
 641
 642         for packet in capture:
 643             try:
 644                 # extract payload_info from packet's payload
 645                 payload_info = self.get_payload_info(packet)
 646                 packet_index = payload_info.index
 647
 648                 self.logger.debug("Verifying packet with index %d"
 649                                   % (packet_index))
 650                 # packet should have arrived on the expected interface
 651                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 652                 self.logger.debug(
 653                     "Got packet on interface %s: src=%u (idx=%u)" %
 654                     (dst_if.name, payload_info.src, packet_index))
 655
 656                 # search for payload_info with same src and dst if_index
 657                 # this will give us the transmitted packet
 658                 next_info = self.get_next_packet_info_for_interface2(
 659                     payload_info.src, dst_sw_if_index,
 660                     last_info[payload_info.src])
 661                 last_info[payload_info.src] = next_info
 662                 # next_info should not be None
 663                 self.assertTrue(next_info is not None)
 664                 # index of tx and rx packets should be equal
 665                 self.assertEqual(packet_index, next_info.index)
 666                 # data field of next_info contains the tx packet
 667                 txed_packet = next_info.data
 668
 669                 self.logger.debug(ppp("Transmitted packet:",
 670                                       txed_packet))  # ppp=Pretty Print Packet
 671
 672                 self.logger.debug(ppp("Received packet:", packet))
 673
 674                 # compare rcvd packet with expected packet using compare_func
 675                 compare_func(txed_packet, packet)
 676
 677             except:
 678                 print packet.command()
 679                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 680                 raise
 681
 682         # have all expected packets arrived?
 683         for i in self.pg_interfaces:
 684             remaining_packet = self.get_next_packet_info_for_interface2(
 685                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 686             self.assertTrue(remaining_packet is None,
 687                             "Interface %s: Packet expected from interface %s "
 688                             "didn't arrive" % (dst_if.name, i.name))
 689
 690
 691 if __name__ == '__main__':
 692     unittest.main(testRunner=VppTestRunner)