Source code for autotest.client.net.net_utils

"""Convenience functions for use by network tests or whomever.

This library is to release in the public repository.
"""

import os
import re
import socket
import struct
import time

from autotest.client import utils
from autotest.client.shared import error

TIMEOUT = 10  # Used for socket timeout and barrier timeout


[docs]class network_utils(object):
[docs] def reset(self, ignore_status=False): utils.system('service network restart', ignore_status=ignore_status)
[docs] def start(self, ignore_status=False): utils.system('service network start', ignore_status=ignore_status)
[docs] def stop(self, ignore_status=False): utils.system('service network stop', ignore_status=ignore_status)
[docs] def list(self): utils.system('/sbin/ifconfig -a')
[docs] def get_ip_local(self, query_ip, netmask="24"): """ Get ip address in local system which can communicate with query_ip. :param query_ip: IP of client which wants to communicate with autotest machine. :return: IP address which can communicate with query_ip """ ip = utils.system_output("ip addr show to %s/%s" % (query_ip, netmask)) ip = re.search(r"inet ([0-9.]*)/", ip) if ip is None: return ip return ip.group(1)
[docs] def disable_ip_local_loopback(self, ignore_status=False): utils.system("echo '1' > /proc/sys/net/ipv4/route/no_local_loopback", ignore_status=ignore_status) utils.system('echo 1 > /proc/sys/net/ipv4/route/flush', ignore_status=ignore_status)
[docs] def enable_ip_local_loopback(self, ignore_status=False): utils.system("echo '0' > /proc/sys/net/ipv4/route/no_local_loopback", ignore_status=ignore_status) utils.system('echo 1 > /proc/sys/net/ipv4/route/flush', ignore_status=ignore_status)
[docs] def process_mpstat(self, mpstat_out, sample_count, loud=True): """Parses mpstat output of the following two forms: 02:10:17 0 0.00 0.00 0.00 0.00 0.00 0.00 \ 0.00 100.00 1012.87 02:10:13 PM 0 0.00 0.00 0.00 0.00 0.00 0.00 \ 0.00 100.00 1019.00 """ mpstat_keys = ['time', 'CPU', 'user', 'nice', 'sys', 'iowait', 'irq', 'soft', 'steal', 'idle', 'intr/s'] if loud: print mpstat_out # Remove the optional AM/PM appearing in time format mpstat_out = mpstat_out.replace('AM', '') mpstat_out = mpstat_out.replace('PM', '') regex = re.compile('(\S+)') stats = [] for line in mpstat_out.splitlines()[3:]: match = regex.findall(line) # Skip the "Average" computed by mpstat. We are gonna compute the # average ourself. Pick only the aggregate 'all' CPU results if match and match[0] != 'Average:' and match[1] == 'all': stats.append(dict(zip(mpstat_keys, match))) if sample_count >= 5: # Throw away first and last sample stats = stats[1:-1] cpu_stats = {} for key in ['user', 'nice', 'sys', 'iowait', 'irq', 'soft', 'steal', 'idle', 'intr/s']: x = [float(row[key]) for row in stats] if len(x): count = len(x) else: print 'net_utils.network_utils.process_mpstat: count is 0!!!\n' count = 1 cpu_stats[key] = sum(x) / count return cpu_stats
[docs]def network(): try: # Here we are importing site net utils only if it exists # pylint: disable=E0611 from autotest.client.net import site_net_utils return site_net_utils.network_utils() except ImportError: return network_utils()
[docs]class network_interface(object): ENABLE, DISABLE = (True, False) def __init__(self, name): self.ethtool = 'ethtool' self._name = name self.exists = self.exists() self.was_down = self.is_down() self.orig_ipaddr = self.get_ipaddr() self.was_loopback_enabled = self.is_loopback_enabled() self._socket = socket.socket(socket.PF_PACKET, socket.SOCK_RAW) self._socket.settimeout(TIMEOUT) self._socket.bind((name, raw_socket.ETH_P_ALL))
[docs] def restore(self): self.set_ipaddr(self.orig_ipaddr) # TODO (msb): The additional conditional guard needs cleanup: # Underlying driver should simply perform a noop # for disabling loopback on an already-disabled device, # instead of returning non-zero exit code. # To avoid sending a RST to the autoserv ssh connection # don't disable loopback until the IP address is restored. if not self.was_loopback_enabled and self.is_loopback_enabled(): self.disable_loopback() if self.was_down: self.down()
[docs] def get_name(self): return self._name
[docs] def parse_ethtool(self, field, match, option='', next_field=''): output = utils.system_output('%s %s %s' % (self.ethtool, option, self._name)) if output: match = re.search('\n\s*%s:\s*(%s)%s' % (field, match, next_field), output, re.S) if match: return match.group(1) return ''
[docs] def get_stats(self): stats = {} stats_path = '/sys/class/net/%s/statistics/' % self._name for stat in os.listdir(stats_path): f = open(stats_path + stat, 'r') if f: stats[stat] = int(f.read()) f.close() return stats
[docs] def get_stats_diff(self, orig_stats): stats = self.get_stats() for stat in stats.keys(): if stat in orig_stats: stats[stat] = stats[stat] - orig_stats[stat] else: stats[stat] = stats[stat] return stats
[docs] def get_driver(self): driver_path = os.readlink('/sys/class/net/%s/device/driver' % self._name) return os.path.basename(driver_path)
[docs] def get_carrier(self): f = open('/sys/class/net/%s/carrier' % self._name) if not f: return '' carrier = f.read().strip() f.close() return carrier
[docs] def is_autoneg_advertised(self): result = self.parse_ethtool('Advertised auto-negotiation', 'Yes|No') return result == 'Yes'
[docs] def get_speed(self): return int(self.parse_ethtool('Speed', '\d+'))
[docs] def is_full_duplex(self): result = self.parse_ethtool('Duplex', 'Full|Half') return result == 'Full'
[docs] def is_autoneg_on(self): result = self.parse_ethtool('Auto-negotiation', 'on|off') return result == 'on'
[docs] def get_wakeon(self): return self.parse_ethtool('Wake-on', '\w+')
[docs] def is_rx_summing_on(self): result = self.parse_ethtool('rx-checksumming', 'on|off', '-k') return result == 'on'
[docs] def is_tx_summing_on(self): result = self.parse_ethtool('tx-checksumming', 'on|off', '-k') return result == 'on'
[docs] def is_scatter_gather_on(self): result = self.parse_ethtool('scatter-gather', 'on|off', '-k') return result == 'on'
[docs] def is_tso_on(self): result = self.parse_ethtool('tcp segmentation offload', 'on|off', '-k') return result == 'on'
[docs] def is_pause_autoneg_on(self): result = self.parse_ethtool('Autonegotiate', 'on|off', '-a') return result == 'on'
[docs] def is_tx_pause_on(self): result = self.parse_ethtool('TX', 'on|off', '-a') return result == 'on'
[docs] def is_rx_pause_on(self): result = self.parse_ethtool('RX', 'on|off', '-a') return result == 'on'
def _set_loopback(self, mode, enable_disable): return utils.system('%s -L %s %s %s' % (self.ethtool, self._name, mode, enable_disable), ignore_status=True)
[docs] def enable_loopback(self): # If bonded do not set loopback mode. # Try mac loopback first then phy loopback # If both fail, raise an error if bond().is_enabled(): raise error.TestError('Unable to enable loopback while ' 'bonding is enabled.') if (self._set_loopback('phyint', 'enable') > 0 and self._set_loopback('mac', 'enable') > 0): raise error.TestError('Unable to enable loopback') # Add a 1 second wait for drivers which do not have # a synchronous loopback enable # TODO (msb); Remove this wait once the drivers are fixed if self.get_driver() in ['tg3', 'bnx2x']: time.sleep(1) self.wait_for_carrier(timeout=30)
[docs] def disable_loopback(self): # Try mac loopback first then phy loopback # If both fail, raise an error if (self._set_loopback('phyint', 'disable') > 0 and self._set_loopback('mac', 'disable') > 0): raise error.TestError('Unable to disable loopback')
[docs] def is_loopback_enabled(self): # Don't try ethtool -l on a bonded host if bond().is_enabled(): return False output = utils.system_output('%s -l %s' % (self.ethtool, self._name)) if output: return 'enabled' in output return False
[docs] def enable_promisc(self): utils.system('ifconfig %s promisc' % self._name)
[docs] def disable_promisc(self): utils.system('ifconfig %s -promisc' % self._name)
[docs] def get_hwaddr(self): f = open('/sys/class/net/%s/address' % self._name) hwaddr = f.read().strip() f.close() return hwaddr
[docs] def set_hwaddr(self, hwaddr): utils.system('ifconfig %s hw ether %s' % (self._name, hwaddr))
[docs] def add_maddr(self, maddr): utils.system('ip maddr add %s dev %s' % (maddr, self._name))
[docs] def del_maddr(self, maddr): utils.system('ip maddr del %s dev %s' % (maddr, self._name))
[docs] def get_ipaddr(self): ipaddr = "0.0.0.0" output = utils.system_output('ifconfig %s' % self._name) if output: match = re.search("inet addr:([\d\.]+)", output) if match: ipaddr = match.group(1) return ipaddr
[docs] def set_ipaddr(self, ipaddr): utils.system('ifconfig %s %s' % (self._name, ipaddr))
# pylint: disable=E0202
[docs] def exists(self): return not utils.system('ifconfig %s' % self._name, ignore_status=True)
[docs] def is_down(self): output = utils.system_output('ifconfig %s' % self._name) if output: return 'UP' not in output return False
[docs] def up(self): utils.system('ifconfig %s up' % self._name)
[docs] def down(self): utils.system('ifconfig %s down' % self._name)
[docs] def wait_for_carrier(self, timeout=60): while timeout and self.get_carrier() != '1': timeout -= 1 time.sleep(1) if timeout == 0: raise error.TestError('Timed out waiting for carrier.')
[docs] def send(self, buf): self._socket.send(buf)
[docs] def recv(self, len): return self._socket.recv(len)
[docs] def flush(self): self._socket.close() self._socket = socket.socket(socket.PF_PACKET, socket.SOCK_RAW) self._socket.settimeout(TIMEOUT) self._socket.bind((self._name, raw_socket.ETH_P_ALL))
[docs]def netif(name): try: # Here we are importing site net utils only if it exists # pylint: disable=E0611 from autotest.client.net import site_net_utils return site_net_utils.network_interface(name) except ImportError: return network_interface(name)
[docs]class bonding(object): """This class implements bonding interface abstraction.""" NO_MODE = 0 AB_MODE = 1 AD_MODE = 2
[docs] def is_enabled(self): raise error.TestError('Undefined')
[docs] def is_bondable(self): raise error.TestError('Undefined')
[docs] def enable(self): raise error.TestError('Undefined')
[docs] def disable(self): raise error.TestError('Undefined')
[docs] def get_mii_status(self): return {}
[docs] def get_mode(self): return bonding.NO_MODE
[docs] def wait_for_state_change(self): """Wait for bonding state change. Wait up to 90 seconds to successfully ping the gateway. This is to know when LACP state change has converged. (0 seconds is 3x lacp timeout, use by protocol) """ netif('eth0').wait_for_carrier(timeout=60) wait_time = 0 while wait_time < 100: time.sleep(10) if not utils.ping_default_gateway(): return True wait_time += 10 return False
[docs] def get_active_interfaces(self): return []
[docs] def get_slave_interfaces(self): return []
[docs]def bond(): try: # Here we are importing site net utils only if it exists # pylint: disable=E0611 from autotest.client.net import site_net_utils return site_net_utils.bonding() except ImportError: return bonding()
[docs]class raw_socket(object): """This class implements an raw socket abstraction.""" ETH_P_ALL = 0x0003 # Use for binding a RAW Socket to all protocols SOCKET_TIMEOUT = 1 def __init__(self, iface_name): """Initialize an interface for use. Args: iface_name: 'eth0' interface name ('eth0, eth1,...') """ self._name = iface_name self._socket = None self._socket_timeout = raw_socket.SOCKET_TIMEOUT socket.setdefaulttimeout(self._socket_timeout) if self._name is None: raise error.TestError('Invalid interface name')
[docs] def socket(self): return self._socket
[docs] def socket_timeout(self): """Get the timeout use by recv_from""" return self._socket_timeout
[docs] def set_socket_timeout(self, timeout): """Set the timeout use by recv_from. Args: timeout: time in seconds """ self._socket_timeout = timeout
[docs] def open(self, protocol=None): """Opens the raw socket to send and receive. Args: protocol : short in host byte order. None if ALL """ if self._socket is not None: raise error.TestError('Raw socket already open') if protocol is None: self._socket = socket.socket(socket.PF_PACKET, socket.SOCK_RAW) self._socket.bind((self._name, self.ETH_P_ALL)) else: self._socket = socket.socket(socket.PF_PACKET, socket.SOCK_RAW, socket.htons(protocol)) self._socket.bind((self._name, self.ETH_P_ALL)) self._socket.settimeout(1) # always running with 1 second timeout
[docs] def close(self): """ Close the raw socket""" if self._socket is not None: self._socket.close() self._socket = None else: raise error.TestError('Raw socket not open')
[docs] def recv(self, timeout): """Synchroneous receive. Receives one packet from the interface and returns its content in a string. Wait up to timeout for the packet if timeout is not 0. This function filters out all the packets that are less than the minimum ethernet packet size (60+crc). Args: timeout: max time in seconds to wait for the read to complete. '0', wait for ever until a valid packet is received Returns: packet: None no packet was received a binary string containing the received packet. time_left: amount of time left in timeout """ if self._socket is None: raise error.TestError('Raw socket not open') time_left = timeout packet = None while time_left or (timeout == 0): try: packet = self._socket.recv(ethernet.ETH_PACKET_MAX_SIZE) if len(packet) >= (ethernet.ETH_PACKET_MIN_SIZE - 4): break packet = None if timeout and time_left: time_left -= raw_socket.SOCKET_TIMEOUT except socket.timeout: packet = None if timeout and time_left: time_left -= raw_socket.SOCKET_TIMEOUT return packet, time_left
[docs] def send(self, packet): """Send an ethernet packet.""" if self._socket is None: raise error.TestError('Raw socket not open') self._socket.send(packet)
[docs] def send_to(self, dst_mac, src_mac, protocol, payload): """Send an ethernet frame. Send an ethernet frame, formating the header. Args: dst_mac: 'byte string' src_mac: 'byte string' protocol: short in host byte order payload: 'byte string' """ if self._socket is None: raise error.TestError('Raw socket not open') try: packet = ethernet.pack(dst_mac, src_mac, protocol, payload) except Exception: raise error.TestError('Invalid Packet') self.send(packet)
[docs] def recv_from(self, dst_mac, src_mac, protocol): """Receive an ethernet frame that matches the dst, src and proto. Filters all received packet to find a matching one, then unpack it and present it to the caller as a frame. Waits up to self._socket_timeout for a matching frame before returning. Args: dst_mac: 'byte string'. None do not use in filter. src_mac: 'byte string'. None do not use in filter. protocol: short in host byte order. None do not use in filter. Returns: ethernet frame: { 'dst' : byte string, 'src' : byte string, 'proto' : short in host byte order, 'payload' : byte string } """ start_time = time.clock() timeout = self._socket_timeout while 1: frame = None packet, timeout = self.recv(timeout) if packet is not None: frame = ethernet.unpack(packet) if ((src_mac is None or frame['src'] == src_mac) and (dst_mac is None or frame['dst'] == dst_mac) and (protocol is None or frame['proto'] == protocol)): break elif (timeout == 0 or time.clock() - start_time > float(self._socket_timeout)): frame = None break else: if (timeout == 0 or time.clock() - start_time > float(self._socket_timeout)): frame = None break continue return frame
[docs]class ethernet(object): """Provide ethernet packet manipulation methods.""" HDR_LEN = 14 # frame header length CHECKSUM_LEN = 4 # frame checksum length # Ethernet payload types - http://standards.ieee.org/regauth/ethertype ETH_TYPE_IP = 0x0800 # IP protocol ETH_TYPE_ARP = 0x0806 # address resolution protocol ETH_TYPE_CDP = 0x2000 # Cisco Discovery Protocol ETH_TYPE_8021Q = 0x8100 # IEEE 802.1Q VLAN tagging ETH_TYPE_IP6 = 0x86DD # IPv6 protocol ETH_TYPE_LOOPBACK = 0x9000 # used to test interfaces ETH_TYPE_LLDP = 0x88CC # LLDP frame type ETH_PACKET_MAX_SIZE = 1518 # maximum ethernet frane size ETH_PACKET_MIN_SIZE = 64 # minimum ethernet frane size ETH_LLDP_DST_MAC = '01:80:C2:00:00:0E' # LLDP destination mac FRAME_KEY_DST_MAC = 'dst' # frame destination mac address FRAME_KEY_SRC_MAC = 'src' # frame source mac address FRAME_KEY_PROTO = 'proto' # frame protocol FRAME_KEY_PAYLOAD = 'payload' # frame payload def __init__(self): pass @staticmethod
[docs] def mac_string_to_binary(hwaddr): """Converts a MAC address text string to byte string. Converts a MAC text string from a text string 'aa:aa:aa:aa:aa:aa' to a byte string 'xxxxxxxxxxxx' Args: hwaddr: a text string containing the MAC address to convert. Returns: A byte string. """ val = ''.join([chr(b) for b in [int(c, 16) for c in hwaddr.split(':', 6)]]) return val
@staticmethod
[docs] def mac_binary_to_string(hwaddr): """Converts a MAC address byte string to text string. Converts a MAC byte string 'xxxxxxxxxxxx' to a text string 'aa:aa:aa:aa:aa:aa' Args: hwaddr: a byte string containing the MAC address to convert. Returns: A text string. """ return "%02x:%02x:%02x:%02x:%02x:%02x" % tuple(map(ord, hwaddr))
@staticmethod
[docs] def pack(dst, src, protocol, payload): """Pack a frame in a byte string. Args: dst: destination mac in byte string format src: src mac address in byte string format protocol: short in network byte order payload: byte string payload data Returns: An ethernet frame with header and payload in a byte string. """ # numbers are converted to network byte order (!) frame = struct.pack("!6s6sH", dst, src, protocol) + payload return frame
@staticmethod
[docs] def unpack(raw_frame): """Unpack a raw ethernet frame. Returns: None on error { 'dst' : byte string, 'src' : byte string, 'proto' : short in host byte order, 'payload' : byte string } """ packet_len = len(raw_frame) if packet_len < ethernet.HDR_LEN: return None payload_len = packet_len - ethernet.HDR_LEN frame = {} frame[ethernet.FRAME_KEY_DST_MAC], \ frame[ethernet.FRAME_KEY_SRC_MAC], \ frame[ethernet.FRAME_KEY_PROTO] = \ struct.unpack("!6s6sH", raw_frame[:ethernet.HDR_LEN]) frame[ethernet.FRAME_KEY_PAYLOAD] = \ raw_frame[ethernet.HDR_LEN:ethernet.HDR_LEN + payload_len] return frame
[docs]def ethernet_packet(): try: # Here we are importing site net utils only if it exists # pylint: disable=E0611 from autotest.client.net import site_net_utils return site_net_utils.ethernet() except ImportError: return ethernet()