# Copyright (c) 2016 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.
"""Implements IPv4 RobotFramework keywords"""
from socket import inet_ntoa
from struct import pack
from abc import ABCMeta, abstractmethod
import copy
from robot.api import logger as log
from robot.api.deco import keyword
from robot.utils.asserts import assert_not_equal
import resources.libraries.python.ssh as ssh
from resources.libraries.python.topology import Topology
from resources.libraries.python.topology import NodeType
from resources.libraries.python.VatExecutor import VatExecutor
from resources.libraries.python.TrafficScriptExecutor\
import TrafficScriptExecutor
class IPv4Node(object):
"""Abstract class of a node in a topology."""
__metaclass__ = ABCMeta
def __init__(self, node_info):
self.node_info = node_info
@staticmethod
def _get_netmask(prefix_length):
bits = 0xffffffff ^ (1 << 32 - prefix_length) - 1
return inet_ntoa(pack('>I', bits))
@abstractmethod
def set_ip(self, interface, address, prefix_length):
"""Configure IPv4 address on interface
:param interface: interface name
:param address:
:param prefix_length:
:type interface: str
:type address: str
:type prefix_length: int
:return: nothing
"""
pass
@abstractmethod
def set_interface_state(self, interface, state):
"""Set interface state
:param interface: interface name string
:param state: one of following values: "up" or "down"
:return: nothing
"""
pass
@abstractmethod
def set_route(self, network, prefix_length, gateway, interface):
"""Configure IPv4 route
:param network: network IPv4 address
:param prefix_length: mask length
:param gateway: IPv4 address of the gateway
:param interface: interface name
:type network: str
:type prefix_length: int
:type gateway: str
:type interface: str
:return: nothing
"""
pass
@abstractmethod
def unset_route(self, network, prefix_length, gateway, interface):
"""Remove specified IPv4 route
:param network: network IPv4 address
:param prefix_length: mask length
:param gateway: IPv4 address of the gateway
:param interface: interface name
:type network: str
:type prefix_length: int
:type gateway: str
:type interface: str
:return: nothing
"""
pass
@abstractmethod
def flush_ip_addresses(self, interface):
"""Flush all IPv4 addresses from specified interface
:param interface: interface name
:type interface: str
:return: nothing
"""
pass
@abstractmethod
def ping(self, destination_address, source_interface):
"""Send an ICMP request to destination node
:param destination_address: address to send the ICMP request
:param source_interface:
:type destination_address: str
:type source_interface: str
:return: nothing
"""
pass
class Tg(IPv4Node):
"""Traffic generator node"""
def __init__(self, node_info):
super(Tg, self).__init__(node_info)
def _execute(self, cmd):
return ssh.exec_cmd_no_error(self.node_info, cmd)
def _sudo_execute(self, cmd):
return ssh.exec_cmd_no_error(self.node_info, cmd, sudo=True)
def set_ip(self, interface, address, prefix_length):
cmd = 'ip -4 addr flush dev {}'.format(interface)
self._sudo_execute(cmd)
cmd = 'ip addr add {}/{} dev {}'.format(address, prefix_length,
interface)
self._sudo_execute(cmd)
# TODO: not ipv4-specific, move to another class
def set_interface_state(self, interface, state):
cmd = 'ip link set {} {}'.format(interface, state)
self._sudo_execute(cmd)
def set_route(self, network, prefix_length, gateway, interface):
netmask = self._get_netmask(prefix_length)
cmd = 'route add -net {} netmask {} gw {}'.\
format(network, netmask, gateway)
self._sudo_execute(cmd)
def unset_route(self, network, prefix_length, gateway, interface):
self._sudo_execute('ip route delete {}/{}'.
format(network, prefix_length))
def arp_ping(self, destination_address, source_interface):
self._sudo_execute('arping -c 1 -I {} {}'.format(source_interface,
destination_address))
def ping(self, destination_address, source_interface):
self._execute('ping -c 1 -w 5 -I {} {}'.format(source_interface,
destination_address))
def flush_ip_addresses(self, interface):
self._sudo_execute('ip addr flush dev {}'.format(interface))
class Dut(IPv4Node):
"""Device under test"""
def __init__(self, node_info):
super(Dut, self).__init__(node_info)
def get_sw_if_index(self, interface):
"""Get sw_if_index of specified interface from current node
:param interface: interface name
:type interface: str
:return: sw_if_index of 'int' type
"""
return Topology().get_interface_sw_index(self.node_info, interface)
def exec_vat(self, script, **args):
"""Wrapper for VAT executor.
:param script: script to execute
:param args: parameters to the script
:type script: str
:type args: dict
:return: nothing
"""
# TODO: check return value
VatExecutor.cmd_from_template(self.node_info, script, **args)
def set_ip(self, interface, address, prefix_length):
self.exec_vat('add_ip_address.vat',
sw_if_index=self.get_sw_if_index(interface),
address=address, prefix_length=prefix_length)
def set_interface_state(self, interface, state):
if state == 'up':
state = 'admin-up link-up'
elif state == 'down':
state = 'admin-down link-down'
else:
raise Exception('Unexpected interface state: {}'.format(state))
self.exec_vat('set_if_state.vat',
sw_if_index=self.get_sw_if_index(interface), state=state)
def set_route(self, network, prefix_length, gateway, interface):
sw_if_index = self.get_sw_if_index(interface)
self.exec_vat('add_route.vat',
network=network, prefix_length=prefix_length,
gateway=gateway, sw_if_index=sw_if_index)
def unset_route(self, network, prefix_length, gateway, interface):
self.exec_vat('del_route.vat', network=network,
prefix_length=prefix_length, gateway=gateway,
sw_if_index=self.get_sw_if_index(interface))
def arp_ping(self, destination_address, source_interface):
pass
def flush_ip_addresses(self, interface):
self.exec_vat('flush_ip_addresses.vat',
sw_if_index=self.get_sw_if_index(interface))
def ping(self, destination_address, source_interface):
pass
def get_node(node_info):
"""Creates a class instance derived from Node based on type.
:param node_info: dictionary containing information on nodes in topology
:return: Class instance that is derived from Node
"""
if node_info['type'] == NodeType.TG:
return Tg(node_info)
elif node_info['type'] == NodeType.DUT:
return Dut(node_info)
else:
raise NotImplementedError('Node type "{}" unsupported!'.
format(node_info['type']))
def get_node_hostname(node_info):
"""Get string identifying specifed node.
:param node_info: Node in the topology.
:type node_info: Dict
:return: String identifying node.
"""
return node_info['host']
class IPv4Util(object):
"""Implements keywords for IPv4 tests."""
ADDRESSES = {} # holds configured IPv4 addresses
PREFIXES = {} # holds configured IPv4 addresses' prefixes
SUBNETS = {} # holds configured IPv4 addresses' subnets
"""
Helper dictionary used when setting up ipv4 addresses in topology
Example value:
'link1': { b'port1': {b'addr': b'192.168.3.1'},
b'port2': {b'addr': b'192.168.3.2'},
b'prefix': 24,
b'subnet': b'192.168.3.0'}
"""
topology_helper = None
@staticmethod
def next_address(subnet):
"""Get next unused IPv4 address from a subnet
:param subnet: holds available IPv4 addresses
:return: tuple (ipv4_address, prefix_length)
"""
for i in range(1, 4):
# build a key and try to get it from address dictionary
interface = 'port{}'.format(i)
if interface in subnet:
addr = subnet[interface]['addr']
del subnet[interface]
return addr, subnet['prefix']
raise Exception('Not enough ipv4 addresses in subnet')
@staticmethod
def next_network(nodes_addr):
"""Get next unused network from dictionary
:param nodes_addr: dictionary of available networks
:return: dictionary describing an IPv4 subnet with addresses
"""
assert_not_equal(len(nodes_addr), 0, 'Not enough networks')
_, subnet = nodes_addr.popitem()
return subnet
@staticmethod
def configure_ipv4_addr_on_node(node, nodes_addr):
"""Configure IPv4 address for all interfaces on a node in topology
:param node: dictionary containing information about node
:param nodes_addr: dictionary containing IPv4 addresses
:return:
"""
for interface, interface_data in node['interfaces'].iteritems():
if interface == 'mgmt':
continue
if interface_data['link'] not in IPv4Util.topology_helper:
IPv4Util.topology_helper[interface_data['link']] = \
IPv4Util.next_network(nodes_addr)
network = IPv4Util.topology_helper[interface_data['link']]
address, prefix = IPv4Util.next_address(network)
get_node(node).set_ip(interface_data['name'], address, prefix)
key = (get_node_hostname(node), interface_data['name'])
IPv4Util.ADDRESSES[key] = address
IPv4Util.PREFIXES[key] = prefix
IPv4Util.SUBNETS[key] = network['subnet']
@staticmethod
def nodes_setup_ipv4_addresses(nodes_info, nodes_addr):
"""Configure IPv4 addresses on all non-management interfaces for each
node in nodes_info
:param nodes_info: dictionary containing information on all nodes
in topology
:param nodes_addr: dictionary containing IPv4 addresses
:return: nothing
"""
IPv4Util.topology_helper = {}
# make a deep copy of nodes_addr because of modifications
nodes_addr_copy = copy.deepcopy(nodes_addr)
for _, node in nodes_info.iteritems():
IPv4Util.configure_ipv4_addr_on_node(node, nodes_addr_copy)
@staticmethod
def nodes_clear_ipv4_addresses(nodes):
"""Clear all addresses from all nodes in topology
:param nodes: dictionary containing information on all nodes
:return: nothing
"""
for _, node in nodes.iteritems():
for interface, interface_data in node['interfaces'].iteritems():
if interface == 'mgmt':
continue
IPv4Util.flush_ip_addresses(interface_data['name'], node)
# TODO: not ipv4-specific, move to another class
@staticmethod
@keyword('Node "${node}" interface "${interface}" is in "${state}" state')
def set_interface_state(node, interface, state):
"""See IPv4Node.set_interface_state for more information.
:param node:
:param interface:
:param state:
:return:
"""
log.debug('Node {} interface {} is in {} state'.format(
get_node_hostname(node), interface, state))
get_node(node).set_interface_state(interface, state)
@staticmethod
@keyword('Node "${node}" interface "${port}" has IPv4 address '
'"${address}" with prefix length "${prefix_length}"')
def set_interface_address(node, interface, address, length):
"""See IPv4Node.set_ip for more information.
:param node:
:param interface:
:param address:
:param length:
:return:
"""
log.debug('Node {} interface {} has IPv4 address {} with prefix '
'length {}'.format(get_node_hostname(node), interface,
address, length))
get_node(node).set_ip(interface, address, int(length))
hostname = get_node_hostname(node)
IPv4Util.ADDRESSES[hostname, interface] = address
IPv4Util.PREFIXES[hostname, interface] = int(length)
# TODO: Calculate subnet from ip address and prefix length.
# IPv4Util.SUBNETS[hostname, interface] =
@staticmethod
@keyword('From node "${node}" interface "${port}" ARP-ping '
'IPv4 address "${ip_address}"')
def arp_ping(node, interface, ip_address):
log.debug('From node {} interface {} ARP-ping IPv4 address {}'.
format(get_node_hostname(node), interface, ip_address))
get_node(node).arp_ping(ip_address, interface)
@staticmethod
@keyword('Node "${node}" routes to IPv4 network "${network}" with prefix '
'length "${prefix_length}" using interface "${interface}" via '
'"${gateway}"')
def set_route(node, network, prefix_length, interface, gateway):
"""See IPv4Node.set_route for more information.
:param node:
:param network:
:param prefix_length:
:param interface:
:param gateway:
:return:
"""
log.debug('Node {} routes to network {} with prefix length {} '
'via {} interface {}'.format(get_node_hostname(node),
network, prefix_length,
gateway, interface))
get_node(node).set_route(network, int(prefix_length),
gateway, interface)
@staticmethod
@keyword('Remove IPv4 route from "${node}" to network "${network}" with '
'prefix length "${prefix_length}" interface "${interface}" via '
'"${gateway}"')
def unset_route(node, network, prefix_length, interface, gateway):
"""See IPv4Node.unset_route for more information.
:param node:
:param network:
:param prefix_length:
:param interface:
:param gateway:
:return:
"""
get_node(node).unset_route(network, prefix_length, gateway, interface)
@staticmethod
@keyword('After ping is sent from node "${src_node}" interface '
'"${src_port}" with destination IPv4 address of node '
'"${dst_node}" interface "${dst_port}" a ping response arrives '
'and TTL is decreased by "${ttl_dec}"')
def send_ping(src_node, src_port, dst_node, dst_port, hops):
"""Send IPv4 ping and wait for response.
:param src_node: Source node.
:param src_port: Source interface.
:param dst_node: Destination node.
:param dst_port: Destination interface.
:param hops: Number of hops between src_node and dst_node.
"""
log.debug('After ping is sent from node "{}" interface "{}" '
'with destination IPv4 address of node "{}" interface "{}" '
'a ping response arrives and TTL is decreased by "${}"'.
format(get_node_hostname(src_node), src_port,
get_node_hostname(dst_node), dst_port, hops))
node = src_node
src_mac = Topology.get_interface_mac(src_node, src_port)
if dst_node['type'] == NodeType.TG:
dst_mac = Topology.get_interface_mac(src_node, src_port)
adj_int = Topology.get_adjacent_interface(src_node, src_port)
first_hop_mac = adj_int['mac_address']
src_ip = IPv4Util.get_ip_addr(src_node, src_port)
dst_ip = IPv4Util.get_ip_addr(dst_node, dst_port)
args = '--src_if "{}" --src_mac "{}" --first_hop_mac "{}" ' \
'--src_ip "{}" --dst_ip "{}" --hops "{}"'\
.format(src_port, src_mac, first_hop_mac, src_ip, dst_ip, hops)
if dst_node['type'] == NodeType.TG:
args += ' --dst_if "{}" --dst_mac "{}"'.format(dst_port, dst_mac)
TrafficScriptExecutor.run_traffic_script_on_node(
"ipv4_ping_ttl_check.py", node, args)
@staticmethod
@keyword('Get IPv4 address of node "${node}" interface "${port}"')
def get_ip_addr(node, port):
"""Get IPv4 address configured on specified interface
:param node: node dictionary
:param port: interface name
:return: IPv4 address of specified interface as a 'str' type
"""
log.debug('Get IPv4 address of node {} interface {}'.
format(get_node_hostname(node), port))
return IPv4Util.ADDRESSES[(get_node_hostname(node), port)]
@staticmethod
@keyword('Get IPv4 address prefix of node "${node}" interface "${port}"')
def get_ip_addr_prefix(node, port):
""" Get IPv4 address prefix for specified interface.
:param node: Node dictionary.
:param port: Interface name.
"""
log.debug('Get IPv4 address prefix of node {} interface {}'.
format(get_node_hostname(node), port))
return IPv4Util.PREFIXES[(get_node_hostname(node), port)]
@staticmethod
@keyword('Get IPv4 subnet of node "${node}" interface "${port}"')
def get_ip_addr_subnet(node, port):
""" Get IPv4 subnet of specified interface.
:param node: Node dictionary.
:param port: Interface name.
"""
log.debug('Get IPv4 subnet of node {} interface {}'.
format(get_node_hostname(node), port))
return IPv4Util.SUBNETS[(get_node_hostname(node), port)]
@staticmethod
@keyword('Flush IPv4 addresses "${port}" "${node}"')
def flush_ip_addresses(port, node):
"""See IPv4Node.flush_ip_addresses for more information.
:param port:
:param node:
:return:
"""
key = (get_node_hostname(node), port)
del IPv4Util.ADDRESSES[key]
del IPv4Util.PREFIXES[key]
del IPv4Util.SUBNETS[key]
get_node(node).flush_ip_addresses(port)