test/vpp_ikev2.py

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