test/test_ipsec_esp.py

   1 import socket
   2 import unittest
   3 from scapy.layers.ipsec import ESP
   4 from scapy.layers.inet import UDP
   5
   6 from framework import VppTestRunner, is_skip_aarch64_set, is_platform_aarch64
   7 from template_ipsec import IpsecTra46Tests, IpsecTun46Tests, TemplateIpsec, \
   8     IpsecTcpTests, IpsecTun4Tests, IpsecTra4Tests, config_tra_params, \
   9     IPsecIPv4Params, IPsecIPv6Params, \
  10     IpsecTra4, IpsecTun4, IpsecTra6, IpsecTun6
  11 from vpp_ipsec import VppIpsecSpd, VppIpsecSpdEntry, VppIpsecSA,\
  12     VppIpsecSpdItfBinding
  13 from vpp_ip_route import VppIpRoute, VppRoutePath
  14 from vpp_ip import DpoProto
  15 from vpp_papi import VppEnum
  16
  17 NUM_PKTS = 67
  18
  19
  20 class ConfigIpsecESP(TemplateIpsec):
  21     encryption_type = ESP
  22     tra4_encrypt_node_name = "esp4-encrypt"
  23     tra4_decrypt_node_name = "esp4-decrypt"
  24     tra6_encrypt_node_name = "esp6-encrypt"
  25     tra6_decrypt_node_name = "esp6-decrypt"
  26     tun4_encrypt_node_name = "esp4-encrypt"
  27     tun4_decrypt_node_name = "esp4-decrypt"
  28     tun6_encrypt_node_name = "esp6-encrypt"
  29     tun6_decrypt_node_name = "esp6-decrypt"
  30
  31     @classmethod
  32     def setUpClass(cls):
  33         super(ConfigIpsecESP, cls).setUpClass()
  34
  35     @classmethod
  36     def tearDownClass(cls):
  37         super(ConfigIpsecESP, cls).tearDownClass()
  38
  39     def setUp(self):
  40         super(ConfigIpsecESP, self).setUp()
  41
  42     def tearDown(self):
  43         super(ConfigIpsecESP, self).tearDown()
  44
  45     def config_network(self, params):
  46         self.net_objs = []
  47         self.tun_if = self.pg0
  48         self.tra_if = self.pg2
  49         self.logger.info(self.vapi.ppcli("show int addr"))
  50
  51         self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
  52         self.tra_spd.add_vpp_config()
  53         self.net_objs.append(self.tra_spd)
  54         self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
  55         self.tun_spd.add_vpp_config()
  56         self.net_objs.append(self.tun_spd)
  57
  58         b = VppIpsecSpdItfBinding(self, self.tun_spd,
  59                                   self.tun_if)
  60         b.add_vpp_config()
  61         self.net_objs.append(b)
  62
  63         b = VppIpsecSpdItfBinding(self, self.tra_spd,
  64                                   self.tra_if)
  65         b.add_vpp_config()
  66         self.net_objs.append(b)
  67
  68         for p in params:
  69             self.config_esp_tra(p)
  70             config_tra_params(p, self.encryption_type)
  71         for p in params:
  72             self.config_esp_tun(p)
  73
  74         for p in params:
  75             d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
  76             r = VppIpRoute(self,  p.remote_tun_if_host, p.addr_len,
  77                            [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
  78                                          0xffffffff,
  79                                          proto=d)])
  80             r.add_vpp_config()
  81             self.net_objs.append(r)
  82
  83         self.logger.info(self.vapi.ppcli("show ipsec all"))
  84
  85     def unconfig_network(self):
  86         for o in reversed(self.net_objs):
  87             o.remove_vpp_config()
  88         self.net_objs = []
  89
  90     def config_esp_tun(self, params):
  91         addr_type = params.addr_type
  92         scapy_tun_sa_id = params.scapy_tun_sa_id
  93         scapy_tun_spi = params.scapy_tun_spi
  94         vpp_tun_sa_id = params.vpp_tun_sa_id
  95         vpp_tun_spi = params.vpp_tun_spi
  96         auth_algo_vpp_id = params.auth_algo_vpp_id
  97         auth_key = params.auth_key
  98         crypt_algo_vpp_id = params.crypt_algo_vpp_id
  99         crypt_key = params.crypt_key
 100         remote_tun_if_host = params.remote_tun_if_host
 101         addr_any = params.addr_any
 102         addr_bcast = params.addr_bcast
 103         e = VppEnum.vl_api_ipsec_spd_action_t
 104         flags = params.flags
 105         salt = params.salt
 106         objs = []
 107
 108         params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
 109                                       auth_algo_vpp_id, auth_key,
 110                                       crypt_algo_vpp_id, crypt_key,
 111                                       self.vpp_esp_protocol,
 112                                       self.tun_if.local_addr[addr_type],
 113                                       self.tun_if.remote_addr[addr_type],
 114                                       flags=flags,
 115                                       salt=salt)
 116         params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
 117                                        auth_algo_vpp_id, auth_key,
 118                                        crypt_algo_vpp_id, crypt_key,
 119                                        self.vpp_esp_protocol,
 120                                        self.tun_if.remote_addr[addr_type],
 121                                        self.tun_if.local_addr[addr_type],
 122                                        flags=flags,
 123                                        salt=salt)
 124         objs.append(params.tun_sa_in)
 125         objs.append(params.tun_sa_out)
 126
 127         params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
 128                                                     scapy_tun_sa_id,
 129                                                     addr_any, addr_bcast,
 130                                                     addr_any, addr_bcast,
 131                                                     socket.IPPROTO_ESP)
 132         params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
 133                                                      scapy_tun_sa_id,
 134                                                      addr_any, addr_bcast,
 135                                                      addr_any, addr_bcast,
 136                                                      socket.IPPROTO_ESP,
 137                                                      is_outbound=0)
 138         objs.append(params.spd_policy_out_any)
 139         objs.append(params.spd_policy_in_any)
 140
 141         objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 142                                      remote_tun_if_host, remote_tun_if_host,
 143                                      self.pg1.remote_addr[addr_type],
 144                                      self.pg1.remote_addr[addr_type],
 145                                      0,
 146                                      priority=10,
 147                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 148                                      is_outbound=0))
 149         objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 150                                      self.pg1.remote_addr[addr_type],
 151                                      self.pg1.remote_addr[addr_type],
 152                                      remote_tun_if_host, remote_tun_if_host,
 153                                      0,
 154                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 155                                      priority=10))
 156         objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 157                                      remote_tun_if_host, remote_tun_if_host,
 158                                      self.pg0.local_addr[addr_type],
 159                                      self.pg0.local_addr[addr_type],
 160                                      0,
 161                                      priority=20,
 162                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 163                                      is_outbound=0))
 164         objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 165                                      self.pg0.local_addr[addr_type],
 166                                      self.pg0.local_addr[addr_type],
 167                                      remote_tun_if_host, remote_tun_if_host,
 168                                      0,
 169                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 170                                      priority=20))
 171         for o in objs:
 172             o.add_vpp_config()
 173         self.net_objs = self.net_objs + objs
 174
 175     def config_esp_tra(self, params):
 176         addr_type = params.addr_type
 177         scapy_tra_sa_id = params.scapy_tra_sa_id
 178         scapy_tra_spi = params.scapy_tra_spi
 179         vpp_tra_sa_id = params.vpp_tra_sa_id
 180         vpp_tra_spi = params.vpp_tra_spi
 181         auth_algo_vpp_id = params.auth_algo_vpp_id
 182         auth_key = params.auth_key
 183         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 184         crypt_key = params.crypt_key
 185         addr_any = params.addr_any
 186         addr_bcast = params.addr_bcast
 187         flags = (VppEnum.vl_api_ipsec_sad_flags_t.
 188                  IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
 189         e = VppEnum.vl_api_ipsec_spd_action_t
 190         flags = params.flags | flags
 191         salt = params.salt
 192         objs = []
 193
 194         params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
 195                                       auth_algo_vpp_id, auth_key,
 196                                       crypt_algo_vpp_id, crypt_key,
 197                                       self.vpp_esp_protocol,
 198                                       flags=flags,
 199                                       salt=salt)
 200         params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
 201                                        auth_algo_vpp_id, auth_key,
 202                                        crypt_algo_vpp_id, crypt_key,
 203                                        self.vpp_esp_protocol,
 204                                        flags=flags,
 205                                        salt=salt)
 206         objs.append(params.tra_sa_in)
 207         objs.append(params.tra_sa_out)
 208
 209         objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 210                                      addr_any, addr_bcast,
 211                                      addr_any, addr_bcast,
 212                                      socket.IPPROTO_ESP))
 213         objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 214                                      addr_any, addr_bcast,
 215                                      addr_any, addr_bcast,
 216                                      socket.IPPROTO_ESP,
 217                                      is_outbound=0))
 218         objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 219                                      self.tra_if.local_addr[addr_type],
 220                                      self.tra_if.local_addr[addr_type],
 221                                      self.tra_if.remote_addr[addr_type],
 222                                      self.tra_if.remote_addr[addr_type],
 223                                      0, priority=10,
 224                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 225                                      is_outbound=0))
 226         objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
 227                                      self.tra_if.local_addr[addr_type],
 228                                      self.tra_if.local_addr[addr_type],
 229                                      self.tra_if.remote_addr[addr_type],
 230                                      self.tra_if.remote_addr[addr_type],
 231                                      0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 232                                      priority=10))
 233         for o in objs:
 234             o.add_vpp_config()
 235         self.net_objs = self.net_objs + objs
 236
 237
 238 class TemplateIpsecEsp(ConfigIpsecESP):
 239     """
 240     Basic test for ipsec esp sanity - tunnel and transport modes.
 241
 242     Below 4 cases are covered as part of this test
 243     1) ipsec esp v4 transport basic test  - IPv4 Transport mode
 244         scenario using HMAC-SHA1-96 integrity algo
 245     2) ipsec esp v4 transport burst test
 246         Above test for 257 pkts
 247     3) ipsec esp 4o4 tunnel basic test    - IPv4 Tunnel mode
 248         scenario using HMAC-SHA1-96 integrity algo
 249     4) ipsec esp 4o4 tunnel burst test
 250         Above test for 257 pkts
 251
 252     TRANSPORT MODE:
 253
 254      ---   encrypt   ---
 255     |pg2| <-------> |VPP|
 256      ---   decrypt   ---
 257
 258     TUNNEL MODE:
 259
 260      ---   encrypt   ---   plain   ---
 261     |pg0| <-------  |VPP| <------ |pg1|
 262      ---             ---           ---
 263
 264      ---   decrypt   ---   plain   ---
 265     |pg0| ------->  |VPP| ------> |pg1|
 266      ---             ---           ---
 267     """
 268
 269     @classmethod
 270     def setUpClass(cls):
 271         super(TemplateIpsecEsp, cls).setUpClass()
 272
 273     @classmethod
 274     def tearDownClass(cls):
 275         super(TemplateIpsecEsp, cls).tearDownClass()
 276
 277     def setUp(self):
 278         super(TemplateIpsecEsp, self).setUp()
 279         self.config_network(self.params.values())
 280
 281     def tearDown(self):
 282         self.unconfig_network()
 283         super(TemplateIpsecEsp, self).tearDown()
 284
 285
 286 class TestIpsecEsp1(TemplateIpsecEsp, IpsecTra46Tests, IpsecTun46Tests):
 287     """ Ipsec ESP - TUN & TRA tests """
 288     pass
 289
 290
 291 class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests):
 292     """ Ipsec ESP - TCP tests """
 293     pass
 294
 295
 296 class TemplateIpsecEspUdp(ConfigIpsecESP):
 297     """
 298     UDP encapped ESP
 299     """
 300
 301     @classmethod
 302     def setUpClass(cls):
 303         super(TemplateIpsecEspUdp, cls).setUpClass()
 304
 305     @classmethod
 306     def tearDownClass(cls):
 307         super(TemplateIpsecEspUdp, cls).tearDownClass()
 308
 309     def setUp(self):
 310         super(TemplateIpsecEspUdp, self).setUp()
 311         self.net_objs = []
 312         self.tun_if = self.pg0
 313         self.tra_if = self.pg2
 314         self.logger.info(self.vapi.ppcli("show int addr"))
 315
 316         p = self.ipv4_params
 317         p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
 318                    IPSEC_API_SAD_FLAG_UDP_ENCAP)
 319         p.nat_header = UDP(sport=5454, dport=4500)
 320
 321         self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
 322         self.tra_spd.add_vpp_config()
 323         VppIpsecSpdItfBinding(self, self.tra_spd,
 324                               self.tra_if).add_vpp_config()
 325
 326         self.config_esp_tra(p)
 327         config_tra_params(p, self.encryption_type)
 328
 329         self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
 330         self.tun_spd.add_vpp_config()
 331         VppIpsecSpdItfBinding(self, self.tun_spd,
 332                               self.tun_if).add_vpp_config()
 333
 334         self.config_esp_tun(p)
 335         self.logger.info(self.vapi.ppcli("show ipsec all"))
 336
 337         d = DpoProto.DPO_PROTO_IP4
 338         VppIpRoute(self,  p.remote_tun_if_host, p.addr_len,
 339                    [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
 340                                  0xffffffff,
 341                                  proto=d)]).add_vpp_config()
 342
 343     def tearDown(self):
 344         super(TemplateIpsecEspUdp, self).tearDown()
 345
 346     def show_commands_at_teardown(self):
 347         self.logger.info(self.vapi.cli("show hardware"))
 348
 349
 350 class TestIpsecEspUdp(TemplateIpsecEspUdp, IpsecTra4Tests):
 351     """ Ipsec NAT-T ESP UDP tests """
 352     pass
 353
 354
 355 @unittest.skipIf(is_skip_aarch64_set and is_platform_aarch64,
 356                  "test doesn't work on aarch64")
 357 class TestIpsecEspAll(ConfigIpsecESP,
 358                       IpsecTra4, IpsecTra6,
 359                       IpsecTun4, IpsecTun6):
 360     """ Ipsec ESP all Algos """
 361
 362     def setUp(self):
 363         super(TestIpsecEspAll, self).setUp()
 364
 365     def tearDown(self):
 366         super(TestIpsecEspAll, self).tearDown()
 367
 368     def test_crypto_algs(self):
 369         """All engines AES-[CBC, GCM]-[128, 192, 256] 3DES-CBC w/ & w/o ESN"""
 370
 371         # foreach VPP crypto engine
 372         engines = ["ia32", "ipsecmb", "openssl"]
 373
 374         # foreach crypto algorithm
 375         algos = [{'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 376                                  IPSEC_API_CRYPTO_ALG_AES_GCM_128),
 377                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 378                                 IPSEC_API_INTEG_ALG_NONE),
 379                   'scapy-crypto': "AES-GCM",
 380                   'scapy-integ': "NULL",
 381                   'key': "JPjyOWBeVEQiMe7h",
 382                   'salt': 0},
 383                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 384                                  IPSEC_API_CRYPTO_ALG_AES_GCM_192),
 385                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 386                                 IPSEC_API_INTEG_ALG_NONE),
 387                   'scapy-crypto': "AES-GCM",
 388                   'scapy-integ': "NULL",
 389                   'key': "JPjyOWBeVEQiMe7h01234567",
 390                   'salt': 1010},
 391                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 392                                  IPSEC_API_CRYPTO_ALG_AES_GCM_256),
 393                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 394                                 IPSEC_API_INTEG_ALG_NONE),
 395                   'scapy-crypto': "AES-GCM",
 396                   'scapy-integ': "NULL",
 397                   'key': "JPjyOWBeVEQiMe7h0123456787654321",
 398                   'salt': 2020},
 399                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 400                                  IPSEC_API_CRYPTO_ALG_AES_CBC_128),
 401                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 402                                 IPSEC_API_INTEG_ALG_SHA1_96),
 403                   'scapy-crypto': "AES-CBC",
 404                   'scapy-integ': "HMAC-SHA1-96",
 405                   'salt': 0,
 406                   'key': "JPjyOWBeVEQiMe7h"},
 407                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 408                                  IPSEC_API_CRYPTO_ALG_AES_CBC_192),
 409                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 410                                 IPSEC_API_INTEG_ALG_SHA1_96),
 411                   'scapy-crypto': "AES-CBC",
 412                   'scapy-integ': "HMAC-SHA1-96",
 413                   'salt': 0,
 414                   'key': "JPjyOWBeVEQiMe7hJPjyOWBe"},
 415                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 416                                  IPSEC_API_CRYPTO_ALG_AES_CBC_256),
 417                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 418                                 IPSEC_API_INTEG_ALG_SHA1_96),
 419                   'scapy-crypto': "AES-CBC",
 420                   'scapy-integ': "HMAC-SHA1-96",
 421                   'salt': 0,
 422                   'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"},
 423                  {'vpp-crypto': (VppEnum.vl_api_ipsec_crypto_alg_t.
 424                                  IPSEC_API_CRYPTO_ALG_3DES_CBC),
 425                   'vpp-integ': (VppEnum.vl_api_ipsec_integ_alg_t.
 426                                 IPSEC_API_INTEG_ALG_SHA1_96),
 427                   'scapy-crypto': "3DES",
 428                   'scapy-integ': "HMAC-SHA1-96",
 429                   'salt': 0,
 430                   'key': "JPjyOWBeVEQiMe7h00112233"}]
 431
 432         # with and without ESN
 433         flags = [0,
 434                  VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN]
 435
 436         #
 437         # loop through the VPP engines
 438         #
 439         for engine in engines:
 440             self.vapi.cli("set crypto handler all %s" % engine)
 441             #
 442             # loop through each of the algorithms
 443             #
 444             for algo in algos:
 445                 # with self.subTest(algo=algo['scapy']):
 446                 for flag in flags:
 447                     #
 448                     # setup up the config paramters
 449                     #
 450                     self.ipv4_params = IPsecIPv4Params()
 451                     self.ipv6_params = IPsecIPv6Params()
 452
 453                     self.params = {self.ipv4_params.addr_type:
 454                                    self.ipv4_params,
 455                                    self.ipv6_params.addr_type:
 456                                    self.ipv6_params}
 457
 458                     for _, p in self.params.items():
 459                         p.auth_algo_vpp_id = algo['vpp-integ']
 460                         p.crypt_algo_vpp_id = algo['vpp-crypto']
 461                         p.crypt_algo = algo['scapy-crypto']
 462                         p.auth_algo = algo['scapy-integ']
 463                         p.crypt_key = algo['key']
 464                         p.salt = algo['salt']
 465                         p.flags = p.flags | flag
 466
 467                     #
 468                     # configure the SPDs. SAs, etc
 469                     #
 470                     self.config_network(self.params.values())
 471
 472                     #
 473                     # run some traffic.
 474                     #  An exhautsive 4o6, 6o4 is not necessary
 475                     #  for each algo
 476                     #
 477                     self.verify_tra_basic6(count=NUM_PKTS)
 478                     self.verify_tra_basic4(count=NUM_PKTS)
 479                     self.verify_tun_66(self.params[socket.AF_INET6],
 480                                        count=NUM_PKTS)
 481                     self.verify_tun_44(self.params[socket.AF_INET],
 482                                        count=NUM_PKTS)
 483
 484                     #
 485                     # remove the SPDs, SAs, etc
 486                     #
 487                     self.unconfig_network()
 488
 489
 490 if __name__ == '__main__':
 491     unittest.main(testRunner=VppTestRunner)