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 from template_ipsec import IpsecTcpTests
  10 from vpp_ipsec import VppIpsecSA, VppIpsecSpd, VppIpsecSpdEntry,\
  11         VppIpsecSpdItfBinding
  12 from vpp_ip_route import VppIpRoute, VppRoutePath
  13 from vpp_ip import DpoProto
  14 from vpp_papi import VppEnum
  15
  16
  17 class TemplateIpsecAh(TemplateIpsec):
  18     """
  19     Basic test for IPSEC using AH transport and Tunnel mode
  20
  21     TRANSPORT MODE:
  22
  23      ---   encrypt   ---
  24     |pg2| <-------> |VPP|
  25      ---   decrypt   ---
  26
  27     TUNNEL MODE:
  28
  29      ---   encrypt   ---   plain   ---
  30     |pg0| <-------  |VPP| <------ |pg1|
  31      ---             ---           ---
  32
  33      ---   decrypt   ---   plain   ---
  34     |pg0| ------->  |VPP| ------> |pg1|
  35      ---             ---           ---
  36     """
  37
  38     @classmethod
  39     def setUpClass(cls):
  40         super(TemplateIpsecAh, cls).setUpClass()
  41
  42     @classmethod
  43     def tearDownClass(cls):
  44         super(TemplateIpsecAh, cls).tearDownClass()
  45
  46     def setUp(self):
  47         super(TemplateIpsecAh, self).setUp()
  48
  49         self.encryption_type = AH
  50         self.tun_if = self.pg0
  51         self.tra_if = self.pg2
  52         self.logger.info(self.vapi.ppcli("show int addr"))
  53
  54         self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
  55         self.tra_spd.add_vpp_config()
  56         VppIpsecSpdItfBinding(self, self.tra_spd,
  57                               self.tra_if).add_vpp_config()
  58         self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
  59         self.tun_spd.add_vpp_config()
  60         VppIpsecSpdItfBinding(self, self.tun_spd,
  61                               self.tun_if).add_vpp_config()
  62
  63         for _, p in self.params.items():
  64             self.config_ah_tra(p)
  65             config_tra_params(p, self.encryption_type)
  66             self.logger.info(self.vapi.ppcli("show ipsec"))
  67         for _, p in self.params.items():
  68             self.config_ah_tun(p)
  69             self.logger.info(self.vapi.ppcli("show ipsec"))
  70         for _, p in self.params.items():
  71             d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
  72             VppIpRoute(self,  p.remote_tun_if_host, p.addr_len,
  73                        [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
  74                                      0xffffffff,
  75                                      proto=d)],
  76                        is_ip6=p.is_ipv6).add_vpp_config()
  77
  78     def tearDown(self):
  79         super(TemplateIpsecAh, self).tearDown()
  80         if not self.vpp_dead:
  81             self.vapi.cli("show hardware")
  82
  83     def config_ah_tun(self, params):
  84         addr_type = params.addr_type
  85         scapy_tun_sa_id = params.scapy_tun_sa_id
  86         scapy_tun_spi = params.scapy_tun_spi
  87         vpp_tun_sa_id = params.vpp_tun_sa_id
  88         vpp_tun_spi = params.vpp_tun_spi
  89         auth_algo_vpp_id = params.auth_algo_vpp_id
  90         auth_key = params.auth_key
  91         crypt_algo_vpp_id = params.crypt_algo_vpp_id
  92         crypt_key = params.crypt_key
  93         remote_tun_if_host = params.remote_tun_if_host
  94         addr_any = params.addr_any
  95         addr_bcast = params.addr_bcast
  96         flags = params.flags
  97         e = VppEnum.vl_api_ipsec_spd_action_t
  98
  99         params.tun_sa_in = VppIpsecSA(self, scapy_tun_sa_id, scapy_tun_spi,
 100                                       auth_algo_vpp_id, auth_key,
 101                                       crypt_algo_vpp_id, crypt_key,
 102                                       self.vpp_ah_protocol,
 103                                       self.tun_if.local_addr[addr_type],
 104                                       self.tun_if.remote_addr[addr_type],
 105                                       flags=flags)
 106         params.tun_sa_in.add_vpp_config()
 107         params.tun_sa_out = VppIpsecSA(self, vpp_tun_sa_id, vpp_tun_spi,
 108                                        auth_algo_vpp_id, auth_key,
 109                                        crypt_algo_vpp_id, crypt_key,
 110                                        self.vpp_ah_protocol,
 111                                        self.tun_if.remote_addr[addr_type],
 112                                        self.tun_if.local_addr[addr_type],
 113                                        flags=flags)
 114         params.tun_sa_out.add_vpp_config()
 115
 116         params.spd_policy_in_any = VppIpsecSpdEntry(self, self.tun_spd,
 117                                                     vpp_tun_sa_id,
 118                                                     addr_any, addr_bcast,
 119                                                     addr_any, addr_bcast,
 120                                                     socket.IPPROTO_AH)
 121         params.spd_policy_in_any.add_vpp_config()
 122         params.spd_policy_out_any = VppIpsecSpdEntry(self, self.tun_spd,
 123                                                      vpp_tun_sa_id,
 124                                                      addr_any, addr_bcast,
 125                                                      addr_any, addr_bcast,
 126                                                      socket.IPPROTO_AH,
 127                                                      is_outbound=0)
 128         params.spd_policy_out_any.add_vpp_config()
 129
 130         VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 131                          remote_tun_if_host,
 132                          remote_tun_if_host,
 133                          self.pg1.remote_addr[addr_type],
 134                          self.pg1.remote_addr[addr_type],
 135                          0, priority=10,
 136                          policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 137                          is_outbound=0).add_vpp_config()
 138         VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 139                          self.pg1.remote_addr[addr_type],
 140                          self.pg1.remote_addr[addr_type],
 141                          remote_tun_if_host,
 142                          remote_tun_if_host,
 143                          0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 144                          priority=10).add_vpp_config()
 145
 146         VppIpsecSpdEntry(self, self.tun_spd, vpp_tun_sa_id,
 147                          remote_tun_if_host,
 148                          remote_tun_if_host,
 149                          self.pg0.local_addr[addr_type],
 150                          self.pg0.local_addr[addr_type],
 151                          0, priority=20,
 152                          policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 153                          is_outbound=0).add_vpp_config()
 154         VppIpsecSpdEntry(self, self.tun_spd, scapy_tun_sa_id,
 155                          self.pg0.local_addr[addr_type],
 156                          self.pg0.local_addr[addr_type],
 157                          remote_tun_if_host,
 158                          remote_tun_if_host,
 159                          0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 160                          priority=20).add_vpp_config()
 161
 162     def config_ah_tra(self, params):
 163         addr_type = params.addr_type
 164         scapy_tra_sa_id = params.scapy_tra_sa_id
 165         scapy_tra_spi = params.scapy_tra_spi
 166         vpp_tra_sa_id = params.vpp_tra_sa_id
 167         vpp_tra_spi = params.vpp_tra_spi
 168         auth_algo_vpp_id = params.auth_algo_vpp_id
 169         auth_key = params.auth_key
 170         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 171         crypt_key = params.crypt_key
 172         addr_any = params.addr_any
 173         addr_bcast = params.addr_bcast
 174         flags = params.flags | (VppEnum.vl_api_ipsec_sad_flags_t.
 175                                 IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
 176         e = VppEnum.vl_api_ipsec_spd_action_t
 177
 178         params.tra_sa_in = VppIpsecSA(self, scapy_tra_sa_id, scapy_tra_spi,
 179                                       auth_algo_vpp_id, auth_key,
 180                                       crypt_algo_vpp_id, crypt_key,
 181                                       self.vpp_ah_protocol,
 182                                       flags=flags)
 183         params.tra_sa_in.add_vpp_config()
 184         params.tra_sa_out = VppIpsecSA(self, vpp_tra_sa_id, vpp_tra_spi,
 185                                        auth_algo_vpp_id, auth_key,
 186                                        crypt_algo_vpp_id, crypt_key,
 187                                        self.vpp_ah_protocol,
 188                                        flags=flags)
 189         params.tra_sa_out.add_vpp_config()
 190
 191         VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 192                          addr_any, addr_bcast,
 193                          addr_any, addr_bcast,
 194                          socket.IPPROTO_AH).add_vpp_config()
 195         VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
 196                          addr_any, addr_bcast,
 197                          addr_any, addr_bcast,
 198                          socket.IPPROTO_AH,
 199                          is_outbound=0).add_vpp_config()
 200
 201         VppIpsecSpdEntry(self, self.tra_spd, vpp_tra_sa_id,
 202                          self.tra_if.local_addr[addr_type],
 203                          self.tra_if.local_addr[addr_type],
 204                          self.tra_if.remote_addr[addr_type],
 205                          self.tra_if.remote_addr[addr_type],
 206                          0, priority=10,
 207                          policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 208                          is_outbound=0).add_vpp_config()
 209         VppIpsecSpdEntry(self, self.tra_spd, scapy_tra_sa_id,
 210                          self.tra_if.local_addr[addr_type],
 211                          self.tra_if.local_addr[addr_type],
 212                          self.tra_if.remote_addr[addr_type],
 213                          self.tra_if.remote_addr[addr_type],
 214                          0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 215                          priority=10).add_vpp_config()
 216
 217
 218 class TestIpsecAh1(TemplateIpsecAh, IpsecTcpTests):
 219     """ Ipsec AH - TCP tests """
 220     pass
 221
 222
 223 class TestIpsecAh2(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
 224     """ Ipsec AH w/ SHA1 """
 225     tra4_encrypt_node_name = "ah4-encrypt"
 226     tra4_decrypt_node_name = "ah4-decrypt"
 227     tra6_encrypt_node_name = "ah6-encrypt"
 228     tra6_decrypt_node_name = "ah6-decrypt"
 229     tun4_encrypt_node_name = "ah4-encrypt"
 230     tun4_decrypt_node_name = "ah4-decrypt"
 231     tun6_encrypt_node_name = "ah6-encrypt"
 232     tun6_decrypt_node_name = "ah6-decrypt"
 233
 234
 235 class TestIpsecAh3(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
 236     """ Ipsec AH w/ SHA1 & ESN """
 237
 238     tra4_encrypt_node_name = "ah4-encrypt"
 239     tra4_decrypt_node_name = "ah4-decrypt"
 240     tra6_encrypt_node_name = "ah6-encrypt"
 241     tra6_decrypt_node_name = "ah6-decrypt"
 242     tun4_encrypt_node_name = "ah4-encrypt"
 243     tun4_decrypt_node_name = "ah4-decrypt"
 244     tun6_encrypt_node_name = "ah6-encrypt"
 245     tun6_decrypt_node_name = "ah6-decrypt"
 246
 247     def setup_params(self):
 248         self.ipv4_params = IPsecIPv4Params()
 249         self.ipv6_params = IPsecIPv6Params()
 250         self.params = {self.ipv4_params.addr_type: self.ipv4_params,
 251                        self.ipv6_params.addr_type: self.ipv6_params}
 252         for _, p in self.params.items():
 253             p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
 254                        IPSEC_API_SAD_FLAG_USE_ESN)
 255
 256
 257 class TestIpsecAh4(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
 258     """ Ipsec AH w/ SHA256 """
 259
 260     tra4_encrypt_node_name = "ah4-encrypt"
 261     tra4_decrypt_node_name = "ah4-decrypt"
 262     tra6_encrypt_node_name = "ah6-encrypt"
 263     tra6_decrypt_node_name = "ah6-decrypt"
 264     tun4_encrypt_node_name = "ah4-encrypt"
 265     tun4_decrypt_node_name = "ah4-decrypt"
 266     tun6_encrypt_node_name = "ah6-encrypt"
 267     tun6_decrypt_node_name = "ah6-decrypt"
 268
 269     def setup_params(self):
 270         self.ipv4_params = IPsecIPv4Params()
 271         self.ipv6_params = IPsecIPv6Params()
 272         self.ipv4_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
 273                                              IPSEC_API_INTEG_ALG_SHA_256_128)
 274         self.ipv4_params.auth_algo = 'SHA2-256-128'  # scapy name
 275         self.ipv6_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
 276                                              IPSEC_API_INTEG_ALG_SHA_256_128)
 277         self.ipv6_params.auth_algo = 'SHA2-256-128'  # scapy name
 278
 279         self.params = {self.ipv4_params.addr_type: self.ipv4_params,
 280                        self.ipv6_params.addr_type: self.ipv6_params}
 281
 282
 283 class TestIpsecAh5(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
 284     """ Ipsec AH w/ SHA384 """
 285
 286     tra4_encrypt_node_name = "ah4-encrypt"
 287     tra4_decrypt_node_name = "ah4-decrypt"
 288     tra6_encrypt_node_name = "ah6-encrypt"
 289     tra6_decrypt_node_name = "ah6-decrypt"
 290     tun4_encrypt_node_name = "ah4-encrypt"
 291     tun4_decrypt_node_name = "ah4-decrypt"
 292     tun6_encrypt_node_name = "ah6-encrypt"
 293     tun6_decrypt_node_name = "ah6-decrypt"
 294
 295     def setup_params(self):
 296         self.ipv4_params = IPsecIPv4Params()
 297         self.ipv6_params = IPsecIPv6Params()
 298         self.ipv4_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
 299                                              IPSEC_API_INTEG_ALG_SHA_384_192)
 300         self.ipv4_params.auth_algo = 'SHA2-384-192'  # scapy name
 301         self.ipv6_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
 302                                              IPSEC_API_INTEG_ALG_SHA_384_192)
 303         self.ipv6_params.auth_algo = 'SHA2-384-192'  # scapy name
 304
 305         self.params = {self.ipv4_params.addr_type: self.ipv4_params,
 306                        self.ipv6_params.addr_type: self.ipv6_params}
 307
 308
 309 class TestIpsecAh6(TemplateIpsecAh, IpsecTra46Tests, IpsecTun46Tests):
 310     """ Ipsec AH w/ SHA512 """
 311
 312     tra4_encrypt_node_name = "ah4-encrypt"
 313     tra4_decrypt_node_name = "ah4-decrypt"
 314     tra6_encrypt_node_name = "ah6-encrypt"
 315     tra6_decrypt_node_name = "ah6-decrypt"
 316     tun4_encrypt_node_name = "ah4-encrypt"
 317     tun4_decrypt_node_name = "ah4-decrypt"
 318     tun6_encrypt_node_name = "ah6-encrypt"
 319     tun6_decrypt_node_name = "ah6-decrypt"
 320
 321     def setup_params(self):
 322         self.ipv4_params = IPsecIPv4Params()
 323         self.ipv6_params = IPsecIPv6Params()
 324         self.ipv4_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
 325                                              IPSEC_API_INTEG_ALG_SHA_512_256)
 326         self.ipv4_params.auth_algo = 'SHA2-512-256'  # scapy name
 327         self.ipv6_params.auth_algo_vpp_id = (VppEnum.vl_api_ipsec_integ_alg_t.
 328                                              IPSEC_API_INTEG_ALG_SHA_512_256)
 329         self.ipv6_params.auth_algo = 'SHA2-512-256'  # scapy name
 330
 331         self.params = {self.ipv4_params.addr_type: self.ipv4_params,
 332                        self.ipv6_params.addr_type: self.ipv6_params}
 333
 334
 335 if __name__ == '__main__':
 336     unittest.main(testRunner=VppTestRunner)