X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=resources%2Flibraries%2Fpython%2FNATUtil.py;h=4ce72d84b480868972e8a2cdb364c1ad1d5357bf;hb=e08706e85b412b1307df3789fdbe747b43c2bd95;hp=f018d38335f54bdb9d4f5ceae554570c613f6c74;hpb=f88a3d9178dfbd73d0479f9aa2f5224e0c89ca1f;p=csit.git diff --git a/resources/libraries/python/NATUtil.py b/resources/libraries/python/NATUtil.py index f018d38335..4ce72d84b4 100644 --- a/resources/libraries/python/NATUtil.py +++ b/resources/libraries/python/NATUtil.py @@ -1,4 +1,4 @@ -# Copyright (c) 2019 Cisco and/or its affiliates. +# Copyright (c) 2020 Cisco and/or its affiliates. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at: @@ -14,17 +14,18 @@ """NAT utilities library.""" from pprint import pformat -from socket import AF_INET, inet_pton - from enum import IntEnum +from ipaddress import IPv4Address from robot.api import logger +from resources.libraries.python.Constants import Constants from resources.libraries.python.InterfaceUtil import InterfaceUtil +from resources.libraries.python.topology import Topology from resources.libraries.python.PapiExecutor import PapiSocketExecutor -class NATConfigFlags(IntEnum): +class NatConfigFlags(IntEnum): """Common NAT plugin APIs""" NAT_IS_NONE = 0x00 NAT_IS_TWICE_NAT = 0x01 @@ -37,12 +38,43 @@ class NATConfigFlags(IntEnum): NAT_IS_EXT_HOST_VALID = 0x80 -class NATUtil(object): +class NatAddrPortAllocAlg(IntEnum): + """NAT Address and port assignment algorithms.""" + NAT_ALLOC_ALG_DEFAULT = 0 + NAT_ALLOC_ALG_MAP_E = 1 + NAT_ALLOC_ALG_PORT_RANGE = 2 + + +class NATUtil: """This class defines the methods to set NAT.""" def __init__(self): pass + @staticmethod + def set_nat44_interface(node, interface, flag): + """Set inside and outside interfaces for NAT44. + + :param node: DUT node. + :param interface: NAT44 interface. + :param flag: Interface NAT configuration flag name. + :type node: dict + :type interface: str + :type flag: str + """ + cmd = u"nat44_interface_add_del_feature" + + err_msg = f"Failed to set {flag} interface {interface} for NAT44 " \ + f"on host {node[u'host']}" + args_in = dict( + sw_if_index=InterfaceUtil.get_sw_if_index(node, interface), + is_add=1, + flags=getattr(NatConfigFlags, flag).value + ) + + with PapiSocketExecutor(node) as papi_exec: + papi_exec.add(cmd, **args_in).get_reply(err_msg) + @staticmethod def set_nat44_interfaces(node, int_in, int_out): """Set inside and outside interfaces for NAT44. @@ -54,34 +86,156 @@ class NATUtil(object): :type int_in: str :type int_out: str """ + NATUtil.set_nat44_interface(node, int_in, u"NAT_IS_INSIDE") + NATUtil.set_nat44_interface(node, int_out, u"NAT_IS_OUTSIDE") - cmd = 'nat44_interface_add_del_feature' + @staticmethod + def set_nat44_address_range( + node, start_ip, end_ip, vrf_id=Constants.BITWISE_NON_ZERO, + flag=u"NAT_IS_NONE"): + """Set NAT44 address range. - int_in_idx = InterfaceUtil.get_sw_if_index(node, int_in) - err_msg = 'Failed to set inside interface {int} for NAT44 on host ' \ - '{host}'.format(int=int_in, host=node['host']) + :param node: DUT node. + :param start_ip: IP range start. + :param end_ip: IP range end. + :param vrf_id: VRF index (Optional). + :param flag: NAT flag name. + :type node: dict + :type start_ip: str + :type end_ip: str + :type vrf_id: int + :type flag: str + """ + cmd = u"nat44_add_del_address_range" + err_msg = f"Failed to set NAT44 address range on host {node[u'host']}" args_in = dict( - sw_if_index=int_in_idx, - is_add=1, - flags=getattr(NATConfigFlags, "NAT_IS_INSIDE").value + is_add=True, + first_ip_address=IPv4Address(str(start_ip)).packed, + last_ip_address=IPv4Address(str(end_ip)).packed, + vrf_id=vrf_id, + flags=getattr(NatConfigFlags, flag).value + ) + + with PapiSocketExecutor(node) as papi_exec: + papi_exec.add(cmd, **args_in).get_reply(err_msg) + + @staticmethod + def show_nat_config(node): + """Show the NAT configuration. + + :param node: DUT node. + :type node: dict + """ + cmd = u"nat_show_config" + err_msg = f"Failed to get NAT configuration on host {node[u'host']}" + + with PapiSocketExecutor(node) as papi_exec: + reply = papi_exec.add(cmd).get_reply(err_msg) + + logger.debug(f"NAT Configuration:\n{pformat(reply)}") + + @staticmethod + def show_nat44_summary(node): + """Show NAT44 summary on the specified topology node. + + :param node: Topology node. + :type node: dict + """ + PapiSocketExecutor.run_cli_cmd(node, u"show nat44 summary") + + @staticmethod + def show_nat_base_data(node): + """Show the NAT base data. + + Used data sources: + + nat_worker_dump + nat44_interface_addr_dump + nat44_address_dump + nat44_static_mapping_dump + nat44_interface_dump + + :param node: DUT node. + :type node: dict + """ + cmds = [ + u"nat_worker_dump", + u"nat44_interface_addr_dump", + u"nat44_address_dump", + u"nat44_static_mapping_dump", + u"nat44_interface_dump", + ] + PapiSocketExecutor.dump_and_log(node, cmds) + + @staticmethod + def show_nat_user_data(node): + """Show the NAT user data. + + Used data sources: + + nat44_user_dump + nat44_user_session_dump + + :param node: DUT node. + :type node: dict + """ + cmds = [ + u"nat44_user_dump", + u"nat44_user_session_dump", + ] + PapiSocketExecutor.dump_and_log(node, cmds) + + # DET44 PAPI calls + # DET44 means deterministic mode of NAT + @staticmethod + def enable_det44_plugin(node, inside_vrf=0, outside_vrf=0): + """Enable DET44 plugin. + + :param node: DUT node. + :param inside_vrf: Inside VRF ID. + :param outside_vrf: Outside VRF ID. + :type node: dict + :type inside_vrf: str or int + :type outside_vrf: str or int + """ + cmd = u"det44_plugin_enable_disable" + err_msg = f"Failed to enable DET44 plugin on the host {node[u'host']}!" + args_in = dict( + enable=True, + inside_vrf=int(inside_vrf), + outside_vrf=int(outside_vrf) ) + with PapiSocketExecutor(node) as papi_exec: papi_exec.add(cmd, **args_in).get_reply(err_msg) - int_out_idx = InterfaceUtil.get_sw_if_index(node, int_out) - err_msg = 'Failed to set outside interface {int} for NAT44 on host ' \ - '{host}'.format(int=int_out, host=node['host']) + @staticmethod + def set_det44_interface(node, if_key, is_inside): + """Enable DET44 feature on the interface. + + :param node: DUT node. + :param if_key: Interface key from topology file of interface + to enable DET44 feature on. + :param is_inside: True if interface is inside, False if outside. + :type node: dict + :type if_key: str + :type is_inside: bool + """ + cmd = u"det44_interface_add_del_feature" + err_msg = f"Failed to enable DET44 feature on the interface {if_key} " \ + f"on the host {node[u'host']}!" args_in = dict( - sw_if_index=int_out_idx, - is_add=1, - flags=getattr(NATConfigFlags, "NAT_IS_OUTSIDE").value + is_add=True, + is_inside=is_inside, + sw_if_index=Topology.get_interface_sw_index(node, if_key) ) + with PapiSocketExecutor(node) as papi_exec: papi_exec.add(cmd, **args_in).get_reply(err_msg) @staticmethod - def set_nat44_deterministic(node, ip_in, subnet_in, ip_out, subnet_out): - """Set deterministic behaviour of NAT44. + def set_det44_mapping(node, ip_in, subnet_in, ip_out, subnet_out): + """Set DET44 mapping. :param node: DUT node. :param ip_in: Inside IP. @@ -94,55 +248,35 @@ class NATUtil(object): :type ip_out: str :type subnet_out: str or int """ - - cmd = 'nat_det_add_del_map' - err_msg = 'Failed to set deterministic behaviour of NAT on host ' \ - '{host}'.format(host=node['host']) + cmd = u"det44_add_del_map" + err_msg = f"Failed to set DET44 mapping on the host {node[u'host']}!" args_in = dict( is_add=True, - in_addr=inet_pton(AF_INET, str(ip_in)), + in_addr=IPv4Address(str(ip_in)).packed, in_plen=int(subnet_in), - out_addr=inet_pton(AF_INET, str(ip_out)), + out_addr=IPv4Address(str(ip_out)).packed, out_plen=int(subnet_out) ) + with PapiSocketExecutor(node) as papi_exec: papi_exec.add(cmd, **args_in).get_reply(err_msg) @staticmethod - def show_nat(node): - """Show the NAT configuration and data. + def show_det44(node): + """Show DET44 data. Used data sources: - nat_show_config - nat_worker_dump - nat44_interface_addr_dump - nat44_address_dump - nat44_static_mapping_dump - nat44_user_dump - nat44_interface_dump - nat44_user_session_dump - nat_det_map_dump + det44_interface_dump + det44_map_dump + det44_session_dump :param node: DUT node. :type node: dict """ - - cmd = 'nat_show_config' - err_msg = 'Failed to get NAT configuration on host {host}'.\ - format(host=node['host']) - with PapiSocketExecutor(node) as papi_exec: - reply = papi_exec.add(cmd).get_reply(err_msg) - logger.debug("NAT Configuration:\n{reply}".format(reply=pformat(reply))) - cmds = [ - "nat_worker_dump", - "nat44_interface_addr_dump", - "nat44_address_dump", - "nat44_static_mapping_dump", - "nat44_user_dump", - "nat44_interface_dump", - "nat44_user_session_dump", - "nat_det_map_dump" + u"det44_interface_dump", + u"det44_map_dump", + u"det44_session_dump", ] PapiSocketExecutor.dump_and_log(node, cmds)