test/test_ipsec_tun_if_esp.py

   1 import unittest
   2 import socket
   3 import copy
   4
   5 from scapy.layers.ipsec import ESP
   6 from scapy.layers.l2 import Ether, Raw, GRE
   7 from scapy.layers.inet import IP, UDP
   8 from scapy.layers.inet6 import IPv6
   9 from framework import VppTestRunner, is_skip_aarch64_set, is_platform_aarch64
  10 from template_ipsec import TemplateIpsec, IpsecTun4Tests, IpsecTun6Tests, \
  11     IpsecTun4, IpsecTun6,  IpsecTcpTests,  config_tun_params
  12 from vpp_ipsec_tun_interface import VppIpsecTunInterface
  13 from vpp_gre_interface import VppGreInterface
  14 from vpp_ipip_tun_interface import VppIpIpTunInterface
  15 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
  16 from vpp_ipsec import VppIpsecSA, VppIpsecTunProtect
  17 from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
  18 from util import ppp
  19 from vpp_papi import VppEnum
  20
  21
  22 class TemplateIpsec4TunIfEsp(TemplateIpsec):
  23     """ IPsec tunnel interface tests """
  24
  25     encryption_type = ESP
  26
  27     @classmethod
  28     def setUpClass(cls):
  29         super(TemplateIpsec4TunIfEsp, cls).setUpClass()
  30
  31     @classmethod
  32     def tearDownClass(cls):
  33         super(TemplateIpsec4TunIfEsp, cls).tearDownClass()
  34
  35     def setUp(self):
  36         super(TemplateIpsec4TunIfEsp, self).setUp()
  37
  38         self.tun_if = self.pg0
  39
  40         p = self.ipv4_params
  41
  42         p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
  43                                         p.scapy_tun_spi, p.crypt_algo_vpp_id,
  44                                         p.crypt_key, p.crypt_key,
  45                                         p.auth_algo_vpp_id, p.auth_key,
  46                                         p.auth_key)
  47         p.tun_if.add_vpp_config()
  48         p.tun_if.admin_up()
  49         p.tun_if.config_ip4()
  50         p.tun_if.config_ip6()
  51
  52         r = VppIpRoute(self, p.remote_tun_if_host, 32,
  53                        [VppRoutePath(p.tun_if.remote_ip4,
  54                                      0xffffffff)])
  55         r.add_vpp_config()
  56         r = VppIpRoute(self, p.remote_tun_if_host6, 128,
  57                        [VppRoutePath(p.tun_if.remote_ip6,
  58                                      0xffffffff,
  59                                      proto=DpoProto.DPO_PROTO_IP6)])
  60         r.add_vpp_config()
  61
  62     def tearDown(self):
  63         super(TemplateIpsec4TunIfEsp, self).tearDown()
  64
  65
  66 class TestIpsec4TunIfEsp1(TemplateIpsec4TunIfEsp, IpsecTun4Tests):
  67     """ Ipsec ESP - TUN tests """
  68     tun4_encrypt_node_name = "esp4-encrypt-tun"
  69     tun4_decrypt_node_name = "esp4-decrypt"
  70
  71     def test_tun_basic64(self):
  72         """ ipsec 6o4 tunnel basic test """
  73         self.tun4_encrypt_node_name = "esp6-encrypt-tun"
  74
  75         self.verify_tun_64(self.params[socket.AF_INET], count=1)
  76
  77     def test_tun_burst64(self):
  78         """ ipsec 6o4 tunnel basic test """
  79         self.tun4_encrypt_node_name = "esp6-encrypt-tun"
  80
  81         self.verify_tun_64(self.params[socket.AF_INET], count=257)
  82
  83     def test_tun_basic_frag44(self):
  84         """ ipsec 4o4 tunnel frag basic test """
  85         self.tun4_encrypt_node_name = "esp4-encrypt-tun"
  86
  87         p = self.ipv4_params
  88
  89         self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
  90                                        [1500, 0, 0, 0])
  91         self.verify_tun_44(self.params[socket.AF_INET],
  92                            count=1, payload_size=1800, n_rx=2)
  93         self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
  94                                        [9000, 0, 0, 0])
  95
  96
  97 class TestIpsec4TunIfEsp2(TemplateIpsec4TunIfEsp, IpsecTcpTests):
  98     """ Ipsec ESP - TCP tests """
  99     pass
 100
 101
 102 class TemplateIpsec6TunIfEsp(TemplateIpsec):
 103     """ IPsec tunnel interface tests """
 104
 105     encryption_type = ESP
 106
 107     def setUp(self):
 108         super(TemplateIpsec6TunIfEsp, self).setUp()
 109
 110         self.tun_if = self.pg0
 111
 112         p = self.ipv6_params
 113         tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
 114                                       p.scapy_tun_spi, p.crypt_algo_vpp_id,
 115                                       p.crypt_key, p.crypt_key,
 116                                       p.auth_algo_vpp_id, p.auth_key,
 117                                       p.auth_key, is_ip6=True)
 118         tun_if.add_vpp_config()
 119         tun_if.admin_up()
 120         tun_if.config_ip6()
 121         tun_if.config_ip4()
 122
 123         r = VppIpRoute(self, p.remote_tun_if_host, 128,
 124                        [VppRoutePath(tun_if.remote_ip6,
 125                                      0xffffffff,
 126                                      proto=DpoProto.DPO_PROTO_IP6)])
 127         r.add_vpp_config()
 128         r = VppIpRoute(self, p.remote_tun_if_host4, 32,
 129                        [VppRoutePath(tun_if.remote_ip4,
 130                                      0xffffffff)])
 131         r.add_vpp_config()
 132
 133     def tearDown(self):
 134         super(TemplateIpsec6TunIfEsp, self).tearDown()
 135
 136
 137 class TestIpsec6TunIfEsp1(TemplateIpsec6TunIfEsp, IpsecTun6Tests):
 138     """ Ipsec ESP - TUN tests """
 139     tun6_encrypt_node_name = "esp6-encrypt-tun"
 140     tun6_decrypt_node_name = "esp6-decrypt"
 141
 142     def test_tun_basic46(self):
 143         """ ipsec 4o6 tunnel basic test """
 144         self.tun6_encrypt_node_name = "esp4-encrypt-tun"
 145         self.verify_tun_46(self.params[socket.AF_INET6], count=1)
 146
 147     def test_tun_burst46(self):
 148         """ ipsec 4o6 tunnel burst test """
 149         self.tun6_encrypt_node_name = "esp4-encrypt-tun"
 150         self.verify_tun_46(self.params[socket.AF_INET6], count=257)
 151
 152
 153 class TestIpsec4MultiTunIfEsp(TemplateIpsec, IpsecTun4):
 154     """ IPsec IPv4 Multi Tunnel interface """
 155
 156     encryption_type = ESP
 157     tun4_encrypt_node_name = "esp4-encrypt-tun"
 158     tun4_decrypt_node_name = "esp4-decrypt"
 159
 160     def setUp(self):
 161         super(TestIpsec4MultiTunIfEsp, self).setUp()
 162
 163         self.tun_if = self.pg0
 164
 165         self.multi_params = []
 166
 167         for ii in range(10):
 168             p = copy.copy(self.ipv4_params)
 169
 170             p.remote_tun_if_host = "1.1.1.%d" % (ii + 1)
 171             p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
 172             p.scapy_tun_spi = p.scapy_tun_spi + ii
 173             p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
 174             p.vpp_tun_spi = p.vpp_tun_spi + ii
 175
 176             p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
 177             p.scapy_tra_spi = p.scapy_tra_spi + ii
 178             p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
 179             p.vpp_tra_spi = p.vpp_tra_spi + ii
 180
 181             config_tun_params(p, self.encryption_type, self.tun_if)
 182             self.multi_params.append(p)
 183
 184             p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
 185                                             p.scapy_tun_spi,
 186                                             p.crypt_algo_vpp_id,
 187                                             p.crypt_key, p.crypt_key,
 188                                             p.auth_algo_vpp_id, p.auth_key,
 189                                             p.auth_key)
 190             p.tun_if.add_vpp_config()
 191             p.tun_if.admin_up()
 192             p.tun_if.config_ip4()
 193
 194             VppIpRoute(self, p.remote_tun_if_host, 32,
 195                        [VppRoutePath(p.tun_if.remote_ip4,
 196                                      0xffffffff)]).add_vpp_config()
 197
 198     def tearDown(self):
 199         super(TestIpsec4MultiTunIfEsp, self).tearDown()
 200
 201     def test_tun_44(self):
 202         """Multiple IPSEC tunnel interfaces """
 203         for p in self.multi_params:
 204             self.verify_tun_44(p, count=127)
 205             c = p.tun_if.get_rx_stats()
 206             self.assertEqual(c['packets'], 127)
 207             c = p.tun_if.get_tx_stats()
 208             self.assertEqual(c['packets'], 127)
 209
 210
 211 class TestIpsec4TunIfEspAll(TemplateIpsec, IpsecTun4):
 212     """ IPsec IPv4 Tunnel interface all Algos """
 213
 214     encryption_type = ESP
 215     tun4_encrypt_node_name = "esp4-encrypt-tun"
 216     tun4_decrypt_node_name = "esp4-decrypt"
 217
 218     def config_network(self, p):
 219         config_tun_params(p, self.encryption_type, self.tun_if)
 220
 221         p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
 222                                         p.scapy_tun_spi,
 223                                         p.crypt_algo_vpp_id,
 224                                         p.crypt_key, p.crypt_key,
 225                                         p.auth_algo_vpp_id, p.auth_key,
 226                                         p.auth_key,
 227                                         salt=p.salt)
 228         p.tun_if.add_vpp_config()
 229         p.tun_if.admin_up()
 230         p.tun_if.config_ip4()
 231         self.logger.info(self.vapi.cli("sh ipsec sa 0"))
 232         self.logger.info(self.vapi.cli("sh ipsec sa 1"))
 233
 234         p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
 235                              [VppRoutePath(p.tun_if.remote_ip4,
 236                                            0xffffffff)])
 237         p.route.add_vpp_config()
 238
 239     def unconfig_network(self, p):
 240         p.tun_if.unconfig_ip4()
 241         p.tun_if.remove_vpp_config()
 242         p.route.remove_vpp_config()
 243
 244     def setUp(self):
 245         super(TestIpsec4TunIfEspAll, self).setUp()
 246
 247         self.tun_if = self.pg0
 248
 249     def tearDown(self):
 250         super(TestIpsec4TunIfEspAll, self).tearDown()
 251
 252     def rekey(self, p):
 253         #
 254         # change the key and the SPI
 255         #
 256         p.crypt_key = 'X' + p.crypt_key[1:]
 257         p.scapy_tun_spi += 1
 258         p.scapy_tun_sa_id += 1
 259         p.vpp_tun_spi += 1
 260         p.vpp_tun_sa_id += 1
 261         p.tun_if.local_spi = p.vpp_tun_spi
 262         p.tun_if.remote_spi = p.scapy_tun_spi
 263
 264         config_tun_params(p, self.encryption_type, self.tun_if)
 265
 266         p.tun_sa_in = VppIpsecSA(self,
 267                                  p.scapy_tun_sa_id,
 268                                  p.scapy_tun_spi,
 269                                  p.auth_algo_vpp_id,
 270                                  p.auth_key,
 271                                  p.crypt_algo_vpp_id,
 272                                  p.crypt_key,
 273                                  self.vpp_esp_protocol,
 274                                  self.tun_if.local_addr[p.addr_type],
 275                                  self.tun_if.remote_addr[p.addr_type],
 276                                  flags=p.flags,
 277                                  salt=p.salt)
 278         p.tun_sa_out = VppIpsecSA(self,
 279                                   p.vpp_tun_sa_id,
 280                                   p.vpp_tun_spi,
 281                                   p.auth_algo_vpp_id,
 282                                   p.auth_key,
 283                                   p.crypt_algo_vpp_id,
 284                                   p.crypt_key,
 285                                   self.vpp_esp_protocol,
 286                                   self.tun_if.remote_addr[p.addr_type],
 287                                   self.tun_if.local_addr[p.addr_type],
 288                                   flags=p.flags,
 289                                   salt=p.salt)
 290         p.tun_sa_in.add_vpp_config()
 291         p.tun_sa_out.add_vpp_config()
 292
 293         self.vapi.ipsec_tunnel_if_set_sa(sw_if_index=p.tun_if.sw_if_index,
 294                                          sa_id=p.tun_sa_in.id,
 295                                          is_outbound=1)
 296         self.vapi.ipsec_tunnel_if_set_sa(sw_if_index=p.tun_if.sw_if_index,
 297                                          sa_id=p.tun_sa_out.id,
 298                                          is_outbound=0)
 299         self.logger.info(self.vapi.cli("sh ipsec sa"))
 300
 301     def test_tun_44(self):
 302         """IPSEC tunnel all algos """
 303
 304         # foreach VPP crypto engine
 305         engines = ["ia32", "ipsecmb", "openssl"]
 306
 307         # foreach crypto algorithm
 308         algos = [{'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 309                                  IPSEC_API_CRYPTO_ALG_AES_GCM_128),
 310                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 311                                 IPSEC_API_INTEG_ALG_NONE),
 312                   'scapy-crypto': "AES-GCM",
 313                   'scapy-integ': "NULL",
 314                   'key': "JPjyOWBeVEQiMe7h",
 315                   'salt': 3333},
 316                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 317                                  IPSEC_API_CRYPTO_ALG_AES_GCM_192),
 318                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 319                                 IPSEC_API_INTEG_ALG_NONE),
 320                   'scapy-crypto': "AES-GCM",
 321                   'scapy-integ': "NULL",
 322                   'key': "JPjyOWBeVEQiMe7hJPjyOWBe",
 323                   'salt': 0},
 324                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 325                                  IPSEC_API_CRYPTO_ALG_AES_GCM_256),
 326                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 327                                 IPSEC_API_INTEG_ALG_NONE),
 328                   'scapy-crypto': "AES-GCM",
 329                   'scapy-integ': "NULL",
 330                   'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
 331                   'salt': 9999},
 332                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 333                                  IPSEC_API_CRYPTO_ALG_AES_CBC_128),
 334                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 335                                 IPSEC_API_INTEG_ALG_SHA1_96),
 336                   'scapy-crypto': "AES-CBC",
 337                   'scapy-integ': "HMAC-SHA1-96",
 338                   'salt': 0,
 339                   'key': "JPjyOWBeVEQiMe7h"},
 340                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 341                                  IPSEC_API_CRYPTO_ALG_AES_CBC_192),
 342                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 343                                 IPSEC_API_INTEG_ALG_SHA1_96),
 344                   'scapy-crypto': "AES-CBC",
 345                   'scapy-integ': "HMAC-SHA1-96",
 346                   'salt': 0,
 347                   'key': "JPjyOWBeVEQiMe7hJPjyOWBe"},
 348                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 349                                  IPSEC_API_CRYPTO_ALG_AES_CBC_256),
 350                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 351                                 IPSEC_API_INTEG_ALG_SHA1_96),
 352                   'scapy-crypto': "AES-CBC",
 353                   'scapy-integ': "HMAC-SHA1-96",
 354                   'salt': 0,
 355                   'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}]
 356
 357         for engine in engines:
 358             self.vapi.cli("set crypto handler all %s" % engine)
 359
 360             #
 361             # loop through each of the algorithms
 362             #
 363             for algo in algos:
 364                 # with self.subTest(algo=algo['scapy']):
 365
 366                 p = copy.copy(self.ipv4_params)
 367                 p.auth_algo_vpp_id = algo['vpp-integ']
 368                 p.crypt_algo_vpp_id = algo['vpp-crypto']
 369                 p.crypt_algo = algo['scapy-crypto']
 370                 p.auth_algo = algo['scapy-integ']
 371                 p.crypt_key = algo['key']
 372                 p.salt = algo['salt']
 373
 374                 self.config_network(p)
 375
 376                 self.verify_tun_44(p, count=127)
 377                 c = p.tun_if.get_rx_stats()
 378                 self.assertEqual(c['packets'], 127)
 379                 c = p.tun_if.get_tx_stats()
 380                 self.assertEqual(c['packets'], 127)
 381
 382                 #
 383                 # rekey the tunnel
 384                 #
 385                 self.rekey(p)
 386                 self.verify_tun_44(p, count=127)
 387
 388                 self.unconfig_network(p)
 389                 p.tun_sa_out.remove_vpp_config()
 390                 p.tun_sa_in.remove_vpp_config()
 391
 392
 393 class TestIpsec6MultiTunIfEsp(TemplateIpsec, IpsecTun6):
 394     """ IPsec IPv6 Multi Tunnel interface """
 395
 396     encryption_type = ESP
 397     tun6_encrypt_node_name = "esp6-encrypt-tun"
 398     tun6_decrypt_node_name = "esp6-decrypt"
 399
 400     def setUp(self):
 401         super(TestIpsec6MultiTunIfEsp, self).setUp()
 402
 403         self.tun_if = self.pg0
 404
 405         self.multi_params = []
 406
 407         for ii in range(10):
 408             p = copy.copy(self.ipv6_params)
 409
 410             p.remote_tun_if_host = "1111::%d" % (ii + 1)
 411             p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
 412             p.scapy_tun_spi = p.scapy_tun_spi + ii
 413             p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
 414             p.vpp_tun_spi = p.vpp_tun_spi + ii
 415
 416             p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
 417             p.scapy_tra_spi = p.scapy_tra_spi + ii
 418             p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
 419             p.vpp_tra_spi = p.vpp_tra_spi + ii
 420
 421             config_tun_params(p, self.encryption_type, self.tun_if)
 422             self.multi_params.append(p)
 423
 424             p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
 425                                             p.scapy_tun_spi,
 426                                             p.crypt_algo_vpp_id,
 427                                             p.crypt_key, p.crypt_key,
 428                                             p.auth_algo_vpp_id, p.auth_key,
 429                                             p.auth_key, is_ip6=True)
 430             p.tun_if.add_vpp_config()
 431             p.tun_if.admin_up()
 432             p.tun_if.config_ip6()
 433
 434             r = VppIpRoute(self, p.remote_tun_if_host, 128,
 435                            [VppRoutePath(p.tun_if.remote_ip6,
 436                                          0xffffffff,
 437                                          proto=DpoProto.DPO_PROTO_IP6)])
 438             r.add_vpp_config()
 439
 440     def tearDown(self):
 441         super(TestIpsec6MultiTunIfEsp, self).tearDown()
 442
 443     def test_tun_66(self):
 444         """Multiple IPSEC tunnel interfaces """
 445         for p in self.multi_params:
 446             self.verify_tun_66(p, count=127)
 447             c = p.tun_if.get_rx_stats()
 448             self.assertEqual(c['packets'], 127)
 449             c = p.tun_if.get_tx_stats()
 450             self.assertEqual(c['packets'], 127)
 451
 452
 453 @unittest.skipIf(is_skip_aarch64_set and is_platform_aarch64,
 454                  "test doesn't work on aarch64")
 455 class TestIpsecGreTebIfEsp(TemplateIpsec,
 456                            IpsecTun4Tests):
 457     """ Ipsec GRE TEB ESP - TUN tests """
 458     tun4_encrypt_node_name = "esp4-encrypt-tun"
 459     tun4_decrypt_node_name = "esp4-decrypt-tun"
 460     encryption_type = ESP
 461     omac = "00:11:22:33:44:55"
 462
 463     def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
 464                          payload_size=100):
 465         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 466                 sa.encrypt(IP(src=self.pg0.remote_ip4,
 467                               dst=self.pg0.local_ip4) /
 468                            GRE() /
 469                            Ether(dst=self.omac) /
 470                            IP(src="1.1.1.1", dst="1.1.1.2") /
 471                            UDP(sport=1144, dport=2233) /
 472                            Raw('X' * payload_size))
 473                 for i in range(count)]
 474
 475     def gen_pkts(self, sw_intf, src, dst, count=1,
 476                  payload_size=100):
 477         return [Ether(dst=self.omac) /
 478                 IP(src="1.1.1.1", dst="1.1.1.2") /
 479                 UDP(sport=1144, dport=2233) /
 480                 Raw('X' * payload_size)
 481                 for i in range(count)]
 482
 483     def verify_decrypted(self, p, rxs):
 484         for rx in rxs:
 485             self.assert_equal(rx[Ether].dst, self.omac)
 486             self.assert_equal(rx[IP].dst, "1.1.1.2")
 487
 488     def verify_encrypted(self, p, sa, rxs):
 489         for rx in rxs:
 490             try:
 491                 pkt = sa.decrypt(rx[IP])
 492                 if not pkt.haslayer(IP):
 493                     pkt = IP(pkt[Raw].load)
 494                 self.assert_packet_checksums_valid(pkt)
 495                 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
 496                 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
 497                 self.assertTrue(pkt.haslayer(GRE))
 498                 e = pkt[Ether]
 499                 self.assertEqual(e[Ether].dst, self.omac)
 500                 self.assertEqual(e[IP].dst, "1.1.1.2")
 501             except (IndexError, AssertionError):
 502                 self.logger.debug(ppp("Unexpected packet:", rx))
 503                 try:
 504                     self.logger.debug(ppp("Decrypted packet:", pkt))
 505                 except:
 506                     pass
 507                 raise
 508
 509     def setUp(self):
 510         super(TestIpsecGreTebIfEsp, self).setUp()
 511
 512         self.tun_if = self.pg0
 513
 514         p = self.ipv4_params
 515
 516         bd1 = VppBridgeDomain(self, 1)
 517         bd1.add_vpp_config()
 518
 519         p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
 520                                   p.auth_algo_vpp_id, p.auth_key,
 521                                   p.crypt_algo_vpp_id, p.crypt_key,
 522                                   self.vpp_esp_protocol,
 523                                   self.pg0.local_ip4,
 524                                   self.pg0.remote_ip4)
 525         p.tun_sa_out.add_vpp_config()
 526
 527         p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
 528                                  p.auth_algo_vpp_id, p.auth_key,
 529                                  p.crypt_algo_vpp_id, p.crypt_key,
 530                                  self.vpp_esp_protocol,
 531                                  self.pg0.remote_ip4,
 532                                  self.pg0.local_ip4)
 533         p.tun_sa_in.add_vpp_config()
 534
 535         self.tun = VppGreInterface(self,
 536                                    self.pg0.local_ip4,
 537                                    self.pg0.remote_ip4,
 538                                    type=(VppEnum.vl_api_gre_tunnel_type_t.
 539                                          GRE_API_TUNNEL_TYPE_TEB))
 540         self.tun.add_vpp_config()
 541
 542         p.tun_protect = VppIpsecTunProtect(self,
 543                                            self.tun,
 544                                            p.tun_sa_out,
 545                                            [p.tun_sa_in])
 546
 547         p.tun_protect.add_vpp_config()
 548
 549         self.tun.admin_up()
 550         self.tun.config_ip4()
 551
 552         VppBridgeDomainPort(self, bd1, self.tun).add_vpp_config()
 553         VppBridgeDomainPort(self, bd1, self.pg1).add_vpp_config()
 554
 555         self.vapi.cli("clear ipsec sa")
 556
 557     def tearDown(self):
 558         self.tun.unconfig_ip4()
 559         super(TestIpsecGreTebIfEsp, self).tearDown()
 560
 561
 562 class TestIpsecGreIfEsp(TemplateIpsec,
 563                         IpsecTun4Tests):
 564     """ Ipsec GRE ESP - TUN tests """
 565     tun4_encrypt_node_name = "esp4-encrypt-tun"
 566     tun4_decrypt_node_name = "esp4-decrypt-tun"
 567     encryption_type = ESP
 568
 569     def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
 570                          payload_size=100):
 571         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 572                 sa.encrypt(IP(src=self.pg0.remote_ip4,
 573                               dst=self.pg0.local_ip4) /
 574                            GRE() /
 575                            IP(src=self.pg1.local_ip4,
 576                               dst=self.pg1.remote_ip4) /
 577                            UDP(sport=1144, dport=2233) /
 578                            Raw('X' * payload_size))
 579                 for i in range(count)]
 580
 581     def gen_pkts(self, sw_intf, src, dst, count=1,
 582                  payload_size=100):
 583         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 584                 IP(src="1.1.1.1", dst="1.1.1.2") /
 585                 UDP(sport=1144, dport=2233) /
 586                 Raw('X' * payload_size)
 587                 for i in range(count)]
 588
 589     def verify_decrypted(self, p, rxs):
 590         for rx in rxs:
 591             self.assert_equal(rx[Ether].dst, self.pg1.remote_mac)
 592             self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
 593
 594     def verify_encrypted(self, p, sa, rxs):
 595         for rx in rxs:
 596             try:
 597                 pkt = sa.decrypt(rx[IP])
 598                 if not pkt.haslayer(IP):
 599                     pkt = IP(pkt[Raw].load)
 600                 self.assert_packet_checksums_valid(pkt)
 601                 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
 602                 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
 603                 self.assertTrue(pkt.haslayer(GRE))
 604                 e = pkt[GRE]
 605                 self.assertEqual(e[IP].dst, "1.1.1.2")
 606             except (IndexError, AssertionError):
 607                 self.logger.debug(ppp("Unexpected packet:", rx))
 608                 try:
 609                     self.logger.debug(ppp("Decrypted packet:", pkt))
 610                 except:
 611                     pass
 612                 raise
 613
 614     def setUp(self):
 615         super(TestIpsecGreIfEsp, self).setUp()
 616
 617         self.tun_if = self.pg0
 618
 619         p = self.ipv4_params
 620
 621         bd1 = VppBridgeDomain(self, 1)
 622         bd1.add_vpp_config()
 623
 624         p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
 625                                   p.auth_algo_vpp_id, p.auth_key,
 626                                   p.crypt_algo_vpp_id, p.crypt_key,
 627                                   self.vpp_esp_protocol,
 628                                   self.pg0.local_ip4,
 629                                   self.pg0.remote_ip4)
 630         p.tun_sa_out.add_vpp_config()
 631
 632         p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
 633                                  p.auth_algo_vpp_id, p.auth_key,
 634                                  p.crypt_algo_vpp_id, p.crypt_key,
 635                                  self.vpp_esp_protocol,
 636                                  self.pg0.remote_ip4,
 637                                  self.pg0.local_ip4)
 638         p.tun_sa_in.add_vpp_config()
 639
 640         self.tun = VppGreInterface(self,
 641                                    self.pg0.local_ip4,
 642                                    self.pg0.remote_ip4)
 643         self.tun.add_vpp_config()
 644
 645         p.tun_protect = VppIpsecTunProtect(self,
 646                                            self.tun,
 647                                            p.tun_sa_out,
 648                                            [p.tun_sa_in])
 649         p.tun_protect.add_vpp_config()
 650
 651         self.tun.admin_up()
 652         self.tun.config_ip4()
 653
 654         VppIpRoute(self, "1.1.1.2", 32,
 655                    [VppRoutePath(self.tun.remote_ip4,
 656                                  0xffffffff)]).add_vpp_config()
 657
 658     def tearDown(self):
 659         self.tun.unconfig_ip4()
 660         super(TestIpsecGreIfEsp, self).tearDown()
 661
 662
 663 class TemplateIpsec4TunProtect(object):
 664     """ IPsec IPv4 Tunnel protect """
 665
 666     def config_sa_tra(self, p):
 667         config_tun_params(p, self.encryption_type, self.tun_if)
 668
 669         p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
 670                                   p.auth_algo_vpp_id, p.auth_key,
 671                                   p.crypt_algo_vpp_id, p.crypt_key,
 672                                   self.vpp_esp_protocol)
 673         p.tun_sa_out.add_vpp_config()
 674
 675         p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
 676                                  p.auth_algo_vpp_id, p.auth_key,
 677                                  p.crypt_algo_vpp_id, p.crypt_key,
 678                                  self.vpp_esp_protocol)
 679         p.tun_sa_in.add_vpp_config()
 680
 681     def config_sa_tun(self, p):
 682         config_tun_params(p, self.encryption_type, self.tun_if)
 683
 684         p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
 685                                   p.auth_algo_vpp_id, p.auth_key,
 686                                   p.crypt_algo_vpp_id, p.crypt_key,
 687                                   self.vpp_esp_protocol,
 688                                   self.tun_if.remote_addr[p.addr_type],
 689                                   self.tun_if.local_addr[p.addr_type])
 690         p.tun_sa_out.add_vpp_config()
 691
 692         p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
 693                                  p.auth_algo_vpp_id, p.auth_key,
 694                                  p.crypt_algo_vpp_id, p.crypt_key,
 695                                  self.vpp_esp_protocol,
 696                                  self.tun_if.remote_addr[p.addr_type],
 697                                  self.tun_if.local_addr[p.addr_type])
 698         p.tun_sa_in.add_vpp_config()
 699
 700     def config_protect(self, p):
 701         p.tun_protect = VppIpsecTunProtect(self,
 702                                            p.tun_if,
 703                                            p.tun_sa_out,
 704                                            [p.tun_sa_in])
 705         p.tun_protect.add_vpp_config()
 706
 707     def config_network(self, p):
 708         p.tun_if = VppIpIpTunInterface(self, self.pg0,
 709                                        self.pg0.local_ip4,
 710                                        self.pg0.remote_ip4)
 711         p.tun_if.add_vpp_config()
 712         p.tun_if.admin_up()
 713         p.tun_if.config_ip4()
 714
 715         p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
 716                              [VppRoutePath(p.tun_if.remote_ip4,
 717                                            0xffffffff)])
 718         p.route.add_vpp_config()
 719
 720     def unconfig_network(self, p):
 721         p.route.remove_vpp_config()
 722         p.tun_if.remove_vpp_config()
 723
 724     def unconfig_protect(self, p):
 725         p.tun_protect.remove_vpp_config()
 726
 727     def unconfig_sa(self, p):
 728         p.tun_sa_out.remove_vpp_config()
 729         p.tun_sa_in.remove_vpp_config()
 730
 731
 732 class TestIpsec4TunProtect(TemplateIpsec,
 733                            TemplateIpsec4TunProtect,
 734                            IpsecTun4):
 735     """ IPsec IPv4 Tunnel protect - transport mode"""
 736
 737     encryption_type = ESP
 738     tun4_encrypt_node_name = "esp4-encrypt-tun"
 739     tun4_decrypt_node_name = "esp4-decrypt-tun"
 740
 741     def setUp(self):
 742         super(TestIpsec4TunProtect, self).setUp()
 743
 744         self.tun_if = self.pg0
 745
 746     def tearDown(self):
 747         super(TestIpsec4TunProtect, self).tearDown()
 748
 749     def test_tun_44(self):
 750         """IPSEC tunnel protect"""
 751
 752         p = self.ipv4_params
 753
 754         self.config_network(p)
 755         self.config_sa_tra(p)
 756         self.config_protect(p)
 757
 758         self.verify_tun_44(p, count=127)
 759         c = p.tun_if.get_rx_stats()
 760         self.assertEqual(c['packets'], 127)
 761         c = p.tun_if.get_tx_stats()
 762         self.assertEqual(c['packets'], 127)
 763
 764         # rekey - create new SAs and update the tunnel protection
 765         np = copy.copy(p)
 766         np.crypt_key = 'X' + p.crypt_key[1:]
 767         np.scapy_tun_spi += 100
 768         np.scapy_tun_sa_id += 1
 769         np.vpp_tun_spi += 100
 770         np.vpp_tun_sa_id += 1
 771         np.tun_if.local_spi = p.vpp_tun_spi
 772         np.tun_if.remote_spi = p.scapy_tun_spi
 773
 774         self.config_sa_tra(np)
 775         self.config_protect(np)
 776         self.unconfig_sa(p)
 777
 778         self.verify_tun_44(np, count=127)
 779         c = p.tun_if.get_rx_stats()
 780         self.assertEqual(c['packets'], 254)
 781         c = p.tun_if.get_tx_stats()
 782         self.assertEqual(c['packets'], 254)
 783
 784         # teardown
 785         self.unconfig_protect(np)
 786         self.unconfig_sa(np)
 787         self.unconfig_network(p)
 788
 789
 790 class TestIpsec4TunProtectTun(TemplateIpsec,
 791                               TemplateIpsec4TunProtect,
 792                               IpsecTun4):
 793     """ IPsec IPv4 Tunnel protect - tunnel mode"""
 794
 795     encryption_type = ESP
 796     tun4_encrypt_node_name = "esp4-encrypt-tun"
 797     tun4_decrypt_node_name = "esp4-decrypt-tun"
 798
 799     def setUp(self):
 800         super(TestIpsec4TunProtectTun, self).setUp()
 801
 802         self.tun_if = self.pg0
 803
 804     def tearDown(self):
 805         super(TestIpsec4TunProtectTun, self).tearDown()
 806
 807     def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
 808                          payload_size=100):
 809         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 810                 sa.encrypt(IP(src=sw_intf.remote_ip4,
 811                               dst=sw_intf.local_ip4) /
 812                            IP(src=src, dst=dst) /
 813                            UDP(sport=1144, dport=2233) /
 814                            Raw('X' * payload_size))
 815                 for i in range(count)]
 816
 817     def gen_pkts(self, sw_intf, src, dst, count=1,
 818                  payload_size=100):
 819         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 820                 IP(src=src, dst=dst) /
 821                 UDP(sport=1144, dport=2233) /
 822                 Raw('X' * payload_size)
 823                 for i in range(count)]
 824
 825     def verify_decrypted(self, p, rxs):
 826         for rx in rxs:
 827             self.assert_equal(rx[IP].dst, self.pg1.remote_ip4)
 828             self.assert_equal(rx[IP].src, p.remote_tun_if_host)
 829             self.assert_packet_checksums_valid(rx)
 830
 831     def verify_encrypted(self, p, sa, rxs):
 832         for rx in rxs:
 833             try:
 834                 pkt = sa.decrypt(rx[IP])
 835                 if not pkt.haslayer(IP):
 836                     pkt = IP(pkt[Raw].load)
 837                 self.assert_packet_checksums_valid(pkt)
 838                 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
 839                 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
 840                 inner = pkt[IP].payload
 841                 self.assertEqual(inner[IP][IP].dst, p.remote_tun_if_host)
 842
 843             except (IndexError, AssertionError):
 844                 self.logger.debug(ppp("Unexpected packet:", rx))
 845                 try:
 846                     self.logger.debug(ppp("Decrypted packet:", pkt))
 847                 except:
 848                     pass
 849                 raise
 850
 851     def test_tun_44(self):
 852         """IPSEC tunnel protect """
 853
 854         p = self.ipv4_params
 855
 856         self.config_network(p)
 857         self.config_sa_tun(p)
 858         self.config_protect(p)
 859
 860         self.verify_tun_44(p, count=127)
 861
 862         c = p.tun_if.get_rx_stats()
 863         self.assertEqual(c['packets'], 127)
 864         c = p.tun_if.get_tx_stats()
 865         self.assertEqual(c['packets'], 127)
 866
 867         # rekey - create new SAs and update the tunnel protection
 868         np = copy.copy(p)
 869         np.crypt_key = 'X' + p.crypt_key[1:]
 870         np.scapy_tun_spi += 100
 871         np.scapy_tun_sa_id += 1
 872         np.vpp_tun_spi += 100
 873         np.vpp_tun_sa_id += 1
 874         np.tun_if.local_spi = p.vpp_tun_spi
 875         np.tun_if.remote_spi = p.scapy_tun_spi
 876
 877         self.config_sa_tun(np)
 878         self.config_protect(np)
 879         self.unconfig_sa(p)
 880
 881         self.verify_tun_44(np, count=127)
 882         c = p.tun_if.get_rx_stats()
 883         self.assertEqual(c['packets'], 254)
 884         c = p.tun_if.get_tx_stats()
 885         self.assertEqual(c['packets'], 254)
 886
 887         # teardown
 888         self.unconfig_protect(np)
 889         self.unconfig_sa(np)
 890         self.unconfig_network(p)
 891
 892
 893 class TemplateIpsec6TunProtect(object):
 894     """ IPsec IPv6 Tunnel protect """
 895
 896     def config_sa_tra(self, p):
 897         config_tun_params(p, self.encryption_type, self.tun_if)
 898
 899         p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
 900                                   p.auth_algo_vpp_id, p.auth_key,
 901                                   p.crypt_algo_vpp_id, p.crypt_key,
 902                                   self.vpp_esp_protocol)
 903         p.tun_sa_out.add_vpp_config()
 904
 905         p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
 906                                  p.auth_algo_vpp_id, p.auth_key,
 907                                  p.crypt_algo_vpp_id, p.crypt_key,
 908                                  self.vpp_esp_protocol)
 909         p.tun_sa_in.add_vpp_config()
 910
 911     def config_sa_tun(self, p):
 912         config_tun_params(p, self.encryption_type, self.tun_if)
 913
 914         p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
 915                                   p.auth_algo_vpp_id, p.auth_key,
 916                                   p.crypt_algo_vpp_id, p.crypt_key,
 917                                   self.vpp_esp_protocol,
 918                                   self.tun_if.remote_addr[p.addr_type],
 919                                   self.tun_if.local_addr[p.addr_type])
 920         p.tun_sa_out.add_vpp_config()
 921
 922         p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
 923                                  p.auth_algo_vpp_id, p.auth_key,
 924                                  p.crypt_algo_vpp_id, p.crypt_key,
 925                                  self.vpp_esp_protocol,
 926                                  self.tun_if.remote_addr[p.addr_type],
 927                                  self.tun_if.local_addr[p.addr_type])
 928         p.tun_sa_in.add_vpp_config()
 929
 930     def config_protect(self, p):
 931         p.tun_protect = VppIpsecTunProtect(self,
 932                                            p.tun_if,
 933                                            p.tun_sa_out,
 934                                            [p.tun_sa_in])
 935         p.tun_protect.add_vpp_config()
 936
 937     def config_network(self, p):
 938         p.tun_if = VppIpIpTunInterface(self, self.pg0,
 939                                        self.pg0.local_ip6,
 940                                        self.pg0.remote_ip6)
 941         p.tun_if.add_vpp_config()
 942         p.tun_if.admin_up()
 943         p.tun_if.config_ip6()
 944
 945         p.route = VppIpRoute(self, p.remote_tun_if_host, 128,
 946                              [VppRoutePath(p.tun_if.remote_ip6,
 947                                            0xffffffff,
 948                                            proto=DpoProto.DPO_PROTO_IP6)])
 949         p.route.add_vpp_config()
 950
 951     def unconfig_network(self, p):
 952         p.route.remove_vpp_config()
 953         p.tun_if.remove_vpp_config()
 954
 955     def unconfig_protect(self, p):
 956         p.tun_protect.remove_vpp_config()
 957
 958     def unconfig_sa(self, p):
 959         p.tun_sa_out.remove_vpp_config()
 960         p.tun_sa_in.remove_vpp_config()
 961
 962
 963 class TestIpsec6TunProtect(TemplateIpsec,
 964                            TemplateIpsec6TunProtect,
 965                            IpsecTun6):
 966     """ IPsec IPv6 Tunnel protect - transport mode"""
 967
 968     encryption_type = ESP
 969     tun6_encrypt_node_name = "esp6-encrypt-tun"
 970     tun6_decrypt_node_name = "esp6-decrypt-tun"
 971
 972     def setUp(self):
 973         super(TestIpsec6TunProtect, self).setUp()
 974
 975         self.tun_if = self.pg0
 976
 977     def tearDown(self):
 978         super(TestIpsec6TunProtect, self).tearDown()
 979
 980     def test_tun_66(self):
 981         """IPSEC tunnel protect"""
 982
 983         p = self.ipv6_params
 984
 985         self.config_network(p)
 986         self.config_sa_tra(p)
 987         self.config_protect(p)
 988
 989         self.verify_tun_66(p, count=127)
 990         c = p.tun_if.get_rx_stats()
 991         self.assertEqual(c['packets'], 127)
 992         c = p.tun_if.get_tx_stats()
 993         self.assertEqual(c['packets'], 127)
 994
 995         # rekey - create new SAs and update the tunnel protection
 996         np = copy.copy(p)
 997         np.crypt_key = 'X' + p.crypt_key[1:]
 998         np.scapy_tun_spi += 100
 999         np.scapy_tun_sa_id += 1
