JumpAvg: Fix string format
[csit.git] / resources / libraries / python / NATUtil.py
1 # Copyright (c) 2021 Cisco and/or its affiliates.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
5 #
6 #     http://www.apache.org/licenses/LICENSE-2.0
7 #
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
13
14 """NAT utilities library."""
15
16 from math import log2, modf
17 from pprint import pformat
18 from enum import IntEnum
19
20 from ipaddress import IPv4Address
21 from robot.api import logger
22
23 from resources.libraries.python.Constants import Constants
24 from resources.libraries.python.InterfaceUtil import InterfaceUtil
25 from resources.libraries.python.topology import Topology
26 from resources.libraries.python.PapiExecutor import PapiSocketExecutor
27
28
29 class NatConfigFlags(IntEnum):
30     """NAT plugin configuration flags"""
31     NAT_IS_NONE = 0x00
32     NAT_IS_TWICE_NAT = 0x01
33     NAT_IS_SELF_TWICE_NAT = 0x02
34     NAT_IS_OUT2IN_ONLY = 0x04
35     NAT_IS_ADDR_ONLY = 0x08
36     NAT_IS_OUTSIDE = 0x10
37     NAT_IS_INSIDE = 0x20
38     NAT_IS_STATIC = 0x40
39     NAT_IS_EXT_HOST_VALID = 0x80
40
41
42 class Nat44ConfigFlags(IntEnum):
43     """NAT44 configuration flags"""
44     NAT44_IS_ENDPOINT_INDEPENDENT = 0x00
45     NAT44_IS_ENDPOINT_DEPENDENT = 0x01
46     NAT44_IS_STATIC_MAPPING_ONLY = 0x02
47     NAT44_IS_CONNECTION_TRACKING = 0x04
48     NAT44_IS_OUT2IN_DPO = 0x08
49
50
51 class NatAddrPortAllocAlg(IntEnum):
52     """NAT Address and port assignment algorithms."""
53     NAT_ALLOC_ALG_DEFAULT = 0
54     NAT_ALLOC_ALG_MAP_E = 1
55     NAT_ALLOC_ALG_PORT_RANGE = 2
56
57
58 class NATUtil:
59     """This class defines the methods to set NAT."""
60
61     def __init__(self):
62         pass
63
64     @staticmethod
65     def enable_nat44_plugin(
66             node, inside_vrf=0, outside_vrf=0, users=0, user_memory=0,
67             sessions=0, session_memory=0, user_sessions=0, mode=u""):
68         """Enable NAT44 plugin.
69
70         :param node: DUT node.
71         :param inside_vrf: Inside VRF ID.
72         :param outside_vrf: Outside VRF ID.
73         :param users: Maximum number of users. Used only in endpoint-independent
74             mode.
75         :param user_memory: User memory size - overwrite auto calculated hash
76             allocation parameter if non-zero.
77         :param sessions: Maximum number of sessions.
78         :param session_memory: Session memory size - overwrite auto calculated
79             hash allocation parameter if non-zero.
80         :param user_sessions: Maximum number of sessions per user. Used only in
81             endpoint-independent mode.
82         :param mode: NAT44 mode. Valid values:
83             - endpoint-independent
84             - endpoint-dependent
85             - static-mapping-only
86             - connection-tracking
87             - out2in-dpo
88         :type node: dict
89         :type inside_vrf: str or int
90         :type outside_vrf: str or int
91         :type users: str or int
92         :type user_memory: str or int
93         :type sessions: str or int
94         :type session_memory: str or int
95         :type user_sessions: str or int
96         :type mode: str
97         """
98         cmd = u"nat44_plugin_enable_disable"
99         err_msg = f"Failed to enable NAT44 plugin on the host {node[u'host']}!"
100         args_in = dict(
101             enable=True,
102             inside_vrf=int(inside_vrf),
103             outside_vrf=int(outside_vrf),
104             users=int(users),
105             user_memory=int(user_memory),
106             sessions=int(sessions),
107             session_memory=int(session_memory),
108             user_sessions=int(user_sessions),
109             flags=getattr(
110                 Nat44ConfigFlags,
111                 f"NAT44_IS_{mode.replace(u'-', u'_').upper()}"
112             ).value
113         )
114
115         with PapiSocketExecutor(node) as papi_exec:
116             papi_exec.add(cmd, **args_in).get_reply(err_msg)
117
118     @staticmethod
119     def set_nat44_interface(node, interface, flag):
120         """Set inside and outside interfaces for NAT44.
121
122         :param node: DUT node.
123         :param interface: NAT44 interface.
124         :param flag: Interface NAT configuration flag name.
125         :type node: dict
126         :type interface: str
127         :type flag: str
128         """
129         cmd = u"nat44_interface_add_del_feature"
130
131         err_msg = f"Failed to set {flag} interface {interface} for NAT44 " \
132             f"on host {node[u'host']}"
133         args_in = dict(
134             sw_if_index=InterfaceUtil.get_sw_if_index(node, interface),
135             is_add=1,
136             flags=getattr(NatConfigFlags, flag).value
137         )
138
139         with PapiSocketExecutor(node) as papi_exec:
140             papi_exec.add(cmd, **args_in).get_reply(err_msg)
141
142     @staticmethod
143     def set_nat44_interfaces(node, int_in, int_out):
144         """Set inside and outside interfaces for NAT44.
145
146         :param node: DUT node.
147         :param int_in: Inside interface.
148         :param int_out: Outside interface.
149         :type node: dict
150         :type int_in: str
151         :type int_out: str
152         """
153         NATUtil.set_nat44_interface(node, int_in, u"NAT_IS_INSIDE")
154         NATUtil.set_nat44_interface(node, int_out, u"NAT_IS_OUTSIDE")
155
156     @staticmethod
157     def set_nat44_address_range(
158             node, start_ip, end_ip, vrf_id=Constants.BITWISE_NON_ZERO,
159             flag=u"NAT_IS_NONE"):
160         """Set NAT44 address range.
161
162         The return value is a callable (zero argument Python function)
163         which can be used to reset NAT state, so repeated trial measurements
164         hit the same slow path.
165
166         :param node: DUT node.
167         :param start_ip: IP range start.
168         :param end_ip: IP range end.
169         :param vrf_id: VRF index (Optional).
170         :param flag: NAT flag name.
171         :type node: dict
172         :type start_ip: str
173         :type end_ip: str
174         :type vrf_id: int
175         :type flag: str
176         :returns: Resetter of the NAT state.
177         :rtype: Callable[[], None]
178         """
179         cmd = u"nat44_add_del_address_range"
180         err_msg = f"Failed to set NAT44 address range on host {node[u'host']}"
181         args_in = dict(
182             is_add=True,
183             first_ip_address=IPv4Address(str(start_ip)).packed,
184             last_ip_address=IPv4Address(str(end_ip)).packed,
185             vrf_id=vrf_id,
186             flags=getattr(NatConfigFlags, flag).value
187         )
188
189         with PapiSocketExecutor(node) as papi_exec:
190             papi_exec.add(cmd, **args_in).get_reply(err_msg)
191
192         # A closure, accessing the variables above.
193         def resetter():
194             """Delete and re-add the NAT range setting."""
195             with PapiSocketExecutor(node) as papi_exec:
196                 args_in[u"is_add"] = False
197                 papi_exec.add(cmd, **args_in)
198                 args_in[u"is_add"] = True
199                 papi_exec.add(cmd, **args_in)
200                 papi_exec.get_replies(err_msg)
201
202         return resetter
203
204     @staticmethod
205     def show_nat44_config(node):
206         """Show the NAT44 plugin running configuration.
207
208         :param node: DUT node.
209         :type node: dict
210         """
211         cmd = u"nat44_show_running_config"
212         err_msg = f"Failed to get NAT44 configuration on host {node[u'host']}"
213
214         with PapiSocketExecutor(node) as papi_exec:
215             reply = papi_exec.add(cmd).get_reply(err_msg)
216
217         logger.debug(f"NAT44 Configuration:\n{pformat(reply)}")
218
219     @staticmethod
220     def show_nat44_summary(node):
221         """Show NAT44 summary on the specified topology node.
222
223         :param node: Topology node.
224         :type node: dict
225         :returns: NAT44 summary data.
226         :rtype: str
227         """
228         return PapiSocketExecutor.run_cli_cmd(node, u"show nat44 summary")
229
230     @staticmethod
231     def show_nat_base_data(node):
232         """Show the NAT base data.
233
234         Used data sources:
235
236             nat_worker_dump
237             nat44_interface_addr_dump
238             nat44_address_dump
239             nat44_static_mapping_dump
240             nat44_interface_dump
241
242         :param node: DUT node.
243         :type node: dict
244         """
245         cmds = [
246             u"nat_worker_dump",
247             u"nat44_interface_addr_dump",
248             u"nat44_address_dump",
249             u"nat44_static_mapping_dump",
250             u"nat44_interface_dump",
251         ]
252         PapiSocketExecutor.dump_and_log(node, cmds)
253
254     @staticmethod
255     def show_nat_user_data(node):
256         """Show the NAT user data.
257
258         Used data sources:
259
260             nat44_user_dump
261             nat44_user_session_dump
262
263         :param node: DUT node.
264         :type node: dict
265         """
266         cmds = [
267             u"nat44_user_dump",
268             u"nat44_user_session_dump",
269         ]
270         PapiSocketExecutor.dump_and_log(node, cmds)
271
272     @staticmethod
273     def compute_max_translations_per_thread(sessions, threads):
274         """Compute value of max_translations_per_thread NAT44 parameter based on
275         total number of worker threads.
276
277         :param sessions: Required number of NAT44 sessions.
278         :param threads: Number of worker threads.
279         :type sessions: int
280         :type threads: int
281         :returns: Value of max_translations_per_thread NAT44 parameter.
282         :rtype: int
283         """
284         # vpp-device tests have not dedicated physical core so
285         # ${dp_count_int} == 0 but we need to use one thread
286         threads = 1 if not int(threads) else int(threads)
287         rest, mult = modf(log2(sessions/(10*threads)))
288         return 2 ** (int(mult) + (1 if rest else 0)) * 10
289
290     @staticmethod
291     def get_nat44_sessions_number(node, proto):
292         """Get number of established NAT44 sessions from actual NAT44 mapping
293         data.
294
295         :param node: DUT node.
296         :param proto: Required protocol - TCP/UDP/ICMP.
297         :type node: dict
298         :type proto: str
299         :returns: Number of established NAT44 sessions.
300         :rtype: int
301         :raises ValueError: If not supported protocol.
302         """
303         nat44_data = dict()
304         if proto in [u"UDP", u"TCP", u"ICMP"]:
305             for line in NATUtil.show_nat44_summary(node).splitlines():
306                 sum_k, sum_v = line.split(u":") if u":" in line \
307                     else (line, None)
308                 nat44_data[sum_k] = sum_v.strip() if isinstance(sum_v, str) \
309                     else sum_v
310         else:
311             raise ValueError(f"Unsupported protocol: {proto}!")
312         return nat44_data.get(f"total {proto.lower()} sessions", 0)
313
314     # DET44 PAPI calls
315     # DET44 means deterministic mode of NAT44
316     @staticmethod
317     def enable_det44_plugin(node, inside_vrf=0, outside_vrf=0):
318         """Enable DET44 plugin.
319
320         :param node: DUT node.
321         :param inside_vrf: Inside VRF ID.
322         :param outside_vrf: Outside VRF ID.
323         :type node: dict
324         :type inside_vrf: str or int
325         :type outside_vrf: str or int
326         """
327         cmd = u"det44_plugin_enable_disable"
328         err_msg = f"Failed to enable DET44 plugin on the host {node[u'host']}!"
329         args_in = dict(
330             enable=True,
331             inside_vrf=int(inside_vrf),
332             outside_vrf=int(outside_vrf)
333         )
334
335         with PapiSocketExecutor(node) as papi_exec:
336             papi_exec.add(cmd, **args_in).get_reply(err_msg)
337
338     @staticmethod
339     def set_det44_interface(node, if_key, is_inside):
340         """Enable DET44 feature on the interface.
341
342         :param node: DUT node.
343         :param if_key: Interface key from topology file of interface
344             to enable DET44 feature on.
345         :param is_inside: True if interface is inside, False if outside.
346         :type node: dict
347         :type if_key: str
348         :type is_inside: bool
349         """
350         cmd = u"det44_interface_add_del_feature"
351         err_msg = f"Failed to enable DET44 feature on the interface {if_key} " \
352             f"on the host {node[u'host']}!"
353         args_in = dict(
354             is_add=True,
355             is_inside=is_inside,
356             sw_if_index=Topology.get_interface_sw_index(node, if_key)
357         )
358
359         with PapiSocketExecutor(node) as papi_exec:
360             papi_exec.add(cmd, **args_in).get_reply(err_msg)
361
362     @staticmethod
363     def set_det44_mapping(node, ip_in, subnet_in, ip_out, subnet_out):
364         """Set DET44 mapping.
365
366         The return value is a callable (zero argument Python function)
367         which can be used to reset NAT state, so repeated trial measurements
368         hit the same slow path.
369
370         :param node: DUT node.
371         :param ip_in: Inside IP.
372         :param subnet_in: Inside IP subnet.
373         :param ip_out: Outside IP.
374         :param subnet_out: Outside IP subnet.
375         :type node: dict
376         :type ip_in: str
377         :type subnet_in: str or int
378         :type ip_out: str
379         :type subnet_out: str or int
380         """
381         cmd = u"det44_add_del_map"
382         err_msg = f"Failed to set DET44 mapping on the host {node[u'host']}!"
383         args_in = dict(
384             is_add=True,
385             in_addr=IPv4Address(str(ip_in)).packed,
386             in_plen=int(subnet_in),
387             out_addr=IPv4Address(str(ip_out)).packed,
388             out_plen=int(subnet_out)
389         )
390
391         with PapiSocketExecutor(node) as papi_exec:
392             papi_exec.add(cmd, **args_in).get_reply(err_msg)
393
394         # A closure, accessing the variables above.
395         def resetter():
396             """Delete and re-add the deterministic NAT mapping."""
397             with PapiSocketExecutor(node) as papi_exec:
398                 args_in[u"is_add"] = False
399                 papi_exec.add(cmd, **args_in)
400                 args_in[u"is_add"] = True
401                 papi_exec.add(cmd, **args_in)
402                 papi_exec.get_replies(err_msg)
403
404         return resetter
405
406     @staticmethod
407     def get_det44_mapping(node):
408         """Get DET44 mapping data.
409
410         :param node: DUT node.
411         :type node: dict
412         :returns: Dictionary of DET44 mapping data.
413         :rtype: dict
414         """
415         cmd = u"det44_map_dump"
416         err_msg = f"Failed to get DET44 mapping data on the host " \
417             f"{node[u'host']}!"
418         args_in = dict()
419         with PapiSocketExecutor(node) as papi_exec:
420             details = papi_exec.add(cmd, **args_in).get_reply(err_msg)
421
422         return details
423
424     @staticmethod
425     def get_det44_sessions_number(node):
426         """Get number of established DET44 sessions from actual DET44 mapping
427         data.
428         :param node: DUT node.
429         :type node: dict
430         :returns: Number of established DET44 sessions.
431         :rtype: int
432         """
433         det44_data = NATUtil.get_det44_mapping(node)
434         return det44_data.get(u"ses_num", 0)
435
436     @staticmethod
437     def show_det44(node):
438         """Show DET44 data.
439
440         Used data sources:
441
442             det44_interface_dump
443             det44_map_dump
444             det44_session_dump
445
446         :param node: DUT node.
447         :type node: dict
448         """
449         cmds = [
450             u"det44_interface_dump",
451             u"det44_map_dump",
452             u"det44_session_dump",
453         ]
454         PapiSocketExecutor.dump_and_log(node, cmds)