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 from vpp_ipsec import VppIpsecSpd, VppIpsecSpdEntry, VppIpsecSA,\
  10         VppIpsecSpdItfBinding
  11 from vpp_ip_route import VppIpRoute, VppRoutePath
  12 from vpp_ip import DpoProto
  13 from vpp_papi import VppEnum
  14
  15
  16 def config_esp_tun(test, params):
  17     addr_type = params.addr_type
  18     scapy_tun_sa_id = params.scapy_tun_sa_id
  19     scapy_tun_spi = params.scapy_tun_spi
  20     vpp_tun_sa_id = params.vpp_tun_sa_id
  21     vpp_tun_spi = params.vpp_tun_spi
  22     auth_algo_vpp_id = params.auth_algo_vpp_id
  23     auth_key = params.auth_key
  24     crypt_algo_vpp_id = params.crypt_algo_vpp_id
  25     crypt_key = params.crypt_key
  26     remote_tun_if_host = params.remote_tun_if_host
  27     addr_any = params.addr_any
  28     addr_bcast = params.addr_bcast
  29     e = VppEnum.vl_api_ipsec_spd_action_t
  30
  31     params.tun_sa_in = VppIpsecSA(test, scapy_tun_sa_id, scapy_tun_spi,
  32                                   auth_algo_vpp_id, auth_key,
  33                                   crypt_algo_vpp_id, crypt_key,
  34                                   test.vpp_esp_protocol,
  35                                   test.tun_if.local_addr[addr_type],
  36                                   test.tun_if.remote_addr[addr_type])
  37     params.tun_sa_in.add_vpp_config()
  38     params.tun_sa_out = VppIpsecSA(test, vpp_tun_sa_id, vpp_tun_spi,
  39                                    auth_algo_vpp_id, auth_key,
  40                                    crypt_algo_vpp_id, crypt_key,
  41                                    test.vpp_esp_protocol,
  42                                    test.tun_if.remote_addr[addr_type],
  43                                    test.tun_if.local_addr[addr_type])
  44     params.tun_sa_out.add_vpp_config()
  45
  46     params.spd_policy_in_any = VppIpsecSpdEntry(test, test.tun_spd,
  47                                                 scapy_tun_sa_id,
  48                                                 addr_any, addr_bcast,
  49                                                 addr_any, addr_bcast,
  50                                                 socket.IPPROTO_ESP)
  51     params.spd_policy_in_any.add_vpp_config()
  52     params.spd_policy_out_any = VppIpsecSpdEntry(test, test.tun_spd,
  53                                                  scapy_tun_sa_id,
  54                                                  addr_any, addr_bcast,
  55                                                  addr_any, addr_bcast,
  56                                                  socket.IPPROTO_ESP,
  57                                                  is_outbound=0)
  58     params.spd_policy_out_any.add_vpp_config()
  59
  60     VppIpsecSpdEntry(test, test.tun_spd, vpp_tun_sa_id,
  61                      remote_tun_if_host, remote_tun_if_host,
  62                      test.pg1.remote_addr[addr_type],
  63                      test.pg1.remote_addr[addr_type],
  64                      0,
  65                      priority=10,
  66                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
  67                      is_outbound=0).add_vpp_config()
  68     VppIpsecSpdEntry(test, test.tun_spd, scapy_tun_sa_id,
  69                      test.pg1.remote_addr[addr_type],
  70                      test.pg1.remote_addr[addr_type],
  71                      remote_tun_if_host, remote_tun_if_host,
  72                      0,
  73                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
  74                      priority=10).add_vpp_config()
  75
  76     VppIpsecSpdEntry(test, test.tun_spd, vpp_tun_sa_id,
  77                      remote_tun_if_host, remote_tun_if_host,
  78                      test.pg0.local_addr[addr_type],
  79                      test.pg0.local_addr[addr_type],
  80                      0,
  81                      priority=20,
  82                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
  83                      is_outbound=0).add_vpp_config()
  84     VppIpsecSpdEntry(test, test.tun_spd, scapy_tun_sa_id,
  85                      test.pg0.local_addr[addr_type],
  86                      test.pg0.local_addr[addr_type],
  87                      remote_tun_if_host, remote_tun_if_host,
  88                      0,
  89                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
  90                      priority=20).add_vpp_config()
  91
  92
  93 def config_esp_tra(test, params):
  94     addr_type = params.addr_type
  95     scapy_tra_sa_id = params.scapy_tra_sa_id
  96     scapy_tra_spi = params.scapy_tra_spi
  97     vpp_tra_sa_id = params.vpp_tra_sa_id
  98     vpp_tra_spi = params.vpp_tra_spi
  99     auth_algo_vpp_id = params.auth_algo_vpp_id
 100     auth_key = params.auth_key
 101     crypt_algo_vpp_id = params.crypt_algo_vpp_id
 102     crypt_key = params.crypt_key
 103     addr_any = params.addr_any
 104     addr_bcast = params.addr_bcast
 105     flags = (VppEnum.vl_api_ipsec_sad_flags_t.
 106              IPSEC_API_SAD_FLAG_USE_ANTI_REPLAY)
 107     e = VppEnum.vl_api_ipsec_spd_action_t
 108     flags = params.flags | flags
 109
 110     params.tra_sa_in = VppIpsecSA(test, scapy_tra_sa_id, scapy_tra_spi,
 111                                   auth_algo_vpp_id, auth_key,
 112                                   crypt_algo_vpp_id, crypt_key,
 113                                   test.vpp_esp_protocol,
 114                                   flags=flags)
 115     params.tra_sa_in.add_vpp_config()
 116     params.tra_sa_out = VppIpsecSA(test, vpp_tra_sa_id, vpp_tra_spi,
 117                                    auth_algo_vpp_id, auth_key,
 118                                    crypt_algo_vpp_id, crypt_key,
 119                                    test.vpp_esp_protocol,
 120                                    flags=flags)
 121     params.tra_sa_out.add_vpp_config()
 122
 123     VppIpsecSpdEntry(test, test.tra_spd, vpp_tra_sa_id,
 124                      addr_any, addr_bcast,
 125                      addr_any, addr_bcast,
 126                      socket.IPPROTO_ESP).add_vpp_config()
 127     VppIpsecSpdEntry(test, test.tra_spd, vpp_tra_sa_id,
 128                      addr_any, addr_bcast,
 129                      addr_any, addr_bcast,
 130                      socket.IPPROTO_ESP,
 131                      is_outbound=0).add_vpp_config()
 132
 133     VppIpsecSpdEntry(test, test.tra_spd, vpp_tra_sa_id,
 134                      test.tra_if.local_addr[addr_type],
 135                      test.tra_if.local_addr[addr_type],
 136                      test.tra_if.remote_addr[addr_type],
 137                      test.tra_if.remote_addr[addr_type],
 138                      0, priority=10,
 139                      policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 140                      is_outbound=0).add_vpp_config()
 141     VppIpsecSpdEntry(test, test.tra_spd, scapy_tra_sa_id,
 142                      test.tra_if.local_addr[addr_type],
 143                      test.tra_if.local_addr[addr_type],
 144                      test.tra_if.remote_addr[addr_type],
 145                      test.tra_if.remote_addr[addr_type],
 146                      0, policy=e.IPSEC_API_SPD_ACTION_PROTECT,
 147                      priority=10).add_vpp_config()
 148
 149
 150 class TemplateIpsecEsp(TemplateIpsec):
 151     """
 152     Basic test for ipsec esp sanity - tunnel and transport modes.
 153
 154     Below 4 cases are covered as part of this test
 155     1) ipsec esp v4 transport basic test  - IPv4 Transport mode
 156         scenario using HMAC-SHA1-96 integrity algo
 157     2) ipsec esp v4 transport burst test
 158         Above test for 257 pkts
 159     3) ipsec esp 4o4 tunnel basic test    - IPv4 Tunnel mode
 160         scenario using HMAC-SHA1-96 integrity algo
 161     4) ipsec esp 4o4 tunnel burst test
 162         Above test for 257 pkts
 163
 164     TRANSPORT MODE:
 165
 166      ---   encrypt   ---
 167     |pg2| <-------> |VPP|
 168      ---   decrypt   ---
 169
 170     TUNNEL MODE:
 171
 172      ---   encrypt   ---   plain   ---
 173     |pg0| <-------  |VPP| <------ |pg1|
 174      ---             ---           ---
 175
 176      ---   decrypt   ---   plain   ---
 177     |pg0| ------->  |VPP| ------> |pg1|
 178      ---             ---           ---
 179     """
 180
 181     @classmethod
 182     def setUpClass(cls):
 183         super(TemplateIpsecEsp, cls).setUpClass()
 184
 185     @classmethod
 186     def tearDownClass(cls):
 187         super(TemplateIpsecEsp, cls).tearDownClass()
 188
 189     def setUp(self):
 190         super(TemplateIpsecEsp, self).setUp()
 191         self.encryption_type = ESP
 192         self.tun_if = self.pg0
 193         self.tra_if = self.pg2
 194         self.logger.info(self.vapi.ppcli("show int addr"))
 195
 196         self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
 197         self.tra_spd.add_vpp_config()
 198         VppIpsecSpdItfBinding(self, self.tra_spd,
 199                               self.tra_if).add_vpp_config()
 200
 201         for _, p in self.params.items():
 202             config_esp_tra(self, p)
 203             config_tra_params(p, self.encryption_type)
 204         self.logger.info(self.vapi.ppcli("show ipsec"))
 205
 206         self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
 207         self.tun_spd.add_vpp_config()
 208         VppIpsecSpdItfBinding(self, self.tun_spd,
 209                               self.tun_if).add_vpp_config()
 210
 211         for _, p in self.params.items():
 212             config_esp_tun(self, p)
 213         self.logger.info(self.vapi.ppcli("show ipsec"))
 214
 215         for _, p in self.params.items():
 216             d = DpoProto.DPO_PROTO_IP6 if p.is_ipv6 else DpoProto.DPO_PROTO_IP4
 217             VppIpRoute(self,  p.remote_tun_if_host, p.addr_len,
 218                        [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
 219                                      0xffffffff,
 220                                      proto=d)],
 221                        is_ip6=p.is_ipv6).add_vpp_config()
 222
 223     def tearDown(self):
 224         super(TemplateIpsecEsp, self).tearDown()
 225         if not self.vpp_dead:
 226             self.vapi.cli("show hardware")
 227
 228
 229 class TestIpsecEsp1(TemplateIpsecEsp, IpsecTra46Tests, IpsecTun46Tests):
 230     """ Ipsec ESP - TUN & TRA tests """
 231     tra4_encrypt_node_name = "esp4-encrypt"
 232     tra4_decrypt_node_name = "esp4-decrypt"
 233     tra6_encrypt_node_name = "esp6-encrypt"
 234     tra6_decrypt_node_name = "esp6-decrypt"
 235     tun4_encrypt_node_name = "esp4-encrypt"
 236     tun4_decrypt_node_name = "esp4-decrypt"
 237     tun6_encrypt_node_name = "esp6-encrypt"
 238     tun6_decrypt_node_name = "esp6-decrypt"
 239
 240
 241 class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests):
 242     """ Ipsec ESP - TCP tests """
 243     pass
 244
 245
 246 class TemplateIpsecEspUdp(TemplateIpsec):
 247     """
 248     UDP encapped ESP
 249     """
 250
 251     @classmethod
 252     def setUpClass(cls):
 253         super(TemplateIpsecEspUdp, cls).setUpClass()
 254
 255     @classmethod
 256     def tearDownClass(cls):
 257         super(TemplateIpsecEspUdp, cls).tearDownClass()
 258
 259     def setUp(self):
 260         super(TemplateIpsecEspUdp, self).setUp()
 261         self.encryption_type = ESP
 262         self.tun_if = self.pg0
 263         self.tra_if = self.pg2
 264         self.logger.info(self.vapi.ppcli("show int addr"))
 265
 266         p = self.ipv4_params
 267         p.flags = (VppEnum.vl_api_ipsec_sad_flags_t.
 268                    IPSEC_API_SAD_FLAG_UDP_ENCAP)
 269         p.nat_header = UDP(sport=5454, dport=4500)
 270
 271         self.tra_spd = VppIpsecSpd(self, self.tra_spd_id)
 272         self.tra_spd.add_vpp_config()
 273         VppIpsecSpdItfBinding(self, self.tra_spd,
 274                               self.tra_if).add_vpp_config()
 275
 276         config_esp_tra(self, p)
 277         config_tra_params(p, self.encryption_type)
 278
 279         self.tun_spd = VppIpsecSpd(self, self.tun_spd_id)
 280         self.tun_spd.add_vpp_config()
 281         VppIpsecSpdItfBinding(self, self.tun_spd,
 282                               self.tun_if).add_vpp_config()
 283
 284         config_esp_tun(self, p)
 285         self.logger.info(self.vapi.ppcli("show ipsec"))
 286
 287         d = DpoProto.DPO_PROTO_IP4
 288         VppIpRoute(self,  p.remote_tun_if_host, p.addr_len,
 289                    [VppRoutePath(self.tun_if.remote_addr[p.addr_type],
 290                                  0xffffffff,
 291                                  proto=d)]).add_vpp_config()
 292
 293     def tearDown(self):
 294         super(TemplateIpsecEspUdp, self).tearDown()
 295         if not self.vpp_dead:
 296             self.vapi.cli("show hardware")
 297
 298
 299 class TestIpsecEspUdp(TemplateIpsecEspUdp, IpsecTra4Tests, IpsecTun4Tests):
 300     """ Ipsec NAT-T ESP UDP tests """
 301     tra4_encrypt_node_name = "esp4-encrypt"
 302     tra4_decrypt_node_name = "esp4-decrypt"
 303     tun4_encrypt_node_name = "esp4-encrypt"
 304     tun4_decrypt_node_name = "esp4-decrypt"
 305     pass
 306
 307
 308 if __name__ == '__main__':
 309     unittest.main(testRunner=VppTestRunner)