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                            is_ip6=p.is_ipv6)
  79             r.add_vpp_config()
  80             self.net_objs.append(r)
  81
  82         self.logger.info(self.vapi.ppcli("show ipsec all"))
  83
  84     def unconfig_network(self):
  85         for o in reversed(self.net_objs):
  86             o.remove_vpp_config()
  87         self.net_objs = []
  88
  89     def config_esp_tun(self, params):
  90         addr_type = params.addr_type
  91         scapy_tun_sa_id = params.scapy_tun_sa_id
  92         scapy_tun_spi = params.scapy_tun_spi
  93         vpp_tun_sa_id = params.vpp_tun_sa_id
  94         vpp_tun_spi = params.vpp_tun_spi
  95         auth_algo_vpp_id = params.auth_algo_vpp_id
  96         auth_key = params.auth_key
  97         crypt_algo_vpp_id = params.crypt_algo_vpp_id
  98         crypt_key = params.crypt_key
  99         remote_tun_if_host = params.remote_tun_if_host
 100         addr_any = params.addr_any
 101         addr_bcast = params.addr_bcast
 102         e = VppEnum.vl_api_ipsec_spd_action_t
 103         objs = []
 104
 105         params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
 106                                       auth_algo_vpp_id, auth_key,
 107                                       crypt_algo_vpp_id, crypt_key,
 108                                       self.vpp_esp_protocol,
 109                                       self.tun_if.local_addr[addr_type],
 110                                       self.tun_if.remote_addr[addr_type])
 111         params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
 112                                        auth_algo_vpp_id, auth_key,
 113                                        crypt_algo_vpp_id, crypt_key,
 114                                        self.vpp_esp_protocol,
 115                                        self.tun_if.remote_addr[addr_type],
 116                                        self.tun_if.local_addr[addr_type])
 117         objs.append(params.tun_sa_in)
 118         objs.append(params.tun_sa_out)
 119
 120         params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
 121                                                     scapy_tun_sa_id,
 122                                                     addr_any, addr_bcast,
 123                                                     addr_any, addr_bcast,
 124                                                     socket.IPPROTO_ESP)
 125         params.spd_policy_out_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                                                      is_outbound=0)
 131         objs.append(params.spd_policy_out_any)
 132         objs.append(params.spd_policy_in_any)
 133
 134         objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 135                                      remote_tun_if_host, remote_tun_if_host,
 136                                      self.pg1.remote_addr[addr_type],
 137                                      self.pg1.remote_addr[addr_type],
 138                                      0,
 139                                      priority=10,
 140                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 141                                      is_outbound=0))
 142         objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 143                                      self.pg1.remote_addr[addr_type],
 144                                      self.pg1.remote_addr[addr_type],
 145                                      remote_tun_if_host, remote_tun_if_host,
 146                                      0,
 147                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 148                                      priority=10))
 149         objs.append(VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 150                                      remote_tun_if_host, remote_tun_if_host,
 151                                      self.pg0.local_addr[addr_type],
 152                                      self.pg0.local_addr[addr_type],
 153                                      0,
 154                                      priority=20,
 155                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 156                                      is_outbound=0))
 157         objs.append(VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 158                                      self.pg0.local_addr[addr_type],
 159                                      self.pg0.local_addr[addr_type],
 160                                      remote_tun_if_host, remote_tun_if_host,
 161                                      0,
 162                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 163                                      priority=20))
 164         for o in objs:
 165             o.add_vpp_config()
 166         self.net_objs = self.net_objs + objs
 167
 168     def config_esp_tra(self, params):
 169         addr_type = params.addr_type
 170         scapy_tra_sa_id = params.scapy_tra_sa_id
 171         scapy_tra_spi = params.scapy_tra_spi
 172         vpp_tra_sa_id = params.vpp_tra_sa_id
 173         vpp_tra_spi = params.vpp_tra_spi
 174         auth_algo_vpp_id = params.auth_algo_vpp_id
 175         auth_key = params.auth_key
 176         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 177         crypt_key = params.crypt_key
 178         addr_any = params.addr_any
 179         addr_bcast = params.addr_bcast
 180         flags = (VppEnum.vl_api_ipsec_sad_flags_t.
 181                  IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
 182         e = VppEnum.vl_api_ipsec_spd_action_t
 183         flags = params.flags | flags
 184         objs = []
 185
 186         params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
 187                                       auth_algo_vpp_id, auth_key,
 188                                       crypt_algo_vpp_id, crypt_key,
 189                                       self.vpp_esp_protocol,
 190                                       flags=flags)
 191         params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
 192                                        auth_algo_vpp_id, auth_key,
 193                                        crypt_algo_vpp_id, crypt_key,
 194                                        self.vpp_esp_protocol,
 195                                        flags=flags)
 196         objs.append(params.tra_sa_in)
 197         objs.append(params.tra_sa_out)
 198
 199         objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 200                                      addr_any, addr_bcast,
 201                                      addr_any, addr_bcast,
 202                                      socket.IPPROTO_ESP))
 203         objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 204                                      addr_any, addr_bcast,
 205                                      addr_any, addr_bcast,
 206                                      socket.IPPROTO_ESP,
 207                                      is_outbound=0))
 208         objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 209                                      self.tra_if.local_addr[addr_type],
 210                                      self.tra_if.local_addr[addr_type],
 211                                      self.tra_if.remote_addr[addr_type],
 212                                      self.tra_if.remote_addr[addr_type],
 213                                      0, priority=10,
 214                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 215                                      is_outbound=0))
 216         objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_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, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 222                                      priority=10))
 223         for o in objs:
 224             o.add_vpp_config()
 225         self.net_objs = self.net_objs + objs
 226
 227
 228 class TemplateIpsecEsp(ConfigIpsecESP):
 229     """
 230     Basic test for ipsec esp sanity - tunnel and transport modes.
 231
 232     Below 4 cases are covered as part of this test
 233     1) ipsec esp v4 transport basic test  - IPv4 Transport mode
 234         scenario using HMAC-SHA1-96 integrity algo
 235     2) ipsec esp v4 transport burst test
 236         Above test for 257 pkts
 237     3) ipsec esp 4o4 tunnel basic test    - IPv4 Tunnel mode
 238         scenario using HMAC-SHA1-96 integrity algo
 239     4) ipsec esp 4o4 tunnel burst test
 240         Above test for 257 pkts
 241
 242     TRANSPORT MODE:
 243
 244      ---   encrypt   ---
 245     |pg2| <-------> |VPP|
 246      ---   decrypt   ---
 247
 248     TUNNEL MODE:
 249
 250      ---   encrypt   ---   plain   ---
 251     |pg0| <-------  |VPP| <------ |pg1|
 252      ---             ---           ---
 253
 254      ---   decrypt   ---   plain   ---
 255     |pg0| ------->  |VPP| ------> |pg1|
 256      ---             ---           ---
 257     """
 258
 259     @classmethod
 260     def setUpClass(cls):
 261         super(TemplateIpsecEsp, cls).setUpClass()
 262
 263     @classmethod
 264     def tearDownClass(cls):
 265         super(TemplateIpsecEsp, cls).tearDownClass()
 266
 267     def setUp(self):
 268         super(TemplateIpsecEsp, self).setUp()
 269         self.config_network(self.params.values())
 270
 271     def tearDown(self):
 272         self.unconfig_network()
 273         super(TemplateIpsecEsp, self).tearDown()
 274
 275
 276 class TestIpsecEsp1(TemplateIpsecEsp, IpsecTra46Tests, IpsecTun46Tests):
 277     """ Ipsec ESP - TUN & TRA tests """
 278     pass
 279
 280
 281 class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests):
 282     """ Ipsec ESP - TCP tests """
 283     pass
 284
 285
 286 class TemplateIpsecEspUdp(ConfigIpsecESP):
 287     """
 288     UDP encapped ESP
 289     """
 290
 291     @classmethod
 292     def setUpClass(cls):
 293         super(TemplateIpsecEspUdp, cls).setUpClass()
 294
 295     @classmethod
 296     def tearDownClass(cls):
 297         super(TemplateIpsecEspUdp, cls).tearDownClass()
 298
 299     def setUp(self):
 300         super(TemplateIpsecEspUdp, self).setUp()
 301         self.net_objs = []
 302         self.tun_if = self.pg0
 303         self.tra_if = self.pg2
 304         self.logger.info(self.vapi.ppcli("show int addr"))
 305
 306         p = self.ipv4_params
 307         p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
 308                    IPSEC_API_SAD_FLAG_UDP_ENCAP)
 309         p.nat_header = UDP(sport=5454, dport=4500)
 310
 311         self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
 312         self.tra_spd.add_vpp_config()
 313         VppIpsecSpdItfBinding(self, self.tra_spd,
 314                               self.tra_if).add_vpp_config()
 315
 316         self.config_esp_tra(p)
 317         config_tra_params(p, self.encryption_type)
 318
 319         self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
 320         self.tun_spd.add_vpp_config()
 321         VppIpsecSpdItfBinding(self, self.tun_spd,
 322                               self.tun_if).add_vpp_config()
 323
 324         self.config_esp_tun(p)
 325         self.logger.info(self.vapi.ppcli("show ipsec all"))
 326
 327         d = DpoProto.DPO_PROTO_IP4
 328         VppIpRoute(self,  p.remote_tun_if_host, p.addr_len,
 329                    [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
 330                                  0xffffffff,
 331                                  proto=d)]).add_vpp_config()
 332
 333     def tearDown(self):
 334         super(TemplateIpsecEspUdp, self).tearDown()
 335         if not self.vpp_dead:
 336             self.vapi.cli("show hardware")
 337
 338
 339 class TestIpsecEspUdp(TemplateIpsecEspUdp, IpsecTra4Tests, IpsecTun4Tests):
 340     """ Ipsec NAT-T ESP UDP tests """
 341     pass
 342
 343
 344 class TestIpsecEspAll(ConfigIpsecESP,
 345                       IpsecTra4, IpsecTra6,
 346                       IpsecTun4, IpsecTun6):
 347     """ Ipsec ESP all Algos """
 348
 349     def setUp(self):
 350         super(TestIpsecEspAll, self).setUp()
 351
 352     def tearDown(self):
 353         super(TestIpsecEspAll, self).tearDown()
 354
 355     def test_crypto_algs(self):
 356         """All engines AES-CBC-[128, 192, 256] w/o ESN"""
 357
 358         # foreach VPP crypto engine
 359         engines = ["ia32", "ipsecmb", "openssl"]
 360
 361         # foreach crypto algorithm
 362         algos = [{'vpp': VppEnum.vl_api_ipsec_crypto_alg_t.
 363                   IPSEC_API_CRYPTO_ALG_AES_CBC_128,
 364                   'scapy': "AES-CBC",
 365                   'key': "JPjyOWBeVEQiMe7h"},
 366                  {'vpp': VppEnum.vl_api_ipsec_crypto_alg_t.
 367                   IPSEC_API_CRYPTO_ALG_AES_CBC_192,
 368                   'scapy': "AES-CBC",
 369                   'key': "JPjyOWBeVEQiMe7hJPjyOWBe"},
 370                  {'vpp': VppEnum.vl_api_ipsec_crypto_alg_t.
 371                   IPSEC_API_CRYPTO_ALG_AES_CBC_256,
 372                   'scapy': "AES-CBC",
 373                   'key': "JPjyOWBeVEQiMe7hJPjyOWBeVEQiMe7h"}]
 374
 375         # bug found in VPP needs fixing with flag
 376         # (VppEnum.vl_api_ipsec_sad_flags_t.IPSEC_API_SAD_FLAG_USE_ESN)
 377         flags = [0]
 378
 379         #
 380         # loop through the VPP engines
 381         #
 382         for engine in engines:
 383             self.vapi.cli("set crypto engine all %s" % engine)
 384
 385             #
 386             # loop through each of the algorithms
 387             #
 388             for algo in algos:
 389                 # with self.subTest(algo=algo['scapy']):
 390                 for flag in flags:
 391                     #
 392                     # setup up the config paramters
 393                     #
 394                     self.ipv4_params = IPsecIPv4Params()
 395                     self.ipv6_params = IPsecIPv6Params()
 396
 397                     self.params = {self.ipv4_params.addr_type:
 398                                    self.ipv4_params,
 399                                    self.ipv6_params.addr_type:
 400                                    self.ipv6_params}
 401
 402                     for _, p in self.params.items():
 403                         p.crypt_algo_vpp_id = algo['vpp']
 404                         p.crypt_algo = algo['scapy']
 405                         p.crypt_key = algo['key']
 406                         p.flags = p.flags | flag
 407
 408                     #
 409                     # configure the SPDs. SAs, etc
 410                     #
 411                     self.config_network(self.params.values())
 412
 413                     #
 414                     # run some traffic.
 415                     #  An exhautsive 4o6, 6o4 is not necessary
 416                     #  for each algo
 417                     #
 418                     self.verify_tra_basic6(count=17)
 419                     self.verify_tra_basic4(count=17)
 420                     self.verify_tun_66(self.params[socket.AF_INET6], 17)
 421                     self.verify_tun_44(self.params[socket.AF_INET], 17)
 422
 423                     #
 424                     # remove the SPDs, SAs, etc
 425                     #
 426                     self.unconfig_network()
 427
 428
 429 if __name__ == '__main__':
 430     unittest.main(testRunner=VppTestRunner)