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