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         anti_replay_window_size=0,
 222     ):
 223         e = VppEnum.vl_api_ipsec_sad_flags_t
 224         self.test = test
 225         self.id = id
 226         self.spi = spi
 227         self.integ_alg = integ_alg
 228         self.integ_key = integ_key
 229         self.crypto_alg = crypto_alg
 230         self.crypto_key = crypto_key
 231         self.proto = proto
 232         self.salt = salt
 233         self.anti_replay_window_size = anti_replay_window_size
 234
 235         self.table_id = 0
 236         self.tun_src = tun_src
 237         self.tun_dst = tun_dst
 238         if not flags:
 239             self.flags = e.IPSEC_API_SAD_FLAG_NONE
 240         else:
 241             self.flags = flags
 242         if tun_src:
 243             self.tun_src = ip_address(text_type(tun_src))
 244             self.flags = self.flags | e.IPSEC_API_SAD_FLAG_IS_TUNNEL
 245         if tun_dst:
 246             self.tun_dst = ip_address(text_type(tun_dst))
 247         self.udp_src = udp_src
 248         self.udp_dst = udp_dst
 249         self.tun_flags = (
 250             VppEnum.vl_api_tunnel_encap_decap_flags_t.TUNNEL_API_ENCAP_DECAP_FLAG_NONE
 251         )
 252         if tun_flags:
 253             self.tun_flags = tun_flags
 254         self.dscp = VppEnum.vl_api_ip_dscp_t.IP_API_DSCP_CS0
 255         if dscp:
 256             self.dscp = dscp
 257         self.hop_limit = 255
 258         if hop_limit:
 259             self.hop_limit = hop_limit
 260
 261     def tunnel_encode(self):
 262         return {
 263             "src": (self.tun_src if self.tun_src else []),
 264             "dst": (self.tun_dst if self.tun_dst else []),
 265             "encap_decap_flags": self.tun_flags,
 266             "dscp": self.dscp,
 267             "hop_limit": self.hop_limit,
 268             "table_id": self.table_id,
 269         }
 270
 271     def add_vpp_config(self):
 272         entry = {
 273             "sad_id": self.id,
 274             "spi": self.spi,
 275             "integrity_algorithm": self.integ_alg,
 276             "integrity_key": {
 277                 "length": len(self.integ_key),
 278                 "data": self.integ_key,
 279             },
 280             "crypto_algorithm": self.crypto_alg,
 281             "crypto_key": {
 282                 "data": self.crypto_key,
 283                 "length": len(self.crypto_key),
 284             },
 285             "protocol": self.proto,
 286             "tunnel": self.tunnel_encode(),
 287             "flags": self.flags,
 288             "salt": self.salt,
 289             "anti_replay_window_size": self.anti_replay_window_size,
 290         }
 291         # don't explicitly send the defaults, let papi fill them in
 292         if self.udp_src:
 293             entry["udp_src_port"] = self.udp_src
 294         if self.udp_dst:
 295             entry["udp_dst_port"] = self.udp_dst
 296         r = self.test.vapi.ipsec_sad_entry_add_v2(entry=entry)
 297         self.stat_index = r.stat_index
 298         self.test.registry.register(self, self.test.logger)
 299         return self
 300
 301     def update_vpp_config(
 302         self, udp_src=None, udp_dst=None, is_tun=False, tun_src=None, tun_dst=None
 303     ):
 304         if is_tun:
 305             if tun_src:
 306                 self.tun_src = ip_address(text_type(tun_src))
 307             if tun_dst:
 308                 self.tun_dst = ip_address(text_type(tun_dst))
 309         if udp_src:
 310             self.udp_src = udp_src
 311         if udp_dst:
 312             self.udp_dst = udp_dst
 313         self.test.vapi.ipsec_sad_entry_update(
 314             sad_id=self.id,
 315             is_tun=is_tun,
 316             tunnel=self.tunnel_encode(),
 317             udp_src_port=udp_src,
 318             udp_dst_port=udp_dst,
 319         )
 320
 321     def remove_vpp_config(self):
 322         self.test.vapi.ipsec_sad_entry_del(id=self.id)
 323
 324     def object_id(self):
 325         return "ipsec-sa-%d" % self.id
 326
 327     def query_vpp_config(self):
 328         e = VppEnum.vl_api_ipsec_sad_flags_t
 329
 330         bs = self.test.vapi.ipsec_sa_v5_dump()
 331         for b in bs:
 332             if b.entry.sad_id == self.id:
 333                 # if udp encap is configured then the ports should match
 334                 # those configured or the default
 335                 if self.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
 336                     if not b.entry.flags & e.IPSEC_API_SAD_FLAG_UDP_ENCAP:
 337                         return False
 338                     if self.udp_src:
 339                         if self.udp_src != b.entry.udp_src_port:
 340                             return False
 341                     else:
 342                         if self.DEFAULT_UDP_PORT != b.entry.udp_src_port:
 343                             return False
 344                     if self.udp_dst:
 345                         if self.udp_dst != b.entry.udp_dst_port:
 346                             return False
 347                     else:
 348                         if self.DEFAULT_UDP_PORT != b.entry.udp_dst_port:
 349                             return False
 350                 return True
 351         return False
 352
 353     def get_stats(self, worker=None):
 354         c = self.test.statistics.get_counter("/net/ipsec/sa")
 355         if worker is None:
 356             total = mk_counter()
 357             for t in c:
 358                 total["packets"] += t[self.stat_index]["packets"]
 359             return total
 360         else:
 361             # +1 to skip main thread
 362             return c[worker + 1][self.stat_index]
 363
 364     def get_err(self, name, worker=None):
 365         c = self.test.statistics.get_counter("/net/ipsec/sa/err/" + name)
 366         if worker is None:
 367             total = 0
 368             for t in c:
 369                 total += t[self.stat_index]
 370             return total
 371         else:
 372             # +1 to skip main thread
 373             return c[worker + 1][self.stat_index]
 374
 375
 376 class VppIpsecTunProtect(VppObject):
 377     """
 378     VPP IPSEC tunnel protection
 379     """
 380
 381     def __init__(self, test, itf, sa_out, sas_in, nh=None):
 382         self.test = test
 383         self.itf = itf
 384         self.sas_in = []
 385         for sa in sas_in:
 386             self.sas_in.append(sa.id)
 387         self.sa_out = sa_out.id
 388         self.nh = nh
 389         if not self.nh:
 390             self.nh = "0.0.0.0"
 391
 392     def update_vpp_config(self, sa_out, sas_in):
 393         self.sas_in = []
 394         for sa in sas_in:
 395             self.sas_in.append(sa.id)
 396         self.sa_out = sa_out.id
 397         self.test.vapi.ipsec_tunnel_protect_update(
 398             tunnel={
 399                 "sw_if_index": self.itf._sw_if_index,
 400                 "n_sa_in": len(self.sas_in),
 401                 "sa_out": self.sa_out,
 402                 "sa_in": self.sas_in,
 403                 "nh": self.nh,
 404             }
 405         )
 406
 407     def object_id(self):
 408         return "ipsec-tun-protect-%s-%s" % (self.itf, self.nh)
 409
 410     def add_vpp_config(self):
 411         self.test.vapi.ipsec_tunnel_protect_update(
 412             tunnel={
 413                 "sw_if_index": self.itf._sw_if_index,
 414                 "n_sa_in": len(self.sas_in),
 415                 "sa_out": self.sa_out,
 416                 "sa_in": self.sas_in,
 417                 "nh": self.nh,
 418             }
 419         )
 420         self.test.registry.register(self, self.test.logger)
 421
 422     def remove_vpp_config(self):
 423         self.test.vapi.ipsec_tunnel_protect_del(
 424             sw_if_index=self.itf.sw_if_index, nh=self.nh
 425         )
 426
 427     def query_vpp_config(self):
 428         bs = self.test.vapi.ipsec_tunnel_protect_dump(sw_if_index=self.itf.sw_if_index)
 429         for b in bs:
 430             if b.tun.sw_if_index == self.itf.sw_if_index and self.nh == str(b.tun.nh):
 431                 return True
 432         return False
 433
 434
 435 class VppIpsecInterface(VppInterface):
 436     """
 437     VPP IPSec interface
 438     """
 439
 440     def __init__(self, test, mode=None, instance=0xFFFFFFFF):
 441         super(VppIpsecInterface, self).__init__(test)
 442
 443         self.mode = mode
 444         if not self.mode:
 445             self.mode = VppEnum.vl_api_tunnel_mode_t.TUNNEL_API_MODE_P2P
 446         self.instance = instance
 447
 448     def add_vpp_config(self):
 449         r = self.test.vapi.ipsec_itf_create(
 450             itf={
 451                 "user_instance": self.instance,
 452                 "mode": self.mode,
 453             }
 454         )
 455         self.set_sw_if_index(r.sw_if_index)
 456         self.test.registry.register(self, self.test.logger)
 457         ts = self.test.vapi.ipsec_itf_dump(sw_if_index=self._sw_if_index)
 458         self.instance = ts[0].itf.user_instance
 459         return self
 460
 461     def remove_vpp_config(self):
 462         self.test.vapi.ipsec_itf_delete(sw_if_index=self._sw_if_index)
 463
 464     def query_vpp_config(self):
 465         ts = self.test.vapi.ipsec_itf_dump(sw_if_index=0xFFFFFFFF)
 466         for t in ts:
 467             if t.itf.sw_if_index == self._sw_if_index:
 468                 return True
 469         return False
 470
 471     def __str__(self):
 472         return self.object_id()
 473
 474     def object_id(self):
 475         return "ipsec%d" % self.instance