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.PapiExecutor import PapiSocketExecutor
  25
  26
  27 class NatConfigFlags(IntEnum):
  28     """Common NAT plugin APIs"""
  29     NAT_IS_NONE = 0x00
  30     NAT_IS_TWICE_NAT = 0x01
  31     NAT_IS_SELF_TWICE_NAT = 0x02
  32     NAT_IS_OUT2IN_ONLY = 0x04
  33     NAT_IS_ADDR_ONLY = 0x08
  34     NAT_IS_OUTSIDE = 0x10
  35     NAT_IS_INSIDE = 0x20
  36     NAT_IS_STATIC = 0x40
  37     NAT_IS_EXT_HOST_VALID = 0x80
  38
  39
  40 class NatAddrPortAllocAlg(IntEnum):
  41     """NAT Address and port assignment algorithms."""
  42     NAT_ALLOC_ALG_DEFAULT = 0
  43     NAT_ALLOC_ALG_MAP_E = 1
  44     NAT_ALLOC_ALG_PORT_RANGE = 2
  45
  46
  47 class NATUtil:
  48     """This class defines the methods to set NAT."""
  49
  50     def __init__(self):
  51         pass
  52
  53     @staticmethod
  54     def set_nat44_interface(node, interface, flag):
  55         """Set inside and outside interfaces for NAT44.
  56
  57         :param node: DUT node.
  58         :param interface: Inside interface.
  59         :param flag: Interface NAT configuration flag name.
  60         :type node: dict
  61         :type interface: str
  62         :type flag: str
  63         """
  64         cmd = u"nat44_interface_add_del_feature"
  65
  66         err_msg = f"Failed to set {flag} interface {interface} for NAT44 " \
  67             f"on host {node[u'host']}"
  68         args_in = dict(
  69             sw_if_index=InterfaceUtil.get_sw_if_index(node, interface),
  70             is_add=1,
  71             flags=getattr(NatConfigFlags, flag).value
  72         )
  73
  74         with PapiSocketExecutor(node) as papi_exec:
  75             papi_exec.add(cmd, **args_in).get_reply(err_msg)
  76
  77     @staticmethod
  78     def set_nat44_interfaces(node, int_in, int_out):
  79         """Set inside and outside interfaces for NAT44.
  80
  81         :param node: DUT node.
  82         :param int_in: Inside interface.
  83         :param int_out: Outside interface.
  84         :type node: dict
  85         :type int_in: str
  86         :type int_out: str
  87         """
  88         NATUtil.set_nat44_interface(node, int_in, u"NAT_IS_INSIDE")
  89         NATUtil.set_nat44_interface(node, int_out, u"NAT_IS_OUTSIDE")
  90
  91     @staticmethod
  92     def set_nat44_deterministic(node, ip_in, subnet_in, ip_out, subnet_out):
  93         """Set deterministic behaviour of NAT44.
  94
  95         :param node: DUT node.
  96         :param ip_in: Inside IP.
  97         :param subnet_in: Inside IP subnet.
  98         :param ip_out: Outside IP.
  99         :param subnet_out: Outside IP subnet.
 100         :type node: dict
 101         :type ip_in: str
 102         :type subnet_in: str or int
 103         :type ip_out: str
 104         :type subnet_out: str or int
 105         """
 106         cmd = u"nat_det_add_del_map"
 107         err_msg = f"Failed to set deterministic behaviour of NAT " \
 108             f"on host {node[u'host']}"
 109         args_in = dict(
 110             is_add=True,
 111             in_addr=IPv4Address(str(ip_in)).packed,
 112             in_plen=int(subnet_in),
 113             out_addr=IPv4Address(str(ip_out)).packed,
 114             out_plen=int(subnet_out)
 115         )
 116
 117         with PapiSocketExecutor(node) as papi_exec:
 118             papi_exec.add(cmd, **args_in).get_reply(err_msg)
 119
 120     @staticmethod
 121     def set_nat44_address_range(
 122             node, start_ip, end_ip, vrf_id=Constants.BITWISE_NON_ZERO,
 123             flag=u"NAT_IS_NONE"):
 124         """Set NAT44 address range.
 125
 126         :param node: DUT node.
 127         :param start_ip: IP range start.
 128         :param end_ip: IP range end.
 129         :param vrf_id: VRF index (Optional).
 130         :param flag: NAT flag name.
 131         :type node: dict
 132         :type start_ip: str
 133         :type end_ip: str
 134         :type vrf_id: int
 135         :type flag: str
 136         """
 137         cmd = u"nat44_add_del_address_range"
 138         err_msg = f"Failed to set NAT44 address range on host {node[u'host']}"
 139         args_in = dict(
 140             is_add=True,
 141             first_ip_address=IPv4Address(str(start_ip)).packed,
 142             last_ip_address=IPv4Address(str(end_ip)).packed,
 143             vrf_id=vrf_id,
 144             flags=getattr(NatConfigFlags, flag).value
 145         )
 146
 147         with PapiSocketExecutor(node) as papi_exec:
 148             papi_exec.add(cmd, **args_in).get_reply(err_msg)
 149
 150     @staticmethod
 151     def show_nat_config(node):
 152         """Show the NAT configuration.
 153
 154         :param node: DUT node.
 155         :type node: dict
 156         """
 157         cmd = u"nat_show_config"
 158         err_msg = f"Failed to get NAT configuration on host {node[u'host']}"
 159
 160         with PapiSocketExecutor(node) as papi_exec:
 161             reply = papi_exec.add(cmd).get_reply(err_msg)
 162
 163         logger.debug(f"NAT Configuration:\n{pformat(reply)}")
 164
 165     @staticmethod
 166     def show_nat44_summary(node):
 167         """Show NAT44 summary on the specified topology node.
 168
 169         :param node: Topology node.
 170         :type node: dict
 171         """
 172         PapiSocketExecutor.run_cli_cmd(node, u"show nat44 summary")
 173
 174     @staticmethod
 175     def show_nat_base_data(node):
 176         """Show the NAT base data.
 177
 178         Used data sources:
 179
 180             nat_worker_dump
 181             nat44_interface_addr_dump
 182             nat44_address_dump
 183             nat44_static_mapping_dump
 184             nat44_interface_dump
 185
 186         :param node: DUT node.
 187         :type node: dict
 188         """
 189         cmd = u"nat_show_config"
 190         err_msg = f"Failed to get NAT base data on host {node[u'host']}"
 191
 192         with PapiSocketExecutor(node) as papi_exec:
 193             reply = papi_exec.add(cmd).get_reply(err_msg)
 194
 195         logger.debug(f"NAT Configuration:\n{pformat(reply)}")
 196
 197         cmds = [
 198             u"nat_worker_dump",
 199             u"nat44_interface_addr_dump",
 200             u"nat44_address_dump",
 201             u"nat44_static_mapping_dump",
 202             u"nat44_interface_dump",
 203         ]
 204         PapiSocketExecutor.dump_and_log(node, cmds)
 205
 206     @staticmethod
 207     def show_nat_user_data(node):
 208         """Show the NAT user data.
 209
 210         Used data sources:
 211
 212             nat44_user_dump
 213             nat44_user_session_dump
 214
 215         :param node: DUT node.
 216         :type node: dict
 217         """
 218         cmd = u"nat_show_config"
 219         err_msg = f"Failed to get NAT user data on host {node[u'host']}"
 220
 221         with PapiSocketExecutor(node) as papi_exec:
 222             reply = papi_exec.add(cmd).get_reply(err_msg)
 223
 224         logger.debug(f"NAT Configuration:\n{pformat(reply)}")
 225
 226         cmds = [
 227             u"nat44_user_dump",
 228             u"nat44_user_session_dump",
 229         ]
 230         PapiSocketExecutor.dump_and_log(node, cmds)