X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=resources%2Flibraries%2Fpython%2FNATUtil.py;h=b43058b23fefd26873fdc04f2952208b64f33b41;hb=f58d415afaacc7565f08817903b0d21f16579eb8;hp=aeeb3bc1a57f4028503a85f22c95cfe0ec4fc488;hpb=287ef8c6dc057220e403a1493cb23450dd8db2b2;p=csit.git diff --git a/resources/libraries/python/NATUtil.py b/resources/libraries/python/NATUtil.py index aeeb3bc1a5..b43058b23f 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,17 @@ """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.PapiExecutor import PapiExecutor +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,49 +37,56 @@ 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_interfaces(node, int_in, int_out): + def set_nat44_interface(node, interface, flag): """Set inside and outside interfaces for NAT44. :param node: DUT node. - :param int_in: Inside interface. - :param int_out: Outside interface. + :param interface: Inside interface. + :param flag: Interface NAT configuration flag name. :type node: dict - :type int_in: str - :type int_out: str + :type interface: str + :type flag: str """ + cmd = u"nat44_interface_add_del_feature" - cmd = 'nat44_interface_add_del_feature' - - 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']) + err_msg = f"Failed to set {flag} interface {interface} for NAT44 " \ + f"on host {node[u'host']}" args_in = dict( - sw_if_index=int_in_idx, + sw_if_index=InterfaceUtil.get_sw_if_index(node, interface), is_add=1, - flags=getattr(NATConfigFlags, "NAT_IS_INSIDE").value + flags=getattr(NatConfigFlags, flag).value ) - with PapiExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_replies(err_msg).\ - verify_reply(err_msg=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']) - args_in = dict( - sw_if_index=int_out_idx, - is_add=1, - flags=getattr(NATConfigFlags, "NAT_IS_OUTSIDE").value - ) - with PapiExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_replies(err_msg). \ - verify_reply(err_msg=err_msg) + 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. + + :param node: DUT node. + :param int_in: Inside interface. + :param int_out: Outside interface. + :type node: dict + :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") @staticmethod def set_nat44_deterministic(node, ip_in, subnet_in, ip_out, subnet_out): @@ -96,57 +103,128 @@ 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"nat_det_add_del_map" + err_msg = f"Failed to set deterministic behaviour of NAT " \ + f"on 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 PapiExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_replies(err_msg). \ - verify_reply(err_msg=err_msg) + + with PapiSocketExecutor(node) as papi_exec: + papi_exec.add(cmd, **args_in).get_reply(err_msg) + + @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. + + :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( + 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(node): - """Show the NAT configuration and data. + 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_show_config nat_worker_dump nat44_interface_addr_dump nat44_address_dump nat44_static_mapping_dump - nat44_user_dump nat44_interface_dump + + :param node: DUT node. + :type node: dict + """ + cmd = u"nat_show_config" + err_msg = f"Failed to get NAT base data 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)}") + + 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 - nat_det_map_dump :param node: DUT node. :type node: dict """ + cmd = u"nat_show_config" + err_msg = f"Failed to get NAT user data on host {node[u'host']}" + + with PapiSocketExecutor(node) as papi_exec: + reply = papi_exec.add(cmd).get_reply(err_msg) - cmd = 'nat_show_config' - err_msg = 'Failed to get NAT configuration on host {host}'.\ - format(host=node['host']) - with PapiExecutor(node) as papi_exec: - data = papi_exec.add(cmd).get_replies(err_msg).\ - verify_reply(err_msg=err_msg) - logger.debug("NAT Configuration:\n{data}".format(data=pformat(data))) + logger.debug(f"NAT Configuration:\n{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"nat44_user_dump", + u"nat44_user_session_dump", ] - PapiExecutor.dump_and_log(node, cmds) + PapiSocketExecutor.dump_and_log(node, cmds)