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
  18 class IPsecIPv4Params:
  19
  20     addr_type = socket.AF_INET
  21     addr_any = "0.0.0.0"
  22     addr_bcast = "255.255.255.255"
  23     addr_len = 32
  24     is_ipv6 = 0
  25
  26     def __init__(self):
  27         self.remote_tun_if_host = '1.1.1.1'
  28         self.remote_tun_if_host6 = '1111::1'
  29
  30         self.scapy_tun_sa_id = 100
  31         self.scapy_tun_spi = 1000
  32         self.vpp_tun_sa_id = 200
  33         self.vpp_tun_spi = 2000
  34
  35         self.scapy_tra_sa_id = 300
  36         self.scapy_tra_spi = 3000
  37         self.vpp_tra_sa_id = 400
  38         self.vpp_tra_spi = 4000
  39
  40         self.outer_hop_limit = 64
  41         self.inner_hop_limit = 255
  42         self.outer_flow_label = 0
  43         self.inner_flow_label = 0x12345
  44
  45         self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
  46                                  IPSEC_API_INTEG_ALG_SHA1_96)
  47         self.auth_algo = 'HMAC-SHA1-96'  # scapy name
  48         self.auth_key = b'C91KUR9GYMm5GfkEvNjX'
  49
  50         self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
  51                                   IPSEC_API_CRYPTO_ALG_AES_CBC_128)
  52         self.crypt_algo = 'AES-CBC'  # scapy name
  53         self.crypt_key = b'JPjyOWBeVEQiMe7h'
  54         self.salt = 0
  55         self.flags = 0
  56         self.nat_header = None
  57         self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
  58                           TUNNEL_API_ENCAP_DECAP_FLAG_NONE)
  59         self.dscp = 0
  60         self.async_mode = False
  61
  62
  63 class IPsecIPv6Params:
  64
  65     addr_type = socket.AF_INET6
  66     addr_any = "0::0"
  67     addr_bcast = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"
  68     addr_len = 128
  69     is_ipv6 = 1
  70
  71     def __init__(self):
  72         self.remote_tun_if_host = '1111:1111:1111:1111:1111:1111:1111:1111'
  73         self.remote_tun_if_host4 = '1.1.1.1'
  74
  75         self.scapy_tun_sa_id = 500
  76         self.scapy_tun_spi = 3001
  77         self.vpp_tun_sa_id = 600
  78         self.vpp_tun_spi = 3000
  79
  80         self.scapy_tra_sa_id = 700
  81         self.scapy_tra_spi = 4001
  82         self.vpp_tra_sa_id = 800
  83         self.vpp_tra_spi = 4000
  84
  85         self.outer_hop_limit = 64
  86         self.inner_hop_limit = 255
  87         self.outer_flow_label = 0
  88         self.inner_flow_label = 0x12345
  89
  90         self.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
  91                                  IPSEC_API_INTEG_ALG_SHA1_96)
  92         self.auth_algo = 'HMAC-SHA1-96'  # scapy name
  93         self.auth_key = b'C91KUR9GYMm5GfkEvNjX'
  94
  95         self.crypt_algo_vpp_id = (VppEnum.vl_api_ipsec_crypto_alg_t.
  96                                   IPSEC_API_CRYPTO_ALG_AES_CBC_128)
  97         self.crypt_algo = 'AES-CBC'  # scapy name
  98         self.crypt_key = b'JPjyOWBeVEQiMe7h'
  99         self.salt = 0
 100         self.flags = 0
 101         self.nat_header = None
 102         self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
 103                           TUNNEL_API_ENCAP_DECAP_FLAG_NONE)
 104         self.dscp = 0
 105         self.async_mode = False
 106
 107
 108 def mk_scapy_crypt_key(p):
 109     if p.crypt_algo in ("AES-GCM", "AES-CTR"):
 110         return p.crypt_key + struct.pack("!I", p.salt)
 111     else:
 112         return p.crypt_key
 113
 114
 115 def config_tun_params(p, encryption_type, tun_if):
 116     ip_class_by_addr_type = {socket.AF_INET: IP, socket.AF_INET6: IPv6}
 117     esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
 118                              IPSEC_API_SAD_FLAG_USE_ESN))
 119     p.tun_dst = tun_if.remote_addr[p.addr_type]
 120     p.tun_src = tun_if.local_addr[p.addr_type]
 121     crypt_key = mk_scapy_crypt_key(p)
 122     p.scapy_tun_sa = SecurityAssociation(
 123         encryption_type, spi=p.vpp_tun_spi,
 124         crypt_algo=p.crypt_algo,
 125         crypt_key=crypt_key,
 126         auth_algo=p.auth_algo, auth_key=p.auth_key,
 127         tunnel_header=ip_class_by_addr_type[p.addr_type](
 128             src=p.tun_dst,
 129             dst=p.tun_src),
 130         nat_t_header=p.nat_header,
 131         esn_en=esn_en)
 132     p.vpp_tun_sa = SecurityAssociation(
 133         encryption_type, spi=p.scapy_tun_spi,
 134         crypt_algo=p.crypt_algo,
 135         crypt_key=crypt_key,
 136         auth_algo=p.auth_algo, auth_key=p.auth_key,
 137         tunnel_header=ip_class_by_addr_type[p.addr_type](
 138             dst=p.tun_dst,
 139             src=p.tun_src),
 140         nat_t_header=p.nat_header,
 141         esn_en=esn_en)
 142
 143
 144 def config_tra_params(p, encryption_type):
 145     esn_en = bool(p.flags & (VppEnum.vl_api_ipsec_sad_flags_t.
 146                              IPSEC_API_SAD_FLAG_USE_ESN))
 147     crypt_key = mk_scapy_crypt_key(p)
 148     p.scapy_tra_sa = SecurityAssociation(
 149         encryption_type,
 150         spi=p.vpp_tra_spi,
 151         crypt_algo=p.crypt_algo,
 152         crypt_key=crypt_key,
 153         auth_algo=p.auth_algo,
 154         auth_key=p.auth_key,
 155         nat_t_header=p.nat_header,
 156         esn_en=esn_en)
 157     p.vpp_tra_sa = SecurityAssociation(
 158         encryption_type,
 159         spi=p.scapy_tra_spi,
 160         crypt_algo=p.crypt_algo,
 161         crypt_key=crypt_key,
 162         auth_algo=p.auth_algo,
 163         auth_key=p.auth_key,
 164         nat_t_header=p.nat_header,
 165         esn_en=esn_en)
 166
 167
 168 class TemplateIpsec(VppTestCase):
 169     """
 170     TRANSPORT MODE:
 171
 172      ------   encrypt   ---
 173     |tra_if| <-------> |VPP|
 174      ------   decrypt   ---
 175
 176     TUNNEL MODE:
 177
 178      ------   encrypt   ---   plain   ---
 179     |tun_if| <-------  |VPP| <------ |pg1|
 180      ------             ---           ---
 181
 182      ------   decrypt   ---   plain   ---
 183     |tun_if| ------->  |VPP| ------> |pg1|
 184      ------             ---           ---
 185     """
 186     tun_spd_id = 1
 187     tra_spd_id = 2
 188
 189     def ipsec_select_backend(self):
 190         """ empty method to be overloaded when necessary """
 191         pass
 192
 193     @classmethod
 194     def setUpClass(cls):
 195         super(TemplateIpsec, cls).setUpClass()
 196
 197     @classmethod
 198     def tearDownClass(cls):
 199         super(TemplateIpsec, cls).tearDownClass()
 200
 201     def setup_params(self):
 202         if not hasattr(self, 'ipv4_params'):
 203             self.ipv4_params = IPsecIPv4Params()
 204         if not hasattr(self, 'ipv6_params'):
 205             self.ipv6_params = IPsecIPv6Params()
 206         self.params = {self.ipv4_params.addr_type: self.ipv4_params,
 207                        self.ipv6_params.addr_type: self.ipv6_params}
 208
 209     def config_interfaces(self):
 210         self.create_pg_interfaces(range(3))
 211         self.interfaces = list(self.pg_interfaces)
 212         for i in self.interfaces:
 213             i.admin_up()
 214             i.config_ip4()
 215             i.resolve_arp()
 216             i.config_ip6()
 217             i.resolve_ndp()
 218
 219     def setUp(self):
 220         super(TemplateIpsec, self).setUp()
 221
 222         self.setup_params()
 223
 224         self.vpp_esp_protocol = (VppEnum.vl_api_ipsec_proto_t.
 225                                  IPSEC_API_PROTO_ESP)
 226         self.vpp_ah_protocol = (VppEnum.vl_api_ipsec_proto_t.
 227                                 IPSEC_API_PROTO_AH)
 228
 229         self.config_interfaces()
 230
 231         self.ipsec_select_backend()
 232
 233     def unconfig_interfaces(self):
 234         for i in self.interfaces:
 235             i.admin_down()
 236             i.unconfig_ip4()
 237             i.unconfig_ip6()
 238
 239     def tearDown(self):
 240         super(TemplateIpsec, self).tearDown()
 241
 242         self.unconfig_interfaces()
 243
 244     def show_commands_at_teardown(self):
 245         self.logger.info(self.vapi.cli("show hardware"))
 246
 247     def gen_encrypt_pkts(self, p, sa, sw_intf, src, dst, count=1,
 248                          payload_size=54):
 249         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 250                 sa.encrypt(IP(src=src, dst=dst) /
 251                            ICMP() / Raw(b'X' * payload_size))
 252                 for i in range(count)]
 253
 254     def gen_encrypt_pkts6(self, p, sa, sw_intf, src, dst, count=1,
 255                           payload_size=54):
 256         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 257                 sa.encrypt(IPv6(src=src, dst=dst,
 258                                 hlim=p.inner_hop_limit,
 259                                 fl=p.inner_flow_label) /
 260                            ICMPv6EchoRequest(id=0, seq=1,
 261                                              data='X' * payload_size))
 262                 for i in range(count)]
 263
 264     def gen_pkts(self, sw_intf, src, dst, count=1, payload_size=54):
 265         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 266                 IP(src=src, dst=dst) / ICMP() / Raw(b'X' * payload_size)
 267                 for i in range(count)]
 268
 269     def gen_pkts6(self, p, sw_intf, src, dst, count=1, payload_size=54):
 270         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 271                 IPv6(src=src, dst=dst,
 272                      hlim=p.inner_hop_limit, fl=p.inner_flow_label) /
 273                 ICMPv6EchoRequest(id=0, seq=1, data='X' * payload_size)
 274                 for i in range(count)]
 275
 276
 277 class IpsecTcp(object):
 278     def verify_tcp_checksum(self):
 279         self.vapi.cli("test http server")
 280         p = self.params[socket.AF_INET]
 281         send = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
 282                 p.scapy_tun_sa.encrypt(IP(src=p.remote_tun_if_host,
 283                                           dst=self.tun_if.local_ip4) /
 284                                        TCP(flags='S', dport=80)))
 285         self.logger.debug(ppp("Sending packet:", send))
 286         recv = self.send_and_expect(self.tun_if, [send], self.tun_if)
 287         recv = recv[0]
 288         decrypted = p.vpp_tun_sa.decrypt(recv[IP])
 289         self.assert_packet_checksums_valid(decrypted)
 290
 291
 292 class IpsecTcpTests(IpsecTcp):
 293     def test_tcp_checksum(self):
 294         """ verify checksum correctness for vpp generated packets """
 295         self.verify_tcp_checksum()
 296
 297
 298 class IpsecTra4(object):
 299     """ verify methods for Transport v4 """
 300     def get_replay_counts(self, p):
 301         replay_node_name = ('/err/%s/SA replayed packet' %
 302                             self.tra4_decrypt_node_name[0])
 303         count = self.statistics.get_err_counter(replay_node_name)
 304
 305         if p.async_mode:
 306             replay_post_node_name = ('/err/%s/SA replayed packet' %
 307                                      self.tra4_decrypt_node_name[p.async_mode])
 308             count += self.statistics.get_err_counter(replay_post_node_name)
 309
 310         return count
 311
 312     def get_hash_failed_counts(self, p):
 313         if ESP == self.encryption_type and p.crypt_algo == "AES-GCM":
 314             hash_failed_node_name = ('/err/%s/ESP decryption failed' %
 315                                      self.tra4_decrypt_node_name[p.async_mode])
 316         else:
 317             hash_failed_node_name = ('/err/%s/Integrity check failed' %
 318                                      self.tra4_decrypt_node_name[p.async_mode])
 319         count = self.statistics.get_err_counter(hash_failed_node_name)
 320
 321         if p.async_mode:
 322             count += self.statistics.get_err_counter(
 323                 '/err/crypto-dispatch/bad-hmac')
 324
 325         return count
 326
 327     def verify_hi_seq_num(self):
 328         p = self.params[socket.AF_INET]
 329         saf = VppEnum.vl_api_ipsec_sad_flags_t
 330         esn_on = p.vpp_tra_sa.esn_en
 331         ar_on = p.flags & saf.IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY
 332
 333         seq_cycle_node_name = \
 334             ('/err/%s/sequence number cycled (packet dropped)' %
 335              self.tra4_encrypt_node_name)
 336         replay_count = self.get_replay_counts(p)
 337         hash_failed_count = self.get_hash_failed_counts(p)
 338         seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
 339
 340         # a few packets so we get the rx seq number above the window size and
 341         # thus can simulate a wrap with an out of window packet
 342         pkts = [(Ether(src=self.tra_if.remote_mac,
 343                        dst=self.tra_if.local_mac) /
 344                  p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 345                                            dst=self.tra_if.local_ip4) /
 346                                         ICMP(),
 347                                         seq_num=seq))
 348                 for seq in range(63, 80)]
 349         recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 350
 351         # these 4 packets will all choose seq-num 0 to decrpyt since none
 352         # are out of window when first checked. however, once #200 has
 353         # decrypted it will move the window to 200 and has #81 is out of
 354         # window. this packet should be dropped.
 355         pkts = [(Ether(src=self.tra_if.remote_mac,
 356                        dst=self.tra_if.local_mac) /
 357                  p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 358                                            dst=self.tra_if.local_ip4) /
 359                                         ICMP(),
 360                                         seq_num=200)),
 361                 (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=81)),
 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=201)),
 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=202))]
 379
 380         # if anti-replay is off then we won't drop #81
 381         n_rx = 3 if ar_on else 4
 382         self.send_and_expect(self.tra_if, pkts, self.tra_if, n_rx=n_rx)
 383         # this packet is one before the wrap
 384         pkts = [(Ether(src=self.tra_if.remote_mac,
 385                        dst=self.tra_if.local_mac) /
 386                  p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 387                                            dst=self.tra_if.local_ip4) /
 388                                         ICMP(),
 389                                         seq_num=203))]
 390         recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 391
 392         # move the window over half way to a wrap
 393         pkts = [(Ether(src=self.tra_if.remote_mac,
 394                        dst=self.tra_if.local_mac) /
 395                  p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 396                                            dst=self.tra_if.local_ip4) /
 397                                         ICMP(),
 398                                         seq_num=0x80000001))]
 399         recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 400
 401         # anti-replay will drop old packets, no anti-replay will not
 402         pkts = [(Ether(src=self.tra_if.remote_mac,
 403                        dst=self.tra_if.local_mac) /
 404                  p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 405                                            dst=self.tra_if.local_ip4) /
 406                                         ICMP(),
 407                                         seq_num=0x44000001))]
 408
 409         if ar_on:
 410             self.send_and_assert_no_replies(self.tra_if, pkts)
 411         else:
 412             recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 413
 414         if esn_on:
 415             #
 416             # validate wrapping the ESN
 417             #
 418
 419             # wrap scapy's TX SA SN
 420             p.scapy_tra_sa.seq_num = 0x100000005
 421
 422             # send a packet that wraps the window for both AR and no AR
 423             pkts = [(Ether(src=self.tra_if.remote_mac,
 424                            dst=self.tra_if.local_mac) /
 425                      p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 426                                                dst=self.tra_if.local_ip4) /
 427                                             ICMP(),
 428                                             seq_num=0x100000005))]
 429
 430             rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 431             for rx in rxs:
 432                 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 433
 434             # move the window forward to half way to the next wrap
 435             pkts = [(Ether(src=self.tra_if.remote_mac,
 436                            dst=self.tra_if.local_mac) /
 437                      p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 438                                                dst=self.tra_if.local_ip4) /
 439                                             ICMP(),
 440                                             seq_num=0x180000005))]
 441
 442             rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 443
 444             # a packet less than 2^30 from the current position is:
 445             #  - AR: out of window and dropped
 446             #  - non-AR: accepted
 447             pkts = [(Ether(src=self.tra_if.remote_mac,
 448                            dst=self.tra_if.local_mac) /
 449                      p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 450                                                dst=self.tra_if.local_ip4) /
 451                                             ICMP(),
 452                                             seq_num=0x170000005))]
 453
 454             if ar_on:
 455                 self.send_and_assert_no_replies(self.tra_if, pkts)
 456             else:
 457                 self.send_and_expect(self.tra_if, pkts, self.tra_if)
 458
 459             # a packet more than 2^30 from the current position is:
 460             #  - AR: out of window and dropped
 461             #  - non-AR: considered a wrap, but since it's not a wrap
 462             #    it won't decrpyt and so will be dropped
 463             pkts = [(Ether(src=self.tra_if.remote_mac,
 464                            dst=self.tra_if.local_mac) /
 465                      p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 466                                                dst=self.tra_if.local_ip4) /
 467                                             ICMP(),
 468                                             seq_num=0x130000005))]
 469
 470             self.send_and_assert_no_replies(self.tra_if, pkts)
 471
 472             # a packet less than 2^30 from the current position and is a
 473             # wrap; (the seq is currently at 0x180000005).
 474             #  - AR: out of window so considered a wrap, so accepted
 475             #  - non-AR: not considered a wrap, so won't decrypt
 476             p.scapy_tra_sa.seq_num = 0x260000005
 477             pkts = [(Ether(src=self.tra_if.remote_mac,
 478                            dst=self.tra_if.local_mac) /
 479                      p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 480                                                dst=self.tra_if.local_ip4) /
 481                                             ICMP(),
 482                                             seq_num=0x260000005))]
 483             if ar_on:
 484                 self.send_and_expect(self.tra_if, pkts, self.tra_if)
 485             else:
 486                 self.send_and_assert_no_replies(self.tra_if, pkts)
 487
 488             #
 489             # window positions are different now for AR/non-AR
 490             #  move non-AR forward
 491             #
 492             if not ar_on:
 493                 # a packet more than 2^30 from the current position and is a
 494                 # wrap; (the seq is currently at 0x180000005).
 495                 #  - AR: accepted
 496                 #  - non-AR: not considered a wrap, so won't decrypt
 497
 498                 pkts = [(Ether(src=self.tra_if.remote_mac,
 499                                dst=self.tra_if.local_mac) /
 500                          p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 501                                                    dst=self.tra_if.local_ip4) /
 502                                                 ICMP(),
 503                                                 seq_num=0x200000005)),
 504                         (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=0x200000006))]
 510                 self.send_and_expect(self.tra_if, pkts, self.tra_if)
 511
 512                 pkts = [(Ether(src=self.tra_if.remote_mac,
 513                                dst=self.tra_if.local_mac) /
 514                          p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 515                                                    dst=self.tra_if.local_ip4) /
 516                                                 ICMP(),
 517                                                 seq_num=0x260000005))]
 518                 self.send_and_expect(self.tra_if, pkts, self.tra_if)
 519
 520     def verify_tra_anti_replay(self):
 521         p = self.params[socket.AF_INET]
 522         esn_en = p.vpp_tra_sa.esn_en
 523
 524         seq_cycle_node_name = \
 525             ('/err/%s/sequence number cycled (packet dropped)' %
 526              self.tra4_encrypt_node_name)
 527         replay_count = self.get_replay_counts(p)
 528         hash_failed_count = self.get_hash_failed_counts(p)
 529         seq_cycle_count = self.statistics.get_err_counter(seq_cycle_node_name)
 530
 531         if ESP == self.encryption_type:
 532             undersize_node_name = ('/err/%s/undersized packet' %
 533                                    self.tra4_decrypt_node_name[0])
 534             undersize_count = self.statistics.get_err_counter(
 535                 undersize_node_name)
 536
 537         #
 538         # send packets with seq numbers 1->34
 539         # this means the window size is still in Case B (see RFC4303
 540         # Appendix A)
 541         #
 542         # for reasons i haven't investigated Scapy won't create a packet with
 543         # seq_num=0
 544         #
 545         pkts = [(Ether(src=self.tra_if.remote_mac,
 546                        dst=self.tra_if.local_mac) /
 547                  p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 548                                            dst=self.tra_if.local_ip4) /
 549                                         ICMP(),
 550                                         seq_num=seq))
 551                 for seq in range(1, 34)]
 552         recv_pkts = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 553
 554         # replayed packets are dropped
 555         self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
 556         replay_count += len(pkts)
 557         self.assertEqual(self.get_replay_counts(p), replay_count)
 558
 559         #
 560         # now send a batch of packets all with the same sequence number
 561         # the first packet in the batch is legitimate, the rest bogus
 562         #
 563         self.vapi.cli("clear error")
 564         self.vapi.cli("clear node counters")
 565         pkts = (Ether(src=self.tra_if.remote_mac,
 566                       dst=self.tra_if.local_mac) /
 567                 p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 568                                           dst=self.tra_if.local_ip4) /
 569                                        ICMP(),
 570                                        seq_num=35))
 571         recv_pkts = self.send_and_expect(self.tra_if, pkts * 8,
 572                                          self.tra_if, n_rx=1)
 573         replay_count += 7
 574         self.assertEqual(self.get_replay_counts(p), replay_count)
 575
 576         #
 577         # now move the window over to 257 (more than one byte) and into Case A
 578         #
 579         self.vapi.cli("clear error")
 580         pkt = (Ether(src=self.tra_if.remote_mac,
 581                      dst=self.tra_if.local_mac) /
 582                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 583                                          dst=self.tra_if.local_ip4) /
 584                                       ICMP(),
 585                                       seq_num=257))
 586         recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 587
 588         # replayed packets are dropped
 589         self.send_and_assert_no_replies(self.tra_if, pkt * 3, timeout=0.2)
 590         replay_count += 3
 591         self.assertEqual(self.get_replay_counts(p), replay_count)
 592
 593         # the window size is 64 packets
 594         # in window are still accepted
 595         pkt = (Ether(src=self.tra_if.remote_mac,
 596                      dst=self.tra_if.local_mac) /
 597                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 598                                          dst=self.tra_if.local_ip4) /
 599                                       ICMP(),
 600                                       seq_num=200))
 601         recv_pkts = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 602
 603         # a packet that does not decrypt does not move the window forward
 604         bogus_sa = SecurityAssociation(self.encryption_type,
 605                                        p.vpp_tra_spi,
 606                                        crypt_algo=p.crypt_algo,
 607                                        crypt_key=mk_scapy_crypt_key(p)[::-1],
 608                                        auth_algo=p.auth_algo,
 609                                        auth_key=p.auth_key[::-1])
 610         pkt = (Ether(src=self.tra_if.remote_mac,
 611                      dst=self.tra_if.local_mac) /
 612                bogus_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 613                                    dst=self.tra_if.local_ip4) /
 614                                 ICMP(),
 615                                 seq_num=350))
 616         self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
 617
 618         hash_failed_count += 17
 619         self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
 620
 621         # a malformed 'runt' packet
 622         #  created by a mis-constructed SA
 623         if (ESP == self.encryption_type and p.crypt_algo != "NULL"):
 624             bogus_sa = SecurityAssociation(self.encryption_type,
 625                                            p.vpp_tra_spi)
 626             pkt = (Ether(src=self.tra_if.remote_mac,
 627                          dst=self.tra_if.local_mac) /
 628                    bogus_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 629                                        dst=self.tra_if.local_ip4) /
 630                                     ICMP(),
 631                                     seq_num=350))
 632             self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
 633
 634             undersize_count += 17
 635             self.assert_error_counter_equal(undersize_node_name,
 636                                             undersize_count)
 637
 638         # which we can determine since this packet is still in the window
 639         pkt = (Ether(src=self.tra_if.remote_mac,
 640                      dst=self.tra_if.local_mac) /
 641                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 642                                          dst=self.tra_if.local_ip4) /
 643                                       ICMP(),
 644                                       seq_num=234))
 645         self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 646
 647         #
 648         # out of window are dropped
 649         #  this is Case B. So VPP will consider this to be a high seq num wrap
 650         #  and so the decrypt attempt will fail
 651         #
 652         pkt = (Ether(src=self.tra_if.remote_mac,
 653                      dst=self.tra_if.local_mac) /
 654                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 655                                          dst=self.tra_if.local_ip4) /
 656                                       ICMP(),
 657                                       seq_num=17))
 658         self.send_and_assert_no_replies(self.tra_if, pkt * 17, timeout=0.2)
 659
 660         if esn_en:
 661             # an out of window error with ESN looks like a high sequence
 662             # wrap. but since it isn't then the verify will fail.
 663             hash_failed_count += 17
 664             self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
 665
 666         else:
 667             replay_count += 17
 668             self.assertEqual(self.get_replay_counts(p), replay_count)
 669
 670         # valid packet moves the window over to 258
 671         pkt = (Ether(src=self.tra_if.remote_mac,
 672                      dst=self.tra_if.local_mac) /
 673                p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 674                                          dst=self.tra_if.local_ip4) /
 675                                       ICMP(),
 676                                       seq_num=258))
 677         rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 678         decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 679
 680         #
 681         # move VPP's SA TX seq-num to just before the seq-number wrap.
 682         # then fire in a packet that VPP should drop on TX because it
 683         # causes the TX seq number to wrap; unless we're using extened sequence
 684         # numbers.
 685         #
 686         self.vapi.cli("test ipsec sa %d seq 0xffffffff" % p.scapy_tra_sa_id)
 687         self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
 688         self.logger.info(self.vapi.ppcli("show ipsec sa 1"))
 689
 690         pkts = [(Ether(src=self.tra_if.remote_mac,
 691                        dst=self.tra_if.local_mac) /
 692                  p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 693                                            dst=self.tra_if.local_ip4) /
 694                                         ICMP(),
 695                                         seq_num=seq))
 696                 for seq in range(259, 280)]
 697
 698         if esn_en:
 699             rxs = self.send_and_expect(self.tra_if, pkts, self.tra_if)
 700
 701             #
 702             # in order for scapy to decrypt its SA's high order number needs
 703             # to wrap
 704             #
 705             p.vpp_tra_sa.seq_num = 0x100000000
 706             for rx in rxs:
 707                 decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 708
 709             #
 710             # wrap scapy's TX high sequence number. VPP is in case B, so it
 711             # will consider this a high seq wrap also.
 712             # The low seq num we set it to will place VPP's RX window in Case A
 713             #
 714             p.scapy_tra_sa.seq_num = 0x100000005
 715             pkt = (Ether(src=self.tra_if.remote_mac,
 716                          dst=self.tra_if.local_mac) /
 717                    p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 718                                              dst=self.tra_if.local_ip4) /
 719                                           ICMP(),
 720                                           seq_num=0x100000005))
 721             rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 722
 723             decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 724
 725             #
 726             # A packet that has seq num between (2^32-64) and 5 is within
 727             # the window
 728             #
 729             p.scapy_tra_sa.seq_num = 0xfffffffd
 730             pkt = (Ether(src=self.tra_if.remote_mac,
 731                          dst=self.tra_if.local_mac) /
 732                    p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 733                                              dst=self.tra_if.local_ip4) /
 734                                           ICMP(),
 735                                           seq_num=0xfffffffd))
 736             rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 737             decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 738
 739             #
 740             # While in case A we cannot wrap the high sequence number again
 741             # because VPP will consider this packet to be one that moves the
 742             # window forward
 743             #
 744             pkt = (Ether(src=self.tra_if.remote_mac,
 745                          dst=self.tra_if.local_mac) /
 746                    p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 747                                              dst=self.tra_if.local_ip4) /
 748                                           ICMP(),
 749                                           seq_num=0x200000999))
 750             self.send_and_assert_no_replies(self.tra_if, [pkt], self.tra_if,
 751                                             timeout=0.2)
 752
 753             hash_failed_count += 1
 754             self.assertEqual(self.get_hash_failed_counts(p), hash_failed_count)
 755
 756             #
 757             # but if we move the window forward to case B, then we can wrap
 758             # again
 759             #
 760             p.scapy_tra_sa.seq_num = 0x100000555
 761             pkt = (Ether(src=self.tra_if.remote_mac,
 762                          dst=self.tra_if.local_mac) /
 763                    p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 764                                              dst=self.tra_if.local_ip4) /
 765                                           ICMP(),
 766                                           seq_num=0x100000555))
 767             rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 768             decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 769
 770             p.scapy_tra_sa.seq_num = 0x200000444
 771             pkt = (Ether(src=self.tra_if.remote_mac,
 772                          dst=self.tra_if.local_mac) /
 773                    p.scapy_tra_sa.encrypt(IP(src=self.tra_if.remote_ip4,
 774                                              dst=self.tra_if.local_ip4) /
 775                                           ICMP(),
 776                                           seq_num=0x200000444))
 777             rx = self.send_and_expect(self.tra_if, [pkt], self.tra_if)
 778             decrypted = p.vpp_tra_sa.decrypt(rx[0][IP])
 779
 780         else:
 781             #
 782             # without ESN TX sequence numbers can't wrap and packets are
 783             # dropped from here on out.
 784             #
 785             self.send_and_assert_no_replies(self.tra_if, pkts, timeout=0.2)
 786             seq_cycle_count += len(pkts)
 787             self.assert_error_counter_equal(seq_cycle_node_name,
 788                                             seq_cycle_count)
 789
 790         # move the security-associations seq number on to the last we used
 791         self.vapi.cli("test ipsec sa %d seq 0x15f" % p.scapy_tra_sa_id)
 792         p.scapy_tra_sa.seq_num = 351
 793         p.vpp_tra_sa.seq_num = 351
 794
 795     def verify_tra_basic4(self, count=1, payload_size=54):
 796         """ ipsec v4 transport basic test """
 797         self.vapi.cli("clear errors")
 798         self.vapi.cli("clear ipsec sa")
 799         try:
 800             p = self.params[socket.AF_INET]
 801             send_pkts = self.gen_encrypt_pkts(p, p.scapy_tra_sa, self.tra_if,
 802                                               src=self.tra_if.remote_ip4,
 803                                               dst=self.tra_if.local_ip4,
 804                                               count=count,
 805                                               payload_size=payload_size)
 806             recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
 807                                              self.tra_if)
 808             for rx in recv_pkts:
 809                 self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
 810                 self.assert_packet_checksums_valid(rx)
 811                 try:
 812                     decrypted = p.vpp_tra_sa.decrypt(rx[IP])
 813                     self.assert_packet_checksums_valid(decrypted)
 814                 except:
 815                     self.logger.debug(ppp("Unexpected packet:", rx))
 816                     raise
 817         finally:
 818             self.logger.info(self.vapi.ppcli("show error"))
 819             self.logger.info(self.vapi.ppcli("show ipsec all"))
 820
 821         pkts = p.tra_sa_in.get_stats()['packets']
 822         self.assertEqual(pkts, count,
 823                          "incorrect SA in counts: expected %d != %d" %
 824                          (count, pkts))
 825         pkts = p.tra_sa_out.get_stats()['packets']
 826         self.assertEqual(pkts, count,
 827                          "incorrect SA out counts: expected %d != %d" %
 828                          (count, pkts))
 829
 830         self.assert_packet_counter_equal(self.tra4_encrypt_node_name, count)
 831         self.assert_packet_counter_equal(self.tra4_decrypt_node_name[0], count)
 832
 833
 834 class IpsecTra4Tests(IpsecTra4):
 835     """ UT test methods for Transport v4 """
 836     def test_tra_anti_replay(self):
 837         """ ipsec v4 transport anti-replay test """
 838         self.verify_tra_anti_replay()
 839
 840     def test_tra_basic(self, count=1):
 841         """ ipsec v4 transport basic test """
 842         self.verify_tra_basic4(count=1)
 843
 844     def test_tra_burst(self):
 845         """ ipsec v4 transport burst test """
 846         self.verify_tra_basic4(count=257)
 847
 848
 849 class IpsecTra6(object):
 850     """ verify methods for Transport v6 """
 851     def verify_tra_basic6(self, count=1, payload_size=54):
 852         self.vapi.cli("clear errors")
 853         self.vapi.cli("clear ipsec sa")
 854         try:
 855             p = self.params[socket.AF_INET6]
 856             send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tra_sa, self.tra_if,
 857                                                src=self.tra_if.remote_ip6,
 858                                                dst=self.tra_if.local_ip6,
 859                                                count=count,
 860                                                payload_size=payload_size)
 861             recv_pkts = self.send_and_expect(self.tra_if, send_pkts,
 862                                              self.tra_if)
 863             for rx in recv_pkts:
 864                 self.assertEqual(len(rx) - len(Ether()) - len(IPv6()),
 865                                  rx[IPv6].plen)
 866                 try:
 867                     decrypted = p.vpp_tra_sa.decrypt(rx[IPv6])
 868                     self.assert_packet_checksums_valid(decrypted)
 869                 except:
 870                     self.logger.debug(ppp("Unexpected packet:", rx))
 871                     raise
 872         finally:
 873             self.logger.info(self.vapi.ppcli("show error"))
 874             self.logger.info(self.vapi.ppcli("show ipsec all"))
 875
 876         pkts = p.tra_sa_in.get_stats()['packets']
 877         self.assertEqual(pkts, count,
 878                          "incorrect SA in counts: expected %d != %d" %
 879                          (count, pkts))
 880         pkts = p.tra_sa_out.get_stats()['packets']
 881         self.assertEqual(pkts, count,
 882                          "incorrect SA out counts: expected %d != %d" %
 883                          (count, pkts))
 884         self.assert_packet_counter_equal(self.tra6_encrypt_node_name, count)
 885         self.assert_packet_counter_equal(self.tra6_decrypt_node_name[0], count)
 886
 887     def gen_encrypt_pkts_ext_hdrs6(self, sa, sw_intf, src, dst, count=1,
 888                                    payload_size=54):
 889         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 890                 sa.encrypt(IPv6(src=src, dst=dst) /
 891                            ICMPv6EchoRequest(id=0, seq=1,
 892                                              data='X' * payload_size))
 893                 for i in range(count)]
 894
 895     def gen_pkts_ext_hdrs6(self, sw_intf, src, dst, count=1, payload_size=54):
 896         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 897                 IPv6(src=src, dst=dst) /
 898                 IPv6ExtHdrHopByHop() /
 899                 IPv6ExtHdrFragment(id=2, offset=200) /
 900                 Raw(b'\xff' * 200)
 901                 for i in range(count)]
 902
 903     def verify_tra_encrypted6(self, p, sa, rxs):
 904         decrypted = []
 905         for rx in rxs:
 906             self.assert_packet_checksums_valid(rx)
 907             try:
 908                 decrypt_pkt = p.vpp_tra_sa.decrypt(rx[IPv6])
 909                 decrypted.append(decrypt_pkt)
 910                 self.assert_equal(decrypt_pkt.src, self.tra_if.local_ip6)
 911                 self.assert_equal(decrypt_pkt.dst, self.tra_if.remote_ip6)
 912             except:
 913                 self.logger.debug(ppp("Unexpected packet:", rx))
 914                 try:
 915                     self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
 916                 except:
 917                     pass
 918                 raise
 919         return decrypted
 920
 921     def verify_tra_66_ext_hdrs(self, p):
 922         count = 63
 923
 924         #
 925         # check we can decrypt with options
 926         #
 927         tx = self.gen_encrypt_pkts_ext_hdrs6(p.scapy_tra_sa, self.tra_if,
 928                                              src=self.tra_if.remote_ip6,
 929                                              dst=self.tra_if.local_ip6,
 930                                              count=count)
 931         self.send_and_expect(self.tra_if, tx, self.tra_if)
 932
 933         #
 934         # injecting a packet from ourselves to be routed of box is a hack
 935         # but it matches an outbout policy, alors je ne regrette rien
 936         #
 937
 938         # one extension before ESP
 939         tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
 940               IPv6(src=self.tra_if.local_ip6,
 941                    dst=self.tra_if.remote_ip6) /
 942               IPv6ExtHdrFragment(id=2, offset=200) /
 943               Raw(b'\xff' * 200))
 944
 945         rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
 946         dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
 947
 948         for dc in dcs:
 949             # for reasons i'm not going to investigate scapy does not
 950             # created the correct headers after decrypt. but reparsing
 951             # the ipv6 packet fixes it
 952             dc = IPv6(raw(dc[IPv6]))
 953             self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
 954
 955         # two extensions before ESP
 956         tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
 957               IPv6(src=self.tra_if.local_ip6,
 958                    dst=self.tra_if.remote_ip6) /
 959               IPv6ExtHdrHopByHop() /
 960               IPv6ExtHdrFragment(id=2, offset=200) /
 961               Raw(b'\xff' * 200))
 962
 963         rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
 964         dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
 965
 966         for dc in dcs:
 967             dc = IPv6(raw(dc[IPv6]))
 968             self.assertTrue(dc[IPv6ExtHdrHopByHop])
 969             self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
 970
 971         # two extensions before ESP, one after
 972         tx = (Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) /
 973               IPv6(src=self.tra_if.local_ip6,
 974                    dst=self.tra_if.remote_ip6) /
 975               IPv6ExtHdrHopByHop() /
 976               IPv6ExtHdrFragment(id=2, offset=200) /
 977               IPv6ExtHdrDestOpt() /
 978               Raw(b'\xff' * 200))
 979
 980         rxs = self.send_and_expect(self.pg2, [tx], self.tra_if)
 981         dcs = self.verify_tra_encrypted6(p, p.vpp_tra_sa, rxs)
 982
 983         for dc in dcs:
 984             dc = IPv6(raw(dc[IPv6]))
 985             self.assertTrue(dc[IPv6ExtHdrDestOpt])
 986             self.assertTrue(dc[IPv6ExtHdrHopByHop])
 987             self.assert_equal(dc[IPv6ExtHdrFragment].id, 2)
 988
 989
 990 class IpsecTra6Tests(IpsecTra6):
 991     """ UT test methods for Transport v6 """
 992     def test_tra_basic6(self):
 993         """ ipsec v6 transport basic test """
 994         self.verify_tra_basic6(count=1)
 995
 996     def test_tra_burst6(self):
 997         """ ipsec v6 transport burst test """
 998         self.verify_tra_basic6(count=257)
 999
