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