test/test_ipsec_esp.py

   1 import socket
   2 import unittest
   3 from scapy.layers.ipsec import ESP
   4
   5 from framework import VppTestRunner
   6 from template_ipsec import IpsecTraTests, IpsecTunTests
   7 from template_ipsec import TemplateIpsec, IpsecTcpTests
   8
   9
  10 class TemplateIpsecEsp(TemplateIpsec):
  11     """
  12     Basic test for ipsec esp sanity - tunnel and transport modes.
  13
  14     Below 4 cases are covered as part of this test
  15     1) ipsec esp v4 transport basic test  - IPv4 Transport mode
  16         scenario using HMAC-SHA1-96 intergrity algo
  17     2) ipsec esp v4 transport burst test
  18         Above test for 257 pkts
  19     3) ipsec esp 4o4 tunnel basic test    - IPv4 Tunnel mode
  20         scenario using HMAC-SHA1-96 intergrity algo
  21     4) ipsec esp 4o4 tunnel burst test
  22         Above test for 257 pkts
  23
  24     TRANSPORT MODE:
  25
  26      ---   encrypt   ---
  27     |pg2| <-------> |VPP|
  28      ---   decrypt   ---
  29
  30     TUNNEL MODE:
  31
  32      ---   encrypt   ---   plain   ---
  33     |pg0| <-------  |VPP| <------ |pg1|
  34      ---             ---           ---
  35
  36      ---   decrypt   ---   plain   ---
  37     |pg0| ------->  |VPP| ------> |pg1|
  38      ---             ---           ---
  39     """
  40
  41     def setUp(self):
  42         super(TemplateIpsecEsp, self).setUp()
  43         self.encryption_type = ESP
  44         self.tun_if = self.pg0
  45         self.tra_if = self.pg2
  46         self.logger.info(self.vapi.ppcli("show int addr"))
  47         self.vapi.ipsec_spd_add_del(self.tra_spd_id)
  48         self.vapi.ipsec_interface_add_del_spd(self.tra_spd_id,
  49                                               self.tra_if.sw_if_index)
  50         for _, p in self.params.items():
  51             self.config_esp_tra(p)
  52             self.configure_sa_tra(p)
  53         self.logger.info(self.vapi.ppcli("show ipsec"))
  54         self.vapi.ipsec_spd_add_del(self.tun_spd_id)
  55         self.vapi.ipsec_interface_add_del_spd(self.tun_spd_id,
  56                                               self.tun_if.sw_if_index)
  57         for _, p in self.params.items():
  58             self.config_esp_tun(p)
  59         self.logger.info(self.vapi.ppcli("show ipsec"))
  60         for _, p in self.params.items():
  61             src = socket.inet_pton(p.addr_type, p.remote_tun_if_host)
  62             self.vapi.ip_add_del_route(
  63                 src, p.addr_len, self.tun_if.remote_addr_n[p.addr_type],
  64                 is_ipv6=p.is_ipv6)
  65
  66     def tearDown(self):
  67         for _, p in self.params.items():
  68             self.unconfig_esp_tun(p)
  69         for _, p in self.params.items():
  70             self.unconfig_esp_tra(p)
  71
  72         self.vapi.ipsec_interface_add_del_spd(self.tun_spd_id,
  73                                               self.tun_if.sw_if_index,
  74                                               is_add=0)
  75         self.vapi.ipsec_spd_add_del(self.tun_spd_id, is_add=0)
  76         self.vapi.ipsec_interface_add_del_spd(self.tra_spd_id,
  77                                               self.tra_if.sw_if_index,
  78                                               is_add=0)
  79         self.vapi.ipsec_spd_add_del(self.tra_spd_id,
  80                                     is_add=0)
  81         for _, p in self.params.items():
  82             src = socket.inet_pton(p.addr_type, p.remote_tun_if_host)
  83             self.vapi.ip_add_del_route(
  84                 src, p.addr_len, self.tun_if.remote_addr_n[p.addr_type],
  85                 is_ipv6=p.is_ipv6, is_add=0)
  86
  87         super(TemplateIpsecEsp, self).tearDown()
  88         if not self.vpp_dead:
  89             self.vapi.cli("show hardware")
  90
  91     def config_esp_tun(self, params):
  92         addr_type = params.addr_type
  93         is_ipv6 = params.is_ipv6
  94         scapy_tun_sa_id = params.scapy_tun_sa_id
  95         scapy_tun_spi = params.scapy_tun_spi
  96         vpp_tun_sa_id = params.vpp_tun_sa_id
  97         vpp_tun_spi = params.vpp_tun_spi
  98         auth_algo_vpp_id = params.auth_algo_vpp_id
  99         auth_key = params.auth_key
 100         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 101         crypt_key = params.crypt_key
 102         remote_tun_if_host = params.remote_tun_if_host
 103         addr_any = params.addr_any
 104         addr_bcast = params.addr_bcast
 105         self.vapi.ipsec_sad_add_del_entry(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_n[addr_type],
 110                                           self.tun_if.remote_addr_n[addr_type],
 111                                           is_tunnel=1, is_tunnel_ipv6=is_ipv6)
 112         self.vapi.ipsec_sad_add_del_entry(vpp_tun_sa_id, vpp_tun_spi,
 113                                           auth_algo_vpp_id, auth_key,
 114                                           crypt_algo_vpp_id, crypt_key,
 115                                           self.vpp_esp_protocol,
 116                                           self.tun_if.remote_addr_n[addr_type],
 117                                           self.tun_if.local_addr_n[addr_type],
 118                                           is_tunnel=1, is_tunnel_ipv6=is_ipv6)
 119         l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any)
 120         l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast)
 121         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id,
 122                                           l_startaddr, l_stopaddr, r_startaddr,
 123                                           r_stopaddr, is_ipv6=is_ipv6,
 124                                           protocol=socket.IPPROTO_ESP)
 125         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id,
 126                                           l_startaddr, l_stopaddr, r_startaddr,
 127                                           r_stopaddr, is_outbound=0,
 128                                           protocol=socket.IPPROTO_ESP,
 129                                           is_ipv6=is_ipv6)
 130         l_startaddr = l_stopaddr = socket.inet_pton(addr_type,
 131                                                     remote_tun_if_host)
 132         r_startaddr = r_stopaddr = self.pg1.remote_addr_n[addr_type]
 133         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, vpp_tun_sa_id,
 134                                           l_startaddr, l_stopaddr, r_startaddr,
 135                                           r_stopaddr, priority=10, policy=3,
 136                                           is_ipv6=is_ipv6, is_outbound=0)
 137         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id,
 138                                           r_startaddr, r_stopaddr, l_startaddr,
 139                                           l_stopaddr, priority=10, policy=3,
 140                                           is_ipv6=is_ipv6)
 141         l_startaddr = l_stopaddr = socket.inet_pton(addr_type,
 142                                                     remote_tun_if_host)
 143         r_startaddr = r_stopaddr = self.pg0.local_addr_n[addr_type]
 144         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, vpp_tun_sa_id,
 145                                           l_startaddr, l_stopaddr, r_startaddr,
 146                                           r_stopaddr, priority=20, policy=3,
 147                                           is_outbound=0, is_ipv6=is_ipv6)
 148         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id,
 149                                           r_startaddr, r_stopaddr, l_startaddr,
 150                                           l_stopaddr, priority=20, policy=3,
 151                                           is_ipv6=is_ipv6)
 152
 153     def unconfig_esp_tun(self, params):
 154         addr_type = params.addr_type
 155         is_ipv6 = params.is_ipv6
 156         scapy_tun_sa_id = params.scapy_tun_sa_id
 157         scapy_tun_spi = params.scapy_tun_spi
 158         vpp_tun_sa_id = params.vpp_tun_sa_id
 159         vpp_tun_spi = params.vpp_tun_spi
 160         auth_algo_vpp_id = params.auth_algo_vpp_id
 161         auth_key = params.auth_key
 162         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 163         crypt_key = params.crypt_key
 164         remote_tun_if_host = params.remote_tun_if_host
 165         addr_any = params.addr_any
 166         addr_bcast = params.addr_bcast
 167         l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any)
 168         l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast)
 169         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id,
 170                                           l_startaddr, l_stopaddr, r_startaddr,
 171                                           r_stopaddr, is_ipv6=is_ipv6,
 172                                           protocol=socket.IPPROTO_ESP,
 173                                           is_add=0)
 174         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id,
 175                                           l_startaddr, l_stopaddr, r_startaddr,
 176                                           r_stopaddr, is_outbound=0,
 177                                           protocol=socket.IPPROTO_ESP,
 178                                           is_ipv6=is_ipv6,
 179                                           is_add=0)
 180         l_startaddr = l_stopaddr = socket.inet_pton(addr_type,
 181                                                     remote_tun_if_host)
 182         r_startaddr = r_stopaddr = self.pg1.remote_addr_n[addr_type]
 183         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, vpp_tun_sa_id,
 184                                           l_startaddr, l_stopaddr, r_startaddr,
 185                                           r_stopaddr, priority=10, policy=3,
 186                                           is_ipv6=is_ipv6, is_outbound=0,
 187                                           is_add=0)
 188         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id,
 189                                           r_startaddr, r_stopaddr, l_startaddr,
 190                                           l_stopaddr, priority=10, policy=3,
 191                                           is_ipv6=is_ipv6, is_add=0)
 192         l_startaddr = l_stopaddr = socket.inet_pton(addr_type,
 193                                                     remote_tun_if_host)
 194         r_startaddr = r_stopaddr = self.pg0.local_addr_n[addr_type]
 195         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, vpp_tun_sa_id,
 196                                           l_startaddr, l_stopaddr, r_startaddr,
 197                                           r_stopaddr, priority=20, policy=3,
 198                                           is_outbound=0, is_ipv6=is_ipv6,
 199                                           is_add=0)
 200         self.vapi.ipsec_spd_add_del_entry(self.tun_spd_id, scapy_tun_sa_id,
 201                                           r_startaddr, r_stopaddr, l_startaddr,
 202                                           l_stopaddr, priority=20, policy=3,
 203                                           is_ipv6=is_ipv6,
 204                                           is_add=0)
 205         self.vapi.ipsec_sad_add_del_entry(scapy_tun_sa_id, scapy_tun_spi,
 206                                           auth_algo_vpp_id, auth_key,
 207                                           crypt_algo_vpp_id, crypt_key,
 208                                           self.vpp_esp_protocol,
 209                                           self.tun_if.local_addr_n[addr_type],
 210                                           self.tun_if.remote_addr_n[addr_type],
 211                                           is_tunnel=1, is_tunnel_ipv6=is_ipv6,
 212                                           is_add=0)
 213         self.vapi.ipsec_sad_add_del_entry(vpp_tun_sa_id, vpp_tun_spi,
 214                                           auth_algo_vpp_id, auth_key,
 215                                           crypt_algo_vpp_id, crypt_key,
 216                                           self.vpp_esp_protocol,
 217                                           self.tun_if.remote_addr_n[addr_type],
 218                                           self.tun_if.local_addr_n[addr_type],
 219                                           is_tunnel=1, is_tunnel_ipv6=is_ipv6,
 220                                           is_add=0)
 221
 222     def config_esp_tra(self, params):
 223         addr_type = params.addr_type
 224         is_ipv6 = params.is_ipv6
 225         scapy_tra_sa_id = params.scapy_tra_sa_id
 226         scapy_tra_spi = params.scapy_tra_spi
 227         vpp_tra_sa_id = params.vpp_tra_sa_id
 228         vpp_tra_spi = params.vpp_tra_spi
 229         auth_algo_vpp_id = params.auth_algo_vpp_id
 230         auth_key = params.auth_key
 231         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 232         crypt_key = params.crypt_key
 233         addr_any = params.addr_any
 234         addr_bcast = params.addr_bcast
 235         self.vapi.ipsec_sad_add_del_entry(scapy_tra_sa_id, scapy_tra_spi,
 236                                           auth_algo_vpp_id, auth_key,
 237                                           crypt_algo_vpp_id, crypt_key,
 238                                           self.vpp_esp_protocol, is_tunnel=0,
 239                                           use_anti_replay=1)
 240         self.vapi.ipsec_sad_add_del_entry(vpp_tra_sa_id, vpp_tra_spi,
 241                                           auth_algo_vpp_id, auth_key,
 242                                           crypt_algo_vpp_id, crypt_key,
 243                                           self.vpp_esp_protocol, is_tunnel=0,
 244                                           use_anti_replay=1)
 245         l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any)
 246         l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast)
 247         self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id,
 248                                           l_startaddr, l_stopaddr, r_startaddr,
 249                                           r_stopaddr, is_ipv6=is_ipv6,
 250                                           protocol=socket.IPPROTO_ESP)
 251         self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id,
 252                                           l_startaddr, l_stopaddr, r_startaddr,
 253                                           r_stopaddr, is_outbound=0,
 254                                           is_ipv6=is_ipv6,
 255                                           protocol=socket.IPPROTO_ESP)
 256         l_startaddr = l_stopaddr = self.tra_if.local_addr_n[addr_type]
 257         r_startaddr = r_stopaddr = self.tra_if.remote_addr_n[addr_type]
 258         self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id,
 259                                           l_startaddr, l_stopaddr, r_startaddr,
 260                                           r_stopaddr, priority=10, policy=3,
 261                                           is_outbound=0, is_ipv6=is_ipv6)
 262         self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, scapy_tra_sa_id,
 263                                           l_startaddr, l_stopaddr, r_startaddr,
 264                                           r_stopaddr, priority=10, policy=3,
 265                                           is_ipv6=is_ipv6)
 266
 267     def unconfig_esp_tra(self, params):
 268         addr_type = params.addr_type
 269         is_ipv6 = params.is_ipv6
 270         scapy_tra_sa_id = params.scapy_tra_sa_id
 271         scapy_tra_spi = params.scapy_tra_spi
 272         vpp_tra_sa_id = params.vpp_tra_sa_id
 273         vpp_tra_spi = params.vpp_tra_spi
 274         auth_algo_vpp_id = params.auth_algo_vpp_id
 275         auth_key = params.auth_key
 276         crypt_algo_vpp_id = params.crypt_algo_vpp_id
 277         crypt_key = params.crypt_key
 278         addr_any = params.addr_any
 279         addr_bcast = params.addr_bcast
 280         l_startaddr = r_startaddr = socket.inet_pton(addr_type, addr_any)
 281         l_stopaddr = r_stopaddr = socket.inet_pton(addr_type, addr_bcast)
 282         self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id,
 283                                           l_startaddr, l_stopaddr, r_startaddr,
 284                                           r_stopaddr, is_ipv6=is_ipv6,
 285                                           protocol=socket.IPPROTO_ESP,
 286                                           is_add=0)
 287         self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id,
 288                                           l_startaddr, l_stopaddr, r_startaddr,
 289                                           r_stopaddr, is_outbound=0,
 290                                           is_ipv6=is_ipv6,
 291                                           protocol=socket.IPPROTO_ESP,
 292                                           is_add=0)
 293         l_startaddr = l_stopaddr = self.tra_if.local_addr_n[addr_type]
 294         r_startaddr = r_stopaddr = self.tra_if.remote_addr_n[addr_type]
 295         self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, vpp_tra_sa_id,
 296                                           l_startaddr, l_stopaddr, r_startaddr,
 297                                           r_stopaddr, priority=10, policy=3,
 298                                           is_outbound=0, is_ipv6=is_ipv6,
 299                                           is_add=0)
 300         self.vapi.ipsec_spd_add_del_entry(self.tra_spd_id, scapy_tra_sa_id,
 301                                           l_startaddr, l_stopaddr, r_startaddr,
 302                                           r_stopaddr, priority=10, policy=3,
 303                                           is_ipv6=is_ipv6,
 304                                           is_add=0)
 305         self.vapi.ipsec_sad_add_del_entry(scapy_tra_sa_id, scapy_tra_spi,
 306                                           auth_algo_vpp_id, auth_key,
 307                                           crypt_algo_vpp_id, crypt_key,
 308                                           self.vpp_esp_protocol, is_tunnel=0,
 309                                           use_anti_replay=1,
 310                                           is_add=0)
 311         self.vapi.ipsec_sad_add_del_entry(vpp_tra_sa_id, vpp_tra_spi,
 312                                           auth_algo_vpp_id, auth_key,
 313                                           crypt_algo_vpp_id, crypt_key,
 314                                           self.vpp_esp_protocol, is_tunnel=0,
 315                                           use_anti_replay=1,
 316                                           is_add=0)
 317
 318
 319 class TestIpsecEsp1(TemplateIpsecEsp, IpsecTraTests, IpsecTunTests):
 320     """ Ipsec ESP - TUN & TRA tests """
 321     tra4_encrypt_node_name = "esp4-encrypt"
 322     tra4_decrypt_node_name = "esp4-decrypt"
 323     tra6_encrypt_node_name = "esp6-encrypt"
 324     tra6_decrypt_node_name = "esp6-decrypt"
 325     tun4_encrypt_node_name = "esp4-encrypt"
 326     tun4_decrypt_node_name = "esp4-decrypt"
 327     tun6_encrypt_node_name = "esp6-encrypt"
 328     tun6_decrypt_node_name = "esp6-decrypt"
 329
 330
 331 class TestIpsecEsp2(TemplateIpsecEsp, IpsecTcpTests):
 332     """ Ipsec ESP - TCP tests """
 333     pass
 334
 335
 336 if __name__ == '__main__':
 337     unittest.main(testRunner=VppTestRunner)