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