test/test_ipsec_ah.py

   1 import socket
   2 import unittest
   3
   4 from scapy.layers.ipsec import AH
   5
   6 from framework import VppTestRunner
   7 from template_ipsec import TemplateIpsec, IpsecTra46Tests, IpsecTun46Tests, \
   8     config_tun_params, config_tra_params, IPsecIPv4Params, IPsecIPv6Params, \
   9     IpsecTra4, IpsecTun4, IpsecTra6, IpsecTun6, \
  10     IpsecTun6HandoffTests, IpsecTun4HandoffTests
  11 from template_ipsec import IpsecTcpTests
  12 from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\
  13         VppIpsecSpdItfBinding
  14 from vpp_ip_route import VppIpRoute, VppRoutePath
  15 from vpp_ip import DpoProto
  16 from vpp_papi import VppEnum
  17
  18
  19 class ConfigIpsecAH(TemplateIpsec):
  20     """
  21     Basic test for IPSEC using AH transport and Tunnel mode
  22
  23     TRANSPORT MODE:
  24
  25      ---   encrypt   ---
  26     |pg2| <-------> |VPP|
  27      ---   decrypt   ---
  28
  29     TUNNEL MODE:
  30
  31      ---   encrypt   ---   plain   ---
  32     |pg0| <-------  |VPP| <------ |pg1|
  33      ---             ---           ---
  34
  35      ---   decrypt   ---   plain   ---
  36     |pg0| ------->  |VPP| ------> |pg1|
  37      ---             ---           ---
  38     """
  39     encryption_type = AH
  40     net_objs = []
  41     tra4_encrypt_node_name = "ah4-encrypt"
  42     tra4_decrypt_node_name = "ah4-decrypt"
  43     tra6_encrypt_node_name = "ah6-encrypt"
  44     tra6_decrypt_node_name = "ah6-decrypt"
  45     tun4_encrypt_node_name = "ah4-encrypt"
  46     tun4_decrypt_node_name = "ah4-decrypt"
  47     tun6_encrypt_node_name = "ah6-encrypt"
  48     tun6_decrypt_node_name = "ah6-decrypt"
  49
  50     @classmethod
  51     def setUpClass(cls):
  52         super(ConfigIpsecAH, cls).setUpClass()
  53
  54     @classmethod
  55     def tearDownClass(cls):
  56         super(ConfigIpsecAH, cls).tearDownClass()
  57
  58     def setUp(self):
  59         super(ConfigIpsecAH, self).setUp()
  60
  61     def tearDown(self):
  62         super(ConfigIpsecAH, self).tearDown()
  63
  64     def config_network(self, params):
  65         self.net_objs = []
  66         self.tun_if = self.pg0
  67         self.tra_if = self.pg2
  68         self.logger.info(self.vapi.ppcli("show int addr"))
  69
  70         self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
  71         self.tra_spd.add_vpp_config()
  72         self.net_objs.append(self.tra_spd)
  73         self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
  74         self.tun_spd.add_vpp_config()
  75         self.net_objs.append(self.tun_spd)
  76
  77         b = VppIpsecSpdItfBinding(self, self.tra_spd,
  78                                   self.tra_if)
  79         b.add_vpp_config()
  80         self.net_objs.append(b)
  81
  82         b = VppIpsecSpdItfBinding(self, self.tun_spd,
  83                                   self.tun_if)
  84         b.add_vpp_config()
  85         self.net_objs.append(b)
  86
  87         for p in params:
  88             self.config_ah_tra(p)
  89             config_tra_params(p, self.encryption_type)
  90         for p in params:
  91             self.config_ah_tun(p)
  92             config_tun_params(p, self.encryption_type, self.tun_if)
  93         for p in params:
  94             d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
  95             r = VppIpRoute(self,  p.remote_tun_if_host, p.addr_len,
  96                            [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
  97                                          0xffffffff,
  98                                          proto=d)])
  99             r.add_vpp_config()
 100             self.net_objs.append(r)
 101         self.logger.info(self.vapi.ppcli("show ipsec all"))
 102
 103     def unconfig_network(self):
 104         for o in reversed(self.net_objs):
 105             o.remove_vpp_config()
 106         self.net_objs = []
 107
 108     def config_ah_tun(self, params):
 109         addr_type = params.addr_type
 110         scapy_tun_sa_id = params.scapy_tun_sa_id
 111         scapy_tun_spi = params.scapy_tun_spi
 112         vpp_tun_sa_id = params.vpp_tun_sa_id
 113         vpp_tun_spi = params.vpp_tun_spi
 114         auth_algo_vpp_id = params.auth_algo_vpp_id
 115         auth_key = params.auth_key
 116         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 117         crypt_key = params.crypt_key
 118         remote_tun_if_host = params.remote_tun_if_host
 119         addr_any = params.addr_any
 120         addr_bcast = params.addr_bcast
 121         flags = params.flags
 122         e = VppEnum.vl_api_ipsec_spd_action_t
 123         objs = []
 124
 125         params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
 126                                       auth_algo_vpp_id, auth_key,
 127                                       crypt_algo_vpp_id, crypt_key,
 128                                       self.vpp_ah_protocol,
 129                                       self.tun_if.local_addr[addr_type],
 130                                       self.tun_if.remote_addr[addr_type],
 131                                       flags=flags)
 132
 133         params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
 134                                        auth_algo_vpp_id, auth_key,
 135                                        crypt_algo_vpp_id, crypt_key,
 136                                        self.vpp_ah_protocol,
 137                                        self.tun_if.remote_addr[addr_type],
 138                                        self.tun_if.local_addr[addr_type],
 139                                        flags=flags)
 140
 141         objs.append(params.tun_sa_in)
 142         objs.append(params.tun_sa_out)
 143
 144         params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
 145                                                     vpp_tun_sa_id,
 146                                                     addr_any, addr_bcast,
 147                                                     addr_any, addr_bcast,
 148                                                     socket.IPPROTO_AH)
 149         params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
 150                                                      vpp_tun_sa_id,
 151                                                      addr_any, addr_bcast,
 152                                                      addr_any, addr_bcast,
 153                                                      socket.IPPROTO_AH,
 154                                                      is_outbound=0)
 155
 156         objs.append(params.spd_policy_out_any)
 157         objs.append(params.spd_policy_in_any)
 158
 159         e1 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 160                               remote_tun_if_host,
 161                               remote_tun_if_host,
 162                               self.pg1.remote_addr[addr_type],
 163                               self.pg1.remote_addr[addr_type],
 164                               0, priority=10,
 165                               policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 166                               is_outbound=0)
 167         e2 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 168                               self.pg1.remote_addr[addr_type],
 169                               self.pg1.remote_addr[addr_type],
 170                               remote_tun_if_host,
 171                               remote_tun_if_host,
 172                               0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 173                               priority=10)
 174         e3 = VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 175                               remote_tun_if_host,
 176                               remote_tun_if_host,
 177                               self.pg0.local_addr[addr_type],
 178                               self.pg0.local_addr[addr_type],
 179                               0, priority=20,
 180                               policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 181                               is_outbound=0)
 182         e4 = VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 183                               self.pg0.local_addr[addr_type],
 184                               self.pg0.local_addr[addr_type],
 185                               remote_tun_if_host,
 186                               remote_tun_if_host,
 187                               0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 188                               priority=20)
 189
 190         objs = objs + [e1, e2, e3, e4]
 191
 192         for o in objs:
 193             o.add_vpp_config()
 194
 195         self.net_objs = self.net_objs + objs
 196
 197     def config_ah_tra(self, params):
 198         addr_type = params.addr_type
 199         scapy_tra_sa_id = params.scapy_tra_sa_id
 200         scapy_tra_spi = params.scapy_tra_spi
 201         vpp_tra_sa_id = params.vpp_tra_sa_id
 202         vpp_tra_spi = params.vpp_tra_spi
 203         auth_algo_vpp_id = params.auth_algo_vpp_id
 204         auth_key = params.auth_key
 205         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 206         crypt_key = params.crypt_key
 207         addr_any = params.addr_any
 208         addr_bcast = params.addr_bcast
 209         flags = params.flags | (VppEnum.vl_api_ipsec_sad_flags_t.
 210                                 IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
 211         e = VppEnum.vl_api_ipsec_spd_action_t
 212         objs = []
 213
 214         params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
 215                                       auth_algo_vpp_id, auth_key,
 216                                       crypt_algo_vpp_id, crypt_key,
 217                                       self.vpp_ah_protocol,
 218                                       flags=flags)
 219         params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
 220                                        auth_algo_vpp_id, auth_key,
 221                                        crypt_algo_vpp_id, crypt_key,
 222                                        self.vpp_ah_protocol,
 223                                        flags=flags)
 224
 225         objs.append(params.tra_sa_in)
 226         objs.append(params.tra_sa_out)
 227
 228         objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 229                                      addr_any, addr_bcast,
 230                                      addr_any, addr_bcast,
 231                                      socket.IPPROTO_AH))
 232         objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
 233                                      addr_any, addr_bcast,
 234                                      addr_any, addr_bcast,
 235                                      socket.IPPROTO_AH,
 236                                      is_outbound=0))
 237         objs.append(VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 238                                      self.tra_if.local_addr[addr_type],
 239                                      self.tra_if.local_addr[addr_type],
 240                                      self.tra_if.remote_addr[addr_type],
 241                                      self.tra_if.remote_addr[addr_type],
 242                                      0, priority=10,
 243                                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 244                                      is_outbound=0))
 245         objs.append(VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
 246                                      self.tra_if.local_addr[addr_type],
 247                                      self.tra_if.local_addr[addr_type],
 248                                      self.tra_if.remote_addr[addr_type],
 249                                      self.tra_if.remote_addr[addr_type],
 250                                      0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 251                                      priority=10))
 252
 253         for o in objs:
 254             o.add_vpp_config()
 255         self.net_objs = self.net_objs + objs
 256
 257
 258 class TemplateIpsecAh(ConfigIpsecAH):
 259     """
 260     Basic test for IPSEC using AH transport and Tunnel mode
 261
 262     TRANSPORT MODE:
 263
 264      ---   encrypt   ---
 265     |pg2| <-------> |VPP|
 266      ---   decrypt   ---
 267
 268     TUNNEL MODE:
 269
 270      ---   encrypt   ---   plain   ---
 271     |pg0| <-------  |VPP| <------ |pg1|
 272      ---             ---           ---
 273
 274      ---   decrypt   ---   plain   ---
 275     |pg0| ------->  |VPP| ------> |pg1|
 276      ---             ---           ---
 277     """
 278     @classmethod
 279     def setUpClass(cls):
 280         super(TemplateIpsecAh, cls).setUpClass()
 281
 282     @classmethod
 283     def tearDownClass(cls):
 284         super(TemplateIpsecAh, cls).tearDownClass()
 285
 286     def setUp(self):
 287         super(TemplateIpsecAh, self).setUp()
 288         self.config_network(self.params.values())
 289
 290     def tearDown(self):
 291         self.unconfig_network()
 292         super(TemplateIpsecAh, self).tearDown()
 293
 294
 295 class TestIpsecAh1(TemplateIpsecAh, IpsecTcpTests):
 296     """ Ipsec AH - TCP tests """
 297     pass
 298
 299
 300 class TestIpsecAh2(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
 301     """ Ipsec AH w/ SHA1 """
 302     pass
 303
 304
 305 class TestIpsecAhHandoff(TemplateIpsecAh,
 306                          IpsecTun6HandoffTests,
 307                          IpsecTun4HandoffTests):
 308     """ Ipsec AH Handoff """
 309     pass
 310
 311
 312 class TestIpsecAhAll(ConfigIpsecAH,
 313                      IpsecTra4, IpsecTra6,
 314                      IpsecTun4, IpsecTun6):
 315     """ Ipsec AH all Algos """
 316
 317     def setUp(self):
 318         super(TestIpsecAhAll, self).setUp()
 319
 320     def tearDown(self):
 321         super(TestIpsecAhAll, self).tearDown()
 322
 323     def test_integ_algs(self):
 324         """All Engines SHA[1_96, 256, 384, 512] w/ & w/o ESN"""
 325         # foreach VPP crypto engine
 326         engines = ["ia32", "ipsecmb", "openssl"]
 327
 328         algos = [{'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
 329                   IPSEC_API_INTEG_ALG_SHA1_96,
 330                   'scapy': "HMAC-SHA1-96"},
 331                  {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
 332                   IPSEC_API_INTEG_ALG_SHA_256_128,
 333                   'scapy': "SHA2-256-128"},
 334                  {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
 335                   IPSEC_API_INTEG_ALG_SHA_384_192,
 336                   'scapy': "SHA2-384-192"},
 337                  {'vpp': VppEnum.vl_api_ipsec_integ_alg_t.
 338                   IPSEC_API_INTEG_ALG_SHA_512_256,
 339                   'scapy': "SHA2-512-256"}]
 340
 341         flags = [0, (VppEnum.vl_api_ipsec_sad_flags_t.
 342                      IPSEC_API_SAD_FLAG_USE_ESN)]
 343
 344         #
 345         # loop through the VPP engines
 346         #
 347         for engine in engines:
 348             self.vapi.cli("set crypto handler all %s" % engine)
 349             #
 350             # loop through each of the algorithms
 351             #
 352             for algo in algos:
 353                 # with self.subTest(algo=algo['scapy']):
 354                 for flag in flags:
 355                     #
 356                     # setup up the config paramters
 357                     #
 358                     self.ipv4_params = IPsecIPv4Params()
 359                     self.ipv6_params = IPsecIPv6Params()
 360
 361                     self.params = {self.ipv4_params.addr_type:
 362                                    self.ipv4_params,
 363                                    self.ipv6_params.addr_type:
 364                                    self.ipv6_params}
 365
 366                     for _, p in self.params.items():
 367                         p.auth_algo_vpp_id = algo['vpp']
 368                         p.auth_algo = algo['scapy']
 369                         p.flags = p.flags | flag
 370
 371                     #
 372                     # configure the SPDs. SAs, etc
 373                     #
 374                     self.config_network(self.params.values())
 375
 376                     #
 377                     # run some traffic.
 378                     #  An exhautsive 4o6, 6o4 is not necessary for each algo
 379                     #
 380                     self.verify_tra_basic6(count=17)
 381                     self.verify_tra_basic4(count=17)
 382                     self.verify_tun_66(self.params[socket.AF_INET6], count=17)
 383                     self.verify_tun_44(self.params[socket.AF_INET], count=17)
 384
 385                     #
 386                     # remove the SPDs, SAs, etc
 387                     #
 388                     self.unconfig_network()
 389
 390
 391 if __name__ == '__main__':
 392     unittest.main(testRunner=VppTestRunner)