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 framework import VppTestRunner, is_skip_aarch64_set, is_platform_aarch64
   9 from template_ipsec import TemplateIpsec, IpsecTun4Tests, IpsecTun6Tests, \
  10     IpsecTun4, IpsecTun6,  IpsecTcpTests,  config_tun_params
  11 from vpp_ipsec_tun_interface import VppIpsecTunInterface, \
  12     VppIpsecGRETunInterface
  13 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
  14 from vpp_ipsec import VppIpsecSA
  15 from vpp_l2 import VppBridgeDomain, VppBridgeDomainPort
  16 from util import ppp
  17 from vpp_papi import VppEnum
  18
  19
  20 class TemplateIpsec4TunIfEsp(TemplateIpsec):
  21     """ IPsec tunnel interface tests """
  22
  23     encryption_type = ESP
  24
  25     @classmethod
  26     def setUpClass(cls):
  27         super(TemplateIpsec4TunIfEsp, cls).setUpClass()
  28
  29     @classmethod
  30     def tearDownClass(cls):
  31         super(TemplateIpsec4TunIfEsp, cls).tearDownClass()
  32
  33     def setUp(self):
  34         super(TemplateIpsec4TunIfEsp, self).setUp()
  35
  36         self.tun_if = self.pg0
  37
  38         p = self.ipv4_params
  39
  40         p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
  41                                         p.scapy_tun_spi, p.crypt_algo_vpp_id,
  42                                         p.crypt_key, p.crypt_key,
  43                                         p.auth_algo_vpp_id, p.auth_key,
  44                                         p.auth_key)
  45         p.tun_if.add_vpp_config()
  46         p.tun_if.admin_up()
  47         p.tun_if.config_ip4()
  48         p.tun_if.config_ip6()
  49
  50         VppIpRoute(self, p.remote_tun_if_host, 32,
  51                    [VppRoutePath(p.tun_if.remote_ip4,
  52                                  0xffffffff)]).add_vpp_config()
  53         VppIpRoute(self, p.remote_tun_if_host6, 128,
  54                    [VppRoutePath(p.tun_if.remote_ip6,
  55                                  0xffffffff,
  56                                  proto=DpoProto.DPO_PROTO_IP6)],
  57                    is_ip6=1).add_vpp_config()
  58
  59     def tearDown(self):
  60         if not self.vpp_dead:
  61             self.vapi.cli("show hardware")
  62         super(TemplateIpsec4TunIfEsp, self).tearDown()
  63
  64
  65 class TestIpsec4TunIfEsp1(TemplateIpsec4TunIfEsp, IpsecTun4Tests):
  66     """ Ipsec ESP - TUN tests """
  67     tun4_encrypt_node_name = "esp4-encrypt"
  68     tun4_decrypt_node_name = "esp4-decrypt"
  69
  70     def test_tun_basic64(self):
  71         """ ipsec 6o4 tunnel basic test """
  72         self.verify_tun_64(self.params[socket.AF_INET], count=1)
  73
  74     def test_tun_burst64(self):
  75         """ ipsec 6o4 tunnel basic test """
  76         self.verify_tun_64(self.params[socket.AF_INET], count=257)
  77
  78     def test_tun_basic_frag44(self):
  79         """ ipsec 4o4 tunnel frag basic test """
  80         p = self.ipv4_params
  81
  82         self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
  83                                        [1500, 0, 0, 0])
  84         self.verify_tun_44(self.params[socket.AF_INET],
  85                            count=1, payload_size=1800, n_rx=2)
  86         self.vapi.sw_interface_set_mtu(p.tun_if.sw_if_index,
  87                                        [9000, 0, 0, 0])
  88
  89
  90 class TestIpsec4TunIfEsp2(TemplateIpsec4TunIfEsp, IpsecTcpTests):
  91     """ Ipsec ESP - TCP tests """
  92     pass
  93
  94
  95 class TemplateIpsec6TunIfEsp(TemplateIpsec):
  96     """ IPsec tunnel interface tests """
  97
  98     encryption_type = ESP
  99
 100     def setUp(self):
 101         super(TemplateIpsec6TunIfEsp, self).setUp()
 102
 103         self.tun_if = self.pg0
 104
 105         p = self.ipv6_params
 106         tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
 107                                       p.scapy_tun_spi, p.crypt_algo_vpp_id,
 108                                       p.crypt_key, p.crypt_key,
 109                                       p.auth_algo_vpp_id, p.auth_key,
 110                                       p.auth_key, is_ip6=True)
 111         tun_if.add_vpp_config()
 112         tun_if.admin_up()
 113         tun_if.config_ip6()
 114         tun_if.config_ip4()
 115
 116         VppIpRoute(self, p.remote_tun_if_host, 128,
 117                    [VppRoutePath(tun_if.remote_ip6,
 118                                  0xffffffff,
 119                                  proto=DpoProto.DPO_PROTO_IP6)],
 120                    is_ip6=1).add_vpp_config()
 121         VppIpRoute(self, p.remote_tun_if_host4, 32,
 122                    [VppRoutePath(tun_if.remote_ip4,
 123                                  0xffffffff)]).add_vpp_config()
 124
 125     def tearDown(self):
 126         if not self.vpp_dead:
 127             self.vapi.cli("show hardware")
 128         super(TemplateIpsec6TunIfEsp, self).tearDown()
 129
 130
 131 class TestIpsec6TunIfEsp1(TemplateIpsec6TunIfEsp, IpsecTun6Tests):
 132     """ Ipsec ESP - TUN tests """
 133     tun6_encrypt_node_name = "esp6-encrypt"
 134     tun6_decrypt_node_name = "esp6-decrypt"
 135
 136     def test_tun_basic46(self):
 137         """ ipsec 4o6 tunnel basic test """
 138         self.verify_tun_46(self.params[socket.AF_INET6], count=1)
 139
 140     def test_tun_burst46(self):
 141         """ ipsec 4o6 tunnel burst test """
 142         self.verify_tun_46(self.params[socket.AF_INET6], count=257)
 143
 144
 145 class TestIpsec4MultiTunIfEsp(TemplateIpsec, IpsecTun4):
 146     """ IPsec IPv4 Multi Tunnel interface """
 147
 148     encryption_type = ESP
 149     tun4_encrypt_node_name = "esp4-encrypt"
 150     tun4_decrypt_node_name = "esp4-decrypt"
 151
 152     def setUp(self):
 153         super(TestIpsec4MultiTunIfEsp, self).setUp()
 154
 155         self.tun_if = self.pg0
 156
 157         self.multi_params = []
 158
 159         for ii in range(10):
 160             p = copy.copy(self.ipv4_params)
 161
 162             p.remote_tun_if_host = "1.1.1.%d" % (ii + 1)
 163             p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
 164             p.scapy_tun_spi = p.scapy_tun_spi + ii
 165             p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
 166             p.vpp_tun_spi = p.vpp_tun_spi + ii
 167
 168             p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
 169             p.scapy_tra_spi = p.scapy_tra_spi + ii
 170             p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
 171             p.vpp_tra_spi = p.vpp_tra_spi + ii
 172
 173             config_tun_params(p, self.encryption_type, self.tun_if)
 174             self.multi_params.append(p)
 175
 176             p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
 177                                             p.scapy_tun_spi,
 178                                             p.crypt_algo_vpp_id,
 179                                             p.crypt_key, p.crypt_key,
 180                                             p.auth_algo_vpp_id, p.auth_key,
 181                                             p.auth_key)
 182             p.tun_if.add_vpp_config()
 183             p.tun_if.admin_up()
 184             p.tun_if.config_ip4()
 185
 186             VppIpRoute(self, p.remote_tun_if_host, 32,
 187                        [VppRoutePath(p.tun_if.remote_ip4,
 188                                      0xffffffff)]).add_vpp_config()
 189
 190     def tearDown(self):
 191         if not self.vpp_dead:
 192             self.vapi.cli("show hardware")
 193         super(TestIpsec4MultiTunIfEsp, self).tearDown()
 194
 195     def test_tun_44(self):
 196         """Multiple IPSEC tunnel interfaces """
 197         for p in self.multi_params:
 198             self.verify_tun_44(p, count=127)
 199             c = p.tun_if.get_rx_stats()
 200             self.assertEqual(c['packets'], 127)
 201             c = p.tun_if.get_tx_stats()
 202             self.assertEqual(c['packets'], 127)
 203
 204
 205 class TestIpsec4TunIfEspAll(TemplateIpsec, IpsecTun4):
 206     """ IPsec IPv4 Tunnel interface all Algos """
 207
 208     encryption_type = ESP
 209     tun4_encrypt_node_name = "esp4-encrypt"
 210     tun4_decrypt_node_name = "esp4-decrypt"
 211
 212     def config_network(self, p):
 213         config_tun_params(p, self.encryption_type, self.tun_if)
 214
 215         p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
 216                                         p.scapy_tun_spi,
 217                                         p.crypt_algo_vpp_id,
 218                                         p.crypt_key, p.crypt_key,
 219                                         p.auth_algo_vpp_id, p.auth_key,
 220                                         p.auth_key,
 221                                         salt=p.salt)
 222         p.tun_if.add_vpp_config()
 223         p.tun_if.admin_up()
 224         p.tun_if.config_ip4()
 225         self.logger.info(self.vapi.cli("sh ipsec sa 0"))
 226         self.logger.info(self.vapi.cli("sh ipsec sa 1"))
 227
 228         p.route = VppIpRoute(self, p.remote_tun_if_host, 32,
 229                              [VppRoutePath(p.tun_if.remote_ip4,
 230                                            0xffffffff)])
 231         p.route.add_vpp_config()
 232
 233     def unconfig_network(self, p):
 234         p.tun_if.unconfig_ip4()
 235         p.tun_if.remove_vpp_config()
 236         p.route.remove_vpp_config()
 237
 238     def setUp(self):
 239         super(TestIpsec4TunIfEspAll, self).setUp()
 240
 241         self.tun_if = self.pg0
 242
 243     def tearDown(self):
 244         super(TestIpsec4TunIfEspAll, self).tearDown()
 245
 246     def test_tun_44(self):
 247         """IPSEC tunnel all algos """
 248
 249         # foreach VPP crypto engine
 250         engines = ["ia32", "ipsecmb", "openssl"]
 251
 252         # foreach crypto algorithm
 253         algos = [{'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 254                                  IPSEC_API_CRYPTO_ALG_AES_GCM_128),
 255                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 256                                 IPSEC_API_INTEG_ALG_NONE),
 257                   'scapy-crypto': "AES-GCM",
 258                   'scapy-integ': "NULL",
 259                   'key': "JPjyOWBeVEQiMe7h",
 260                   'salt': 3333},
 261                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 262                                  IPSEC_API_CRYPTO_ALG_AES_GCM_192),
 263                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 264                                 IPSEC_API_INTEG_ALG_NONE),
 265                   'scapy-crypto': "AES-GCM",
 266                   'scapy-integ': "NULL",
 267                   'key': "JPjyOWBeVEQiMe7hJPjyOWBe",
 268                   'salt': 0},
 269                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 270                                  IPSEC_API_CRYPTO_ALG_AES_GCM_256),
 271                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 272                                 IPSEC_API_INTEG_ALG_NONE),
 273                   'scapy-crypto': "AES-GCM",
 274                   'scapy-integ': "NULL",
 275                   'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h",
 276                   'salt': 9999},
 277                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 278                                  IPSEC_API_CRYPTO_ALG_AES_CBC_128),
 279                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 280                                 IPSEC_API_INTEG_ALG_SHA1_96),
 281                   'scapy-crypto': "AES-CBC",
 282                   'scapy-integ': "HMAC-SHA1-96",
 283                   'salt': 0,
 284                   'key': "JPjyOWBeVEQiMe7h"},
 285                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 286                                  IPSEC_API_CRYPTO_ALG_AES_CBC_192),
 287                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 288                                 IPSEC_API_INTEG_ALG_SHA1_96),
 289                   'scapy-crypto': "AES-CBC",
 290                   'scapy-integ': "HMAC-SHA1-96",
 291                   'salt': 0,
 292                   'key': "JPjyOWBeVEQiMe7hJPjyOWBe"},
 293                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 294                                  IPSEC_API_CRYPTO_ALG_AES_CBC_256),
 295                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 296                                 IPSEC_API_INTEG_ALG_SHA1_96),
 297                   'scapy-crypto': "AES-CBC",
 298                   'scapy-integ': "HMAC-SHA1-96",
 299                   'salt': 0,
 300                   'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}]
 301
 302         for engine in engines:
 303             self.vapi.cli("set crypto handler all %s" % engine)
 304
 305             #
 306             # loop through each of the algorithms
 307             #
 308             for algo in algos:
 309                 # with self.subTest(algo=algo['scapy']):
 310
 311                 p = copy.copy(self.ipv4_params)
 312                 p.auth_algo_vpp_id = algo['vpp-integ']
 313                 p.crypt_algo_vpp_id = algo['vpp-crypto']
 314                 p.crypt_algo = algo['scapy-crypto']
 315                 p.auth_algo = algo['scapy-integ']
 316                 p.crypt_key = algo['key']
 317                 p.salt = algo['salt']
 318
 319                 self.config_network(p)
 320
 321                 self.verify_tun_44(p, count=127)
 322                 c = p.tun_if.get_rx_stats()
 323                 self.assertEqual(c['packets'], 127)
 324                 c = p.tun_if.get_tx_stats()
 325                 self.assertEqual(c['packets'], 127)
 326
 327                 self.unconfig_network(p)
 328
 329
 330 class TestIpsec6MultiTunIfEsp(TemplateIpsec, IpsecTun6):
 331     """ IPsec IPv6 Multi Tunnel interface """
 332
 333     encryption_type = ESP
 334     tun6_encrypt_node_name = "esp6-encrypt"
 335     tun6_decrypt_node_name = "esp6-decrypt"
 336
 337     def setUp(self):
 338         super(TestIpsec6MultiTunIfEsp, self).setUp()
 339
 340         self.tun_if = self.pg0
 341
 342         self.multi_params = []
 343
 344         for ii in range(10):
 345             p = copy.copy(self.ipv6_params)
 346
 347             p.remote_tun_if_host = "1111::%d" % (ii + 1)
 348             p.scapy_tun_sa_id = p.scapy_tun_sa_id + ii
 349             p.scapy_tun_spi = p.scapy_tun_spi + ii
 350             p.vpp_tun_sa_id = p.vpp_tun_sa_id + ii
 351             p.vpp_tun_spi = p.vpp_tun_spi + ii
 352
 353             p.scapy_tra_sa_id = p.scapy_tra_sa_id + ii
 354             p.scapy_tra_spi = p.scapy_tra_spi + ii
 355             p.vpp_tra_sa_id = p.vpp_tra_sa_id + ii
 356             p.vpp_tra_spi = p.vpp_tra_spi + ii
 357
 358             config_tun_params(p, self.encryption_type, self.tun_if)
 359             self.multi_params.append(p)
 360
 361             p.tun_if = VppIpsecTunInterface(self, self.pg0, p.vpp_tun_spi,
 362                                             p.scapy_tun_spi,
 363                                             p.crypt_algo_vpp_id,
 364                                             p.crypt_key, p.crypt_key,
 365                                             p.auth_algo_vpp_id, p.auth_key,
 366                                             p.auth_key, is_ip6=True)
 367             p.tun_if.add_vpp_config()
 368             p.tun_if.admin_up()
 369             p.tun_if.config_ip6()
 370
 371             VppIpRoute(self, p.remote_tun_if_host, 128,
 372                        [VppRoutePath(p.tun_if.remote_ip6,
 373                                      0xffffffff,
 374                                      proto=DpoProto.DPO_PROTO_IP6)],
 375                        is_ip6=1).add_vpp_config()
 376
 377     def tearDown(self):
 378         if not self.vpp_dead:
 379             self.vapi.cli("show hardware")
 380         super(TestIpsec6MultiTunIfEsp, self).tearDown()
 381
 382     def test_tun_66(self):
 383         """Multiple IPSEC tunnel interfaces """
 384         for p in self.multi_params:
 385             self.verify_tun_66(p, count=127)
 386             c = p.tun_if.get_rx_stats()
 387             self.assertEqual(c['packets'], 127)
 388             c = p.tun_if.get_tx_stats()
 389             self.assertEqual(c['packets'], 127)
 390
 391
 392 class TemplateIpsecGRETunIfEsp(TemplateIpsec):
 393     """ IPsec GRE tunnel interface tests """
 394
 395     encryption_type = ESP
 396     omac = "00:11:22:33:44:55"
 397
 398     def gen_encrypt_pkts(self, sa, sw_intf, src, dst, count=1,
 399                          payload_size=100):
 400         return [Ether(src=sw_intf.remote_mac, dst=sw_intf.local_mac) /
 401                 sa.encrypt(IP(src=self.pg0.remote_ip4,
 402                               dst=self.pg0.local_ip4) /
 403                            GRE() /
 404                            Ether(dst=self.omac) /
 405                            IP(src="1.1.1.1", dst="1.1.1.2") /
 406                            UDP(sport=1144, dport=2233) /
 407                            Raw('X' * payload_size))
 408                 for i in range(count)]
 409
 410     def gen_pkts(self, sw_intf, src, dst, count=1,
 411                  payload_size=100):
 412         return [Ether(dst=self.omac) /
 413                 IP(src="1.1.1.1", dst="1.1.1.2") /
 414                 UDP(sport=1144, dport=2233) /
 415                 Raw('X' * payload_size)
 416                 for i in range(count)]
 417
 418     def verify_decrypted(self, p, rxs):
 419         for rx in rxs:
 420             self.assert_equal(rx[Ether].dst, self.omac)
 421             self.assert_equal(rx[IP].dst, "1.1.1.2")
 422
 423     def verify_encrypted(self, p, sa, rxs):
 424         for rx in rxs:
 425             try:
 426                 pkt = sa.decrypt(rx[IP])
 427                 if not pkt.haslayer(IP):
 428                     pkt = IP(pkt[Raw].load)
 429                 self.assert_packet_checksums_valid(pkt)
 430                 self.assert_equal(pkt[IP].dst, self.pg0.remote_ip4)
 431                 self.assert_equal(pkt[IP].src, self.pg0.local_ip4)
 432                 self.assertTrue(pkt.haslayer(GRE))
 433                 e = pkt[Ether]
 434                 self.assertEqual(e[Ether].dst, self.omac)
 435                 self.assertEqual(e[IP].dst, "1.1.1.2")
 436             except (IndexError, AssertionError):
 437                 self.logger.debug(ppp("Unexpected packet:", rx))
 438                 try:
 439                     self.logger.debug(ppp("Decrypted packet:", pkt))
 440                 except:
 441                     pass
 442                 raise
 443
 444     def setUp(self):
 445         super(TemplateIpsecGRETunIfEsp, self).setUp()
 446
 447         self.tun_if = self.pg0
 448
 449         p = self.ipv4_params
 450
 451         bd1 = VppBridgeDomain(self, 1)
 452         bd1.add_vpp_config()
 453
 454         p.tun_sa_out = VppIpsecSA(self, p.scapy_tun_sa_id, p.scapy_tun_spi,
 455                                   p.auth_algo_vpp_id, p.auth_key,
 456                                   p.crypt_algo_vpp_id, p.crypt_key,
 457                                   self.vpp_esp_protocol,
 458                                   self.pg0.local_ip4,
 459                                   self.pg0.remote_ip4)
 460         p.tun_sa_out.add_vpp_config()
 461
 462         p.tun_sa_in = VppIpsecSA(self, p.vpp_tun_sa_id, p.vpp_tun_spi,
 463                                  p.auth_algo_vpp_id, p.auth_key,
 464                                  p.crypt_algo_vpp_id, p.crypt_key,
 465                                  self.vpp_esp_protocol,
 466                                  self.pg0.remote_ip4,
 467                                  self.pg0.local_ip4)
 468         p.tun_sa_in.add_vpp_config()
 469
 470         self.tun = VppIpsecGRETunInterface(self, self.pg0,
 471                                            p.tun_sa_out.id,
 472                                            p.tun_sa_in.id)
 473
 474         self.tun.add_vpp_config()
 475         self.tun.admin_up()
 476         self.tun.config_ip4()
 477
 478         VppIpRoute(self, p.remote_tun_if_host, 32,
 479                    [VppRoutePath(self.tun.remote_ip4,
 480                                  0xffffffff)]).add_vpp_config()
 481         VppBridgeDomainPort(self, bd1, self.tun).add_vpp_config()
 482         VppBridgeDomainPort(self, bd1, self.pg1).add_vpp_config()
 483
 484     def tearDown(self):
 485         if not self.vpp_dead:
 486             self.vapi.cli("show hardware")
 487         self.tun.unconfig_ip4()
 488         super(TemplateIpsecGRETunIfEsp, self).tearDown()
 489
 490
 491 @unittest.skipIf(is_skip_aarch64_set and is_platform_aarch64,
 492                  "test doesn't work on aarch64")
 493 class TestIpsecGRETunIfEsp1(TemplateIpsecGRETunIfEsp, IpsecTun4Tests):
 494     """ Ipsec GRE ESP - TUN tests """
 495     tun4_encrypt_node_name = "esp4-encrypt"
 496     tun4_decrypt_node_name = "esp4-decrypt"
 497
 498 if __name__ == '__main__':
 499     unittest.main(testRunner=VppTestRunner)