1000         np.vpp_tun_spi += 100
1001         np.vpp_tun_sa_id += 1
1002         np.tun_if.local_spi = p.vpp_tun_spi
1003         np.tun_if.remote_spi = p.scapy_tun_spi
1004
1005         self.config_sa_tra(np)
1006         self.config_protect(np)
1007         self.unconfig_sa(p)
1008
1009         self.verify_tun_66(np, count=127)
1010         c = p.tun_if.get_rx_stats()
1011         self.assertEqual(c['packets'], 254)
1012         c = p.tun_if.get_tx_stats()
1013         self.assertEqual(c['packets'], 254)
1014
1015         # 3 phase rekey
1016         #  1) add two input SAs [old, new]
1017         #  2) swap output SA to [new]
1018         #  3) use only [new] input SA
1019         np3 = copy.copy(np)
1020         np3.crypt_key = 'Z' + p.crypt_key[1:]
1021         np3.scapy_tun_spi += 100
1022         np3.scapy_tun_sa_id += 1
1023         np3.vpp_tun_spi += 100
1024         np3.vpp_tun_sa_id += 1
1025         np3.tun_if.local_spi = p.vpp_tun_spi
1026         np3.tun_if.remote_spi = p.scapy_tun_spi
1027
1028         self.config_sa_tra(np3)
1029
1030         # step 1;
1031         p.tun_protect.update_vpp_config(np.tun_sa_out,
1032                                         [np.tun_sa_in, np3.tun_sa_in])
1033         self.verify_tun_66(np, np, count=127)
1034         self.verify_tun_66(np3, np, count=127)
1035
1036         # step 2;
1037         p.tun_protect.update_vpp_config(np3.tun_sa_out,
1038                                         [np.tun_sa_in, np3.tun_sa_in])
1039         self.verify_tun_66(np, np3, count=127)
1040         self.verify_tun_66(np3, np3, count=127)
1041
1042         # step 1;
1043         p.tun_protect.update_vpp_config(np3.tun_sa_out,
1044                                         [np3.tun_sa_in])
1045         self.verify_tun_66(np3, np3, count=127)
1046         self.verify_drop_tun_66(np, count=127)
1047
1048         c = p.tun_if.get_rx_stats()
1049         self.assertEqual(c['packets'], 127*7)
1050         c = p.tun_if.get_tx_stats()
1051         self.assertEqual(c['packets'], 127*7)
1052         self.unconfig_sa(np)
1053
1054         # teardown
1055         self.unconfig_protect(np3)
1056         self.unconfig_sa(np3)
1057         self.unconfig_network(p)
1058
1059
1060 class TestIpsec6TunProtectTun(TemplateIpsec,
1061                               TemplateIpsec6TunProtect,
1062                               IpsecTun6):
1063     """ IPsec IPv6 Tunnel protect - tunnel mode"""
1064
1065     encryption_type = ESP
1066     tun6_encrypt_node_name = "esp6-encrypt-tun"
1067     tun6_decrypt_node_name = "esp6-decrypt-tun"
1068
1069     def setUp(self):
1070         super(TestIpsec6TunProtectTun, self).setUp()
1071
1072         self.tun_if = self.pg0
1073
1074     def tearDown(self):
1075         super(TestIpsec6TunProtectTun, self).tearDown()
1076
1077     def gen_encrypt_pkts6(self, sa, sw_intf, src, dst, count=1,
1078                           payload_size=100):
1079         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
1080                 sa.encrypt(IPv6(src=sw_intf.remote_ip6,
1081                                 dst=sw_intf.local_ip6) /
1082                            IPv6(src=src, dst=dst) /
1083                            UDP(sport=1166, dport=2233) /
1084                            Raw('X' * payload_size))
1085                 for i in range(count)]
1086
1087     def gen_pkts6(self, sw_intf, src, dst, count=1,
1088                   payload_size=100):
1089         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
1090                 IPv6(src=src, dst=dst) /
1091                 UDP(sport=1166, dport=2233) /
1092                 Raw('X' * payload_size)
1093                 for i in range(count)]
1094
1095     def verify_decrypted6(self, p, rxs):
1096         for rx in rxs:
1097             self.assert_equal(rx[IPv6].dst, self.pg1.remote_ip6)
1098             self.assert_equal(rx[IPv6].src, p.remote_tun_if_host)
1099             self.assert_packet_checksums_valid(rx)
1100
1101     def verify_encrypted6(self, p, sa, rxs):
1102         for rx in rxs:
1103             try:
1104                 pkt = sa.decrypt(rx[IPv6])
1105                 if not pkt.haslayer(IPv6):
1106                     pkt = IPv6(pkt[Raw].load)
1107                 self.assert_packet_checksums_valid(pkt)
1108                 self.assert_equal(pkt[IPv6].dst, self.pg0.remote_ip6)
1109                 self.assert_equal(pkt[IPv6].src, self.pg0.local_ip6)
1110                 inner = pkt[IPv6].payload
1111                 self.assertEqual(inner[IPv6][IPv6].dst, p.remote_tun_if_host)
1112
1113             except (IndexError, AssertionError):
1114                 self.logger.debug(ppp("Unexpected packet:", rx))
1115                 try:
1116                     self.logger.debug(ppp("Decrypted packet:", pkt))
1117                 except:
1118                     pass
1119                 raise
1120
1121     def test_tun_66(self):
1122         """IPSEC tunnel protect """
1123
1124         p = self.ipv6_params
1125
1126         self.config_network(p)
1127         self.config_sa_tun(p)
1128         self.config_protect(p)
1129
1130         self.verify_tun_66(p, count=127)
1131
1132         c = p.tun_if.get_rx_stats()
1133         self.assertEqual(c['packets'], 127)
1134         c = p.tun_if.get_tx_stats()
1135         self.assertEqual(c['packets'], 127)
1136
1137         # rekey - create new SAs and update the tunnel protection
1138         np = copy.copy(p)
1139         np.crypt_key = 'X' + p.crypt_key[1:]
1140         np.scapy_tun_spi += 100
1141         np.scapy_tun_sa_id += 1
1142         np.vpp_tun_spi += 100
1143         np.vpp_tun_sa_id += 1
1144         np.tun_if.local_spi = p.vpp_tun_spi
1145         np.tun_if.remote_spi = p.scapy_tun_spi
1146
1147         self.config_sa_tun(np)
1148         self.config_protect(np)
1149         self.unconfig_sa(p)
1150
1151         self.verify_tun_66(np, count=127)
1152         c = p.tun_if.get_rx_stats()
1153         self.assertEqual(c['packets'], 254)
1154         c = p.tun_if.get_tx_stats()
1155         self.assertEqual(c['packets'], 254)
1156
1157         # teardown
1158         self.unconfig_protect(np)
1159         self.unconfig_sa(np)
1160         self.unconfig_network(p)
1161
1162
1163 if __name__ == '__main__':
1164     unittest.main(testRunner=VppTestRunner)