1 # Copyright (c) 2020 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:
6 # http://www.apache.org/licenses/LICENSE-2.0
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.
14 """NAT utilities library."""
16 from pprint import pformat
17 from enum import IntEnum
19 from ipaddress import IPv4Address
20 from robot.api import logger
22 from resources.libraries.python.Constants import Constants
23 from resources.libraries.python.InterfaceUtil import InterfaceUtil
24 from resources.libraries.python.topology import Topology
25 from resources.libraries.python.PapiExecutor import PapiSocketExecutor
28 class NatConfigFlags(IntEnum):
29 """Common NAT plugin APIs"""
31 NAT_IS_TWICE_NAT = 0x01
32 NAT_IS_SELF_TWICE_NAT = 0x02
33 NAT_IS_OUT2IN_ONLY = 0x04
34 NAT_IS_ADDR_ONLY = 0x08
38 NAT_IS_EXT_HOST_VALID = 0x80
41 class NatAddrPortAllocAlg(IntEnum):
42 """NAT Address and port assignment algorithms."""
43 NAT_ALLOC_ALG_DEFAULT = 0
44 NAT_ALLOC_ALG_MAP_E = 1
45 NAT_ALLOC_ALG_PORT_RANGE = 2
49 """This class defines the methods to set NAT."""
55 def set_nat44_interface(node, interface, flag):
56 """Set inside and outside interfaces for NAT44.
58 :param node: DUT node.
59 :param interface: NAT44 interface.
60 :param flag: Interface NAT configuration flag name.
65 cmd = u"nat44_interface_add_del_feature"
67 err_msg = f"Failed to set {flag} interface {interface} for NAT44 " \
68 f"on host {node[u'host']}"
70 sw_if_index=InterfaceUtil.get_sw_if_index(node, interface),
72 flags=getattr(NatConfigFlags, flag).value
75 with PapiSocketExecutor(node) as papi_exec:
76 papi_exec.add(cmd, **args_in).get_reply(err_msg)
79 def set_nat44_interfaces(node, int_in, int_out):
80 """Set inside and outside interfaces for NAT44.
82 :param node: DUT node.
83 :param int_in: Inside interface.
84 :param int_out: Outside interface.
89 NATUtil.set_nat44_interface(node, int_in, u"NAT_IS_INSIDE")
90 NATUtil.set_nat44_interface(node, int_out, u"NAT_IS_OUTSIDE")
93 def set_nat44_address_range(
94 node, start_ip, end_ip, vrf_id=Constants.BITWISE_NON_ZERO,
96 """Set NAT44 address range.
98 :param node: DUT node.
99 :param start_ip: IP range start.
100 :param end_ip: IP range end.
101 :param vrf_id: VRF index (Optional).
102 :param flag: NAT flag name.
109 cmd = u"nat44_add_del_address_range"
110 err_msg = f"Failed to set NAT44 address range on host {node[u'host']}"
113 first_ip_address=IPv4Address(str(start_ip)).packed,
114 last_ip_address=IPv4Address(str(end_ip)).packed,
116 flags=getattr(NatConfigFlags, flag).value
119 with PapiSocketExecutor(node) as papi_exec:
120 papi_exec.add(cmd, **args_in).get_reply(err_msg)
123 def show_nat_config(node):
124 """Show the NAT configuration.
126 :param node: DUT node.
129 cmd = u"nat_show_config"
130 err_msg = f"Failed to get NAT configuration on host {node[u'host']}"
132 with PapiSocketExecutor(node) as papi_exec:
133 reply = papi_exec.add(cmd).get_reply(err_msg)
135 logger.debug(f"NAT Configuration:\n{pformat(reply)}")
138 def show_nat44_summary(node):
139 """Show NAT44 summary on the specified topology node.
141 :param node: Topology node.
144 PapiSocketExecutor.run_cli_cmd(node, u"show nat44 summary")
147 def show_nat_base_data(node):
148 """Show the NAT base data.
153 nat44_interface_addr_dump
155 nat44_static_mapping_dump
158 :param node: DUT node.
163 u"nat44_interface_addr_dump",
164 u"nat44_address_dump",
165 u"nat44_static_mapping_dump",
166 u"nat44_interface_dump",
168 PapiSocketExecutor.dump_and_log(node, cmds)
171 def show_nat_user_data(node):
172 """Show the NAT user data.
177 nat44_user_session_dump
179 :param node: DUT node.
184 u"nat44_user_session_dump",
186 PapiSocketExecutor.dump_and_log(node, cmds)
189 # DET44 means deterministic mode of NAT
191 def enable_det44_plugin(node, inside_vrf=0, outside_vrf=0):
192 """Enable DET44 plugin.
194 :param node: DUT node.
195 :param inside_vrf: Inside VRF ID.
196 :param outside_vrf: Outside VRF ID.
198 :type inside_vrf: str or int
199 :type outside_vrf: str or int
201 cmd = u"det44_plugin_enable_disable"
202 err_msg = f"Failed to enable DET44 plugin on the host {node[u'host']}!"
205 inside_vrf=int(inside_vrf),
206 outside_vrf=int(outside_vrf)
209 with PapiSocketExecutor(node) as papi_exec:
210 papi_exec.add(cmd, **args_in).get_reply(err_msg)
213 def set_det44_interface(node, if_key, is_inside):
214 """Enable DET44 feature on the interface.
216 :param node: DUT node.
217 :param if_key: Interface key from topology file of interface
218 to enable DET44 feature on.
219 :param is_inside: True if interface is inside, False if outside.
222 :type is_inside: bool
224 cmd = u"det44_interface_add_del_feature"
225 err_msg = f"Failed to enable DET44 feature on the interface {if_key} " \
226 f"on the host {node[u'host']}!"
230 sw_if_index=Topology.get_interface_sw_index(node, if_key)
233 with PapiSocketExecutor(node) as papi_exec:
234 papi_exec.add(cmd, **args_in).get_reply(err_msg)
237 def set_det44_mapping(node, ip_in, subnet_in, ip_out, subnet_out):
238 """Set DET44 mapping.
240 :param node: DUT node.
241 :param ip_in: Inside IP.
242 :param subnet_in: Inside IP subnet.
243 :param ip_out: Outside IP.
244 :param subnet_out: Outside IP subnet.
247 :type subnet_in: str or int
249 :type subnet_out: str or int
251 cmd = u"det44_add_del_map"
252 err_msg = f"Failed to set DET44 mapping on the host {node[u'host']}!"
255 in_addr=IPv4Address(str(ip_in)).packed,
256 in_plen=int(subnet_in),
257 out_addr=IPv4Address(str(ip_out)).packed,
258 out_plen=int(subnet_out)
261 with PapiSocketExecutor(node) as papi_exec:
262 papi_exec.add(cmd, **args_in).get_reply(err_msg)
265 def show_det44(node):
274 :param node: DUT node.
278 u"det44_interface_dump",
280 u"det44_session_dump",
282 PapiSocketExecutor.dump_and_log(node, cmds)