src/plugins/ikev2/test/vpp_ikev2.py

   1 from vpp_object import VppObject
   2 from vpp_papi import VppEnum
   3
   4
   5 class AuthMethod:
   6     v = {'rsa-sig': 1,
   7          'shared-key': 2}
   8
   9     @staticmethod
  10     def value(key): return AuthMethod.v[key]
  11
  12
  13 class IDType:
  14     v = {'ip4-addr': 1,
  15          'fqdn': 2}
  16
  17     @staticmethod
  18     def value(key): return IDType.v[key]
  19
  20
  21 class Profile(VppObject):
  22     """ IKEv2 profile """
  23     def __init__(self, test, profile_name):
  24         self.test = test
  25         self.vapi = test.vapi
  26         self.profile_name = profile_name
  27
  28     def add_auth(self, method, data, is_hex=False):
  29         if isinstance(method, int):
  30             m = method
  31         elif isinstance(method, str):
  32             m = AuthMethod.value(method)
  33         else:
  34             raise Exception('unsupported type {}'.format(method))
  35         self.auth = {'auth_method': m,
  36                      'data': data,
  37                      'is_hex': is_hex}
  38
  39     def add_local_id(self, id_type, data):
  40         if isinstance(id_type, str):
  41             t = IDType.value(id_type)
  42         self.local_id = {'id_type': t,
  43                          'data': data,
  44                          'is_local': True}
  45
  46     def add_remote_id(self, id_type, data):
  47         if isinstance(id_type, str):
  48             t = IDType.value(id_type)
  49         self.remote_id = {'id_type': t,
  50                           'data': data,
  51                           'is_local': False}
  52
  53     def add_local_ts(self, start_addr, end_addr, start_port=0, end_port=0xffff,
  54                      proto=0):
  55         self.local_ts = {'is_local': True,
  56                          'proto': proto,
  57                          'start_port': start_port,
  58                          'end_port': end_port,
  59                          'start_addr': start_addr,
  60                          'end_addr': end_addr}
  61
  62     def add_remote_ts(self, start_addr, end_addr, start_port=0,
  63                       end_port=0xffff, proto=0):
  64         self.remote_ts = {'is_local': False,
  65                           'proto': proto,
  66                           'start_port': start_port,
  67                           'end_port': end_port,
  68                           'start_addr': start_addr,
  69                           'end_addr': end_addr}
  70
  71     def object_id(self):
  72         return 'ikev2-profile-%s' % self.profile_name
  73
  74     def remove_vpp_config(self):
  75         self.vapi.ikev2_profile_add_del(name=self.profile_name, is_add=False)
  76
  77     def add_vpp_config(self):
  78         self.vapi.ikev2_profile_add_del(name=self.profile_name, is_add=True)
  79         if hasattr(self, 'auth'):
  80             self.vapi.ikev2_profile_set_auth(name=self.profile_name,
  81                                              data_len=len(self.auth['data']),
  82                                              **self.auth)
  83         if hasattr(self, 'local_id'):
  84             self.vapi.ikev2_profile_set_id(name=self.profile_name,
  85                                            data_len=len(self.local_id
  86                                                         ['data']),
  87                                            **self.local_id)
  88         if hasattr(self, 'remote_id'):
  89             self.vapi.ikev2_profile_set_id(name=self.profile_name,
  90                                            data_len=len(self.remote_id
  91                                                         ['data']),
  92                                            **self.remote_id)
  93         if hasattr(self, 'local_ts'):
  94             self.vapi.ikev2_profile_set_ts(name=self.profile_name,
  95                                            **self.local_ts)
  96         if hasattr(self, 'remote_ts'):
  97             self.vapi.ikev2_profile_set_ts(name=self.profile_name,
  98                                            **self.remote_ts)
  99
 100     def query_vpp_config(self):
 101         raise NotImplementedError()