test/template_ipsec.py

   1 import unittest
   2 import socket
   3 import struct
   4
   5 from scapy.layers.inet import IP, ICMP, TCP, UDP
   6 from scapy.layers.ipsec import SecurityAssociation, ESP
   7 from scapy.layers.l2 import Ether
   8 from scapy.packet import raw, Raw
   9 from scapy.layers.inet6 import (
  10     IPv6,
  11     ICMPv6EchoRequest,
  12     IPv6ExtHdrHopByHop,
  13     IPv6ExtHdrFragment,
  14     IPv6ExtHdrDestOpt,
  15 )
  16
  17
  18 from framework import VppTestCase, VppTestRunner
  19 from util import ppp, reassemble4, fragment_rfc791, fragment_rfc8200
  20 from vpp_papi import VppEnum
  21
  22 from vpp_ipsec import VppIpsecSpd, VppIpsecSpdEntry, VppIpsecSpdItfBinding
  23 from ipaddress import ip_address
  24 from re import search
  25 from os import popen
  26
  27
  28 class IPsecIPv4Params:
  29     addr_type = socket.AF_INET
  30     addr_any = "0.0.0.0"
  31     addr_bcast = "255.255.255.255"
  32     addr_len = 32
  33     is_ipv6 = 0
  34
  35     def __init__(self):
  36         self.remote_tun_if_host = "1.1.1.1"
  37         self.remote_tun_if_host6 = "1111::1"
  38
  39         self.scapy_tun_sa_id = 100
  40         self.scapy_tun_spi = 1000
  41         self.vpp_tun_sa_id = 200
  42         self.vpp_tun_spi = 2000
  43
  44         self.scapy_tra_sa_id = 300
  45         self.scapy_tra_spi = 3000
  46         self.vpp_tra_sa_id = 400
  47         self.vpp_tra_spi = 4000
  48
  49         self.outer_hop_limit = 64
  50         self.inner_hop_limit = 255
  51         self.outer_flow_label = 0
  52         self.inner_flow_label = 0x12345
  53
  54         self.auth_algo_vpp_id = (
  55             VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA1_96
  56         )
  57         self.auth_algo = "HMAC-SHA1-96"  # scapy name
  58         self.auth_key = b"C91KUR9GYMm5GfkEvNjX"
  59
  60         self.crypt_algo_vpp_id = (
  61             VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_128
  62         )
  63         self.crypt_algo = "AES-CBC"  # scapy name
  64         self.crypt_key = b"JPjyOWBeVEQiMe7h"
  65         self.salt = 0
  66         self.flags = 0
  67         self.nat_header = None
  68         self.tun_flags = (
  69             VppEnum.vl_api_tunnel_encap_decap_flags_t.TUNNEL_API_ENCAP_DECAP_FLAG_NONE
  70         )
  71         self.dscp = 0
  72         self.async_mode = False
  73
  74
  75 class IPsecIPv6Params:
  76     addr_type = socket.AF_INET6
  77     addr_any = "0::0"
  78     addr_bcast = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"
  79     addr_len = 128
  80     is_ipv6 = 1
  81
  82     def __init__(self):
  83         self.remote_tun_if_host = "1111:1111:1111:1111:1111:1111:1111:1111"
  84         self.remote_tun_if_host4 = "1.1.1.1"
  85
  86         self.scapy_tun_sa_id = 500
  87         self.scapy_tun_spi = 3001
  88         self.vpp_tun_sa_id = 600
  89         self.vpp_tun_spi = 3000
  90
  91         self.scapy_tra_sa_id = 700
  92         self.scapy_tra_spi = 4001
  93         self.vpp_tra_sa_id = 800
  94         self.vpp_tra_spi = 4000
  95
  96         self.outer_hop_limit = 64
  97         self.inner_hop_limit = 255
  98         self.outer_flow_label = 0
  99         self.inner_flow_label = 0x12345
 100
 101         self.auth_algo_vpp_id = (
 102             VppEnum.vl_api_ipsec_integ_alg_t.IPSEC_API_INTEG_ALG_SHA1_96
 103         )
 104         self.auth_algo = "HMAC-SHA1-96"  # scapy name
 105         self.auth_key = b"C91KUR9GYMm5GfkEvNjX"
 106
 107         self.crypt_algo_vpp_id = (
 108             VppEnum.vl_api_ipsec_crypto_alg_t.IPSEC_API_CRYPTO_ALG_AES_CBC_128
 109         )
 110         self.crypt_algo = "AES-CBC"  # scapy name
 111         self.crypt_key = b"JPjyOWBeVEQiMe7h"
 112         self.salt = 0
 113         self.flags = 0
 114         self.nat_header = None
 115         self.tun_flags = (
 116             VppEnum.vl_api_tunnel_encap_decap_flags_t.TUNNEL_API_ENCAP_DECAP_FLAG_NONE
 117         )
 118         self.dscp = 0
 119         self.async_mode = False
 120
 121
 122 def mk_scapy_crypt_key(p):
 123     if p.crypt_algo in ("AES-GCM", "AES-CTR", "AES-NULL-GMAC"):
 124         return p.crypt_key + struct.pack("!I", p.salt)
 125     else:
 126         return p.crypt_key
 127
 128
 129 def config_tun_params(p, encryption_type, tun_if):
 130     ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
 131     esn_en = bool(
 132         p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
 133     )
 134     p.tun_dst = tun_if.remote_addr[p.addr_type]
 135     p.tun_src = tun_if.local_addr[p.addr_type]
 136     crypt_key = mk_scapy_crypt_key(p)
 137     p.scapy_tun_sa = SecurityAssociation(
 138         encryption_type,
 139         spi=p.scapy_tun_spi,
 140         crypt_algo=p.crypt_algo,
 141         crypt_key=crypt_key,
 142         auth_algo=p.auth_algo,
 143         auth_key=p.auth_key,
 144         tunnel_header=ip_class_by_addr_type[p.addr_type](src=p.tun_dst, dst=p.tun_src),
 145         nat_t_header=p.nat_header,
 146         esn_en=esn_en,
 147     )
 148     p.vpp_tun_sa = SecurityAssociation(
 149         encryption_type,
 150         spi=p.vpp_tun_spi,
 151         crypt_algo=p.crypt_algo,
 152         crypt_key=crypt_key,
 153         auth_algo=p.auth_algo,
 154         auth_key=p.auth_key,
 155         tunnel_header=ip_class_by_addr_type[p.addr_type](dst=p.tun_dst, src=p.tun_src),
 156         nat_t_header=p.nat_header,
 157         esn_en=esn_en,
 158     )
 159
 160
 161 def config_tra_params(p, encryption_type):
 162     esn_en = bool(
 163         p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
 164     )
 165     crypt_key = mk_scapy_crypt_key(p)
 166     p.scapy_tra_sa = SecurityAssociation(
 167         encryption_type,
 168         spi=p.scapy_tra_spi,
 169         crypt_algo=p.crypt_algo,
 170         crypt_key=crypt_key,
 171         auth_algo=p.auth_algo,
 172         auth_key=p.auth_key,
 173         nat_t_header=p.nat_header,
 174         esn_en=esn_en,
 175     )
 176     p.vpp_tra_sa = SecurityAssociation(
 177         encryption_type,
 178         spi=p.vpp_tra_spi,
 179         crypt_algo=p.crypt_algo,
 180         crypt_key=crypt_key,
 181         auth_algo=p.auth_algo,
 182         auth_key=p.auth_key,
 183         nat_t_header=p.nat_header,
 184         esn_en=esn_en,
 185     )
 186
 187
 188 class TemplateIpsec(VppTestCase):
 189     """
 190     TRANSPORT MODE::
 191
 192          ------   encrypt   ---
 193         |tra_if| <-------> |VPP|
 194          ------   decrypt   ---
 195
 196     TUNNEL MODE::
 197
 198          ------   encrypt   ---   plain   ---
 199         |tun_if| <-------  |VPP| <------ |pg1|
 200          ------             ---           ---
 201
 202          ------   decrypt   ---   plain   ---
 203         |tun_if| ------->  |VPP| ------> |pg1|
 204          ------             ---           ---
 205     """
 206
 207     tun_spd_id = 1
 208     tra_spd_id = 2
 209
 210     def ipsec_select_backend(self):
 211         """empty method to be overloaded when necessary"""
 212         pass
 213
 214     @classmethod
 215     def setUpClass(cls):
 216         super(TemplateIpsec, cls).setUpClass()
 217
 218     @classmethod
 219     def tearDownClass(cls):
 220         super(TemplateIpsec, cls).tearDownClass()
 221
 222     def setup_params(self):
 223         if not hasattr(self, "ipv4_params"):
 224             self.ipv4_params = IPsecIPv4Params()
 225         if not hasattr(self, "ipv6_params"):
 226             self.ipv6_params = IPsecIPv6Params()
 227         self.params = {
 228             self.ipv4_params.addr_type: self.ipv4_params,
 229             self.ipv6_params.addr_type: self.ipv6_params,
 230         }
 231
 232     def config_interfaces(self):
 233         self.create_pg_interfaces(range(3))
 234         self.interfaces = list(self.pg_interfaces)
 235         for i in self.interfaces:
 236             i.admin_up()
 237             i.config_ip4()
 238             i.resolve_arp()
 239             i.config_ip6()
 240             i.resolve_ndp()
 241
 242     def setUp(self):
 243         super(TemplateIpsec, self).setUp()
 244
 245         self.setup_params()
 246
 247         self.vpp_esp_protocol = VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_ESP
 248         self.vpp_ah_protocol = VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_AH
 249
 250         self.config_interfaces()
 251
 252         self.ipsec_select_backend()
 253
 254     def unconfig_interfaces(self):
 255         for i in self.interfaces:
 256             i.admin_down()
 257             i.unconfig_ip4()
 258             i.unconfig_ip6()
 259
 260     def tearDown(self):
 261         super(TemplateIpsec, self).tearDown()
 262
 263         self.unconfig_interfaces()
 264
 265     def show_commands_at_teardown(self):
 266         self.logger.info(self.vapi.cli("show hardware"))
 267
 268     def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1, payload_size=54):
 269         return [
 270             Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
 271             / sa.encrypt(IP(src=src, dst=dst) / ICMP() / Raw(b"X" * payload_size))
 272             for i in range(count)
 273         ]
 274
 275     def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1, payload_size=54):
 276         return [
 277             Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
 278             / sa.encrypt(
 279                 IPv6(src=src, dst=dst, hlim=p.inner_hop_limit, fl=p.inner_flow_label)
 280                 / ICMPv6EchoRequest(id=0, seq=1, data="X" * payload_size)
 281             )
 282             for i in range(count)
 283         ]
 284
 285     def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
 286         return [
 287             Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
 288             / IP(src=src, dst=dst)
 289             / ICMP()
 290             / Raw(b"X" * payload_size)
 291             for i in range(count)
 292         ]
 293
 294     def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=54):
 295         return [
 296             Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
 297             / IPv6(src=src, dst=dst, hlim=p.inner_hop_limit, fl=p.inner_flow_label)
 298             / ICMPv6EchoRequest(id=0, seq=1, data="X" * payload_size)
 299             for i in range(count)
 300         ]
 301
 302
 303 class IpsecTcp(object):
 304     def verify_tcp_checksum(self):
 305         # start http cli server listener on http://0.0.0.0:80
 306         self.vapi.cli("http cli server")
 307         p = self.params[socket.AF_INET]
 308         send = Ether(
 309             src=self.tun_if.remote_mac, dst=self.tun_if.local_mac
 310         ) / p.scapy_tun_sa.encrypt(
 311             IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
 312             / TCP(flags="S", dport=80)
 313         )
 314         self.logger.debug(ppp("Sending packet:", send))
 315         recv = self.send_and_expect(self.tun_if, [send], self.tun_if)
 316         recv = recv[0]
 317         decrypted = p.vpp_tun_sa.decrypt(recv[IP])
 318         self.assert_packet_checksums_valid(decrypted)
 319
 320
 321 class IpsecTcpTests(IpsecTcp):
 322     def test_tcp_checksum(self):
 323         """verify checksum correctness for vpp generated packets"""
 324         self.verify_tcp_checksum()
 325
 326
 327 class IpsecTra4(object):
 328     """verify methods for Transport v4"""
 329
 330     def get_replay_counts(self, p):
 331         replay_node_name = "/err/%s/replay" % self.tra4_decrypt_node_name[0]
 332         count = self.statistics.get_err_counter(replay_node_name)
 333
 334         if p.async_mode:
 335             replay_post_node_name = (
 336                 "/err/%s/replay" % self.tra4_decrypt_node_name[p.async_mode]
 337             )
 338             count += self.statistics.get_err_counter(replay_post_node_name)
 339
 340         return count
 341
 342     def get_hash_failed_counts(self, p):
 343         if ESP == self.encryption_type and p.crypt_algo in ("AES-GCM", "AES-NULL-GMAC"):
 344             hash_failed_node_name = (
 345                 "/err/%s/decryption_failed" % self.tra4_decrypt_node_name[p.async_mode]
 346             )
 347         else:
 348             hash_failed_node_name = (
 349                 "/err/%s/integ_error" % self.tra4_decrypt_node_name[p.async_mode]
 350             )
 351         count = self.statistics.get_err_counter(hash_failed_node_name)
 352
 353         if p.async_mode:
 354             count += self.statistics.get_err_counter("/err/crypto-dispatch/bad-hmac")
 355
 356         return count
 357
 358     def verify_hi_seq_num(self):
 359         p = self.params[socket.AF_INET]
 360         saf = VppEnum.vl_api_ipsec_sad_flags_t
 361         esn_on = p.vpp_tra_sa.esn_en
 362         ar_on = p.flags & saf.IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY
 363
 364         seq_cycle_node_name = "/err/%s/seq_cycled" % self.tra4_encrypt_node_name
 365         replay_count = self.get_replay_counts(p)
 366         hash_failed_count = self.get_hash_failed_counts(p)
 367         seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
 368
 369         # a few packets so we get the rx seq number above the window size and
 370         # thus can simulate a wrap with an out of window packet
 371         pkts = [
 372             (
 373                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 374                 / p.scapy_tra_sa.encrypt(
 375                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 376                     seq_num=seq,
 377                 )
 378             )
 379             for seq in range(63, 80)
 380         ]
 381         recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 382
 383         # these 4 packets will all choose seq-num 0 to decrpyt since none
 384         # are out of window when first checked. however, once #200 has
 385         # decrypted it will move the window to 200 and has #81 is out of
 386         # window. this packet should be dropped.
 387         pkts = [
 388             (
 389                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 390                 / p.scapy_tra_sa.encrypt(
 391                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 392                     seq_num=200,
 393                 )
 394             ),
 395             (
 396                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 397                 / p.scapy_tra_sa.encrypt(
 398                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 399                     seq_num=81,
 400                 )
 401             ),
 402             (
 403                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 404                 / p.scapy_tra_sa.encrypt(
 405                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 406                     seq_num=201,
 407                 )
 408             ),
 409             (
 410                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 411                 / p.scapy_tra_sa.encrypt(
 412                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 413                     seq_num=202,
 414                 )
 415             ),
 416         ]
 417
 418         # if anti-replay is off then we won't drop #81
 419         n_rx = 3 if ar_on else 4
 420         self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=n_rx)
 421         # this packet is one before the wrap
 422         pkts = [
 423             (
 424                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 425                 / p.scapy_tra_sa.encrypt(
 426                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 427                     seq_num=203,
 428                 )
 429             )
 430         ]
 431         recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 432
 433         # a replayed packet, then an out of window, then a legit
 434         # tests that a early failure on the batch doesn't affect subsequent packets.
 435         pkts = [
 436             (
 437                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 438                 / p.scapy_tra_sa.encrypt(
 439                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 440                     seq_num=203,
 441                 )
 442             ),
 443             (
 444                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 445                 / p.scapy_tra_sa.encrypt(
 446                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 447                     seq_num=81,
 448                 )
 449             ),
 450             (
 451                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 452                 / p.scapy_tra_sa.encrypt(
 453                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 454                     seq_num=204,
 455                 )
 456             ),
 457         ]
 458         n_rx = 1 if ar_on else 3
 459         recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=n_rx)
 460
 461         # move the window over half way to a wrap
 462         pkts = [
 463             (
 464                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 465                 / p.scapy_tra_sa.encrypt(
 466                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 467                     seq_num=0x80000001,
 468                 )
 469             )
 470         ]
 471         recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 472
 473         # anti-replay will drop old packets, no anti-replay will not
 474         pkts = [
 475             (
 476                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 477                 / p.scapy_tra_sa.encrypt(
 478                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 479                     seq_num=0x44000001,
 480                 )
 481             )
 482         ]
 483
 484         if ar_on:
 485             self.send_and_assert_no_replies(self.tra_if, pkts)
 486         else:
 487             recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 488
 489         if esn_on:
 490             #
 491             # validate wrapping the ESN
 492             #
 493
 494             # wrap scapy's TX SA SN
 495             p.scapy_tra_sa.seq_num = 0x100000005
 496
 497             # send a packet that wraps the window for both AR and no AR
 498             pkts = [
 499                 (
 500                     Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 501                     / p.scapy_tra_sa.encrypt(
 502                         IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
 503                         / ICMP(),
 504                         seq_num=0x100000005,
 505                     )
 506                 )
 507             ]
 508
 509             rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 510             for rx in rxs:
 511                 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 512
 513             # move the window forward to half way to the next wrap
 514             pkts = [
 515                 (
 516                     Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 517                     / p.scapy_tra_sa.encrypt(
 518                         IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
 519                         / ICMP(),
 520                         seq_num=0x180000005,
 521                     )
 522                 )
 523             ]
 524
 525             rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 526
 527             # a packet less than 2^30 from the current position is:
 528             #  - AR: out of window and dropped
 529             #  - non-AR: accepted
 530             pkts = [
 531                 (
 532                     Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 533                     / p.scapy_tra_sa.encrypt(
 534                         IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
 535                         / ICMP(),
 536                         seq_num=0x170000005,
 537                     )
 538                 )
 539             ]
 540
 541             if ar_on:
 542                 self.send_and_assert_no_replies(self.tra_if, pkts)
 543             else:
 544                 self.send_and_expect(self.tra_if, pkts, self.tra_if)
 545
 546             # a packet more than 2^30 from the current position is:
 547             #  - AR: out of window and dropped
 548             #  - non-AR: considered a wrap, but since it's not a wrap
 549             #    it won't decrpyt and so will be dropped
 550             pkts = [
 551                 (
 552                     Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 553                     / p.scapy_tra_sa.encrypt(
 554                         IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
 555                         / ICMP(),
 556                         seq_num=0x130000005,
 557                     )
 558                 )
 559             ]
 560
 561             self.send_and_assert_no_replies(self.tra_if, pkts)
 562
 563             # a packet less than 2^30 from the current position and is a
 564             # wrap; (the seq is currently at 0x180000005).
 565             #  - AR: out of window so considered a wrap, so accepted
 566             #  - non-AR: not considered a wrap, so won't decrypt
 567             p.scapy_tra_sa.seq_num = 0x260000005
 568             pkts = [
 569                 (
 570                     Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 571                     / p.scapy_tra_sa.encrypt(
 572                         IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
 573                         / ICMP(),
 574                         seq_num=0x260000005,
 575                     )
 576                 )
 577             ]
 578             if ar_on:
 579                 self.send_and_expect(self.tra_if, pkts, self.tra_if)
 580             else:
 581                 self.send_and_assert_no_replies(self.tra_if, pkts)
 582
 583             #
 584             # window positions are different now for AR/non-AR
 585             #  move non-AR forward
 586             #
 587             if not ar_on:
 588                 # a packet more than 2^30 from the current position and is a
 589                 # wrap; (the seq is currently at 0x180000005).
 590                 #  - AR: accepted
 591                 #  - non-AR: not considered a wrap, so won't decrypt
 592
 593                 pkts = [
 594                     (
 595                         Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 596                         / p.scapy_tra_sa.encrypt(
 597                             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
 598                             / ICMP(),
 599                             seq_num=0x200000005,
 600                         )
 601                     ),
 602                     (
 603                         Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 604                         / p.scapy_tra_sa.encrypt(
 605                             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
 606                             / ICMP(),
 607                             seq_num=0x200000006,
 608                         )
 609                     ),
 610                 ]
 611                 self.send_and_expect(self.tra_if, pkts, self.tra_if)
 612
 613                 pkts = [
 614                     (
 615                         Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 616                         / p.scapy_tra_sa.encrypt(
 617                             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4)
 618                             / ICMP(),
 619                             seq_num=0x260000005,
 620                         )
 621                     )
 622                 ]
 623                 self.send_and_expect(self.tra_if, pkts, self.tra_if)
 624
 625     def verify_tra_anti_replay(self):
 626         p = self.params[socket.AF_INET]
 627         esn_en = p.vpp_tra_sa.esn_en
 628
 629         seq_cycle_node_name = "/err/%s/seq_cycled" % self.tra4_encrypt_node_name
 630         replay_count = self.get_replay_counts(p)
 631         hash_failed_count = self.get_hash_failed_counts(p)
 632         seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
 633         hash_err = "integ_error"
 634
 635         if ESP == self.encryption_type:
 636             undersize_node_name = "/err/%s/runt" % self.tra4_decrypt_node_name[0]
 637             undersize_count = self.statistics.get_err_counter(undersize_node_name)
 638             # For AES-GCM an error in the hash is reported as a decryption failure
 639             if p.crypt_algo in ("AES-GCM", "AES-NULL-GMAC"):
 640                 hash_err = "decryption_failed"
 641         # In async mode, we don't report errors in the hash.
 642         if p.async_mode:
 643             hash_err = ""
 644
 645         #
 646         # send packets with seq numbers 1->34
 647         # this means the window size is still in Case B (see RFC4303
 648         # Appendix A)
 649         #
 650         # for reasons i haven't investigated Scapy won't create a packet with
 651         # seq_num=0
 652         #
 653         pkts = [
 654             (
 655                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 656                 / p.scapy_tra_sa.encrypt(
 657                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 658                     seq_num=seq,
 659                 )
 660             )
 661             for seq in range(1, 34)
 662         ]
 663         recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 664
 665         # replayed packets are dropped
 666         self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
 667         replay_count += len(pkts)
 668         self.assertEqual(self.get_replay_counts(p), replay_count)
 669         err = p.tra_sa_in.get_err("replay")
 670         self.assertEqual(err, replay_count)
 671
 672         #
 673         # now send a batch of packets all with the same sequence number
 674         # the first packet in the batch is legitimate, the rest bogus
 675         #
 676         self.vapi.cli("clear error")
 677         self.vapi.cli("clear node counters")
 678         pkts = Ether(
 679             src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 680         ) / p.scapy_tra_sa.encrypt(
 681             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 682             seq_num=35,
 683         )
 684         recv_pkts = self.send_and_expect(self.tra_if, pkts * 8, self.tra_if, n_rx=1)
 685         replay_count += 7
 686         self.assertEqual(self.get_replay_counts(p), replay_count)
 687         err = p.tra_sa_in.get_err("replay")
 688         self.assertEqual(err, replay_count)
 689
 690         #
 691         # now move the window over to 257 (more than one byte) and into Case A
 692         #
 693         self.vapi.cli("clear error")
 694         pkt = Ether(
 695             src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 696         ) / p.scapy_tra_sa.encrypt(
 697             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 698             seq_num=257,
 699         )
 700         recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 701
 702         # replayed packets are dropped
 703         self.send_and_assert_no_replies(self.tra_if, pkt * 3, timeout=0.2)
 704         replay_count += 3
 705         self.assertEqual(self.get_replay_counts(p), replay_count)
 706         err = p.tra_sa_in.get_err("replay")
 707         self.assertEqual(err, replay_count)
 708
 709         # the window size is 64 packets
 710         # in window are still accepted
 711         pkt = Ether(
 712             src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 713         ) / p.scapy_tra_sa.encrypt(
 714             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 715             seq_num=200,
 716         )
 717         recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 718
 719         # a packet that does not decrypt does not move the window forward
 720         bogus_sa = SecurityAssociation(
 721             self.encryption_type,
 722             p.scapy_tra_spi,
 723             crypt_algo=p.crypt_algo,
 724             crypt_key=mk_scapy_crypt_key(p)[::-1],
 725             auth_algo=p.auth_algo,
 726             auth_key=p.auth_key[::-1],
 727         )
 728         pkt = Ether(
 729             src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 730         ) / bogus_sa.encrypt(
 731             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 732             seq_num=350,
 733         )
 734         self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
 735
 736         hash_failed_count += 17
 737         self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
 738         if hash_err != "":
 739             err = p.tra_sa_in.get_err(hash_err)
 740             self.assertEqual(err, hash_failed_count)
 741
 742         # a malformed 'runt' packet
 743         #  created by a mis-constructed SA
 744         if ESP == self.encryption_type and p.crypt_algo != "NULL":
 745             bogus_sa = SecurityAssociation(self.encryption_type, p.scapy_tra_spi)
 746             pkt = Ether(
 747                 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 748             ) / bogus_sa.encrypt(
 749                 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 750                 seq_num=350,
 751             )
 752             self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
 753
 754             undersize_count += 17
 755             self.assert_error_counter_equal(undersize_node_name, undersize_count)
 756             err = p.tra_sa_in.get_err("runt")
 757             self.assertEqual(err, undersize_count)
 758
 759         # which we can determine since this packet is still in the window
 760         pkt = Ether(
 761             src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 762         ) / p.scapy_tra_sa.encrypt(
 763             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 764             seq_num=234,
 765         )
 766         self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 767
 768         #
 769         # out of window are dropped
 770         #  this is Case B. So VPP will consider this to be a high seq num wrap
 771         #  and so the decrypt attempt will fail
 772         #
 773         pkt = Ether(
 774             src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 775         ) / p.scapy_tra_sa.encrypt(
 776             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 777             seq_num=17,
 778         )
 779         self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
 780
 781         if esn_en:
 782             # an out of window error with ESN looks like a high sequence
 783             # wrap. but since it isn't then the verify will fail.
 784             hash_failed_count += 17
 785             self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
 786             if hash_err != "":
 787                 err = p.tra_sa_in.get_err(hash_err)
 788                 self.assertEqual(err, hash_failed_count)
 789
 790         else:
 791             replay_count += 17
 792             self.assertEqual(self.get_replay_counts(p), replay_count)
 793             err = p.tra_sa_in.get_err("replay")
 794             self.assertEqual(err, replay_count)
 795
 796         # valid packet moves the window over to 258
 797         pkt = Ether(
 798             src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 799         ) / p.scapy_tra_sa.encrypt(
 800             IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 801             seq_num=258,
 802         )
 803         rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 804         decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 805
 806         #
 807         # move VPP's SA TX seq-num to just before the seq-number wrap.
 808         # then fire in a packet that VPP should drop on TX because it
 809         # causes the TX seq number to wrap; unless we're using extened sequence
 810         # numbers.
 811         #
 812         self.vapi.cli("test ipsec sa %d seq 0xffffffff" % p.vpp_tra_sa_id)
 813         self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
 814         self.logger.info(self.vapi.ppcli("show ipsec sa 1"))
 815
 816         pkts = [
 817             (
 818                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 819                 / p.scapy_tra_sa.encrypt(
 820                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 821                     seq_num=seq,
 822                 )
 823             )
 824             for seq in range(259, 280)
 825         ]
 826
 827         if esn_en:
 828             rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 829
 830             #
 831             # in order for scapy to decrypt its SA's high order number needs
 832             # to wrap
 833             #
 834             p.vpp_tra_sa.seq_num = 0x100000000
 835             for rx in rxs:
 836                 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 837
 838             #
 839             # wrap scapy's TX high sequence number. VPP is in case B, so it
 840             # will consider this a high seq wrap also.
 841             # The low seq num we set it to will place VPP's RX window in Case A
 842             #
 843             p.scapy_tra_sa.seq_num = 0x100000005
 844             pkt = Ether(
 845                 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 846             ) / p.scapy_tra_sa.encrypt(
 847                 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 848                 seq_num=0x100000005,
 849             )
 850             rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 851
 852             decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 853
 854             #
 855             # A packet that has seq num between (2^32-64) and 5 is within
 856             # the window
 857             #
 858             p.scapy_tra_sa.seq_num = 0xFFFFFFFD
 859             pkt = Ether(
 860                 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 861             ) / p.scapy_tra_sa.encrypt(
 862                 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 863                 seq_num=0xFFFFFFFD,
 864             )
 865             rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 866             decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 867
 868             #
 869             # While in case A we cannot wrap the high sequence number again
 870             # because VPP will consider this packet to be one that moves the
 871             # window forward
 872             #
 873             pkt = Ether(
 874                 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 875             ) / p.scapy_tra_sa.encrypt(
 876                 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 877                 seq_num=0x200000999,
 878             )
 879             self.send_and_assert_no_replies(
 880                 self.tra_if, [pkt], self.tra_if, timeout=0.2
 881             )
 882
 883             hash_failed_count += 1
 884             self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
 885             if hash_err != "":
 886                 err = p.tra_sa_in.get_err(hash_err)
 887                 self.assertEqual(err, hash_failed_count)
 888
 889             #
 890             # but if we move the window forward to case B, then we can wrap
 891             # again
 892             #
 893             p.scapy_tra_sa.seq_num = 0x100000555
 894             pkt = Ether(
 895                 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 896             ) / p.scapy_tra_sa.encrypt(
 897                 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 898                 seq_num=0x100000555,
 899             )
 900             rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 901             decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 902
 903             p.scapy_tra_sa.seq_num = 0x200000444
 904             pkt = Ether(
 905                 src=self.tra_if.remote_mac, dst=self.tra_if.local_mac
 906             ) / p.scapy_tra_sa.encrypt(
 907                 IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 908                 seq_num=0x200000444,
 909             )
 910             rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 911             decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 912
 913         else:
 914             #
 915             # without ESN TX sequence numbers can't wrap and packets are
 916             # dropped from here on out.
 917             #
 918             self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
 919             seq_cycle_count += len(pkts)
 920             self.assert_error_counter_equal(seq_cycle_node_name, seq_cycle_count)
 921             err = p.tra_sa_out.get_err("seq_cycled")
 922             self.assertEqual(err, seq_cycle_count)
 923
 924         # move the security-associations seq number on to the last we used
 925         self.vapi.cli("test ipsec sa %d seq 0x15f" % p.scapy_tra_sa_id)
 926         p.scapy_tra_sa.seq_num = 351
 927         p.vpp_tra_sa.seq_num = 351
 928
 929     def verify_tra_lost(self):
 930         p = self.params[socket.AF_INET]
 931         esn_en = p.vpp_tra_sa.esn_en
 932
 933         #
 934         # send packets with seq numbers 1->34
 935         # this means the window size is still in Case B (see RFC4303
 936         # Appendix A)
 937         #
 938         # for reasons i haven't investigated Scapy won't create a packet with
 939         # seq_num=0
 940         #
 941         pkts = [
 942             (
 943                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 944                 / p.scapy_tra_sa.encrypt(
 945                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 946                     seq_num=seq,
 947                 )
 948             )
 949             for seq in range(1, 3)
 950         ]
 951         self.send_and_expect(self.tra_if, pkts, self.tra_if)
 952
 953         self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
 954
 955         # skip a sequence number
 956         pkts = [
 957             (
 958                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 959                 / p.scapy_tra_sa.encrypt(
 960                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 961                     seq_num=seq,
 962                 )
 963             )
 964             for seq in range(4, 6)
 965         ]
 966         self.send_and_expect(self.tra_if, pkts, self.tra_if)
 967
 968         self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
 969
 970         # the lost packet are counted untill we get up past the first
 971         # sizeof(replay_window) packets
 972         pkts = [
 973             (
 974                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 975                 / p.scapy_tra_sa.encrypt(
 976                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 977                     seq_num=seq,
 978                 )
 979             )
 980             for seq in range(6, 100)
 981         ]
 982         self.send_and_expect(self.tra_if, pkts, self.tra_if)
 983
 984         self.assertEqual(p.tra_sa_in.get_err("lost"), 1)
 985
 986         # lost of holes in the sequence
 987         pkts = [
 988             (
 989                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
 990                 / p.scapy_tra_sa.encrypt(
 991                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
 992                     seq_num=seq,
 993                 )
 994             )
 995             for seq in range(100, 200, 2)
 996         ]
 997         self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=50)
 998
 999         pkts = [
1000             (
1001                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
1002                 / p.scapy_tra_sa.encrypt(
1003                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
1004                     seq_num=seq,
1005                 )
1006             )
1007             for seq in range(200, 300)
1008         ]
1009         self.send_and_expect(self.tra_if, pkts, self.tra_if)
1010
1011         self.assertEqual(p.tra_sa_in.get_err("lost"), 51)
1012
1013         # a big hole in the seq number space
1014         pkts = [
1015             (
1016                 Ether(src=self.tra_if.remote_mac, dst=self.tra_if.local_mac)
1017                 / p.scapy_tra_sa.encrypt(
1018                     IP(src=self.tra_if.remote_ip4, dst=self.tra_if.local_ip4) / ICMP(),
1019                     seq_num=seq,
1020                 )
1021             )
1022             for seq in range(400, 500)
1023         ]
1024         self.send_and_expect(self.tra_if, pkts, self.tra_if)
1025
1026         self.assertEqual(p.tra_sa_in.get_err("lost"), 151)
1027
1028     def verify_tra_basic4(self, count=1, payload_size=54):
1029         """ipsec v4 transport basic test"""
1030         self.vapi.cli("clear errors")
1031         self.vapi.cli("clear ipsec sa")
1032         try:
1033             p = self.params[socket.AF_INET]
1034             send_pkts = self.gen_encrypt_pkts(
1035                 p,
1036                 p.scapy_tra_sa,
1037                 self.tra_if,
1038                 src=self.tra_if.remote_ip4,
1039                 dst=self.tra_if.local_ip4,
1040                 count=count,
1041                 payload_size=payload_size,
1042             )
1043             recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
1044             for rx in recv_pkts:
1045                 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
1046                 self.assert_packet_checksums_valid(rx)
1047                 try:
1048                     decrypted = p.vpp_tra_sa.decrypt(rx[IP])
1049                     self.assert_packet_checksums_valid(decrypted)
1050                 except:
1051                     self.logger.debug(ppp("Unexpected packet:", rx))
1052                     raise
1053         finally:
1054             self.logger.info(self.vapi.ppcli("show error"))
1055             self.logger.info(self.vapi.ppcli("show ipsec all"))
1056
1057         pkts = p.tra_sa_in.get_stats()["packets"]
1058         self.assertEqual(
1059             pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1060         )
1061         pkts = p.tra_sa_out.get_stats()["packets"]
1062         self.assertEqual(
1063             pkts, count, "incorrect SA out counts: expected %d != %d" % (count, pkts)
1064         )
1065         self.assertEqual(p.tra_sa_out.get_err("lost"), 0)
1066         self.assertEqual(p.tra_sa_in.get_err("lost"), 0)
1067
1068         self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
1069         self.assert_packet_counter_equal(self.tra4_decrypt_node_name[0], count)
1070
1071
1072 class IpsecTra4Tests(IpsecTra4):
1073     """UT test methods for Transport v4"""
1074
1075     def test_tra_anti_replay(self):
1076         """ipsec v4 transport anti-replay test"""
1077         self.verify_tra_anti_replay()
1078
1079     def test_tra_lost(self):
1080         """ipsec v4 transport lost packet test"""
1081         self.verify_tra_lost()
1082
1083     def test_tra_basic(self, count=1):
1084         """ipsec v4 transport basic test"""
1085         self.verify_tra_basic4(count=1)
1086
1087     def test_tra_burst(self):
1088         """ipsec v4 transport burst test"""
1089         self.verify_tra_basic4(count=257)
1090
1091
1092 class IpsecTra6(object):
1093     """verify methods for Transport v6"""
1094
1095     def verify_tra_basic6(self, count=1, payload_size=54):
1096         self.vapi.cli("clear errors")
1097         self.vapi.cli("clear ipsec sa")
1098         try:
1099             p = self.params[socket.AF_INET6]
1100             send_pkts = self.gen_encrypt_pkts6(
1101                 p,
1102                 p.scapy_tra_sa,
1103                 self.tra_if,
1104                 src=self.tra_if.remote_ip6,
1105                 dst=self.tra_if.local_ip6,
1106                 count=count,
1107                 payload_size=payload_size,
1108             )
1109             recv_pkts = self.send_and_expect(self.tra_if, send_pkts, self.tra_if)
1110             for rx in recv_pkts:
1111                 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()), rx[IPv6].plen)
1112                 try:
1113                     decrypted = p.vpp_tra_sa.decrypt(rx[IPv6])
1114                     self.assert_packet_checksums_valid(decrypted)
1115                 except:
1116                     self.logger.debug(ppp("Unexpected packet:", rx))
1117                     raise
1118         finally:
1119             self.logger.info(self.vapi.ppcli("show error"))
1120             self.logger.info(self.vapi.ppcli("show ipsec all"))
1121
1122         pkts = p.tra_sa_in.get_stats()["packets"]
1123         self.assertEqual(
1124             pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1125         )
1126         pkts = p.tra_sa_out.get_stats()["packets"]
1127         self.assertEqual(
1128             pkts, count, "incorrect SA out counts: expected %d != %d" % (count, pkts)
1129         )
1130         self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
1131         self.assert_packet_counter_equal(self.tra6_decrypt_node_name[0], count)
1132
1133     def gen_encrypt_pkts_ext_hdrs6(
1134         self, sa, sw_intf, src, dst, count=1, payload_size=54
1135     ):
1136         return [
1137             Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
1138             / sa.encrypt(
1139                 IPv6(src=src, dst=dst)
1140                 / ICMPv6EchoRequest(id=0, seq=1, data="X" * payload_size)
1141             )
1142             for i in range(count)
1143         ]
1144
1145     def gen_pkts_ext_hdrs6(self, sw_intf, src, dst, count=1, payload_size=54):
1146         return [
1147             Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac)
1148             / IPv6(src=src, dst=dst)
1149             / IPv6ExtHdrHopByHop()
1150             / IPv6ExtHdrFragment(id=2, offset=200)
1151             / Raw(b"\xff" * 200)
1152             for i in range(count)
1153         ]
1154
1155     def verify_tra_encrypted6(self, p, sa, rxs):
1156         decrypted = []
1157         for rx in rxs:
1158             self.assert_packet_checksums_valid(rx)
1159             try:
1160                 decrypt_pkt = p.vpp_tra_sa.decrypt(rx[IPv6])
1161                 decrypted.append(decrypt_pkt)
1162                 self.assert_equal(decrypt_pkt.src, self.tra_if.local_ip6)
1163                 self.assert_equal(decrypt_pkt.dst, self.tra_if.remote_ip6)
1164             except:
1165                 self.logger.debug(ppp("Unexpected packet:", rx))
1166                 try:
1167                     self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1168                 except:
1169                     pass
1170                 raise
1171         return decrypted
1172
1173     def verify_tra_66_ext_hdrs(self, p):
1174         count = 63
1175
1176         #
1177         # check we can decrypt with options
1178         #
1179         tx = self.gen_encrypt_pkts_ext_hdrs6(
1180             p.scapy_tra_sa,
1181             self.tra_if,
1182             src=self.tra_if.remote_ip6,
1183             dst=self.tra_if.local_ip6,
1184             count=count,
1185         )
1186         self.send_and_expect(self.tra_if, tx, self.tra_if)
1187
1188         #
1189         # injecting a packet from ourselves to be routed of box is a hack
1190         # but it matches an outbout policy, alors je ne regrette rien
1191         #
1192
1193         # one extension before ESP
1194         tx = (
1195             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1196             / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1197             / IPv6ExtHdrFragment(id=2, offset=200)
1198             / Raw(b"\xff" * 200)
1199         )
1200
1201         rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1202         dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1203
1204         for dc in dcs:
1205             # for reasons i'm not going to investigate scapy does not
1206             # created the correct headers after decrypt. but reparsing
1207             # the ipv6 packet fixes it
1208             dc = IPv6(raw(dc[IPv6]))
1209             self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1210
1211         # two extensions before ESP
1212         tx = (
1213             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1214             / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1215             / IPv6ExtHdrHopByHop()
1216             / IPv6ExtHdrFragment(id=2, offset=200)
1217             / Raw(b"\xff" * 200)
1218         )
1219
1220         rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1221         dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1222
1223         for dc in dcs:
1224             dc = IPv6(raw(dc[IPv6]))
1225             self.assertTrue(dc[IPv6ExtHdrHopByHop])
1226             self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1227
1228         # two extensions before ESP, one after
1229         tx = (
1230             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
1231             / IPv6(src=self.tra_if.local_ip6, dst=self.tra_if.remote_ip6)
1232             / IPv6ExtHdrHopByHop()
1233             / IPv6ExtHdrFragment(id=2, offset=200)
1234             / IPv6ExtHdrDestOpt()
1235             / Raw(b"\xff" * 200)
1236         )
1237
1238         rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
1239         dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
1240
1241         for dc in dcs:
1242             dc = IPv6(raw(dc[IPv6]))
1243             self.assertTrue(dc[IPv6ExtHdrDestOpt])
1244             self.assertTrue(dc[IPv6ExtHdrHopByHop])
1245             self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
1246
1247
1248 class IpsecTra6Tests(IpsecTra6):
1249     """UT test methods for Transport v6"""
1250
1251     def test_tra_basic6(self):
1252         """ipsec v6 transport basic test"""
1253         self.verify_tra_basic6(count=1)
1254
1255     def test_tra_burst6(self):
1256         """ipsec v6 transport burst test"""
1257         self.verify_tra_basic6(count=257)
1258
1259
1260 class IpsecTra6ExtTests(IpsecTra6):
1261     def test_tra_ext_hdrs_66(self):
1262         """ipsec 6o6 tra extension headers test"""
1263         self.verify_tra_66_ext_hdrs(self.params[socket.AF_INET6])
1264
1265
1266 class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
1267     """UT test methods for Transport v6 and v4"""
1268
1269     pass
1270
1271
1272 class IpsecTun4(object):
1273     """verify methods for Tunnel v4"""
1274
1275     def verify_counters4(self, p, count, n_frags=None, worker=None):
1276         if not n_frags:
1277             n_frags = count
1278         if hasattr(p, "spd_policy_in_any"):
1279             pkts = p.spd_policy_in_any.get_stats(worker)["packets"]
1280             self.assertEqual(
1281                 pkts,
1282                 count,
1283                 "incorrect SPD any policy: expected %d != %d" % (count, pkts),
1284             )
1285
1286         if hasattr(p, "tun_sa_in"):
1287             pkts = p.tun_sa_in.get_stats(worker)["packets"]
1288             self.assertEqual(
1289                 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1290             )
1291             pkts = p.tun_sa_out.get_stats(worker)["packets"]
1292             self.assertEqual(
1293                 pkts,
1294                 n_frags,
1295                 "incorrect SA out counts: expected %d != %d" % (count, pkts),
1296             )
1297
1298         self.assert_packet_counter_equal(self.tun4_encrypt_node_name, n_frags)
1299         self.assert_packet_counter_equal(self.tun4_decrypt_node_name[0], count)
1300
1301     def verify_decrypted(self, p, rxs):
1302         for rx in rxs:
1303             self.assert_equal(rx[IP].src, p.remote_tun_if_host)
1304             self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
1305             self.assert_packet_checksums_valid(rx)
1306
1307     def verify_esp_padding(self, sa, esp_payload, decrypt_pkt):
1308         align = sa.crypt_algo.block_size
1309         if align < 4:
1310             align = 4
1311         exp_len = (len(decrypt_pkt) + 2 + (align - 1)) & ~(align - 1)
1312         exp_len += sa.crypt_algo.iv_size
1313         exp_len += sa.crypt_algo.icv_size or sa.auth_algo.icv_size
1314         self.assertEqual(exp_len, len(esp_payload))
1315
1316     def verify_encrypted(self, p, sa, rxs):
1317         decrypt_pkts = []
1318         for rx in rxs:
1319             if p.nat_header:
1320                 self.assertEqual(rx[UDP].dport, p.nat_header.dport)
1321             self.assert_packet_checksums_valid(rx)
1322             self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
1323             try:
1324                 rx_ip = rx[IP]
1325                 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
1326                 if not decrypt_pkt.haslayer(IP):
1327                     decrypt_pkt = IP(decrypt_pkt[Raw].load)
1328                 if rx_ip.proto == socket.IPPROTO_ESP:
1329                     self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
1330                 decrypt_pkts.append(decrypt_pkt)
1331                 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1332                 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1333             except:
1334                 self.logger.debug(ppp("Unexpected packet:", rx))
1335                 try:
1336                     self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1337                 except:
1338                     pass
1339                 raise
1340         pkts = reassemble4(decrypt_pkts)
1341         for pkt in pkts:
1342             self.assert_packet_checksums_valid(pkt)
1343
1344     def verify_tun_44(self, p, count=1, payload_size=64, n_rx=None):
1345         self.vapi.cli("clear errors")
1346         self.vapi.cli("clear ipsec counters")
1347         self.vapi.cli("clear ipsec sa")
1348         if not n_rx:
1349             n_rx = count
1350         try:
1351             send_pkts = self.gen_encrypt_pkts(
1352                 p,
1353                 p.scapy_tun_sa,
1354                 self.tun_if,
1355                 src=p.remote_tun_if_host,
1356                 dst=self.pg1.remote_ip4,
1357                 count=count,
1358                 payload_size=payload_size,
1359             )
1360             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1361             self.verify_decrypted(p, recv_pkts)
1362
1363             send_pkts = self.gen_pkts(
1364                 self.pg1,
1365                 src=self.pg1.remote_ip4,
1366                 dst=p.remote_tun_if_host,
1367                 count=count,
1368                 payload_size=payload_size,
1369             )
1370             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if, n_rx)
1371             self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1372
1373             for rx in recv_pkts:
1374                 self.assertEqual(rx[IP].src, p.tun_src)
1375                 self.assertEqual(rx[IP].dst, p.tun_dst)
1376
1377         finally:
1378             self.logger.info(self.vapi.ppcli("show error"))
1379             self.logger.info(self.vapi.ppcli("show ipsec all"))
1380
1381         self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
1382         self.logger.info(self.vapi.ppcli("show ipsec sa 4"))
1383         self.verify_counters4(p, count, n_rx)
1384
1385     def verify_tun_dropped_44(self, p, count=1, payload_size=64, n_rx=None):
1386         self.vapi.cli("clear errors")
1387         if not n_rx:
1388             n_rx = count
1389         try:
1390             send_pkts = self.gen_encrypt_pkts(
1391                 p,
1392                 p.scapy_tun_sa,
1393                 self.tun_if,
1394                 src=p.remote_tun_if_host,
1395                 dst=self.pg1.remote_ip4,
1396                 count=count,
1397             )
1398             self.send_and_assert_no_replies(self.tun_if, send_pkts)
1399
1400             send_pkts = self.gen_pkts(
1401                 self.pg1,
1402                 src=self.pg1.remote_ip4,
1403                 dst=p.remote_tun_if_host,
1404                 count=count,
1405                 payload_size=payload_size,
1406             )
1407             self.send_and_assert_no_replies(self.pg1, send_pkts)
1408
1409         finally:
1410             self.logger.info(self.vapi.ppcli("show error"))
1411             self.logger.info(self.vapi.ppcli("show ipsec all"))
1412
1413     def verify_tun_reass_44(self, p):
1414         self.vapi.cli("clear errors")
1415         self.vapi.ip_reassembly_enable_disable(
1416             sw_if_index=self.tun_if.sw_if_index, enable_ip4=True
1417         )
1418
1419         try:
1420             send_pkts = self.gen_encrypt_pkts(
1421                 p,
1422                 p.scapy_tun_sa,
1423                 self.tun_if,
1424                 src=p.remote_tun_if_host,
1425                 dst=self.pg1.remote_ip4,
1426                 payload_size=1900,
1427                 count=1,
1428             )
1429             send_pkts = fragment_rfc791(send_pkts[0], 1400)
1430             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1, n_rx=1)
1431             self.verify_decrypted(p, recv_pkts)
1432
1433             send_pkts = self.gen_pkts(
1434                 self.pg1, src=self.pg1.remote_ip4, dst=p.remote_tun_if_host, count=1
1435             )
1436             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1437             self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1438
1439         finally:
1440             self.logger.info(self.vapi.ppcli("show error"))
1441             self.logger.info(self.vapi.ppcli("show ipsec all"))
1442
1443         self.verify_counters4(p, 1, 1)
1444         self.vapi.ip_reassembly_enable_disable(
1445             sw_if_index=self.tun_if.sw_if_index, enable_ip4=False
1446         )
1447
1448     def verify_tun_64(self, p, count=1):
1449         self.vapi.cli("clear errors")
1450         self.vapi.cli("clear ipsec sa")
1451         try:
1452             send_pkts = self.gen_encrypt_pkts6(
1453                 p,
1454                 p.scapy_tun_sa,
1455                 self.tun_if,
1456                 src=p.remote_tun_if_host6,
1457                 dst=self.pg1.remote_ip6,
1458                 count=count,
1459             )
1460             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1461             for recv_pkt in recv_pkts:
1462                 self.assert_equal(recv_pkt[IPv6].src, p.remote_tun_if_host6)
1463                 self.assert_equal(recv_pkt[IPv6].dst, self.pg1.remote_ip6)
1464                 self.assert_packet_checksums_valid(recv_pkt)
1465             send_pkts = self.gen_pkts6(
1466                 p,
1467                 self.pg1,
1468                 src=self.pg1.remote_ip6,
1469                 dst=p.remote_tun_if_host6,
1470                 count=count,
1471             )
1472             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1473             for recv_pkt in recv_pkts:
1474                 try:
1475                     decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IP])
1476                     if not decrypt_pkt.haslayer(IPv6):
1477                         decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1478                     self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1479                     self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host6)
1480                     self.assert_packet_checksums_valid(decrypt_pkt)
1481                 except:
1482                     self.logger.error(ppp("Unexpected packet:", recv_pkt))
1483                     try:
1484                         self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1485                     except:
1486                         pass
1487                     raise
1488         finally:
1489             self.logger.info(self.vapi.ppcli("show error"))
1490             self.logger.info(self.vapi.ppcli("show ipsec all"))
1491
1492         self.verify_counters4(p, count)
1493
1494     def verify_keepalive(self, p):
1495         # the sizeof Raw is calculated to pad to the minimum ehternet
1496         # frame size of 64 btyes
1497         pkt = (
1498             Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1499             / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1500             / UDP(sport=333, dport=4500)
1501             / Raw(b"\xff")
1502             / Padding(0 * 21)
1503         )
1504         self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1505         self.assert_error_counter_equal(
1506             "/err/%s/nat_keepalive" % self.tun4_input_node, 31
1507         )
1508
1509         pkt = (
1510             Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1511             / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1512             / UDP(sport=333, dport=4500)
1513             / Raw(b"\xfe")
1514         )
1515         self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1516         self.assert_error_counter_equal("/err/%s/too_short" % self.tun4_input_node, 31)
1517
1518         pkt = (
1519             Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1520             / IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4)
1521             / UDP(sport=333, dport=4500)
1522             / Raw(b"\xfe")
1523             / Padding(0 * 21)
1524         )
1525         self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1526         self.assert_error_counter_equal("/err/%s/too_short" % self.tun4_input_node, 62)
1527
1528
1529 class IpsecTun4Tests(IpsecTun4):
1530     """UT test methods for Tunnel v4"""
1531
1532     def test_tun_basic44(self):
1533         """ipsec 4o4 tunnel basic test"""
1534         self.verify_tun_44(self.params[socket.AF_INET], count=1)
1535         self.tun_if.admin_down()
1536         self.tun_if.resolve_arp()
1537         self.tun_if.admin_up()
1538         self.verify_tun_44(self.params[socket.AF_INET], count=1)
1539
1540     def test_tun_reass_basic44(self):
1541         """ipsec 4o4 tunnel basic reassembly test"""
1542         self.verify_tun_reass_44(self.params[socket.AF_INET])
1543
1544     def test_tun_burst44(self):
1545         """ipsec 4o4 tunnel burst test"""
1546         self.verify_tun_44(self.params[socket.AF_INET], count=127)
1547
1548
1549 class IpsecTun6(object):
1550     """verify methods for Tunnel v6"""
1551
1552     def verify_counters6(self, p_in, p_out, count, worker=None):
1553         if hasattr(p_in, "tun_sa_in"):
1554             pkts = p_in.tun_sa_in.get_stats(worker)["packets"]
1555             self.assertEqual(
1556                 pkts, count, "incorrect SA in counts: expected %d != %d" % (count, pkts)
1557             )
1558         if hasattr(p_out, "tun_sa_out"):
1559             pkts = p_out.tun_sa_out.get_stats(worker)["packets"]
1560             self.assertEqual(
1561                 pkts,
1562                 count,
1563                 "incorrect SA out counts: expected %d != %d" % (count, pkts),
1564             )
1565         self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count)
1566         self.assert_packet_counter_equal(self.tun6_decrypt_node_name[0], count)
1567
1568     def verify_decrypted6(self, p, rxs):
1569         for rx in rxs:
1570             self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1571             self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1572             self.assert_packet_checksums_valid(rx)
1573
1574     def verify_encrypted6(self, p, sa, rxs):
1575         for rx in rxs:
1576             self.assert_packet_checksums_valid(rx)
1577             self.assertEqual(len(rx) - len(Ether()) - len(IPv6()), rx[IPv6].plen)
1578             self.assert_equal(rx[IPv6].hlim, p.outer_hop_limit)
1579             if p.outer_flow_label:
1580                 self.assert_equal(rx[IPv6].fl, p.outer_flow_label)
1581             try:
1582                 decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IPv6])
1583                 if not decrypt_pkt.haslayer(IPv6):
1584                     decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1585                 self.assert_packet_checksums_valid(decrypt_pkt)
1586                 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1587                 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1588                 self.assert_equal(decrypt_pkt.hlim, p.inner_hop_limit - 1)
1589                 self.assert_equal(decrypt_pkt.fl, p.inner_flow_label)
1590             except:
1591                 self.logger.debug(ppp("Unexpected packet:", rx))
1592                 try:
1593                     self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1594                 except:
1595                     pass
1596                 raise
1597
1598     def verify_drop_tun_tx_66(self, p_in, count=1, payload_size=64):
1599         self.vapi.cli("clear errors")
1600         self.vapi.cli("clear ipsec sa")
1601
1602         send_pkts = self.gen_pkts6(
1603             p_in,
1604             self.pg1,
1605             src=self.pg1.remote_ip6,
1606             dst=p_in.remote_tun_if_host,
1607             count=count,
1608             payload_size=payload_size,
1609         )
1610         self.send_and_assert_no_replies(self.tun_if, send_pkts)
1611         self.logger.info(self.vapi.cli("sh punt stats"))
1612
1613     def verify_drop_tun_rx_66(self, p_in, count=1, payload_size=64):
1614         self.vapi.cli("clear errors")
1615         self.vapi.cli("clear ipsec sa")
1616
1617         send_pkts = self.gen_encrypt_pkts6(
1618             p_in,
1619             p_in.scapy_tun_sa,
1620             self.tun_if,
1621             src=p_in.remote_tun_if_host,
1622             dst=self.pg1.remote_ip6,
1623             count=count,
1624         )
1625         self.send_and_assert_no_replies(self.tun_if, send_pkts)
1626
1627     def verify_drop_tun_66(self, p_in, count=1, payload_size=64):
1628         self.verify_drop_tun_tx_66(p_in, count=count, payload_size=payload_size)
1629         self.verify_drop_tun_rx_66(p_in, count=count, payload_size=payload_size)
1630
1631     def verify_tun_66(self, p_in, p_out=None, count=1, payload_size=64):
1632         self.vapi.cli("clear errors")
1633         self.vapi.cli("clear ipsec sa")
1634         if not p_out:
1635             p_out = p_in
1636         try:
1637             send_pkts = self.gen_encrypt_pkts6(
1638                 p_in,
1639                 p_in.scapy_tun_sa,
1640                 self.tun_if,
1641                 src=p_in.remote_tun_if_host,
1642                 dst=self.pg1.remote_ip6,
1643                 count=count,
1644                 payload_size=payload_size,
1645             )
1646             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1647             self.verify_decrypted6(p_in, recv_pkts)
1648
1649             send_pkts = self.gen_pkts6(
1650                 p_in,
1651                 self.pg1,
1652                 src=self.pg1.remote_ip6,
1653                 dst=p_out.remote_tun_if_host,
1654                 count=count,
1655                 payload_size=payload_size,
1656             )
1657             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1658             self.verify_encrypted6(p_out, p_out.vpp_tun_sa, recv_pkts)
1659
1660             for rx in recv_pkts:
1661                 self.assertEqual(rx[IPv6].src, p_out.tun_src)
1662                 self.assertEqual(rx[IPv6].dst, p_out.tun_dst)
1663
1664         finally:
1665             self.logger.info(self.vapi.ppcli("show error"))
1666             self.logger.info(self.vapi.ppcli("show ipsec all"))
1667         self.verify_counters6(p_in, p_out, count)
1668
1669     def verify_tun_reass_66(self, p):
1670         self.vapi.cli("clear errors")
1671         self.vapi.ip_reassembly_enable_disable(
1672             sw_if_index=self.tun_if.sw_if_index, enable_ip6=True
1673         )
1674
1675         try:
1676             send_pkts = self.gen_encrypt_pkts6(
1677                 p,
1678                 p.scapy_tun_sa,
1679                 self.tun_if,
1680                 src=p.remote_tun_if_host,
1681                 dst=self.pg1.remote_ip6,
1682                 count=1,
1683                 payload_size=1850,
1684             )
1685             send_pkts = fragment_rfc8200(send_pkts[0], 1, 1400, self.logger)
1686             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1, n_rx=1)
1687             self.verify_decrypted6(p, recv_pkts)
1688
1689             send_pkts = self.gen_pkts6(
1690                 p,
1691                 self.pg1,
1692                 src=self.pg1.remote_ip6,
1693                 dst=p.remote_tun_if_host,
1694                 count=1,
1695                 payload_size=64,
1696             )
1697             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1698             self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1699         finally:
1700             self.logger.info(self.vapi.ppcli("show error"))
1701             self.logger.info(self.vapi.ppcli("show ipsec all"))
1702         self.verify_counters6(p, p, 1)
1703         self.vapi.ip_reassembly_enable_disable(
1704             sw_if_index=self.tun_if.sw_if_index, enable_ip6=False
1705         )
1706
1707     def verify_tun_46(self, p, count=1):
1708         """ipsec 4o6 tunnel basic test"""
1709         self.vapi.cli("clear errors")
1710         self.vapi.cli("clear ipsec sa")
1711         try:
1712             send_pkts = self.gen_encrypt_pkts(
1713                 p,
1714                 p.scapy_tun_sa,
1715                 self.tun_if,
1716                 src=p.remote_tun_if_host4,
1717                 dst=self.pg1.remote_ip4,
1718                 count=count,
1719             )
1720             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1721             for recv_pkt in recv_pkts:
1722                 self.assert_equal(recv_pkt[IP].src, p.remote_tun_if_host4)
1723                 self.assert_equal(recv_pkt[IP].dst, self.pg1.remote_ip4)
1724                 self.assert_packet_checksums_valid(recv_pkt)
1725             send_pkts = self.gen_pkts(
1726                 self.pg1,
1727                 src=self.pg1.remote_ip4,
1728                 dst=p.remote_tun_if_host4,
1729                 count=count,
1730             )
1731             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1732             for recv_pkt in recv_pkts:
1733                 try:
1734                     decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IPv6])
1735                     if not decrypt_pkt.haslayer(IP):
1736                         decrypt_pkt = IP(decrypt_pkt[Raw].load)
1737                     self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1738                     self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host4)
1739                     self.assert_packet_checksums_valid(decrypt_pkt)
1740                 except:
1741                     self.logger.debug(ppp("Unexpected packet:", recv_pkt))
1742                     try:
1743                         self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1744                     except:
1745                         pass
1746                     raise
1747         finally:
1748             self.logger.info(self.vapi.ppcli("show error"))
1749             self.logger.info(self.vapi.ppcli("show ipsec all"))
1750         self.verify_counters6(p, p, count)
1751
1752     def verify_keepalive(self, p):
1753         # the sizeof Raw is calculated to pad to the minimum ehternet
1754         # frame size of 64 btyes
1755         pkt = (
1756             Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1757             / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
1758             / UDP(sport=333, dport=4500)
1759             / Raw(b"\xff")
1760             / Padding(0 * 1)
1761         )
1762         self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1763         self.assert_error_counter_equal(
1764             "/err/%s/nat_keepalive" % self.tun6_input_node, 31
1765         )
1766
1767         pkt = (
1768             Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1769             / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
1770             / UDP(sport=333, dport=4500)
1771             / Raw(b"\xfe")
1772         )
1773         self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1774         self.assert_error_counter_equal("/err/%s/too_short" % self.tun6_input_node, 31)
1775
1776         pkt = (
1777             Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac)
1778             / IPv6(src=p.remote_tun_if_host, dst=self.tun_if.local_ip6)
1779             / UDP(sport=333, dport=4500)
1780             / Raw(b"\xfe")
1781             / Padding(0 * 21)
1782         )
1783         self.send_and_assert_no_replies(self.tun_if, pkt * 31)
1784         self.assert_error_counter_equal("/err/%s/too_short" % self.tun6_input_node, 62)
1785
1786
1787 class IpsecTun6Tests(IpsecTun6):
1788     """UT test methods for Tunnel v6"""
1789
1790     def test_tun_basic66(self):
1791         """ipsec 6o6 tunnel basic test"""
1792         self.verify_tun_66(self.params[socket.AF_INET6], count=1)
1793
1794     def test_tun_reass_basic66(self):
1795         """ipsec 6o6 tunnel basic reassembly test"""
1796         self.verify_tun_reass_66(self.params[socket.AF_INET6])
1797
1798     def test_tun_burst66(self):
1799         """ipsec 6o6 tunnel burst test"""
1800         self.verify_tun_66(self.params[socket.AF_INET6], count=257)
1801
1802
1803 class IpsecTun6HandoffTests(IpsecTun6):
1804     """UT test methods for Tunnel v6 with multiple workers"""
1805
1806     vpp_worker_count = 2
1807
1808     def test_tun_handoff_66(self):
1809         """ipsec 6o6 tunnel worker hand-off test"""
1810         self.vapi.cli("clear errors")
1811         self.vapi.cli("clear ipsec sa")
1812
1813         N_PKTS = 15
1814         p = self.params[socket.AF_INET6]
1815
1816         # inject alternately on worker 0 and 1. all counts on the SA
1817         # should be against worker 0
1818         for worker in [0, 1, 0, 1]:
1819             send_pkts = self.gen_encrypt_pkts6(
1820                 p,
1821                 p.scapy_tun_sa,
1822                 self.tun_if,
1823                 src=p.remote_tun_if_host,
1824                 dst=self.pg1.remote_ip6,
1825                 count=N_PKTS,
1826             )
1827             recv_pkts = self.send_and_expect(
1828                 self.tun_if, send_pkts, self.pg1, worker=worker
1829             )
1830             self.verify_decrypted6(p, recv_pkts)
1831
1832             send_pkts = self.gen_pkts6(
1833                 p,
1834                 self.pg1,
1835                 src=self.pg1.remote_ip6,
1836                 dst=p.remote_tun_if_host,
1837                 count=N_PKTS,
1838             )
1839             recv_pkts = self.send_and_expect(
1840                 self.pg1, send_pkts, self.tun_if, worker=worker
1841             )
1842             self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1843
1844         # all counts against the first worker that was used
1845         self.verify_counters6(p, p, 4 * N_PKTS, worker=0)
1846
1847
1848 class IpsecTun4HandoffTests(IpsecTun4):
1849     """UT test methods for Tunnel v4 with multiple workers"""
1850
1851     vpp_worker_count = 2
1852
1853     def test_tun_handooff_44(self):
1854         """ipsec 4o4 tunnel worker hand-off test"""
1855         self.vapi.cli("clear errors")
1856         self.vapi.cli("clear ipsec sa")
1857
1858         N_PKTS = 15
1859         p = self.params[socket.AF_INET]
1860
1861         # inject alternately on worker 0 and 1. all counts on the SA
1862         # should be against worker 0
1863         for worker in [0, 1, 0, 1]:
1864             send_pkts = self.gen_encrypt_pkts(
1865                 p,
1866                 p.scapy_tun_sa,
1867                 self.tun_if,
1868                 src=p.remote_tun_if_host,
1869                 dst=self.pg1.remote_ip4,
1870                 count=N_PKTS,
1871             )
1872             recv_pkts = self.send_and_expect(
1873                 self.tun_if, send_pkts, self.pg1, worker=worker
1874             )
1875             self.verify_decrypted(p, recv_pkts)
1876
1877             send_pkts = self.gen_pkts(
1878                 self.pg1,
1879                 src=self.pg1.remote_ip4,
1880                 dst=p.remote_tun_if_host,
1881                 count=N_PKTS,
1882             )
1883             recv_pkts = self.send_and_expect(
1884                 self.pg1, send_pkts, self.tun_if, worker=worker
1885             )
1886             self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1887
1888         # all counts against the first worker that was used
1889         self.verify_counters4(p, 4 * N_PKTS, worker=0)
1890
1891
1892 class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
1893     """UT test methods for Tunnel v6 & v4"""
1894
1895     pass
1896
1897
1898 class IPSecIPv4Fwd(VppTestCase):
1899     """Test IPSec by capturing and verifying IPv4 forwarded pkts"""
1900
1901     @classmethod
1902     def setUpConstants(cls):
1903         super(IPSecIPv4Fwd, cls).setUpConstants()
1904
1905     def setUp(self):
1906         super(IPSecIPv4Fwd, self).setUp()
1907         # store SPD objects so we can remove configs on tear down
1908         self.spd_objs = []
1909         self.spd_policies = []
1910
1911     def tearDown(self):
1912         # remove SPD policies
1913         for obj in self.spd_policies:
1914             obj.remove_vpp_config()
1915         self.spd_policies = []
1916         # remove SPD items (interface bindings first, then SPD)
1917         for obj in reversed(self.spd_objs):
1918             obj.remove_vpp_config()
1919         self.spd_objs = []
1920         # close down pg intfs
1921         for pg in self.pg_interfaces:
1922             pg.unconfig_ip4()
1923             pg.admin_down()
1924         super(IPSecIPv4Fwd, self).tearDown()
1925
1926     def create_interfaces(self, num_ifs=2):
1927         # create interfaces pg0 ... pg<num_ifs>
1928         self.create_pg_interfaces(range(num_ifs))
1929         for pg in self.pg_interfaces:
1930             # put the interface up
1931             pg.admin_up()
1932             # configure IPv4 address on the interface
1933             pg.config_ip4()
1934             # resolve ARP, so that we know VPP MAC
1935             pg.resolve_arp()
1936         self.logger.info(self.vapi.ppcli("show int addr"))
1937
1938     def spd_create_and_intf_add(self, spd_id, pg_list):
1939         spd = VppIpsecSpd(self, spd_id)
1940         spd.add_vpp_config()
1941         self.spd_objs.append(spd)
1942         for pg in pg_list:
1943             spdItf = VppIpsecSpdItfBinding(self, spd, pg)
1944             spdItf.add_vpp_config()
1945             self.spd_objs.append(spdItf)
1946
1947     def get_policy(self, policy_type):
1948         e = VppEnum.vl_api_ipsec_spd_action_t
1949         if policy_type == "protect":
1950             return e.IPSEC_API_SPD_ACTION_PROTECT
1951         elif policy_type == "bypass":
1952             return e.IPSEC_API_SPD_ACTION_BYPASS
1953         elif policy_type == "discard":
1954             return e.IPSEC_API_SPD_ACTION_DISCARD
1955         else:
1956             raise Exception("Invalid policy type: %s", policy_type)
1957
1958     def spd_add_rem_policy(
1959         self,
1960         spd_id,
1961         src_if,
1962         dst_if,
1963         proto,
1964         is_out,
1965         priority,
1966         policy_type,
1967         remove=False,
1968         all_ips=False,
1969         ip_range=False,
1970         local_ip_start=ip_address("0.0.0.0"),
1971         local_ip_stop=ip_address("255.255.255.255"),
1972         remote_ip_start=ip_address("0.0.0.0"),
1973         remote_ip_stop=ip_address("255.255.255.255"),
1974         remote_port_start=0,
1975         remote_port_stop=65535,
1976         local_port_start=0,
1977         local_port_stop=65535,
1978     ):
1979         spd = VppIpsecSpd(self, spd_id)
1980
1981         if all_ips:
1982             src_range_low = ip_address("0.0.0.0")
1983             src_range_high = ip_address("255.255.255.255")
1984             dst_range_low = ip_address("0.0.0.0")
1985             dst_range_high = ip_address("255.255.255.255")
1986
1987         elif ip_range:
1988             src_range_low = local_ip_start
1989             src_range_high = local_ip_stop
1990             dst_range_low = remote_ip_start
1991             dst_range_high = remote_ip_stop
1992
1993         else:
1994             src_range_low = src_if.remote_ip4
1995             src_range_high = src_if.remote_ip4
1996             dst_range_low = dst_if.remote_ip4
1997             dst_range_high = dst_if.remote_ip4
1998
1999         spdEntry = VppIpsecSpdEntry(
2000             self,
2001             spd,
2002             0,
2003             src_range_low,
2004             src_range_high,
2005             dst_range_low,
2006             dst_range_high,
2007             proto,
2008             priority=priority,
2009             policy=self.get_policy(policy_type),
2010             is_outbound=is_out,
2011             remote_port_start=remote_port_start,
2012             remote_port_stop=remote_port_stop,
2013             local_port_start=local_port_start,
2014             local_port_stop=local_port_stop,
2015         )
2016
2017         if remove is False:
2018             spdEntry.add_vpp_config()
2019             self.spd_policies.append(spdEntry)
2020         else:
2021             spdEntry.remove_vpp_config()
2022             self.spd_policies.remove(spdEntry)
2023         self.logger.info(self.vapi.ppcli("show ipsec all"))
2024         return spdEntry
2025
2026     def create_stream(self, src_if, dst_if, pkt_count, src_prt=1234, dst_prt=5678):
2027         packets = []
2028         for i in range(pkt_count):
2029             # create packet info stored in the test case instance
2030             info = self.create_packet_info(src_if, dst_if)
2031             # convert the info into packet payload
2032             payload = self.info_to_payload(info)
2033             # create the packet itself
2034             p = (
2035                 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
2036                 / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
2037                 / UDP(sport=src_prt, dport=dst_prt)
2038                 / Raw(payload)
2039             )
2040             # store a copy of the packet in the packet info
2041             info.data = p.copy()
2042             # append the packet to the list
2043             packets.append(p)
2044         # return the created packet list
2045         return packets
2046
2047     def verify_capture(self, src_if, dst_if, capture):
2048         packet_info = None
2049         for packet in capture:
2050             try:
2051                 ip = packet[IP]
2052                 udp = packet[UDP]
2053                 # convert the payload to packet info object
2054                 payload_info = self.payload_to_info(packet)
2055                 # make sure the indexes match
2056                 self.assert_equal(
2057                     payload_info.src, src_if.sw_if_index, "source sw_if_index"
2058                 )
2059                 self.assert_equal(
2060                     payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
2061                 )
2062                 packet_info = self.get_next_packet_info_for_interface2(
2063                     src_if.sw_if_index, dst_if.sw_if_index, packet_info
2064                 )
2065                 # make sure we didn't run out of saved packets
2066                 self.assertIsNotNone(packet_info)
2067                 self.assert_equal(
2068                     payload_info.index, packet_info.index, "packet info index"
2069                 )
2070                 saved_packet = packet_info.data  # fetch the saved packet
2071                 # assert the values match
2072                 self.assert_equal(ip.src, saved_packet[IP].src, "IP source address")
2073                 # ... more assertions here
2074                 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
2075             except Exception as e:
2076                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2077                 raise
2078         remaining_packet = self.get_next_packet_info_for_interface2(
2079             src_if.sw_if_index, dst_if.sw_if_index, packet_info
2080         )
2081         self.assertIsNone(
2082             remaining_packet,
2083             "Interface %s: Packet expected from interface "
2084             "%s didn't arrive" % (dst_if.name, src_if.name),
2085         )
2086
2087     def verify_policy_match(self, pkt_count, spdEntry):
2088         self.logger.info("XXXX %s %s", str(spdEntry), str(spdEntry.get_stats()))
2089         matched_pkts = spdEntry.get_stats().get("packets")
2090         self.logger.info("Policy %s matched: %d pkts", str(spdEntry), matched_pkts)
2091         self.assert_equal(pkt_count, matched_pkts)
2092
2093
2094 class SpdFlowCacheTemplate(IPSecIPv4Fwd):
2095     @classmethod
2096     def setUpConstants(cls):
2097         super(SpdFlowCacheTemplate, cls).setUpConstants()
2098         # Override this method with required cmdline parameters e.g.
2099         # cls.vpp_cmdline.extend(["ipsec", "{",
2100         #                         "ipv4-outbound-spd-flow-cache on",
2101         #                         "}"])
2102         # cls.logger.info("VPP modified cmdline is %s" % " "
2103         #                 .join(cls.vpp_cmdline))
2104
2105     def setUp(self):
2106         super(SpdFlowCacheTemplate, self).setUp()
2107
2108     def tearDown(self):
2109         super(SpdFlowCacheTemplate, self).tearDown()
2110
2111     def get_spd_flow_cache_entries(self, outbound):
2112         """'show ipsec spd' output:
2113         ipv4-inbound-spd-flow-cache-entries: 0
2114         ipv4-outbound-spd-flow-cache-entries: 0
2115         """
2116         show_ipsec_reply = self.vapi.cli("show ipsec spd")
2117         # match the relevant section of 'show ipsec spd' output
2118         if outbound:
2119             regex_match = re.search(
2120                 "ipv4-outbound-spd-flow-cache-entries: (.*)",
2121                 show_ipsec_reply,
2122                 re.DOTALL,
2123             )
2124         else:
2125             regex_match = re.search(
2126                 "ipv4-inbound-spd-flow-cache-entries: (.*)", show_ipsec_reply, re.DOTALL
2127             )
2128         if regex_match is None:
2129             raise Exception(
2130                 "Unable to find spd flow cache entries \
2131                 in 'show ipsec spd' CLI output - regex failed to match"
2132             )
2133         else:
2134             try:
2135                 num_entries = int(regex_match.group(1))
2136             except ValueError:
2137                 raise Exception(
2138                     "Unable to get spd flow cache entries \
2139                 from 'show ipsec spd' string: %s",
2140                     regex_match.group(0),
2141                 )
2142             self.logger.info("%s", regex_match.group(0))
2143         return num_entries
2144
2145     def verify_num_outbound_flow_cache_entries(self, expected_elements):
2146         self.assertEqual(
2147             self.get_spd_flow_cache_entries(outbound=True), expected_elements
2148         )
2149
2150     def verify_num_inbound_flow_cache_entries(self, expected_elements):
2151         self.assertEqual(
2152             self.get_spd_flow_cache_entries(outbound=False), expected_elements
2153         )
2154
2155     def crc32_supported(self):
2156         # lscpu is part of util-linux package, available on all Linux Distros
2157         stream = os.popen("lscpu")
2158         cpu_info = stream.read()
2159         # feature/flag "crc32" on Aarch64 and "sse4_2" on x86
2160         # see vppinfra/crc32.h
2161         if "crc32" or "sse4_2" in cpu_info:
2162             self.logger.info("\ncrc32 supported:\n" + cpu_info)
2163             return True
2164         else:
2165             self.logger.info("\ncrc32 NOT supported:\n" + cpu_info)
2166             return False
2167
2168
2169 class IPSecIPv6Fwd(VppTestCase):
2170     """Test IPSec by capturing and verifying IPv6 forwarded pkts"""
2171
2172     @classmethod
2173     def setUpConstants(cls):
2174         super(IPSecIPv6Fwd, cls).setUpConstants()
2175
2176     def setUp(self):
2177         super(IPSecIPv6Fwd, self).setUp()
2178         # store SPD objects so we can remove configs on tear down
2179         self.spd_objs = []
2180         self.spd_policies = []
2181
2182     def tearDown(self):
2183         # remove SPD policies
2184         for obj in self.spd_policies:
2185             obj.remove_vpp_config()
2186         self.spd_policies = []
2187         # remove SPD items (interface bindings first, then SPD)
2188         for obj in reversed(self.spd_objs):
2189             obj.remove_vpp_config()
2190         self.spd_objs = []
2191         # close down pg intfs
2192         for pg in self.pg_interfaces:
2193             pg.unconfig_ip6()
2194             pg.admin_down()
2195         super(IPSecIPv6Fwd, self).tearDown()
2196
2197     def create_interfaces(self, num_ifs=2):
2198         # create interfaces pg0 ... pg<num_ifs>
2199         self.create_pg_interfaces(range(num_ifs))
2200         for pg in self.pg_interfaces:
2201             # put the interface up
2202             pg.admin_up()
2203             # configure IPv6 address on the interface
2204             pg.config_ip6()
2205             pg.resolve_ndp()
2206         self.logger.info(self.vapi.ppcli("show int addr"))
2207
2208     def spd_create_and_intf_add(self, spd_id, pg_list):
2209         spd = VppIpsecSpd(self, spd_id)
2210         spd.add_vpp_config()
2211         self.spd_objs.append(spd)
2212         for pg in pg_list:
2213             spdItf = VppIpsecSpdItfBinding(self, spd, pg)
2214             spdItf.add_vpp_config()
2215             self.spd_objs.append(spdItf)
2216
2217     def get_policy(self, policy_type):
2218         e = VppEnum.vl_api_ipsec_spd_action_t
2219         if policy_type == "protect":
2220             return e.IPSEC_API_SPD_ACTION_PROTECT
2221         elif policy_type == "bypass":
2222             return e.IPSEC_API_SPD_ACTION_BYPASS
2223         elif policy_type == "discard":
2224             return e.IPSEC_API_SPD_ACTION_DISCARD
2225         else:
2226             raise Exception("Invalid policy type: %s", policy_type)
2227
2228     def spd_add_rem_policy(
2229         self,
2230         spd_id,
2231         src_if,
2232         dst_if,
2233         proto,
2234         is_out,
2235         priority,
2236         policy_type,
2237         remove=False,
2238         all_ips=False,
2239         ip_range=False,
2240         local_ip_start=ip_address("0::0"),
2241         local_ip_stop=ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"),
2242         remote_ip_start=ip_address("0::0"),
2243         remote_ip_stop=ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"),
2244         remote_port_start=0,
2245         remote_port_stop=65535,
2246         local_port_start=0,
2247         local_port_stop=65535,
2248     ):
2249         spd = VppIpsecSpd(self, spd_id)
2250
2251         if all_ips:
2252             src_range_low = ip_address("0::0")
2253             src_range_high = ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")
2254             dst_range_low = ip_address("0::0")
2255             dst_range_high = ip_address("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")
2256
2257         elif ip_range:
2258             src_range_low = local_ip_start
2259             src_range_high = local_ip_stop
2260             dst_range_low = remote_ip_start
2261             dst_range_high = remote_ip_stop
2262
2263         else:
2264             src_range_low = src_if.remote_ip6
2265             src_range_high = src_if.remote_ip6
2266             dst_range_low = dst_if.remote_ip6
2267             dst_range_high = dst_if.remote_ip6
2268
2269         spdEntry = VppIpsecSpdEntry(
2270             self,
2271             spd,
2272             0,
2273             src_range_low,
2274             src_range_high,
2275             dst_range_low,
2276             dst_range_high,
2277             proto,
2278             priority=priority,
2279             policy=self.get_policy(policy_type),
2280             is_outbound=is_out,
2281             remote_port_start=remote_port_start,
2282             remote_port_stop=remote_port_stop,
2283             local_port_start=local_port_start,
2284             local_port_stop=local_port_stop,
2285         )
2286
2287         if remove is False:
2288             spdEntry.add_vpp_config()
2289             self.spd_policies.append(spdEntry)
2290         else:
2291             spdEntry.remove_vpp_config()
2292             self.spd_policies.remove(spdEntry)
2293         self.logger.info(self.vapi.ppcli("show ipsec all"))
2294         return spdEntry
2295
2296     def create_stream(self, src_if, dst_if, pkt_count, src_prt=1234, dst_prt=5678):
2297         packets = []
2298         for i in range(pkt_count):
2299             # create packet info stored in the test case instance
2300             info = self.create_packet_info(src_if, dst_if)
2301             # convert the info into packet payload
2302             payload = self.info_to_payload(info)
2303             # create the packet itself
2304             p = (
2305                 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
2306                 / IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)
2307                 / UDP(sport=src_prt, dport=dst_prt)
2308                 / Raw(payload)
2309             )
2310             # store a copy of the packet in the packet info
2311             info.data = p.copy()
2312             # append the packet to the list
2313             packets.append(p)
2314         # return the created packet list
2315         return packets
2316
2317     def verify_capture(self, src_if, dst_if, capture):
2318         packet_info = None
2319         for packet in capture:
2320             try:
2321                 ip = packet[IPv6]
2322                 udp = packet[UDP]
2323                 # convert the payload to packet info object
2324                 payload_info = self.payload_to_info(packet)
2325                 # make sure the indexes match
2326                 self.assert_equal(
2327                     payload_info.src, src_if.sw_if_index, "source sw_if_index"
2328                 )
2329                 self.assert_equal(
2330                     payload_info.dst, dst_if.sw_if_index, "destination sw_if_index"
2331                 )
2332                 packet_info = self.get_next_packet_info_for_interface2(
2333                     src_if.sw_if_index, dst_if.sw_if_index, packet_info
2334                 )
2335                 # make sure we didn't run out of saved packets
2336                 self.assertIsNotNone(packet_info)
2337                 self.assert_equal(
2338                     payload_info.index, packet_info.index, "packet info index"
2339                 )
2340                 saved_packet = packet_info.data  # fetch the saved packet
2341                 # assert the values match
2342                 self.assert_equal(ip.src, saved_packet[IPv6].src, "IP source address")
2343                 # ... more assertions here
2344                 self.assert_equal(udp.sport, saved_packet[UDP].sport, "UDP source port")
2345             except Exception as e:
2346                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2347                 raise
2348         remaining_packet = self.get_next_packet_info_for_interface2(
2349             src_if.sw_if_index, dst_if.sw_if_index, packet_info
2350         )
2351         self.assertIsNone(
2352             remaining_packet,
2353             "Interface %s: Packet expected from interface "
2354             "%s didn't arrive" % (dst_if.name, src_if.name),
2355         )
2356
2357     def verify_policy_match(self, pkt_count, spdEntry):
2358         self.logger.info("XXXX %s %s", str(spdEntry), str(spdEntry.get_stats()))
2359         matched_pkts = spdEntry.get_stats().get("packets")
2360         self.logger.info("Policy %s matched: %d pkts", str(spdEntry), matched_pkts)
2361         self.assert_equal(pkt_count, matched_pkts)
2362
2363
2364 if __name__ == "__main__":
2365     unittest.main(testRunner=VppTestRunner)