test/vpp_ikev2.py

   1 from ipaddress import IPv4Address, AddressValueError
   2 from vpp_object import VppObject
   3 from vpp_papi import VppEnum
   4
   5
   6 class AuthMethod:
   7     v = {"rsa-sig": 1, "shared-key": 2}
   8
   9     @staticmethod
  10     def value(key):
  11         return AuthMethod.v[key]
  12
  13
  14 class IDType:
  15     v = {"ip4-addr": 1, "fqdn": 2, "ip6-addr": 5}
  16
  17     @staticmethod
  18     def value(key):
  19         return IDType.v[key]
  20
  21
  22 class Profile(VppObject):
  23     """IKEv2 profile"""
  24
  25     def __init__(self, test, profile_name):
  26         self.test = test
  27         self.vapi = test.vapi
  28         self.profile_name = profile_name
  29         self.udp_encap = False
  30         self.natt = True
  31
  32     def disable_natt(self):
  33         self.natt = False
  34
  35     def add_auth(self, method, data, is_hex=False):
  36         if isinstance(method, int):
  37             m = method
  38         elif isinstance(method, str):
  39             m = AuthMethod.value(method)
  40         else:
  41             raise Exception("unsupported type {}".format(method))
  42         self.auth = {"auth_method": m, "data": data, "is_hex": is_hex}
  43
  44     def add_local_id(self, id_type, data):
  45         if isinstance(id_type, str):
  46             t = IDType.value(id_type)
  47         self.local_id = {"id_type": t, "data": data, "is_local": True}
  48
  49     def add_remote_id(self, id_type, data):
  50         if isinstance(id_type, str):
  51             t = IDType.value(id_type)
  52         self.remote_id = {"id_type": t, "data": data, "is_local": False}
  53
  54     def add_local_ts(
  55         self, start_addr, end_addr, start_port=0, end_port=0xFFFF, proto=0, is_ip4=True
  56     ):
  57         self.ts_is_ip4 = is_ip4
  58         self.local_ts = {
  59             "is_local": True,
  60             "protocol_id": proto,
  61             "start_port": start_port,
  62             "end_port": end_port,
  63             "start_addr": start_addr,
  64             "end_addr": end_addr,
  65         }
  66
  67     def add_remote_ts(
  68         self, start_addr, end_addr, start_port=0, end_port=0xFFFF, proto=0
  69     ):
  70         try:
  71             IPv4Address(start_addr)
  72             is_ip4 = True
  73         except AddressValueError:
  74             is_ip4 = False
  75         self.ts_is_ip4 = is_ip4
  76         self.remote_ts = {
  77             "is_local": False,
  78             "protocol_id": proto,
  79             "start_port": start_port,
  80             "end_port": end_port,
  81             "start_addr": start_addr,
  82             "end_addr": end_addr,
  83         }
  84
  85     def add_responder_hostname(self, hn):
  86         self.responder_hostname = hn
  87
  88     def add_responder(self, responder):
  89         self.responder = responder
  90
  91     def add_ike_transforms(self, tr):
  92         self.ike_transforms = tr
  93
  94     def add_esp_transforms(self, tr):
  95         self.esp_transforms = tr
  96
  97     def set_udp_encap(self, udp_encap):
  98         self.udp_encap = udp_encap
  99
 100     def set_lifetime_data(self, data):
 101         self.lifetime_data = data
 102
 103     def set_ipsec_over_udp_port(self, port):
 104         self.ipsec_udp_port = {"is_set": 1, "port": port}
 105
 106     def set_tunnel_interface(self, sw_if_index):
 107         self.tun_itf = sw_if_index
 108
 109     def object_id(self):
 110         return "ikev2-profile-%s" % self.profile_name
 111
 112     def remove_vpp_config(self):
 113         self.vapi.ikev2_profile_add_del(name=self.profile_name, is_add=False)
 114
 115     def add_vpp_config(self):
 116         self.vapi.ikev2_profile_add_del(name=self.profile_name, is_add=True)
 117         if hasattr(self, "auth"):
 118             self.vapi.ikev2_profile_set_auth(
 119                 name=self.profile_name, data_len=len(self.auth["data"]), **self.auth
 120             )
 121         if hasattr(self, "local_id"):
 122             self.vapi.ikev2_profile_set_id(
 123                 name=self.profile_name,
 124                 data_len=len(self.local_id["data"]),
 125                 **self.local_id,
 126             )
 127         if hasattr(self, "remote_id"):
 128             self.vapi.ikev2_profile_set_id(
 129                 name=self.profile_name,
 130                 data_len=len(self.remote_id["data"]),
 131                 **self.remote_id,
 132             )
 133         if hasattr(self, "local_ts"):
 134             self.vapi.ikev2_profile_set_ts(name=self.profile_name, ts=self.local_ts)
 135
 136         if hasattr(self, "remote_ts"):
 137             self.vapi.ikev2_profile_set_ts(name=self.profile_name, ts=self.remote_ts)
 138
 139         if hasattr(self, "responder"):
 140             self.vapi.ikev2_set_responder(
 141                 name=self.profile_name, responder=self.responder
 142             )
 143
 144         if hasattr(self, "responder_hostname"):
 145             print(self.responder_hostname)
 146             self.vapi.ikev2_set_responder_hostname(
 147                 name=self.profile_name, **self.responder_hostname
 148             )
 149
 150         if hasattr(self, "ike_transforms"):
 151             self.vapi.ikev2_set_ike_transforms(
 152                 name=self.profile_name, tr=self.ike_transforms
 153             )
 154
 155         if hasattr(self, "esp_transforms"):
 156             self.vapi.ikev2_set_esp_transforms(
 157                 name=self.profile_name, tr=self.esp_transforms
 158             )
 159
 160         if self.udp_encap:
 161             self.vapi.ikev2_profile_set_udp_encap(name=self.profile_name)
 162
 163         if hasattr(self, "lifetime_data"):
 164             self.vapi.ikev2_set_sa_lifetime(
 165                 name=self.profile_name, **self.lifetime_data
 166             )
 167
 168         if hasattr(self, "ipsec_udp_port"):
 169             self.vapi.ikev2_profile_set_ipsec_udp_port(
 170                 name=self.profile_name, **self.ipsec_udp_port
 171             )
 172         if hasattr(self, "tun_itf"):
 173             self.vapi.ikev2_set_tunnel_interface(
 174                 name=self.profile_name, sw_if_index=self.tun_itf
 175             )
 176
 177         if not self.natt:
 178             self.vapi.ikev2_profile_disable_natt(name=self.profile_name)
 179
 180     def query_vpp_config(self):
 181         res = self.vapi.ikev2_profile_dump()
 182         for r in res:
 183             if r.profile.name == self.profile_name:
 184                 return r.profile
 185         return None