diff options
Diffstat (limited to 'resources/libraries/python/NATUtil.py')
| -rw-r--r-- | resources/libraries/python/NATUtil.py | 148 |
1 files changed, 115 insertions, 33 deletions
diff --git a/resources/libraries/python/NATUtil.py b/resources/libraries/python/NATUtil.py index 2d5c1c7b76..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,16 +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 PapiSocketExecutor -class NATConfigFlags(IntEnum): +class NatConfigFlags(IntEnum): """Common NAT plugin APIs""" NAT_IS_NONE = 0x00 NAT_IS_TWICE_NAT = 0x01 @@ -36,6 +37,13 @@ class NATConfigFlags(IntEnum): NAT_IS_EXT_HOST_VALID = 0x80 +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.""" @@ -43,41 +51,42 @@ class NATUtil: 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" - int_in_idx = InterfaceUtil.get_sw_if_index(node, int_in) - err_msg = f"Failed to set inside interface {int_in} for NAT44 " \ + 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, u"NAT_IS_INSIDE").value + flags=getattr(NatConfigFlags, flag).value ) 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 = f"Failed to set outside interface {int_out} for NAT44 " \ - f"on host {node[u'host']}" - args_in = dict( - sw_if_index=int_out_idx, - is_add=1, - flags=getattr(NATConfigFlags, u"NAT_IS_OUTSIDE").value - ) + @staticmethod + def set_nat44_interfaces(node, int_in, int_out): + """Set inside and outside interfaces for NAT44. - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_reply(err_msg) + :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): @@ -99,9 +108,9 @@ class NATUtil: 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) ) @@ -109,26 +118,76 @@ class NATUtil: papi_exec.add(cmd, **args_in).get_reply(err_msg) @staticmethod - def show_nat(node): - """Show the NAT configuration and data. + 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_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 - 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 configuration on host {node[u'host']}" + 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) @@ -140,9 +199,32 @@ class NATUtil: u"nat44_interface_addr_dump", u"nat44_address_dump", u"nat44_static_mapping_dump", - u"nat44_user_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 + """ + 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) + + logger.debug(f"NAT Configuration:\n{pformat(reply)}") + + cmds = [ + u"nat44_user_dump", u"nat44_user_session_dump", - u"nat_det_map_dump" ] PapiSocketExecutor.dump_and_log(node, cmds) |