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=socket.IPPROTO_RAW,
  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 update_vpp_config(
 299         self, udp_src=None, udp_dst=None, is_tun=False, tun_src=None, tun_dst=None
 300     ):
 301         if is_tun:
 302             if tun_src:
 303                 self.tun_src = ip_address(text_type(tun_src))
 304             if tun_dst:
 305                 self.tun_dst = ip_address(text_type(tun_dst))
 306         if udp_src:
 307             self.udp_src = udp_src
 308         if udp_dst:
 309             self.udp_dst = udp_dst
 310         self.test.vapi.ipsec_sad_entry_update(
 311             sad_id=self.id,
 312             is_tun=is_tun,
 313             tunnel=self.tunnel_encode(),
 314             udp_src_port=udp_src,
 315             udp_dst_port=udp_dst,
 316         )
 317
 318     def remove_vpp_config(self):
 319         self.test.vapi.ipsec_sad_entry_del(id=self.id)
 320
 321     def object_id(self):
 322         return "ipsec-sa-%d" % self.id
 323
 324     def query_vpp_config(self):
 325         e = VppEnum.vl_api_ipsec_sad_flags_t
 326
 327         bs = self.test.vapi.ipsec_sa_v3_dump()
 328         for b in bs:
 329             if b.entry.sad_id == self.id:
 330                 # if udp encap is configured then the ports should match
 331                 # those configured or the default
 332                 if self.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
 333                     if not b.entry.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
 334                         return False
 335                     if self.udp_src:
 336                         if self.udp_src != b.entry.udp_src_port:
 337                             return False
 338                     else:
 339                         if self.DEFAULT_UDP_PORT != b.entry.udp_src_port:
 340                             return False
 341                     if self.udp_dst:
 342                         if self.udp_dst != b.entry.udp_dst_port:
 343                             return False
 344                     else:
 345                         if self.DEFAULT_UDP_PORT != b.entry.udp_dst_port:
 346                             return False
 347                 return True
 348         return False
 349
 350     def get_stats(self, worker=None):
 351         c = self.test.statistics.get_counter("/net/ipsec/sa")
 352         if worker is None:
 353             total = mk_counter()
 354             for t in c:
 355                 total["packets"] += t[self.stat_index]["packets"]
 356             return total
 357         else:
 358             # +1 to skip main thread
 359             return c[worker + 1][self.stat_index]
 360
 361     def get_err(self, name, worker=None):
 362         c = self.test.statistics.get_counter("/net/ipsec/sa/err/" + name)
 363         if worker is None:
 364             total = 0
 365             for t in c:
 366                 total += t[self.stat_index]
 367             return total
 368         else:
 369             # +1 to skip main thread
 370             return c[worker + 1][self.stat_index]
 371
 372
 373 class VppIpsecTunProtect(VppObject):
 374     """
 375     VPP IPSEC tunnel protection
 376     """
 377
 378     def __init__(self, test, itf, sa_out, sas_in, nh=None):
 379         self.test = test
 380         self.itf = itf
 381         self.sas_in = []
 382         for sa in sas_in:
 383             self.sas_in.append(sa.id)
 384         self.sa_out = sa_out.id
 385         self.nh = nh
 386         if not self.nh:
 387             self.nh = "0.0.0.0"
 388
 389     def update_vpp_config(self, sa_out, sas_in):
 390         self.sas_in = []
 391         for sa in sas_in:
 392             self.sas_in.append(sa.id)
 393         self.sa_out = sa_out.id
 394         self.test.vapi.ipsec_tunnel_protect_update(
 395             tunnel={
 396                 "sw_if_index": self.itf._sw_if_index,
 397                 "n_sa_in": len(self.sas_in),
 398                 "sa_out": self.sa_out,
 399                 "sa_in": self.sas_in,
 400                 "nh": self.nh,
 401             }
 402         )
 403
 404     def object_id(self):
 405         return "ipsec-tun-protect-%s-%s" % (self.itf, self.nh)
 406
 407     def add_vpp_config(self):
 408         self.test.vapi.ipsec_tunnel_protect_update(
 409             tunnel={
 410                 "sw_if_index": self.itf._sw_if_index,
 411                 "n_sa_in": len(self.sas_in),
 412                 "sa_out": self.sa_out,
 413                 "sa_in": self.sas_in,
 414                 "nh": self.nh,
 415             }
 416         )
 417         self.test.registry.register(self, self.test.logger)
 418
 419     def remove_vpp_config(self):
 420         self.test.vapi.ipsec_tunnel_protect_del(
 421             sw_if_index=self.itf.sw_if_index, nh=self.nh
 422         )
 423
 424     def query_vpp_config(self):
 425         bs = self.test.vapi.ipsec_tunnel_protect_dump(sw_if_index=self.itf.sw_if_index)
 426         for b in bs:
 427             if b.tun.sw_if_index == self.itf.sw_if_index and self.nh == str(b.tun.nh):
 428                 return True
 429         return False
 430
 431
 432 class VppIpsecInterface(VppInterface):
 433     """
 434     VPP IPSec interface
 435     """
 436
 437     def __init__(self, test, mode=None, instance=0xFFFFFFFF):
 438         super(VppIpsecInterface, self).__init__(test)
 439
 440         self.mode = mode
 441         if not self.mode:
 442             self.mode = VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_P2P
 443         self.instance = instance
 444
 445     def add_vpp_config(self):
 446         r = self.test.vapi.ipsec_itf_create(
 447             itf={
 448                 "user_instance": self.instance,
 449                 "mode": self.mode,
 450             }
 451         )
 452         self.set_sw_if_index(r.sw_if_index)
 453         self.test.registry.register(self, self.test.logger)
 454         ts = self.test.vapi.ipsec_itf_dump(sw_if_index=self._sw_if_index)
 455         self.instance = ts[0].itf.user_instance
 456         return self
 457
 458     def remove_vpp_config(self):
 459         self.test.vapi.ipsec_itf_delete(sw_if_index=self._sw_if_index)
 460
 461     def query_vpp_config(self):
 462         ts = self.test.vapi.ipsec_itf_dump(sw_if_index=0xFFFFFFFF)
 463         for t in ts:
 464             if t.itf.sw_if_index == self._sw_if_index:
 465                 return True
 466         return False
 467
 468     def __str__(self):
 469         return self.object_id()
 470
 471     def object_id(self):
 472         return "ipsec%d" % self.instance