1000
1001 class IpsecTra6ExtTests(IpsecTra6):
1002     def test_tra_ext_hdrs_66(self):
1003         """ ipsec 6o6 tra extension headers test """
1004         self.verify_tra_66_ext_hdrs(self.params[socket.AF_INET6])
1005
1006
1007 class IpsecTra46Tests(IpsecTra4Tests, IpsecTra6Tests):
1008     """ UT test methods for Transport v6 and v4"""
1009     pass
1010
1011
1012 class IpsecTun4(object):
1013     """ verify methods for Tunnel v4 """
1014     def verify_counters4(self, p, count, n_frags=None, worker=None):
1015         if not n_frags:
1016             n_frags = count
1017         if (hasattr(p, "spd_policy_in_any")):
1018             pkts = p.spd_policy_in_any.get_stats(worker)['packets']
1019             self.assertEqual(pkts, count,
1020                              "incorrect SPD any policy: expected %d != %d" %
1021                              (count, pkts))
1022
1023         if (hasattr(p, "tun_sa_in")):
1024             pkts = p.tun_sa_in.get_stats(worker)['packets']
1025             self.assertEqual(pkts, count,
1026                              "incorrect SA in counts: expected %d != %d" %
1027                              (count, pkts))
1028             pkts = p.tun_sa_out.get_stats(worker)['packets']
1029             self.assertEqual(pkts, n_frags,
1030                              "incorrect SA out counts: expected %d != %d" %
1031                              (count, pkts))
1032
1033         self.assert_packet_counter_equal(self.tun4_encrypt_node_name, n_frags)
1034         self.assert_packet_counter_equal(self.tun4_decrypt_node_name[0], count)
1035
1036     def verify_decrypted(self, p, rxs):
1037         for rx in rxs:
1038             self.assert_equal(rx[IP].src, p.remote_tun_if_host)
1039             self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
1040             self.assert_packet_checksums_valid(rx)
1041
1042     def verify_esp_padding(self, sa, esp_payload, decrypt_pkt):
1043         align = sa.crypt_algo.block_size
1044         if align < 4:
1045             align = 4
1046         exp_len = (len(decrypt_pkt) + 2 + (align - 1)) & ~(align - 1)
1047         exp_len += sa.crypt_algo.iv_size
1048         exp_len += sa.crypt_algo.icv_size or sa.auth_algo.icv_size
1049         self.assertEqual(exp_len, len(esp_payload))
1050
1051     def verify_encrypted(self, p, sa, rxs):
1052         decrypt_pkts = []
1053         for rx in rxs:
1054             if p.nat_header:
1055                 self.assertEqual(rx[UDP].dport, 4500)
1056             self.assert_packet_checksums_valid(rx)
1057             self.assertEqual(len(rx) - len(Ether()), rx[IP].len)
1058             try:
1059                 rx_ip = rx[IP]
1060                 decrypt_pkt = p.vpp_tun_sa.decrypt(rx_ip)
1061                 if not decrypt_pkt.haslayer(IP):
1062                     decrypt_pkt = IP(decrypt_pkt[Raw].load)
1063                 if rx_ip.proto == socket.IPPROTO_ESP:
1064                     self.verify_esp_padding(sa, rx_ip[ESP].data, decrypt_pkt)
1065                 decrypt_pkts.append(decrypt_pkt)
1066                 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1067                 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1068             except:
1069                 self.logger.debug(ppp("Unexpected packet:", rx))
1070                 try:
1071                     self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1072                 except:
1073                     pass
1074                 raise
1075         pkts = reassemble4(decrypt_pkts)
1076         for pkt in pkts:
1077             self.assert_packet_checksums_valid(pkt)
1078
1079     def verify_tun_44(self, p, count=1, payload_size=64, n_rx=None):
1080         self.vapi.cli("clear errors")
1081         self.vapi.cli("clear ipsec counters")
1082         self.vapi.cli("clear ipsec sa")
1083         if not n_rx:
1084             n_rx = count
1085         try:
1086             send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
1087                                               src=p.remote_tun_if_host,
1088                                               dst=self.pg1.remote_ip4,
1089                                               count=count,
1090                                               payload_size=payload_size)
1091             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1092             self.verify_decrypted(p, recv_pkts)
1093
1094             send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
1095                                       dst=p.remote_tun_if_host, count=count,
1096                                       payload_size=payload_size)
1097             recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1098                                              self.tun_if, n_rx)
1099             self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1100
1101             for rx in recv_pkts:
1102                 self.assertEqual(rx[IP].src, p.tun_src)
1103                 self.assertEqual(rx[IP].dst, p.tun_dst)
1104
1105         finally:
1106             self.logger.info(self.vapi.ppcli("show error"))
1107             self.logger.info(self.vapi.ppcli("show ipsec all"))
1108
1109         self.logger.info(self.vapi.ppcli("show ipsec sa 0"))
1110         self.logger.info(self.vapi.ppcli("show ipsec sa 4"))
1111         self.verify_counters4(p, count, n_rx)
1112
1113     def verify_tun_dropped_44(self, p, count=1, payload_size=64, n_rx=None):
1114         self.vapi.cli("clear errors")
1115         if not n_rx:
1116             n_rx = count
1117         try:
1118             send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
1119                                               src=p.remote_tun_if_host,
1120                                               dst=self.pg1.remote_ip4,
1121                                               count=count)
1122             self.send_and_assert_no_replies(self.tun_if, send_pkts)
1123
1124             send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
1125                                       dst=p.remote_tun_if_host, count=count,
1126                                       payload_size=payload_size)
1127             self.send_and_assert_no_replies(self.pg1, send_pkts)
1128
1129         finally:
1130             self.logger.info(self.vapi.ppcli("show error"))
1131             self.logger.info(self.vapi.ppcli("show ipsec all"))
1132
1133     def verify_tun_reass_44(self, p):
1134         self.vapi.cli("clear errors")
1135         self.vapi.ip_reassembly_enable_disable(
1136             sw_if_index=self.tun_if.sw_if_index, enable_ip4=True)
1137
1138         try:
1139             send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
1140                                               src=p.remote_tun_if_host,
1141                                               dst=self.pg1.remote_ip4,
1142                                               payload_size=1900,
1143                                               count=1)
1144             send_pkts = fragment_rfc791(send_pkts[0], 1400)
1145             recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1146                                              self.pg1, n_rx=1)
1147             self.verify_decrypted(p, recv_pkts)
1148
1149             send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
1150                                       dst=p.remote_tun_if_host, count=1)
1151             recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1152                                              self.tun_if)
1153             self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1154
1155         finally:
1156             self.logger.info(self.vapi.ppcli("show error"))
1157             self.logger.info(self.vapi.ppcli("show ipsec all"))
1158
1159         self.verify_counters4(p, 1, 1)
1160         self.vapi.ip_reassembly_enable_disable(
1161             sw_if_index=self.tun_if.sw_if_index, enable_ip4=False)
1162
1163     def verify_tun_64(self, p, count=1):
1164         self.vapi.cli("clear errors")
1165         self.vapi.cli("clear ipsec sa")
1166         try:
1167             send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
1168                                                src=p.remote_tun_if_host6,
1169                                                dst=self.pg1.remote_ip6,
1170                                                count=count)
1171             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1172             for recv_pkt in recv_pkts:
1173                 self.assert_equal(recv_pkt[IPv6].src, p.remote_tun_if_host6)
1174                 self.assert_equal(recv_pkt[IPv6].dst, self.pg1.remote_ip6)
1175                 self.assert_packet_checksums_valid(recv_pkt)
1176             send_pkts = self.gen_pkts6(p, self.pg1, src=self.pg1.remote_ip6,
1177                                        dst=p.remote_tun_if_host6, count=count)
1178             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1179             for recv_pkt in recv_pkts:
1180                 try:
1181                     decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IP])
1182                     if not decrypt_pkt.haslayer(IPv6):
1183                         decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1184                     self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1185                     self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host6)
1186                     self.assert_packet_checksums_valid(decrypt_pkt)
1187                 except:
1188                     self.logger.error(ppp("Unexpected packet:", recv_pkt))
1189                     try:
1190                         self.logger.debug(
1191                             ppp("Decrypted packet:", decrypt_pkt))
1192                     except:
1193                         pass
1194                     raise
1195         finally:
1196             self.logger.info(self.vapi.ppcli("show error"))
1197             self.logger.info(self.vapi.ppcli("show ipsec all"))
1198
1199         self.verify_counters4(p, count)
1200
1201     def verify_keepalive(self, p):
1202         pkt = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
1203                IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4) /
1204                UDP(sport=333, dport=4500) /
1205                Raw(b'\xff'))
1206         self.send_and_assert_no_replies(self.tun_if, pkt*31)
1207         self.assert_error_counter_equal(
1208             '/err/%s/NAT Keepalive' % self.tun4_input_node, 31)
1209
1210         pkt = (Ether(src=self.tun_if.remote_mac, dst=self.tun_if.local_mac) /
1211                IP(src=p.remote_tun_if_host, dst=self.tun_if.local_ip4) /
1212                UDP(sport=333, dport=4500) /
1213                Raw(b'\xfe'))
1214         self.send_and_assert_no_replies(self.tun_if, pkt*31)
1215         self.assert_error_counter_equal(
1216             '/err/%s/Too Short' % self.tun4_input_node, 31)
1217
1218
1219 class IpsecTun4Tests(IpsecTun4):
1220     """ UT test methods for Tunnel v4 """
1221     def test_tun_basic44(self):
1222         """ ipsec 4o4 tunnel basic test """
1223         self.verify_tun_44(self.params[socket.AF_INET], count=1)
1224         self.tun_if.admin_down()
1225         self.tun_if.resolve_arp()
1226         self.tun_if.admin_up()
1227         self.verify_tun_44(self.params[socket.AF_INET], count=1)
1228
1229     def test_tun_reass_basic44(self):
1230         """ ipsec 4o4 tunnel basic reassembly test """
1231         self.verify_tun_reass_44(self.params[socket.AF_INET])
1232
1233     def test_tun_burst44(self):
1234         """ ipsec 4o4 tunnel burst test """
1235         self.verify_tun_44(self.params[socket.AF_INET], count=127)
1236
1237
1238 class IpsecTun6(object):
1239     """ verify methods for Tunnel v6 """
1240     def verify_counters6(self, p_in, p_out, count, worker=None):
1241         if (hasattr(p_in, "tun_sa_in")):
1242             pkts = p_in.tun_sa_in.get_stats(worker)['packets']
1243             self.assertEqual(pkts, count,
1244                              "incorrect SA in counts: expected %d != %d" %
1245                              (count, pkts))
1246         if (hasattr(p_out, "tun_sa_out")):
1247             pkts = p_out.tun_sa_out.get_stats(worker)['packets']
1248             self.assertEqual(pkts, count,
1249                              "incorrect SA out counts: expected %d != %d" %
1250                              (count, pkts))
1251         self.assert_packet_counter_equal(self.tun6_encrypt_node_name, count)
1252         self.assert_packet_counter_equal(self.tun6_decrypt_node_name[0], count)
1253
1254     def verify_decrypted6(self, p, rxs):
1255         for rx in rxs:
1256             self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1257             self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1258             self.assert_packet_checksums_valid(rx)
1259
1260     def verify_encrypted6(self, p, sa, rxs):
1261         for rx in rxs:
1262             self.assert_packet_checksums_valid(rx)
1263             self.assertEqual(len(rx) - len(Ether()) - len(IPv6()),
1264                              rx[IPv6].plen)
1265             self.assert_equal(rx[IPv6].hlim, p.outer_hop_limit)
1266             if p.outer_flow_label:
1267                 self.assert_equal(rx[IPv6].fl, p.outer_flow_label)
1268             try:
1269                 decrypt_pkt = p.vpp_tun_sa.decrypt(rx[IPv6])
1270                 if not decrypt_pkt.haslayer(IPv6):
1271                     decrypt_pkt = IPv6(decrypt_pkt[Raw].load)
1272                 self.assert_packet_checksums_valid(decrypt_pkt)
1273                 self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip6)
1274                 self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host)
1275                 self.assert_equal(decrypt_pkt.hlim, p.inner_hop_limit - 1)
1276                 self.assert_equal(decrypt_pkt.fl, p.inner_flow_label)
1277             except:
1278                 self.logger.debug(ppp("Unexpected packet:", rx))
1279                 try:
1280                     self.logger.debug(ppp("Decrypted packet:", decrypt_pkt))
1281                 except:
1282                     pass
1283                 raise
1284
1285     def verify_drop_tun_66(self, p_in, count=1, payload_size=64):
1286         self.vapi.cli("clear errors")
1287         self.vapi.cli("clear ipsec sa")
1288
1289         send_pkts = self.gen_encrypt_pkts6(p_in, p_in.scapy_tun_sa,
1290                                            self.tun_if,
1291                                            src=p_in.remote_tun_if_host,
1292                                            dst=self.pg1.remote_ip6,
1293                                            count=count)
1294         self.send_and_assert_no_replies(self.tun_if, send_pkts)
1295         self.logger.info(self.vapi.cli("sh punt stats"))
1296
1297     def verify_tun_66(self, p_in, p_out=None, count=1, payload_size=64):
1298         self.vapi.cli("clear errors")
1299         self.vapi.cli("clear ipsec sa")
1300         if not p_out:
1301             p_out = p_in
1302         try:
1303             send_pkts = self.gen_encrypt_pkts6(p_in, p_in.scapy_tun_sa,
1304                                                self.tun_if,
1305                                                src=p_in.remote_tun_if_host,
1306                                                dst=self.pg1.remote_ip6,
1307                                                count=count,
1308                                                payload_size=payload_size)
1309             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1310             self.verify_decrypted6(p_in, recv_pkts)
1311
1312             send_pkts = self.gen_pkts6(p_in, self.pg1, src=self.pg1.remote_ip6,
1313                                        dst=p_out.remote_tun_if_host,
1314                                        count=count,
1315                                        payload_size=payload_size)
1316             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1317             self.verify_encrypted6(p_out, p_out.vpp_tun_sa, recv_pkts)
1318
1319             for rx in recv_pkts:
1320                 self.assertEqual(rx[IPv6].src, p_out.tun_src)
1321                 self.assertEqual(rx[IPv6].dst, p_out.tun_dst)
1322
1323         finally:
1324             self.logger.info(self.vapi.ppcli("show error"))
1325             self.logger.info(self.vapi.ppcli("show ipsec all"))
1326         self.verify_counters6(p_in, p_out, count)
1327
1328     def verify_tun_reass_66(self, p):
1329         self.vapi.cli("clear errors")
1330         self.vapi.ip_reassembly_enable_disable(
1331             sw_if_index=self.tun_if.sw_if_index, enable_ip6=True)
1332
1333         try:
1334             send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
1335                                                src=p.remote_tun_if_host,
1336                                                dst=self.pg1.remote_ip6,
1337                                                count=1,
1338                                                payload_size=1850)
1339             send_pkts = fragment_rfc8200(send_pkts[0], 1, 1400, self.logger)
1340             recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1341                                              self.pg1, n_rx=1)
1342             self.verify_decrypted6(p, recv_pkts)
1343
1344             send_pkts = self.gen_pkts6(p, self.pg1, src=self.pg1.remote_ip6,
1345                                        dst=p.remote_tun_if_host,
1346                                        count=1,
1347                                        payload_size=64)
1348             recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1349                                              self.tun_if)
1350             self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1351         finally:
1352             self.logger.info(self.vapi.ppcli("show error"))
1353             self.logger.info(self.vapi.ppcli("show ipsec all"))
1354         self.verify_counters6(p, p, 1)
1355         self.vapi.ip_reassembly_enable_disable(
1356             sw_if_index=self.tun_if.sw_if_index, enable_ip6=False)
1357
1358     def verify_tun_46(self, p, count=1):
1359         """ ipsec 4o6 tunnel basic test """
1360         self.vapi.cli("clear errors")
1361         self.vapi.cli("clear ipsec sa")
1362         try:
1363             send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
1364                                               src=p.remote_tun_if_host4,
1365                                               dst=self.pg1.remote_ip4,
1366                                               count=count)
1367             recv_pkts = self.send_and_expect(self.tun_if, send_pkts, self.pg1)
1368             for recv_pkt in recv_pkts:
1369                 self.assert_equal(recv_pkt[IP].src, p.remote_tun_if_host4)
1370                 self.assert_equal(recv_pkt[IP].dst, self.pg1.remote_ip4)
1371                 self.assert_packet_checksums_valid(recv_pkt)
1372             send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
1373                                       dst=p.remote_tun_if_host4,
1374                                       count=count)
1375             recv_pkts = self.send_and_expect(self.pg1, send_pkts, self.tun_if)
1376             for recv_pkt in recv_pkts:
1377                 try:
1378                     decrypt_pkt = p.vpp_tun_sa.decrypt(recv_pkt[IPv6])
1379                     if not decrypt_pkt.haslayer(IP):
1380                         decrypt_pkt = IP(decrypt_pkt[Raw].load)
1381                     self.assert_equal(decrypt_pkt.src, self.pg1.remote_ip4)
1382                     self.assert_equal(decrypt_pkt.dst, p.remote_tun_if_host4)
1383                     self.assert_packet_checksums_valid(decrypt_pkt)
1384                 except:
1385                     self.logger.debug(ppp("Unexpected packet:", recv_pkt))
1386                     try:
1387                         self.logger.debug(ppp("Decrypted packet:",
1388                                               decrypt_pkt))
1389                     except:
1390                         pass
1391                     raise
1392         finally:
1393             self.logger.info(self.vapi.ppcli("show error"))
1394             self.logger.info(self.vapi.ppcli("show ipsec all"))
1395         self.verify_counters6(p, p, count)
1396
1397
1398 class IpsecTun6Tests(IpsecTun6):
1399     """ UT test methods for Tunnel v6 """
1400
1401     def test_tun_basic66(self):
1402         """ ipsec 6o6 tunnel basic test """
1403         self.verify_tun_66(self.params[socket.AF_INET6], count=1)
1404
1405     def test_tun_reass_basic66(self):
1406         """ ipsec 6o6 tunnel basic reassembly test """
1407         self.verify_tun_reass_66(self.params[socket.AF_INET6])
1408
1409     def test_tun_burst66(self):
1410         """ ipsec 6o6 tunnel burst test """
1411         self.verify_tun_66(self.params[socket.AF_INET6], count=257)
1412
1413
1414 class IpsecTun6HandoffTests(IpsecTun6):
1415     """ UT test methods for Tunnel v6 with multiple workers """
1416     vpp_worker_count = 2
1417
1418     def test_tun_handoff_66(self):
1419         """ ipsec 6o6 tunnel worker hand-off test """
1420         self.vapi.cli("clear errors")
1421         self.vapi.cli("clear ipsec sa")
1422
1423         N_PKTS = 15
1424         p = self.params[socket.AF_INET6]
1425
1426         # inject alternately on worker 0 and 1. all counts on the SA
1427         # should be against worker 0
1428         for worker in [0, 1, 0, 1]:
1429             send_pkts = self.gen_encrypt_pkts6(p, p.scapy_tun_sa, self.tun_if,
1430                                                src=p.remote_tun_if_host,
1431                                                dst=self.pg1.remote_ip6,
1432                                                count=N_PKTS)
1433             recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1434                                              self.pg1, worker=worker)
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=N_PKTS)
1440             recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1441                                              self.tun_if, worker=worker)
1442             self.verify_encrypted6(p, p.vpp_tun_sa, recv_pkts)
1443
1444         # all counts against the first worker that was used
1445         self.verify_counters6(p, p, 4*N_PKTS, worker=0)
1446
1447
1448 class IpsecTun4HandoffTests(IpsecTun4):
1449     """ UT test methods for Tunnel v4 with multiple workers """
1450     vpp_worker_count = 2
1451
1452     def test_tun_handooff_44(self):
1453         """ ipsec 4o4 tunnel worker hand-off test """
1454         self.vapi.cli("clear errors")
1455         self.vapi.cli("clear ipsec sa")
1456
1457         N_PKTS = 15
1458         p = self.params[socket.AF_INET]
1459
1460         # inject alternately on worker 0 and 1. all counts on the SA
1461         # should be against worker 0
1462         for worker in [0, 1, 0, 1]:
1463             send_pkts = self.gen_encrypt_pkts(p, p.scapy_tun_sa, self.tun_if,
1464                                               src=p.remote_tun_if_host,
1465                                               dst=self.pg1.remote_ip4,
1466                                               count=N_PKTS)
1467             recv_pkts = self.send_and_expect(self.tun_if, send_pkts,
1468                                              self.pg1, worker=worker)
1469             self.verify_decrypted(p, recv_pkts)
1470
1471             send_pkts = self.gen_pkts(self.pg1, src=self.pg1.remote_ip4,
1472                                       dst=p.remote_tun_if_host,
1473                                       count=N_PKTS)
1474             recv_pkts = self.send_and_expect(self.pg1, send_pkts,
1475                                              self.tun_if, worker=worker)
1476             self.verify_encrypted(p, p.vpp_tun_sa, recv_pkts)
1477
1478         # all counts against the first worker that was used
1479         self.verify_counters4(p, 4*N_PKTS, worker=0)
1480
1481
1482 class IpsecTun46Tests(IpsecTun4Tests, IpsecTun6Tests):
1483     """ UT test methods for Tunnel v6 & v4 """
1484     pass
1485
1486
1487 if __name__ == '__main__':
1488     unittest.main(testRunner=VppTestRunner)