-# 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:
"""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
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."""
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):
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)
)
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)
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)
Code Review / csit.git / blobdiff