test/vpp_ipsec.py

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