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