resources/libraries/python/NATUtil.py

   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:
   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 pprint import pformat
  17 from enum import IntEnum
  18
  19 from ipaddress import IPv4Address
  20 from robot.api import logger
  21
  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
  26
  27
  28 class NatConfigFlags(IntEnum):
  29     """Common NAT plugin APIs"""
  30     NAT_IS_NONE = 0x00
  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
  35     NAT_IS_OUTSIDE = 0x10
  36     NAT_IS_INSIDE = 0x20
  37     NAT_IS_STATIC = 0x40
  38     NAT_IS_EXT_HOST_VALID = 0x80
  39
  40
  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
  46
  47
  48 class NATUtil:
  49     """This class defines the methods to set NAT."""
  50
  51     def __init__(self):
  52         pass
  53
  54     @staticmethod
  55     def set_nat44_interface(node, interface, flag):
  56         """Set inside and outside interfaces for NAT44.
  57
  58         :param node: DUT node.
  59         :param interface: NAT44 interface.
  60         :param flag: Interface NAT configuration flag name.
  61         :type node: dict
  62         :type interface: str
  63         :type flag: str
  64         """
  65         cmd = u"nat44_interface_add_del_feature"
  66
  67         err_msg = f"Failed to set {flag} interface {interface} for NAT44 " \
  68             f"on host {node[u'host']}"
  69         args_in = dict(
  70             sw_if_index=InterfaceUtil.get_sw_if_index(node, interface),
  71             is_add=1,
  72             flags=getattr(NatConfigFlags, flag).value
  73         )
  74
  75         with PapiSocketExecutor(node) as papi_exec:
  76             papi_exec.add(cmd, **args_in).get_reply(err_msg)
  77
  78     @staticmethod
  79     def set_nat44_interfaces(node, int_in, int_out):
  80         """Set inside and outside interfaces for NAT44.
  81
  82         :param node: DUT node.
  83         :param int_in: Inside interface.
  84         :param int_out: Outside interface.
  85         :type node: dict
  86         :type int_in: str
  87         :type int_out: str
  88         """
  89         NATUtil.set_nat44_interface(node, int_in, u"NAT_IS_INSIDE")
  90         NATUtil.set_nat44_interface(node, int_out, u"NAT_IS_OUTSIDE")
  91
  92     @staticmethod
  93     def set_nat44_address_range(
  94             node, start_ip, end_ip, vrf_id=Constants.BITWISE_NON_ZERO,
  95             flag=u"NAT_IS_NONE"):
  96         """Set NAT44 address range.
  97
  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.
 103         :type node: dict
 104         :type start_ip: str
 105         :type end_ip: str
 106         :type vrf_id: int
 107         :type flag: str
 108         """
 109         cmd = u"nat44_add_del_address_range"
 110         err_msg = f"Failed to set NAT44 address range on host {node[u'host']}"
 111         args_in = dict(
 112             is_add=True,
 113             first_ip_address=IPv4Address(str(start_ip)).packed,
 114             last_ip_address=IPv4Address(str(end_ip)).packed,
 115             vrf_id=vrf_id,
 116             flags=getattr(NatConfigFlags, flag).value
 117         )
 118
 119         with PapiSocketExecutor(node) as papi_exec:
 120             papi_exec.add(cmd, **args_in).get_reply(err_msg)
 121
 122     @staticmethod
 123     def show_nat_config(node):
 124         """Show the NAT configuration.
 125
 126         :param node: DUT node.
 127         :type node: dict
 128         """
 129         cmd = u"nat_show_config"
 130         err_msg = f"Failed to get NAT configuration on host {node[u'host']}"
 131
 132         with PapiSocketExecutor(node) as papi_exec:
 133             reply = papi_exec.add(cmd).get_reply(err_msg)
 134
 135         logger.debug(f"NAT Configuration:\n{pformat(reply)}")
 136
 137     @staticmethod
 138     def show_nat44_summary(node):
 139         """Show NAT44 summary on the specified topology node.
 140
 141         :param node: Topology node.
 142         :type node: dict
 143         """
 144         PapiSocketExecutor.run_cli_cmd(node, u"show nat44 summary")
 145
 146     @staticmethod
 147     def show_nat_base_data(node):
 148         """Show the NAT base data.
 149
 150         Used data sources:
 151
 152             nat_worker_dump
 153             nat44_interface_addr_dump
 154             nat44_address_dump
 155             nat44_static_mapping_dump
 156             nat44_interface_dump
 157
 158         :param node: DUT node.
 159         :type node: dict
 160         """
 161         cmds = [
 162             u"nat_worker_dump",
 163             u"nat44_interface_addr_dump",
 164             u"nat44_address_dump",
 165             u"nat44_static_mapping_dump",
 166             u"nat44_interface_dump",
 167         ]
 168         PapiSocketExecutor.dump_and_log(node, cmds)
 169
 170     @staticmethod
 171     def show_nat_user_data(node):
 172         """Show the NAT user data.
 173
 174         Used data sources:
 175
 176             nat44_user_dump
 177             nat44_user_session_dump
 178
 179         :param node: DUT node.
 180         :type node: dict
 181         """
 182         cmds = [
 183             u"nat44_user_dump",
 184             u"nat44_user_session_dump",
 185         ]
 186         PapiSocketExecutor.dump_and_log(node, cmds)
 187
 188     @staticmethod
 189     def compute_max_translations_per_thread(sessions, threads):
 190         """Compute value of max_translations_per_thread NAT44 parameter based on
 191         total number of worker threads.
 192
 193         :param sessions: Required number of NAT44 sessions.
 194         :param threads: Number of worker threads.
 195         :type sessions: int
 196         :type threads: int
 197         :returns: Value of max_translations_per_thread NAT44 parameter.
 198         :rtype: int
 199         """
 200         from math import log2, modf
 201         rest, mult = modf(log2(sessions/(10*threads)))
 202         return 2 ** (int(mult) + (1 if rest else 0)) * 10
 203
 204     # DET44 PAPI calls
 205     # DET44 means deterministic mode of NAT44
 206     @staticmethod
 207     def enable_det44_plugin(node, inside_vrf=0, outside_vrf=0):
 208         """Enable DET44 plugin.
 209
 210         :param node: DUT node.
 211         :param inside_vrf: Inside VRF ID.
 212         :param outside_vrf: Outside VRF ID.
 213         :type node: dict
 214         :type inside_vrf: str or int
 215         :type outside_vrf: str or int
 216         """
 217         cmd = u"det44_plugin_enable_disable"
 218         err_msg = f"Failed to enable DET44 plugin on the host {node[u'host']}!"
 219         args_in = dict(
 220             enable=True,
 221             inside_vrf=int(inside_vrf),
 222             outside_vrf=int(outside_vrf)
 223         )
 224
 225         with PapiSocketExecutor(node) as papi_exec:
 226             papi_exec.add(cmd, **args_in).get_reply(err_msg)
 227
 228     @staticmethod
 229     def set_det44_interface(node, if_key, is_inside):
 230         """Enable DET44 feature on the interface.
 231
 232         :param node: DUT node.
 233         :param if_key: Interface key from topology file of interface
 234             to enable DET44 feature on.
 235         :param is_inside: True if interface is inside, False if outside.
 236         :type node: dict
 237         :type if_key: str
 238         :type is_inside: bool
 239         """
 240         cmd = u"det44_interface_add_del_feature"
 241         err_msg = f"Failed to enable DET44 feature on the interface {if_key} " \
 242             f"on the host {node[u'host']}!"
 243         args_in = dict(
 244             is_add=True,
 245             is_inside=is_inside,
 246             sw_if_index=Topology.get_interface_sw_index(node, if_key)
 247         )
 248
 249         with PapiSocketExecutor(node) as papi_exec:
 250             papi_exec.add(cmd, **args_in).get_reply(err_msg)
 251
 252     @staticmethod
 253     def set_det44_mapping(node, ip_in, subnet_in, ip_out, subnet_out):
 254         """Set DET44 mapping.
 255
 256         :param node: DUT node.
 257         :param ip_in: Inside IP.
 258         :param subnet_in: Inside IP subnet.
 259         :param ip_out: Outside IP.
 260         :param subnet_out: Outside IP subnet.
 261         :type node: dict
 262         :type ip_in: str
 263         :type subnet_in: str or int
 264         :type ip_out: str
 265         :type subnet_out: str or int
 266         """
 267         cmd = u"det44_add_del_map"
 268         err_msg = f"Failed to set DET44 mapping on the host {node[u'host']}!"
 269         args_in = dict(
 270             is_add=True,
 271             in_addr=IPv4Address(str(ip_in)).packed,
 272             in_plen=int(subnet_in),
 273             out_addr=IPv4Address(str(ip_out)).packed,
 274             out_plen=int(subnet_out)
 275         )
 276
 277         with PapiSocketExecutor(node) as papi_exec:
 278             papi_exec.add(cmd, **args_in).get_reply(err_msg)
 279
 280     @staticmethod
 281     def get_det44_mapping(node):
 282         """Get DET44 mapping data.
 283
 284         :param node: DUT node.
 285         :type node: dict
 286         :returns: Dictionary of DET44 mapping data.
 287         :rtype: dict
 288         """
 289         cmd = u"det44_map_dump"
 290         err_msg = f"Failed to get DET44 mapping data on the host " \
 291             f"{node[u'host']}!"
 292         args_in = dict()
 293         with PapiSocketExecutor(node) as papi_exec:
 294             details = papi_exec.add(cmd, **args_in).get_reply(err_msg)
 295
 296         return details
 297
 298     @staticmethod
 299     def get_det44_sessions_number(node):
 300         """Get number of established DET44 sessions from actual DET44 mapping
 301         data.
 302
 303         :param node: DUT node.
 304         :type node: dict
 305         :returns: Number of established DET44 sessions.
 306         :rtype: int
 307         """
 308         det44_data = NATUtil.get_det44_mapping(node)
 309
 310         return det44_data.get(u"ses_num", 0)
 311
 312     @staticmethod
 313     def show_det44(node):
 314         """Show DET44 data.
 315
 316         Used data sources:
 317
 318             det44_interface_dump
 319             det44_map_dump
 320             det44_session_dump
 321
 322         :param node: DUT node.
 323         :type node: dict
 324         """
 325         cmds = [
 326             u"det44_interface_dump",
 327             u"det44_map_dump",
 328             u"det44_session_dump",
 329         ]
 330         PapiSocketExecutor.dump_and_log(node, cmds)