""" test framework utilities """
+import ipaddress
+import logging
import socket
-import sys
+from socket import AF_INET6
import os.path
-from abc import abstractmethod, ABCMeta
-from scapy.utils6 import in6_mactoifaceid
+import platform
+from copy import deepcopy
+from collections import UserDict
+import scapy.compat
from scapy.layers.l2 import Ether
-from scapy.packet import Raw
from scapy.layers.inet import IP
-from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, IPv6ExtHdrRouting,\
- IPv6ExtHdrHopByHop
+from scapy.layers.inet6 import (
+ IPv6,
+ IPv6ExtHdrFragment,
+ IPv6ExtHdrRouting,
+ IPv6ExtHdrHopByHop,
+)
+from scapy.packet import Raw
from scapy.utils import hexdump
-from socket import AF_INET6
+from scapy.utils6 import in6_mactoifaceid
+
from io import BytesIO
-from vpp_mac import mactobinary
+from vpp_papi import mac_pton
+
+# Set up an empty logger for the testcase that can be overridden as necessary
+null_logger = logging.getLogger("VppTestCase.util")
+null_logger.addHandler(logging.NullHandler())
+
+
+def pr(packet):
+ return packet.__repr__()
def ppp(headline, packet):
- """ Return string containing the output of scapy packet.show() call. """
- o = BytesIO()
- old_stdout = sys.stdout
- sys.stdout = o
- print(headline)
- hexdump(packet)
- print("")
- packet.show()
- sys.stdout = old_stdout
- return o.getvalue()
+ """Return string containing headline and output of scapy packet.show()"""
+ return "%s\n%s\n\n%s\n" % (
+ headline,
+ hexdump(packet, dump=True),
+ packet.show(dump=True),
+ )
def ppc(headline, capture, limit=10):
- """ Return string containing ppp() printout for a capture.
+ """Return string containing ppp() printout for a capture.
:param headline: printed as first line of output
:param capture: packets to print
tail = ""
if limit < len(capture):
tail = "\nPrint limit reached, %s out of %s packets printed" % (
- limit, len(capture))
- body = "".join([ppp("Packet #%s:" % count, p)
- for count, p in zip(range(0, limit), capture)])
+ limit,
+ len(capture),
+ )
+ body = "".join(
+ [ppp("Packet #%s:" % count, p) for count, p in zip(range(0, limit), capture)]
+ )
return "%s\n%s%s" % (headline, body, tail)
def ip4_range(ip4, s, e):
- tmp = ip4.rsplit('.', 1)[0]
+ tmp = ip4.rsplit(".", 1)[0]
return ("%s.%d" % (tmp, i) for i in range(s, e))
-def ip4n_range(ip4n, s, e):
- ip4 = socket.inet_ntop(socket.AF_INET, ip4n)
- return (socket.inet_pton(socket.AF_INET, ip)
- for ip in ip4_range(ip4, s, e))
+def mcast_ip_to_mac(ip):
+ ip = ipaddress.ip_address(ip)
+ if not ip.is_multicast:
+ raise ValueError("Must be multicast address.")
+ ip_as_int = int(ip)
+ if ip.version == 4:
+ mcast_mac = "01:00:5e:%02x:%02x:%02x" % (
+ (ip_as_int >> 16) & 0x7F,
+ (ip_as_int >> 8) & 0xFF,
+ ip_as_int & 0xFF,
+ )
+ else:
+ mcast_mac = "33:33:%02x:%02x:%02x:%02x" % (
+ (ip_as_int >> 24) & 0xFF,
+ (ip_as_int >> 16) & 0xFF,
+ (ip_as_int >> 8) & 0xFF,
+ ip_as_int & 0xFF,
+ )
+ return mcast_mac
+# wrapper around scapy library function.
def mk_ll_addr(mac):
- euid = in6_mactoifaceid(mac)
+ euid = in6_mactoifaceid(str(mac))
addr = "fe80::" + euid
return addr
def ip6_normalize(ip6):
- return socket.inet_ntop(socket.AF_INET6,
- socket.inet_pton(socket.AF_INET6, ip6))
+ return socket.inet_ntop(socket.AF_INET6, socket.inet_pton(socket.AF_INET6, ip6))
def get_core_path(tempdir):
def get_core_pattern():
- with open("/proc/sys/kernel/core_pattern", "r") as f:
- corefmt = f.read().strip()
+ if platform.uname().system == "FreeBSD":
+ import sysctl
+
+ corefmt = sysctl.filter("kern.corefile")[0].value
+ elif platform.uname().system == "Linux":
+ with open("/proc/sys/kernel/core_pattern", "r") as f:
+ corefmt = f.read().strip()
return corefmt
if corefmt.startswith("|"):
logger.error(
"WARNING: redirecting the core dump through a"
- " filter may result in truncated dumps.")
+ " filter may result in truncated dumps."
+ )
logger.error(
" You may want to check the filter settings"
" or uninstall it and edit the"
- " /proc/sys/kernel/core_pattern accordingly.")
- logger.error(
- " current core pattern is: %s" % corefmt)
-
+ " /proc/sys/kernel/core_pattern accordingly."
+ )
+ logger.error(" current core pattern is: %s" % corefmt)
-class NumericConstant(object):
- __metaclass__ = ABCMeta
+class NumericConstant:
desc_dict = {}
- @abstractmethod
def __init__(self, value):
self._value = value
return ""
-class Host(object):
- """ Generic test host "connected" to VPPs interface. """
+class Host:
+ """Generic test host "connected" to VPPs interface."""
@property
def mac(self):
- """ MAC address """
+ """MAC address"""
return self._mac
@property
def bin_mac(self):
- """ MAC address """
- return mactobinary(self._mac)
+ """MAC address"""
+ return mac_pton(self._mac)
@property
def ip4(self):
- """ IPv4 address - string """
+ """IPv4 address - string"""
return self._ip4
@property
def ip4n(self):
- """ IPv4 address of remote host - raw, suitable as API parameter."""
+ """IPv4 address of remote host - raw, suitable as API parameter."""
return socket.inet_pton(socket.AF_INET, self._ip4)
@property
def ip6(self):
- """ IPv6 address - string """
+ """IPv6 address - string"""
return self._ip6
@property
def ip6n(self):
- """ IPv6 address of remote host - raw, suitable as API parameter."""
+ """IPv6 address of remote host - raw, suitable as API parameter."""
return socket.inet_pton(socket.AF_INET6, self._ip6)
@property
def ip6_ll(self):
- """ IPv6 link-local address - string """
+ """IPv6 link-local address - string"""
return self._ip6_ll
@property
def ip6n_ll(self):
- """ IPv6 link-local address of remote host -
+ """IPv6 link-local address of remote host -
raw, suitable as API parameter."""
return socket.inet_pton(socket.AF_INET6, self._ip6_ll)
def __eq__(self, other):
if isinstance(other, Host):
- return (self.mac == other.mac and
- self.ip4 == other.ip4 and
- self.ip6 == other.ip6 and
- self.ip6_ll == other.ip6_ll)
+ return (
+ self.mac == other.mac
+ and self.ip4 == other.ip4
+ and self.ip6 == other.ip6
+ and self.ip6_ll == other.ip6_ll
+ )
else:
return False
return not self.__eq__(other)
def __repr__(self):
- return "Host { mac:%s ip4:%s ip6:%s ip6_ll:%s }" % (self.mac,
- self.ip4,
- self.ip6,
- self.ip6_ll)
+ return "Host { mac:%s ip4:%s ip6:%s ip6_ll:%s }" % (
+ self.mac,
+ self.ip4,
+ self.ip6,
+ self.ip6_ll,
+ )
def __hash__(self):
return hash(self.__repr__())
self._ip6_ll = ip6_ll
-class ForeignAddressFactory(object):
- count = 0
- prefix_len = 24
- net_template = '10.10.10.{}'
- net = net_template.format(0) + '/' + str(prefix_len)
-
- def get_ip4(self):
- if self.count > 255:
- raise Exception("Network host address exhaustion")
- self.count += 1
- return self.net_template.format(self.count)
-
-
-class L4_Conn():
- """ L4 'connection' tied to two VPP interfaces """
+class L4_Conn:
+ """L4 'connection' tied to two VPP interfaces"""
def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
self.testcase = testcase
s1 = 1 - side
src_if = self.ifs[s0]
dst_if = self.ifs[s1]
- layer_3 = [IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
- IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)]
- merged_l4args = {'sport': self.ports[s0], 'dport': self.ports[s1]}
+ layer_3 = [
+ IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
+ IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6),
+ ]
+ merged_l4args = {"sport": self.ports[s0], "dport": self.ports[s1]}
merged_l4args.update(l4args)
- p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
- layer_3[is_ip6] /
- self.l4proto(**merged_l4args) /
- Raw(payload))
+ p = (
+ Ether(dst=src_if.local_mac, src=src_if.remote_mac)
+ / layer_3[is_ip6]
+ / self.l4proto(**merged_l4args)
+ / Raw(payload)
+ )
return p
def send(self, side, flags=None, payload=""):
l4args = {}
if flags is not None:
- l4args['flags'] = flags
- self.ifs[side].add_stream(self.pkt(side,
- l4args=l4args, payload=payload))
+ l4args["flags"] = flags
+ self.ifs[side].add_stream(self.pkt(side, l4args=l4args, payload=payload))
self.ifs[1 - side].enable_capture()
self.testcase.pg_start()
L4_CONN_SIDE_ONE = 1
-class LoggerWrapper(object):
- def __init__(self, logger=None):
- self._logger = logger
-
- def debug(self, *args, **kwargs):
- if self._logger:
- self._logger.debug(*args, **kwargs)
-
- def error(self, *args, **kwargs):
- if self._logger:
- self._logger.error(*args, **kwargs)
-
-
-def fragment_rfc791(packet, fragsize, _logger=None):
+def fragment_rfc791(packet, fragsize, logger=null_logger):
"""
Fragment an IPv4 packet per RFC 791
:param packet: packet to fragment
:note: IP options are not supported
:returns: list of fragments
"""
- logger = LoggerWrapper(_logger)
logger.debug(ppp("Fragmenting packet:", packet))
- packet = packet.__class__(str(packet)) # recalculate all values
+ packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
if len(packet[IP].options) > 0:
raise Exception("Not implemented")
if len(packet) <= fragsize:
pre_ip_len = len(packet) - len(packet[IP])
ip_header_len = packet[IP].ihl * 4
- hex_packet = str(packet)
- hex_headers = hex_packet[:(pre_ip_len + ip_header_len)]
- hex_payload = hex_packet[(pre_ip_len + ip_header_len):]
+ hex_packet = scapy.compat.raw(packet)
+ hex_headers = hex_packet[: (pre_ip_len + ip_header_len)]
+ hex_payload = hex_packet[(pre_ip_len + ip_header_len) :]
pkts = []
ihl = packet[IP].ihl
otl = len(packet[IP])
- nfb = (fragsize - pre_ip_len - ihl * 4) / 8
+ nfb = int((fragsize - pre_ip_len - ihl * 4) / 8)
fo = packet[IP].frag
- p = packet.__class__(hex_headers + hex_payload[:nfb * 8])
+ p = packet.__class__(hex_headers + hex_payload[: nfb * 8])
p[IP].flags = "MF"
p[IP].frag = fo
p[IP].len = ihl * 4 + nfb * 8
del p[IP].chksum
pkts.append(p)
- p = packet.__class__(hex_headers + hex_payload[nfb * 8:])
+ p = packet.__class__(hex_headers + hex_payload[nfb * 8 :])
p[IP].len = otl - nfb * 8
p[IP].frag = fo + nfb
del p[IP].chksum
- more_fragments = fragment_rfc791(p, fragsize, _logger)
+ more_fragments = fragment_rfc791(p, fragsize, logger)
pkts.extend(more_fragments)
return pkts
-def fragment_rfc8200(packet, identification, fragsize, _logger=None):
+def fragment_rfc8200(packet, identification, fragsize, logger=null_logger):
"""
Fragment an IPv6 packet per RFC 8200
:param packet: packet to fragment
:note: IP options are not supported
:returns: list of fragments
"""
- logger = LoggerWrapper(_logger)
- packet = packet.__class__(str(packet)) # recalculate all values
+ packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
if len(packet) <= fragsize:
return [packet]
logger.debug(ppp("Fragmenting packet:", packet))
routing_hdr = counter
elif l.__class__ is IPv6ExtHdrHopByHop:
hop_by_hop_hdr = counter
- elif seen_ipv6 and not upper_layer and \
- not l.__class__.__name__.startswith('IPv6ExtHdr'):
+ elif (
+ seen_ipv6
+ and not upper_layer
+ and not l.__class__.__name__.startswith("IPv6ExtHdr")
+ ):
upper_layer = counter
counter = counter + 1
l = packet.getlayer(counter)
logger.debug(
- "Layers seen: IPv6(#%s), Routing(#%s), HopByHop(#%s), upper(#%s)" %
- (ipv6_nr, routing_hdr, hop_by_hop_hdr, upper_layer))
+ "Layers seen: IPv6(#%s), Routing(#%s), HopByHop(#%s), upper(#%s)"
+ % (ipv6_nr, routing_hdr, hop_by_hop_hdr, upper_layer)
+ )
if upper_layer is None:
raise Exception("Upper layer header not found in IPv6 packet")
logger.debug(ppp("Per-fragment headers:", per_fragment_headers))
ext_and_upper_layer = packet.getlayer(last_per_fragment_hdr)[1]
- hex_payload = str(ext_and_upper_layer)
+ hex_payload = scapy.compat.raw(ext_and_upper_layer)
logger.debug("Payload length is %s" % len(hex_payload))
logger.debug(ppp("Ext and upper layer:", ext_and_upper_layer))
fragment_ext_hdr = IPv6ExtHdrFragment()
logger.debug(ppp("Fragment header:", fragment_ext_hdr))
- if len(per_fragment_headers) + len(fragment_ext_hdr) +\
- len(ext_and_upper_layer) - len(ext_and_upper_layer.payload)\
- > fragsize:
- raise Exception("Cannot fragment this packet - MTU too small "
- "(%s, %s, %s, %s, %s)" % (
- len(per_fragment_headers), len(fragment_ext_hdr),
- len(ext_and_upper_layer),
- len(ext_and_upper_layer.payload), fragsize))
+ len_ext_and_upper_layer_payload = len(ext_and_upper_layer.payload)
+ if not len_ext_and_upper_layer_payload and hasattr(ext_and_upper_layer, "data"):
+ len_ext_and_upper_layer_payload = len(ext_and_upper_layer.data)
+
+ if (
+ len(per_fragment_headers)
+ + len(fragment_ext_hdr)
+ + len(ext_and_upper_layer)
+ - len_ext_and_upper_layer_payload
+ > fragsize
+ ):
+ raise Exception(
+ "Cannot fragment this packet - MTU too small "
+ "(%s, %s, %s, %s, %s)"
+ % (
+ len(per_fragment_headers),
+ len(fragment_ext_hdr),
+ len(ext_and_upper_layer),
+ len_ext_and_upper_layer_payload,
+ fragsize,
+ )
+ )
orig_nh = packet[IPv6].nh
p = per_fragment_headers
del p[IPv6].nh
p = p / fragment_ext_hdr
del p[IPv6ExtHdrFragment].nh
- first_payload_len_nfb = (fragsize - len(p)) / 8
- p = p / Raw(hex_payload[:first_payload_len_nfb * 8])
+ first_payload_len_nfb = int((fragsize - len(p)) / 8)
+ p = p / Raw(hex_payload[: first_payload_len_nfb * 8])
del p[IPv6].plen
p[IPv6ExtHdrFragment].nh = orig_nh
p[IPv6ExtHdrFragment].id = identification
p[IPv6ExtHdrFragment].offset = 0
p[IPv6ExtHdrFragment].m = 1
- p = p.__class__(str(p))
+ p = p.__class__(scapy.compat.raw(p))
logger.debug(ppp("Fragment %s:" % len(pkts), p))
pkts.append(p)
offset = first_payload_len_nfb * 8
del p[IPv6].nh
p = p / fragment_ext_hdr
del p[IPv6ExtHdrFragment].nh
- l_nfb = (fragsize - len(p)) / 8
- p = p / Raw(hex_payload[offset:offset + l_nfb * 8])
+ l_nfb = int((fragsize - len(p)) / 8)
+ p = p / Raw(hex_payload[offset : offset + l_nfb * 8])
p[IPv6ExtHdrFragment].nh = orig_nh
p[IPv6ExtHdrFragment].id = identification
- p[IPv6ExtHdrFragment].offset = offset / 8
+ p[IPv6ExtHdrFragment].offset = int(offset / 8)
p[IPv6ExtHdrFragment].m = 1
- p = p.__class__(str(p))
+ p = p.__class__(scapy.compat.raw(p))
logger.debug(ppp("Fragment %s:" % len(pkts), p))
pkts.append(p)
offset = offset + l_nfb * 8
first = listoffragments[0]
buffer.seek(20)
for pkt in listoffragments:
- buffer.seek(pkt[IP].frag*8)
+ buffer.seek(pkt[IP].frag * 8)
buffer.write(bytes(pkt[IP].payload))
first.len = len(buffer.getvalue()) + 20
first.flags = 0
- del(first.chksum)
+ del first.chksum
if return_ip:
header = bytes(first[IP])[:20]
return first[IP].__class__(header + buffer.getvalue())
def reassemble4(listoffragments):
return reassemble4_core(listoffragments, True)
+
+
+class UnexpectedPacketError(Exception):
+ def __init__(self, packet, msg=""):
+ self.packet = packet
+ self.msg = msg
+
+ def __str__(self):
+ return f"\nUnexpected packet:\n{pr(self.packet)}{self.msg}"
+
+
+def recursive_dict_merge(dict_base, dict_update):
+ """Recursively merge base dict with update dict, return merged dict"""
+ for key in dict_update:
+ if key in dict_base:
+ if type(dict_update[key]) is dict:
+ dict_base[key] = recursive_dict_merge(dict_base[key], dict_update[key])
+ else:
+ dict_base[key] = dict_update[key]
+ else:
+ dict_base[key] = dict_update[key]
+ return dict_base
+
+
+class StatsDiff(UserDict):
+ """
+ Diff dictionary is a dictionary of dictionaries of interesting stats:
+
+ diff_dictionary =
+ {
+ "err" : { '/error/counter1' : 4, },
+ sw_if_index1 : { '/stat/segment/counter1' : 5,
+ '/stat/segment/counter2' : 6,
+ },
+ sw_if_index2 : { '/stat/segment/counter1' : 7,
+ },
+ }
+
+ It describes a per sw-if-index diffset, where each key is stat segment
+ path and value is the expected change for that counter for sw-if-index.
+ Special case string "err" is used for error counters, which are not per
+ sw-if-index.
+ """
+
+ __slots__ = () # prevent setting properties to act like a dictionary
+
+ def __init__(self, data):
+ super().__init__(data)
+
+ def __or__(self, other):
+ return recursive_dict_merge(deepcopy(self.data), other)
Code Review / vpp.git / blobdiff