1 from ipaddress import IPv4Address, AddressValueError
2 from vpp_object import VppObject
6 v = {"rsa-sig": 1, "shared-key": 2}
10 return AuthMethod.v[key]
14 v = {"ip4-addr": 1, "fqdn": 2, "ip6-addr": 5}
21 class Profile(VppObject):
24 def __init__(self, test, profile_name):
27 self.profile_name = profile_name
28 self.udp_encap = False
31 def disable_natt(self):
34 def add_auth(self, method, data, is_hex=False):
35 if isinstance(method, int):
37 elif isinstance(method, str):
38 m = AuthMethod.value(method)
40 raise Exception("unsupported type {}".format(method))
41 self.auth = {"auth_method": m, "data": data, "is_hex": is_hex}
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}
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}
54 self, start_addr, end_addr, start_port=0, end_port=0xFFFF, proto=0, is_ip4=True
56 self.ts_is_ip4 = is_ip4
60 "start_port": start_port,
62 "start_addr": start_addr,
67 self, start_addr, end_addr, start_port=0, end_port=0xFFFF, proto=0
70 IPv4Address(start_addr)
72 except AddressValueError:
74 self.ts_is_ip4 = is_ip4
78 "start_port": start_port,
80 "start_addr": start_addr,
84 def add_responder_hostname(self, hn):
85 self.responder_hostname = hn
87 def add_responder(self, responder):
88 self.responder = responder
90 def add_ike_transforms(self, tr):
91 self.ike_transforms = tr
93 def add_esp_transforms(self, tr):
94 self.esp_transforms = tr
96 def set_udp_encap(self, udp_encap):
97 self.udp_encap = udp_encap
99 def set_lifetime_data(self, data):
100 self.lifetime_data = data
102 def set_ipsec_over_udp_port(self, port):
103 self.ipsec_udp_port = {"is_set": 1, "port": port}
105 def set_tunnel_interface(self, sw_if_index):
106 self.tun_itf = sw_if_index
109 return "ikev2-profile-%s" % self.profile_name
111 def remove_vpp_config(self):
112 self.vapi.ikev2_profile_add_del(name=self.profile_name, is_add=False)
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
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"]),
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"]),
132 if hasattr(self, "local_ts"):
133 self.vapi.ikev2_profile_set_ts(name=self.profile_name, ts=self.local_ts)
135 if hasattr(self, "remote_ts"):
136 self.vapi.ikev2_profile_set_ts(name=self.profile_name, ts=self.remote_ts)
138 if hasattr(self, "responder"):
139 self.vapi.ikev2_set_responder(
140 name=self.profile_name, responder=self.responder
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
149 if hasattr(self, "ike_transforms"):
150 self.vapi.ikev2_set_ike_transforms(
151 name=self.profile_name, tr=self.ike_transforms
154 if hasattr(self, "esp_transforms"):
155 self.vapi.ikev2_set_esp_transforms(
156 name=self.profile_name, tr=self.esp_transforms
160 self.vapi.ikev2_profile_set_udp_encap(name=self.profile_name)
162 if hasattr(self, "lifetime_data"):
163 self.vapi.ikev2_set_sa_lifetime(
164 name=self.profile_name, **self.lifetime_data
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
171 if hasattr(self, "tun_itf"):
172 self.vapi.ikev2_set_tunnel_interface(
173 name=self.profile_name, sw_if_index=self.tun_itf
177 self.vapi.ikev2_profile_disable_natt(name=self.profile_name)
179 def query_vpp_config(self):
180 res = self.vapi.ikev2_profile_dump()
182 if r.profile.name == self.profile_name: