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     # DET44 PAPI calls
 189     # DET44 means deterministic mode of NAT44
 190     @staticmethod
 191     def enable_det44_plugin(node, inside_vrf=0, outside_vrf=0):
 192         """Enable DET44 plugin.
 193
 194         :param node: DUT node.
 195         :param inside_vrf: Inside VRF ID.
 196         :param outside_vrf: Outside VRF ID.
 197         :type node: dict
 198         :type inside_vrf: str or int
 199         :type outside_vrf: str or int
 200         """
 201         cmd = u"det44_plugin_enable_disable"
 202         err_msg = f"Failed to enable DET44 plugin on the host {node[u'host']}!"
 203         args_in = dict(
 204             enable=True,
 205             inside_vrf=int(inside_vrf),
 206             outside_vrf=int(outside_vrf)
 207         )
 208
 209         with PapiSocketExecutor(node) as papi_exec:
 210             papi_exec.add(cmd, **args_in).get_reply(err_msg)
 211
 212     @staticmethod
 213     def set_det44_interface(node, if_key, is_inside):
 214         """Enable DET44 feature on the interface.
 215
 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.
 220         :type node: dict
 221         :type if_key: str
 222         :type is_inside: bool
 223         """
 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']}!"
 227         args_in = dict(
 228             is_add=True,
 229             is_inside=is_inside,
 230             sw_if_index=Topology.get_interface_sw_index(node, if_key)
 231         )
 232
 233         with PapiSocketExecutor(node) as papi_exec:
 234             papi_exec.add(cmd, **args_in).get_reply(err_msg)
 235
 236     @staticmethod
 237     def set_det44_mapping(node, ip_in, subnet_in, ip_out, subnet_out):
 238         """Set DET44 mapping.
 239
 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.
 245         :type node: dict
 246         :type ip_in: str
 247         :type subnet_in: str or int
 248         :type ip_out: str
 249         :type subnet_out: str or int
 250         """
 251         cmd = u"det44_add_del_map"
 252         err_msg = f"Failed to set DET44 mapping on the host {node[u'host']}!"
 253         args_in = dict(
 254             is_add=True,
 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)
 259         )
 260
 261         with PapiSocketExecutor(node) as papi_exec:
 262             papi_exec.add(cmd, **args_in).get_reply(err_msg)
 263
 264     @staticmethod
 265     def get_det44_mapping(node):
 266         """Get DET44 mapping data.
 267
 268         :param node: DUT node.
 269         :type node: dict
 270         :returns: Dictionary of DET44 mapping data.
 271         :rtype: dict
 272         """
 273         cmd = u"det44_map_dump"
 274         err_msg = f"Failed to get DET44 mapping data on the host " \
 275             f"{node[u'host']}!"
 276         args_in = dict()
 277         with PapiSocketExecutor(node) as papi_exec:
 278             details = papi_exec.add(cmd, **args_in).get_reply(err_msg)
 279
 280         return details
 281
 282     @staticmethod
 283     def get_det44_sessions_number(node):
 284         """Get number of established DET44 sessions from actual DET44 mapping
 285         data.
 286
 287         :param node: DUT node.
 288         :type node: dict
 289         :returns: Number of established DET44 sessions.
 290         :rtype: int
 291         """
 292         det44_data = NATUtil.get_det44_mapping(node)
 293
 294         return det44_data.get(u"ses_num", 0)
 295
 296     @staticmethod
 297     def show_det44(node):
 298         """Show DET44 data.
 299
 300         Used data sources:
 301
 302             det44_interface_dump
 303             det44_map_dump
 304             det44_session_dump
 305
 306         :param node: DUT node.
 307         :type node: dict
 308         """
 309         cmds = [
 310             u"det44_interface_dump",
 311             u"det44_map_dump",
 312             u"det44_session_dump",
 313         ]
 314         PapiSocketExecutor.dump_and_log(node, cmds)