test/vpp_ipsec.py

   1 from vpp_object import VppObject
   2 from ipaddress import ip_address
   3 from vpp_papi import VppEnum
   4 from vpp_interface import VppInterface
   5
   6 try:
   7     text_type = unicode
   8 except NameError:
   9     text_type = str
  10
  11
  12 def mk_counter():
  13     return {'packets': 0, 'bytes': 0}
  14
  15
  16 class VppIpsecSpd(VppObject):
  17     """
  18     VPP SPD DB
  19     """
  20
  21     def __init__(self, test, id):
  22         self.test = test
  23         self.id = id
  24
  25     def add_vpp_config(self):
  26         self.test.vapi.ipsec_spd_add_del(self.id)
  27         self.test.registry.register(self, self.test.logger)
  28
  29     def remove_vpp_config(self):
  30         self.test.vapi.ipsec_spd_add_del(self.id, is_add=0)
  31
  32     def object_id(self):
  33         return "ipsec-spd-%d" % self.id
  34
  35     def query_vpp_config(self):
  36         spds = self.test.vapi.ipsec_spds_dump()
  37         for spd in spds:
  38             if spd.spd_id == self.id:
  39                 return True
  40         return False
  41
  42
  43 class VppIpsecSpdItfBinding(VppObject):
  44     """
  45     VPP SPD DB to interface binding
  46     (i.e. this SPD is used on this interface)
  47     """
  48
  49     def __init__(self, test, spd, itf):
  50         self.test = test
  51         self.spd = spd
  52         self.itf = itf
  53
  54     def add_vpp_config(self):
  55         self.test.vapi.ipsec_interface_add_del_spd(self.spd.id,
  56                                                    self.itf.sw_if_index)
  57         self.test.registry.register(self, self.test.logger)
  58
  59     def remove_vpp_config(self):
  60         self.test.vapi.ipsec_interface_add_del_spd(self.spd.id,
  61                                                    self.itf.sw_if_index,
  62                                                    is_add=0)
  63
  64     def object_id(self):
  65         return "bind-%s-to-%s" % (self.spd.id, self.itf)
  66
  67     def query_vpp_config(self):
  68         bs = self.test.vapi.ipsec_spd_interface_dump()
  69         for b in bs:
  70             if b.sw_if_index == self.itf.sw_if_index:
  71                 return True
  72         return False
  73
  74
  75 class VppIpsecSpdEntry(VppObject):
  76     """
  77     VPP SPD DB Entry
  78     """
  79
  80     def __init__(self, test, spd, sa_id,
  81                  local_start, local_stop,
  82                  remote_start, remote_stop,
  83                  proto,
  84                  priority=100,
  85                  policy=None,
  86                  is_outbound=1,
  87                  remote_port_start=0,
  88                  remote_port_stop=65535,
  89                  local_port_start=0,
  90                  local_port_stop=65535):
  91         self.test = test
  92         self.spd = spd
  93         self.sa_id = sa_id
  94         self.local_start = ip_address(text_type(local_start))
  95         self.local_stop = ip_address(text_type(local_stop))
  96         self.remote_start = ip_address(text_type(remote_start))
  97         self.remote_stop = ip_address(text_type(remote_stop))
  98         self.proto = proto
  99         self.is_outbound = is_outbound
 100         self.priority = priority
 101         if not policy:
 102             self.policy = (VppEnum.vl_api_ipsec_spd_action_t.
 103                            IPSEC_API_SPD_ACTION_BYPASS)
 104         else:
 105             self.policy = policy
 106         self.is_ipv6 = (0 if self.local_start.version == 4 else 1)
 107         self.local_port_start = local_port_start
 108         self.local_port_stop = local_port_stop
 109         self.remote_port_start = remote_port_start
 110         self.remote_port_stop = remote_port_stop
 111
 112     def add_vpp_config(self):
 113         rv = self.test.vapi.ipsec_spd_entry_add_del(
 114             self.spd.id,
 115             self.sa_id,
 116             self.local_start,
 117             self.local_stop,
 118             self.remote_start,
 119             self.remote_stop,
 120             protocol=self.proto,
 121             is_ipv6=self.is_ipv6,
 122             is_outbound=self.is_outbound,
 123             priority=self.priority,
 124             policy=self.policy,
 125             local_port_start=self.local_port_start,
 126             local_port_stop=self.local_port_stop,
 127             remote_port_start=self.remote_port_start,
 128             remote_port_stop=self.remote_port_stop)
 129         self.stat_index = rv.stat_index
 130         self.test.registry.register(self, self.test.logger)
 131         return self
 132
 133     def remove_vpp_config(self):
 134         self.test.vapi.ipsec_spd_entry_add_del(
 135             self.spd.id,
 136             self.sa_id,
 137             self.local_start,
 138             self.local_stop,
 139             self.remote_start,
 140             self.remote_stop,
 141             protocol=self.proto,
 142             is_ipv6=self.is_ipv6,
 143             is_outbound=self.is_outbound,
 144             priority=self.priority,
 145             policy=self.policy,
 146             local_port_start=self.local_port_start,
 147             local_port_stop=self.local_port_stop,
 148             remote_port_start=self.remote_port_start,
 149             remote_port_stop=self.remote_port_stop,
 150             is_add=0)
 151
 152     def object_id(self):
 153         return "spd-entry-%d-%d-%d-%d-%d-%d" % (self.spd.id,
 154                                                 self.priority,
 155                                                 self.policy,
 156                                                 self.is_outbound,
 157                                                 self.is_ipv6,
 158                                                 self.remote_port_start)
 159
 160     def query_vpp_config(self):
 161         ss = self.test.vapi.ipsec_spd_dump(self.spd.id)
 162         for s in ss:
 163             if s.entry.sa_id == self.sa_id and \
 164                s.entry.is_outbound == self.is_outbound and \
 165                s.entry.priority == self.priority and \
 166                s.entry.policy == self.policy and \
 167                s.entry.remote_address_start == self.remote_start and \
 168                s.entry.remote_port_start == self.remote_port_start:
 169                 return True
 170         return False
 171
 172     def get_stats(self, worker=None):
 173         c = self.test.statistics.get_counter("/net/ipsec/policy")
 174         if worker is None:
 175             total = mk_counter()
 176             for t in c:
 177                 total['packets'] += t[self.stat_index]['packets']
 178             return total
 179         else:
 180             # +1 to skip main thread
 181             return c[worker+1][self.stat_index]
 182
 183
 184 class VppIpsecSA(VppObject):
 185     """
 186     VPP SAD Entry
 187     """
 188
 189     DEFAULT_UDP_PORT = 4500
 190
 191     def __init__(self, test, id, spi,
 192                  integ_alg, integ_key,
 193                  crypto_alg, crypto_key,
 194                  proto,
 195                  tun_src=None, tun_dst=None,
 196                  flags=None, salt=0, tun_flags=None,
 197                  dscp=None,
 198                  udp_src=None, udp_dst=None, hop_limit=None):
 199         e = VppEnum.vl_api_ipsec_sad_flags_t
 200         self.test = test
 201         self.id = id
 202         self.spi = spi
 203         self.integ_alg = integ_alg
 204         self.integ_key = integ_key
 205         self.crypto_alg = crypto_alg
 206         self.crypto_key = crypto_key
 207         self.proto = proto
 208         self.salt = salt
 209
 210         self.table_id = 0
 211         self.tun_src = tun_src
 212         self.tun_dst = tun_dst
 213         if not flags:
 214             self.flags = e.IPSEC_API_SAD_FLAG_NONE
 215         else:
 216             self.flags = flags
 217         if (tun_src):
 218             self.tun_src = ip_address(text_type(tun_src))
 219             self.flags = self.flags | e.IPSEC_API_SAD_FLAG_IS_TUNNEL
 220             if (self.tun_src.version == 6):
 221                 self.flags = self.flags | e.IPSEC_API_SAD_FLAG_IS_TUNNEL_V6
 222         if (tun_dst):
 223             self.tun_dst = ip_address(text_type(tun_dst))
 224         self.udp_src = udp_src
 225         self.udp_dst = udp_dst
 226         self.tun_flags = (VppEnum.vl_api_tunnel_encap_decap_flags_t.
 227                           TUNNEL_API_ENCAP_DECAP_FLAG_NONE)
 228         if tun_flags:
 229             self.tun_flags = tun_flags
 230         self.dscp = VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_CS0
 231         if dscp:
 232             self.dscp = dscp
 233         self.hop_limit = 255
 234         if hop_limit:
 235             self.hop_limit = hop_limit
 236
 237     def tunnel_encode(self):
 238         return {'src': (self.tun_src if self.tun_src else []),
 239                 'dst': (self.tun_dst if self.tun_dst else []),
 240                 'encap_decap_flags': self.tun_flags,
 241                 'dscp': self.dscp,
 242                 'hop_limit': self.hop_limit,
 243                 'table_id': self.table_id
 244                 }
 245
 246     def add_vpp_config(self):
 247         entry = {
 248             'sad_id': self.id,
 249             'spi': self.spi,
 250             'integrity_algorithm': self.integ_alg,
 251             'integrity_key': {
 252                 'length': len(self.integ_key),
 253                 'data': self.integ_key,
 254             },
 255             'crypto_algorithm': self.crypto_alg,
 256             'crypto_key': {
 257                 'data': self.crypto_key,
 258                 'length': len(self.crypto_key),
 259             },
 260             'protocol': self.proto,
 261             'tunnel': self.tunnel_encode(),
 262             'flags': self.flags,
 263             'salt': self.salt
 264         }
 265         # don't explicitly send the defaults, let papi fill them in
 266         if self.udp_src:
 267             entry['udp_src_port'] = self.udp_src
 268         if self.udp_dst:
 269             entry['udp_dst_port'] = self.udp_dst
 270         r = self.test.vapi.ipsec_sad_entry_add_del_v3(is_add=1, entry=entry)
 271         self.stat_index = r.stat_index
 272         self.test.registry.register(self, self.test.logger)
 273         return self
 274
 275     def remove_vpp_config(self):
 276         r = self.test.vapi.ipsec_sad_entry_add_del_v3(
 277             is_add=0,
 278             entry={
 279                 'sad_id': self.id,
 280                 'spi': self.spi,
 281                 'integrity_algorithm': self.integ_alg,
 282                 'integrity_key': {
 283                     'length': len(self.integ_key),
 284                     'data': self.integ_key,
 285                 },
 286                 'crypto_algorithm': self.crypto_alg,
 287                 'crypto_key': {
 288                     'data': self.crypto_key,
 289                     'length': len(self.crypto_key),
 290                 },
 291                 'protocol': self.proto,
 292                 'tunnel': self.tunnel_encode(),
 293                 'salt': self.salt
 294             })
 295
 296     def object_id(self):
 297         return "ipsec-sa-%d" % self.id
 298
 299     def query_vpp_config(self):
 300         e = VppEnum.vl_api_ipsec_sad_flags_t
 301
 302         bs = self.test.vapi.ipsec_sa_v3_dump()
 303         for b in bs:
 304             if b.entry.sad_id == self.id:
 305                 # if udp encap is configured then the ports should match
 306                 # those configured or the default
 307                 if (self.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP):
 308                     if not b.entry.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
 309                         return False
 310                     if self.udp_src:
 311                         if self.udp_src != b.entry.udp_src_port:
 312                             return False
 313                     else:
 314                         if self.DEFAULT_UDP_PORT != b.entry.udp_src_port:
 315                             return False
 316                     if self.udp_dst:
 317                         if self.udp_dst != b.entry.udp_dst_port:
 318                             return False
 319                     else:
 320                         if self.DEFAULT_UDP_PORT != b.entry.udp_dst_port:
 321                             return False
 322                 return True
 323         return False
 324
 325     def get_stats(self, worker=None):
 326         c = self.test.statistics.get_counter("/net/ipsec/sa")
 327         if worker is None:
 328             total = mk_counter()
 329             for t in c:
 330                 total['packets'] += t[self.stat_index]['packets']
 331             return total
 332         else:
 333             # +1 to skip main thread
 334             return c[worker+1][self.stat_index]
 335
 336
 337 class VppIpsecTunProtect(VppObject):
 338     """
 339     VPP IPSEC tunnel protection
 340     """
 341
 342     def __init__(self, test, itf, sa_out, sas_in, nh=None):
 343         self.test = test
 344         self.itf = itf
 345         self.sas_in = []
 346         for sa in sas_in:
 347             self.sas_in.append(sa.id)
 348         self.sa_out = sa_out.id
 349         self.nh = nh
 350         if not self.nh:
 351             self.nh = "0.0.0.0"
 352
 353     def update_vpp_config(self, sa_out, sas_in):
 354         self.sas_in = []
 355         for sa in sas_in:
 356             self.sas_in.append(sa.id)
 357         self.sa_out = sa_out.id
 358         self.test.vapi.ipsec_tunnel_protect_update(
 359             tunnel={
 360                 'sw_if_index': self.itf._sw_if_index,
 361                 'n_sa_in': len(self.sas_in),
 362                 'sa_out': self.sa_out,
 363                 'sa_in': self.sas_in,
 364                 'nh': self.nh})
 365
 366     def object_id(self):
 367         return "ipsec-tun-protect-%s-%s" % (self.itf, self.nh)
 368
 369     def add_vpp_config(self):
 370         self.test.vapi.ipsec_tunnel_protect_update(
 371             tunnel={
 372                 'sw_if_index': self.itf._sw_if_index,
 373                 'n_sa_in': len(self.sas_in),
 374                 'sa_out': self.sa_out,
 375                 'sa_in': self.sas_in,
 376                 'nh': self.nh})
 377         self.test.registry.register(self, self.test.logger)
 378
 379     def remove_vpp_config(self):
 380         self.test.vapi.ipsec_tunnel_protect_del(
 381             sw_if_index=self.itf.sw_if_index,
 382             nh=self.nh)
 383
 384     def query_vpp_config(self):
 385         bs = self.test.vapi.ipsec_tunnel_protect_dump(
 386             sw_if_index=self.itf.sw_if_index)
 387         for b in bs:
 388             if b.tun.sw_if_index == self.itf.sw_if_index and \
 389                self.nh == str(b.tun.nh):
 390                 return True
 391         return False
 392
 393
 394 class VppIpsecInterface(VppInterface):
 395     """
 396     VPP IPSec interface
 397     """
 398
 399     def __init__(self, test, mode=None, instance=0xffffffff):
 400         super(VppIpsecInterface, self).__init__(test)
 401
 402         self.mode = mode
 403         if not self.mode:
 404             self.mode = (VppEnum.vl_api_tunnel_mode_t.
 405                          TUNNEL_API_MODE_P2P)
 406         self.instance = instance
 407
 408     def add_vpp_config(self):
 409         r = self.test.vapi.ipsec_itf_create(itf={
 410             'user_instance': self.instance,
 411             'mode': self.mode,
 412         })
 413         self.set_sw_if_index(r.sw_if_index)
 414         self.test.registry.register(self, self.test.logger)
 415         return self
 416
 417     def remove_vpp_config(self):
 418         self.test.vapi.ipsec_itf_delete(sw_if_index=self._sw_if_index)
 419
 420     def query_vpp_config(self):
 421         ts = self.test.vapi.ipsec_itf_dump(sw_if_index=0xffffffff)
 422         for t in ts:
 423             if t.tunnel.sw_if_index == self._sw_if_index:
 424                 return True
 425         return False
 426
 427     def __str__(self):
 428         return self.object_id()
 429
 430     def object_id(self):
 431         return "ipsec-%d" % self._sw_if_index