src/plugins/ikev2/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,
   8          'shared-key': 2}
   9
  10     @staticmethod
  11     def value(key): return AuthMethod.v[key]
  12
  13
  14 class IDType:
  15     v = {'ip4-addr': 1,
  16          'fqdn': 2,
  17          'ip6-addr': 5}
  18
  19     @staticmethod
  20     def value(key): return IDType.v[key]
  21
  22
  23 class Profile(VppObject):
  24     """ IKEv2 profile """
  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
  31     def add_auth(self, method, data, is_hex=False):
  32         if isinstance(method, int):
  33             m = method
  34         elif isinstance(method, str):
  35             m = AuthMethod.value(method)
  36         else:
  37             raise Exception('unsupported type {}'.format(method))
  38         self.auth = {'auth_method': m,
  39                      'data': data,
  40                      'is_hex': is_hex}
  41
  42     def add_local_id(self, id_type, data):
  43         if isinstance(id_type, str):
  44             t = IDType.value(id_type)
  45         self.local_id = {'id_type': t,
  46                          'data': data,
  47                          '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,
  53                           'data': data,
  54                           'is_local': False}
  55
  56     def add_local_ts(self, start_addr, end_addr, start_port=0, end_port=0xffff,
  57                      proto=0, is_ip4=True):
  58         self.ts_is_ip4 = is_ip4
  59         self.local_ts = {'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     def add_remote_ts(self, start_addr, end_addr, start_port=0,
  67                       end_port=0xffff, proto=0):
  68         try:
  69             IPv4Address(start_addr)
  70             is_ip4 = True
  71         except AddressValueError:
  72             is_ip4 = False
  73         self.ts_is_ip4 = is_ip4
  74         self.remote_ts = {'is_local': False,
  75                           'protocol_id': proto,
  76                           'start_port': start_port,
  77                           'end_port': end_port,
  78                           'start_addr': start_addr,
  79                           'end_addr': end_addr}
  80
  81     def add_responder(self, responder):
  82         self.responder = responder
  83
  84     def add_ike_transforms(self, tr):
  85         self.ike_transforms = tr
  86
  87     def add_esp_transforms(self, tr):
  88         self.esp_transforms = tr
  89
  90     def set_udp_encap(self, udp_encap):
  91         self.udp_encap = udp_encap
  92
  93     def set_lifetime_data(self, data):
  94         self.lifetime_data = data
  95
  96     def set_ipsec_over_udp_port(self, port):
  97         self.ipsec_udp_port = {'is_set': 1,
  98                                'port': port}
  99
 100     def set_tunnel_interface(self, sw_if_index):
 101         self.tun_itf = sw_if_index
 102
 103     def object_id(self):
 104         return 'ikev2-profile-%s' % self.profile_name
 105
 106     def remove_vpp_config(self):
 107         self.vapi.ikev2_profile_add_del(name=self.profile_name, is_add=False)
 108
 109     def add_vpp_config(self):
 110         self.vapi.ikev2_profile_add_del(name=self.profile_name, is_add=True)
 111         if hasattr(self, 'auth'):
 112             self.vapi.ikev2_profile_set_auth(name=self.profile_name,
 113                                              data_len=len(self.auth['data']),
 114                                              **self.auth)
 115         if hasattr(self, 'local_id'):
 116             self.vapi.ikev2_profile_set_id(name=self.profile_name,
 117                                            data_len=len(self.local_id
 118                                                         ['data']),
 119                                            **self.local_id)
 120         if hasattr(self, 'remote_id'):
 121             self.vapi.ikev2_profile_set_id(name=self.profile_name,
 122                                            data_len=len(self.remote_id
 123                                                         ['data']),
 124                                            **self.remote_id)
 125         if hasattr(self, 'local_ts'):
 126             self.vapi.ikev2_profile_set_ts(name=self.profile_name,
 127                                            ts=self.local_ts)
 128
 129         if hasattr(self, 'remote_ts'):
 130             self.vapi.ikev2_profile_set_ts(name=self.profile_name,
 131                                            ts=self.remote_ts)
 132
 133         if hasattr(self, 'responder'):
 134             self.vapi.ikev2_set_responder(name=self.profile_name,
 135                                           responder=self.responder)
 136
 137         if hasattr(self, 'ike_transforms'):
 138             self.vapi.ikev2_set_ike_transforms(name=self.profile_name,
 139                                                tr=self.ike_transforms)
 140
 141         if hasattr(self, 'esp_transforms'):
 142             self.vapi.ikev2_set_esp_transforms(name=self.profile_name,
 143                                                tr=self.esp_transforms)
 144
 145         if self.udp_encap:
 146             self.vapi.ikev2_profile_set_udp_encap(name=self.profile_name)
 147
 148         if hasattr(self, 'lifetime_data'):
 149             self.vapi.ikev2_set_sa_lifetime(name=self.profile_name,
 150                                             **self.lifetime_data)
 151
 152         if hasattr(self, 'ipsec_udp_port'):
 153             self.vapi.ikev2_profile_set_ipsec_udp_port(name=self.profile_name,
 154                                                        **self.ipsec_udp_port)
 155         if hasattr(self, 'tun_itf'):
 156             self.vapi.ikev2_set_tunnel_interface(name=self.profile_name,
 157                                                  sw_if_index=self.tun_itf)
 158
 159     def query_vpp_config(self):
 160         res = self.vapi.ikev2_profile_dump()
 161         for r in res:
 162             if r.profile.name == self.profile_name:
 163                 return r.profile
 164         return None