# Copyright (c) 2019 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: # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tap utilities library.""" from ipaddress import ip_address from robot.api import logger from resources.libraries.python.Constants import Constants from resources.libraries.python.InterfaceUtil import InterfaceUtil from resources.libraries.python.L2Util import L2Util from resources.libraries.python.PapiExecutor import PapiSocketExecutor from resources.libraries.python.topology import Topology class Tap: """Tap utilities.""" @staticmethod def add_tap_interface(node, tap_name, mac=None): """Add tap interface with name and optionally with MAC. :param node: Node to add tap on. :param tap_name: Tap interface name for linux tap. :param mac: Optional MAC address for VPP tap. :type node: dict :type tap_name: str :type mac: str :returns: Returns a interface index. :rtype: int """ cmd = u"tap_create_v2" args = dict( id=Constants.BITWISE_NON_ZERO, use_random_mac=0 if mac else 1, mac_address=L2Util.mac_to_bin(mac) if mac else 6 * b"\0", host_namespace=64 * b"\0", host_mac_addr=6 * b"\0", host_if_name_set=1, host_if_name=tap_name.encode(encoding=u"utf-8") + (64 - len(tap_name)) * b"\0", host_bridge=64 * b"\0", host_ip4_addr=4 * b"\0", host_ip6_addr=16 * b"\0", host_ip4_gw=4 * b"\0", host_ip6_gw=16 * b"\0" ) err_msg = f"Failed to create tap interface {tap_name} " \ f"on host {node[u'host']}" with PapiSocketExecutor(node) as papi_exec: sw_if_index = papi_exec.add(cmd, **args).get_sw_if_index(err_msg) if_key = Topology.add_new_port(node, u"tap") Topology.update_interface_sw_if_index(node, if_key, sw_if_index) Topology.update_interface_name(node, if_key, tap_name) if mac is None: mac = Tap.vpp_get_tap_interface_mac(node, tap_name) Topology.update_interface_mac_address(node, if_key, mac) tap_dev_name = Tap.vpp_get_tap_dev_name(node, tap_name) Topology.update_interface_tap_dev_name(node, if_key, tap_dev_name) return sw_if_index @staticmethod def vpp_get_tap_dev_name(node, host_if_name): """Get VPP tap interface name from hardware interfaces dump. :param node: DUT node. :param host_if_name: Tap host interface name. :type node: dict :type host_if_name: str :returns: VPP tap interface dev_name. :rtype: str """ return Tap.tap_dump(node, host_if_name).get(u"dev_name") @staticmethod def vpp_get_tap_interface_mac(node, interface_name): """Get tap interface MAC address from interfaces dump. :param node: DUT node. :param interface_name: Tap interface name. :type node: dict :type interface_name: str :returns: Tap interface MAC address. :rtype: str """ return InterfaceUtil.vpp_get_interface_mac(node, interface_name) @staticmethod def tap_dump(node, name=None): """Get all TAP interface data from the given node, or data about a specific TAP interface. :param node: VPP node to get data from. :param name: Optional name of a specific TAP interface. :type node: dict :type name: str :returns: Dictionary of information about a specific TAP interface, or a List of dictionaries containing all TAP data for the given node. :rtype: dict or list """ def process_tap_dump(tap_dump): """Process tap dump. :param tap_dump: Tap interface dump. :type tap_dump: dict :returns: Processed tap interface dump. :rtype: dict """ tap_dump[u"host_mac_addr"] = L2Util.bin_to_mac( tap_dump[u"host_mac_addr"] ) tap_dump[u"host_ip4_addr"] = ip_address(tap_dump[u"host_ip4_addr"]) tap_dump[u"host_ip6_addr"] = ip_address(tap_dump[u"host_ip6_addr"]) return tap_dump cmd = u"sw_interface_tap_v2_dump" err_msg = f"Failed to get TAP dump on host {node[u'host']}" with PapiSocketExecutor(node) as papi_exec: details = papi_exec.add(cmd).get_details(err_msg) data = list() if name is None else dict() for dump in details: if name is None: data.append(process_tap_dump(dump)) elif dump.get(u"host_if_name") == name: data = process_tap_dump(dump) break logger.debug(f"TAP data:\n{data}") return data