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